如何在微信小程序环境下将文件上传到OSS

配置依赖。

<dependency>
    <groupId>com.aliyun.oss</groupId>
    <artifactId>aliyun-sdk-oss</artifactId>
    <version>3.17.4</version>
</dependency>

配置环境变量。

• macOS/Linux/Unix系统。

  export OSS_ACCESS_KEY_ID=<ALIBABA_CLOUD_ACCESS_KEY_ID>
  export OSS_ACCESS_KEY_SECRET=<ALIBABA_CLOUD_ACCESS_KEY_SECRET>
  export OSS_STS_ROLE_ARN=<ROLE_ARN>

• Windows系统。

  set OSS_ACCESS_KEY_ID=<ALIBABA_CLOUD_ACCESS_KEY_ID>
  set OSS_ACCESS_KEY_SECRET=<ALIBABA_CLOUD_ACCESS_KEY_SECRET>
  set OSS_STS_ROLE_ARN=<ROLE_ARN>

API接口示例。

package com.example.demo.controller;

import com.example.demo.util.ECSGenerateSignature;
import com.example.demo.util.RAMGenerateSignature;
import com.fasterxml.jackson.core.JsonProcessingException;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class VxController {

    /**
     * 使用环境变量中的AK获取临时访问凭证，计算签名信息，返回小程序端。
     * @return
     * @throws JsonProcessingException
     */
    @GetMapping("/generate_signature")
    public String generate_signature() throws JsonProcessingException {
        RAMGenerateSignature ramGenerateSignature = new RAMGenerateSignature();
        return ramGenerateSignature.getSignature();
    }

}

签名信息工具类示例。

package com.example.demo.util;

import com.aliyun.oss.common.utils.BinaryUtil;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.commons.codec.binary.Base64;
import com.aliyuncs.DefaultAcsClient;
import com.aliyuncs.IAcsClient;
import com.aliyuncs.auth.sts.AssumeRoleRequest;
import com.aliyuncs.auth.sts.AssumeRoleResponse;
import com.aliyuncs.exceptions.ClientException;
import com.aliyuncs.profile.DefaultProfile;

import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import java.time.Instant;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;


/**
 * 使用环境变量中的AK获取临时访问凭证计算签名
 */
public class RAMGenerateSignature {

