Azure Functions App中OAuth2令牌的安全持久化存储问询

为什么之前的方法行不通 #

• 应用设置（App Settings）是只读的运行时配置：通过os.environ修改的只是当前函数进程内的内存变量，不会持久化到Azure后台的应用设置中，重启或新实例启动后就会丢失。
• 直接修改应用设置需要调用Azure管理API，不仅权限控制复杂，还会导致配置刷新延迟，不适合高频更新的令牌场景。

推荐的安全持久化方案 #

1. Azure Key Vault（最安全的首选方案） #

Key Vault是Azure专门用于存储敏感信息的服务，自带加密和细粒度权限控制，完美适配令牌这类敏感数据。

操作步骤：

• 为你的Function App启用系统分配托管标识。
• 创建Azure Key Vault，并给托管标识分配Key Vault Secrets User（读取）和Key Vault Secrets Officer（写入）权限。
• 安装Azure Key Vault SDK：pip install azure-keyvault-secrets azure-identity

Python代码示例：

from azure.identity import DefaultAzureCredential
from azure.keyvault.secrets import SecretClient
import time

# 初始化Key Vault客户端
VAULT_URL = "https://your-keyvault-name.vault.azure.net/"
credential = DefaultAzureCredential()
secret_client = SecretClient(vault_url=VAULT_URL, credential=credential)

def get_token():
    # 尝试从Key Vault读取现有令牌
    try:
        access_token_secret = secret_client.get_secret("ACCESS_TOKEN")
        expires_on_secret = secret_client.get_secret("AUTH_EXPIRES_ON")
        expires_on = int(expires_on_secret.value)
        
        # 检查令牌是否未过期（提前5分钟刷新）
        if time.time() < expires_on - 300:
            return access_token_secret.value, secret_client.get_secret("REFRESH_TOKEN").value
    except Exception as e:
        print("读取令牌失败，将获取新令牌:", e)
    
    # 调用认证API获取新令牌（替换为你的现有认证逻辑）
    auth_response = your_auth_function()
    new_access_token = auth_response.get("access_token")
    new_refresh_token = auth_response.get("refresh_token")
    new_expires_on = auth_response.get("expires_on")
    
    # 将新令牌写入Key Vault
    secret_client.set_secret("ACCESS_TOKEN", new_access_token)
    secret_client.set_secret("REFRESH_TOKEN", new_refresh_token)
    secret_client.set_secret("AUTH_EXPIRES_ON", str(new_expires_on))
    
    return new_access_token, new_refresh_token

2. Azure Storage Blob（低成本方案） #

如果不需要Key Vault的高级权限控制，可使用加密的Blob存储，存储序列化的令牌数据。

操作步骤：

• 创建Azure存储账户，启用Blob服务加密（默认启用）。
• 创建Blob容器，设置私有访问权限，给Function App的托管标识分配Storage Blob Data Contributor权限。
• 安装SDK：pip install azure-storage-blob azure-identity

Python代码示例：

from azure.identity import DefaultAzureCredential
from azure.storage.blob import BlobServiceClient
import json
import time

STORAGE_ACCOUNT_URL = "https://your-storage-account-name.blob.core.windows.net/"
CONTAINER_NAME = "token-container"
BLOB_NAME = "tokens.json"

credential = DefaultAzureCredential()
blob_service_client = BlobServiceClient(STORAGE_ACCOUNT_URL, credential=credential)
container_client = blob_service_client.get_container_client(CONTAINER_NAME)

def get_token():
    # 读取Blob中的令牌
    try:
        blob_client = container_client.get_blob_client(BLOB_NAME)
        token_data = json.loads(blob_client.download_blob().readall())
        
        if time.time() < token_data["expires_on"] - 300:
            return token_data["access_token"], token_data["refresh_token"]
    except Exception as e:
        print("读取令牌失败，将获取新令牌:", e)
    
    # 获取新令牌（替换为你的现有认证逻辑）
    auth_response = your_auth_function()
    token_data = {
        "access_token": auth_response.get("access_token"),
        "refresh_token": auth_response.get("refresh_token"),
        "expires_on": auth_response.get("expires_on")
    }
    
    # 写入Blob
    blob_client = container_client.get_blob_client(BLOB_NAME)
    blob_client.upload_blob(json.dumps(token_data), overwrite=True)
    
    return token_data["access_token"], token_data["refresh_token"]

3. Azure Redis Cache（高性能自动过期方案） #

如果令牌需要高频访问，Redis的内存存储+自动过期特性可以减少频繁读写持久化存储的开销。

操作步骤：

• 创建Azure Redis Cache实例，启用持久化（可选，防止缓存丢失）。
• 获取Redis连接字符串，存储在Function App的应用设置中。
• 安装SDK：pip install redis

Python代码示例：

import redis
import json
import time
import os

REDIS_CONN_STR = os.environ["REDIS_CONNECTION_STRING"]
r = redis.from_url(REDIS_CONN_STR)

def get_token():
    # 读取Redis中的令牌
    token_data_str = r.get("oauth_tokens")
    if token_data_str:
        token_data = json.loads(token_data_str)
        if time.time() < token_data["expires_on"] - 300:
            return token_data["access_token"], token_data["refresh_token"]
    
    # 获取新令牌（替换为你的现有认证逻辑）
    auth_response = your_auth_function()
    token_data = {
        "access_token": auth_response.get("access_token"),
        "refresh_token": auth_response.get("refresh_token"),
        "expires_on": auth_response.get("expires_on")
    }
    
    # 写入Redis，设置过期时间为令牌有效期（可选）
    expire_seconds = token_data["expires_on"] - int(time.time())
    r.setex("oauth_tokens", expire_seconds, json.dumps(token_data))
    
    return token_data["access_token"], token_data["refresh_token"]

方案选择建议 #

• 优先选Azure Key Vault：令牌属于敏感数据，Key Vault的安全防护级别最高，符合合规要求。
• 低成本需求选Storage Blob：适合令牌更新频率不高的场景。
• 高性能需求选Redis Cache：适合高频触发的Function，减少认证请求次数。

内容的提问来源于stack exchange，提问作者Daniel Garcia