1 前端直传

在客户端通过JavaScript代码完成签名，无需过多配置，即可实现直传，非常方便。
但是
客户端通过JavaScript把AccesssKey ID和AccessKey Secret写在代码里面有泄露的风险，
强烈建议使用服务端签名后直传或者STS临时授权访问OSS。

2 后端上传

(1) 原理

image.png
和数据直传到OSS相比，以上方法存在以下缺点：

  • 上传慢：用户数据需先上传到应用服务器，之后再上传到OSS，网络传输时间比直传到OSS多一倍。如果用户数据不通过应用服务器中转，而是直传到OSS，速度将大大提升。而且OSS采用BGP带宽，能保证各地各运营商之间的传输速度。
  • 扩展性差：如果后续用户数量逐渐增加，则应用服务器会成为瓶颈。
  • 费用高：需要准备多台应用服务器。由于OSS上行流量是免费的，如果数据直传到OSS，将节省多台应用服务器的费用。

    (2) 代码

    pip install oss2

  1. import oss2
  3. OSS_BASE_PATH = "https://cncms-server.oss-cn-beijing.aliyuncs.com/"
  4. end_point = 'https://oss-cn-beijing.aliyuncs.com'
  5. bucket_name = 'cncms-server'
  7. class OssFileStorage():
  8.     def __init__(self, root_store_path: str):
  9.         self.auth = oss2.Auth(access_key_id, access_key_secret)
  10.         self.bucket = oss2.Bucket(self.auth, end_point, bucket_name)
  11.         self.root_store_path = root_store_path  # 根存储路径
  13.     def file_upload(self, pk: str, category: str, file_name: str, file_content: str) -> dict:
  14.         '''文件上传'''
  15.         # 拼接文件路径
  16.         object_path = self.root_store_path + "/" + pk + "/" + category + "/" + file_name
  17.         # 上传文件
  18.         self.bucket.put_object(object_path, file_content)
  19.         # 转义
  20.         quote_file_name = urllib.parse.quote(file_name)
  21.         object_path = self.root_store_path + "/" + pk + "/" + category + "/" + quote_file_name
  23.         file_url = OSS_BASE_PATH + object_path
  24.         result = {
  25.             "object_path": object_path,
  26.             "file_url": file_url
  27.         }
  28.         return result
  30.     def delete_file(self, file_url: str):
  31.         '''删除文件'''
  32.         object_path = file_url.replace(OSS_BASE_PATH, "")
  33.         self.bucket.delete_object(object_path)
  35.     def is_file_exist(self, filename):
  36.         ret = self.bucket.object_exists(filename)
  37.         return ret
  39.     def generate_download_url(self, url):
  40.         if not url:
  41.             return None
  42.         # 从数据库存储的完整路径中获取object_name
  43.         object_name = url.replace(OSS_BASE_PATH, "")
  44.         # 反转义, 避免路径中包含中文后不能加AccesskeyID
  45.         object_name = urllib.parse.unquote(object_name)
  47.         # 生成下载文件的签名URL，有效时间为1天。
  48.         # 生成签名URL时，OSS默认会对Object完整路径中的正斜线（/）进行转义，从而导致生成的签名URL无法直接使用。
  49.         # 设置slash_safe为True，OSS不会对Object完整路径中的正斜线（/）进行转义，此时生成的签名URL可以直接使用。
  50.         download_url = self.bucket.sign_url('GET', object_name, 24*60*60, slash_safe=True)
  51.         return download_url
  1. @bp_hard.route('/hardwares/export', methods=['GET'])
  2. @auth.login_required
  3. @auth.module_hard
  4. def export_hardware():
  5.     """导出硬件"""
  6.     xls_name = ['硬件']
  7.     fit_hards = filter_hards(request, xls_name)
  8.     # 生成excel文件
  9.     file_name = '_'.join(xls_name) + '.xlsx'
  10.     file_path = './temp/' + file_name
  11.     xlsx_export.export_xlsx_hards(fit_hards, file_path)
  12.     # 同步上传
  13.     file_content = open(file_path, 'rb').read()
  14.     ret = fs.file_upload('excel', 'hard', file_name, file_content)
  15.     file_url = ret['file_url']
  16.     result = {'fileUrl': fs.generate_download_url(file_url)}
  17.     return jsonify(errno=RET.OK, errmsg='导出硬件数据成功', result=result)

3 服务端签名后, 前端上传

(1) 原理

