Python 官方文档:入门教程 => 点击学习
签名 import base64 import JSON import time from datetime import datetime import hMac from
签名
import base64
import JSON
import time
from datetime import datetime
import hMac
from hashlib import sha1
access_key_id = ''
# 请填写您的AccessKeySecret。
access_key_secret = ''
# host的格式为 bucketname.endpoint ,请替换为您的真实信息。
host = ''
# callback_url为 上传回调服务器的URL,请将下面的IP和Port配置为您自己的真实信息。
callback_url = ""
# 用户上传文件时指定的前缀。
upload_dir = 'user-dir-prefix/'
expire_time = 1200
expire_syncpoint = int(time.time() + expire_time)
policy_dict = {
'expiration': datetime.utcfromtimestamp(expire_syncpoint).isofORMat() + 'Z',
'conditions': [
{"bucket": "test-paige"},
['starts-with', '$key', 'user/test/']
]
}
policy = json.dumps(policy_dict).strip()
policy_encode = base64.b64encode(policy.encode())
signature = base64.encodebytes(hmac.new(access_key_secret.encode(), policy_encode, sha1).digest())
callback_dict = {
'callbackUrl': callback_url,
'callbackBody': 'filename=${object}&size=${size}&mimeType=${mimeType}&height=${imageInfo.height}&width=${'
'imageInfo.width}',
'callbackBodyType': 'application/json'
}
callback = base64.b64encode(json.dumps(callback_dict).strip().encode()).decode()
var = {
'accessid': access_key_id,
'host': host,
'policy': policy_encode.decode(),
'signature': signature.decode().strip(),
'expire': expire_syncpoint,
'callback': callback
}
回调验签
import asyncio
import base64
import time
import aioMysql
import rsa
from aioHttp import WEB, ClientSession
from urllib import parse
import uuid
def success(msg='', data=None):
if data is None:
data = {}
dict_data = {
'code': 1,
'msg': msg,
'data': data
}
return web.json_response(dict_data)
def failed(msg='', data=None):
if data is None:
data = {}
dict_data = {
'code': 0,
'msg': msg,
'data': data
}
return web.json_response(dict_data)
async def handle(request):
"""
获取连接池
:param web.BaseRequest request:
:return:
"""
authorization_base64 = request.headers['authorization']
x_oss_pub_key_url_base64 = request.headers['x-oss-pub-key-url']
pub_key_url = base64.b64decode(x_oss_pub_key_url_base64.encode())
authorization = base64.b64decode(authorization_base64.encode())
path = request.path
async with ClientSession() as session:
async with session.get(pub_key_url.decode()) as resp:
pub_key_body = await resp.text()
pubkey = rsa.PublicKey.load_pkcs1_openssl_pem(pub_key_body.encode())
body = await request.content.read()
auth_str = parse.unquote(path) + '\n' + body.decode()
parse_url = parse.parse_qs(body.decode())
print(parse_url)
try:
rsa.verify(auth_str.encode(), authorization, pubkey)
pool = request.app['mysql_pool']
async with pool.acquire() as conn:
async with conn.cursor() as cur:
id = str(uuid.uuid4())
url = parse_url['filename'][0]
mime = parse_url['mimeType'][0]
disk = 'oss'
time_at = time.strftime("%Y-%m-%d %H:%I:%S", time.localtime())
sql = "INSERT INTO media(id,url,mime,disk,created_at,updated_at) VALUES(%s,%s,%s,%s,%s,%s)"
await cur.execute(sql, (id, url, mime, disk, time_at, time_at))
await conn.commit()
dict_data = {
'id': id,
'url': url,
'cdn_url': 'https://cdn.***.net' + '/' + url,
'mime': mime,
'disk': disk,
'created_at': time_at,
'updated_at': time_at,
}
return success(data=dict_data)
except rsa.pkcs1.VerificationError:
return failed(msg='验证错误')
async def init(loop):
# 创建连接池
mysql_pool = await aiomysql.create_pool(host='127.0.0.1', port=3306,
user='', passWord='',
db='', loop=loop)
async def on_shutdown(application):
"""
接收到关闭信号时,要先关闭连接池,并等待连接池关闭成功.
:param web.Application application:
:return:
"""
application['mysql_pool'].close()
# 没有下面这句话会报错 RuntimeError: Event loop is closed ,因为连接池没有真正关关闭程序就关闭了,引发python的报错
await application['mysql_pool'].wait_closed()
application = web.Application()
application.on_shutdown.append(on_shutdown)
# 把连接池放到 application 实例中
application['mysql_pool'] = mysql_pool
application.add_routes([web.get('/', handle), web.post('/oss', handle)])
return application
if __name__ == '__main__':
loop = asyncio.get_event_loop()
application = loop.run_until_complete(init(loop))
web.run_app(application, host='127.0.0.1')
loop.close()
到此这篇关于Python 阿里云oss实现直传签名与回调验证的文章就介绍到这了,更多相关python 直传签名与回调验证内容请搜索编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程网!
--结束END--
本文标题: python 阿里云oss实现直传签名与回调验证的示例方法
本文链接: https://www.lsjlt.com/news/122536.html(转载时请注明来源链接)
有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341
下载Word文档到电脑,方便收藏和打印~
2024-03-01
2024-03-01
2024-03-01
2024-02-29
2024-02-29
2024-02-29
2024-02-29
2024-02-29
2024-02-29
2024-02-29
回答
回答
回答
回答
回答
回答
回答
回答
回答
回答
0