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) + '\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 直传签名与回调验证内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们!

时间: 2021-03-27

python3个性签名设计实现代码

本文实例为大家分享了python个性签名设计的具体代码,供大家参考,具体内容如下 参考博客:Python GUI Tkinter简单实现个性签名设计 参考博客:python3爬虫之设计签名小程序 Code from tkinter import * from tkinter import messagebox import requests import re from PIL import Image,ImageTk #模拟浏览器发送请求 def download(): startUrl =

python3爬虫之设计签名小程序

本文实例为大家分享了python3设计签名小程序的具体代码,供大家参考,具体内容如下 首先,上一下要做的效果图: 先是这样一个丑陋的界面(我尽力了的真的!) 然后随便输入名字 然后点击按钮会显示出对应的个性签名: 这个是怎么实现的呢? 其实这个是将一个签名网站http://www.uustv.com/的内容爬下来显示了而已: 源代码如下: from tkinter import * import requests from tkinter import messagebox import re

PYTHON实现SIGN签名的过程解析

sign签名是用于提供给外部(第三方)调用的接口,调用方需要提供正确的appkey钥匙才能调用,确保了接口的安全性. 签名参数sign生成方法: 假设有请求参数如下: appkey = "111222333" body = { "username": "Test", "Password": "123456", "mail": "", "sign":

python rsa实现数据加密和解密、签名加密和验签功能

本篇文章主要说明python库rsa生成密钥对,数据的加密解密,api接口的签名和验签,如有抄袭,请留言联系我. 先安装 pip install rsa 安装好后,请看代码 """ 注意: api签名时:签名用私钥,验签用公钥 数据加密时:加密用公钥,解密用私钥 密钥和公钥保存的图片会在下边展示出来 """ import rsa # 对api的签名机制进行验证:签名用私钥,验签用公钥 class RsaEncrypt: def __init__(

Python生成个性签名图片获取GUI过程解析

这篇文章主要介绍了Python生成个性签名图片获取GUI过程解析,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 先来看看程序运行的样子: 所以,程序的原理是 从一个url = ' http://www.uustv.com/ '中爬取个性签名的图片. 在该网页中审查该图片的元素,然后找到该图片的imgur 使用正则表达式直接定位 程序的后面使用到了tkinter的 GUI简易界面,用于和用户的简单交互,非常方便. 整个程序的完整代码如下: (有任

Python GUI Tkinter简单实现个性签名设计

一.Tkinter的介绍和简单教程 Tkinter 是 Python 的标准 GUI 库.Python 使用 Tkinter 可以快速的创建 GUI 应用程序. 由于 Tkinter 是内置到 python 的安装包中.只要安装好 Python 之后就能 import Tkinter 库.而且 IDLE 也是用 Tkinter 编写而成.对于简单的图形界面 Tkinter 还是能应付自如. 注意:Python3.x 版本使用的库名为 tkinter,即首写字母 T 为小写. import tki

对python函数签名的方法详解

函数签名对象,表示调用函数的方式,即定义了函数的输入和输出. 在Python中,可以使用标准库inspect的一些方法或类,来操作或创建函数签名. 获取函数签名及参数 使用标准库的signature方法,获取函数签名对象:通过函数签名的parameters属性,获取函数参数. # 注意是小写的signature from inspect import signature def foo(value): return value # 获取函数签名 foo_sig = signature(foo)

Python中格式化format()方法详解

 Python中格式化format()方法详解 Python中格式化输出字符串使用format()函数, 字符串即类, 可以使用方法; Python是完全面向对象的语言, 任何东西都是对象; 字符串的参数使用{NUM}进行表示,0, 表示第一个参数,1, 表示第二个参数, 以后顺次递加; 使用":", 指定代表元素需要的操作, 如":.3"小数点三位, ":8"占8个字符空间等; 还可以添加特定的字母, 如: 'b' - 二进制. 将数字以2为基

Python sep参数使用方法详解

这篇文章主要介绍了Python sep参数使用方法详解,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 Python中sep不是函数,它是print函数的一个参数,用来定义输出数据之间的间隔符号. 具体用法如下: 同时打印多个字符串的时候,每个字符串都会自动默认以空格作为每个字符串之间的间隔. print("abc", "uuu", "oop") # abc uuu oop 如果在多个字符串的最后

shell 使用数组作为函数参数的方法(详解)

因工作需要,需要使用shell开发一些小工具,当使用数组作为函数参数时,发现只能传递数组的第一个元素,数组后面的元素不能传递到函数内. #!/bin/bash function showArr(){ arr=$1 for i in ${arr[*]}; do echo $i done } regions=("GZ" "SH" "BJ") showArr $regions exit 0 把代码保存为test.sh后执行,只输出了第一个元素. ./t

python生成式的send()方法(详解)

随便在网上找了找,感觉都是讲半天讲不清楚,这里写一下. def generator(): while True: receive=yield 1 print('extra'+str(receive)) g=generator() print(next(g)) print(g.send(111)) print(next(g)) 输出: 1 extra111 1 extraNone 1 为什么会这样呢,点进send就能看到一句话 send:Resumes the generator and "sen

inux下gettimeofday函数windows替换方法(详解)

实例如下: #include <time.h> #ifdef WIN32 # include <windows.h> #else # include <sys/time.h> #endif #ifdef WIN32 int gettimeofday(struct timeval *tp, void *tzp) { time_t clock; struct tm tm; SYSTEMTIME wtm; GetLocalTime(&wtm); tm.tm_year

vue iview组件表格 render函数的使用方法详解

如果要在标签中加入属性,例如img 中src属性   a标签中href属性  此时需要用到 attrs 来加入而不是props { title: '操作', key: 'action', align: 'center', render: function (h, params) { return h('div', [ h('Button', { props: { type: 'primary', size: 'small' }, style: { marginRight: '8px' }, on

把JSON数据格式转换为Python的类对象方法详解(两种方法)

JOSN字符串转换为自定义类实例对象 有时候我们有这种需求就是把一个JSON字符串转换为一个具体的Python类的实例,比如你接收到这样一个JSON字符串如下: {"Name": "Tom", "Sex": "Male", "BloodType": "A", "Hobbies": ["篮球", "足球"]} 我需要把这个转换为具

对python 自定义协议的方法详解

前面说到最近在写python的一些东西,然后和另外一位小伙伴定义了协议,然后昨天我有一部分东西没理解对,昨天上午我自己重写了一遍接收和发送的全部逻辑,昨天下午补了压力测试的脚本,自测没问题之后告知联调的小伙伴. 结果上午还是出了一点问题,然后我们两对代码,他写了一个python的实现.还好最后我这边没问题.(我也害怕是我这边出问题啊,所以我自己的代码都自己检查了好几遍) 简单放一下他的实现: import struct import ctypes class E(Exception): def

Python底层封装实现方法详解

这篇文章主要介绍了Python底层封装实现方法详解,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 事实上,python封装特性的实现纯属"投机取巧",之所以类对象无法直接调用私有方法和属性,是因为底层实现时,python偷偷改变了它们的名称. python在底层实现时,将它们的名称都偷偷改成了"_类名__属性(方法)名"的格式 class Person: def setname(self, name): if le