You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何在Python中使用OAuth2令牌实现Blob存储的用户特定身份验证?

如何在Python中使用OAuth2令牌实现Blob存储的用户特定身份验证?

嗨,我来帮你梳理下怎么实现这个需求!其实你要的方案完全可行,核心思路是让前端把用户登录Azure AD时拿到的、针对Blob存储的OAuth2令牌传给Azure函数,然后函数用这个令牌直接访问Blob存储,不用再依赖托管身份。下面是具体的步骤和代码示例:

核心思路拆解

  1. 前端获取正确范围的令牌:用户登录Azure AD时,要请求https://storage.azure.com/.default这个范围的权限,这样Azure AD颁发的令牌才拥有访问Blob存储的权限。之后把这个令牌通过请求头(比如Authorization: Bearer <用户令牌>)传给你的Azure函数。
  2. 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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.04.16 09:15:29