AI导哈

Vault代理凭证管理

使用HashiCorp Vault管理的5分钟TTL GCP凭证，通过短时效令牌和基于角色的访问控制来缓解代理应用中的安全威胁。

  • by ChrisAdkin8
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 ChrisAdkin8/vault-mcp-agents

使用说明

保管库mcp代理人

此回购演示了针对以下三种威胁的缓解措施 代理应用程序的OWASP前10名（2026） 使用HashiCorp Vault代理的5分钟TTL GCP凭据：

OWASP ID 威胁 此仓库如何减轻威胁
ASI03 身份和特权滥用 Vault代理所有凭据——人类身份验证、代理身份（AppRole）和GCP令牌都以5分钟的硬性TTL颁发。磁盘上不存在长期密钥。
ASI02 工具误用与利用 YAML策略映射每个 (human_role, agent_id) 与明确的工具搭配。MCP服务器只公开允许的工具，因此代理不能调用其范围之外的工具。
ASI10 流氓特工 短期凭据和每个代理的工具范围包含爆炸半径——受感染的代理只能到达其允许的工具，任何被盗的GCP令牌都会在几分钟内过期。

备注

尽管Vault代理的凭证是用于GCP的，但此回购所说明的原则是有效的 Azure和AWS也同样适用。

演示场景

两个LangChain代理调用单独的MCP服务器，每个服务器的功能由 组合身份 呼叫代理和经过身份验证的人类用户。GCP凭据通过HashiCorp Vault进行代理——应用程序中不存在长期服务帐户密钥。

为什么是5分钟证书？

长期使用的GCP凭据是安全事件的常见来源。一个有效期为一小时的被盗OAuth2令牌为攻击者提供了一个广泛的窗口来窃取数据或配置资源。通过将凭证租约限制在5分钟，该项目执行了以下原则 最短持续时间特权：Vault发行的每个GCP令牌在攻击者能够通过横向移动实际发现并利用它之前就会过期。

下图显示了对比。在第一种情况下，受感染的代理持有永不过期的静态凭据——攻击者有无限的时间枚举资源、泄露数据和枢轴。在第二种情况下，同样的妥协会产生一个在5分钟内到期的令牌，将一个持久的后门变成一个短暂的、有界的事件。

5分钟的上限由两个独立的层执行，因此在发行代币之前，双方必须达成一致：

层 配置 它控制什么
Vault GCP模拟帐户 ttl = "300" 在每个 vault_gcp_secret_impersonated_account 在Terraform 服务器侧天花板中——Vault将其作为 lifetime GCP的 generateAccessToken API，因此代币在5分钟后真正过期
应用程序策略 max_gcp_token_ttl: "5m" 在 policies/capabilities.yaml 客户端防护——应用程序策略声明了审计和深度防御的预期最大TTL

这个项目展示了什么

问题 如何处理
人工身份验证 Vault用户通行证（可插入LDAP/OIDC）
代理身份 每个代理的Vault AppRole
GCP证书颁发 保管GCP机密引擎→ 短期OAuth2令牌
工具级访问控制 YAML策略文件映射 (human_role, agent_id) → 允许的MCP工具
代理商↔ MCP通信 MCP stdio传输 IdentityContext 通过环境变量传递
代理框架 LangChain create_tool_calling_agent 使用改编自MCP的工具

体系结构概述

看 建筑.md 用于图表和模式描述。

快速开始

先决条件

  • Python 3.11–3.13 （3.14是 不 支持——见下文）
  • Docker（用于本地Vault）
  • 启用了API（存储、BigQuery、计算）的GCP项目
  • LLM API密钥（Anthropic或OpenAI）

检查你的Python版本

python3 --version

如果输出显示3.14或更高版本，则需要显式使用早期版本。 在macOS上使用Homebrew：

brew install python@3.13
python3.13 --version   # confirm it prints Python 3.13.x

为什么不是Python 3.14？ Python 3.14改变了类型注释的方式 在运行时进行评估。LangChain基于Pydantic的类触发 TypeError: 'function' object is not subscriptable 在进口期间， 使图书馆在3.14+上无法使用。

1.启动Vault

docker compose up -d
export VAULT_ADDR=http://127.0.0.1:8200
export VAULT_TOKEN=dev-root-token

2.配置保险库

bash scripts/setup_vault.sh

这将创建三个测试用户（alice/operator、bob/analysis、carol/viewer）并编写Vault策略。

2b。配置GCP机密引擎（Terraform）

GCP secrets引擎是通过Terraform配置的，Terraform创建GCP服务帐户，授予其必要的IAM角色，并配置Vault——磁盘上没有服务帐户密钥文件。

先决条件：

  • 地形 >= 1.5
  • gcloud CLI已通过身份验证： gcloud auth application-default login
