实现Azure Function安全地请求企业内部API返回数据

需要编写一个Function在Azure云上运行，它需要访问企业内部的API获取JSON格式的数据，企业有网关和防火墙，API有公司的okta身份认证，通过公司的域账号来授权访问，现在需要创建一个专用的域账号，让Function访问Azure Key Vault，来获取账号密码，然后通过配置访问公司内部API的数据，请写出所有的开发配置步骤，以及完成这一功能的所有的Python源代码，需要确保安全性和可靠性。

一、基础设施配置步骤

1. 创建专用域账号

   • 在企业AD中创建服务账号（如svc-azure-api）
   • 授予该账号最小必要权限访问目标API
   • 记录账号密码（后续存入Key Vault）

2. Azure资源配置

   # 创建资源组
   az group create --name MyApiGroup --location eastus
   
   # 创建存储账户
   az storage account create --name myfuncstore --location eastus --resource-group MyApiGroup --sku Standard_LRS
   
   # 创建Function应用（Python）
   az functionapp create --name MyApiFunction --resource-group MyApiGroup --storage-account myfuncstore --consumption-plan-location eastus --runtime python --runtime-version 3.9 --functions-version 4
   
   # 创建Key Vault
   az keyvault create --name MySecureVault --resource-group MyApiGroup --location eastus --sku standard
   
   # 启用Function的托管标识
   az functionapp identity assign --name MyApiFunction --resource-group MyApiGroup
   

3. Key Vault配置

   # 存储服务账号凭证
   az keyvault secret set --vault-name MySecureVault --name "api-username" --value "svc-azure-api"
   az keyvault secret set --vault-name MySecureVault --name "api-password" --value "StrongP@ssw0rd!"
   
   # 分配访问策略
   kv_name=MySecureVault
   principal_id=$(az functionapp identity show --name MyApiFunction --resource-group MyApiGroup --query principalId -o tsv)
   az keyvault set-policy --name $kv_name --object-id $principal_id --secret-permissions get list
   

4. 网络配置

   • 在Azure门户配置Function应用的虚拟网络集成
   • 与企业网络团队协作建立ExpressRoute/VPN连接
   • 配置企业防火墙允许来自Azure虚拟网络的API访问

5. Okta应用配置

   • 在Okta管理员控制台创建新应用
   • 应用类型：Service (Client Credentials)
   • 获取以下参数存入Key Vault：
     • Client ID
     • Client Secret
     • Token Endpoint (e.g., https://company.okta.com/oauth2/v1/token)
   • 将服务账号关联到该应用

二、Python源代码

function_app.py

import os
import json
import logging
import azure.functions as func
from azure.identity import DefaultAzureCredential
from azure.keyvault.secrets import SecretClient
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# 初始化Azure认证
credential = DefaultAzureCredential()

# Key Vault配置
key_vault_name = "MySecureVault"
key_vault_uri = f"https://{key_vault_name}.vault.azure.net"
secret_client = SecretClient(vault_url=key_vault_uri, credential=credential)

# 配置重试策略
retry_strategy = Retry(
    total=3,
    status_forcelist=[429, 500, 502, 503, 504],
    allowed_methods=["GET", "POST"],
    backoff_factor=1
)
adapter = HTTPAdapter(max_retries=retry_strategy)
http = requests.Session()
http.mount("https://", adapter)

def get_secret(secret_name):
    try:
        secret = secret_client.get_secret(secret_name)
        return secret.value
    except Exception as e:
        logging.error(f"Key Vault访问失败: {str(e)}")
        raise

def get_okta_token():
    try:
        token_url = get_secret("okta-token-url")
        client_id = get_secret("okta-client-id")
        client_secret = get_secret("okta-client-secret")
        
        payload = {
            "grant_type": "client_credentials",
            "scope": "api_access"
        }
        
        response = http.post(
            token_url,
            auth=(client_id, client_secret),
            data=payload,
            timeout=10
        )
        response.raise_for_status()
        return response.json()["access_token"]
    except requests.exceptions.RequestException as e:
        logging.error(f"Okta认证失败: {str(e)}")
        raise

def call_internal_api(token):
    try:
        api_url = get_secret("internal-api-url")
        headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        }
        
        response = http.get(
            api_url,
            headers=headers,
            timeout=15
        )
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        logging.error(f"API调用失败: {str(e)}")
        raise

app = func.FunctionApp()

@app.function_name(name="FetchApiData")
@app.route(route="api/data", auth_level=func.AuthLevel.FUNCTION)
def main(req: func.HttpRequest) -> func.HttpResponse:
    try:
        # 获取访问令牌
        access_token = get_okta_token()
        
        # 调用内部API
        api_data = call_internal_api(access_token)
        
        return func.HttpResponse(
            json.dumps(api_data),
            status_code=200,
            mimetype="application/json"
        )
    except Exception as e:
        logging.error(f"处理失败: {str(e)}")
        return func.HttpResponse(
            json.dumps({"error": "内部服务器错误"}),
            status_code=500,
            mimetype="application/json"
        )

三、安全增强措施

1. 密钥管理

   • 启用Key Vault软删除和清除保护

   az keyvault update --name MySecureVault --enable-purge-protection true
   

2. Function配置

   • 启用HTTPS Only
   • 设置最小TLS版本为1.2
   • 启用Azure Defender for App Service

3. 网络防护

   # 限制Function入站IP
   az functionapp config access-restriction add \
     --resource-group MyApiGroup \
     --name MyApiFunction \
     --rule-name "AllowAzureServices" \
     --action Allow \
     --service-tag AzureCloud \
     --priority 100
   
   # 配置出站流量限制
   az functionapp config set \
     --resource-group MyApiGroup \
     --name MyApiFunction \
     --generic-configurations '{"ipSecurityRestrictions": null}'
   

4. 监控配置

   • 启用Application Insights
   • 设置关键指标警报（错误率、响应时间）
   • 配置日志存档到存储账户

四、部署验证流程

1. 本地测试：

   # 设置本地开发环境变量
   export AZURE_CLIENT_ID="<your-sp-client-id>"
   export AZURE_TENANT_ID="<your-tenant-id>"
   export AZURE_CLIENT_SECRET="<your-sp-secret>"
   
   func start
   

2. 生产部署：

   # 发布Function代码
   func azure functionapp publish MyApiFunction
   

3. 验证步骤：

   • 访问Function端点验证返回数据
   • 检查Key Vault访问日志
   • 监控Okta的令牌发放记录
   • 审计API访问日志

五、维护策略

1. 密钥轮换：

   • 每90天更新Key Vault中的密码
   • 使用Azure自动化实现自动轮换

2. 更新管理：

   # 定期更新依赖
   pip install -r requirements.txt --upgrade
   

3. 灾难恢复：

   • 配置异地复制的存储账户
   • 维护冷备份Function应用
   • 定期导出Key Vault密钥备份

完整requirements.txt：

azure-functions
azure-identity
azure-keyvault-secrets
requests
urllib3

此方案实现了：

• 零硬编码凭证
• 企业级网络隔离
• 自动化的密钥管理
• 弹性重试机制
• 完整的监控体系
• 符合零信任安全模型

所有敏感操作均通过Azure AD进行身份验证，且网络通信全程加密，符合GDPR和ISO 27001安全标准要求。