使用Identity出站认证安全管理敏感信息

出站认证（Outbound Authentication）

出站认证是指智能体运行时在访问外部系统时所使用的身份认证机制。与入站认证（Inbound Authentication，验证调用方身份）不同，出站认证解决的是“Agent 如何安全地证明自己的身份并获取访问外部资源的凭证”。

AgentArts Identity 服务通过以下组件实现出站认证：

凭证生命周期

支持的认证类型

OAuth2凭证提供者支持的Vendor：

OAuth2认证流程支持两种模式：

步骤一：创建出站身份

出站身份是凭证获取的主体，智能体运行时通过出站身份标识自身，并向Identity服务请求凭证。

步骤二：在SDK中获取敏感信息

完成控制台配置后，在智能体代码中使用Identity SDK获取凭证。SDK 提供两种获取模式：显式调用和装饰器自动注入。

方式一：显式调用

适用于需要精细控制凭证获取时机的场景。

from agentarts.sdk.service.identity import IdentityClient

# 初始化 Identity 客户端
client = IdentityClient(region="cn-southwest-2")

# 步骤 a：获取出站访问令牌（Outbound Access Token）
# 方式 1：使用用户 Token 创建
workload_token = client.create_workload_access_token(
    workload_name="my-agent-outbound",
    user_token="user-oauth-token"  # 从运行时上下文获取
)

# 方式 2：使用用户 ID 创建
# workload_token = client.create_workload_access_token_for_user_id(
#     workload_name="my-agent-outbound",
#     user_id="user-123"
# )

# 步骤 b：根据凭证提供者类型获取对应凭证

# 获取 API Key
api_key = client.get_resource_api_key(
    provider_name="openai-api-key",
    workload_access_token=workload_token
)

# 获取 OAuth2 Access Token（支持指定 scopes）
# oauth2_token = client.get_resource_oauth2_token(
#     provider_name="github-oauth",
#     scopes=["user:email", "read:user"],
#     workload_access_token=workload_token
# )

# 获取 STS Token（临时 AK/SK/SecurityToken）
# sts_token = client.get_resource_sts_token(
#     provider_name="huaweicloud-iam",
#     workload_access_token=workload_token,
#     agency_session_name="agent-session"
# )
# print(f"Access Key: {sts_token.access_key_id}")
# print(f"Secret Key: {sts_token.secret_access_key}")
# print(f"Security Token: {sts_token.security_token}")

方式二：装饰器自动注入（推荐）

使用装饰器模式时，SDK会自动获取出站访问令牌并注入凭证到函数参数中，开发者无需手动管理令牌生命周期。推荐使用此方式，可减少代码量并自动处理凭证刷新。

from typing import Optional
from agentarts.sdk import require_access_token, require_api_key, require_sts_token
from agentarts.sdk.identity.types import StsCredentials

# API Key 自动注入（同步函数）
@require_api_key(provider_name="openai-api-key")
def call_llm(api_key: Optional[str] = None):
    """api_key 参数由 SDK 自动注入，无需手动传入"""
    import openai
    client = openai.OpenAI(api_key=api_key)
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": "Hello!"}]
    )
    return response.choices[0].message.content

# OAuth2 Access Token 自动注入 - USER_FEDERATION 模式（异步函数）
@require_access_token(
    provider_name="github-oauth",
    scopes=["user:email", "read:user"],
    auth_flow="USER_FEDERATION"
)
async def fetch_github_user_data(access_token: Optional[str] = None):
    """access_token 参数由 SDK 自动注入"""
    import httpx
    async with httpx.AsyncClient() as http_client:
        resp = await http_client.get(
            "https://api.github.com/user",
            headers={"Authorization": f"Bearer {access_token}"}
        )
        return resp.json()