cd terraform
cp terraform.tfvars.example terraform.tfvars
# Edit terraform.tfvars — set gcp_project_id to your GCP project
terraform init
terraform apply

您还必须在中设置相同的GCP项目ID config/settings.yaml 以便MCP数据服务器可以将其传递给GCS和BigQuery客户端：

# config/settings.yaml
gcp:
  project_id: "your-gcp-project-id"   # must match terraform.tfvars
  region: "us-central1"

为什么需要这个？ 数据代理使用Vault中的OAuth2访问令牌 而不是服务帐户密钥文件。与基于密钥的凭据不同，OAuth2 令牌不携带项目元数据，因此GCP客户端库无法 自动推断项目。项目ID在 settings.yaml 是 通过了 IdentityContext 每次GCS和BigQuery客户端调用。

这将创建：

  • Vault本身的GCP服务帐户(vault-gcp-secrets@<project>.iam.gserviceaccount.com)
  • IAM绑定 serviceAccountAdmin, serviceAccountKeyAdmin, serviceAccountTokenCreator，以及 projectIamAdmin
  • 每个代理都有专门的GCP服务帐户(data-agent-gcp, compute-agent-gcp)各自的IAM角色
  • Vault GCP机密引擎，配备两个 冒充账户 (data-agent-gcp 和 compute-agent-gcp)，每个配置有 5分钟令牌TTL (ttl = "300")

拆卸： terraform destroy。参见 terraform/README.md 了解全部细节。

3.创建并激活虚拟环境

python3.13 -m venv .venv
source .venv/bin/activate      # Linux / macOS
# .venv\Scripts\activate       # Windows

您必须在每次新的shell会话中激活虚拟环境，然后 运行或开发项目：

source .venv/bin/activate

4.安装

pip install -e ".[dev]"

5.设置LLM API密钥

export ANTHROPIC_API_KEY=sk-ant-...
# or
export OPENAI_API_KEY=sk-...

6.配置config/settings.yaml文件

将您的GCP项目id添加到config/settings.yaml文件中，这是该文件底部的摘录 其去向如下：

# GCP project default
gcp:
  project_id: "<Your GCP project id goes here>"
  region: "us-central1"

7.跑步

vault-mcp-agents
# or
python -m vault_mcp_agents.main --verbose

系统将提示您登录，选择一个代理，然后用自然语言与之交互。

安装脚本(scripts/setup_vault.sh)创建三个具有不同访问级别的预配置用户：

用户名 密码 角色 访问级别
alice alice-pass 操作员 对所有GCP工具的完全读/写/删除访问权限
bob bob-pass 分析员 只读GCS+BigQuery，计算能力有限
carol carol-pass 查看器 最小化只读数据访问

8.运行测试

pytest -v

策略引擎、会话和身份上下文的测试在没有Vault或GCP的情况下运行。

测试5分钟凭证租赁

如何验证

有三个级别的验证，从快速单元测试到完整的端到端证明。

级别1：单元测试（无需基础设施）

测试套件验证策略引擎是否能够解决 "5m" 对于每个角色：

pytest tests/test_policy_engine.py -v -k "five_minute"

这运行 test_all_roles_get_five_minute_ttl，它断言操作员、分析师和观众都会收到 max_gcp_token_ttl == "5m".

级别2:Vault CLI（需要运行Vault+GCP secrets引擎）

直接从Vault读取令牌并检查报告的TTL：

export VAULT_ADDR=http://127.0.0.1:8200
export VAULT_TOKEN=dev-root-token

# Request a token from the data-agent impersonated account
vault read gcp/impersonated-account/data-agent-gcp/token

该回复包括 token_ttl 现场。在5分钟配置下，该值将为 300 （秒）或更短。

您还可以确认模拟的帐户配置：

vault read gcp/impersonated-account/data-agent-gcp

这 ttl 字段应显示 5m （或 300s).

第三级：端到端验证（需要Vault+GCP项目）

此测试通过Vault获得真实的GCP令牌，立即使用它，等待它过期，并确认GCP拒绝过时的令牌：

source .venv/bin/activate
python - <<'PYEOF'
import subprocess
import time
import hvac
from google.oauth2.credentials import Credentials
from google.cloud import storage

# --- Step 0: Get GCP project ID from gcloud ---
project_id = subprocess.run(
    ["gcloud", "config", "get-value", "project"],
    capture_output=True, text=True, check=True,
).stdout.strip()
print(f"GCP project: {project_id}")

