---

前言

小程序业务，涉及到上传图片的功能，刚开始使用的是腾讯低代码平台，发现上传很方便，无需关注逻辑，但是有很多潜在的问题，无法定制开发。之前使用过阿里云的OSS，感觉还可以，就打算直接上手，在微信小程序环境内上传文件到阿里云存储。

本次是通过服务器生成签名，客户端拿着签名直接进行上传操作，减少客户端操作

---

一、微信小程序上传方法

微信小程序文件上传 官方文档。由于小程序内部环境，暂时无法使用其他方式进行上传。

是通过调用wx.uploadFile(Object object)方法进行上传，示例如下：

wx.chooseImage({
  success (res) {
    const tempFilePaths = res.tempFilePaths
    wx.uploadFile({
      url: 'https://example.weixin.qq.com/upload', //仅为示例，非真实的接口地址
      filePath: tempFilePaths[0],
      name: 'file',
      formData: {
        'user': 'test'
      },
      success (res){
        const data = res.data
        //do something
      }
    })
  }
})

二、阿里云OSS

阿里云官方 微信小程序实践 官方文档

有了文档，那么，接下来就可以根据文档进行定制操作

1.配置跨域访问（参考文档）

2.获取上传签名（重点）

获取签名有多种方式，因为我使用的是Python环境，本次就通过Python来实现如何在服务器获取上传签名

服务器签名准备所需要的数据

bucket_name_endpoint	Bucket域名
callback_url			上传完成之后，阿里云会回调该地址
access_key_id			访问授权key
access_key_secret		访问授权secret

access_key_id和access_key_secret 为了安全，建议使用RAM授权特定存储权限

RAM权限策略如下：

{
    "Statement": [
        {
            "Action": "oss:*",
            "Effect": "Allow",
            "Resource": [
                "acs:oss:*:*:devapp-storage",
                "acs:oss:*:*:devapp-storage/*"
            ]
        }
    ],
    "Version": "1"
}

devapp-storage 是本人的存储名称，这个名称要更换为自己的存储名

分析下具体流程

1. 小程序请求服务器，获取上传token
2. 小程序根据token，进行上传文件
3. 上传成功之后，如果配置回调，则阿里云回调服务器
4. 进行数据库的存储以及资源展示

服务器代码如下：

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# filename: alioss
# date: 2022/5/26

import datetime
import hmac
import json
import logging
import time
import urllib.request
from hashlib import sha1 as sha
import base64
import urllib.request
from Crypto.Signature import PKCS1_v1_5
from Crypto.Hash import MD5
from Crypto.PublicKey import RSA

logger = logging.getLogger(__name__)


def verify_auth(auth_str, authorization_base64, pub_key):
    """
    校验签名是否正确（MD5 + RAS）
    :param auth_str: 文本信息
    :param authorization_base64: 签名信息
    :param pub_key: 公钥
    :return: 若签名验证正确返回 True 否则返回 False
    """
    pub_key_load = RSA.importKey(pub_key)
    auth_md5 = MD5.new(auth_str.encode())
    result = False
    try:
        result = PKCS1_v1_5.new(pub_key_load).verify(auth_md5, base64.b64decode(authorization_base64.encode()))
    except Exception as e:
        logger.error(f"oss callback authorization verify failed! Exception:{e}")
    return result


def get_iso_8601(expire):
    gmt = datetime.datetime.utcfromtimestamp(expire).isoformat()
    gmt += 'Z'
    return gmt


def get_pub_key(pub_key_url_base64):
    """ 抽取出 public key 公钥 """
    pub_key_url = base64.b64decode(pub_key_url_base64.encode()).decode()
    if pub_key_url.find("://gosspublic.alicdn.com/") == -1:
        raise Exception('public key error')
    ## 可配置缓存，将公钥缓存下来
    url_reader = urllib.request.urlopen(pub_key_url)
    pub_key = url_reader.read()
    return pub_key