# OAuth2 Access Token 自动注入 - M2M 模式（异步函数）
@require_access_token(
    provider_name="internal-service-oauth",
    auth_flow="M2M"
)
async def call_internal_service(access_token: Optional[str] = None):
    """M2M 模式，服务间直接认证"""
    import httpx
    async with httpx.AsyncClient() as http_client:
        resp = await http_client.post(
            "https://internal.example.com/api",
            headers={"Authorization": f"Bearer {access_token}"}
        )
        return resp.json()

# STS Token 自动注入（异步函数）
@require_sts_token(
    provider_name="huaweicloud-iam",
    agency_session_name="agent-session"
)
async def access_huawei_resource(sts_credentials: Optional[StsCredentials] = None):
    """sts_credentials 参数由 SDK 自动注入"""
    print(f"AK: {sts_credentials.access_key_id}")
    print(f"SK: {sts_credentials.secret_access_key}")
    print(f"SecurityToken: {sts_credentials.security_token}")
    # 使用 STS 凭证访问华为云资源
    # ...

步骤三：在Agent代码中安全使用凭证

以下是一个完整的Agent示例，展示在智能体运行时中使用Identity出站认证访问外部LLM服务的完整流程。

"""
AgentArts 智能体示例：使用 Identity 出站认证安全获取 API Key 并调用 LLM
"""
from typing import Optional
from agentarts.sdk import require_api_key
from agentarts.sdk.runtime import AgentArtsRuntimeApp

# 使用装饰器自动注入 API Key
# provider_name 对应步骤1中创建的身份名称
@require_api_key(provider_name="openai-api-key")
def generate_response(user_input: str, api_key: Optional[str] = None) -> str:
    """
    使用 Identity 注入的 API Key 调用 LLM 生成回复。

    Args:
        user_input: 用户输入文本
        api_key: 由 SDK 装饰器自动注入，无需手动传入

    Returns:
        LLM 生成的回复文本
    """
    import openai

    # 使用 Identity 注入的 API Key，而非环境变量
    client = openai.OpenAI(api_key=api_key)

    try:
        response = client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": "你是一个有帮助的助手。"},
                {"role": "user", "content": user_input}
            ],
            temperature=0.7
        )
        return response.choices[0].message.content
    except Exception as e:
        # 凭证获取或 API 调用失败时的降级处理
        # 注意：错误日志中不要打印 api_key 内容
        print(f"LLM 调用失败: {type(e).__name__}")
        return "抱歉，服务暂时不可用，请稍后重试。"

# 使用 AgentArts SDK 封装为标准运行时应用
app = AgentArtsRuntimeApp()

@app.invocation
def handle_invocation(request):
    """处理智能体调用请求"""
    user_input = request.get("input", "")
    response = generate_response(user_input)
    return {"output": response}

if __name__ == "__main__":
    app.run()  # 默认监听 8080 端口

Python SDK完整示例：多凭证类型场景

以下示例展示在一个Agent中同时使用API Key、OAuth2和STS三种凭证类型：

"""
AgentArts 智能体示例：同时使用多种凭证类型
- API Key：调用 LLM 服务
- OAuth2：访问 GitHub API
- STS Token：访问华为云 OBS
"""
from typing import Optional
from agentarts.sdk import require_api_key, require_access_token, require_sts_token
from agentarts.sdk.identity.types import StsCredentials
from agentarts.sdk.runtime import AgentArtsRuntimeApp

app = AgentArtsRuntimeApp()


# ========== API Key 场景：调用 LLM ==========
@require_api_key(provider_name="openai-api-key")
def call_llm(prompt: str, api_key: Optional[str] = None) -> str:
    """使用 API Key 调用 LLM"""
    import openai
    client = openai.OpenAI(api_key=api_key)
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content


# ========== OAuth2 场景：访问 GitHub ==========
@require_access_token(
    provider_name="github-oauth",
    scopes=["user:email", "read:user"],
    auth_flow="USER_FEDERATION"
)
async def get_github_user_info(access_token: Optional[str] = None) -> dict:
    """使用 OAuth2 Token 访问 GitHub 用户信息"""
    import httpx
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            "https://api.github.com/user",
            headers={"Authorization": f"Bearer {access_token}"}
        )
        return resp.json()


