Java

在Maven項目中，導入以下依賴。

<!-- https://mvnrepository.com/artifact/com.aliyun/credentials-java -->
<dependency>
    <groupId>com.aliyun</groupId>
    <artifactId>credentials-java</artifactId>
    <version>0.3.4</version>
</dependency>

<dependency>
    <groupId>com.aliyun.kms</groupId>
    <artifactId>kms-transfer-client</artifactId>
    <version>0.1.0</version>
</dependency>

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

您可以參考如下代碼來完成服務端獲取臨時訪問憑證并計算POST簽名：

package com.sanfeng.oss.serversigneddirectupload.demos.web;

import com.aliyun.sts20150401.models.AssumeRoleResponse;
import com.aliyun.sts20150401.models.AssumeRoleResponseBody;
import com.aliyun.tea.TeaException;
import com.aliyun.oss.common.utils.BinaryUtil;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.commons.codec.binary.Base64;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.GetMapping;

import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import java.time.*;
import java.time.format.DateTimeFormatter;
import java.util.*;

@Controller
public class WebController {

    //OSS基礎信息 替換為實際的 bucket 名稱和 region-id
    String bucket = "web-direct-upload";
    String region = "cn-hangzhou";

    String host = "http://" + bucket + ".oss-" + region + ".aliyuncs.com";
    //指定上傳到OSS的文件前綴。
    String upload_dir = "dir";
    //指定過期時間，單位為秒。
    Long expire_time = 3600L;

    /**
     * 通過指定有效的時長（秒）生成過期時間。
     * @param seconds 有效時長（秒）。
     * @return ISO8601 時間字符串，如："2014-12-01T12:00:00.000Z"。
     */
    public static String generateExpiration(long seconds) {
        // 獲取當前時間戳（以秒為單位）
        long now = Instant.now().getEpochSecond();
        // 計算過期時間的時間戳
        long expirationTime = now + seconds;
        // 將時間戳轉換為Instant對象，并格式化為ISO8601格式
        Instant instant = Instant.ofEpochSecond(expirationTime);
        // 定義時區
        ZoneId zone = ZoneId.systemDefault();  // 使用系統默認時區
        // 將 Instant 轉換為 ZonedDateTime
        ZonedDateTime zonedDateTime = instant.atZone(zone);
        // 定義日期時間格式，例如2023-12-03T13:00:00.000Z
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
        // 格式化日期時間
        String formattedDate = zonedDateTime.format(formatter);
        // 輸出結果
        return formattedDate;
    }
    //初始化STS Client
    public static com.aliyun.sts20150401.Client createStsClient() throws Exception {
        // 工程代碼泄露可能會導致 AccessKey 泄露，并威脅賬號下所有資源的安全性。以下代碼示例僅供參考。
        // 建議使用更安全的 STS 方式。
        com.aliyun.teaopenapi.models.Config config = new com.aliyun.teaopenapi.models.Config()
                    // 必填，請確保代碼運行環境設置了環境變量 OSS_ACCESS_KEY_ID。
                .setAccessKeyId(System.getenv("OSS_ACCESS_KEY_ID"))
                // 必填，請確保代碼運行環境設置了環境變量 OSS_ACCESS_KEY_SECRET。
                .setAccessKeySecret(System.getenv("OSS_ACCESS_KEY_SECRET"));
        // Endpoint 請參考 https://api.aliyun.com/product/Sts
        config.endpoint = "sts.cn-hangzhou.aliyuncs.com";
        return new com.aliyun.sts20150401.Client(config);
    }