image.png
文档:
https://help.aliyun.com/document_detail/91848.html
image.png
第一个GET请求的响应

  1. {
  2.     "accessid": "LTAI5tA7ijeV6BCwdeHWy5hQ", 
  3.     "host": "http://wstest1.oss-cn-beijing.aliyuncs.com", 
  4.     "policy": "eyJleHBpcmF0aW9uIjogIjIwMjItMDEtMDVUMDU6MjM6NTlaIiwgImNvbmRpdGlvbnMiOiBbWyJzdGFydHMtd2l0aCIsICIka2V5IiwgInVzZXItZGlyLXByZWZpeC8iXV19", 
  5.     "signature": "dzqZQaklHQcA6K4Rk+6pZ5/RP7M=", 
  6.     "expire": 1641360239, 
  7.     "dir": "user-dir-prefix/", 
  8.     "callback": "eyJjYWxsYmFja1VybCI6ICJodHRwOi8vNDcuMTA4LjE3MC4xOTU6NDcxMjMiLCAiY2FsbGJhY2tCb2R5IjogImZpbGVuYW1lPSR7b2JqZWN0fSZzaXplPSR7c2l6ZX0mbWltZVR5cGU9JHttaW1lVHlwZX0maGVpZ2h0PSR7aW1hZ2VJbmZvLmhlaWdodH0md2lkdGg9JHtpbWFnZUluZm8ud2lkdGh9IiwgImNhbGxiYWNrQm9keVR5cGUiOiAiYXBwbGljYXRpb24veC13d3ctZm9ybS11cmxlbmNvZGVkIn0="
  9. }

(2) 代码

1) appserver.py

  1. # -*- coding: UTF-8 -*-
  3. import socket
  4. import base64
  5. import sys
  6. import time
  7. import datetime
  8. import json
  9. import hmac
  10. from hashlib import sha1 as sha
  12. import httpserver
  14. # 请填写您的AccessKeyId。
  15. access_key_id = 'LTAI5tA7ijeV6BCwdeHWy5hQ'
  16. # 请填写您的AccessKeySecret。
  17. access_key_secret = 'bDikXRrncXK1A9JCMrwV1c89WN2ySr'
  18. # host的格式为 bucketname.endpoint ，请替换为您的真实信息。
  19. host = 'http://wstest1.oss-cn-beijing.aliyuncs.com'
  20. # callback_url为 上传回调服务器的URL，请将下面的IP和Port配置为您自己的真实信息。
  21. callback_url = "http://192.168.1.103:47123"
  22. # 用户上传文件时指定的前缀。
  23. upload_dir = 'user-dir-prefix/'
  24. expire_time = 3000000
  27. def get_iso_8601(expire):
  28.     gmt = datetime.datetime.utcfromtimestamp(expire).isoformat()
  29.     gmt += 'Z'
  30.     return gmt
  33. def get_token():
  34.     now = int(time.time())
  35.     expire_syncpoint = now + expire_time
  36.     expire = get_iso_8601(expire_syncpoint)
  38.     policy_dict = {}
  39.     policy_dict['expiration'] = expire
  40.     condition_array = []
  41.     array_item = []
  42.     array_item.append('starts-with')
  43.     array_item.append('$key')
  44.     array_item.append(upload_dir)
  45.     condition_array.append(array_item)
  46.     policy_dict['conditions'] = condition_array
  47.     policy = json.dumps(policy_dict).strip()
  48.     policy_encode = base64.b64encode(policy.encode())
  49.     h = hmac.new(access_key_secret.encode(), policy_encode, sha)
  50.     sign_result = base64.encodestring(h.digest()).strip()
  52.     callback_dict = {}
  53.     callback_dict['callbackUrl'] = callback_url
  54.     callback_dict['callbackBody'] = 'filename=${object}&size=${size}&mimeType=${mimeType}' \
  55.                                     '&height=${imageInfo.height}&width=${imageInfo.width}'
  56.     callback_dict['callbackBodyType'] = 'application/x-www-form-urlencoded'
  57.     callback_param = json.dumps(callback_dict).strip()
  58.     base64_callback_body = base64.b64encode(callback_param.encode())
  60.     token_dict = {}
  61.     token_dict['accessid'] = access_key_id
  62.     token_dict['host'] = host
  63.     token_dict['policy'] = policy_encode.decode()
  64.     token_dict['signature'] = sign_result.decode()
  65.     token_dict['expire'] = expire_syncpoint
  66.     token_dict['dir'] = upload_dir
  67.     token_dict['callback'] = base64_callback_body.decode()
  68.     result = json.dumps(token_dict)
  69.     return result
  72. def get_local_ip():
  73.     """
  74.     获取本机 IPV4 地址
  75.     :return: 成功返回本机 IP 地址，否则返回空
  76.     """
  77.     try:
  78.         csocket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
  79.         csocket.connect(('8.8.8.8', 80))
  80.         (addr, port) = csocket.getsockname()
  81.         csocket.close()
  82.         return addr
  83.     except socket.error:
  84.         return ""
  87. def do_POST(server):
  88.     """
  89.     启用 POST 调用处理逻辑
  90.     :param server: Web HTTP Server 服务
  91.     :return:
  92.     """
  93.     print("********************* do_POST ")
  94.     # get public key
  95.     pub_key_url = ''
  97.     try:
  98.         pub_key_url_base64 = server.headers['x-oss-pub-key-url']
  99.         pub_key = httpserver.get_pub_key(pub_key_url_base64)
  100.     except Exception as e:
  101.         print(str(e))
  102.         print('Get pub key failed! pub_key_url : ' + pub_key_url)
  103.         server.send_response(400)
  104.         server.end_headers()
  105.         return
  107.     # get authorization
  108.     authorization_base64 = server.headers['authorization']
  110.     # get callback body
  111.     content_length = server.headers['content-length']
  112.     callback_body = server.rfile.read(int(content_length))
  114.     # compose authorization string
  115.     auth_str = ''
  116.     pos = server.path.find('?')
  117.     if -1 == pos:
  118.         auth_str = server.path + '\n' + callback_body.decode()
  119.     else:
  120.         auth_str = httpserver.get_http_request_unquote(server.path[0:pos]) + server.path[pos:] + '\n' + callback_body
  122.     result = httpserver.verrify(auth_str, authorization_base64, pub_key)
  124.     if not result:
  125.         print('Authorization verify failed!')
  126.         print('Public key : %s' % (pub_key))
  127.         print('Auth string : %s' % (auth_str))
  128.         server.send_response(400)
  129.         server.end_headers()
  130.         return
  132.     # response to OSS
  133.     resp_body = '{"Status":"OK"}'
  134.     server.send_response(200)
  135.     server.send_header('Content-Type', 'application/json')
  136.     server.send_header('Content-Length', str(len(resp_body)))
  137.     server.end_headers()
  138.     server.wfile.write(resp_body.encode())
  141. def do_GET(server):
  142.     """
  143.     启用 Get 调用处理逻辑
  144.     :param server: Web HTTP Server 服务
  145.     :return:
  146.     """
  147.     print("********************* do_GET ")
  148.     token = get_token()
  149.     server.send_response(200)
  150.     server.send_header('Access-Control-Allow-Methods', 'POST')
  151.     server.send_header('Access-Control-Allow-Origin', '*')
  152.     server.send_header('Content-Type', 'text/html; charset=UTF-8')
  153.     server.end_headers()
  154.     server.wfile.write(token.encode())
  157. if '__main__' == __name__:
  158.     # 在服务器中, 0.0.0.0指的是本机上的所有IPV4地址, 如果一个主机有两个IP地址,
  159.     # 192.168.1.1 和 10.1.2.1, 并且该主机上的一个服务监听的地址是0.0.0.0, 那么通过两个IP地址都能够访问该服务。
  160.     # server_ip = get_local_ip() 若用户希望监听本机外网IPV4地址，则采用本行代码并注释掉下一行代码
  161.     server_ip = "0.0.0.0"
  162.     server_port = 47123
  163.     if len(sys.argv) == 2:
  164.         server_port = int(sys.argv[1])
  165.     if len(sys.argv) == 3:
  166.         server_ip = sys.argv[1]
  167.         server_port = int(sys.argv[2])
  168.     print("App server is running on http://%s:%s " % (server_ip, server_port))
  170.     server = httpserver.MyHTTPServer(server_ip, server_port)
  171.     server.serve_forever()