# ========== STS Token 场景：访问华为云 OBS ==========
@require_sts_token(
    provider_name="huaweicloud-iam",
    agency_session_name="obs-access-session"
)
async def upload_to_obs(
    file_data: bytes,
    bucket: str,
    object_key: str,
    sts_credentials: Optional[StsCredentials] = None
) -> str:
    """使用 STS Token 上传文件到 OBS"""
    from obs import ObsClient

    obs_client = ObsClient(
        access_key_id=sts_credentials.access_key_id,
        secret_access_key=sts_credentials.secret_access_key,
        security_token=sts_credentials.security_token,
        server="obs.cn-southwest-2.myhuaweicloud.com"
    )

    resp = obs_client.putContent(bucket, object_key, file_data)
    if resp.status < 300:
        return f"obs://{bucket}/{object_key}"
    else:
        raise Exception(f"OBS 上传失败: {resp.errorCode}")


# ========== 主入口 ==========
@app.invocation
async def handle_invocation(request):
    """处理智能体调用请求"""
    task_type = request.get("task", "")

    if task_type == "chat":
        result = call_llm(request.get("input", ""))
        return {"output": result}

    elif task_type == "github":
        user_info = await get_github_user_info()
        return {"output": f"GitHub 用户: {user_info.get('login', 'unknown')}"}

    elif task_type == "upload":
        file_data = request.get("file_data", "").encode()
        url = await upload_to_obs(
            file_data=file_data,
            bucket="my-bucket",
            object_key=f"uploads/{request.get('filename', 'file.txt')}"
        )
        return {"output": f"文件已上传: {url}"}

    return {"output": "未知任务类型"}


if __name__ == "__main__":
    app.run()

凭证过期与刷新处理

Identity SDK的装饰器模式会自动处理短期凭证的获取和刷新。对于需要手动管理凭证生命周期的场景，参考以下模式：

"""
凭证过期处理与自动重试模式
"""
from typing import Optional
import time
from functools import wraps
from agentarts.sdk.service.identity import IdentityClient

# 初始化 Identity 客户端
identity_client = IdentityClient(region="cn-southwest-2")

# 缓存凭证（仅缓存非敏感的标识信息，不缓存凭证本身）
_credential_cache = {}
_credential_expiry = {}


def get_fresh_api_key(provider_name: str, workload_token: str, max_retries: int = 3) -> str:
    """
    获取 API Key，支持自动重试。

    Args:
        provider_name: 凭证提供者名称
        workload_token: 出站访问令牌
        max_retries: 最大重试次数

    Returns:
        API Key 字符串

    Raises:
        RuntimeError: 重试次数耗尽后仍无法获取凭证
    """
    last_error = None

    for attempt in range(max_retries):
        try:
            api_key = identity_client.get_resource_api_key(
                provider_name=provider_name,
                workload_access_token=workload_token
            )
            return api_key
        except Exception as e:
            last_error = e
            error_type = type(e).__name__
            # 不打印凭证相关信息，仅打印错误类型
            print(f"凭证获取失败 (尝试 {attempt + 1}/{max_retries}): {error_type}")
            # 指数退避等待
            time.sleep(2 ** attempt)

    raise RuntimeError(f"凭证获取失败，已重试 {max_retries} 次。最后错误: {type(last_error).__name__}")