    public String getSignature() throws JsonProcessingException {

        //获取发送STS请求基础信息
        String accessKeyId = System.getenv("OSS_ACCESS_KEY_ID");    //环境变量中获取access_key_id
        String accessKeySecret = System.getenv("OSS_ACCESS_KEY_SECRET");    //环境变量中获取access_key_secret
        String roleArnForOssUpload = System.getenv("OSS_STS_ROLE_ARN");    //环境变量中获取ARN
        String regionId = "cn-hangzhou";    //发起STS请求所在的地域
        String roleSessionName = "<YOUR_ROLE_SESSION_NAME>";    //色会话名称，用来区分不同的令牌，可自定义
        Long durationSeconds = 3600L;   //临时访问凭证的有效时间

        //初始化客户端
        DefaultProfile profile = DefaultProfile.getProfile(regionId, accessKeyId, accessKeySecret);
        IAcsClient client = new DefaultAcsClient(profile);

        AssumeRoleRequest request = new AssumeRoleRequest();
        request.setRoleArn(roleArnForOssUpload);
        request.setRoleSessionName(roleSessionName);
        request.setDurationSeconds(durationSeconds);

        //定义STS临时访问凭证变量
        String STSaccessKeyId = null;
        String STSsecretAccessKey = null;
        String securityToken = null;

        try {
            AssumeRoleResponse response = client.getAcsResponse(request);

            //将请求返回的STS临时访问凭证赋值到自定义变量中
            STSaccessKeyId = response.getCredentials().getAccessKeyId();
            STSsecretAccessKey = response.getCredentials().getAccessKeySecret();
            securityToken = response.getCredentials().getSecurityToken();

        } catch (ClientException e) {
            e.printStackTrace();
        }

        //格式化请求日期
        long now = System.currentTimeMillis() / 1000;
        ZonedDateTime dtObj = ZonedDateTime.ofInstant(Instant.ofEpochSecond(now), ZoneId.of("UTC"));
        ZonedDateTime dtObjPlus3h = dtObj.plusHours(3);
        //请求时间
        DateTimeFormatter dtObj1Formatter = DateTimeFormatter.ofPattern("yyyyMMdd'T'HHmmss'Z'");
        String dtObj1 = dtObj.format(dtObj1Formatter);
        //请求日期
        DateTimeFormatter dtObj2Formatter = DateTimeFormatter.ofPattern("yyyyMMdd");
        String dtObj2 = dtObj.format(dtObj2Formatter);
        //请求过期时间
        DateTimeFormatter expirationTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
        String expirationTime = dtObjPlus3h.format(expirationTimeFormatter);

        // 创建policy
        ObjectMapper mapper = new ObjectMapper();

        Map<String, Object> policy = new HashMap<>();
        policy.put("expiration", expirationTime);

        List<Object> conditions = new ArrayList<>();

        Map<String, String> bucketCondition = new HashMap<>();
        bucketCondition.put("bucket", "bucketname");  //请替换为目标bucket名称
        conditions.add(bucketCondition);

        Map<String, String> signatureVersionCondition = new HashMap<>();
        signatureVersionCondition.put("x-oss-signature-version", "OSS4-HMAC-SHA256");
        conditions.add(signatureVersionCondition);

        Map<String, String> credentialCondition = new HashMap<>();
        credentialCondition.put("x-oss-credential",  STSaccessKeyId + "/" + dtObj2 + "/cn-hangzhou/oss/aliyun_v4_request"); // 替换为实际的 access key id
        conditions.add(credentialCondition);

        Map<String, String> token = new HashMap<>();
        token.put("x-oss-security-token", securityToken);
        conditions.add(token);

        Map<String, String> dateCondition = new HashMap<>();
        dateCondition.put("x-oss-date", dtObj1);
        conditions.add(dateCondition);

        policy.put("conditions", conditions);
        String jsonPolicy = mapper.writeValueAsString(policy);

        //构造待签名字符串（StringToSign）
        String stringToSign = new String(Base64.encodeBase64(jsonPolicy.getBytes()));

        //计算SigningKey
        byte[] dateKey = hmacsha256(("aliyun_v4" + STSsecretAccessKey).getBytes(), dtObj2);
        byte[] dateRegionKey = hmacsha256(dateKey, "cn-hangzhou");
        byte[] dateRegionServiceKey = hmacsha256(dateRegionKey, "oss");
        byte[] signingKey = hmacsha256(dateRegionServiceKey, "aliyun_v4_request");

        //计算Signature
        byte[] result = hmacsha256(signingKey, stringToSign);
        String signature = BinaryUtil.toHex(result);

        Map<String, String> messageMap = new HashMap<>();
        messageMap.put("security_token", securityToken);
        messageMap.put("signature", signature);
        messageMap.put("x_oss_date", dtObj1);
        messageMap.put("x_oss_credential", STSaccessKeyId + "/" + dtObj2 + "/cn-hangzhou/oss/aliyun_v4_request");
        messageMap.put("x_oss_signature_version", "OSS4-HMAC-SHA256");
        messageMap.put("policy", stringToSign);
        ObjectMapper objectMapper = new ObjectMapper();
        //打印返回至客户端的签名信息
//        System.out.println(objectMapper.writeValueAsString(messageMap));
        return objectMapper.writeValueAsString(messageMap);
    }

    /**
     * 使用HMAC-SHA256算法计算给定密钥和数据的哈希值的静态方法
     * @param key
     * @param data
     * @return
     */
    public static byte[] hmacsha256(byte[] key,String data){
        try {
            SecretKeySpec secretKeySpec = new SecretKeySpec(key, "HmacSHA256");
            Mac mac = Mac.getInstance("HmacSHA256");
            mac.init(secretKeySpec);
            byte[] hmacBytes = mac.doFinal(data.getBytes());
            return hmacBytes;
        }catch (Exception e){
            throw new RuntimeException("Failed to calculate HMAC-SHA256", e);
        }
    }
}

配置依赖。

pip install oss2  
pip install alibabacloud_tea_openapi alibabacloud_sts20150401 alibabacloud_credentials

配置环境变量。

• macOS/Linux/Unix

  export OSS_ACCESS_KEY_ID=<ALIBABA_CLOUD_ACCESS_KEY_ID>
  export OSS_ACCESS_KEY_SECRET=<ALIBABA_CLOUD_ACCESS_KEY_SECRET>
  export OSS_STS_ROLE_ARN=<ROLE_ARN>

• Windows

  set OSS_ACCESS_KEY_ID=<ALIBABA_CLOUD_ACCESS_KEY_ID>
  set OSS_ACCESS_KEY_SECRET=<ALIBABA_CLOUD_ACCESS_KEY_SECRET>
  set OSS_STS_ROLE_ARN=<ROLE_ARN>

代码示例。

from flask import Flask, jsonify
import base64
import hmac
import hashlib
import os
import datetime
import json
import time
from alibabacloud_tea_openapi.models import Config
from alibabacloud_sts20150401.client import Client as Sts20150401Client
from alibabacloud_sts20150401 import models as sts_20150401_models
from alibabacloud_credentials.client import Client as CredentialClient
import os

app = Flask(__name__)

