服務端實現阿里云OSS直傳

介紹

阿里云上傳 OSS 有兩種方式，一種是普通上傳，一種是客戶端直傳。

普通上傳，就是需要先將文件上傳到服務端，然后調用接口將文件上傳到阿里云。

當然這種方案經常出現不合理的使用方式，即客戶端充當服務端的角色，在本地直接通過AK/SK，調用阿里云接口上傳文件。

不建議客戶端直接這樣做，一旦AK/SK泄露，存在很大的安全隱患，有可能被盜用。

客戶端直傳

通過服務器下發上傳令牌，客戶端通過使用臨時令牌，直接上傳到阿里云 OSS。

這種是最佳方案，不僅安全，而且不占用服務器帶寬，傳輸速度快。

接下來，本文圍繞這種方案介紹如何實現，并以 Python 語言為例，其他語言的實現類似，可以從參考文檔下載相應的 Demo

配置權限

創建用戶并生成 AK，SK

這里建議專門為 API 調用創建一個用戶，然后生成AK，SK，記錄好AK，SK以便后續使用，并授予權限。

點擊權限管理，添加 AliyunSTSAssumeRoleAccess (調用STS服務AssumeRole接口的權限)

在這里插入圖片描述

創建角色

創建一個角色，創建成功后，記錄角色的 ARN，后面代碼中會用到，為了方便演示，這里將角色命名為：ramossuploadonly

在這里插入圖片描述

添加權限策略

導航欄找到權限策略，點擊創建，點擊“腳本編輯”, 在文本框中輸入以下內容，將 <Bucket名稱> 替換成自己的 Bucket 名稱，然后點擊保存,

例如，bucket 名稱為 oss-upload-demo，則 Resource 填寫為 "acs:oss:*:*:oss-upload-demo/*"

{"Version": "1","Statement": [{"Effect": "Allow","Action": "oss:PutObject","Resource": "acs:oss:*:*:<Bucket名稱>/*"}]
}

點擊保存，例如可將名稱命名為 oss-upload-policy

為角色授權

回到剛剛添加的角色 ramossuploadonly，點擊新增授權，從權限策略中搜索剛剛添加的 oss-upload-policy，點擊確認新增授權。

這樣權限就配置完成了。

創建 bucket

打開對象存儲 OSS，點擊創建 Bucket，在彈窗中輸入 bucket 名稱

在這里插入圖片描述

需要注意的是，需要記住這里的選擇地域，后面代碼中會用到，OSS上傳需要指定地域，本文中選擇北京。

編寫代碼

安裝依賴

主要用到以下的依賴包, requirements.txt 文件內容如下

Flask
alibabacloud-credentials
alibabacloud-tea-openapi
alibabacloud-sts20150401
oss2

使用 pip 安裝：

pip install -r requirements.txt

創建 main.py 文件

基于官方 demo，主要修改以下內容：

access_key_id = '###AK###'
access_key_secret = '###SK###'
role_arn_for_oss_upload = '###acs:ram::19920XXX5721:role/roleoss###'# 自定義會話名稱
role_session_name = 'yourRoleSessionName'# 替換為實際的bucket名稱、region_id、host
bucket = ' oss-upload-demo'
region_id = 'cn-beijing'
host = 'http://oss-upload-demo.oss-cn-beijing.aliyuncs.com'

這里 access_key_id, access_key_secret 為最開始創建用戶后，拿到的 AK，SK 的值，替換成相應的內容。

role_arn_for_oss_upload 為創建的 RAM 角色ramossuploadonly時，拿到的 ARN，可以打開角色詳情找到ARN。

bucket 為自己創建的 bucket 名稱，本示例中為 oss-upload-demo
region_id 為 bucket 所在的區域，本示例中為 cn-beijing
host 為 bucket 的訪問地址，本示例中為 http://oss-upload-demo.oss-cn-beijing.aliyuncs.com, 根據不同的區域，訪問地址不同，可以通過OSS地域和訪問域名, 找到 bucket 對應地域的訪問域名，選擇外網 Endpoint

在這里插入圖片描述

需要注意的是，官方 demo 中，需要全搜索 cn-hangzhou, 替換掉 bucket 地域的 ID

完整內容如下：