class AliOss(object):

    def __init__(self, access_key_id, access_key_secret, bucket_name_endpoint, callback_url):
        """
        :param access_key_id:
        :param access_key_secret:
        :param bucket_name_endpoint:    Bucket 域名
        :param callback_url:    回调地址
        """
        self.bucket_name_endpoint = bucket_name_endpoint
        self.callback_url = callback_url
        self.access_key_id = access_key_id
        self.access_key_secret = access_key_secret

    def make_token(self, upload_dir, expire_time=300, content_max_length=10 * 1024 * 1024):
        """
        :param upload_dir:   上传目录
        :param expire_time:     token生效时间
        :param content_max_length:  最大上传文件大小
        :return:
        """
        expire_time = int(time.time()) + expire_time
        policy_dict = {
            'expiration': get_iso_8601(expire_time),
            'conditions': [
                ['starts-with', '$key', upload_dir],
                ["content-length-range", 0, content_max_length]
            ]
        }
        policy = json.dumps(policy_dict).strip()
        policy_encode = base64.b64encode(policy.encode())
        h = hmac.new(self.access_key_secret.encode(), policy_encode, sha)
        sign_result = base64.encodebytes(h.digest()).strip()

        callback_dict = {
            'callbackUrl': self.callback_url,
            'callbackBody': 'filename=${object}&size=${size}&mimeType=${mimeType}',
            'callbackBodyType': 'application/x-www-form-urlencoded'
        }
        callback_param = json.dumps(callback_dict).strip()
        base64_callback_body = base64.b64encode(callback_param.encode())

        token_dict = {
            'access_key_id': self.access_key_id, 'host': self.bucket_name_endpoint,
            'policy': policy_encode.decode(),
            'signature': sign_result.decode(), 'expire': expire_time,
            'callback': base64_callback_body.decode()
        }
        return token_dict

    @staticmethod
    def callback_verify(request_headers, request_body):

        pub_key_url = ''

        try:
            pub_key_url_base64 = request_headers['HTTP_X_OSS_PUB_KEY_URL']
            pub_key = get_pub_key(pub_key_url_base64)
        except Exception as e:
            logger.error(f"Get pub key failed! pub_key_url {pub_key_url} Exception:{e}")
            return {"status": 400, "msg": "Get pub key failed!"}

        # get authorization
        authorization_base64 = request_headers['HTTP_AUTHORIZATION']

        # get callback body
        content_length = int(request_headers['CONTENT_LENGTH'])
        callback_body = request_body[:content_length]

        query_string = request_headers['QUERY_STRING']
        path_info = request_headers['PATH_INFO']
        query_string = '?' + query_string if query_string else ''
        auth_str = path_info + query_string + '\n' + callback_body.decode()
        result = verify_auth(auth_str, authorization_base64, pub_key)

        if not result:
            return {"status": 400, "msg": "authorization verify failed!"}

        return {"status": 200, "Status": "OK"}

客户端请求token方法，客户端通过post请求，携带文件名参数

class Upload2View(APIView):


    def post(self, request):
        """
        获取上传的token
        :param request:
        :return:
        """
        filename = request.data.get('filename', '')
        f_type = filename.split(".")[-1]
        if not f_type or (f_type and f_type not in settings.FILE_UPLOAD_ALLOW):
            return ApiResponse(code=1001, msg='上传类型错误')

        upload_dir = f"{request.user.pk}/"
        random_file = make_from_user_uuid(request.user.username)
        upload_key = f"{upload_dir}{random_file}.{f_type}{settings.FILE_UPLOAD_TMP_KEY}"
       

        access_key_id = "L******************"			# 填写自己的
        access_key_secret = "e2a5m1**************"      # 填写自己的
        bucket_name_endpoint = "devapp-storage.************ncs.com"    # 填写自己的
        callback_url = f"https://*********/api/v1/server/callback"     # 填写自己的

        alioss = AliOss(access_key_id, access_key_secret, bucket_name_endpoint, callback_url)
        upload_token = alioss.make_token(upload_dir)

        data = {
            "upload_token": upload_token,
            "upload_key": upload_key
        }

        return ApiResponse(data=data)

阿里云回调方法,阿里云是通过post进行回调

 class AliOSSCallBackView(APIView):
    authentication_classes = []

    def post(self, request):
        logging.error(request.META)
        access_token = request.query_params.get('access_token')
        result = AliOss.callback_verify(request.META, request.body)
        response = ApiResponse(**result)
        if response.status_code == 200:
            upload_key = request.data.get('filename')
            size = request.data.get('size')
            PictureInfo.objects.create(pic_uid_name=upload_key, pic_size=size)
			# 可以进行一些操作，比如数据入库
        return response

三、微信小程序封装上传方法

根据服务器生成的签名数据，通过下面方法就行封装，操作测试，记得把域名校验关闭

function uploadFile(filePath, upload_token, upload_key, success, fail) {
  if (!filePath) {
    fail && fail();
    return;
  }

  const { access_key_id, policy, signature, host, callback } = upload_token


  //小程序直传oss
  wx.uploadFile({
    url: `https://${host}`,
    filePath: filePath,
    name: 'file', //必须填file
    header: {
      "Content-Type": "multipart/form-data"
    },
    formData: {
      'key': upload_key,
      'policy': policy,
      'OSSAccessKeyId': access_key_id,
      'signature': signature,
      'callback': callback,
      // 'x-oss-security-token': security_token, 
      'success_action_status': '200'
    },
    success: function (res) {
      if (res.statusCode != 200) {
        fail && fail(new Error('上传错误:' + JSON.stringify(res)))
        return;
      }
      success(res);
    },
    fail: function (err) {
      fail && fail(err);
    },
  })

}

module.exports = uploadFile;

---

测试

小程序上传

服务器可以看到

---

总结

本次介绍就到此结束了，希望刚入门的小伙伴能有收获。

相关服务器源码 点击git

相关客户端源码 点击git

源码包含 阿里云oss存储的两种上传方式，sts授权和URL签名上传。STS授权比较繁琐，并且sts请求有并发限制，建议小伙伴使用URL签名直传。