配置依賴。

<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)