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 직전 서명과 리셋 검증 내용은 저희 이전의 글을 검색하거나 아래의 관련 글을 계속 훑어보십시오. 앞으로 많은 응원 부탁드립니다!

좋은 웹페이지 즐겨찾기