AI Agent 身份与 PKI:自主 AI 时代的机器身份治理
2026 年的身份危机:当 AI Agent 有了"手和脚"
2026 年 4 月,CloudSecurity Alliance 发布了一份白皮书《The Non-Human Identity Governance Vacuum》,揭示了一个令人不安的现实:88% 的企业已经在生产环境中使用 AI Agent,但只有 12% 有完善的身份治理策略。
这不是一个理论问题。2026 年初,安全研究人员发现多起 AI Agent 被恶意利用的案例:
- Shadow Agent 问题:企业员工在未经 IT 部门批准的情况下,将 AI Agent 接入内部系统,这些 Agent 使用不受控的 API Key 访问敏感数据
- 身份冒充:恶意 Agent 伪造服务身份,绕过传统的基于 IP 的信任机制
- 权限蔓延:一个 Agent 从"只读查询"开始,通过自动扩展权限,最终获得了生产环境的完全控制权
我们需要一个新的身份治理范式。
从人类身份到机器身份:PKI 的角色演进
传统 PKI 的设计假设
传统 PKI 体系(X.509)建立在一系列隐含假设之上:
- 主体是人或组织:证书 Subject 通常是域名、邮箱或组织名称
- 身份相对稳定:一张证书有效期 1-2 年,主体身份在此期间不变
- 申请流程有人工参与:CSR 生成 → CA 审核 → 签发,每一步都有人类审批
- 信任基于预先建立的根:信任锚是预先植入浏览器/操作系统的根 CA
AI Agent 的身份需求
一个 AI Agent 的身份系统需要回答三个核心问题:
| 问题 | 传统 PKI 的回答 | AI Agent 需要的回答 |
|---|---|---|
| 你是谁? | 域名、组织名 | 这个 Agent 属于哪个团队、执行什么任务 |
| 你能做什么? | 基本约束(CA:TRUE 等) | 精确的权限范围:能访问哪些 API、能操作哪些资源 |
| 你从哪里来? | 签发 CA 链 | 运行在哪个沙箱/容器/K8s 命名空间 |
SPIFFE:为机器身份而生的标准
SPIFFE(Secure Production Identity Framework for Everyone)是云原生计算基金会(CNCF)托管的项目,专门为解决机器身份问题而设计。2026 年,SPIRE(SPIFFE 的生产级实现)已经在 Google、Uber、Bloomberg 等公司的生产环境中运行,管理着数百万个工作负载的身份。
SPIFFE 的核心概念:
SPIFFE ID: spiffe://trust-domain/ns/default/sa/my-agent
│ │ │ │
│ │ │ └─ Kubernetes ServiceAccount
│ │ └─────── 命名空间
│ └────────────── 服务账户
└───────────────────────────── 信任域SPIFFE vs X.509 的映射关系:
| SPIFFE 概念 | X.509 对应 | 说明 |
|---|---|---|
| SPIFFE ID | URI Subject Alternative Name | 唯一标识 |
| Trust Domain | Root CA | 信任根 |
| Workload | 证书持有者 | 运行中的进程 |
| SVID | X.509 证书 | 身份凭证 |
| Bundle | CA 证书包 | 信任锚分发 |
架构设计:三层机器身份治理模型
基于当前最佳实践,一个完整的 AI Agent 身份治理体系应该包含三层:
┌─────────────────────────────────────────────────────┐
│ 第三层:信任治理 │
│ 策略引擎 → 审计日志 → 实时吊销 → 合规报告 │
├─────────────────────────────────────────────────────┤
│ 第二层:身份签发 │
│ SPIRE Server → CA 桥接 → 证书签发 → 自动轮换 │
├─────────────────────────────────────────────────────┤
│ 第一层:身份验证 │
│ mTLS 握手 → SPIFFE ID 校验 → 权限映射 → 访问控制 │
└─────────────────────────────────────────────────────┘第一层:双向认证(mTLS)
所有 Agent 之间的通信必须使用 mTLS,确保双方的身份都可验证。
传统 TLS(单向):
Client ──验证 Server 证书──→ Server
mTLS(双向):
Client ←──验证 Server 证书──→ Server
↑ ↑
└──验证 Client 证书──────────┘在 AI Agent 场景中,"Client"和"Server"都是 Agent,双向认证确保:
- 调用方只能访问它有权访问的服务
- 被调用方只能接受来自合法 Agent 的请求
- 任何中间人攻击都会被检测到
第二层:自动身份签发与轮换
Agent 的身份凭证必须是短期的(几小时到几天),并且自动轮换。
为什么不能用长期证书:
- Agent 被攻破时,长期证书意味着长期风险
- Agent 完成任务后应该立即失去访问权限
- 短期凭证天然限制了"权限蔓延"的时间窗口
┌──────────┐ ┌──────────┐ ┌──────────┐
│ Agent │────→│ SPIRE │────→│ CA │
│ (Workload)│ │ Agent │ │ (Issuer) │
└──────────┘ └──────────┘ └──────────┘
│ │ │
│ 1. 启动时请求 │ │
│ SVID │ │
│────────────────→│ │
│ │ 2. 验证 Agent │
│ │ 身份 (K8s SA, │
│ │ Attestor) │
│ │────────────────→│
│ │ │ 3. 签发
│ │ │ X.509-SVID
│ │←────────────────│
│ 4. 返回 SVID │ │
│ (有效期: 1h) │ │
│←────────────────│ │
│ │ │
│ 5. 50分钟后 │ │
│ 自动续期 │ │
│────────────────→│ │第三层:策略与审计
仅有身份验证不够,还需要:
- 策略引擎:基于 SPIFFE ID 的细粒度访问控制
- 审计日志:每次认证的完整记录
- 实时吊销:发现异常时立即撤销凭证
工程实践:从零搭建 Agent 身份治理
下面我们在一个 Kubernetes 环境中,用 Python 实现一个简化的 Agent 身份治理系统。
环境准备
# 需要安装
pip install cryptography==48.0.0 gmssl>=3.0.0关于国密算法的选择:当前版本的cryptography(48.0)尚未原生支持 SM2 密钥生成,但支持 SM3 哈希。本文示例使用gmssl库处理 SM2 运算,使用cryptography处理 X.509 证书结构。在生产环境中,推荐使用国密加密机或 HSM 进行密钥生成和签名。
Step 1:构建内部 CA
首先,我们需要一个内部 CA 来签发 Agent 的证书:
# internal_ca.py - 内部 CA 实现
import datetime
from pathlib import Path
from cryptography import x509
from cryptography.x509.oid import NameOID
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import ec
CA_DIR = Path("code/agent-pki")
def generate_ca():
"""生成自签名根 CA 证书"""
CA_DIR.mkdir(parents=True, exist_ok=True)
# 生成 ECDSA 密钥对(P-256,生产环境应使用 SM2)
private_key = ec.generate_private_key(ec.SECP256R1())
subject = issuer = x509.Name([
x509.NameAttribute(NameOID.COUNTRY_NAME, "CN"),
x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, "Beijing"),
x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Agent PKI Demo CA"),
x509.NameAttribute(NameOID.COMMON_NAME, "Agent PKI Root CA"),
])
cert = (
x509.CertificateBuilder()
.subject_name(subject)
.issuer_name(issuer)
.public_key(private_key.public_key())
.serial_number(x509.random_serial_number())
.not_valid_before(datetime.datetime.utcnow())
.not_valid_after(datetime.datetime.utcnow() + datetime.timedelta(days=3650))
.add_extension(
x509.BasicConstraints(ca=True, path_length=None),
critical=True,
)
.add_extension(
x509.KeyUsage(
digital_signature=True,
key_cert_sign=True,
crl_sign=True,
content_commitment=False,
key_encipherment=False,
data_encipherment=False,
key_agreement=False,
encipher_only=False,
decipher_only=False,
),
critical=True,
)
.sign(private_key, hashes.SHA256())
)
with open(CA_DIR / "ca-key.pem", "wb") as f:
f.write(private_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.BestAvailableEncryption(b"ca-password")
))
with open(CA_DIR / "ca-cert.pem", "wb") as f:
f.write(cert.public_bytes(serialization.Encoding.PEM))
print(f"[CA] 根 CA 证书已生成: {CA_DIR / 'ca-cert.pem'}")
return private_key, cert
if __name__ == "__main__":
generate_ca()Step 2:为 Agent 签发身份证书
Agent 证书与传统服务器证书的关键区别在于 Subject Alternative Name (SAN) 的设计。我们用 URI 格式的 SAN 来表示 Agent 的 SPIFFE ID:
# agent_cert.py - Agent 身份证书签发
import datetime
from pathlib import Path
from cryptography import x509
from cryptography.x509.oid import NameOID
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import ec
CA_DIR = Path("code/agent-pki")
def load_ca():
"""加载 CA 私钥和证书"""
from cryptography.hazmat.primitives.serialization import load_pem_private_key
with open(CA_DIR / "ca-key.pem", "rb") as f:
key = load_pem_private_key(f.read(), password=b"ca-password")
with open(CA_DIR / "ca-cert.pem", "rb") as f:
cert = x509.load_pem_x509_certificate(f.read())
return key, cert
def issue_agent_cert(
agent_name: str,
trust_domain: str = "demo.internal",
namespace: str = "default",
service_account: str = "default",
valid_hours: int = 24,
allowed_apis: list = None,
):
"""为 Agent 签发身份证书"""
ca_key, ca_cert = load_ca()
# 生成 Agent 密钥对
agent_key = ec.generate_private_key(ec.SECP256R1())
# 构建 SPIFFE ID
spiffe_id = f"spiffe://{trust_domain}/ns/{namespace}/sa/{service_account}"
agent_id = f"agent-{agent_name}-{datetime.datetime.utcnow().strftime('%Y%m%d')}"
subject = x509.Name([
x509.NameAttribute(NameOID.COUNTRY_NAME, "CN"),
x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Agent PKI Demo"),
x509.NameAttribute(NameOID.COMMON_NAME, agent_name),
x509.NameAttribute(NameOID.SERIAL_NUMBER, agent_id),
])
# SAN: SPIFFE ID + DNS 名称
dns_name = f"{agent_name}.{namespace}.svc.cluster.local"
san_names = [
x509.UniformResourceIdentifier(spiffe_id),
x509.DNSName(dns_name),
]
# 自定义扩展:权限声明( OID 1.3.6.1.4.1.99999.1)
perms = ",".join(allowed_apis).encode() if allowed_apis else b"*"
perm_ext = x509.UnrecognizedExtension(
x509.ObjectIdentifier("1.3.6.1.4.1.99999.1"), perms,
)
now = datetime.datetime.utcnow()
cert = (
x509.CertificateBuilder()
.subject_name(subject)
.issuer_name(ca_cert.subject)
.public_key(agent_key.public_key())
.serial_number(x509.random_serial_number())
.not_valid_before(now)
.not_valid_after(now + datetime.timedelta(hours=valid_hours))
.add_extension(x509.SubjectAlternativeName(san_names), critical=False)
.add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
.add_extension(x509.ExtendedKeyUsage([
x509.oid.ExtendedKeyUsageOID.SERVER_AUTH,
x509.oid.ExtendedKeyUsageOID.CLIENT_AUTH,
]), critical=False)
.add_extension(perm_ext, critical=False)
.sign(ca_key, hashes.SHA256())
)
output_dir = CA_DIR / "agents" / agent_name
output_dir.mkdir(parents=True, exist_ok=True)
key_path = output_dir / "agent-key.pem"
cert_path = output_dir / "agent-cert.pem"
with open(key_path, "wb") as f:
f.write(agent_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption(),
))
with open(cert_path, "wb") as f:
f.write(cert.public_bytes(serialization.Encoding.PEM))
print(f"[Agent] 证书已签发: {cert_path}")
print(f" SPIFFE ID: {spiffe_id}")
print(f" 有效期: {valid_hours}h")
print(f" 允许 API: {allowed_apis or '全部'}")
return str(key_path), str(cert_path)
if __name__ == "__main__":
issue_agent_cert(
agent_name="data-query",
namespace="analytics",
service_account="data-query-sa",
valid_hours=8,
allowed_apis=["GET /api/v1/data", "POST /api/v1/query"],
)Step 3:mTLS 服务端实现
# mtls_server.py - mTLS 服务端(身份验证 + 权限控制)
import ssl
import json
import http.server
import socketserver
from pathlib import Path
from cryptography import x509
from cryptography.x509.oid import NameOID
CA_DIR = Path("code/agent-pki")
PERMISSIONS_OID = x509.ObjectIdentifier("1.3.6.1.4.1.99999.1")
def extract_spiffe_id(cert):
"""从证书 SAN 中提取 SPIFFE ID"""
try:
san = cert.extensions.get_extension_for_class(x509.SubjectAlternativeName)
for name in san.value:
if isinstance(name, x509.UniformResourceIdentifier):
if name.value.startswith("spiffe://"):
return name.value
except x509.ExtensionNotFound:
pass
return None
def extract_permissions(cert):
"""从证书扩展中提取权限列表"""
try:
ext = cert.extensions.get_extension_for_oid(PERMISSIONS_OID)
perms = ext.value.value.decode()
return ["*"] if perms == "*" else perms.split(",")
except x509.ExtensionNotFound:
return []
class AgentAuthorizedHandler(http.server.BaseHTTPRequestHandler):
def do_GET(self):
peer_cert = self.connection.getpeercert(binary_form=True)
if not peer_cert:
self.send_error(401, "缺少客户端证书")
return
cert = x509.load_der_x509_certificate(peer_cert)
spiffe_id = extract_spiffe_id(cert)
permissions = extract_permissions(cert)
if not spiffe_id:
self.send_error(403, "证书缺少 SPIFFE ID")
return
requested = f"GET {self.path}"
if permissions != ["*"] and requested not in permissions:
self.send_error(403,
f"Agent {spiffe_id} 无权访问 {requested}\n"
f"允许的权限: {permissions}")
return
response = {
"status": "authorized",
"agent": {
"spiffe_id": spiffe_id,
"permissions": permissions,
"cert_subject": cert.subject.get_attributes_for_oid(
NameOID.COMMON_NAME)[0].value,
"expires": cert.not_valid_after.isoformat(),
},
"resource": self.path,
}
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps(response, indent=2, ensure_ascii=False).encode())
def log_message(self, format, *args):
pass # 生产环境使用结构化日志
def run_server(host="0.0.0.0", port=8443):
ca_cert_path = str(CA_DIR / "ca-cert.pem")
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
ssl_context.verify_mode = ssl.CERT_REQUIRED
ssl_context.load_verify_locations(ca_cert_path)
ssl_context.load_cert_chain(
certfile=str(CA_DIR / "agents" / "server" / "agent-cert.pem"),
keyfile=str(CA_DIR / "agents" / "server" / "agent-key.pem"),
)
server = socketserver.TCPServer((host, port), AgentAuthorizedHandler)
server.socket = ssl_context.wrap_socket(server.socket, server_side=True)
print(f"[Server] mTLS 服务端运行在 https://{host}:{port}")
server.serve_forever()
if __name__ == "__main__":
run_server()Step 4:mTLS 客户端实现
# mtls_client.py - mTLS 客户端(Agent 端)
import ssl
import json
import urllib.request
from pathlib import Path
CA_DIR = Path("code/agent-pki")
def make_authenticated_request(url, agent_name, namespace="default"):
"""以 Agent 身份发起 mTLS 请求"""
agent_dir = CA_DIR / "agents" / agent_name
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ssl_context.verify_mode = ssl.CERT_REQUIRED
ssl_context.load_verify_locations(str(CA_DIR / "ca-cert.pem"))
ssl_context.load_cert_chain(
certfile=str(agent_dir / "agent-cert.pem"),
keyfile=str(agent_dir / "agent-key.pem"),
)
req = urllib.request.Request(url)
try:
with urllib.request.urlopen(req, context=ssl_context) as resp:
return json.loads(resp.read())
except urllib.error.HTTPError as e:
return {"error": f"HTTP {e.code}", "detail": e.read().decode()}
if __name__ == "__main__":
result = make_authenticated_request(
url="https://localhost:8443/api/v1/data",
agent_name="data-query",
namespace="analytics",
)
print(json.dumps(result, indent=2, ensure_ascii=False))Step 5:验证完整流程
# 1. 生成 CA
python3 internal_ca.py
# 2. 为服务端和 Agent 签发证书
python3 -c "
from agent_cert import issue_agent_cert
issue_agent_cert('server', valid_hours=720)
issue_agent_cert('data-query', namespace='analytics',
service_account='data-query-sa', valid_hours=8,
allowed_apis=['GET /api/v1/data', 'POST /api/v1/query'])
"
# 3. 启动服务端(终端 1)
python3 mtls_server.py
# 4. Agent 发起请求(终端 2)
python3 mtls_client.py预期输出:
{
"status": "authorized",
"agent": {
"spiffe_id": "spiffe://demo.internal/ns/analytics/sa/data-query-sa",
"permissions": ["GET /api/v1/data", "POST /api/v1/query"],
"cert_subject": "data-query",
"expires": "2026-06-13T09:00:00"
},
"resource": "/api/v1/data"
}踩坑记录:生产环境的 5 个坑
坑 1:证书有效期与 Agent 生命周期的匹配
现象:Agent 证书过期后,正在执行的任务突然中断。
原因:证书有效期设为 24 小时,但 Agent 的批处理任务可能运行 48 小时。
解决方案:使用证书窗口机制——在新证书签发后,旧证书保留 2 小时宽限期:
def should_rotate_cert(cert_path, rotation_threshold=0.8):
"""检查证书是否需要轮换"""
with open(cert_path, "rb") as f:
cert = x509.load_pem_x509_certificate(f.read())
remaining = cert.not_valid_after - datetime.datetime.utcnow()
total = cert.not_valid_after - cert.not_valid_before
return (remaining.total_seconds() / total.total_seconds()) < (1 - rotation_threshold)坑 2:CA 根密钥的安全存储
现象:CA 私钥以明文文件形式存储在磁盘上。
解决方案:生产环境中必须使用 HSM(硬件安全模块) 或 KMS(密钥管理服务):
| 方案 | 适用场景 | 国密合规 |
|---|---|---|
| 本地 HSM(如 SafeNet) | 传统数据中心 | 支持 SM2 |
| 云 KMS(如阿里云 KMS) | 云上环境 | 支持 SM2/SM4 |
| HashiCorp Vault | 混合云 | 需配置 SM2 后端 |
| 国密加密机 | 高安全场景 | 完全合规 |
坑 3:SPIFFE ID 与权限的耦合
现象:每次修改 Agent 权限都需要重新签发证书。
解决方案:将 SPIFFE ID 与权限解耦——证书只包含 SPIFFE ID,权限由策略引擎动态查询:
证书(身份): spiffe://demo.internal/ns/analytics/sa/data-query-sa
│
▼
策略引擎(权限): 查询 → 返回当前允许的 API 列表
│
▼
实时生效: 下次请求自动使用最新权限坑 4:国密算法的兼容性
现象:部分 Python 库不支持 SM2/SM3 算法。
解决方案:使用 gmssl 库处理 SM2 运算:
from gmssl import sm2, sm3, func
# SM2 密钥生成
crypt_sm2 = sm2.CryptSM2(public_key="", private_key="")
# SM3 哈希
hash_value = sm3.sm3_hash(func.bytes_to_list(b"Hello, SM3"))
print(f"SM3 hash: {hash_value}")
# 输出: 21b937fed61e685b8ac08c67fe9a3300437f2ca44547dea06e0cfe30219fdc4c注意:cryptography 48.0 版本尚未原生支持 SM2 密钥生成,但支持 SM3 哈希算法。在生产环境中,推荐使用国密加密机或 HSM 进行 SM2 密钥生成和签名操作。
坑 5:证书吊销的实时性
现象:Agent 被攻破后,其证书在过期前仍然有效。
解决方案:对于短期证书(几小时有效期),不需要吊销机制——让证书自然过期即可。对于长期证书,使用 OCSP Stapling 或短效 CRL。
短期证书策略:
证书有效期 = Agent 任务最大执行时间 + 缓冲(如 2h)
不需要吊销,自然过期即可
长期证书策略:
使用 OCSP 响应器或 CRL 分发点
配合实时威胁检测系统行业趋势:2026 年的机器身份管理
趋势 1:从 API Key 到 X.509 证书
2026 年,越来越多的企业正在从"API Key 认证"迁移到"基于 X.509 证书的 mTLS 认证"。
| 维度 | API Key | X.509 证书 |
|---|---|---|
| 安全性 | 低(静态密钥) | 高(非对称加密) |
| 有效期 | 长期(手动轮换) | 短期(自动轮换) |
| 身份绑定 | 弱(谁拿到谁用) | 强(绑定到进程/容器) |
| 审计 | 困难 | 完整(证书链可追溯) |
| 国密支持 | 不适用 | 支持 SM2/SM3 |
趋势 2:SPIFFE 成为云原生身份标准
2026 年,SPIFFE/SPIRE 已经被以下组织采用:
- Google:BeyondCorp 零信任架构的基础
- Uber:管理数百万微服务身份
- Bloomberg:金融交易系统的机器身份
- Pinterest:从 API Key 迁移到 SPIFFE
趋势 3:AI Agent 身份成为新的安全边界
根据 CyberArk 2026 年的预测,AI Agent 将成为 2026 年最大的非人类身份风险。关键挑战包括:
- Agent 的"最小权限"如何定义
- Agent 之间的信任链如何建立
- Agent 的行为审计如何实现
总结
AI Agent 的身份治理不是 PKI 的"新应用场景",而是 PKI 体系的范式升级。
核心要点:
- 身份标识从"域名"到"SPIFFE ID":用 URI 格式唯一标识每个 Agent
- 证书有效期从"年"到"小时":短期凭证 + 自动轮换 = 最小权限的时间维度
- 权限管理从"硬编码"到"动态策略":证书只负责身份,策略引擎负责权限
- 信任根从"公共 CA"到"内部 CA":机器身份不需要公共信任,内部 CA 足够
- 国密算法是必选项:在涉及国内业务的场景中,SM2/SM3 提供合规保障
- 评估当前 AI Agent 的身份管理现状(API Key?mTLS?无认证?)
- 在测试环境部署 SPIRE,体验自动身份签发
- 从最关键的服务开始,逐步迁移到 mTLS
- 建立 Agent 身份的生命周期管理流程(创建 → 授权 → 轮换 → 销毁)
参考来源
- CloudSecurity Alliance: The Non-Human Identity Governance Vacuum (2026)
- CyberArk: AI Agents and Identity Risks in 2026
- SPIFFE 官方文档
- NIST SP 800-204C: Implementing Zero Trust for Non-Human Identities
- AppViewX: PQC Readiness in 2026
- Keyfactor: Securing Agentic AI
- GM/T 0054-2018 信息系统密码应用基本要求
- RFC 8998: SM2 算法在 TLS 1.3 中的应用