    //獲取STS臨時憑證
    public static AssumeRoleResponseBody.AssumeRoleResponseBodyCredentials getCredential() throws Exception {
        com.aliyun.sts20150401.Client client = WebController.createStsClient();
        com.aliyun.sts20150401.models.AssumeRoleRequest assumeRoleRequest = new com.aliyun.sts20150401.models.AssumeRoleRequest()
                .setRoleArn(System.getenv("OSS_ACCESS_KEY_ID"))
                .setRoleSessionName("yourRoleSessionName");// 自定義會話名稱
        com.aliyun.teautil.models.RuntimeOptions runtime = new com.aliyun.teautil.models.RuntimeOptions();
        try {
            // 復制代碼運行請自行打印 API 的返回值
            AssumeRoleResponse response = client.assumeRoleWithOptions(assumeRoleRequest, runtime);
            // credentials里包含了后續要用到的AccessKeyId、AccessKeySecret和SecurityToken。
            return response.body.credentials;
        } catch (TeaException error) {
            // 此處僅做打印展示，請謹慎對待異常處理，在工程項目中切勿直接忽略異常。
            // 錯誤 message
            System.out.println(error.getMessage());
            // 診斷地址
            System.out.println(error.getData().get("Recommend"));
            com.aliyun.teautil.Common.assertAsString(error.message);
        } catch (Exception _error) {
            TeaException error = new TeaException(_error.getMessage(), _error);
            // 此處僅做打印展示，請謹慎對待異常處理，在工程項目中切勿直接忽略異常。
            // 錯誤 message
            System.out.println(error.getMessage());
            // 診斷地址
            System.out.println(error.getData().get("Recommend"));
            com.aliyun.teautil.Common.assertAsString(error.message);
        }
        return null;
    }
    @GetMapping("/get_post_signature_for_oss_upload")
    public ResponseEntity<Map<String, String>> getPostSignatureForOssUpload() throws Exception {
        AssumeRoleResponseBody.AssumeRoleResponseBodyCredentials sts_data = getCredential();

        String accesskeyid =  sts_data.accessKeyId;
        String accesskeysecret =  sts_data.accessKeySecret;
        String securitytoken =  sts_data.securityToken;
        System.out.println("sts accesskeyid:"+accesskeyid);
        System.out.println("sts accesskeysecret:"+accesskeysecret);
        System.out.println("sts securitytoken:"+securitytoken);

        //獲取x-oss-credential里的date，當前日期，格式為yyyyMMdd
        LocalDate today = LocalDate.now();
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd");
        String date = today.format(formatter);

        //獲取x-oss-date
        ZonedDateTime now = ZonedDateTime.now().withZoneSameInstant(java.time.ZoneOffset.UTC);
        DateTimeFormatter formatter2 = DateTimeFormatter.ofPattern("yyyyMMdd'T'HHmmss'Z'");
        String x_oss_date = now.format(formatter2);

        // 步驟1：創建policy。
        String x_oss_credential = accesskeyid + "/" + date + "/" + region + "/oss/aliyun_v4_request";
        
        ObjectMapper mapper = new ObjectMapper();

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

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

        Map<String, String> bucketCondition = new HashMap<>();
        bucketCondition.put("bucket", bucket);
        conditions.add(bucketCondition);

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

        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", x_oss_credential); // 替換為實際的 access key id
        conditions.add(credentialCondition);

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

        conditions.add(Arrays.asList("content-length-range", 1, 10240000));
        conditions.add(Arrays.asList("eq", "$success_action_status", "200"));
        conditions.add(Arrays.asList("starts-with", "$key", upload_dir));

        policy.put("conditions", conditions);

        String jsonPolicy = mapper.writeValueAsString(policy);

        // 步驟2：構造待簽名字符串（StringToSign）。
        String stringToSign = new String(Base64.encodeBase64(jsonPolicy.getBytes()));
        // System.out.println("stringToSign: " + stringToSign);

        // 步驟3：計算SigningKey。
        byte[] dateKey = hmacsha256(("aliyun_v4" + accesskeysecret).getBytes(), date);
        byte[] dateRegionKey = hmacsha256(dateKey, region);
        byte[] dateRegionServiceKey = hmacsha256(dateRegionKey, "oss");
        byte[] signingKey = hmacsha256(dateRegionServiceKey, "aliyun_v4_request");
        // System.out.println("signingKey: " + BinaryUtil.toBase64String(signingKey));

        // 步驟4：計算Signature。
        byte[] result = hmacsha256(signingKey, stringToSign);
        String signature = BinaryUtil.toHex(result);
        // System.out.println("signature:" + signature);

        Map<String, String> response = new HashMap<>();
        // 將數據添加到 map 中
        response.put("version", "OSS4-HMAC-SHA256");
        // 這里是易錯點，不能直接傳policy，需要做一下Base64編碼
        response.put("policy", stringToSign);
        response.put("credential", x_oss_credential);
        response.put("ossdate", x_oss_date);
        response.put("signature", signature);
        response.put("token", securitytoken);
        response.put("dir", upload_dir);
        response.put("host", host);
        // 返回帶有狀態碼 200 (OK) 的 ResponseEntity，返回給Web端，進行PostObject操作
        return ResponseEntity.ok(response);
    }
    public static byte[] hmacsha256(byte[] key, String data) {
        try {
            // 初始化HMAC密鑰規格，指定算法為HMAC-SHA256并使用提供的密鑰。
            SecretKeySpec secretKeySpec = new SecretKeySpec(key, "HmacSHA256");

            // 獲取Mac實例，并通過getInstance方法指定使用HMAC-SHA256算法。
            Mac mac = Mac.getInstance("HmacSHA256");
            // 使用密鑰初始化Mac對象。
            mac.init(secretKeySpec);

            // 執行HMAC計算，通過doFinal方法接收需要計算的數據并返回計算結果的數組。
            byte[] hmacBytes = mac.doFinal(data.getBytes());

            return hmacBytes;
        } catch (Exception e) {
            throw new RuntimeException("Failed to calculate HMAC-SHA256", e);
        }
    }
}

