做后端的人迟早会碰到这个问题：业务要允许用户上传文件，文件存储在阿里云 OSS，但你不能把 AccessKey 直接暴露给前端或让文件绕过权限验证。我在工程里折腾过几轮之后，确定下来的标准方案就是 Python 后端对接阿里云 STS，签发出临时访问凭证，再通过 Flask 暴露 API，让内部系统拿着临时凭证把文件安全上传到 OSS。这篇文章会把这套方案的完整思路、核心代码、权限配置和排障过程全部讲清楚，适合正在搭文件上传服务、或者想把现有上传模块从"长期密钥裸奔"升级为"临时凭证"的开发者参考。
这套方案解决的核心痛点，概括起来就一句话：长期密钥不安全、不可控、到期收回成本高；而 STS 临时凭证天然有时效、有范围，就算被截获，也只是在窗口期里多了一个受限身份，不会让整个 Bucket 暴露在风险中。下面从设计思路开始，一步步把整条链路拆开讲透。
1. 为什么选 STS + Flask 这条路线
1.1 长期密钥的痛点和 STS 的定位
先说说我为什么不愿意用长期 AccessKey 直接做上传。长期凭证有两个绕不开的麻烦：不好收回，也不好精细控制范围。如果 AccessKey 泄露出去了，你得去控制台重新生成密钥，然后所有引用它的系统都要跟着改配置。就算你不把 AccessKey 暴露给前端，只让它躺在后端代码里，一旦代码仓库权限管理不严，密钥照样可能被"顺走"。
STS 做的事情，简单理解就是"用长期密钥换一张短期入场券"。这个入场券由阿里云 STS 服务签发，包含一组临时 AccessKeyId、AccessKeySecret 和 SecurityToken，默认有效期最短 15 分钟、最长 1 小时。过期之后自动作废，你不需要去吊销，也不需要通知所有调用方重新换密钥。调用方只要在这段时间内完成上传，之后凭证就自动失效——这就把"密钥轮换"这种事后补救动作，变成了"凭证过期"这种事前和事中兜底。
具体到 OSS 上传场景，STS 还有一个隐藏优势：它可以配合 RAM 角色做到非常细的授权。我会给角色配置一个类似"只能对 images 这个 Bucket 下面的 upload/ 目录执行 PutObject"的策略，就算前端 JS 文件里出现一段临时凭证，攻击者拿到的权限也仅限于往那个目录里写文件，看不了文件列表、删不了库、改不了别人的对象。这种精细程度是长期密钥很难做到的。
1.2 为什么服务端选 Flask 而不是其他框架
本质上，一个"签发临时凭证 + 代理上传"的服务，职责很小，不需要引入重型框架。用 Flask 的最直接原因是轻量：没有强制的 ORM、没有自带数据库迁移、没有复杂的工程骨架，几十行 Python 就能把接口跑起来。我之前用 Django 做过类似的内部工具，建项目、配 settings、注册 app 的流程比接口本身还费时间；换 Flask 之后，前后不到一个小时就立了一个可用的服务。
需要注意，我并不是让 Flask 承担所有业务逻辑。这层 API 只负责两件事：校验调用方身份、返回临时凭证或接收上传请求，然后把文件流转交给 OSS SDK。任何重计算、异步处理、格式转换都不要堆在这一层，否则 Flask 的轻量优势会被业务复杂度抵消。如果你的团队对 Python Web 生态没那么熟，这个选型依然是门槛最低的。
1.3 整体架构：先分清两种上传模式
后面代码里会反复涉及两个概念，这里必须先讲清楚。
第一种是"服务端代传"。客户端把文件 POST 到 Flask API，Flask 拿到文件流，自己再去调用 STS 获取临时凭证，然后以临时凭证身份把文件写入 OSS，最后返回一个可访问的 URL。这种模式适合内部系统、文件不大、并发量可控的场景，开发上最简单，但也意味着服务器要承接文件的中转流量。
第二种是"客户端直传"。Flask 只负责签发并返回 STS 临时凭证，前端拿到凭证后直接用阿里云 OSS 的 JS SDK 把文件传上去。文件不经过应用服务器，带宽压力全部落到了 OSS 的接入层，支持大文件分片上传，适合面向公众用户的高并发产品。
这两条路看着不同，底层完全一样，都依赖 STS 的 AssumeRole 流程。先把 STS 和权限这一层打通，选哪种模式就是切换接口形态的问题而已。
2. 动手前的硬准备：RAM 角色与权限策略
2.1 创建 RAM 角色并配置信任策略
整个流程的起点是创建一个 RAM 角色。登录阿里云 RAM 控制台，选择"角色管理"，创建一个"用户自定义"角色，可信实体选"阿里云账号"。这里最关键的一项配置是信任策略，它规定"谁有资格扮演这个角色"。我的策略大概长这样：
{ "Version": "1", "Statement": [ { "Effect": "Allow", "Action": "sts:AssumeRole", "Principal": { "RAM": ["acs:ram::<阿里云账号UID>:root"] } } ] }如果生产环境里有多个 RAM 用户，可以把 Principal 从 root 收得更紧，改成具体的用户，例如 acs:ram:: :user/oss-upload-bot。我这边用 root 写法的原因，是实际持有长期密钥的 RAM 用户本身就是专用的运维账号，已经和其他账号做了隔离；但是如果你团队里账号划分不清晰，建议还是不要把信任范围放宽到 root。
2.2 角色授权策略：按最小权限设计
给角色附加权限策略时，最容易犯的错是顺手放一个"AliyunOSSFullAccess"。这份策略确实能让所有功能跑通，但它也意味着扮演成功的临时凭证能对名下所有 Bucket 做任何操作，包括删除、列举、修改权限。安全起见，我每次都会按最小权限写策略：所有上传都落到同一个目录下，策略里只授权这个目录的写入和读取：
{ "Version": "1", "Statement": [ { "Effect": "Allow", "Action": ["oss:PutObject"], "Resource": [ "acs:oss:*:*:my-bucket/upload/*" ] } ] }如果之后还需要生成下载链接，再补一条 oss:GetObject；需要列出文件目录，再补 oss:ListObjects。我的建议是第一次实现只放 PutObject，等某个具体需求确实出现，再逐条追加权限。权限这东西，放出去容易，哪天想收回来，运行中的老旧业务十有八九会"炸"，所以起步越窄越好。
2.3 准备环境：依赖安装与配置项
代码层面依赖三个库：Flask、oss2、阿里云 STS SDK。直接在虚拟环境里安装：
pip install flask oss2 aliyun-python-sdk-sts这里提醒一句，aliyun-python-sdk-sts 对 aliyun-python-sdk-core 的版本有依赖，如果你之前装过旧版 core，建议一起升级，避免出现循环依赖或者方法签名不匹配。Python 版本我建议 3.10 及以上，倒不是说低版本跑不了，而是新版对类型标注和异步支持更好，后面扩展起来舒服。
配置项不要写死在源码里。我统一放到环境变量，部署时直接通过进程环境注入：
export ALIBABA_CLOUD_ACCESS_KEY_ID="LTAI5t****" export ALIBABA_CLOUD_ACCESS_KEY_SECRET="your-long-term-secret" export STS_ROLE_ARN="acs:ram::<UID>:role/oss-uploader" export OSS_BUCKET_NAME="my-bucket" export OSS_ENDPOINT="oss-cn-hangzhou.aliyuncs.com" export OSS_BASE_DIR="upload"3. Flask API 核心实现：临时凭证签发与文件上传
3.1 签发 STS 临时凭证的完整代码
先写一个独立的模块，负责调用 AssumeRole。我一般单独建一个 sts_client.py，方便后续其他服务复用：
import json from aliyunsdkcore.client import AcsClient from aliyunsdksts.request.v20150401 import AssumeRoleRequest def get_assume_role_token( access_key_id: str, access_key_secret: str, role_arn: str, session_name: str = "flask-app", duration_seconds: int = 3600, region: str = "cn-hangzhou", ) -> dict: client = AcsClient(access_key_id, access_key_secret, region) request = AssumeRoleRequest.AssumeRoleRequest() request.set_RoleArn(role_arn) request.set_RoleSessionName(session_name) request.set_DurationSeconds(duration_seconds) response = client.do_action_with_exception(request) result = json.loads(response) return result["Credentials"]返回结果里就是经典的四个字段：AccessKeyId、AccessKeySecret、SecurityToken、Expiration。有一点值得明确：这个临时 AccessKeyId 的典型前缀是 STS.，和我们长期密钥的格式完全不同。看到这个前缀，就应该意识到当前的身份是扮演出来的临时身份，只能用于限定时长的操作。
3.2 用临时凭证完成 OSS 上传
Flask 侧的服务端代传模式，核心实现如下：
import os import uuid import time from flask import Flask, request, jsonify import oss2 app = Flask(__name__) ACCESS_KEY_ID = os.environ["ALIBABA_CLOUD_ACCESS_KEY_ID"] ACCESS_KEY_SECRET = os.environ["ALIBABA_CLOUD_ACCESS_KEY_SECRET"] STS_ROLE_ARN = os.environ["STS_ROLE_ARN"] OSS_BUCKET_NAME = os.environ["OSS_BUCKET_NAME"] OSS_ENDPOINT = os.environ.get("OSS_ENDPOINT", "oss-cn-hangzhou.aliyuncs.com") OSS_BASE_DIR = os.environ.get("OSS_BASE_DIR", "upload") # 全局简单缓存，避免每次都重新调用 STS _global_cache = {} def _refresh_token_if_needed(): now = time.time() if "expire_at" not in _global_cache or now >= _global_cache["expire_at"]: cred = get_assume_role_token( ACCESS_KEY_ID, ACCESS_KEY_SECRET, STS_ROLE_ARN, session_name=f"flask-app-{int(now)}", duration_seconds=3600, ) _global_cache["cred"] = cred _global_cache["expire_at"] = iso_to_timestamp(cred["Expiration"]) - 300 return _global_cache["cred"] @app.post("/api/oss/upload") def upload_file(): if "file" not in request.files: return jsonify({"error": "file is required"}), 400 file = request.files["file"] # 通过 uuid 隔离多人同名文件 key = f"{OSS_BASE_DIR}/{uuid.uuid4().hex}/{file.filename}" cred = _refresh_token_if_needed() auth = oss2.StsAuth( cred["AccessKeyId"], cred["AccessKeySecret"], cred["SecurityToken"], ) bucket = oss2.Bucket(auth, OSS_ENDPOINT, OSS_BUCKET_NAME) bucket.put_object(key, file.stream) url = f"https://{OSS_BUCKET_NAME}.{OSS_ENDPOINT}/{key}" return jsonify({"key": key, "url": url}), 200这段代码里最关键的一行是 oss2.StsAuth。oss2 的 Bucket 构造支持三种认证方式：Auth（长期密钥）、StsAuth（临时凭证）、AnonymousAuth（匿名访问）。使用临时凭证时，SecurityToken 必须作为第三个参数传进去，漏传的话 OSS 端签名校验直接失败，返回 AccessDenied。
至于在对象 key 中插入 uuid 片段，是为了避免不同用户上传同名文件相互覆盖。只要你还在做多用户上传，这个习惯一定要养成。后面要做日志追溯时，uuid 也是一条很高效的索引线索。
3.3 关键参数解析：DurationSeconds 与 RoleSessionName
DurationSeconds 的边界要记清楚：最小值 900，也就是 15 分钟；最大值默认 3600，也就是 1 小时。如果角色配置了更短的最大会话时间，参数再大也没有用，实际有效期取两者的较小值。
我默认用的是 3600，但从安全角度讲，纯上传型接口其实用不着 1 小时。前端传到后端、后端写 OSS，整个过程在正常网络下就是秒级到分钟级的事，所以我更推荐把 DurationSeconds 缩短到 900 到 1800 之间。凭证越短，被截获后的破坏窗口越小。
RoleSessionName 的作用很多人会忽略。在云审计的日志里，每次 AssumeRole 都会被记录，RoleSessionName 就是用来区分具体发起方的标签。我会让每个用户传入一个唯一会话名，例如 user_123 或 order_20250101_001，后续出了问题，通过日志能直接定位是哪一次调用、哪个用户、上传了哪些对象。这个字段只能包含字母、数字和.@-_等字符，别放中文或者超长字符串。
4. 直传模式：前端拿临时凭证直接传 OSS
4.1 直传解决什么问题
如果你做的是面向公众用户的产品，我强烈建议用直传模式。原因很简单：服务端代传时，用户上传一个 100MB 的视频，你的 ECS 要先收 100MB，再转发 100MB 到 OSS，服务器带宽被硬生生吞掉一份中转流量。高峰期并发上来，网络 I/O 和 CPU 都容易被打满，服务器成了瓶颈，OSS 却闲着。
直传模式下，Flask 只负责下发临时凭证，之后浏览器和 OSS 之间直接建立上传链路。文件内容不经过应用服务器，带宽压力全部落在 OSS 上，应用服务器只需要处理一个轻量的凭证请求。如果你的应用有图片压缩、视频转码这类需求，可以等直传完成后再触发异步处理任务，而不是让上传本身阻塞在应用层。
4.2 后端签发凭证接口
直传模式下 Flask 接口变得更简单，它不再接收文件，只返回凭证：
@app.post("/api/sts/token") def issue_sts_token(): # 实际生产环境里，这里必须校验调用方身份，比如 Header API Key cred = get_assume_role_token( ACCESS_KEY_ID, ACCESS_KEY_SECRET, STS_ROLE_ARN, session_name=f"user-{request.remote_addr}", duration_seconds=900, ) return jsonify({ "access_key_id": cred["AccessKeyId"], "access_key_secret": cred["AccessKeySecret"], "security_token": cred["SecurityToken"], "expiration": cred["Expiration"], "bucket": OSS_BUCKET_NAME, "endpoint": OSS_ENDPOINT, })注意这里 DurationSeconds 我直接给了 900 秒。原因是浏览器从拉取凭证到完成上传，通常只有几十秒，完全没有必要让凭证活得更久。另外，session_name 这里我偷懒用了 IP，但真实生产环境更合理的做法是拼接用户 ID 和会话 ID，因为同一个出口 IP 可能对应多个用户，会降低审计日志的准确性。
4.3 前端配合与 CORS 配置
前端拿到凭证后，直接用阿里云官方 SDK 或者简单 fetch 写文件：
const resp = await fetch("/api/sts/token", { method: "POST" }); const data = await resp.json(); const client = new OSS({ region: "oss-cn-hangzhou", accessKeyId: data.access_key_id, accessKeySecret: data.access_key_secret, stsToken: data.security_token, bucket: data.bucket }); const result = await client.put("upload/example.png", file); console.log(result.url);浏览器直传 OSS 会遇到跨域问题。OSS Bucket 的"跨域设置"里必须增加一条允许来源为你的域名，Allowed 方法至少要包含 PUT、POST、GET、HEAD，还要允许 OPTIONS 预检请求，因为 OSS SDK 在上传前会先发一个 OPTIONS 请求确认服务器允许跨域。我第一次在线上配置时就把 Allowed Header 漏了，前端一直报 CORS 错误，排查半天才发现是预检请求被拦。
提示：CORS 配置里的 Allowed Origin，生产环境尽量不要用
*。精确到你的业务域名，能避免其他已知站点借用你 Bucket 的跨域能力。本地调试时再用通配符或者 localhost 单独开一条规则。
5. 踩坑实录与排查技巧
5.1 常见错误码速查表
我把自己遇到过的和身边同事踩过的典型报错整理成一个速查表，遇到问题先按这个思路过一遍。
| 错误码 / 报错信息 | 出现场景 | 排查方向 |
|---|---|---|
| InvalidAccessKeyId.NotFound | 使用了 STS 返回的临时 AccessKeyId，但请求里没带 SecurityToken | 检查 oss2.StsAuth 是否传入第三个参数 |
| AccessDenied | 临时凭证没有对应目录的写权限 | 检查 RAM 角色授权策略的 Resource 是否正确 |
| SecurityTokenExpired | 凭证超过有效期 | 重新调用 AssumeRole，并实现提前刷新的缓存逻辑 |
| InvalidParameter | DurationSeconds 超范围，或 RoleArn 格式错误 | 检查 900 ≤ duration ≤ 3600，RoleArn 前缀必须是 acs:ram:: |
| SignatureDoesNotMatch | 服务器时钟偏移，或 Secret 抄错 | 校准系统时间，确认长期 AccessKeySecret 和角色 ARN 无误 |
| BucketAlreadyExists | 创建 bucket 时名称被占用 | bucket 全局唯一，换名字或使用已有 bucket |
5.2 我在实际工程中踩过的三个坑
第一个坑是临时凭证缓存策略没做好。我第一次实现时，把凭证缓存写成了"进程启动时获取一次，之后一直复用"，结果跑了一个小时左右上传开始大面积报 SecurityTokenExpired。改成每次上传都重新调用 STS 后，又发现高并发下 STS 接口调用量太大。最终方案就是在内存里缓存凭证，并且在真正过期前 5 分钟提前刷新。这样 STS 的调用频率被压到瓶颈以下，上传链路又不中断。
第二个坑是中文文件名的编码问题。OSS 控制台里看文件名没问题，但通过浏览器访问 URL 时出现乱码。排查后确认是 key 里直接包含中文和空格，链路各层对 URL 的编码处理不一致。后来我统一改成服务端生成 uuid 文件名作为 OSS 对象 key，原始文件名在 API 的响应字段里单独返回，展示层自己去解码，彻底避开 URL 编码的坑。
第三个坑是前端直传模式下的 CORS 预检。本地开发环境用 localhost:5000 去调 OSS 没问题，一换到线上域名就报跨域。检查发现 OSS 控制台的跨域规则里只配了 GET，没配 OPTIONS，也没有把实际的请求头加入 Allowed Header。把跨域规则补齐后，问题当场解决。还有一个容易忽略的细节：Bucket 的跨域规则是分区域生效的，如果你用了多个 Region，每个 Bucket 都要单独配。
6. 生产环境的安全加固建议
6.1 密钥管理：环境变量只是一个起点
环境变量解决了"密钥不写死在代码里"的问题，但它仍然意味着运维能直接读到明文。更稳妥的方案是把密钥放到密钥管理类服务中，比如阿里云 KMS，或者外部 Vault，应用启动时再动态拉取。我见过不少本可以避免的事故，背后都是同一个原因：AccessKeySecret 被写进配置文件并推到 Git 仓库，后来仓库权限失控，密钥跟着泄漏。
我自己的习惯是：长期密钥只存在于部署环境的内存中，任何日志、响应体、数据库字段都不能出现完整的 AccessKeySecret。前端哪怕只瞥到一眼，都应该立即轮换。如果你发现日志里有 SDL 工具扫出来的密钥字样，宁愿多花 5 分钟重新生成，也不要侥幸它只是一次性打印。
6.2 签发临时凭证的接口必须加鉴权
这一点再怎么强调都不过分：开放一个没有任何校验的 STS 签发接口，等于把 RAM 角色的完整能力变成公共 API。只要有人拿到你的接口地址，他就能不停地换取临时凭证，然后利用角色权限往 OSS 里传垃圾文件，甚至制造流量账单。
最简单的做法是给调用方分配 API Key，放在请求头里校验。如果是面向普通用户的网页，则要求登录态，并且在签发时绑定用户 ID、限制来源 IP。理论上还应给每个账号设置 STS 签发频率上限，防止被恶意利用者循环调用。
6.3 凭证时效、目录隔离与审计闭环
在实际部署里，我会把"签发凭证"和"文件上传"拆成两个独立服务。Flask 只负责前者，后者沉淀成一个公共上传组件，供不同业务复用。这样每个业务可以有自己的角色和目录，即使某个业务线被攻击，损失也不会横向扩散到其他业务线。
策略上，建议每个模块分配一个独立的前缀目录，比如 upload/order/、upload/user/avatar/，然后在策略 Resource 里分别匹配。临时凭证的时长按业务类型设：上传头像这类小文件 15 分钟足够，而视频转码这类大对象上传，可以适当放宽到 30 分钟，但最好不要超过 1 小时。
最后，开启云审计的 STS 调用记录。每天花一分钟扫一眼 AssumeRole 日志，重点看来源 IP 和 RoleSessionName 是否和业务预期一致。发现异常调用，立即禁用对应 RAM 用户或收紧角色的信任策略，同时轮换长期密钥。这套闭环跑顺之后，文件上传这条链路的安全性就基本可控了。
按照我个人的习惯，在一开始就预留好"凭证签发接口和文件上传组件分离"的扩展点，后面迁移新业务非常省事。回看整个实现，最值得记住的一点其实是：STS 并不是把技术方案变复杂，而是把安全从"密钥轮换"这种事后补救动作，前移成了"凭证过期"这种事前和事中兜底。如果你现在还在用长期 AccessKey 做上传服务，下次改版时优先考虑迁移到这套方案，哪怕只是把凭证有效期调短一点，整体的攻击暴露面也已经缩小了一个量级。
系统)
