如何在Python中使用OAuth2令牌实现Blob存储的用户特定身份验证?
如何在Python中使用OAuth2令牌实现Blob存储的用户特定身份验证?
嗨,我来帮你梳理下怎么实现这个需求!其实你要的方案完全可行,核心思路是让前端把用户登录Azure AD时拿到的、针对Blob存储的OAuth2令牌传给Azure函数,然后函数用这个令牌直接访问Blob存储,不用再依赖托管身份。下面是具体的步骤和代码示例:
核心思路拆解
- 前端获取正确范围的令牌:用户登录Azure AD时,要请求
https://storage.azure.com/.default这个范围的权限,这样Azure AD颁发的令牌才拥有访问Blob存储的权限。之后把这个令牌通过请求头(比如Authorization: Bearer <用户令牌>)传给你的Azure函数。 - Azure函数用令牌创建Blob客户端:在函数里,我们可以用Azure SDK提供的工具把用户令牌包装成可用的凭证,然后初始化Blob服务客户端,就像用托管身份时一样操作Blob。
具体代码实现
首先确保你的函数安装了必要的Python包:
pip install azure-storage-blob azure-identity pyjwt
然后是Azure函数的核心代码:
import os from azure.storage.blob import BlobServiceClient from azure.identity import StaticTokenCredential from azure.core.credentials import AccessToken import jwt from datetime import datetime, timedelta def main(req): # 从请求头提取用户的OAuth2令牌 auth_header = req.headers.get('Authorization') if not auth_header or not auth_header.startswith('Bearer '): return "缺少或无效的身份验证头", 401 user_access_token = auth_header.split(' ')[1] # 解析令牌的过期时间(如果前端没主动传的话) try: # 不验证签名,只解析payload获取过期时间 token_payload = jwt.decode(user_access_token, options={"verify_signature": False}) expires_timestamp = token_payload['exp'] expires_on = datetime.fromtimestamp(expires_timestamp) except Exception as e: # 解析失败的话,暂时设置1小时过期(建议让前端传过期时间更可靠) expires_on = datetime.utcnow() + timedelta(hours=1) # 把用户令牌包装成Azure SDK认可的AccessToken对象 access_token = AccessToken(token=user_access_token, expires_on=expires_on.timestamp()) # 用StaticTokenCredential来封装令牌,供Blob客户端使用 user_credential = StaticTokenCredential(access_token) # 初始化Blob服务客户端 storage_account_name = os.environ.get('STORAGE_ACCOUNT_NAME') blob_service_client = BlobServiceClient( account_url=f"https://{storage_account_name}.blob.core.windows.net", credential=user_credential ) # 示例:获取存储账户下的容器列表 containers = list(blob_service_client.list_containers()) return {"容器列表": [container.name for container in containers]}
关键注意事项
- 权限配置:要给登录的用户分配Blob存储的RBAC角色,比如「Storage Blob Data Reader」(只读)或「Storage Blob Data Contributor」(读写),否则即使令牌有效,也会被Blob存储拒绝访问。
- 令牌范围:前端请求令牌时必须指定
https://storage.azure.com/.default作为范围,不然Azure AD不会给令牌赋予Blob存储的访问权限。 - 过期时间处理:令牌的过期时间一定要正确设置,
StaticTokenCredential会根据这个判断令牌是否有效。解析JWT的exp字段是最准确的方式,比手动假设过期时间靠谱。
你之前在azure.identity文档里没找到相关方法,是因为这个场景用的是StaticTokenCredential——它专门用来处理已经获取到的令牌,把它转换成Azure SDK客户端能识别的凭证类型,完美匹配你的需求。
备注:内容来源于stack exchange,提问作者spinning_out_of_time
相关产品推荐
相关产品推荐