def hmacsha256(key, data):
    """
    计算HMAC-SHA256哈希值的函数

    :param key: 用于计算哈希的密钥，字节类型
    :param data: 要进行哈希计算的数据，字符串类型
    :return: 计算得到的HMAC-SHA256哈希值，字节类型
    """
    try:
        mac = hmac.new(key, data.encode(), hashlib.sha256)
        hmacBytes = mac.digest()
        return hmacBytes
    except Exception as e:
        raise RuntimeError(f"Failed to calculate HMAC-SHA256 due to {e}")


@app.route('/generate_signature', methods=['GET'])
def generate_signature():
    """
    处理生成签名信息的请求，执行相关逻辑流程，包括获取环境变量、创建策略、构造待签名字符串、计算签名等操作，
    并返回生成的签名信息。
    :return: JSON格式的响应，包含签名信息的字典，格式如下：
    {
        "policy": "policy字符串",
        "x-oss-signature-version": "OSS4-HMAC-SHA256",
        "x-oss-credential": "accesskeyid/日期/cn-hangzhou/oss/aliyun_v4_request",
        "x-oss-date": "请求时间",
        "signature": "签名",
        "security_token": "安全令牌"
    }
    """
    access_key_id = os.environ.get('OSS_ACCESS_KEY_ID')
    access_key_secret = os.environ.get('OSS_ACCESS_KEY_SECRET')
    role_arn_for_oss_upload = os.environ.get('OSS_STS_ROLE_ARN')

    # 自定义会话名称
    role_session_name = 'role_session_name'  

    # 指定过期时间，单位为秒
    expire_time = 3600  

    bucket = 'examplebucket'
    region_id = 'cn-hangzhou'

    # 初始化配置，直接传递凭据
    config = Config(
        region_id=region_id,
        access_key_id=access_key_id,
        access_key_secret=access_key_secret
    )

    # 创建 STS 客户端并获取临时凭证
    sts_client = Sts20150401Client(config=config)
    assume_role_request = sts_20150401_models.AssumeRoleRequest(
        role_arn=role_arn_for_oss_upload,
        role_session_name=role_session_name
    )
    response = sts_client.assume_role(assume_role_request)
    token_data = response.body.credentials.to_map()

    # 使用 STS 返回的临时凭据
    sts_accesskeyid = token_data['AccessKeyId']
    sts_accesskeysecret = token_data['AccessKeySecret']
    security_token = token_data['SecurityToken']

    now = int(time.time())

    # 将时间戳转换为datetime对象
    dt_obj = datetime.datetime.utcfromtimestamp(now)
    # 在当前时间增加3小时，设置为请求的过期时间
    dt_obj_plus_3h = dt_obj + datetime.timedelta(hours=3)

    # 请求时间
    dt_obj_1 = dt_obj.strftime('%Y%m%dT%H%M%S') + 'Z'
    # 请求日期
    dt_obj_2 = dt_obj.strftime('%Y%m%d')
    # 请求过期时间
    expiration_time = dt_obj_plus_3h.strftime('%Y-%m-%dT%H:%M:%S.000Z')

    # 步骤1：创建policy。
    policy = {
        "expiration": expiration_time,
        "conditions": [
            {"bucket": "cjl3"},   #请替换为目标Bucket名称
            {"x-oss-signature-version": "OSS4-HMAC-SHA256"},
            {"x-oss-credential": f"{sts_accesskeyid}/{dt_obj_2}/cn-hangzhou/oss/aliyun_v4_request"},
            {"x-oss-security-token": security_token},
            {"x-oss-date": dt_obj_1},
        ]
    }
    policy_str = json.dumps(policy).strip()

    # 步骤2：构造待签名字符串（StringToSign）
    stringToSign = base64.b64encode(policy_str.encode()).decode()

    # 步骤3：计算SigningKey
    dateKey = hmacsha256(("aliyun_v4" + sts_accesskeysecret).encode(), dt_obj_2)
    dateRegionKey = hmacsha256(dateKey, "cn-hangzhou")
    dateRegionServiceKey = hmacsha256(dateRegionKey, "oss")
    signingKey = hmacsha256(dateRegionServiceKey, "aliyun_v4_request")

    # 步骤4：计算Signature
    result = hmacsha256(signingKey, stringToSign)
    signature = result.hex()

    return jsonify({
        "policy": stringToSign,
        "x_oss_signature_version": "OSS4-HMAC-SHA256",
        "x_oss_credential": f"{sts_accesskeyid}/{dt_obj_2}/cn-hangzhou/oss/aliyun_v4_request",
        "x_oss_date": dt_obj_1,
        "signature": signature,
        "security_token": security_token
    })


if __name__ == "__main__":
    app.run(host='0.0.0.0', port=5000)