def with_api_key_retry(provider_name: str, max_retries: int = 3):
    """
    装饰器：为需要 API Key 的函数添加自动重试逻辑。

    当凭证获取失败时自动重试，而非直接失败。
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    # 每次尝试都重新获取凭证，不使用缓存
                    workload_token = identity_client.create_workload_access_token(
                        workload_name="my-agent-outbound",
                        user_token=kwargs.get("user_token", "")
                    )
                    api_key = get_fresh_api_key(
                        provider_name=provider_name,
                        workload_token=workload_token,
                        max_retries=1  # 内部不再重试，由外层控制
                    )
                    kwargs["api_key"] = api_key
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt < max_retries - 1:
                        print(f"调用失败 (尝试 {attempt + 1}/{max_retries}): {type(e).__name__}")
                        time.sleep(2 ** attempt)
                    else:
                        raise
        return wrapper
    return decorator


# 使用示例
@with_api_key_retry(provider_name="openai-api-key", max_retries=3)
def call_llm_with_retry(user_input: str, api_key: Optional[str] = None) -> str:
    """带自动重试的 LLM 调用"""
    import openai
    client = openai.OpenAI(api_key=api_key)
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": user_input}]
    )
    return response.choices[0].message.content

从环境变量迁移到Identity出站认证

以下示例展示如何将现有环境变量方式逐步迁移到 Identity 出站认证，支持过渡期双模式运行：

"""
迁移辅助器：从环境变量模式平滑迁移到 Identity 出站认证
支持双模式运行，便于灰度迁移
"""
import os
from typing import Optional
from agentarts.sdk import require_api_key


# 迁移开关：通过环境变量控制是否启用 Identity 出站认证
USE_IDENTITY_AUTH = os.getenv("AGENTARTS_USE_IDENTITY_AUTH", "false").lower() == "true"


def get_api_key(provider_name: str, env_var_name: str) -> Optional[str]:
    """
    获取 API Key，支持双模式：
    - Identity 出站认证模式（推荐）
    - 环境变量模式（迁移过渡期使用）

    Args:
        provider_name: Identity 凭证提供者名称
        env_var_name: 环境变量名称（迁移过渡期使用）

    Returns:
        API Key 字符串
    """
    if USE_IDENTITY_AUTH:
        # Identity 模式：通过 SDK 获取
        # 使用装饰器需要包装在函数中
        @require_api_key(provider_name=provider_name)
        def _get_key(api_key: Optional[str] = None) -> str:
            return api_key
        return _get_key()
    else:
        # 环境变量模式：迁移过渡期使用
        # 添加告警日志提醒迁移
        import logging
        logging.warning(
            f"正在使用环境变量 {env_var_name} 获取凭证，"
            "建议迁移到 Identity 出站认证。"
            "设置 AGENTARTS_USE_IDENTITY_AUTH=true 启用 Identity 模式。"
        )
        return os.getenv(env_var_name)


# 使用示例
def call_external_api(user_input: str) -> str:
    """调用外部 API，自动适配 Identity 或环境变量模式"""
    api_key = get_api_key(
        provider_name="openai-api-key",
        env_var_name="OPENAI_API_KEY"
    )

    if not api_key:
        raise RuntimeError("无法获取 API Key，请检查 Identity 配置或环境变量")

    import openai
    client = openai.OpenAI(api_key=api_key)
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": user_input}]
    )
    return response.choices[0].message.content

凭证最小权限原则

凭证轮换策略

"""
凭证轮换管理示例
建议通过 CI/CD 管道定期更新凭证提供者
"""
from agentarts.sdk.service.identity import IdentityClient

client = IdentityClient(region="cn-southwest-2")

def rotate_api_key(provider_name: str, new_api_key: str):
    """
    轮换 API Key 凭证提供者。

    在 Identity 服务中更新凭证后，Agent 运行时下次获取凭证时
    将自动获取新值，无需重启运行时。

    建议轮换周期：
    - API Key：每 90 天
    - OAuth2 Client Secret：每 180 天
    - STS Token：自动过期，无需手动轮换
    """
    # 删除旧凭证提供者
    # 注意：SDK 暂不提供直接更新接口，需删除后重建
    # client.delete_credential_provider(name=provider_name)

    # 创建新的凭证提供者
    provider = client.create_api_key_credential_provider(
        name=provider_name,
        api_key=new_api_key
    )
    print(f"凭证轮换完成: {provider_name}")

轮换凭证时，旧凭证提供者删除前需确保没有正在进行的请求使用旧凭证。建议在业务低峰期执行轮换操作。

日志脱敏

在Agent代码中，严禁将凭证内容输出到日志。以下规则必须严格遵守：

"""
日志脱敏示例
"""
import logging

logger = logging.getLogger(__name__)

# 配置日志过滤器，过滤敏感信息
class CredentialRedactionFilter(logging.Filter):
    """过滤日志中的敏感信息"""

    SENSITIVE_PATTERNS = [
        "api_key", "api-key", "apikey",
        "access_token", "access-token", "accesstoken",
        "secret", "password", "credential",
        "authorization", "bearer"
    ]

    def filter(self, record):
        message = str(record.getMessage())
        for pattern in self.SENSITIVE_PATTERNS:
            if pattern in message.lower():
                # 标记为敏感，但保留错误类型
                record.msg = f"[REDACTED: 含敏感字段 {pattern}]"
                return True
        return True

# 应用过滤器
logger.addFilter(CredentialRedactionFilter())


@require_api_key(provider_name="openai-api-key")
def safe_call(api_key: Optional[str] = None):
    """正确做法：日志中不包含凭证"""
    logger.info("开始调用 LLM API")  # 正确
    # logger.info(f"使用 API Key: {api_key}")  # 错误！禁止这样做

    try:
        # 调用逻辑...
        pass
    except Exception as e:
        logger.error(f"调用失败: {type(e).__name__}")  # 正确：仅记录错误类型
        # logger.error(f"调用失败: {e}")  # 有风险：异常信息可能包含凭证

错误处理与降级策略

当凭证获取失败时，Agent应具备合理的降级策略，而非直接崩溃：

"""
凭证获取失败的降级策略
"""
from typing import Optional
from agentarts.sdk import require_api_key
import logging

logger = logging.getLogger(__name__)


class FallbackResponse:
    """降级响应"""
    RATE_LIMITED = "服务繁忙，请稍后重试"
    AUTH_FAILED = "服务暂时不可用，请稍后重试"
    UNKNOWN_ERROR = "服务异常，请稍后重试"


def call_with_fallback(provider_name: str, fallback_message: str = FallbackResponse.UNKNOWN_ERROR):
    """
    装饰器：为函数添加凭证获取失败降级逻辑

    Args:
        provider_name: Identity 凭证提供者名称
        fallback_message: 降级时返回的消息
    """
    def decorator(func):
        @require_api_key(provider_name=provider_name)
        def wrapper(*args, api_key: Optional[str] = None, **kwargs):
            if api_key is None:
                # 凭证获取失败，返回降级消息
                logger.warning(f"凭证获取失败，触发降级: {func.__name__}")
                return fallback_message

            try:
                return func(*args, api_key=api_key, **kwargs)
            except Exception as e:
                error_type = type(e).__name__
                logger.error(f"函数执行失败: {func.__name__}, 错误类型: {error_type}")

                # 根据错误类型返回不同降级消息
                if "RateLimit" in error_type:
                    return FallbackResponse.RATE_LIMITED
                elif "Auth" in error_type:
                    return FallbackResponse.AUTH_FAILED
                else:
                    return fallback_message
        return wrapper
    return decorator


# 使用示例
@call_with_fallback(
    provider_name="openai-api-key",
    fallback_message="AI 助手暂时无法响应，请稍后重试。"
)
def generate_response(user_input: str, api_key: Optional[str] = None) -> str:
    """生成回复，带降级保护"""
    import openai
    client = openai.OpenAI(api_key=api_key)
    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": user_input}]
    )
    return response.choices[0].message.content