python 阿里云oss实现直传签名与回调验证的示例方法
作者:weixin_54126636 发布时间:2021-12-08 00:30:18
标签:python,直传签名,回调验证
签名
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()
来源:https://blog.csdn.net/weixin_54126636/article/details/113731760
0
投稿
猜你喜欢
- python3.6.4安装opencv3.4.2使用pip安装OpenCV直接安装最新版:pip3 install opencv_pytho
- 文档介绍利用python写“猜数字”,“猜词语”,“谁是卧底”这三个游戏,从而快速掌握python编程的入门知识,包括python语法/列表
- 在昨天的文章,《 block 和 inline 的区别是?》里,我给大家留了个问题——LI 元素到底是block level 的,还是 in
- 做 Web 开发少不了要与模板引擎打交道。我陆续也接触了 Python 的不少模板引擎,感觉可以总结一下了。一、首先按照我的熟悉程度列一下:
- 一、说明早上看到Python使用pickle进行序列化和反序列化,然后发现面临的一个获取不到返回值的框架,似乎可以通过在框架中先序列化,然后
- pycharm永久激活码6ZUMD7WWWU-eyJsaWNlbnNlSWQiOiI2WlVNRDdXV1dVIiwibGljZW5zZWV
- 随着jQuery、Mootools、prototype等知名的JavaScript框架的应用变的越来越强大,浏览器对最新版本CSS属性的支持
- 一、引言有一定 Python 编程经验的人估计十有八九使用过异常,异常对于程序的健壮性是毋庸置疑的。二、使用异常对数据进行初始化在某些条件下
- 一、Python简介Python是一种面向对象的解释型计算机程序设计语言。Python是纯粹的自由软件, 源代码和解释器CPython遵循
- 客户用的数据库是mysql,而研发好的产品支持oracle,为了让客户掏腰包,我们必须把数据库环境从oracle转向mysql。我们在转换的
- 本文实例讲述了python实现string和dict的相互转换方法。分享给大家供大家参考,具体如下:字典(dict)转为字符串(string
- 在这个自动化时代,我们有很多重复无聊的工作要做。 想想这些你不再需要一次又一次地做的无聊的事情,让它自动化,让你的生活更轻松。 那么在本文中
- 近日无事,想起以前曾打算过要做一个定时重启或关机的工具,便花了一点时间以hta的形式写了个,名为"Windows Timer&qu
- 本文实例讲述了python获取一组数据里最大值max函数用法。分享给大家供大家参考。具体如下:# 最简单的max(1, 2)max('
- 本文实例为大家分享了python实现五子棋游戏的具体代码,供大家参考,具体内容如下checkerboard.pyfrom collectio
- 前言我们在上一期学习了关于Python 迭代器Iterator详情相关的概念,满足迭代器需要符合两个条件实现__iter__()方
- 1、把这段拷到DW里,存成HTML文件。<HTML><HEAD><TITLE>move backgrou
- 描述微软开发的两个动态库存在后门允许用户查看ASP文件源程序和下载整个网站详细随IIS和Frontpage Extention server
- 1 实验环境(1)服务端:本实验基于虚拟机win2008系统的WAMP环境进行,该环境相关配置过程参考文章《【语言环境】WAMP环境部署及优
- 一、查询操作可以使用Dataframe的index属性和columns属性获取行、列索引。import pandas as pddata =