python 阿里云oss实现直传签名与回调验证的示例方法
更新时间:2021年3月29日 15:00 点击:2132
签名
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 直传签名与回调验证内容请搜索猪先飞以前的文章或继续浏览下面的相关文章希望大家以后多多支持猪先飞!
相关文章
- 这篇文章主要介绍了python-opencv-画外接矩形框的实例代码,代码简单易懂,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下...2021-09-04
Python astype(np.float)函数使用方法解析
这篇文章主要介绍了Python astype(np.float)函数使用方法解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下...2020-06-08- 2022虎年新年即将来临,小编为大家带来了一个利用Python编写的虎年烟花特效,堪称全网最绚烂,文中的示例代码简洁易懂,感兴趣的同学可以动手试一试...2022-02-14
- 在本篇文章里小编给大家分享的是一篇关于python中numpy.empty()函数实例讲解内容,对此有兴趣的朋友们可以学习下。...2021-02-06
python-for x in range的用法(注意要点、细节)
这篇文章主要介绍了python-for x in range的用法,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧...2021-05-10- 这篇文章主要介绍了Python 图片转数组,二进制互转操作,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧...2021-03-09
- 这篇文章主要介绍了Python中的imread()函数用法说明,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧...2021-03-16
- 这篇文章主要介绍了python如何实现b站直播自动发送弹幕,帮助大家更好的理解和学习使用python,感兴趣的朋友可以了解下...2021-02-20
python Matplotlib基础--如何添加文本和标注
这篇文章主要介绍了python Matplotlib基础--如何添加文本和标注,帮助大家更好的利用Matplotlib绘制图表,感兴趣的朋友可以了解下...2021-01-26- 这篇文章主要介绍了解决python 使用openpyxl读写大文件的坑,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧...2021-03-13
- 今天小编就为大家分享一篇python 计算方位角实例(根据两点的坐标计算),具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧...2020-04-27
- 这篇文章主要为大家详细介绍了python实现双色球随机选号,文中示例代码介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们可以参考一下...2020-05-02
- 在本篇文章里小编给大家整理的是一篇关于python中使用np.delete()的实例方法,对此有兴趣的朋友们可以学习参考下。...2021-02-01
- 这篇文章主要介绍了使用Python的pencolor函数实现渐变色功能,本文通过实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下...2021-03-09
- 这篇文章主要介绍了python自动化办公操作PPT的实现,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...2021-02-05
Python getsizeof()和getsize()区分详解
这篇文章主要介绍了Python getsizeof()和getsize()区分详解,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧...2020-11-20- 这篇文章主要介绍了解决python 两个时间戳相减出现结果错误的问题,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧...2021-03-12
- 这篇文章主要为大家详细介绍了python实现学生通讯录管理系统,文中示例代码介绍的非常详细,具有一定的参考价值,感兴趣的小伙伴们可以参考一下...2021-02-25
- 这篇文章主要介绍了PyTorch一小时掌握之迁移学习篇,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下...2021-09-08
- 这篇文章主要介绍了Python绘制的爱心树与表白代码,本文通过实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下...2021-04-06