Python Flask 对接阿里云 STS：OSS 临时凭证安全上传方案

做后端的人迟早会碰到这个问题业务要允许用户上传文件文件存储在阿里云 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 APIFlask 拿到文件流自己再去调用 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_IDLTAI5t**** export ALIBABA_CLOUD_ACCESS_KEY_SECRETyour-long-term-secret export STS_ROLE_ARNacs:ram::UID:role/oss-uploader export OSS_BUCKET_NAMEmy-bucket export OSS_ENDPOINToss-cn-hangzhou.aliyuncs.com export OSS_BASE_DIRupload3. 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_namefflask-app-{int(now)}, duration_seconds3600, ) _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 fhttps://{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 与 RoleSessionNameDurationSeconds 的边界要记清楚最小值 900也就是 15 分钟最大值默认 3600也就是 1 小时。如果角色配置了更短的最大会话时间参数再大也没有用实际有效期取两者的较小值。我默认用的是 3600但从安全角度讲纯上传型接口其实用不着 1 小时。前端传到后端、后端写 OSS整个过程在正常网络下就是秒级到分钟级的事所以我更推荐把 DurationSeconds 缩短到 900 到 1800 之间。凭证越短被截获后的破坏窗口越小。RoleSessionName 的作用很多人会忽略。在云审计的日志里每次 AssumeRole 都会被记录RoleSessionName 就是用来区分具体发起方的标签。我会让每个用户传入一个唯一会话名例如 user_123 或 order_20250101_001后续出了问题通过日志能直接定位是哪一次调用、哪个用户、上传了哪些对象。这个字段只能包含字母、数字和.-_等字符别放中文或者超长字符串。4. 直传模式前端拿临时凭证直接传 OSS4.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_namefuser-{request.remote_addr}, duration_seconds900, ) 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并实现提前刷新的缓存逻辑InvalidParameterDurationSeconds 超范围或 RoleArn 格式错误检查 900 ≤ duration ≤ 3600RoleArn 前缀必须是 acs:ram::SignatureDoesNotMatch服务器时钟偏移或 Secret 抄错校准系统时间确认长期 AccessKeySecret 和角色 ARN 无误BucketAlreadyExists创建 bucket 时名称被占用bucket 全局唯一换名字或使用已有 bucket5.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 做上传服务下次改版时优先考虑迁移到这套方案哪怕只是把凭证有效期调短一点整体的攻击暴露面也已经缩小了一个量级。