Python

• 執行以下命令安裝依賴。

  pip install flask
  pip install alibabacloud_tea_openapi alibabacloud_sts20150401 alibabacloud_credentials

• 請參考如下代碼來完成Python服務端獲取臨時訪問憑證STStoken并構建上傳策略以計算POST簽名。

  from flask import Flask, render_template, jsonify
  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
  import json
  import base64
  import hmac
  import datetime
  import time
  import hashlib
  
  app = Flask(__name__)
  
  # 配置環境變量 OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_ID, OSS_STS_ROLE_ARN。
  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'  
  
  # 替換為實際的bucket名稱和region_id
  bucket = 'examplebucket'
  region_id = 'cn-hangzhou'
  
  host = f'http://{bucket}.oss-cn-hangzhou.aliyuncs.com'
  
  # 指定過期時間，單位為秒
  expire_time = 3600  
  # 指定上傳到OSS的文件前綴。
  upload_dir = 'dir'
  
  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("/")
  def hello_world():
      return render_template('index.html')
  
  @app.route('/get_post_signature_for_oss_upload', methods=['GET'])
  def generate_upload_params():
      # 初始化配置，直接傳遞憑據
      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 返回的臨時憑據
      temp_access_key_id = token_data['AccessKeyId']
      temp_access_key_secret = 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')
      
      # 構建 Policy 并生成簽名
      policy = {
          "expiration": expiration_time,
          "conditions": [
              ["eq", "$success_action_status", "200"],
              {"x-oss-signature-version": "OSS4-HMAC-SHA256"},
              {"x-oss-credential": f"{temp_access_key_id}/{dt_obj_2}/cn-hangzhou/oss/aliyun_v4_request"},
              {"x-oss-security-token": security_token},
              {"x-oss-date": dt_obj_1},  
          ]
      }
      print(dt_obj_1)
      policy_str = json.dumps(policy).strip()
  
      # 步驟2：構造待簽名字符串（StringToSign）
      stringToSign = base64.b64encode(policy_str.encode()).decode()
  
      # 步驟3：計算SigningKey
      dateKey = hmacsha256(("aliyun_v4" + temp_access_key_secret).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()
  
      # 組織返回數據
      response_data = {
          'policy': stringToSign,  #表單域
          'x_oss_signature_version': "OSS4-HMAC-SHA256",  #指定簽名的版本和算法，固定值為OSS4-HMAC-SHA256
          'x_oss_credential': f"{temp_access_key_id}/{dt_obj_2}/cn-hangzhou/oss/aliyun_v4_request",  #指明派生密鑰的參數集
          'x_oss_date': dt_obj_1,  #請求的時間
          'signature': signature,  #簽名認證描述信息
          'host': host,
          'dir': upload_dir,
          'security_token': security_token  #安全令牌
  
      }
      return jsonify(response_data)
  
  if __name__ == "__main__":
      app.run(host="0.0.0.0", port=8000)