python 아리운oss가 직전 서명과 리셋 검증을 실현하는 예시 방법
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) + '
' + 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 직전 서명과 리셋 검증 내용은 저희 이전의 글을 검색하거나 아래의 관련 글을 계속 훑어보십시오. 앞으로 많은 응원 부탁드립니다!
이 내용에 흥미가 있습니까?
현재 기사가 여러분의 문제를 해결하지 못하는 경우 AI 엔진은 머신러닝 분석(스마트 모델이 방금 만들어져 부정확한 경우가 있을 수 있음)을 통해 가장 유사한 기사를 추천합니다:
로마 숫자를 정수로 또는 그 반대로 변환그 중 하나는 로마 숫자를 정수로 변환하는 함수를 만드는 것이었고 두 번째는 그 반대를 수행하는 함수를 만드는 것이었습니다. 문자만 포함합니다'I', 'V', 'X', 'L', 'C', 'D', 'M' ; 문자열이 ...
텍스트를 자유롭게 공유하거나 복사할 수 있습니다.하지만 이 문서의 URL은 참조 URL로 남겨 두십시오.
CC BY-SA 2.5, CC BY-SA 3.0 및 CC BY-SA 4.0에 따라 라이센스가 부여됩니다.