av一区二区在线观看_亚洲男人的天堂网站_日韩亚洲视频_在线成人免费_欧美日韩精品免费观看视频_久草视

您的位置:首頁技術文章
文章詳情頁

python 阿里云oss實現直傳簽名與回調驗證的示例方法

瀏覽:80日期:2022-06-23 18:32:48

簽名

import base64import jsonimport timefrom datetime import datetimeimport hmacfrom hashlib import sha1access_key_id = ’’# 請填寫您的AccessKeySecret。access_key_secret = ’’# host的格式為 bucketname.endpoint ,請替換為您的真實信息。host = ’’# callback_url為 上傳回調服務器的URL,請將下面的IP和Port配置為您自己的真實信息。callback_url = ''# 用戶上傳文件時指定的前綴。upload_dir = ’user-dir-prefix/’expire_time = 1200expire_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 asyncioimport base64import timeimport aiomysqlimport rsafrom aiohttp import web, ClientSessionfrom urllib import parseimport uuiddef 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 applicationif __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 編程
相關文章:
主站蜘蛛池模板: 亚洲一区二区三区免费在线观看 | 色综合99| 日韩视频成人 | 国产精品免费在线 | 91免费看片| 人人干人人看 | 久久久国产一区 | 久久综合888 | 久久黄色网 | 欧美日韩视频在线 | 国产欧美精品一区二区色综合朱莉 | 91精品国产91综合久久蜜臀 | 日韩视频免费 | 成人在线精品视频 | 国产精品久久久久久久久久久免费看 | 日韩免费 | 国产一区中文字幕 | 我想看一级黄色毛片 | 欧美日韩国产三级 | 国产精品久久视频 | 亚洲国产成人av好男人在线观看 | 免费观看一级黄色录像 | 免费观看一级特黄欧美大片 | 亚州激情| 国产在线一区观看 | 成人性视频在线播放 | 午夜影院在线观看视频 | 超碰成人av | 一级国产精品一级国产精品片 | 欧美一区二区三区免费在线观看 | 久久精品视频亚洲 | 天天澡天天狠天天天做 | 国产一区二区三区在线看 | 日本高清aⅴ毛片免费 | 久久久久国产精品一区 | 91影片 | 久久精品视频一区二区三区 | www.日韩欧美| 成人免费视频网站 | 涩涩视频在线观看 | 久久久国产一区二区 |