验证轮换

概览

本指南会告诉你如何验证：在云 provider 中轮换后的 secret，会在下一次 automation kickoff 立即被拾取 - 无需重新部署，也无需重启 worker。它只在你已经配置了 Workload Identity-backed credential 时才相关（AWS、GCP、Azure）。Static-credential 部署在轮换后需要重新部署；这部分无需验证。

下面的示例使用一个很小的自包含 crew：一个 tool、一个 agent、一个 task。crew prompt 本身不会引用 secret value - 取而代之的是，tool 会从 os.environ 读取它，并报告自己看到的 SHA-256 fingerprint。先在你的云 provider 中轮换 secret，再 kickoff 一次，fingerprint 就会发生变化。

Note

为什么用 fingerprint，而不是原始值？把原始 secret 写进 LLM 输出和 trace logs 会造成泄露风险。fingerprint 足以确认“值变了”，同时不会把实际值写到任何可观察的位置。

前提条件

在运行这个验证前：

• 已配置 WI-backed Secret Provider Credential（AWS、GCP、Azure）。
• 你的 deployment 上有一个 Secret = true 的环境变量，key 为 API_KEY（或你偏好的名称 - 请让下方 tool 与之匹配），它引用的是你云 provider 中的一个 secret。
• 你有办法在云 provider 中更新 secret value（CLI access 或 cloud console）。
• 你有办法通过 HTTP 触发部署 kickoff（curl、Postman，或 CrewAI Platform 的 Run tab）。

第 1 步 - 搭建一个 Verification Crew

创建一个 classic crew project，因为这个示例会通过 crew.py 接入一个 Python tool：

crewai create crew rotation_verifier --classic --skip_provider
cd rotation_verifier

第 2 步 - 添加 Credential Echo Tool

将 src/rotation_verifier/tools/custom_tool.py 替换为一个读取 secret-backed env var 并返回 fingerprint 的 tool：

"""Tool that verifies a runtime-injected secret without leaking the value.

Reads the secret-backed env var (populated by the workload-identity
secrets manager at kickoff time) and returns a stable fingerprint. Never
echo raw credential values into LLM output or logs in production code —
the fingerprint alone is sufficient to confirm rotation worked.
"""

from __future__ import annotations

import hashlib
import os

from crewai.tools import BaseTool


# Match the deployment environment variable's `key` field.
ENV_VAR_NAME = "API_KEY"


class CredentialEchoTool(BaseTool):
    name: str = "credential_echo"
    description: str = (
        "Read the API credential from the worker's environment and return a "
        "fingerprint summary. Use this exactly once when asked to verify the "
        "current credential. Takes no arguments."
    )

    def _run(self) -> str:
        value = os.environ.get(ENV_VAR_NAME)
        if not value:
            return (
                f"ERROR: {ENV_VAR_NAME} env var is not set. The workload-"
                "identity secret fetch did not run, or the deployment is "
                "missing the secret-backed env var."
            )
        fingerprint = hashlib.sha256(value.encode()).hexdigest()[:12]
        return f"Authenticated. credential.fingerprint=sha256:{fingerprint}"

第 3 步 - 替换默认 Agent 和 Task 配置

这个 crew 只有一个 agent 和一个 task - 它们的描述绝不会提到 secret value，因此 task key 在轮换之间保持稳定。

credential_checker:
  role: >
    Credential Verifier
  goal: >
    Confirm that the workload-identity-backed secret reached this worker
    process and report a fingerprint of the current value.
  backstory: >
    You are a no-nonsense reliability engineer responsible for verifying
    that secrets fetched at runtime via workload identity are present
    and fresh. You always use the credential_echo tool exactly once and
    report the result verbatim — you never make up values.

verify_credential_task:
  description: >
    Use the credential_echo tool to read the runtime-injected credential
    and produce a one-line confirmation. The current year is {current_year}
    (use it only in the timestamp; do not transform the credential output).
  expected_output: >
    A single line in the form:
    "[{current_year}] <credential_echo tool's exact output>"
  agent: credential_checker

第 4 步 - 连接 Crew Class

from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task
from crewai.agents.agent_builder.base_agent import BaseAgent

from rotation_verifier.tools.credential_echo_tool import CredentialEchoTool


@CrewBase
class RotationVerifierCrew():
    """Single-task crew that verifies a workload-identity-backed secret
    was successfully fetched at runtime.

    Rotate the underlying secret in the cloud provider, kickoff again, and
    the credential fingerprint in the agent's report changes — without any
    re-deploy, worker restart, or input change. The crew prompt itself
    never references the secret value.
    """

    agents: list[BaseAgent]
    tasks: list[Task]

    @agent
    def credential_checker(self) -> Agent:
        return Agent(
            config=self.agents_config["credential_checker"],
            tools=[CredentialEchoTool()],
            verbose=True,
        )

    @task
    def verify_credential_task(self) -> Task:
        return Task(config=self.tasks_config["verify_credential_task"])

    @crew
    def crew(self) -> Crew:
        return Crew(
            agents=self.agents,
            tasks=self.tasks,
            process=Process.sequential,
            verbose=True,
        )