# --- Step 1: Get a short-lived token from Vault ---
client = hvac.Client(url="http://127.0.0.1:8200", token="dev-root-token")
resp = client.secrets.gcp.generate_impersonated_account_oauth2_access_token(
    name="data-agent-gcp", mount_point="gcp"
)
token = resp["data"]["token"]
ttl   = resp["data"].get("token_ttl", "unknown")
print(f"Vault issued token with TTL: {ttl}s")
assert int(ttl) <= 300, f"TTL {ttl}s exceeds 5-minute ceiling!"

# --- Step 2: Use the token immediately (should succeed) ---
creds = Credentials(token=token)
sc = storage.Client(credentials=creds, project=project_id)
buckets = [b.name for b in sc.list_buckets()]
print(f"T+0s:   SUCCESS — found {len(buckets)} bucket(s)")

# --- Step 3: Wait past expiry ---
wait = int(ttl) + 10
print(f"Waiting {wait}s for token to expire...")
time.sleep(wait)

# --- Step 4: Retry with the same token (should fail) ---
try:
    sc2 = storage.Client(credentials=creds, project=project_id)
    list(sc2.list_buckets())
    print("T+expired: UNEXPECTED SUCCESS — token should have expired")
except Exception as e:
    print(f"T+expired: EXPECTED FAILURE — {type(e).__name__}: {e}")
PYEOF

成功的测试运行如下：

GCP project: my-gcp-project
Vault issued token with TTL: 300s
T+0s:   SUCCESS — found 3 bucket(s)
Waiting 310s for token to expire...
T+expired: EXPECTED FAILURE — Forbidden: ...

项目结构

vault-mcp-agents/
├── config/
│   └── settings.yaml              # Vault, agent, MCP, LLM configuration
├── policies/
│   └── capabilities.yaml          # (role, agent) → allowed tools
├── scripts/
│   └── setup_vault.sh             # Vault dev provisioning (auth, policies, users)
├── terraform/
│   ├── main.tf                    # GCP SAs + Vault GCP secrets engine + impersonated accounts
│   ├── variables.tf               # Input variables (project ID, region, etc.)
│   ├── outputs.tf                 # Impersonated account names, SA email
│   ├── versions.tf                # Provider version constraints
│   ├── terraform.tfvars.example   # Template for local variable values
│   └── README.md                  # Terraform-specific usage guide
├── src/vault_mcp_agents/
│   ├── main.py                    # CLI entry point
│   ├── auth/
│   │   ├── vault_authenticator.py # Human login via Vault
│   │   └── session.py             # Immutable session context
│   ├── vault/
│   │   └── gcp_credentials.py     # GCP token retrieval from Vault
│   ├── policy/
│   │   └── engine.py              # YAML policy resolution
│   ├── mcp/
│   │   ├── identity_context.py    # Composite identity (agent + human)
│   │   ├── base_server.py         # Base MCP server with tool filtering
│   │   ├── data_server.py         # GCS + BigQuery MCP tools
│   │   └── compute_server.py      # GCE Compute MCP tools
│   ├── agents/
│   │   ├── factory.py             # Builds LangChain agents from config
│   │   └── mcp_langchain_adapter.py # MCP tools → LangChain tools
│   └── prompt/
│       └── cli.py                 # Interactive login + conversation UI
├── tests/
│   ├── conftest.py                # Shared fixtures
│   ├── test_policy_engine.py      # Policy resolution tests
│   ├── test_session.py            # Session behaviour tests
│   ├── test_identity_context.py   # Serialisation round-trip tests
│   └── test_capability_filter.py  # End-to-end capability filtering
├── docker-compose.yaml
└── pyproject.toml

定制

添加新的MCP工具

  1. 将处理程序方法添加到相关服务器(data_server.py 或 compute_server.py).
  2. 在中注册 _register_all_tools().
  3. 将工具名称添加到中的相应角色 policies/capabilities.yaml.
  4. 策略引擎的测试将捕获任何缺失的条目。

添加新角色

  1. 创建Vault策略（例如。 auditor-policy).
  2. 在中添加角色映射 vault_authenticator.py → _POLICY_TO_ROLE.
  3. 添加a auditor: 阻止到 policies/capabilities.yaml 每个代理的工具列表。

切换身份验证方法

改变 vault.auth_method 在 settings.yaml 到 ldap 或 oidcThe VaultAuthenticator 代表相应 hvac 身份验证后端。

故障排除

ModuleNotFoundError: No module named 'vault_mcp_agents'

运行时发生此错误 python -m vault_mcp_agents.main 没有 活动虚拟环境或未先安装包。

修复： 创建并激活虚拟环境，然后安装软件包：

python3.13 -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"

如果你已经有了 .venv 目录，你只需要激活它：

source .venv/bin/activate

为什么需要venv？ macOS上由Homebrew管理的Python 外部管理 和块 系统范围的 pip install虚拟环境将项目的 依赖关系并使 vault_mcp_agents 包装可进口。

许可证

麻省理工学院

相关 MCP