from flask import Flask, render_template, jsonify, request
from alibabacloud_tea_openapi.models import Config
from alibabacloud_sts20150401.client import Client as Sts20150401Client
from alibabacloud_sts20150401 import models as sts_20150401_models
import os
import json
import base64
import hmac
import datetime
import time
import hashlibimport oss2app = Flask(__name__)# 配置環境變量 OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET, 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')access_key_id = '###AK###'
access_key_secret = '###SK###'
role_arn_for_oss_upload = '###acs:ram::19920XXX5721:role/roleoss###'# 自定義會話名稱
role_session_name = 'yourRoleSessionName'# 替換為實際的bucket名稱、region_id、host
bucket = ' oss-upload-demo'
region_id = 'cn-beijing'
host = 'http://oss-upload-demo.oss-cn-beijing.aliyuncs.com'# 指定過期時間，單位為秒
expire_time =  1000# 指定上傳到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 hmacBytesexcept 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')# 定義回調參數Base64編碼函數。def encode_callback(callback_params):cb_str = json.dumps(callback_params).strip()return oss2.compat.to_string(base64.b64encode(oss2.compat.to_bytes(cb_str)))# 構建回調配置并 Base64 編碼callback_config = {"callbackUrl": "http://x.x.x.x/images/callback",  # 替換為您的回調服務器地址"callbackBody": "bucket=${bucket}&object=${object}&etag=${etag}&size=${size}","callbackBodyType": "application/x-www-form-urlencoded"}encoded_callback = encode_callback(callback_config)# 構建 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}/{region_id}/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：計算SigningKeydateKey = hmacsha256(("aliyun_v4" + temp_access_key_secret).encode(), dt_obj_2)dateRegionKey = hmacsha256(dateKey, region_id)dateRegionServiceKey = hmacsha256(dateRegionKey, "oss")signingKey = hmacsha256(dateRegionServiceKey, "aliyun_v4_request")# 步驟4：計算Signatureresult = 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-beijing/oss/aliyun_v4_request",  # 指明派生密鑰的參數集'x_oss_date': dt_obj_1,  # 請求的時間'signature': signature,  # 簽名認證描述信息'host': host,'dir': upload_dir,'security_token': security_token,  # 安全令牌#'callback': encoded_callback   # 返回 Base64 編碼的回調配置}return jsonify(response_data)if __name__ == "__main__":app.run(host="127.0.0.1", port=8000)  # 如果需要監聽其他地址如0.0.0.0，需要您自行在服務端添加認證機制

這里下載好官方給出的 Demo

創建 html 頁面

在項目中創建目錄 templates，然后創建一個 index.html 文件，內容為：

<!DOCTYPE html>
<html lang="zh-CN">
<head><meta charset="UTF-8" /><meta name="viewport" content="width=device-width, initial-scale=1.0" /><title>服務端生成簽名上傳文件到OSS</title>
</head>
<body>
<div class="container"><form><div class="mb-3"><label for="file" class="form-label">選擇文件:</label><input type="file" class="form-control" id="file" name="file" required /></div><button type="submit" class="btn btn-primary">上傳</button></form><div id="callback-info" class="mt-3" style="display: none;"><h4>回調信息:</h4><pre id="callback-content"></pre></div>
</div><script type="text/javascript">
document.addEventListener('DOMContentLoaded', function () {const form = document.querySelector("form");const fileInput = document.querySelector("#file");const callbackInfo = document.querySelector("#callback-info");const callbackContent = document.querySelector("#callback-content");form.addEventListener("submit", (event) => {event.preventDefault();const file = fileInput.files[0];if (!file) {alert('請選擇一個文件再上傳。');return;}const filename = file.name;fetch("/get_post_signature_for_oss_upload", { method: "GET" }).then((response) => {if (!response.ok) {throw new Error("獲取簽名失敗");}return response.json();}).then((data) => {let formData = new FormData();formData.append("success_action_status", "200");formData.append("policy", data.policy);formData.append("x-oss-signature", data.signature);formData.append("x-oss-signature-version", "OSS4-HMAC-SHA256");formData.append("x-oss-credential", data.x_oss_credential);formData.append("x-oss-date", data.x_oss_date);formData.append("key", data.dir + file.name); // 文件名formData.append("x-oss-security-token", data.security_token);formData.append("callback", data.callback);  // 添加回調參數formData.append("file", file); // file 必須為最后一個表單域return fetch(data.host, {method: "POST",body: formData});}).then((response) => {if (response.ok) {console.log("上傳成功");alert("文件已上傳");return response.json();  // 解析回調信息} else {console.log("上傳失敗", response);alert("上傳失敗，請稍后再試");}}).then((callbackData) => {if (callbackData) {callbackContent.textContent = JSON.stringify(callbackData, null, 2);callbackInfo.style.display = "block";}}).catch((error) => {console.error("發生錯誤:", error);});});
});
</script>
</body>
</html>