第 5 步 - 部署并配置 Secret Env Var

把这个 crew 部署到 CrewAI Platform，就像部署任何其他 crew 一样。然后在该 deployment 的 Environment Variables 页面：

• Key: API_KEY（必须与 tool 中的 ENV_VAR_NAME 一致）
• Value Source: 你在 AWS WI 或 GCP WI 中设置的 WI-backed credential
• Secret Name: 你云 provider 的 Secret Manager 中该 secret 的名称

第 6 步 - 运行第一次 Kickoff

把 <DEPLOYMENT_AUTH_TOKEN> 和 <DEPLOYMENT_HOST> 替换成你 deployment 的 Run tab 中提供的值。

curl -m 60 \
  -H "Authorization: Bearer <DEPLOYMENT_AUTH_TOKEN>" \
  -H "Content-Type: application/json" \
  -X POST https://<DEPLOYMENT_HOST>/kickoff \
  -d '{"inputs":{"current_year":"2026"}}'

当 kickoff 完成后（几秒钟），检查 agent 的输出。你会看到：

[2026] Authenticated. credential.fingerprint=sha256:004421b993c9

记下 fingerprint。这个 hash 唯一对应于你云 provider 中当前的 secret value。

第 7 步 - 在云 provider 中轮换 Secret

AWS

aws secretsmanager update-secret \
  --region <REGION> \
  --secret-id <SECRET_NAME> \
  --secret-string "rotated value"

GCP

添加一个新版本（Secret Manager 始终读取 latest）：

echo -n "rotated value" | gcloud secrets versions add <SECRET_NAME> \
  --data-file=- \
  --project=<YOUR_PROJECT_ID>

Azure

az keyvault secret set \
  --vault-name <VAULT_NAME> \
  --name <SECRET_NAME> \
  --value "rotated value"

第 8 步 - 运行第二次 Kickoff 并比较

curl -m 60 \
  -H "Authorization: Bearer <DEPLOYMENT_AUTH_TOKEN>" \
  -H "Content-Type: application/json" \
  -X POST https://<DEPLOYMENT_HOST>/kickoff \
  -d '{"inputs":{"current_year":"2026"}}'

现在 agent 的输出会显示一个不同的 fingerprint：

[2026] Authenticated. credential.fingerprint=sha256:e2fc89848f72

这证明轮换后的值已经被正在运行的部署拾取，且无需重新部署、无需 worker restart，也无需其他操作。

这能验证什么，以及不能验证什么

验证了：

• CrewAI Platform 到 WI OIDC token 的签发正常。
• 云侧信任（AWS 的 IAM OIDC provider、GCP 的 Workload Identity Pool、Azure 的 Federated Identity Credential）接受该 token。
• 云侧身份（IAM Role / GCP service account / Entra App Registration）有权读取 secret。
• secret value 在 kickoff 时进入了 worker process 的 os.environ。
• 后续轮换会传播到下一次 kickoff。

未验证：

• 你的真实生产 crews 是否能优雅处理轮换 - 例如，长时间运行的任务如果只在启动时读取一次 env var，就会一直使用旧值直到任务结束。请据此设计：在使用点读取 secrets，而不是在 module import 时读取。

为什么不把 secret 直接写进 prompt？

一种更直观的 demo 方式，是把 secret value 直接放进 task description 里（例如，“Research about {api_key}”），然后检查 prompt。**不要这样做。**有两个原因：

1. 它会把 secret 泄露到 LLM 调用 trace 和 provider 侧日志中。 任何能访问 trace 的人都能读到它。
2. 它会在每次 kickoff 时改变 task description。 CrewAI Platform 通过 task description 的 MD5 hash 来识别 task；轮换值意味着每次 kickoff hash 都会变化，从而破坏 deploy-time → runtime 的 task mapping。症状是：task 记录会一直显示为 pending_run，或者多 task crew 中只有部分 task 被注册。

本指南中的 tool-based 模式同时规避了这两个问题：prompt 是静态的，tool 在运行时读取 env var，而 LLM 只会看到该值的 fingerprint。

下一步

• 返回 Secrets Manager 概览
• 验证完成后，可以删除这个 verification crew。真实 crews 应该遵循相同模式：secrets 通过 tool 内部的 os.environ 访问，绝不要直接替换进 prompt。