2) httpserver.py

  1. # -*- coding: utf-8 -*-
  3. """
  4. 兼容Python 2.X 与 3.X 版本
  5. """
  7. import appserver
  9. import base64
  10. import urllib.request
  11. from http.server import HTTPServer, BaseHTTPRequestHandler
  12. from Crypto.Signature import PKCS1_v1_5
  13. from Crypto.Hash import MD5
  14. from Crypto.PublicKey import RSA
  17. def verrify(auth_str, authorization_base64, pub_key):
  18.     """
  19.     校验签名是否正确（MD5 + RAS）
  20.     :param auth_str: 文本信息
  21.     :param authorization_base64: 签名信息
  22.     :param pub_key: 公钥
  23.     :return: 若签名验证正确返回 True 否则返回 False
  24.     """
  25.     pub_key_load = RSA.importKey(pub_key)
  26.     auth_md5 = MD5.new(auth_str.encode())
  27.     result = False
  28.     try:
  29.         result = PKCS1_v1_5.new(pub_key_load).verify(auth_md5, base64.b64decode(authorization_base64.encode()))
  30.     except Exception as e:
  31.         print(e)
  32.         result = False
  33.     return result
  35. def get_http_request_unquote(url):
  36.     return urllib.request.unquote(url)
  38. def get_pub_key(pub_key_url_base64):
  39.     """ 抽取出 public key 公钥 """
  40.     pub_key_url = base64.b64decode(pub_key_url_base64.encode())
  41.     url_reader = urllib.request.urlopen(pub_key_url.decode())
  42.     pub_key = url_reader.read()
  43.     return pub_key
  45. class MyHTTPRequestHandler(BaseHTTPRequestHandler):
  46.     def do_POST(self):
  47.         """处理 POST 请求"""
  48.         return appserver.do_POST(self)
  50.     def do_GET(self):
  51.         """处理 GET 请求"""
  52.         appserver.do_GET(self)
  54. class MyHTTPServer(HTTPServer):
  55.     def __init__(self, host, port):
  56.         print("run app server by python3!")
  57.         HTTPServer.__init__(self,  (host, port), MyHTTPRequestHandler)