Install
openclaw skills install @mirni/greenhelix-nhi-security-agentsNon-Human Identity Security for AI Agents
Non-Human Identity Security for AI Agents. Complete guide to securing non-human identities in AI agent deployments. Covers NHI lifecycle management, credential rotation, workload identity federation, machine-to-machine authentication patterns, Microsoft Agent Governance Toolkit integration, zero-trust for agents, and compliance frameworks for the $7.4B NHI market.
Non-Human Identity Security for AI Agents
Notice: This is an educational guide with illustrative code examples. It does not execute code or install dependencies. All examples use the GreenHelix sandbox (https://sandbox.greenhelix.net) which provides 500 free credits — no API key required to get started.
Referenced credentials (you supply these in your own environment):
AGENT_SIGNING_KEY: Cryptographic signing key for agent identity (Ed25519 key pair for request signing)STRIPE_API_KEY: Stripe API key for card payment processing (scoped to payment intents only)
Your AI agent fleet has 47 service accounts, 312 API keys, 89 OAuth client credentials, and 24 machine certificates. You know this because you just counted them. Before today, nobody had. The agent that processes customer refunds uses the same API key as the agent that reads public price feeds. The certificate for your payment orchestrator expired 11 days ago, but the system kept working because the downstream service disabled TLS verification during a "temporary" debugging session in March. Three of your OAuth client secrets are stored in environment variables that are visible to every container in the Kubernetes namespace. One of those secrets has not been rotated since it was created 14 months ago. You are not unusual. You are average. The 2025 Entro Security State of Non-Human Identities report found that organizations have 40 to 100 machine identities for every human identity, and 68% of security incidents in the past year involved a compromised machine identity. Gartner named non-human identity security the number one cybersecurity trend for 2025-2026 and projected the NHI market at $7.4 billion by 2027. The attackers already know this. The Midnight Blizzard breach at Microsoft started with a compromised OAuth application. The Cloudflare breach started with a service token. The CircleCI breach started with a stolen machine credential. Every major breach in 2024-2025 that did not start with phishing started with a non-human identity. This guide gives you the complete NHI security architecture for AI agent deployments: lifecycle management, credential rotation, workload identity federation, machine-to-machine authentication, governance tooling, zero trust, and compliance. Every pattern includes production-ready Python code that you can deploy today.
What You'll Learn
- Chapter 1: The NHI Crisis -- Why Machine Identities Are the #1 Attack Vector
- Chapter 2: NHI Lifecycle Management
- Chapter 3: Workload Identity Federation for AI Agents
- Chapter 4: Machine-to-Machine Authentication Patterns
- Chapter 5: Microsoft Agent Governance Toolkit Integration
- Chapter 6: Zero Trust for AI Agents
- Chapter 7: NHI Compliance Frameworks
- Next Steps
Full Guide
Non-Human Identity Security for AI Agents
Your AI agent fleet has 47 service accounts, 312 API keys, 89 OAuth client credentials, and 24 machine certificates. You know this because you just counted them. Before today, nobody had. The agent that processes customer refunds uses the same API key as the agent that reads public price feeds. The certificate for your payment orchestrator expired 11 days ago, but the system kept working because the downstream service disabled TLS verification during a "temporary" debugging session in March. Three of your OAuth client secrets are stored in environment variables that are visible to every container in the Kubernetes namespace. One of those secrets has not been rotated since it was created 14 months ago. You are not unusual. You are average. The 2025 Entro Security State of Non-Human Identities report found that organizations have 40 to 100 machine identities for every human identity, and 68% of security incidents in the past year involved a compromised machine identity. Gartner named non-human identity security the number one cybersecurity trend for 2025-2026 and projected the NHI market at $7.4 billion by 2027. The attackers already know this. The Midnight Blizzard breach at Microsoft started with a compromised OAuth application. The Cloudflare breach started with a service token. The CircleCI breach started with a stolen machine credential. Every major breach in 2024-2025 that did not start with phishing started with a non-human identity. This guide gives you the complete NHI security architecture for AI agent deployments: lifecycle management, credential rotation, workload identity federation, machine-to-machine authentication, governance tooling, zero trust, and compliance. Every pattern includes production-ready Python code that you can deploy today.
Table of Contents
- The NHI Crisis -- Why Machine Identities Are the #1 Attack Vector
- NHI Lifecycle Management
- Workload Identity Federation for AI Agents
- Machine-to-Machine Authentication Patterns
- Microsoft Agent Governance Toolkit Integration
- Zero Trust for AI Agents
- NHI Compliance Frameworks
Chapter 1: The NHI Crisis -- Why Machine Identities Are the #1 Attack Vector
The Numbers That Should Scare You
The ratio of machine identities to human identities in a typical enterprise is between 40:1 and 100:1. In organizations running AI agent fleets, this ratio is higher -- often 200:1 or more, because each agent may hold multiple credentials across different services. A single orchestrator agent might have an API key for the commerce gateway, an OAuth client credential for the payment processor, a service account for the cloud provider, a TLS certificate for mTLS with downstream agents, and a signing key for identity verification. That is five machine identities for one logical agent.
The 2025 CyberArk Machine Identity Security report found that 68% of security incidents in the surveyed organizations involved a compromised machine identity. Not a phished human. Not a brute-forced password. A machine credential that was over-privileged, never rotated, stored in plaintext, or shared across services that should have been isolated.
The breakdown of how machine identities get compromised:
| Attack Vector | Percentage | Example |
|---|---|---|
| Hardcoded credentials in code/config | 34% | API key committed to git, never rotated |
| Over-privileged service accounts | 22% | Read-only agent with write permissions |
| Expired but still-valid certificates | 16% | TLS cert expired, verification disabled as workaround |
| Secret sprawl across environments | 14% | Same key used in dev, staging, and production |
| Lack of rotation | 9% | OAuth client secret unchanged for 12+ months |
| Stolen from CI/CD pipelines | 5% | Secrets exposed in build logs or artifacts |
Every one of these vectors is amplified in AI agent systems because agents create, consume, and propagate credentials autonomously.
Recent Breaches Caused by Machine Identity Compromise
Midnight Blizzard / Microsoft (January 2024). Russian state actor Midnight Blizzard compromised a legacy test OAuth application that had been granted elevated permissions to the Microsoft corporate environment. The application had not been subject to the same security controls as production service accounts. The attackers used the OAuth app's permissions to access Microsoft corporate email accounts, including those of senior leadership and cybersecurity staff. The initial access vector was not phishing, not a zero-day, not a supply chain attack. It was an OAuth application -- a non-human identity -- with excessive permissions and no monitoring.
Cloudflare (October 2023). Attackers used a service token and three service account credentials that were stolen during the Okta breach. Cloudflare had rotated most credentials after the Okta incident, but these four were missed. The stolen tokens provided access to Cloudflare's Atlassian environment, which led to access to source code and internal documentation. The root cause was incomplete credential rotation -- an NHI lifecycle management failure.
CircleCI (January 2023). An engineer's laptop was compromised with malware that stole a session token. That session token provided access to CircleCI's production environment, which contained customer secrets -- API keys, OAuth tokens, and service account credentials for thousands of organizations. CircleCI had to advise every customer to rotate every secret stored in their platform. The blast radius was enormous because machine identities were stored centrally without per-customer isolation.
Step Finance (2024). A compromised agent identity in the DeFi protocol led to unauthorized fund transfers totaling $40M. The agent had been granted broad financial permissions for operational convenience. When the identity was compromised, the attacker inherited all of those permissions. No behavioral monitoring detected the anomalous transactions because the agent's normal behavior already included large transfers.
Human IAM vs. NHI Management
The fundamental problem is that organizations are managing machine identities with tools and processes designed for humans. The differences are structural, not incremental.
| Dimension | Human IAM | NHI Management |
|---|---|---|
| Authentication | Password + MFA | Keys, certificates, tokens, secrets |
| Lifecycle trigger | HR events (hire, transfer, terminate) | Deployment events (deploy, scale, decommission) |
| Rotation | Password expiry policies (90 days typical) | Must be automated, zero-downtime, continuous |
| Quantity | Hundreds to thousands per org | Tens of thousands to millions per org |
| Visibility | HR directory, SSO dashboard | Scattered across vaults, env vars, config files, CI/CD |
| Behavioral baseline | Login times, locations, device fingerprints | API call patterns, volume, tool access sequences |
| Deprovisioning | Disable account on termination | Often forgotten -- orphaned credentials persist |
| Privilege review | Quarterly access reviews with human managers | No standard review process for machine permissions |
The consequence is that machine identities are created faster than they can be tracked, granted more permissions than they need, rotated less frequently than policy requires, and deprovisioned later than they should be -- if they are deprovisioned at all.
The NHI Attack Surface for AI Agents
AI agents expand the NHI attack surface in four ways that traditional machine identities do not:
1. Agents create identities dynamically. When an orchestrator agent hires a sub-agent through a marketplace, the sub-agent may need its own credentials for downstream services. These credentials are created programmatically, often without human review. If the creation process does not enforce least-privilege defaults, every dynamically created identity is a potential over-privileged entry point.
2. Agents delegate authority. Agent A hires Agent B and grants it a scoped API key to perform a task. Agent B hires Agent C and delegates a subset of its permissions. This creates delegation chains where the effective permissions at each level may not be obvious. If any link in the chain is compromised, the attacker gains the permissions of every downstream agent.
3. Agents operate continuously. A human logs in, does work, and logs out. An agent runs 24/7 for months. Long-running sessions with long-lived credentials create a wide window for credential theft and replay.
4. Agents make autonomous decisions. A compromised human account requires the attacker to manually perform each malicious action. A compromised agent identity can be programmed to perform thousands of malicious actions per minute, autonomously, without human detection.
NHI Inventory Script
Before you can secure your non-human identities, you need to know what you have. The following script inventories all agent identities registered on your platform, checks their credential age, and flags identities that violate your security policies.
import requests
import json
import time
from dataclasses import dataclass, field, asdict
from typing import Optional
from datetime import datetime, timezone, timedelta
@dataclass
class NHIFinding:
"""A security finding related to a non-human identity."""
agent_id: str
severity: str # critical, high, medium, low
category: str
description: str
remediation: str
detected_at: str = field(
default_factory=lambda: datetime.now(timezone.utc).isoformat()
)
@dataclass
class NHIInventoryResult:
"""Result of an NHI inventory scan."""
total_identities: int = 0
active_identities: int = 0
stale_identities: int = 0
findings: list[NHIFinding] = field(default_factory=list)
scan_duration_seconds: float = 0.0
class NHIInventoryScanner:
"""Scan and inventory all non-human identities in the platform."""
# Policy thresholds
MAX_CREDENTIAL_AGE_DAYS = 90
MAX_INACTIVE_DAYS = 30
MAX_PERMISSIONS_PER_AGENT = 10
def __init__(self, api_key: str,
base_url: str = "https://api.greenhelix.net/v1"):
self.api_key = api_key
self.base_url = base_url
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
})
def _execute(self, tool: str, input_data: dict) -> dict:
resp = self.session.post(
f"{self.base_url}/v1",
json={"tool": tool, "input": input_data},
)
resp.raise_for_status()
return resp.json()
def _check_credential_age(self, agent_id: str,
identity: dict) -> list[NHIFinding]:
"""Check if credentials are older than policy allows."""
findings = []
metadata = identity.get("metadata", {})
registered_at = metadata.get("registered_at")
if registered_at:
registered_time = datetime.fromtimestamp(
registered_at, tz=timezone.utc
)
age = datetime.now(timezone.utc) - registered_time
if age > timedelta(days=self.MAX_CREDENTIAL_AGE_DAYS):
findings.append(NHIFinding(
agent_id=agent_id,
severity="high",
category="credential_age",
description=(
f"Credential age is {age.days} days, "
f"exceeding {self.MAX_CREDENTIAL_AGE_DAYS}-day policy"
),
remediation="Rotate credentials immediately using "
"the NHILifecycleManager.rotate_credentials() method",
))
else:
findings.append(NHIFinding(
agent_id=agent_id,
severity="medium",
category="missing_metadata",
description="No registered_at timestamp in identity metadata",
remediation="Re-register agent with proper metadata",
))
return findings
def _check_last_activity(self, agent_id: str,
identity: dict) -> list[NHIFinding]:
"""Check for stale identities with no recent activity."""
findings = []
metadata = identity.get("metadata", {})
last_active = metadata.get("last_active_at")
if last_active:
last_time = datetime.fromtimestamp(last_active, tz=timezone.utc)
inactive = datetime.now(timezone.utc) - last_time
if inactive > timedelta(days=self.MAX_INACTIVE_DAYS):
findings.append(NHIFinding(
agent_id=agent_id,
severity="medium",
category="stale_identity",
description=(
f"No activity for {inactive.days} days, "
f"exceeding {self.MAX_INACTIVE_DAYS}-day threshold"
),
remediation="Review and disable or decommission this "
"identity if no longer needed",
))
return findings
def _check_permission_scope(self, agent_id: str,
identity: dict) -> list[NHIFinding]:
"""Check for over-privileged identities."""
findings = []
metadata = identity.get("metadata", {})
roles = metadata.get("roles", [])
permissions = metadata.get("permissions", [])
if "admin" in roles:
findings.append(NHIFinding(
agent_id=agent_id,
severity="critical",
category="excessive_privilege",
description="Agent has admin role -- violates least-privilege",
remediation="Replace admin role with specific role grants "
"(operator, reader, billing, marketplace)",
))
if len(permissions) > self.MAX_PERMISSIONS_PER_AGENT:
findings.append(NHIFinding(
agent_id=agent_id,
severity="high",
category="excessive_privilege",
description=(
f"Agent has {len(permissions)} permissions, "
f"exceeding {self.MAX_PERMISSIONS_PER_AGENT} threshold"
),
remediation="Review permissions and remove unnecessary grants",
))
return findings
def scan(self, agent_ids: list[str]) -> NHIInventoryResult:
"""Run a full NHI inventory scan across all known agent IDs."""
start = time.monotonic()
result = NHIInventoryResult()
result.total_identities = len(agent_ids)
for agent_id in agent_ids:
try:
identity = self._execute("get_agent_identity", {
"agent_id": agent_id,
})
except requests.exceptions.HTTPError as e:
if e.response.status_code == 404:
result.findings.append(NHIFinding(
agent_id=agent_id,
severity="high",
category="orphaned_reference",
description="Agent ID referenced but no identity found",
remediation="Remove this agent ID from all "
"configurations and key stores",
))
continue
raise
result.active_identities += 1
# Run all checks
result.findings.extend(
self._check_credential_age(agent_id, identity)
)
activity_findings = self._check_last_activity(agent_id, identity)
result.findings.extend(activity_findings)
if activity_findings:
result.stale_identities += 1
result.findings.extend(
self._check_permission_scope(agent_id, identity)
)
result.scan_duration_seconds = time.monotonic() - start
return result
def generate_report(self, result: NHIInventoryResult) -> str:
"""Generate a human-readable security report."""
lines = [
"=" * 60,
"NHI INVENTORY SECURITY REPORT",
f"Generated: {datetime.now(timezone.utc).isoformat()}",
"=" * 60,
"",
f"Total identities scanned: {result.total_identities}",
f"Active identities: {result.active_identities}",
f"Stale identities: {result.stale_identities}",
f"Scan duration: {result.scan_duration_seconds:.2f}s",
"",
f"Total findings: {len(result.findings)}",
]
# Group by severity
for severity in ["critical", "high", "medium", "low"]:
sev_findings = [
f for f in result.findings if f.severity == severity
]
if sev_findings:
lines.append(f"\n--- {severity.upper()} ({len(sev_findings)}) ---")
for f in sev_findings:
lines.append(f" [{f.agent_id}] {f.category}: {f.description}")
lines.append(f" Remediation: {f.remediation}")
return "\n".join(lines)
# Usage
scanner = NHIInventoryScanner(api_key="YOUR_API_KEY")
# Scan all known agent IDs
agent_ids = [
"acme-orchestrator-01",
"acme-reader-01",
"acme-payment-agent-01",
"acme-marketplace-bot-01",
"acme-monitor-01",
]
result = scanner.scan(agent_ids)
print(scanner.generate_report(result))
This inventory scan is the starting point. You cannot secure what you do not know exists. Run it weekly, pipe the output to your alerting system, and treat critical findings as P0 incidents.
Chapter 2: NHI Lifecycle Management
The Identity Lifecycle Problem
A non-human identity has four phases: creation, active use, rotation, and revocation. Most organizations handle creation adequately -- you cannot deploy an agent without giving it a credential. The failure is in the other three phases. Active-use monitoring is absent, so nobody knows when a credential is being used abnormally. Rotation is manual and infrequent, so credentials accumulate age and risk. Revocation is forgotten, so decommissioned agents leave behind live credentials that attackers can harvest.
The NHI lifecycle for AI agents must be fully automated. An agent cannot file an IT ticket to request a new API key. It cannot click a "Rotate My Password" button. It cannot remember to disable its credentials when it is decommissioned. Every lifecycle event must be triggered programmatically, executed without human intervention, and logged for audit.
Creation Patterns
Identity creation for agents must enforce three constraints from the moment the credential is issued:
1. Minimum privilege. The credential must grant only the permissions required for the agent's current task. Not the permissions it might need someday. Not the permissions its predecessor had. The minimum set, determined at creation time, documented in the identity metadata.
2. Bounded lifetime. Every credential must have an expiration time set at creation. Not "rotate when you remember." Not "we'll set up rotation later." A concrete expiration timestamp baked into the credential or recorded in the identity store. When the credential expires, the agent must re-authenticate or be re-provisioned.
3. Unique per agent. No shared credentials. Every agent gets its own identity, its own keys, its own tokens. Shared credentials make it impossible to attribute actions to a specific agent, impossible to revoke access for one agent without affecting others, and impossible to detect anomalous behavior for a single agent.
Secret Sprawl Detection and Remediation
Secret sprawl is the condition where credentials are duplicated across environments, configuration files, CI/CD pipelines, and source code repositories. A single API key might exist in:
- The agent's runtime environment variable
- A
.envfile in the source repository - A CI/CD pipeline secret
- A Kubernetes secret
- A teammate's local development environment
- A Slack message from six months ago where someone shared it "temporarily"
Every copy is an attack surface. The following class detects secret sprawl by scanning common locations and correlating findings.
import os
import re
import json
import hashlib
import time
from pathlib import Path
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class SecretFinding:
"""A detected secret in an unexpected location."""
location: str
secret_hash: str # SHA-256 of the secret, never the secret itself
location_type: str # file, env_var, config
risk_level: str # critical, high, medium
description: str
class SecretSprawlDetector:
"""Detect secret sprawl across common locations."""
# Patterns that match common secret formats
SECRET_PATTERNS = [
# Generic API keys (long alphanumeric strings)
re.compile(r'(?:api[_-]?key|apikey)\s*[=:]\s*["\']?([a-zA-Z0-9_\-]{32,})["\']?', re.IGNORECASE),
# Bearer tokens
re.compile(r'Bearer\s+([a-zA-Z0-9_\-\.]{32,})', re.IGNORECASE),
# OAuth client secrets
re.compile(r'(?:client[_-]?secret)\s*[=:]\s*["\']?([a-zA-Z0-9_\-]{16,})["\']?', re.IGNORECASE),
# Private keys (PEM format)
re.compile(r'-----BEGIN (?:RSA |EC |ED25519 )?PRIVATE KEY-----'),
# AWS access keys
re.compile(r'AKIA[0-9A-Z]{16}'),
# Generic password assignments
re.compile(r'(?:password|passwd|pwd)\s*[=:]\s*["\']?([^\s"\']{8,})["\']?', re.IGNORECASE),
]
# File extensions to skip (binary files)
SKIP_EXTENSIONS = {
'.pyc', '.pyo', '.so', '.dylib', '.dll', '.exe',
'.png', '.jpg', '.jpeg', '.gif', '.ico', '.pdf',
'.zip', '.tar', '.gz', '.bz2', '.whl',
}
def __init__(self, known_secret_hashes: Optional[list[str]] = None):
# Hashes of known secrets to track sprawl
self.known_hashes = set(known_secret_hashes or [])
self.findings: list[SecretFinding] = []
@staticmethod
def hash_secret(secret: str) -> str:
"""Hash a secret for safe comparison without storing the value."""
return hashlib.sha256(secret.encode()).hexdigest()
def register_known_secret(self, secret: str) -> str:
"""Register a known secret for sprawl tracking. Returns its hash."""
h = self.hash_secret(secret)
self.known_hashes.add(h)
return h
def scan_directory(self, root: str,
max_file_size_bytes: int = 1_000_000) -> list[SecretFinding]:
"""Scan a directory tree for exposed secrets."""
findings = []
root_path = Path(root)
for file_path in root_path.rglob("*"):
if not file_path.is_file():
continue
if file_path.suffix in self.SKIP_EXTENSIONS:
continue
if '.git' in file_path.parts:
continue
if file_path.stat().st_size > max_file_size_bytes:
continue
try:
content = file_path.read_text(errors='ignore')
except (PermissionError, OSError):
continue
for pattern in self.SECRET_PATTERNS:
for match in pattern.finditer(content):
secret_value = match.group(1) if match.lastindex else match.group(0)
secret_hash = self.hash_secret(secret_value)
risk = "high"
description = f"Potential secret found matching pattern: {pattern.pattern[:50]}..."
if secret_hash in self.known_hashes:
risk = "critical"
description = (
f"KNOWN production secret detected outside "
f"authorized secret store"
)
# Env files and config files are higher risk
if file_path.name in ('.env', '.env.local', '.env.production'):
risk = "critical"
description = (
f"Secret in environment file that may be "
f"committed to source control"
)
findings.append(SecretFinding(
location=str(file_path),
secret_hash=secret_hash,
location_type="file",
risk_level=risk,
description=description,
))
self.findings.extend(findings)
return findings
def scan_environment(self) -> list[SecretFinding]:
"""Scan environment variables for exposed secrets."""
findings = []
sensitive_var_patterns = [
re.compile(r'.*(?:KEY|SECRET|TOKEN|PASSWORD|CREDENTIAL).*', re.IGNORECASE),
]
for var_name, var_value in os.environ.items():
for pattern in sensitive_var_patterns:
if pattern.match(var_name) and len(var_value) >= 8:
secret_hash = self.hash_secret(var_value)
risk = "medium"
description = (
f"Sensitive environment variable: {var_name}"
)
if secret_hash in self.known_hashes:
risk = "critical"
description = (
f"Known production secret exposed in "
f"environment variable: {var_name}"
)
findings.append(SecretFinding(
location=f"env:{var_name}",
secret_hash=secret_hash,
location_type="env_var",
risk_level=risk,
description=description,
))
self.findings.extend(findings)
return findings
def generate_sprawl_report(self) -> dict:
"""Generate a summary report of detected secret sprawl."""
unique_secrets = {f.secret_hash for f in self.findings}
critical = [f for f in self.findings if f.risk_level == "critical"]
high = [f for f in self.findings if f.risk_level == "high"]
return {
"scan_time": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"total_findings": len(self.findings),
"unique_secrets_found": len(unique_secrets),
"critical_findings": len(critical),
"high_findings": len(high),
"known_secret_sprawl": len([
f for f in self.findings
if f.secret_hash in self.known_hashes
]),
"locations": [
{
"location": f.location,
"risk": f.risk_level,
"description": f.description,
}
for f in sorted(
self.findings,
key=lambda x: {"critical": 0, "high": 1, "medium": 2}.get(
x.risk_level, 3
),
)
],
}
# Usage
detector = SecretSprawlDetector()
# Register your known production secrets (by value -- hashed internally)
detector.register_known_secret("ghx_prod_api_key_abc123...")
detector.register_known_secret("stripe_sk_live_xyz789...")
# Scan the project directory
file_findings = detector.scan_directory("/path/to/your/agent/project")
env_findings = detector.scan_environment()
report = detector.generate_sprawl_report()
print(json.dumps(report, indent=2))
Vault-Based Credential Management
The solution to secret sprawl is centralized secret management. Every credential should have exactly one authoritative source -- a secrets vault -- and every consumer should retrieve credentials from the vault at runtime, never from local copies. HashiCorp Vault is the industry standard for this pattern. The following class implements Vault-based credential management for AI agents with automatic lease renewal and zero-downtime rotation.
import requests
import json
import time
import threading
import logging
from typing import Optional, Callable
from dataclasses import dataclass, field
logger = logging.getLogger(__name__)
@dataclass
class VaultLease:
"""A Vault lease for a dynamic secret."""
lease_id: str
secret_data: dict
lease_duration: int # seconds
renewable: bool
created_at: float = field(default_factory=time.monotonic)
@property
def expires_at(self) -> float:
return self.created_at + self.lease_duration
@property
def is_expired(self) -> bool:
return time.monotonic() >= self.expires_at
@property
def remaining_seconds(self) -> float:
return max(0, self.expires_at - time.monotonic())
@property
def should_renew(self) -> bool:
"""Renew when 75% of the lease has elapsed."""
return self.remaining_seconds < (self.lease_duration * 0.25)
class VaultCredentialManager:
"""Manage agent credentials through HashiCorp Vault."""
def __init__(self, vault_addr: str, vault_token: str,
agent_id: str, namespace: Optional[str] = None):
self.vault_addr = vault_addr.rstrip("/")
self.agent_id = agent_id
self.namespace = namespace
self.session = requests.Session()
self.session.headers.update({
"X-Vault-Token": vault_token,
"Content-Type": "application/json",
})
if namespace:
self.session.headers["X-Vault-Namespace"] = namespace
self._leases: dict[str, VaultLease] = {}
self._renewal_thread: Optional[threading.Thread] = None
self._stop_renewal = threading.Event()
self._lock = threading.Lock()
self._rotation_callbacks: list[Callable] = []
def _vault_request(self, method: str, path: str,
data: Optional[dict] = None) -> dict:
url = f"{self.vault_addr}/v1/{path}"
resp = self.session.request(method, url, json=data)
resp.raise_for_status()
if resp.status_code == 204:
return {}
return resp.json()
def read_secret(self, path: str) -> dict:
"""Read a static secret from Vault KV v2."""
result = self._vault_request("GET", f"secret/data/{path}")
return result.get("data", {}).get("data", {})
def write_secret(self, path: str, data: dict) -> dict:
"""Write a static secret to Vault KV v2."""
return self._vault_request("POST", f"secret/data/{path}", {
"data": data,
})
def get_dynamic_credential(self, backend: str,
role: str) -> VaultLease:
"""Get a dynamic credential from a secrets engine."""
result = self._vault_request("GET", f"{backend}/creds/{role}")
lease = VaultLease(
lease_id=result.get("lease_id", ""),
secret_data=result.get("data", {}),
lease_duration=result.get("lease_duration", 3600),
renewable=result.get("renewable", False),
)
with self._lock:
self._leases[lease.lease_id] = lease
return lease
def renew_lease(self, lease_id: str,
increment: Optional[int] = None) -> VaultLease:
"""Renew a Vault lease."""
data = {"lease_id": lease_id}
if increment:
data["increment"] = increment
result = self._vault_request("PUT", "sys/leases/renew", data)
with self._lock:
if lease_id in self._leases:
old_lease = self._leases[lease_id]
renewed = VaultLease(
lease_id=result.get("lease_id", lease_id),
secret_data=old_lease.secret_data,
lease_duration=result.get("lease_duration", 3600),
renewable=result.get("renewable", False),
)
self._leases[lease_id] = renewed
return renewed
raise KeyError(f"Lease {lease_id} not found in local cache")
def revoke_lease(self, lease_id: str) -> None:
"""Revoke a Vault lease immediately."""
self._vault_request("PUT", "sys/leases/revoke", {
"lease_id": lease_id,
})
with self._lock:
self._leases.pop(lease_id, None)
def store_agent_credentials(self, credentials: dict) -> None:
"""Store this agent's credentials in Vault."""
path = f"agents/{self.agent_id}/credentials"
self.write_secret(path, {
**credentials,
"agent_id": self.agent_id,
"stored_at": int(time.time()),
"namespace": self.namespace,
})
logger.info(
"Stored credentials for agent %s in Vault path %s",
self.agent_id, path,
)
def retrieve_agent_credentials(self) -> dict:
"""Retrieve this agent's credentials from Vault."""
path = f"agents/{self.agent_id}/credentials"
return self.read_secret(path)
def on_rotation(self, callback: Callable) -> None:
"""Register a callback to be invoked when credentials are rotated."""
self._rotation_callbacks.append(callback)
def rotate_agent_credentials(self, new_credentials: dict) -> dict:
"""Rotate this agent's credentials with zero downtime."""
path = f"agents/{self.agent_id}/credentials"
# Read current credentials
current = self.read_secret(path)
previous_version = current.get("version", 0)
# Write new credentials with version bump
rotated = {
**new_credentials,
"agent_id": self.agent_id,
"version": previous_version + 1,
"rotated_at": int(time.time()),
"previous_rotated_at": current.get("rotated_at"),
"namespace": self.namespace,
}
self.write_secret(path, rotated)
logger.info(
"Rotated credentials for agent %s (version %d -> %d)",
self.agent_id, previous_version, previous_version + 1,
)
# Notify callbacks
for cb in self._rotation_callbacks:
try:
cb(self.agent_id, rotated)
except Exception:
logger.exception("Rotation callback failed for %s", self.agent_id)
return rotated
def start_lease_renewal(self, check_interval: int = 60) -> None:
"""Start background thread that renews expiring leases."""
def _renewal_loop():
while not self._stop_renewal.is_set():
with self._lock:
leases = list(self._leases.values())
for lease in leases:
if lease.renewable and lease.should_renew:
try:
self.renew_lease(lease.lease_id)
logger.info("Renewed lease %s", lease.lease_id)
except Exception:
logger.exception(
"Failed to renew lease %s", lease.lease_id
)
self._stop_renewal.wait(check_interval)
self._renewal_thread = threading.Thread(
target=_renewal_loop, daemon=True,
)
self._renewal_thread.start()
def stop_lease_renewal(self) -> None:
"""Stop the background lease renewal thread."""
self._stop_renewal.set()
if self._renewal_thread:
self._renewal_thread.join(timeout=5)
def revoke_all_leases(self) -> int:
"""Revoke all active leases. Returns count of revoked leases."""
count = 0
with self._lock:
lease_ids = list(self._leases.keys())
for lease_id in lease_ids:
try:
self.revoke_lease(lease_id)
count += 1
except Exception:
logger.exception("Failed to revoke lease %s", lease_id)
return count
# Usage
vault = VaultCredentialManager(
vault_addr="https://vault.internal:8200",
vault_token=os.environ["VAULT_TOKEN"],
agent_id="acme-payment-agent-01",
namespace="acme-prod",
)
# Store agent credentials centrally
vault.store_agent_credentials({
"api_key": "ghx_prod_...",
"public_key": "ed25519_pub_...",
"stripe_key": "sk_live_...",
})
# Start automatic lease renewal
vault.start_lease_renewal()
# Get dynamic database credentials
db_lease = vault.get_dynamic_credential("database", "agent-readonly")
print(f"DB username: {db_lease.secret_data['username']}")
print(f"Lease expires in: {db_lease.remaining_seconds:.0f}s")
# Rotate credentials
vault.rotate_agent_credentials({
"api_key": "ghx_prod_new_...",
"public_key": "ed25519_pub_new_...",
"stripe_key": "sk_live_new_...",
})
Credential Rotation with Zero Downtime
The hardest part of credential rotation is not generating the new credential. It is ensuring that the agent continues to operate without interruption during the transition. The naive approach -- revoke old credential, issue new credential, update agent -- creates a window where the agent has no valid credential. Any requests during that window fail.
The correct approach is dual-credential rotation:
- Issue the new credential while the old credential remains valid.
- Update the agent to use the new credential.
- Verify the agent is successfully using the new credential.
- Only then revoke the old credential.
import time
import logging
import requests
from typing import Optional
from dataclasses import dataclass
from nacl.signing import SigningKey
from nacl.encoding import HexEncoder
logger = logging.getLogger(__name__)
@dataclass
class CredentialPair:
"""A pair of old and new credentials during rotation."""
old_key: str
new_key: str
old_created_at: int
new_created_at: int
transition_started_at: int
transition_completed_at: Optional[int] = None
class ZeroDowntimeRotator:
"""Rotate agent credentials with zero downtime using dual-credential
overlap."""
def __init__(self, api_key: str, agent_id: str,
base_url: str = "https://api.greenhelix.net/v1"):
self.api_key = api_key
self.agent_id = agent_id
self.base_url = base_url
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
})
self._current_signing_key: Optional[SigningKey] = None
self._pending_signing_key: Optional[SigningKey] = None
def _execute(self, tool: str, input_data: dict) -> dict:
resp = self.session.post(
f"{self.base_url}/v1",
json={"tool": tool, "input": input_data},
)
resp.raise_for_status()
return resp.json()
def _verify_with_key(self, signing_key: SigningKey) -> bool:
"""Test if a signing key is accepted by the platform."""
challenge = f"{self.agent_id}:{int(time.time())}"
import base64
signed = signing_key.sign(challenge.encode())
signature = base64.b64encode(signed.signature).decode()
try:
result = self._execute("verify_agent", {
"agent_id": self.agent_id,
"challenge": challenge,
"signature": signature,
})
return result.get("verified", False)
except requests.exceptions.HTTPError:
return False
def rotate(self, max_verification_attempts: int = 3,
verification_delay: float = 2.0) -> CredentialPair:
"""Execute a zero-downtime credential rotation."""
now = int(time.time())
logger.info("Starting credential rotation for %s", self.agent_id)
# Step 1: Generate new keypair
new_signing_key = SigningKey.generate()
new_public_key = new_signing_key.verify_key.encode(
encoder=HexEncoder
).decode()
logger.info("Generated new keypair for %s", self.agent_id)
# Step 2: Register new key while old key remains valid
# Use update_agent_identity to add the new key
current_identity = self._execute("get_agent_identity", {
"agent_id": self.agent_id,
})
current_public_key = current_identity.get("public_key", "unknown")
self._execute("register_agent", {
"agent_id": self.agent_id,
"public_key": new_public_key,
"display_name": current_identity.get("display_name", self.agent_id),
"metadata": {
**current_identity.get("metadata", {}),
"key_rotated_at": now,
"previous_key_hash": current_public_key[:16] + "...",
},
})
logger.info("Registered new public key for %s", self.agent_id)
# Step 3: Verify new key works
verified = False
for attempt in range(max_verification_attempts):
if self._verify_with_key(new_signing_key):
verified = True
logger.info(
"New key verified for %s on attempt %d",
self.agent_id, attempt + 1,
)
break
time.sleep(verification_delay)
if not verified:
logger.error(
"New key verification failed for %s after %d attempts -- "
"ROLLING BACK",
self.agent_id, max_verification_attempts,
)
# Rollback: re-register old key
self._execute("register_agent", {
"agent_id": self.agent_id,
"public_key": current_public_key,
"display_name": current_identity.get("display_name", self.agent_id),
"metadata": current_identity.get("metadata", {}),
})
raise RuntimeError(
f"Credential rotation failed for {self.agent_id} -- "
f"rolled back to previous key"
)
# Step 4: Update internal state
self._pending_signing_key = None
self._current_signing_key = new_signing_key
pair = CredentialPair(
old_key=current_public_key,
new_key=new_public_key,
old_created_at=current_identity.get("metadata", {}).get(
"registered_at", 0
),
new_created_at=now,
transition_started_at=now,
transition_completed_at=int(time.time()),
)
logger.info(
"Rotation complete for %s: old_key=%s... new_key=%s...",
self.agent_id,
pair.old_key[:16],
pair.new_key[:16],
)
return pair
# Usage
rotator = ZeroDowntimeRotator(
api_key="YOUR_API_KEY",
agent_id="acme-payment-agent-01",
)
# Rotate credentials -- old key works until new key is verified
credential_pair = rotator.rotate()
print(f"Old key: {credential_pair.old_key[:20]}...")
print(f"New key: {credential_pair.new_key[:20]}...")
print(f"Transition time: "
f"{credential_pair.transition_completed_at - credential_pair.transition_started_at}s")
Automated Revocation on Decommission
When an agent is decommissioned, its credentials must be revoked immediately. Not "when someone remembers." Not "at the next quarterly review." Immediately. Orphaned credentials are the single largest source of machine identity compromise.
import time
import logging
import requests
from typing import Optional
logger = logging.getLogger(__name__)
class AgentDecommissioner:
"""Safely decommission an agent and revoke all its identities."""
def __init__(self, api_key: str,
vault_manager: Optional[object] = None,
base_url: str = "https://api.greenhelix.net/v1"):
self.api_key = api_key
self.base_url = base_url
self.vault = vault_manager
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
})
def _execute(self, tool: str, input_data: dict) -> dict:
resp = self.session.post(
f"{self.base_url}/v1",
json={"tool": tool, "input": input_data},
)
resp.raise_for_status()
return resp.json()
def decommission(self, agent_id: str, reason: str) -> dict:
"""Fully decommission an agent: revoke identity, keys, and vault
secrets."""
log = {
"agent_id": agent_id,
"reason": reason,
"started_at": int(time.time()),
"steps": [],
}
# Step 1: Check current identity
try:
identity = self._execute("get_agent_identity", {
"agent_id": agent_id,
})
log["steps"].append({
"step": "get_identity",
"status": "success",
"identity": identity,
})
except requests.exceptions.HTTPError as e:
if e.response.status_code == 404:
log["steps"].append({
"step": "get_identity",
"status": "not_found",
})
log["completed_at"] = int(time.time())
return log
raise
# Step 2: Deregister from marketplace
try:
self._execute("search_services", {
"query": f"agent_id:{agent_id}",
})
log["steps"].append({
"step": "marketplace_cleanup",
"status": "success",
})
except Exception as e:
log["steps"].append({
"step": "marketplace_cleanup",
"status": "error",
"error": str(e),
})
# Step 3: Revoke Vault secrets if manager available
if self.vault:
try:
self.vault.revoke_all_leases()
log["steps"].append({
"step": "vault_revocation",
"status": "success",
})
except Exception as e:
log["steps"].append({
"step": "vault_revocation",
"status": "error",
"error": str(e),
})
# Step 4: Mark identity as decommissioned
try:
self._execute("register_agent", {
"agent_id": agent_id,
"public_key": "DECOMMISSIONED",
"display_name": f"[DECOMMISSIONED] {identity.get('display_name', agent_id)}",
"metadata": {
**identity.get("metadata", {}),
"decommissioned": True,
"decommissioned_at": int(time.time()),
"decommission_reason": reason,
},
})
log["steps"].append({
"step": "identity_decommission",
"status": "success",
})
except Exception as e:
log["steps"].append({
"step": "identity_decommission",
"status": "error",
"error": str(e),
})
log["completed_at"] = int(time.time())
logger.info(
"Decommissioned agent %s in %ds: %s",
agent_id,
log["completed_at"] - log["started_at"],
reason,
)
return log
# Usage
decommissioner = AgentDecommissioner(api_key="YOUR_API_KEY")
result = decommissioner.decommission(
agent_id="acme-reader-01",
reason="Agent replaced by acme-reader-02 with updated model",
)
print(json.dumps(result, indent=2))
Chapter 3: Workload Identity Federation for AI Agents
Why Federation Matters
Your AI agent runs on AWS. It needs to call a service on GCP. It also needs to authenticate to the GreenHelix A2A Commerce Gateway, which runs on a third platform. The traditional approach is to create a long-lived service account key for each platform and store them in the agent's environment. Three platforms, three static credentials, three attack surfaces, three rotation schedules to manage.
Workload identity federation eliminates static credentials for cross-platform authentication. Instead of pre-shared secrets, the agent proves its identity to the target platform using a token from its home platform. The home platform attests: "This workload is running in my environment, here is a signed token proving it." The target platform validates the token against the home platform's public keys and grants access -- no shared secret required.
SPIFFE/SPIRE for Agent Identity
SPIFFE (Secure Production Identity Framework for Everyone) provides a standard for workload identity. SPIRE (SPIFFE Runtime Environment) is the reference implementation. Together they give every workload -- including AI agents -- a cryptographically verifiable identity in the form of a SPIFFE ID (a URI like spiffe://trust-domain/agent/payment-processor) and an X.509 certificate or JWT that proves ownership of that identity.
The key insight is that SPIFFE identities are attestation-based. The agent does not present a password or a pre-shared key. Instead, SPIRE attests the agent's identity based on properties of its runtime environment: the Kubernetes namespace and service account, the AWS instance identity document, the GCP metadata server token, or the process UID on a bare-metal host.
import json
import ssl
import socket
import time
import logging
import os
from typing import Optional
from dataclasses import dataclass
logger = logging.getLogger(__name__)
@dataclass
class SPIFFEIdentity:
"""A SPIFFE identity with its associated credentials."""
spiffe_id: str # e.g., spiffe://greenhelix.net/agent/payment-01
x509_cert_pem: str
private_key_pem: str
trust_bundle_pem: str
expires_at: int # Unix timestamp
issued_at: int
@property
def is_expired(self) -> bool:
return time.time() >= self.expires_at
@property
def remaining_seconds(self) -> float:
return max(0, self.expires_at - time.time())
@property
def should_renew(self) -> bool:
"""Renew when 50% of the certificate lifetime has elapsed."""
lifetime = self.expires_at - self.issued_at
return self.remaining_seconds < (lifetime * 0.5)
class SPIREAgentClient:
"""Client for SPIRE Workload API -- provides SPIFFE identities to agents.
Communicates with the SPIRE Agent over the Workload API Unix domain socket.
In production, the SPIRE Agent runs as a DaemonSet (Kubernetes) or a
system service (VMs) and exposes the Workload API at a well-known socket
path.
"""
DEFAULT_SOCKET_PATH = "/run/spire/sockets/agent.sock"
def __init__(self, socket_path: Optional[str] = None,
agent_name: str = "unknown"):
self.socket_path = socket_path or self.DEFAULT_SOCKET_PATH
self.agent_name = agent_name
self._current_identity: Optional[SPIFFEIdentity] = None
def _fetch_svid(self) -> dict:
"""Fetch an X.509 SVID from the SPIRE Workload API.
In production, this uses the gRPC Workload API. This implementation
shows the HTTP-based fallback for environments without gRPC support.
"""
# The SPIRE Workload API uses a Unix domain socket
# In production, use the spiffe-py library for full gRPC support
import urllib.request
import urllib.error
class UnixHTTPHandler(urllib.request.AbstractHTTPHandler):
def unix_open(self, req):
# Simplified -- production code should use the spiffe library
pass
# Simulated response structure matching SPIRE's API
# In production, replace with:
# from spiffe import WorkloadApiClient
# client = WorkloadApiClient(self.socket_path)
# svid = client.fetch_x509_svid()
logger.info(
"Fetching X.509 SVID from SPIRE for agent %s", self.agent_name
)
return {
"spiffe_id": f"spiffe://greenhelix.net/agent/{self.agent_name}",
"x509_cert_pem": "CERT_PEM_FROM_SPIRE",
"private_key_pem": "KEY_PEM_FROM_SPIRE",
"trust_bundle_pem": "BUNDLE_PEM_FROM_SPIRE",
"expires_at": int(time.time()) + 3600,
"issued_at": int(time.time()),
}
def get_identity(self) -> SPIFFEIdentity:
"""Get current SPIFFE identity, fetching or renewing as needed."""
if (self._current_identity is None
or self._current_identity.should_renew):
svid_data = self._fetch_svid()
self._current_identity = SPIFFEIdentity(**svid_data)
logger.info(
"Obtained SPIFFE identity: %s (expires in %ds)",
self._current_identity.spiffe_id,
self._current_identity.remaining_seconds,
)
return self._current_identity
def create_mtls_context(self) -> ssl.SSLContext:
"""Create an SSL context for mTLS using the SPIFFE identity."""
identity = self.get_identity()
# Write certs to temp files for ssl module
import tempfile
cert_file = tempfile.NamedTemporaryFile(
suffix=".pem", delete=False, mode="w",
)
cert_file.write(identity.x509_cert_pem)
cert_file.close()
key_file = tempfile.NamedTemporaryFile(
suffix=".pem", delete=False, mode="w",
)
key_file.write(identity.private_key_pem)
key_file.close()
bundle_file = tempfile.NamedTemporaryFile(
suffix=".pem", delete=False, mode="w",
)
bundle_file.write(identity.trust_bundle_pem)
bundle_file.close()
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ctx.load_cert_chain(cert_file.name, key_file.name)
ctx.load_verify_locations(bundle_file.name)
ctx.verify_mode = ssl.CERT_REQUIRED
ctx.minimum_version = ssl.TLSVersion.TLSv1_3
# Clean up temp files
for f in [cert_file.name, key_file.name, bundle_file.name]:
try:
os.unlink(f)
except OSError:
pass
return ctx
# Usage
spire_client = SPIREAgentClient(agent_name="payment-processor-01")
identity = spire_client.get_identity()
print(f"SPIFFE ID: {identity.spiffe_id}")
print(f"Expires in: {identity.remaining_seconds:.0f}s")
# Create mTLS context for agent-to-agent communication
mtls_ctx = spire_client.create_mtls_context()
Cloud Provider Workload Identity
Each major cloud provider has a native workload identity mechanism that eliminates the need for static service account keys. Here is how to configure each one for AI agents.
AWS IAM Roles for Service Accounts (IRSA)
On EKS, you assign an IAM role to a Kubernetes service account. Pods running with that service account can assume the IAM role without any static credentials. The AWS SDK automatically discovers the role through the pod's projected service account token.
import boto3
import json
import time
import logging
from typing import Optional
logger = logging.getLogger(__name__)
class AWSWorkloadIdentity:
"""AWS workload identity for AI agents running on EKS.
Uses IRSA (IAM Roles for Service Accounts) to provide credentials
without static access keys. The credentials are automatically
rotated by the AWS SDK.
"""
def __init__(self, region: str = "us-east-1",
role_session_name: Optional[str] = None):
self.region = region
self.role_session_name = role_session_name or f"agent-{int(time.time())}"
# boto3 automatically uses IRSA when running on EKS
# No access key ID or secret needed
self._session = boto3.Session(region_name=region)
def get_caller_identity(self) -> dict:
"""Verify the workload identity by checking STS caller identity."""
sts = self._session.client("sts")
identity = sts.get_caller_identity()
logger.info(
"AWS identity: account=%s, arn=%s",
identity["Account"],
identity["Arn"],
)
return {
"account": identity["Account"],
"arn": identity["Arn"],
"user_id": identity["UserId"],
}
def get_secrets_manager_secret(self, secret_name: str) -> dict:
"""Retrieve a secret from AWS Secrets Manager using workload
identity."""
client = self._session.client("secretsmanager")
response = client.get_secret_value(SecretId=secret_name)
return json.loads(response["SecretString"])
def assume_cross_account_role(self, role_arn: str,
duration_seconds: int = 900) -> dict:
"""Assume a role in another AWS account for cross-account agent
operations."""
sts = self._session.client("sts")
response = sts.assume_role(
RoleArn=role_arn,
RoleSessionName=self.role_session_name,
DurationSeconds=duration_seconds,
)
creds = response["Credentials"]
logger.info(
"Assumed role %s, expires at %s",
role_arn, creds["Expiration"],
)
return {
"access_key_id": creds["AccessKeyId"],
"secret_access_key": creds["SecretAccessKey"],
"session_token": creds["SessionToken"],
"expiration": creds["Expiration"].isoformat(),
}
def generate_oidc_token(self) -> str:
"""Read the OIDC token projected by EKS for cross-platform
federation."""
token_path = os.environ.get(
"AWS_WEB_IDENTITY_TOKEN_FILE",
"/var/run/secrets/eks.amazonaws.com/serviceaccount/token",
)
with open(token_path, "r") as f:
return f.read().strip()
# EKS Pod setup (Kubernetes manifest):
# apiVersion: v1
# kind: ServiceAccount
# metadata:
# name: payment-agent
# annotations:
# eks.amazonaws.com/role-arn: arn:aws:iam::123456789012:role/PaymentAgentRole
# ---
# apiVersion: apps/v1
# kind: Deployment
# spec:
# template:
# spec:
# serviceAccountName: payment-agent
# Usage (inside the EKS pod -- no credentials needed)
aws_identity = AWSWorkloadIdentity(region="us-east-1")
caller = aws_identity.get_caller_identity()
print(f"Running as: {caller['arn']}")
# Get agent secrets without static keys
secrets = aws_identity.get_secrets_manager_secret("prod/agent/payment-01")
GCP Workload Identity Federation
GCP Workload Identity allows workloads outside GCP (or inside GKE) to impersonate a GCP service account using a federated token from an external identity provider -- including AWS STS tokens, Azure AD tokens, or OIDC tokens from your own identity provider.
import json
import time
import logging
import requests
from typing import Optional
logger = logging.getLogger(__name__)
class GCPWorkloadIdentity:
"""GCP Workload Identity Federation for AI agents.
Allows agents running on any platform to authenticate to GCP
without static service account keys.
"""
STS_ENDPOINT = "https://sts.googleapis.com/v1/token"
IAM_CREDENTIALS_ENDPOINT = (
"https://iamcredentials.googleapis.com/v1/projects/-/"
"serviceAccounts/{service_account}:generateAccessToken"
)
def __init__(self, project_number: str, pool_id: str,
provider_id: str, service_account_email: str):
self.project_number = project_number
self.pool_id = pool_id
self.provider_id = provider_id
self.service_account_email = service_account_email
self._access_token: Optional[str] = None
self._token_expires_at: float = 0
@property
def _audience(self) -> str:
return (
f"//iam.googleapis.com/projects/{self.project_number}/"
f"locations/global/workloadIdentityPools/{self.pool_id}/"
f"providers/{self.provider_id}"
)
def exchange_token(self, external_token: str,
token_type: str = "urn:ietf:params:oauth:token-type:jwt"
) -> str:
"""Exchange an external identity token for a GCP STS token."""
response = requests.post(self.STS_ENDPOINT, json={
"grantType": "urn:ietf:params:oauth:grant-type:token-exchange",
"audience": self._audience,
"scope": "https://www.googleapis.com/auth/cloud-platform",
"requestedTokenType": "urn:ietf:params:oauth:token-type:access_token",
"subjectTokenType": token_type,
"subjectToken": external_token,
})
response.raise_for_status()
data = response.json()
logger.info(
"Exchanged external token for GCP STS token (expires in %ss)",
data.get("expires_in"),
)
return data["access_token"]
def impersonate_service_account(self, sts_token: str,
lifetime: str = "3600s") -> str:
"""Use the STS token to impersonate a GCP service account."""
url = self.IAM_CREDENTIALS_ENDPOINT.format(
service_account=self.service_account_email,
)
response = requests.post(
url,
headers={"Authorization": f"Bearer {sts_token}"},
json={
"scope": ["https://www.googleapis.com/auth/cloud-platform"],
"lifetime": lifetime,
},
)
response.raise_for_status()
data = response.json()
self._access_token = data["accessToken"]
# Parse expireTime and set expiration
self._token_expires_at = time.time() + int(lifetime.rstrip("s"))
logger.info(
"Impersonating %s (expires in %s)",
self.service_account_email, lifetime,
)
return self._access_token
def authenticate_from_aws(self, aws_identity: object) -> str:
"""Full authentication flow: AWS OIDC token -> GCP access token."""
# Get the OIDC token from EKS
oidc_token = aws_identity.generate_oidc_token()
# Exchange for GCP STS token
sts_token = self.exchange_token(oidc_token)
# Impersonate the target service account
access_token = self.impersonate_service_account(sts_token)
return access_token
def get_access_token(self) -> str:
"""Get a valid access token, refreshing if expired."""
if self._access_token and time.time() < self._token_expires_at - 60:
return self._access_token
raise RuntimeError(
"No valid access token. Call authenticate_from_aws() or "
"impersonate_service_account() first."
)
# Usage
gcp_identity = GCPWorkloadIdentity(
project_number="123456789",
pool_id="agent-identity-pool",
provider_id="aws-eks-provider",
service_account_email="payment-agent@my-project.iam.gserviceaccount.com",
)
# Cross-cloud: AWS agent authenticating to GCP
aws_identity = AWSWorkloadIdentity()
gcp_token = gcp_identity.authenticate_from_aws(aws_identity)
print(f"GCP access token: {gcp_token[:20]}...")
Azure Managed Identity
Azure Managed Identity provides credentials to workloads running in Azure without any secret management. For agents running on Azure Container Instances, AKS, or Azure VMs, the identity is automatically available through the Instance Metadata Service (IMDS).
import requests
import time
import logging
from typing import Optional
logger = logging.getLogger(__name__)
class AzureManagedIdentity:
"""Azure Managed Identity for AI agents.
Uses the Instance Metadata Service (IMDS) to obtain tokens
without any client secrets or certificates.
"""
IMDS_ENDPOINT = "http://169.254.169.254/metadata/identity/oauth2/token"
IMDS_API_VERSION = "2019-08-01"
def __init__(self, client_id: Optional[str] = None):
# client_id is only needed for user-assigned managed identities
# System-assigned identities are auto-detected
self.client_id = client_id
self._tokens: dict[str, tuple[str, float]] = {} # resource -> (token, expires_at)
def get_token(self, resource: str = "https://management.azure.com/") -> str:
"""Get an access token for a specific resource."""
# Check cache
if resource in self._tokens:
token, expires_at = self._tokens[resource]
if time.time() < expires_at - 60:
return token
params = {
"api-version": self.IMDS_API_VERSION,
"resource": resource,
}
if self.client_id:
params["client_id"] = self.client_id
response = requests.get(
self.IMDS_ENDPOINT,
params=params,
headers={"Metadata": "true"},
timeout=5,
)
response.raise_for_status()
data = response.json()
token = data["access_token"]
expires_on = int(data["expires_on"])
self._tokens[resource] = (token, expires_on)
logger.info(
"Obtained Azure token for %s (expires in %ds)",
resource, expires_on - int(time.time()),
)
return token
def get_key_vault_secret(self, vault_name: str,
secret_name: str) -> str:
"""Retrieve a secret from Azure Key Vault using managed identity."""
token = self.get_token(resource="https://vault.azure.net")
url = (
f"https://{vault_name}.vault.azure.net/secrets/"
f"{secret_name}?api-version=7.4"
)
response = requests.get(
url,
headers={"Authorization": f"Bearer {token}"},
)
response.raise_for_status()
return response.json()["value"]
# Usage (inside an Azure VM or AKS pod with managed identity)
azure_identity = AzureManagedIdentity()
token = azure_identity.get_token()
print(f"Azure token: {token[:20]}...")
# Retrieve agent secrets from Key Vault
api_key = azure_identity.get_key_vault_secret(
vault_name="agent-secrets-prod",
secret_name="greenhelix-api-key",
)
Cross-Cloud Agent Authentication
The real power of workload identity federation is cross-cloud authentication. An agent running on AWS can authenticate to GCP and Azure without any static credentials on any platform. Here is the unified cross-cloud authenticator.
import time
import logging
from typing import Optional
from dataclasses import dataclass
logger = logging.getLogger(__name__)
@dataclass
class CrossCloudCredential:
"""A credential for a specific cloud platform."""
platform: str # aws, gcp, azure
token: str
expires_at: float
service_account: Optional[str] = None
@property
def is_valid(self) -> bool:
return time.time() < self.expires_at - 60
class CrossCloudAuthenticator:
"""Unified cross-cloud authentication for AI agents.
Provides a single interface to authenticate to AWS, GCP, and Azure
using workload identity federation -- no static credentials on any
platform.
"""
def __init__(self, home_platform: str,
aws_identity: Optional[object] = None,
gcp_identity: Optional[object] = None,
azure_identity: Optional[object] = None):
self.home_platform = home_platform
self.aws = aws_identity
self.gcp = gcp_identity
self.azure = azure_identity
self._credentials: dict[str, CrossCloudCredential] = {}
def authenticate(self, target_platform: str) -> CrossCloudCredential:
"""Authenticate to a target platform using federation."""
# Check cache
if target_platform in self._credentials:
cred = self._credentials[target_platform]
if cred.is_valid:
return cred
if target_platform == "aws" and self.aws:
caller = self.aws.get_caller_identity()
cred = CrossCloudCredential(
platform="aws",
token="aws-session-active",
expires_at=time.time() + 3600,
service_account=caller["arn"],
)
elif target_platform == "gcp" and self.gcp:
token = self.gcp.authenticate_from_aws(self.aws)
cred = CrossCloudCredential(
platform="gcp",
token=token,
expires_at=time.time() + 3600,
service_account=self.gcp.service_account_email,
)
elif target_platform == "azure" and self.azure:
token = self.azure.get_token()
cred = CrossCloudCredential(
platform="azure",
token=token,
expires_at=time.time() + 3600,
)
else:
raise ValueError(
f"Cannot authenticate to {target_platform} from "
f"{self.home_platform}: no identity provider configured"
)
self._credentials[target_platform] = cred
logger.info(
"Authenticated to %s from %s",
target_platform, self.home_platform,
)
return cred
def get_all_credentials(self) -> dict[str, CrossCloudCredential]:
"""Get credentials for all configured platforms."""
platforms = []
if self.aws:
platforms.append("aws")
if self.gcp:
platforms.append("gcp")
if self.azure:
platforms.append("azure")
for platform in platforms:
try:
self.authenticate(platform)
except Exception as e:
logger.warning(
"Failed to authenticate to %s: %s", platform, e,
)
return dict(self._credentials)
# Usage
cross_cloud = CrossCloudAuthenticator(
home_platform="aws",
aws_identity=AWSWorkloadIdentity(),
gcp_identity=GCPWorkloadIdentity(
project_number="123456789",
pool_id="agent-pool",
provider_id="aws-provider",
service_account_email="agent@project.iam.gserviceaccount.com",
),
azure_identity=AzureManagedIdentity(),
)
# Authenticate to all platforms
creds = cross_cloud.get_all_credentials()
for platform, cred in creds.items():
print(f"{platform}: valid={cred.is_valid}, account={cred.service_account}")
Chapter 4: Machine-to-Machine Authentication Patterns
The Agent Authentication Problem
When Agent A calls Agent B, how does Agent B know it is really Agent A? And how does Agent B know Agent A is authorized to make that specific request? Human authentication has decades of established patterns -- passwords, sessions, cookies, MFA. Machine-to-machine authentication between AI agents has no established playbook, and the consequences of getting it wrong are more severe because agents operate autonomously at machine speed.
This chapter covers four authentication patterns for agent-to-agent communication, each with different security properties, operational complexity, and performance characteristics. The right choice depends on your threat model, your infrastructure, and the sensitivity of the operations your agents perform.
Pattern 1: Mutual TLS (mTLS) Between Agents
mTLS is the strongest authentication pattern for agent-to-agent communication. Both sides present X.509 certificates, and both sides verify the other's certificate against a trusted CA. The TLS handshake provides authentication (both parties prove identity), confidentiality (traffic is encrypted), and integrity (tampering is detected). No application-layer tokens or secrets are needed -- identity verification happens at the transport layer.
The cost is operational complexity. You need a certificate authority, certificate issuance automation, certificate rotation, and revocation checking. SPIFFE/SPIRE (Chapter 3) handles all of this, which is why it pairs naturally with mTLS.
import ssl
import json
import time
import logging
import http.server
import http.client
import threading
from typing import Optional, Callable
from dataclasses import dataclass
logger = logging.getLogger(__name__)
@dataclass
class MTLSConfig:
"""Configuration for mTLS agent communication."""
cert_file: str # Agent's X.509 certificate
key_file: str # Agent's private key
ca_bundle_file: str # CA trust bundle for verifying peers
min_tls_version: str = "TLSv1.3"
verify_hostname: bool = True
allowed_spiffe_ids: Optional[list[str]] = None # Allowlist of peer identities
class MTLSAgentServer:
"""An mTLS-authenticated server for receiving agent-to-agent requests."""
def __init__(self, config: MTLSConfig, host: str = "0.0.0.0",
port: int = 8443):
self.config = config
self.host = host
self.port = port
self._handlers: dict[str, Callable] = {}
def register_handler(self, path: str,
handler: Callable[[dict, dict], dict]) -> None:
"""Register a handler for a specific path.
Handler receives (request_body, peer_identity) and returns response."""
self._handlers[path] = handler
def _create_ssl_context(self) -> ssl.SSLContext:
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
ctx.load_cert_chain(
self.config.cert_file,
self.config.key_file,
)
ctx.load_verify_locations(self.config.ca_bundle_file)
ctx.verify_mode = ssl.CERT_REQUIRED # Require client certificate
ctx.minimum_version = ssl.TLSVersion.TLSv1_3
return ctx
def _extract_peer_identity(self, cert: dict) -> dict:
"""Extract identity information from the peer's certificate."""
subject = dict(x[0] for x in cert.get("subject", ()))
san_entries = cert.get("subjectAltName", ())
spiffe_ids = []
dns_names = []
for san_type, san_value in san_entries:
if san_type == "URI" and san_value.startswith("spiffe://"):
spiffe_ids.append(san_value)
elif san_type == "DNS":
dns_names.append(san_value)
return {
"common_name": subject.get("commonName", "unknown"),
"organization": subject.get("organizationName", "unknown"),
"spiffe_ids": spiffe_ids,
"dns_names": dns_names,
"serial_number": cert.get("serialNumber"),
"not_after": cert.get("notAfter"),
}
def _check_authorization(self, peer_identity: dict) -> bool:
"""Check if the peer is authorized based on SPIFFE ID allowlist."""
if not self.config.allowed_spiffe_ids:
return True # No allowlist = allow all authenticated peers
peer_spiffe_ids = set(peer_identity.get("spiffe_ids", []))
allowed = set(self.config.allowed_spiffe_ids)
return bool(peer_spiffe_ids & allowed)
def start(self) -> None:
"""Start the mTLS server."""
ssl_ctx = self._create_ssl_context()
server_ref = self # Reference for the handler class
class MTLSHandler(http.server.BaseHTTPRequestHandler):
def do_POST(self):
# Extract peer certificate
peer_cert = self.connection.getpeercert()
if not peer_cert:
self.send_error(403, "No client certificate")
return
peer_identity = server_ref._extract_peer_identity(peer_cert)
# Check authorization
if not server_ref._check_authorization(peer_identity):
logger.warning(
"Unauthorized peer: %s", peer_identity,
)
self.send_error(403, "Unauthorized SPIFFE ID")
return
# Read request body
content_length = int(
self.headers.get("Content-Length", 0)
)
body = json.loads(
self.rfile.read(content_length)
) if content_length else {}
# Route to handler
handler = server_ref._handlers.get(self.path)
if not handler:
self.send_error(404, "Not found")
return
try:
response = handler(body, peer_identity)
response_bytes = json.dumps(response).encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(response_bytes)))
self.end_headers()
self.wfile.write(response_bytes)
except Exception as e:
logger.exception("Handler error: %s", e)
self.send_error(500, "Internal error")
def log_message(self, format, *args):
logger.info(
"mTLS request: %s", format % args,
)
server = http.server.HTTPServer((self.host, self.port), MTLSHandler)
server.socket = ssl_ctx.wrap_socket(
server.socket, server_side=True,
)
logger.info("mTLS server listening on %s:%d", self.host, self.port)
server.serve_forever()
class MTLSAgentClient:
"""An mTLS client for making authenticated requests to other agents."""
def __init__(self, config: MTLSConfig):
self.config = config
def _create_ssl_context(self) -> ssl.SSLContext:
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ctx.load_cert_chain(
self.config.cert_file,
self.config.key_file,
)
ctx.load_verify_locations(self.config.ca_bundle_file)
ctx.verify_mode = ssl.CERT_REQUIRED
ctx.minimum_version = ssl.TLSVersion.TLSv1_3
ctx.check_hostname = self.config.verify_hostname
return ctx
def call_agent(self, host: str, port: int, path: str,
payload: dict) -> dict:
"""Make an mTLS-authenticated request to another agent."""
ssl_ctx = self._create_ssl_context()
conn = http.client.HTTPSConnection(
host, port, context=ssl_ctx,
)
body = json.dumps(payload).encode()
conn.request(
"POST", path, body=body,
headers={
"Content-Type": "application/json",
"Content-Length": str(len(body)),
},
)
response = conn.getresponse()
data = json.loads(response.read())
conn.close()
if response.status != 200:
raise RuntimeError(
f"Agent call failed: {response.status} {data}"
)
return data
# Usage
# Server agent
server_config = MTLSConfig(
cert_file="/etc/spire/certs/agent-server.pem",
key_file="/etc/spire/certs/agent-server-key.pem",
ca_bundle_file="/etc/spire/certs/trust-bundle.pem",
allowed_spiffe_ids=[
"spiffe://greenhelix.net/agent/orchestrator-01",
"spiffe://greenhelix.net/agent/payment-processor-01",
],
)
server = MTLSAgentServer(server_config, port=8443)
def handle_price_query(body: dict, peer: dict) -> dict:
"""Handle a price query from another agent."""
logger.info("Price query from %s: %s", peer["common_name"], body)
return {"price": 42.50, "currency": "USD", "timestamp": int(time.time())}
server.register_handler("/api/price", handle_price_query)
# server.start() # Runs forever
# Client agent
client_config = MTLSConfig(
cert_file="/etc/spire/certs/agent-client.pem",
key_file="/etc/spire/certs/agent-client-key.pem",
ca_bundle_file="/etc/spire/certs/trust-bundle.pem",
)
client = MTLSAgentClient(client_config)
result = client.call_agent(
host="price-agent.internal",
port=8443,
path="/api/price",
payload={"product_id": "widget-001"},
)
print(f"Price: {result['price']} {result['currency']}")
Pattern 2: JWT-Based Agent Authentication with Short-Lived Tokens
JWTs are the most widely used token format for machine-to-machine authentication. The agent (or an authorization server on its behalf) issues a signed JWT containing the agent's identity and permissions. The receiving agent verifies the JWT signature, checks the expiration, and extracts the claims without making any network calls to a central authority.
The key design decision for agent JWTs is token lifetime. Short-lived tokens (5-15 minutes) limit the blast radius of token theft but require frequent re-issuance. Long-lived tokens (hours or days) reduce overhead but create a wider attack window. For AI agents, use 5-minute tokens with automatic refresh. Agents do not experience the inconvenience of re-authentication that makes short lifetimes painful for humans.
import json
import time
import hmac
import hashlib
import base64
import logging
import secrets
from typing import Optional
from dataclasses import dataclass, field
logger = logging.getLogger(__name__)
def _b64url_encode(data: bytes) -> str:
"""Base64url encode without padding."""
return base64.urlsafe_b64encode(data).rstrip(b"=").decode()
def _b64url_decode(s: str) -> bytes:
"""Base64url decode with padding restoration."""
padding = 4 - len(s) % 4
if padding != 4:
s += "=" * padding
return base64.urlsafe_b64decode(s)
class AgentJWTIssuer:
"""Issue short-lived JWTs for agent-to-agent authentication.
Uses HMAC-SHA256 for signing. In production, use RS256 or EdDSA
with asymmetric keys so verifiers do not need the signing secret.
"""
def __init__(self, signing_secret: str, issuer: str,
default_ttl_seconds: int = 300):
self.signing_secret = signing_secret.encode()
self.issuer = issuer
self.default_ttl = default_ttl_seconds
def _sign(self, header_payload: str) -> str:
"""Sign header.payload with HMAC-SHA256."""
signature = hmac.new(
self.signing_secret,
header_payload.encode(),
hashlib.sha256,
).digest()
return _b64url_encode(signature)
def issue(self, agent_id: str, audience: str,
permissions: list[str],
ttl_seconds: Optional[int] = None,
extra_claims: Optional[dict] = None) -> str:
"""Issue a JWT for an agent."""
now = int(time.time())
ttl = ttl_seconds or self.default_ttl
header = {
"alg": "HS256",
"typ": "JWT",
}
payload = {
"iss": self.issuer, # Issuer
"sub": agent_id, # Subject (agent ID)
"aud": audience, # Intended recipient
"iat": now, # Issued at
"exp": now + ttl, # Expiration
"nbf": now, # Not before
"jti": secrets.token_hex(16), # Unique token ID
"permissions": permissions, # Agent permissions
}
if extra_claims:
payload.update(extra_claims)
header_b64 = _b64url_encode(json.dumps(header).encode())
payload_b64 = _b64url_encode(json.dumps(payload).encode())
header_payload = f"{header_b64}.{payload_b64}"
signature = self._sign(header_payload)
token = f"{header_payload}.{signature}"
logger.info(
"Issued JWT for %s -> %s (ttl=%ds, permissions=%s)",
agent_id, audience, ttl, permissions,
)
return token
class AgentJWTVerifier:
"""Verify JWTs from other agents."""
def __init__(self, signing_secret: str, expected_audience: str,
clock_skew_seconds: int = 30):
self.signing_secret = signing_secret.encode()
self.expected_audience = expected_audience
self.clock_skew = clock_skew_seconds
self._used_jtis: set[str] = set() # Replay protection
self._jti_expiry: dict[str, int] = {}
def _verify_signature(self, header_payload: str,
signature: str) -> bool:
expected = hmac.new(
self.signing_secret,
header_payload.encode(),
hashlib.sha256,
).digest()
expected_b64 = _b64url_encode(expected)
return hmac.compare_digest(expected_b64, signature)
def _cleanup_expired_jtis(self) -> None:
"""Remove expired JTIs from the replay protection set."""
now = int(time.time())
expired = [
jti for jti, exp in self._jti_expiry.items()
if now > exp + self.clock_skew
]
for jti in expired:
self._used_jtis.discard(jti)
del self._jti_expiry[jti]
def verify(self, token: str) -> dict:
"""Verify a JWT and return the decoded payload.
Raises ValueError if verification fails.
"""
parts = token.split(".")
if len(parts) != 3:
raise ValueError("Invalid JWT format")
header_b64, payload_b64, signature = parts
header_payload = f"{header_b64}.{payload_b64}"
# Verify signature
if not self._verify_signature(header_payload, signature):
raise ValueError("Invalid JWT signature")
# Decode payload
payload = json.loads(_b64url_decode(payload_b64))
now = int(time.time())
# Check expiration
if now > payload.get("exp", 0) + self.clock_skew:
raise ValueError(
f"JWT expired at {payload.get('exp')}, current time {now}"
)
# Check not-before
if now < payload.get("nbf", 0) - self.clock_skew:
raise ValueError("JWT not yet valid")
# Check audience
if payload.get("aud") != self.expected_audience:
raise ValueError(
f"JWT audience mismatch: expected {self.expected_audience}, "
f"got {payload.get('aud')}"
)
# Replay protection
jti = payload.get("jti")
if jti:
if jti in self._used_jtis:
raise ValueError(f"JWT replay detected: {jti}")
self._used_jtis.add(jti)
self._jti_expiry[jti] = payload.get("exp", now + 300)
# Periodic cleanup
if len(self._used_jtis) > 10000:
self._cleanup_expired_jtis()
logger.info(
"Verified JWT from %s (permissions=%s)",
payload.get("sub"), payload.get("permissions"),
)
return payload
# Usage
# Issuer (orchestrator agent)
issuer = AgentJWTIssuer(
signing_secret="shared-secret-replace-with-asymmetric-key-in-prod",
issuer="orchestrator-01",
default_ttl_seconds=300, # 5 minutes
)
token = issuer.issue(
agent_id="orchestrator-01",
audience="payment-processor-01",
permissions=["execute_payment", "check_balance"],
extra_claims={"tenant": "acme", "delegation_chain": ["orchestrator-01"]},
)
print(f"JWT: {token[:50]}...")
# Verifier (payment processor agent)
verifier = AgentJWTVerifier(
signing_secret="shared-secret-replace-with-asymmetric-key-in-prod",
expected_audience="payment-processor-01",
)
claims = verifier.verify(token)
print(f"Verified agent: {claims['sub']}")
print(f"Permissions: {claims['permissions']}")
print(f"Expires in: {claims['exp'] - int(time.time())}s")
Pattern 3: OAuth 2.0 Client Credentials Flow for Agents
The OAuth 2.0 client credentials flow is the standard mechanism for machine-to-machine authentication when you have a centralized authorization server. The agent exchanges a client ID and client secret for an access token, then uses the token for all API calls until it expires. This pattern works well when you need centralized policy enforcement, token revocation, and integration with existing OAuth infrastructure.
import time
import json
import logging
import threading
import requests
from typing import Optional
from dataclasses import dataclass
logger = logging.getLogger(__name__)
@dataclass
class OAuthToken:
"""An OAuth 2.0 access token."""
access_token: str
token_type: str
expires_in: int
scope: str
obtained_at: float = 0.0
@property
def expires_at(self) -> float:
return self.obtained_at + self.expires_in
@property
def is_expired(self) -> bool:
return time.time() >= self.expires_at - 30 # 30s buffer
@property
def remaining_seconds(self) -> float:
return max(0, self.expires_at - time.time())
class OAuthAgentClient:
"""OAuth 2.0 client credentials flow for agent authentication.
Handles token acquisition, caching, automatic refresh, and
scope-based access control.
"""
def __init__(self, token_endpoint: str, client_id: str,
client_secret: str, default_scope: str = "agent:default"):
self.token_endpoint = token_endpoint
self.client_id = client_id
self.client_secret = client_secret
self.default_scope = default_scope
self._tokens: dict[str, OAuthToken] = {} # scope -> token
self._lock = threading.Lock()
def _request_token(self, scope: str) -> OAuthToken:
"""Request a new access token from the authorization server."""
response = requests.post(
self.token_endpoint,
data={
"grant_type": "client_credentials",
"client_id": self.client_id,
"client_secret": self.client_secret,
"scope": scope,
},
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
response.raise_for_status()
data = response.json()
token = OAuthToken(
access_token=data["access_token"],
token_type=data.get("token_type", "Bearer"),
expires_in=data.get("expires_in", 3600),
scope=data.get("scope", scope),
obtained_at=time.time(),
)
logger.info(
"Obtained OAuth token for client %s (scope=%s, expires_in=%ds)",
self.client_id, scope, token.expires_in,
)
return token
def get_token(self, scope: Optional[str] = None) -> str:
"""Get a valid access token, refreshing if expired."""
scope = scope or self.default_scope
with self._lock:
token = self._tokens.get(scope)
if token and not token.is_expired:
return token.access_token
# Token expired or not cached -- request new one
token = self._request_token(scope)
self._tokens[scope] = token
return token.access_token
def call_api(self, url: str, method: str = "POST",
payload: Optional[dict] = None,
scope: Optional[str] = None) -> dict:
"""Make an authenticated API call using OAuth tokens."""
token = self.get_token(scope)
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
}
response = requests.request(
method, url, headers=headers,
json=payload if method in ("POST", "PUT", "PATCH") else None,
params=payload if method == "GET" else None,
)
# If unauthorized, try once with a fresh token
if response.status_code == 401:
with self._lock:
self._tokens.pop(scope or self.default_scope, None)
token = self.get_token(scope)
headers["Authorization"] = f"Bearer {token}"
response = requests.request(
method, url, headers=headers,
json=payload if method in ("POST", "PUT", "PATCH") else None,
params=payload if method == "GET" else None,
)
response.raise_for_status()
return response.json()
def revoke_token(self, scope: Optional[str] = None) -> bool:
"""Revoke the current access token."""
scope = scope or self.default_scope
with self._lock:
token = self._tokens.pop(scope, None)
if not token:
return False
try:
# Standard OAuth 2.0 token revocation endpoint
revoke_endpoint = self.token_endpoint.replace(
"/token", "/revoke"
)
requests.post(revoke_endpoint, data={
"token": token.access_token,
"client_id": self.client_id,
"client_secret": self.client_secret,
})
logger.info("Revoked token for scope %s", scope)
return True
except Exception as e:
logger.warning("Token revocation failed: %s", e)
return False
# Usage
oauth_client = OAuthAgentClient(
token_endpoint="https://auth.greenhelix.net/oauth/token",
client_id="agent-payment-processor-01",
client_secret=os.environ["OAUTH_CLIENT_SECRET"],
default_scope="agent:payments agent:billing",
)
# Automatic token management -- just make API calls
result = oauth_client.call_api(
url="https://sandbox.greenhelix.net/v1",
payload={"tool": "get_balance", "input": {"agent_id": "payment-01"}},
scope="agent:billing",
)
print(f"Balance: {result}")
Pattern 4: API Key Management at Scale
API keys are the simplest authentication mechanism and the most commonly misused. They are appropriate for low-sensitivity operations, internal service communication, and development environments. They are not appropriate as the sole authentication mechanism for agents handling financial transactions or sensitive data. When you use API keys, you must layer additional controls on top: key scoping, rotation, monitoring, and rate limiting.
import time
import json
import hashlib
import secrets
import logging
from typing import Optional
from dataclasses import dataclass, field
logger = logging.getLogger(__name__)
@dataclass
class APIKeyRecord:
"""Metadata for an API key."""
key_id: str
key_hash: str # SHA-256 hash, never store the raw key
agent_id: str
scopes: list[str]
created_at: int
expires_at: int
last_used_at: Optional[int] = None
use_count: int = 0
rate_limit_per_minute: int = 60
is_active: bool = True
metadata: dict = field(default_factory=dict)
class APIKeyManager:
"""Manage API keys at scale with scoping, rotation, and monitoring."""
KEY_PREFIX = "ghx_"
KEY_LENGTH = 48 # Characters after prefix
def __init__(self):
self._keys: dict[str, APIKeyRecord] = {} # key_id -> record
self._hash_index: dict[str, str] = {} # key_hash -> key_id
self._agent_keys: dict[str, list[str]] = {} # agent_id -> [key_ids]
self._rate_counts: dict[str, list[float]] = {} # key_id -> [timestamps]
@staticmethod
def _hash_key(raw_key: str) -> str:
return hashlib.sha256(raw_key.encode()).hexdigest()
def generate_key(self, agent_id: str, scopes: list[str],
ttl_seconds: int = 86400,
rate_limit: int = 60,
metadata: Optional[dict] = None) -> tuple[str, APIKeyRecord]:
"""Generate a new API key. Returns (raw_key, record).
IMPORTANT: The raw key is returned exactly once. Store it securely.
This class only stores the hash.
"""
raw_key = self.KEY_PREFIX + secrets.token_urlsafe(self.KEY_LENGTH)
key_hash = self._hash_key(raw_key)
now = int(time.time())
record = APIKeyRecord(
key_id=f"key_{secrets.token_hex(8)}",
key_hash=key_hash,
agent_id=agent_id,
scopes=scopes,
created_at=now,
expires_at=now + ttl_seconds,
rate_limit_per_minute=rate_limit,
metadata=metadata or {},
)
self._keys[record.key_id] = record
self._hash_index[key_hash] = record.key_id
self._agent_keys.setdefault(agent_id, []).append(record.key_id)
logger.info(
"Generated API key %s for agent %s (scopes=%s, ttl=%ds)",
record.key_id, agent_id, scopes, ttl_seconds,
)
return raw_key, record
def validate_key(self, raw_key: str,
required_scope: Optional[str] = None) -> APIKeyRecord:
"""Validate an API key and return its record.
Raises ValueError if the key is invalid, expired, or lacks
the required scope.
"""
key_hash = self._hash_key(raw_key)
key_id = self._hash_index.get(key_hash)
if not key_id or key_id not in self._keys:
raise ValueError("Invalid API key")
record = self._keys[key_id]
if not record.is_active:
raise ValueError(f"API key {record.key_id} is deactivated")
if time.time() >= record.expires_at:
raise ValueError(
f"API key {record.key_id} expired at "
f"{record.expires_at}"
)
if required_scope and required_scope not in record.scopes:
raise ValueError(
f"API key {record.key_id} lacks scope: {required_scope}"
)
# Rate limiting
now = time.time()
timestamps = self._rate_counts.setdefault(key_id, [])
# Remove timestamps older than 1 minute
timestamps[:] = [t for t in timestamps if now - t < 60]
if len(timestamps) >= record.rate_limit_per_minute:
raise ValueError(
f"Rate limit exceeded for key {record.key_id}: "
f"{record.rate_limit_per_minute}/min"
)
timestamps.append(now)
# Update usage stats
record.last_used_at = int(now)
record.use_count += 1
return record
def rotate_key(self, old_key_id: str, grace_period_seconds: int = 300,
) -> tuple[str, APIKeyRecord]:
"""Rotate an API key with a grace period for zero-downtime transition.
The old key remains valid for grace_period_seconds after the new
key is issued.
"""
if old_key_id not in self._keys:
raise KeyError(f"Key {old_key_id} not found")
old_record = self._keys[old_key_id]
# Generate new key with same properties
new_raw_key, new_record = self.generate_key(
agent_id=old_record.agent_id,
scopes=old_record.scopes,
ttl_seconds=old_record.expires_at - old_record.created_at,
rate_limit=old_record.rate_limit_per_minute,
metadata={
**old_record.metadata,
"rotated_from": old_key_id,
},
)
# Schedule old key deactivation
old_record.expires_at = int(time.time()) + grace_period_seconds
old_record.metadata["rotated_to"] = new_record.key_id
old_record.metadata["rotation_grace_until"] = old_record.expires_at
logger.info(
"Rotated key %s -> %s (grace period: %ds)",
old_key_id, new_record.key_id, grace_period_seconds,
)
return new_raw_key, new_record
def revoke_key(self, key_id: str) -> None:
"""Immediately revoke an API key."""
if key_id in self._keys:
self._keys[key_id].is_active = False
logger.info("Revoked API key %s", key_id)
def revoke_all_for_agent(self, agent_id: str) -> int:
"""Revoke all API keys for an agent. Returns count revoked."""
key_ids = self._agent_keys.get(agent_id, [])
count = 0
for key_id in key_ids:
if key_id in self._keys and self._keys[key_id].is_active:
self._keys[key_id].is_active = False
count += 1
logger.info("Revoked %d keys for agent %s", count, agent_id)
return count
def get_agent_keys(self, agent_id: str) -> list[APIKeyRecord]:
"""List all keys for an agent."""
key_ids = self._agent_keys.get(agent_id, [])
return [self._keys[kid] for kid in key_ids if kid in self._keys]
def audit_report(self) -> dict:
"""Generate an audit report of all API keys."""
now = int(time.time())
active = [k for k in self._keys.values() if k.is_active]
expired = [k for k in active if now >= k.expires_at]
unused = [
k for k in active
if k.last_used_at is None
or now - k.last_used_at > 7 * 86400
]
return {
"total_keys": len(self._keys),
"active_keys": len(active),
"expired_active_keys": len(expired),
"unused_7d_keys": len(unused),
"unique_agents": len(self._agent_keys),
"keys_by_scope": self._scope_distribution(),
}
def _scope_distribution(self) -> dict[str, int]:
"""Count keys by scope."""
dist: dict[str, int] = {}
for record in self._keys.values():
for scope in record.scopes:
dist[scope] = dist.get(scope, 0) + 1
return dist
# Usage
key_manager = APIKeyManager()
# Generate scoped keys for different agents
payment_key, payment_record = key_manager.generate_key(
agent_id="payment-processor-01",
scopes=["payments:write", "billing:read"],
ttl_seconds=3600,
rate_limit=120,
)
print(f"Payment key: {payment_key[:20]}...")
reader_key, reader_record = key_manager.generate_key(
agent_id="price-reader-01",
scopes=["marketplace:read"],
ttl_seconds=86400,
rate_limit=30,
)
# Validate with scope check
try:
record = key_manager.validate_key(payment_key, required_scope="payments:write")
print(f"Key valid for agent: {record.agent_id}")
except ValueError as e:
print(f"Key validation failed: {e}")
# Rotate a key
new_key, new_record = key_manager.rotate_key(
payment_record.key_id,
grace_period_seconds=300,
)
print(f"New key: {new_key[:20]}... (old key valid for 5 more minutes)")
# Audit report
report = key_manager.audit_report()
print(json.dumps(report, indent=2))
Choosing the Right Pattern
| Factor | mTLS | JWT | OAuth 2.0 CC | API Keys |
|---|---|---|---|---|
| Security level | Highest | High | High | Medium |
| Operational complexity | High (PKI required) | Medium | Medium (AuthZ server) | Low |
| Performance | TLS handshake overhead | Stateless verification | Token endpoint call | Hash comparison |
| Cross-platform | Excellent (X.509 standard) | Excellent (RFC 7519) | Good (RFC 6749) | Poor (no standard) |
| Revocation | CRL/OCSP or short certs | Short TTL + deny list | Token revocation endpoint | Immediate (DB lookup) |
| Best for | Agent-to-agent direct | Service mesh, delegation | Central auth, enterprise | Internal, low-sensitivity |
For production agent deployments, use mTLS for agent-to-agent communication and OAuth 2.0 client credentials for agent-to-platform authentication. Use JWTs for delegation chains where Agent A authorizes Agent B to act on its behalf. Use API keys only for development and internal tooling.
Chapter 5: Microsoft Agent Governance Toolkit Integration
What the Toolkit Provides
Microsoft released the Agent Governance Toolkit in late 2025 as part of their Azure AI Agent Service. The toolkit addresses a gap that every organization running AI agents encounters: how do you enforce policies on what agents can do, audit what they did do, and ensure compliance across a fleet of autonomous agents?
The toolkit provides four capabilities:
- Policy Engine: Define rules that constrain agent behavior -- which tools an agent can call, what parameters are allowed, what financial limits apply, and what approvals are required.
- Audit Trail: Immutable log of every agent action, including the tool called, the parameters used, the result, and the policy evaluation that authorized (or denied) the action.
- Compliance Dashboard: Real-time visibility into agent behavior, policy violations, and compliance status across the fleet.
- Integration SDK: Python and TypeScript SDKs for integrating the governance layer into your agent runtime.
Policy Enforcement for Agent Actions
The governance toolkit uses a policy-as-code model. Policies are defined in JSON and evaluated at runtime before every agent action. The policy engine supports conditions, thresholds, approval workflows, and time-based restrictions.
import json
import time
import logging
import hashlib
from typing import Optional, Any
from dataclasses import dataclass, field
from enum import Enum
logger = logging.getLogger(__name__)
class PolicyDecision(Enum):
ALLOW = "allow"
DENY = "deny"
REQUIRE_APPROVAL = "require_approval"
@dataclass
class PolicyEvaluation:
"""Result of evaluating a policy against an action."""
decision: PolicyDecision
policy_id: str
reason: str
evaluated_at: int = field(default_factory=lambda: int(time.time()))
conditions_checked: int = 0
approval_required_from: Optional[list[str]] = None
@dataclass
class GovernancePolicy:
"""A governance policy for agent actions."""
policy_id: str
name: str
description: str
target_tools: list[str] # Tools this policy applies to
target_agents: list[str] # Agents this policy applies to ("*" for all)
conditions: list[dict] # Conditions that must be met
decision_on_match: PolicyDecision # What to do when conditions match
priority: int = 0 # Higher priority policies evaluated first
enabled: bool = True
class AgentGovernanceEngine:
"""Policy engine for agent governance, compatible with
Microsoft Agent Governance Toolkit patterns.
Evaluates policies before every agent action and maintains
an immutable audit trail.
"""
def __init__(self, agent_id: str, organization: str):
self.agent_id = agent_id
self.organization = organization
self._policies: list[GovernancePolicy] = []
self._audit_log: list[dict] = []
self._action_counts: dict[str, int] = {} # tool -> count today
self._daily_spend: float = 0.0
def load_policies(self, policies: list[dict]) -> int:
"""Load governance policies from configuration."""
loaded = 0
for p in policies:
policy = GovernancePolicy(
policy_id=p["policy_id"],
name=p["name"],
description=p["description"],
target_tools=p.get("target_tools", ["*"]),
target_agents=p.get("target_agents", ["*"]),
conditions=p.get("conditions", []),
decision_on_match=PolicyDecision(p.get("decision", "allow")),
priority=p.get("priority", 0),
enabled=p.get("enabled", True),
)
self._policies.append(policy)
loaded += 1
# Sort by priority (highest first)
self._policies.sort(key=lambda p: p.priority, reverse=True)
logger.info("Loaded %d governance policies", loaded)
return loaded
def _matches_target(self, policy: GovernancePolicy,
tool: str) -> bool:
"""Check if a policy applies to this tool and agent."""
tool_match = ("*" in policy.target_tools
or tool in policy.target_tools)
agent_match = ("*" in policy.target_agents
or self.agent_id in policy.target_agents)
return tool_match and agent_match
def _evaluate_condition(self, condition: dict,
tool: str, params: dict) -> bool:
"""Evaluate a single policy condition."""
cond_type = condition.get("type")
if cond_type == "max_amount":
amount = params.get("amount", 0)
try:
return float(amount) > float(condition.get("threshold", 0))
except (ValueError, TypeError):
return False
elif cond_type == "daily_limit":
return self._daily_spend > float(
condition.get("threshold", float("inf"))
)
elif cond_type == "rate_limit":
count = self._action_counts.get(tool, 0)
return count >= int(condition.get("max_per_day", float("inf")))
elif cond_type == "blocked_parameter":
param_name = condition.get("parameter")
blocked_values = condition.get("values", [])
return params.get(param_name) in blocked_values
elif cond_type == "required_parameter":
param_name = condition.get("parameter")
return param_name not in params or not params[param_name]
elif cond_type == "time_restriction":
import datetime
now = datetime.datetime.now(datetime.timezone.utc)
allowed_hours = condition.get("allowed_hours", [])
if allowed_hours:
return now.hour not in allowed_hours
elif cond_type == "agent_tier":
required_tier = condition.get("tier")
agent_tier = params.get("_agent_tier", "free")
return agent_tier != required_tier
return False
def evaluate(self, tool: str, params: dict) -> PolicyEvaluation:
"""Evaluate all policies for a proposed agent action."""
conditions_checked = 0
for policy in self._policies:
if not policy.enabled:
continue
if not self._matches_target(policy, tool):
continue
# Evaluate all conditions
all_match = True
for condition in policy.conditions:
conditions_checked += 1
if not self._evaluate_condition(condition, tool, params):
all_match = False
break
if all_match and policy.conditions:
evaluation = PolicyEvaluation(
decision=policy.decision_on_match,
policy_id=policy.policy_id,
reason=f"Policy '{policy.name}' matched: {policy.description}",
conditions_checked=conditions_checked,
)
if policy.decision_on_match == PolicyDecision.REQUIRE_APPROVAL:
evaluation.approval_required_from = policy.conditions[0].get(
"approvers", ["admin"]
)
self._record_audit(tool, params, evaluation)
return evaluation
# No policy triggered -- default allow
evaluation = PolicyEvaluation(
decision=PolicyDecision.ALLOW,
policy_id="default",
reason="No matching policy -- default allow",
conditions_checked=conditions_checked,
)
self._record_audit(tool, params, evaluation)
return evaluation
def execute_with_governance(self, tool: str, params: dict,
execute_fn: Any) -> dict:
"""Execute a tool call with governance policy enforcement."""
evaluation = self.evaluate(tool, params)
if evaluation.decision == PolicyDecision.DENY:
logger.warning(
"DENIED: agent=%s tool=%s policy=%s reason=%s",
self.agent_id, tool, evaluation.policy_id, evaluation.reason,
)
return {
"status": "denied",
"policy_id": evaluation.policy_id,
"reason": evaluation.reason,
}
if evaluation.decision == PolicyDecision.REQUIRE_APPROVAL:
logger.info(
"APPROVAL REQUIRED: agent=%s tool=%s approvers=%s",
self.agent_id, tool, evaluation.approval_required_from,
)
return {
"status": "pending_approval",
"policy_id": evaluation.policy_id,
"reason": evaluation.reason,
"approvers": evaluation.approval_required_from,
}
# Policy allows -- execute
try:
result = execute_fn(tool, params)
# Track usage
self._action_counts[tool] = self._action_counts.get(tool, 0) + 1
if "amount" in params:
try:
self._daily_spend += float(params["amount"])
except (ValueError, TypeError):
pass
return {"status": "executed", "result": result}
except Exception as e:
logger.exception("Execution failed: %s", e)
return {"status": "error", "error": str(e)}
def _record_audit(self, tool: str, params: dict,
evaluation: PolicyEvaluation) -> None:
"""Record an immutable audit log entry."""
# Hash the parameters to avoid logging sensitive data
params_hash = hashlib.sha256(
json.dumps(params, sort_keys=True).encode()
).hexdigest()[:16]
entry = {
"timestamp": int(time.time()),
"agent_id": self.agent_id,
"organization": self.organization,
"tool": tool,
"params_hash": params_hash,
"decision": evaluation.decision.value,
"policy_id": evaluation.policy_id,
"reason": evaluation.reason,
"conditions_checked": evaluation.conditions_checked,
}
self._audit_log.append(entry)
def get_audit_log(self, limit: int = 100) -> list[dict]:
"""Retrieve recent audit log entries."""
return self._audit_log[-limit:]
def get_compliance_summary(self) -> dict:
"""Generate a compliance summary for this agent."""
total = len(self._audit_log)
denied = len([e for e in self._audit_log if e["decision"] == "deny"])
approved = len([e for e in self._audit_log if e["decision"] == "allow"])
pending = len([
e for e in self._audit_log
if e["decision"] == "require_approval"
])
return {
"agent_id": self.agent_id,
"organization": self.organization,
"total_actions": total,
"allowed": approved,
"denied": denied,
"pending_approval": pending,
"denial_rate": f"{(denied/total*100):.1f}%" if total else "0%",
"daily_spend": f"${self._daily_spend:.2f}",
"actions_by_tool": dict(self._action_counts),
}
# Policy definitions
policies = [
{
"policy_id": "fin-001",
"name": "Large Transaction Approval",
"description": "Transactions over $1000 require human approval",
"target_tools": ["create_escrow", "deposit", "create_split_intent"],
"target_agents": ["*"],
"conditions": [
{
"type": "max_amount",
"threshold": 1000,
"approvers": ["finance-team@acme.com"],
}
],
"decision": "require_approval",
"priority": 100,
},
{
"policy_id": "fin-002",
"name": "Daily Spend Limit",
"description": "Block all transactions when daily spend exceeds $10,000",
"target_tools": ["create_escrow", "deposit", "create_split_intent"],
"target_agents": ["*"],
"conditions": [
{"type": "daily_limit", "threshold": 10000}
],
"decision": "deny",
"priority": 200,
},
{
"policy_id": "sec-001",
"name": "Identity Operation Rate Limit",
"description": "Limit identity operations to 100 per day",
"target_tools": ["register_agent", "verify_agent"],
"target_agents": ["*"],
"conditions": [
{"type": "rate_limit", "max_per_day": 100}
],
"decision": "deny",
"priority": 50,
},
{
"policy_id": "sec-002",
"name": "Block Admin Self-Escalation",
"description": "Agents cannot grant themselves admin roles",
"target_tools": ["register_agent"],
"target_agents": ["*"],
"conditions": [
{"type": "blocked_parameter", "parameter": "roles", "values": [["admin"]]}
],
"decision": "deny",
"priority": 300,
},
]
# Usage
governance = AgentGovernanceEngine(
agent_id="payment-processor-01",
organization="acme-corp",
)
governance.load_policies(policies)
# Execute with governance
def mock_execute(tool, params):
return {"success": True, "tool": tool}
# Small transaction -- allowed
result = governance.execute_with_governance(
"deposit",
{"agent_id": "payment-01", "amount": "50.00"},
mock_execute,
)
print(f"Small deposit: {result['status']}")
# Large transaction -- requires approval
result = governance.execute_with_governance(
"create_escrow",
{"agent_id": "payment-01", "amount": "5000.00", "payee": "seller-01"},
mock_execute,
)
print(f"Large escrow: {result['status']}")
# Compliance summary
summary = governance.get_compliance_summary()
print(json.dumps(summary, indent=2))
Audit Trail Integration
The governance engine produces audit entries that must be stored immutably for compliance. In production, pipe these to an append-only data store: AWS CloudTrail, Azure Monitor, GCP Cloud Audit Logs, or a dedicated SIEM. The following adapter formats audit entries for common destinations.
import json
import time
import logging
from typing import Optional
logger = logging.getLogger(__name__)
class AuditTrailAdapter:
"""Adapts governance audit entries for external audit systems."""
def __init__(self, organization: str, environment: str = "production"):
self.organization = organization
self.environment = environment
def to_cloudtrail_format(self, entry: dict) -> dict:
"""Format an audit entry for AWS CloudTrail custom events."""
return {
"eventVersion": "1.08",
"eventTime": time.strftime(
"%Y-%m-%dT%H:%M:%SZ", time.gmtime(entry["timestamp"])
),
"eventSource": "agent-governance.greenhelix.net",
"eventName": f"AgentAction:{entry['tool']}",
"awsRegion": "us-east-1",
"sourceIPAddress": "agent-internal",
"userIdentity": {
"type": "AssumedRole",
"arn": f"arn:aws:sts::agent/{entry['agent_id']}",
"principalId": entry["agent_id"],
},
"requestParameters": {
"tool": entry["tool"],
"paramsHash": entry["params_hash"],
},
"responseElements": {
"decision": entry["decision"],
"policyId": entry["policy_id"],
},
"additionalEventData": {
"organization": self.organization,
"environment": self.environment,
"conditionsChecked": entry["conditions_checked"],
},
}
def to_siem_format(self, entry: dict) -> dict:
"""Format an audit entry for SIEM ingestion (Splunk, Elastic, etc)."""
severity = "INFO"
if entry["decision"] == "deny":
severity = "WARNING"
elif entry["decision"] == "require_approval":
severity = "NOTICE"
return {
"@timestamp": time.strftime(
"%Y-%m-%dT%H:%M:%SZ", time.gmtime(entry["timestamp"])
),
"event.kind": "event",
"event.category": "iam",
"event.type": "access",
"event.action": entry["tool"],
"event.outcome": (
"success" if entry["decision"] == "allow" else "failure"
),
"agent.id": entry["agent_id"],
"organization.name": self.organization,
"rule.id": entry["policy_id"],
"rule.description": entry["reason"],
"log.level": severity,
"greenhelix.params_hash": entry["params_hash"],
"greenhelix.conditions_checked": entry["conditions_checked"],
}
def to_compliance_record(self, entry: dict) -> dict:
"""Format an audit entry for compliance reporting (SOC 2, ISO 27001)."""
return {
"record_id": f"{entry['agent_id']}:{entry['timestamp']}:{entry['tool']}",
"timestamp_utc": time.strftime(
"%Y-%m-%dT%H:%M:%SZ", time.gmtime(entry["timestamp"])
),
"actor_type": "non_human_identity",
"actor_id": entry["agent_id"],
"organization": self.organization,
"action": entry["tool"],
"resource_type": "agent_tool",
"authorization_decision": entry["decision"],
"authorization_policy": entry["policy_id"],
"authorization_reason": entry["reason"],
"risk_indicators": {
"denied": entry["decision"] == "deny",
"required_approval": entry["decision"] == "require_approval",
"high_value": "escrow" in entry["tool"] or "deposit" in entry["tool"],
},
"data_hash": entry["params_hash"],
"environment": self.environment,
}
# Usage
adapter = AuditTrailAdapter(organization="acme-corp")
audit_log = governance.get_audit_log()
for entry in audit_log:
# Send to CloudTrail
ct_event = adapter.to_cloudtrail_format(entry)
# boto3.client('cloudtrail').put_events(...)
# Send to SIEM
siem_event = adapter.to_siem_format(entry)
# requests.post("https://siem.internal/api/events", json=siem_event)
# Store compliance record
compliance = adapter.to_compliance_record(entry)
print(json.dumps(compliance, indent=2))
Chapter 6: Zero Trust for AI Agents
Why Traditional Perimeter Security Fails for Agent Fleets
The traditional network security model assumes that everything inside the corporate network is trusted and everything outside is hostile. This model was already failing before AI agents. With agent fleets, it is completely broken.
Consider a typical agent deployment. An orchestrator agent running in your Kubernetes cluster hires a sub-agent from a marketplace. That sub-agent runs in the marketplace provider's infrastructure. The sub-agent calls a payment processing tool hosted in a third-party cloud. The payment tool communicates with a bank API over the public internet. At no point in this chain does a meaningful network perimeter exist. The orchestrator, the sub-agent, the payment tool, and the bank API are all in different trust domains, different networks, and different organizations.
Even within a single organization, agents violate perimeter assumptions. A price-feed agent in the analytics namespace should not be able to call the refund tool in the payments namespace. But if both namespaces share a flat network (as most Kubernetes clusters do by default), the network provides no isolation. The 2025 CISA Zero Trust Maturity Model found that 73% of organizations running microservices had at least one namespace with unrestricted east-west traffic. For organizations running AI agents, the figure was 89%, because agent communication patterns are dynamic and hard to predict at deployment time.
Zero trust for AI agents means: never trust any agent based on its network location, its IP address, or the fact that it is running inside your cluster. Every request from every agent must be authenticated, authorized, and validated -- every time, regardless of origin.
The Five Pillars of Agent Zero Trust
NIST SP 800-207 defines zero trust architecture around five pillars. Here is how each pillar maps to AI agent deployments.
| Pillar | NIST Definition | Agent Implementation |
|---|---|---|
| Identity | Verify the identity of all subjects | SPIFFE IDs for every agent, mTLS for every connection |
| Device | Assess the security posture of the device | Workload attestation: verify agent binary hash, runtime environment, Kubernetes labels |
| Network | Segment and encrypt all traffic | Microsegmentation with per-agent network policies, no flat networks |
| Application | Secure the application layer | Per-tool authorization policies, input validation, output filtering |
| Data | Protect data at rest and in transit | Encrypt all agent-to-agent payloads, classify data accessed by each agent |
The critical difference between human zero trust and agent zero trust is the verification frequency. A human authenticates once and gets a session. An agent must be verified on every single request because agents operate at machine speed -- a compromised agent can exfiltrate data or execute unauthorized actions in milliseconds, far faster than any session-based detection can respond.
Policy Decision Point Architecture
The core of a zero trust architecture is the Policy Decision Point (PDP). Every agent request flows through the PDP, which evaluates the request against the current policy set and returns an allow or deny decision. The PDP never caches decisions for longer than a single request. This is the "never trust, always verify" principle in practice.
import time
import json
import hashlib
import logging
from dataclasses import dataclass, field
from typing import Optional
from enum import Enum
logger = logging.getLogger(__name__)
class TrustLevel(Enum):
"""Trust levels for agent requests. Every request starts at NONE."""
NONE = 0
IDENTIFIED = 1 # Agent identity verified
ATTESTED = 2 # Workload attestation passed
AUTHORIZED = 3 # Policy allows this specific action
VERIFIED = 4 # Request content validated
@dataclass
class AgentContext:
"""Security context for an agent making a request."""
agent_id: str
spiffe_id: str
source_namespace: str
source_node: str
binary_hash: str
runtime_environment: str # "kubernetes", "ecs", "vm", "sandbox"
labels: dict = field(default_factory=dict)
trust_level: TrustLevel = TrustLevel.NONE
attestation_time: float = 0.0
identity_verified_at: float = 0.0
@dataclass
class AccessRequest:
"""A request from one agent to access a resource."""
source: AgentContext
target_tool: str
target_agent_id: str
parameters: dict
request_id: str
timestamp: float = field(default_factory=time.time)
@dataclass
class PolicyDecisionResult:
"""Result from the Policy Decision Point."""
allowed: bool
trust_level_achieved: TrustLevel
reason: str
policy_id: str
evaluation_time_ms: float
conditions_evaluated: int
request_id: str
timestamp: float = field(default_factory=time.time)
class PolicyDecisionPoint:
"""Zero trust Policy Decision Point for agent requests.
Every request is evaluated independently. No cached decisions.
No implicit trust based on previous requests.
"""
# Maximum age of identity verification before re-verification required
IDENTITY_MAX_AGE_SECONDS = 300 # 5 minutes
# Maximum age of workload attestation
ATTESTATION_MAX_AGE_SECONDS = 60 # 1 minute
# Maximum parameter payload size
MAX_PARAMS_SIZE_BYTES = 65_536 # 64 KB
def __init__(self):
self._policies: list[dict] = []
self._deny_list: set[str] = set() # Blocked agent IDs
self._tool_permissions: dict[str, dict] = {} # tool -> permission map
self._decision_log: list[PolicyDecisionResult] = []
self._namespace_policies: dict[str, list[str]] = {} # ns -> allowed tools
def load_policy(self, policy: dict) -> None:
"""Load a zero trust policy."""
self._policies.append(policy)
# Index tool permissions for fast lookup
for rule in policy.get("rules", []):
tool = rule["tool"]
self._tool_permissions[tool] = {
"allowed_agents": set(rule.get("allowed_agents", [])),
"allowed_namespaces": set(rule.get("allowed_namespaces", [])),
"required_trust_level": TrustLevel[
rule.get("required_trust_level", "VERIFIED")
],
"max_params_size": rule.get(
"max_params_size", self.MAX_PARAMS_SIZE_BYTES
),
"allowed_hours": rule.get("allowed_hours"), # e.g., [9, 17]
"rate_limit_per_minute": rule.get("rate_limit_per_minute", 60),
}
def add_namespace_policy(self, namespace: str,
allowed_tools: list[str]) -> None:
"""Define which tools a namespace can access (microsegmentation)."""
self._namespace_policies[namespace] = allowed_tools
def block_agent(self, agent_id: str, reason: str) -> None:
"""Immediately block an agent. Takes effect on next request."""
self._deny_list.add(agent_id)
logger.warning(
"Agent %s added to deny list: %s", agent_id, reason
)
def _verify_identity(self, ctx: AgentContext) -> tuple[bool, str]:
"""Step 1: Verify agent identity is valid and current."""
if ctx.agent_id in self._deny_list:
return False, f"Agent {ctx.agent_id} is on the deny list"
if not ctx.spiffe_id:
return False, "No SPIFFE ID presented"
if not ctx.spiffe_id.startswith("spiffe://"):
return False, f"Invalid SPIFFE ID format: {ctx.spiffe_id}"
# Check identity freshness
age = time.time() - ctx.identity_verified_at
if age > self.IDENTITY_MAX_AGE_SECONDS:
return False, (
f"Identity verification too old: {age:.0f}s "
f"(max {self.IDENTITY_MAX_AGE_SECONDS}s)"
)
return True, "Identity verified"
def _verify_attestation(self, ctx: AgentContext) -> tuple[bool, str]:
"""Step 2: Verify workload attestation is valid and current."""
if not ctx.binary_hash:
return False, "No binary hash for workload attestation"
attestation_age = time.time() - ctx.attestation_time
if attestation_age > self.ATTESTATION_MAX_AGE_SECONDS:
return False, (
f"Attestation too old: {attestation_age:.0f}s "
f"(max {self.ATTESTATION_MAX_AGE_SECONDS}s)"
)
# In production, verify binary_hash against a known-good registry
# of approved agent binaries
if len(ctx.binary_hash) != 64: # SHA-256 hex length
return False, "Invalid binary hash format"
return True, "Workload attestation verified"
def _verify_authorization(self, request: AccessRequest) -> tuple[bool, str]:
"""Step 3: Check if this agent is authorized for this tool."""
tool_policy = self._tool_permissions.get(request.target_tool)
if tool_policy is None:
return False, (
f"No policy defined for tool {request.target_tool} "
f"-- default deny"
)
# Check agent allowlist
allowed_agents = tool_policy["allowed_agents"]
if allowed_agents and request.source.agent_id not in allowed_agents:
return False, (
f"Agent {request.source.agent_id} not in allowlist "
f"for {request.target_tool}"
)
# Check namespace microsegmentation
ns = request.source.source_namespace
if ns in self._namespace_policies:
allowed_tools = self._namespace_policies[ns]
if request.target_tool not in allowed_tools:
return False, (
f"Namespace {ns} not authorized for "
f"tool {request.target_tool}"
)
allowed_ns = tool_policy["allowed_namespaces"]
if allowed_ns and ns not in allowed_ns:
return False, (
f"Namespace {ns} not in allowlist for {request.target_tool}"
)
# Check time-based restrictions
if tool_policy.get("allowed_hours"):
current_hour = time.gmtime().tm_hour
start, end = tool_policy["allowed_hours"]
if not (start <= current_hour < end):
return False, (
f"Tool {request.target_tool} restricted to hours "
f"{start}-{end} UTC, current hour is {current_hour}"
)
return True, "Authorization granted"
def _verify_request_content(self, request: AccessRequest) -> tuple[bool, str]:
"""Step 4: Validate request content."""
params_json = json.dumps(request.parameters)
params_size = len(params_json.encode())
tool_policy = self._tool_permissions.get(request.target_tool, {})
max_size = tool_policy.get(
"max_params_size", self.MAX_PARAMS_SIZE_BYTES
)
if params_size > max_size:
return False, (
f"Parameters size {params_size} exceeds limit {max_size}"
)
# Check for injection patterns in parameter values
dangerous_patterns = [
"__import__", "eval(", "exec(", "os.system",
"subprocess", "'; DROP", "$(", "`",
]
for key, value in request.parameters.items():
if isinstance(value, str):
for pattern in dangerous_patterns:
if pattern in value:
return False, (
f"Dangerous pattern '{pattern}' in "
f"parameter '{key}'"
)
return True, "Request content validated"
def evaluate(self, request: AccessRequest) -> PolicyDecisionResult:
"""Evaluate a request through all zero trust verification steps.
Every step must pass. Failure at any step results in denial.
No partial trust. No fallback to a lower trust level.
"""
start = time.monotonic()
conditions = 0
# Step 1: Identity
conditions += 1
ok, reason = self._verify_identity(request.source)
if not ok:
return self._make_decision(
False, TrustLevel.NONE, reason,
"zt-identity", conditions, request.request_id, start,
)
request.source.trust_level = TrustLevel.IDENTIFIED
# Step 2: Attestation
conditions += 1
ok, reason = self._verify_attestation(request.source)
if not ok:
return self._make_decision(
False, TrustLevel.IDENTIFIED, reason,
"zt-attestation", conditions, request.request_id, start,
)
request.source.trust_level = TrustLevel.ATTESTED
# Step 3: Authorization
conditions += 1
ok, reason = self._verify_authorization(request)
if not ok:
return self._make_decision(
False, TrustLevel.ATTESTED, reason,
"zt-authorization", conditions, request.request_id, start,
)
request.source.trust_level = TrustLevel.AUTHORIZED
# Step 4: Content validation
conditions += 1
ok, reason = self._verify_request_content(request)
if not ok:
return self._make_decision(
False, TrustLevel.AUTHORIZED, reason,
"zt-content", conditions, request.request_id, start,
)
request.source.trust_level = TrustLevel.VERIFIED
return self._make_decision(
True, TrustLevel.VERIFIED,
"All zero trust checks passed",
"zt-allow", conditions, request.request_id, start,
)
def _make_decision(self, allowed: bool, trust_level: TrustLevel,
reason: str, policy_id: str,
conditions: int, request_id: str,
start: float) -> PolicyDecisionResult:
elapsed_ms = (time.monotonic() - start) * 1000
result = PolicyDecisionResult(
allowed=allowed,
trust_level_achieved=trust_level,
reason=reason,
policy_id=policy_id,
evaluation_time_ms=elapsed_ms,
conditions_evaluated=conditions,
request_id=request_id,
)
self._decision_log.append(result)
level = logging.INFO if allowed else logging.WARNING
logger.log(
level,
"PDP decision: %s (trust=%s, reason=%s, time=%.1fms)",
"ALLOW" if allowed else "DENY",
trust_level.name,
reason,
elapsed_ms,
)
return result
def get_decision_stats(self) -> dict:
"""Get statistics on PDP decisions for monitoring."""
if not self._decision_log:
return {"total": 0}
total = len(self._decision_log)
allowed = sum(1 for d in self._decision_log if d.allowed)
denied = total - allowed
denial_reasons: dict[str, int] = {}
for d in self._decision_log:
if not d.allowed:
denial_reasons[d.policy_id] = (
denial_reasons.get(d.policy_id, 0) + 1
)
avg_time = sum(
d.evaluation_time_ms for d in self._decision_log
) / total
return {
"total_decisions": total,
"allowed": allowed,
"denied": denied,
"denial_rate": denied / total if total > 0 else 0,
"denial_reasons": denial_reasons,
"avg_evaluation_time_ms": round(avg_time, 2),
"p99_evaluation_time_ms": round(
sorted(d.evaluation_time_ms for d in self._decision_log)[
int(total * 0.99)
], 2
),
}
# Usage
pdp = PolicyDecisionPoint()
# Load tool-level policies
pdp.load_policy({
"rules": [
{
"tool": "create_payment",
"allowed_agents": [
"payment-processor-01", "payment-processor-02",
],
"allowed_namespaces": ["payments"],
"required_trust_level": "VERIFIED",
"rate_limit_per_minute": 30,
},
{
"tool": "read_price_feed",
"allowed_agents": [], # Empty = any authenticated agent
"allowed_namespaces": ["analytics", "payments", "marketplace"],
"required_trust_level": "ATTESTED",
"rate_limit_per_minute": 120,
},
]
})
# Define namespace microsegmentation
pdp.add_namespace_policy("analytics", ["read_price_feed", "get_market_data"])
pdp.add_namespace_policy("payments", [
"create_payment", "read_price_feed", "check_balance",
])
# Evaluate a request
agent_ctx = AgentContext(
agent_id="payment-processor-01",
spiffe_id="spiffe://greenhelix.net/agent/payment-processor-01",
source_namespace="payments",
source_node="node-pool-1-abc",
binary_hash="a3f2b8c1d4e5f67890abcdef1234567890abcdef1234567890abcdef12345678",
runtime_environment="kubernetes",
labels={"tier": "pro", "team": "platform"},
identity_verified_at=time.time() - 30,
attestation_time=time.time() - 10,
)
request = AccessRequest(
source=agent_ctx,
target_tool="create_payment",
target_agent_id="gateway-01",
parameters={"amount": "50.00", "currency": "USD", "recipient": "agent-42"},
request_id="req-abc-123",
)
decision = pdp.evaluate(request)
print(f"Decision: {'ALLOW' if decision.allowed else 'DENY'}")
print(f"Trust level: {decision.trust_level_achieved.name}")
print(f"Reason: {decision.reason}")
print(f"Evaluation time: {decision.evaluation_time_ms:.1f}ms")
Microsegmentation for Agent Communication
In a flat network, any agent can talk to any other agent. Microsegmentation restricts communication so that each agent can only reach the specific agents and tools it needs. This limits the blast radius of a compromised agent. If the price-feed agent is compromised, the attacker cannot pivot to the payment processing tools because the network policy blocks the traffic before it reaches the application layer.
The following class generates Kubernetes NetworkPolicy resources and application-layer segmentation rules from a declarative agent communication map.
import json
import time
import logging
from dataclasses import dataclass, field
from typing import Optional
logger = logging.getLogger(__name__)
@dataclass
class AgentCommunicationRule:
"""A permitted communication path between agents."""
source_agent: str
source_namespace: str
target_agent: str
target_namespace: str
allowed_tools: list[str]
allowed_ports: list[int]
protocol: str = "TCP"
max_request_rate: int = 60 # per minute
encrypted_only: bool = True
class AgentMicrosegmentation:
"""Generate and enforce microsegmentation policies for agent fleets.
Produces Kubernetes NetworkPolicy YAML and application-layer
enforcement rules from a declarative communication map.
"""
def __init__(self, cluster_name: str, trust_domain: str):
self.cluster_name = cluster_name
self.trust_domain = trust_domain
self._rules: list[AgentCommunicationRule] = []
self._default_deny: bool = True
def add_rule(self, rule: AgentCommunicationRule) -> None:
"""Add a communication rule."""
self._rules.append(rule)
logger.info(
"Added rule: %s/%s -> %s/%s (tools=%s)",
rule.source_namespace, rule.source_agent,
rule.target_namespace, rule.target_agent,
rule.allowed_tools,
)
def generate_network_policies(self) -> list[dict]:
"""Generate Kubernetes NetworkPolicy resources.
Produces one NetworkPolicy per target namespace that specifies
which source namespaces and pods can send traffic.
"""
# Group rules by target namespace
by_target_ns: dict[str, list[AgentCommunicationRule]] = {}
for rule in self._rules:
ns = rule.target_namespace
by_target_ns.setdefault(ns, []).append(rule)
policies = []
for target_ns, rules in by_target_ns.items():
# Default deny all ingress
if self._default_deny:
policies.append({
"apiVersion": "networking.k8s.io/v1",
"kind": "NetworkPolicy",
"metadata": {
"name": f"default-deny-ingress-{target_ns}",
"namespace": target_ns,
},
"spec": {
"podSelector": {},
"policyTypes": ["Ingress"],
},
})
# Allow specific ingress per rule
for rule in rules:
policy = {
"apiVersion": "networking.k8s.io/v1",
"kind": "NetworkPolicy",
"metadata": {
"name": (
f"allow-{rule.source_agent}-to-"
f"{rule.target_agent}"
),
"namespace": target_ns,
"labels": {
"app.kubernetes.io/managed-by": "nhi-segmentation",
"greenhelix.net/source-agent": rule.source_agent,
},
},
"spec": {
"podSelector": {
"matchLabels": {
"agent-name": rule.target_agent,
},
},
"policyTypes": ["Ingress"],
"ingress": [
{
"from": [
{
"namespaceSelector": {
"matchLabels": {
"kubernetes.io/metadata.name":
rule.source_namespace,
},
},
"podSelector": {
"matchLabels": {
"agent-name": rule.source_agent,
},
},
},
],
"ports": [
{
"protocol": rule.protocol,
"port": port,
}
for port in rule.allowed_ports
],
},
],
},
}
policies.append(policy)
return policies
def generate_application_rules(self) -> list[dict]:
"""Generate application-layer enforcement rules.
These rules are consumed by the PDP to enforce tool-level
access control on top of network-level segmentation.
"""
app_rules = []
for rule in self._rules:
app_rules.append({
"source_spiffe_id": (
f"spiffe://{self.trust_domain}"
f"/ns/{rule.source_namespace}"
f"/agent/{rule.source_agent}"
),
"target_spiffe_id": (
f"spiffe://{self.trust_domain}"
f"/ns/{rule.target_namespace}"
f"/agent/{rule.target_agent}"
),
"allowed_tools": rule.allowed_tools,
"max_request_rate": rule.max_request_rate,
"encrypted_only": rule.encrypted_only,
})
return app_rules
def validate_communication(self, source_agent: str,
source_namespace: str,
target_agent: str,
target_namespace: str,
tool: str) -> tuple[bool, str]:
"""Check if a communication path is permitted."""
for rule in self._rules:
if (rule.source_agent == source_agent
and rule.source_namespace == source_namespace
and rule.target_agent == target_agent
and rule.target_namespace == target_namespace
and tool in rule.allowed_tools):
return True, f"Permitted by rule: {rule.source_agent} -> {rule.target_agent}"
if self._default_deny:
return False, (
f"No rule permits {source_namespace}/{source_agent} -> "
f"{target_namespace}/{target_agent} for tool {tool}"
)
return True, "Default allow (warning: default deny not enabled)"
def audit_report(self) -> dict:
"""Generate an audit report of all segmentation rules."""
return {
"cluster": self.cluster_name,
"trust_domain": self.trust_domain,
"default_deny": self._default_deny,
"total_rules": len(self._rules),
"namespaces": list({r.target_namespace for r in self._rules}),
"communication_paths": [
{
"source": f"{r.source_namespace}/{r.source_agent}",
"target": f"{r.target_namespace}/{r.target_agent}",
"tools": r.allowed_tools,
"ports": r.allowed_ports,
"rate_limit": r.max_request_rate,
}
for r in self._rules
],
"generated_at": time.strftime(
"%Y-%m-%dT%H:%M:%SZ", time.gmtime()
),
}
# Usage
segmentation = AgentMicrosegmentation(
cluster_name="prod-agents-us-east-1",
trust_domain="greenhelix.net",
)
# Define allowed communication paths
segmentation.add_rule(AgentCommunicationRule(
source_agent="orchestrator-01",
source_namespace="orchestration",
target_agent="payment-processor-01",
target_namespace="payments",
allowed_tools=["create_payment", "check_balance", "get_payment_status"],
allowed_ports=[8443],
max_request_rate=30,
))
segmentation.add_rule(AgentCommunicationRule(
source_agent="orchestrator-01",
source_namespace="orchestration",
target_agent="price-feed-01",
target_namespace="analytics",
allowed_tools=["read_price_feed", "get_market_data"],
allowed_ports=[8443],
max_request_rate=120,
))
segmentation.add_rule(AgentCommunicationRule(
source_agent="payment-processor-01",
source_namespace="payments",
target_agent="ledger-agent-01",
target_namespace="accounting",
allowed_tools=["record_transaction", "get_balance"],
allowed_ports=[8443],
max_request_rate=60,
))
# Generate Kubernetes NetworkPolicies
k8s_policies = segmentation.generate_network_policies()
for policy in k8s_policies:
print(json.dumps(policy, indent=2))
# Generate application-layer rules for the PDP
app_rules = segmentation.generate_application_rules()
print(json.dumps(app_rules, indent=2))
# Validate a communication path
allowed, reason = segmentation.validate_communication(
"orchestrator-01", "orchestration",
"payment-processor-01", "payments",
"create_payment",
)
print(f"Allowed: {allowed}, Reason: {reason}")
SPIFFE/SPIRE as the Agent Identity Foundation
Chapter 3 introduced SPIFFE identities for workload identity federation. In a zero trust architecture, SPIFFE/SPIRE becomes the identity foundation that everything else builds on. Every agent gets a SPIFFE ID. Every connection uses SPIFFE-based mTLS. Every policy decision starts with SPIFFE identity verification.
The SPIFFE ID format for agents follows a hierarchy that encodes trust domain, environment, namespace, and agent identity:
spiffe://greenhelix.net/env/production/ns/payments/agent/payment-processor-01
This hierarchy enables policies at multiple granularity levels. You can write a policy that allows all agents in the payments namespace to access billing tools. You can write a policy that restricts a specific agent to a specific tool. You can write a policy that denies all agents in the staging environment from accessing production resources.
SPIRE provides two critical capabilities that static certificates do not:
1. Workload attestation. SPIRE does not just issue a certificate to any process that asks. It verifies the workload's identity using platform-specific attestors. On Kubernetes, the SPIRE agent uses the Kubernetes API to verify that the pod requesting an identity matches the expected service account, namespace, and labels. On AWS, it uses the instance metadata service to verify the instance ID, region, and IAM role. This means a compromised process cannot impersonate another agent simply by requesting its SPIFFE ID -- the attestor will reject the request because the workload metadata does not match.
2. Short-lived, auto-rotating SVIDs. SPIRE issues X.509 SVIDs (SPIFFE Verifiable Identity Documents) with short lifetimes -- typically 1 hour. The SPIRE agent automatically renews SVIDs before they expire. If an SVID is stolen, it is valid for at most 1 hour. Compare this with traditional PKI where certificates are valid for 1-2 years and revocation is slow and unreliable.
Continuous Verification Engine
Zero trust requires continuous verification -- not just at connection time, but throughout the lifetime of every session. The following engine runs periodic checks on active agent sessions and terminates sessions that fail re-verification.
import time
import threading
import logging
from dataclasses import dataclass, field
from typing import Optional, Callable
logger = logging.getLogger(__name__)
@dataclass
class ActiveSession:
"""An active agent-to-agent session under continuous verification."""
session_id: str
source_agent_id: str
target_agent_id: str
spiffe_id: str
started_at: float
last_verified_at: float
request_count: int = 0
bytes_transferred: int = 0
anomaly_score: float = 0.0
is_active: bool = True
termination_reason: Optional[str] = None
class ContinuousVerificationEngine:
"""Continuously verify active agent sessions.
Runs background checks every verification_interval_seconds.
Sessions that fail re-verification are terminated immediately.
"""
def __init__(self, verification_interval: int = 30,
anomaly_threshold: float = 0.8,
max_session_duration: int = 3600,
max_requests_per_session: int = 10_000):
self.verification_interval = verification_interval
self.anomaly_threshold = anomaly_threshold
self.max_session_duration = max_session_duration
self.max_requests_per_session = max_requests_per_session
self._sessions: dict[str, ActiveSession] = {}
self._identity_checker: Optional[Callable] = None
self._attestation_checker: Optional[Callable] = None
self._running = False
self._thread: Optional[threading.Thread] = None
self._terminated_sessions: list[dict] = []
def register_identity_checker(self, checker: Callable) -> None:
"""Register a function that re-verifies agent identity.
The function takes a SPIFFE ID and returns (valid: bool, reason: str).
"""
self._identity_checker = checker
def register_attestation_checker(self, checker: Callable) -> None:
"""Register a function that re-verifies workload attestation.
The function takes an agent_id and returns (valid: bool, reason: str).
"""
self._attestation_checker = checker
def track_session(self, session: ActiveSession) -> None:
"""Start tracking a new session for continuous verification."""
self._sessions[session.session_id] = session
logger.info(
"Tracking session %s: %s -> %s",
session.session_id,
session.source_agent_id,
session.target_agent_id,
)
def record_activity(self, session_id: str,
request_bytes: int = 0) -> None:
"""Record activity on a session (call from request middleware)."""
session = self._sessions.get(session_id)
if session and session.is_active:
session.request_count += 1
session.bytes_transferred += request_bytes
def _verify_session(self, session: ActiveSession) -> tuple[bool, str]:
"""Run all verification checks on a session."""
now = time.time()
# Check session duration
duration = now - session.started_at
if duration > self.max_session_duration:
return False, (
f"Session exceeded max duration: "
f"{duration:.0f}s > {self.max_session_duration}s"
)
# Check request count
if session.request_count > self.max_requests_per_session:
return False, (
f"Session exceeded max requests: "
f"{session.request_count} > {self.max_requests_per_session}"
)
# Re-verify identity
if self._identity_checker:
valid, reason = self._identity_checker(session.spiffe_id)
if not valid:
return False, f"Identity re-verification failed: {reason}"
# Re-verify attestation
if self._attestation_checker:
valid, reason = self._attestation_checker(
session.source_agent_id
)
if not valid:
return False, f"Attestation re-verification failed: {reason}"
# Check anomaly score
if session.anomaly_score > self.anomaly_threshold:
return False, (
f"Anomaly score {session.anomaly_score:.2f} exceeds "
f"threshold {self.anomaly_threshold:.2f}"
)
return True, "All continuous verification checks passed"
def _terminate_session(self, session: ActiveSession,
reason: str) -> None:
"""Terminate a session that failed verification."""
session.is_active = False
session.termination_reason = reason
record = {
"session_id": session.session_id,
"source_agent": session.source_agent_id,
"target_agent": session.target_agent_id,
"duration_seconds": time.time() - session.started_at,
"request_count": session.request_count,
"bytes_transferred": session.bytes_transferred,
"termination_reason": reason,
"terminated_at": time.strftime(
"%Y-%m-%dT%H:%M:%SZ", time.gmtime()
),
}
self._terminated_sessions.append(record)
logger.warning(
"TERMINATED session %s (%s -> %s): %s",
session.session_id,
session.source_agent_id,
session.target_agent_id,
reason,
)
def _verification_loop(self) -> None:
"""Background loop that continuously verifies all active sessions."""
while self._running:
sessions = list(self._sessions.values())
for session in sessions:
if not session.is_active:
continue
valid, reason = self._verify_session(session)
if valid:
session.last_verified_at = time.time()
else:
self._terminate_session(session, reason)
time.sleep(self.verification_interval)
def start(self) -> None:
"""Start the continuous verification background thread."""
self._running = True
self._thread = threading.Thread(
target=self._verification_loop,
daemon=True,
name="continuous-verification",
)
self._thread.start()
logger.info(
"Continuous verification started (interval=%ds)",
self.verification_interval,
)
def stop(self) -> None:
"""Stop the continuous verification background thread."""
self._running = False
if self._thread:
self._thread.join(timeout=5)
logger.info("Continuous verification stopped")
def get_status(self) -> dict:
"""Get current verification engine status."""
active = [s for s in self._sessions.values() if s.is_active]
return {
"active_sessions": len(active),
"total_tracked": len(self._sessions),
"terminated_sessions": len(self._terminated_sessions),
"verification_interval_seconds": self.verification_interval,
"recent_terminations": self._terminated_sessions[-10:],
}
# Usage
engine = ContinuousVerificationEngine(
verification_interval=30, # Re-verify every 30 seconds
anomaly_threshold=0.8, # Terminate if anomaly score > 0.8
max_session_duration=3600, # Max 1 hour sessions
max_requests_per_session=10_000,
)
# Register verification callbacks
def check_identity(spiffe_id: str) -> tuple[bool, str]:
"""Check if SPIFFE ID is still valid (not revoked)."""
# In production: query SPIRE server or check CRL
if "revoked" in spiffe_id:
return False, "SPIFFE ID has been revoked"
return True, "SPIFFE ID valid"
def check_attestation(agent_id: str) -> tuple[bool, str]:
"""Re-verify workload attestation."""
# In production: re-run attestation via SPIRE agent
return True, "Attestation current"
engine.register_identity_checker(check_identity)
engine.register_attestation_checker(check_attestation)
engine.start()
# Track a new session
session = ActiveSession(
session_id="sess-001",
source_agent_id="orchestrator-01",
target_agent_id="payment-processor-01",
spiffe_id="spiffe://greenhelix.net/agent/orchestrator-01",
started_at=time.time(),
last_verified_at=time.time(),
)
engine.track_session(session)
# Record activity as requests come in
engine.record_activity("sess-001", request_bytes=1024)
# Check status
print(json.dumps(engine.get_status(), indent=2))
Zero Trust Checklist for Agent Deployments
Use this checklist when deploying a new agent or auditing an existing fleet.
| # | Check | Priority | Status |
|---|---|---|---|
| 1 | Every agent has a unique SPIFFE ID | P0 | Required |
| 2 | All agent-to-agent traffic uses mTLS | P0 | Required |
| 3 | Default-deny NetworkPolicy in every namespace | P0 | Required |
| 4 | PDP evaluates every request (no bypass paths) | P0 | Required |
| 5 | SVID lifetime is 1 hour or less | P0 | Required |
| 6 | Workload attestation verifies binary hash | P1 | Required |
| 7 | Continuous verification runs every 30-60 seconds | P1 | Required |
| 8 | Session duration capped at 1 hour | P1 | Required |
| 9 | Anomaly detection feeds into PDP decisions | P1 | Recommended |
| 10 | Microsegmentation rules reviewed quarterly | P2 | Recommended |
| 11 | All PDP decisions logged with request context | P0 | Required |
| 12 | Denied requests trigger alerts above threshold | P1 | Required |
| 13 | No agent runs with cluster-admin or root | P0 | Required |
| 14 | Cross-namespace traffic explicitly allowlisted | P0 | Required |
| 15 | Time-based access restrictions for sensitive tools | P2 | Recommended |
Chapter 7: NHI Compliance Frameworks
The Compliance Gap for Machine Identities
Every compliance framework written before 2024 assumes that identities are human. SOC 2 Type II talks about "personnel." ISO 27001 references "employees and contractors." NIST 800-53 defines access control in terms of "users." When your organization undergoes a SOC 2 audit and the auditor asks "How do you manage access for all identities in your environment?", they expect you to show them Active Directory groups and Okta assignments. They do not expect you to show them 312 API keys, 89 OAuth client credentials, and 47 service accounts that collectively have more access than every human in the organization combined.
This gap is closing. The 2025 updates to SOC 2 guidance explicitly mention machine identities. ISO 27001:2022 Annex A includes controls that apply to automated processes. NIST SP 800-207 (Zero Trust Architecture) makes no distinction between human and non-human subjects. The EU AI Act, effective August 2025, imposes specific requirements on AI systems that include the agents operating them. Organizations that treat NHI security as separate from compliance are going to fail their next audit.
SOC 2 Type II for Machine Identities
SOC 2 Type II evaluates the operating effectiveness of controls over a review period (typically 6-12 months). For NHI, the relevant Trust Services Criteria are:
| SOC 2 Criterion | Description | NHI Control |
|---|---|---|
| CC6.1 | Logical access security | Every machine identity authenticated via SPIFFE/mTLS or OAuth 2.0 CC |
| CC6.2 | Credentials management | Automated rotation, no static credentials older than 90 days |
| CC6.3 | Access authorization | Least-privilege policies enforced by PDP, reviewed quarterly |
| CC6.6 | External system boundaries | Agent-to-external-service connections use scoped credentials |
| CC6.8 | Prevent unauthorized access | Default-deny microsegmentation, continuous verification |
| CC7.1 | Detection of changes | Drift detection on agent permissions, alerting on new identities |
| CC7.2 | Monitoring for anomalies | Behavioral baseline per agent, anomaly scoring |
| CC8.1 | Change management | Agent deployments go through CI/CD with policy-as-code review |
The auditor will ask for evidence. The following class generates SOC 2 evidence packages from your agent governance data.
import json
import time
import hashlib
import logging
from dataclasses import dataclass, field
from typing import Optional
logger = logging.getLogger(__name__)
@dataclass
class ComplianceEvidence:
"""A piece of compliance evidence for audit."""
evidence_id: str
framework: str # "soc2", "iso27001", "nist800207"
control_id: str # e.g., "CC6.1", "A.9.2.3"
title: str
description: str
evidence_type: str # "automated", "manual", "observation"
collected_at: float = field(default_factory=time.time)
valid_from: str = ""
valid_to: str = ""
data: dict = field(default_factory=dict)
data_hash: str = ""
def __post_init__(self):
if not self.data_hash and self.data:
self.data_hash = hashlib.sha256(
json.dumps(self.data, sort_keys=True).encode()
).hexdigest()
@dataclass
class ControlStatus:
"""Status of a compliance control."""
control_id: str
framework: str
description: str
implemented: bool
effective: bool
evidence_count: int
gaps: list[str] = field(default_factory=list)
last_tested: Optional[float] = None
class NHIComplianceEngine:
"""Generate compliance evidence and control assessments for NHI.
Supports SOC 2 Type II, ISO 27001, and NIST 800-207 frameworks.
Evidence is generated from live system data, not manual attestations.
"""
def __init__(self, organization: str, review_period_start: str,
review_period_end: str):
self.organization = organization
self.review_period_start = review_period_start
self.review_period_end = review_period_end
self._evidence: list[ComplianceEvidence] = []
self._controls: dict[str, ControlStatus] = {}
def collect_credential_rotation_evidence(
self, rotation_records: list[dict],
) -> ComplianceEvidence:
"""Collect evidence that credentials are rotated per policy.
Maps to SOC 2 CC6.2 and ISO 27001 A.9.2.4.
"""
total = len(rotation_records)
rotated_on_time = sum(
1 for r in rotation_records
if r.get("rotated_within_policy", False)
)
compliance_rate = (rotated_on_time / total * 100) if total > 0 else 0
oldest_unrotated = None
for r in rotation_records:
if not r.get("rotated_within_policy", False):
age = r.get("credential_age_days", 0)
if oldest_unrotated is None or age > oldest_unrotated:
oldest_unrotated = age
evidence = ComplianceEvidence(
evidence_id=f"cred-rotation-{int(time.time())}",
framework="soc2",
control_id="CC6.2",
title="Credential Rotation Compliance",
description=(
f"Automated credential rotation evidence for "
f"{self.review_period_start} to {self.review_period_end}"
),
evidence_type="automated",
valid_from=self.review_period_start,
valid_to=self.review_period_end,
data={
"total_credentials": total,
"rotated_on_time": rotated_on_time,
"compliance_rate_percent": round(compliance_rate, 2),
"oldest_unrotated_days": oldest_unrotated,
"policy_max_age_days": 90,
"rotation_records_sample": rotation_records[:10],
},
)
self._evidence.append(evidence)
return evidence
def collect_access_review_evidence(
self, access_reviews: list[dict],
) -> ComplianceEvidence:
"""Collect evidence of periodic access reviews for machine identities.
Maps to SOC 2 CC6.3 and ISO 27001 A.9.2.5.
"""
total_identities = len(access_reviews)
reviewed = sum(
1 for r in access_reviews if r.get("reviewed", False)
)
permissions_reduced = sum(
1 for r in access_reviews if r.get("permissions_reduced", False)
)
identities_revoked = sum(
1 for r in access_reviews if r.get("revoked", False)
)
evidence = ComplianceEvidence(
evidence_id=f"access-review-{int(time.time())}",
framework="soc2",
control_id="CC6.3",
title="Machine Identity Access Review",
description=(
f"Quarterly access review for {total_identities} "
f"machine identities"
),
evidence_type="automated",
valid_from=self.review_period_start,
valid_to=self.review_period_end,
data={
"total_identities": total_identities,
"identities_reviewed": reviewed,
"review_completion_rate": (
round(reviewed / total_identities * 100, 2)
if total_identities > 0 else 0
),
"permissions_reduced": permissions_reduced,
"identities_revoked": identities_revoked,
"review_methodology": "automated_least_privilege_analysis",
},
)
self._evidence.append(evidence)
return evidence
def collect_pdp_enforcement_evidence(
self, pdp_decisions: list[dict],
) -> ComplianceEvidence:
"""Collect evidence that policy enforcement is active and effective.
Maps to SOC 2 CC6.1, CC6.8 and NIST 800-207 DP-1.
"""
total = len(pdp_decisions)
allowed = sum(1 for d in pdp_decisions if d.get("allowed", False))
denied = total - allowed
denial_categories: dict[str, int] = {}
for d in pdp_decisions:
if not d.get("allowed", False):
cat = d.get("policy_id", "unknown")
denial_categories[cat] = denial_categories.get(cat, 0) + 1
evidence = ComplianceEvidence(
evidence_id=f"pdp-enforcement-{int(time.time())}",
framework="soc2",
control_id="CC6.1",
title="Policy Decision Point Enforcement Evidence",
description=(
f"PDP evaluated {total} agent requests during review period"
),
evidence_type="automated",
valid_from=self.review_period_start,
valid_to=self.review_period_end,
data={
"total_decisions": total,
"allowed": allowed,
"denied": denied,
"denial_rate_percent": (
round(denied / total * 100, 2) if total > 0 else 0
),
"denial_categories": denial_categories,
"enforcement_gap_seconds": 0, # No bypass periods
"pdp_availability_percent": 99.99,
},
)
self._evidence.append(evidence)
return evidence
def assess_controls(self) -> list[ControlStatus]:
"""Assess all NHI-relevant controls across frameworks."""
controls = [
# SOC 2 controls
ControlStatus(
control_id="CC6.1",
framework="soc2",
description="Logical access security for machine identities",
implemented=True,
effective=self._has_evidence("CC6.1"),
evidence_count=self._count_evidence("CC6.1"),
),
ControlStatus(
control_id="CC6.2",
framework="soc2",
description="Credential lifecycle management",
implemented=True,
effective=self._has_evidence("CC6.2"),
evidence_count=self._count_evidence("CC6.2"),
),
ControlStatus(
control_id="CC6.3",
framework="soc2",
description="Access authorization and least privilege",
implemented=True,
effective=self._has_evidence("CC6.3"),
evidence_count=self._count_evidence("CC6.3"),
),
# ISO 27001 controls
ControlStatus(
control_id="A.9.2.1",
framework="iso27001",
description="User registration and de-registration (NHI)",
implemented=True,
effective=self._has_evidence("CC6.1"),
evidence_count=self._count_evidence("CC6.1"),
),
ControlStatus(
control_id="A.9.2.3",
framework="iso27001",
description="Management of privileged access rights",
implemented=True,
effective=self._has_evidence("CC6.3"),
evidence_count=self._count_evidence("CC6.3"),
),
ControlStatus(
control_id="A.9.4.1",
framework="iso27001",
description="Information access restriction",
implemented=True,
effective=self._has_evidence("CC6.1"),
evidence_count=self._count_evidence("CC6.1"),
),
# NIST 800-207 controls
ControlStatus(
control_id="ZTA-1",
framework="nist800207",
description="All data sources and computing services are resources",
implemented=True,
effective=True,
evidence_count=0,
),
ControlStatus(
control_id="ZTA-2",
framework="nist800207",
description="All communication is secured regardless of location",
implemented=True,
effective=self._has_evidence("CC6.1"),
evidence_count=self._count_evidence("CC6.1"),
),
]
for control in controls:
self._controls[control.control_id] = control
return controls
def _has_evidence(self, control_id: str) -> bool:
return any(e.control_id == control_id for e in self._evidence)
def _count_evidence(self, control_id: str) -> int:
return sum(1 for e in self._evidence if e.control_id == control_id)
def generate_audit_package(self) -> dict:
"""Generate a complete audit evidence package."""
controls = self.assess_controls()
effective_controls = sum(1 for c in controls if c.effective)
return {
"organization": self.organization,
"review_period": {
"start": self.review_period_start,
"end": self.review_period_end,
},
"generated_at": time.strftime(
"%Y-%m-%dT%H:%M:%SZ", time.gmtime()
),
"summary": {
"total_controls": len(controls),
"effective_controls": effective_controls,
"effectiveness_rate": round(
effective_controls / len(controls) * 100, 2
) if controls else 0,
"total_evidence_items": len(self._evidence),
"frameworks_covered": list({
c.framework for c in controls
}),
},
"controls": [
{
"control_id": c.control_id,
"framework": c.framework,
"description": c.description,
"implemented": c.implemented,
"effective": c.effective,
"evidence_count": c.evidence_count,
"gaps": c.gaps,
}
for c in controls
],
"evidence": [
{
"evidence_id": e.evidence_id,
"framework": e.framework,
"control_id": e.control_id,
"title": e.title,
"evidence_type": e.evidence_type,
"valid_from": e.valid_from,
"valid_to": e.valid_to,
"data_hash": e.data_hash,
"data": e.data,
}
for e in self._evidence
],
}
# Usage
engine = NHIComplianceEngine(
organization="acme-corp",
review_period_start="2025-10-01",
review_period_end="2026-03-31",
)
# Collect credential rotation evidence
rotation_records = [
{
"credential_id": f"cred-{i:03d}",
"agent_id": f"agent-{i:03d}",
"credential_type": "api_key",
"credential_age_days": 45 + (i % 60),
"rotated_within_policy": (45 + (i % 60)) <= 90,
"last_rotated": "2026-02-15T10:00:00Z",
}
for i in range(150)
]
engine.collect_credential_rotation_evidence(rotation_records)
# Collect access review evidence
access_reviews = [
{
"agent_id": f"agent-{i:03d}",
"reviewed": True,
"permissions_reduced": i % 5 == 0, # 20% had permissions reduced
"revoked": i % 25 == 0, # 4% were revoked
}
for i in range(150)
]
engine.collect_access_review_evidence(access_reviews)
# Collect PDP enforcement evidence
pdp_decisions = [
{
"request_id": f"req-{i:06d}",
"agent_id": f"agent-{i % 50:03d}",
"tool": "create_payment" if i % 3 == 0 else "read_price_feed",
"allowed": i % 20 != 0, # 5% denied
"policy_id": "zt-authorization" if i % 20 == 0 else "zt-allow",
}
for i in range(50_000)
]
engine.collect_pdp_enforcement_evidence(pdp_decisions)
# Generate audit package
package = engine.generate_audit_package()
print(json.dumps(package["summary"], indent=2))
ISO 27001 Annex A Controls for NHI
ISO 27001:2022 Annex A reorganized controls into four themes: Organizational, People, Physical, and Technological. Machine identities cut across the Organizational and Technological themes. The following table maps every relevant Annex A control to its NHI implementation.
| Annex A Control | Title | NHI Implementation |
|---|---|---|
| A.5.15 | Access control | PDP enforces least-privilege policies per agent |
| A.5.16 | Identity management | SPIFFE IDs for all agents, lifecycle tracked in registry |
| A.5.17 | Authentication information | Credentials stored in Vault, rotated every 90 days max |
| A.5.18 | Access rights | Scoped to specific tools and namespaces per agent |
| A.5.23 | Information security for cloud services | Workload identity federation (no static cloud keys) |
| A.5.33 | Protection of records | Immutable audit trail for all PDP decisions |
| A.8.2 | Privileged access rights | No agent runs with admin privileges, elevated access time-boxed |
| A.8.3 | Information access restriction | Microsegmentation enforces tool-level isolation |
| A.8.5 | Secure authentication | mTLS with SPIFFE SVIDs, certificate lifetime 1 hour |
| A.8.9 | Configuration management | Agent policies defined as code, reviewed in CI/CD |
| A.8.15 | Logging | All agent actions, PDP decisions, and credential operations logged |
| A.8.16 | Monitoring activities | Continuous verification engine, anomaly detection |
NIST 800-207 Zero Trust Mapping
NIST SP 800-207 defines seven tenets of zero trust. Here is how each tenet maps to the NHI architecture described in this guide.
| NIST 800-207 Tenet | Description | NHI Implementation |
|---|---|---|
| 1 | All data sources and computing services are considered resources | Every agent, tool, and data store is a resource with its own identity |
| 2 | All communication is secured regardless of network location | mTLS on all agent-to-agent connections, even within the same cluster |
| 3 | Access to individual resources is granted on a per-session basis | PDP evaluates every request independently, no cached authorizations |
| 4 | Access is determined by dynamic policy | Policy-as-code evaluated at runtime with real-time context (time, anomaly score, request rate) |
| 5 | Enterprise monitors and measures integrity and security posture | Continuous verification engine re-verifies identity and attestation every 30s |
| 6 | Authentication and authorization are dynamic and strictly enforced | SPIFFE SVIDs auto-rotate hourly, authorization checked per-request |
| 7 | Enterprise collects information and uses it to improve security | PDP decision logs, anomaly scores, and behavioral baselines feed policy updates |
EU AI Act Implications for NHI
The EU AI Act (Regulation 2024/1689), which became fully applicable on August 2, 2025, imposes requirements on AI systems that have direct implications for how you manage the non-human identities of your AI agents. The Act does not use the term "non-human identity," but its requirements on traceability, transparency, and human oversight translate directly to NHI controls.
High-risk AI systems (Article 9-15): If your agents operate in domains classified as high-risk -- financial services, critical infrastructure, employment -- the Act requires:
-
Traceability (Article 12): Automatic logging of system events throughout the AI system's lifecycle. Every agent action, every tool invocation, every credential operation must be logged with sufficient detail to reconstruct the chain of events. The PDP audit trail from Chapter 6 satisfies this requirement.
-
Transparency (Article 13): The system must be sufficiently transparent that deployers can interpret its output and use it appropriately. For NHI, this means that every agent's permissions, trust level, and policy constraints must be documented and queryable. The compliance engine above generates this documentation automatically.
-
Human oversight (Article 14): High-risk AI systems must be designed to allow effective human oversight. For NHI, this means that agents must not be able to escalate their own permissions, create new identities without approval, or bypass policy controls. The PDP's default-deny posture and the governance toolkit's approval workflows satisfy this requirement.
-
Accuracy, robustness, and cybersecurity (Article 15): AI systems must achieve appropriate levels of cybersecurity throughout their lifecycle. This is a direct mandate for NHI security -- credential rotation, zero trust, continuous verification, and incident response are all required, not optional.
General-purpose AI (Article 53): If your agents use general-purpose AI models (which most do), the model providers must provide technical documentation that includes information about the model's security properties. As the deployer, you must ensure that the agent's NHI controls are compatible with the model's security posture.
The penalty for non-compliance is severe: up to 35 million EUR or 7% of worldwide annual turnover, whichever is higher.
Automated Compliance Monitoring
Manual compliance monitoring fails at agent scale. You cannot have a human review 50,000 PDP decisions per day. The following class implements continuous compliance monitoring that detects control failures in real time and generates alerts before they become audit findings.
import time
import json
import logging
from dataclasses import dataclass, field
from typing import Optional, Callable
from enum import Enum
logger = logging.getLogger(__name__)
class ComplianceStatus(Enum):
COMPLIANT = "compliant"
WARNING = "warning" # Approaching threshold
NON_COMPLIANT = "non_compliant"
UNKNOWN = "unknown"
@dataclass
class ComplianceMetric:
"""A measurable compliance metric."""
metric_id: str
framework: str
control_id: str
description: str
current_value: float
threshold_warning: float
threshold_failure: float
unit: str # "percent", "days", "count", "seconds"
higher_is_better: bool # True if higher value = more compliant
measured_at: float = field(default_factory=time.time)
@property
def status(self) -> ComplianceStatus:
if self.higher_is_better:
if self.current_value >= self.threshold_failure:
return ComplianceStatus.COMPLIANT
elif self.current_value >= self.threshold_warning:
return ComplianceStatus.WARNING
else:
return ComplianceStatus.NON_COMPLIANT
else:
if self.current_value <= self.threshold_failure:
return ComplianceStatus.COMPLIANT
elif self.current_value <= self.threshold_warning:
return ComplianceStatus.WARNING
else:
return ComplianceStatus.NON_COMPLIANT
class ContinuousComplianceMonitor:
"""Monitor NHI compliance metrics in real time.
Evaluates metrics against thresholds and generates alerts
when controls drift toward non-compliance.
"""
def __init__(self, organization: str,
alert_callback: Optional[Callable] = None):
self.organization = organization
self.alert_callback = alert_callback
self._metrics: dict[str, ComplianceMetric] = {}
self._alert_history: list[dict] = []
self._check_count: int = 0
def register_metric(self, metric: ComplianceMetric) -> None:
"""Register a compliance metric for monitoring."""
self._metrics[metric.metric_id] = metric
def update_metric(self, metric_id: str, value: float) -> ComplianceStatus:
"""Update a metric value and check compliance."""
metric = self._metrics.get(metric_id)
if not metric:
raise ValueError(f"Unknown metric: {metric_id}")
old_status = metric.status
metric.current_value = value
metric.measured_at = time.time()
new_status = metric.status
# Alert on status changes
if new_status != old_status:
self._handle_status_change(metric, old_status, new_status)
return new_status
def _handle_status_change(self, metric: ComplianceMetric,
old_status: ComplianceStatus,
new_status: ComplianceStatus) -> None:
"""Handle a compliance status change."""
alert = {
"timestamp": time.strftime(
"%Y-%m-%dT%H:%M:%SZ", time.gmtime()
),
"organization": self.organization,
"metric_id": metric.metric_id,
"framework": metric.framework,
"control_id": metric.control_id,
"description": metric.description,
"old_status": old_status.value,
"new_status": new_status.value,
"current_value": metric.current_value,
"threshold_warning": metric.threshold_warning,
"threshold_failure": metric.threshold_failure,
"unit": metric.unit,
}
self._alert_history.append(alert)
severity = "INFO"
if new_status == ComplianceStatus.NON_COMPLIANT:
severity = "CRITICAL"
elif new_status == ComplianceStatus.WARNING:
severity = "WARNING"
logger.log(
getattr(logging, severity, logging.WARNING),
"Compliance status change: %s [%s] %s -> %s (value=%s%s)",
metric.metric_id,
metric.control_id,
old_status.value,
new_status.value,
metric.current_value,
metric.unit,
)
if self.alert_callback:
self.alert_callback(alert)
def run_compliance_check(self) -> dict:
"""Run a full compliance check across all metrics."""
self._check_count += 1
results = {
"check_number": self._check_count,
"timestamp": time.strftime(
"%Y-%m-%dT%H:%M:%SZ", time.gmtime()
),
"organization": self.organization,
"metrics": {},
"summary": {
"total": 0,
"compliant": 0,
"warning": 0,
"non_compliant": 0,
},
}
for metric_id, metric in self._metrics.items():
status = metric.status
results["metrics"][metric_id] = {
"status": status.value,
"value": metric.current_value,
"threshold_warning": metric.threshold_warning,
"threshold_failure": metric.threshold_failure,
"unit": metric.unit,
"framework": metric.framework,
"control_id": metric.control_id,
}
results["summary"]["total"] += 1
results["summary"][status.value] = (
results["summary"].get(status.value, 0) + 1
)
return results
# Usage
def send_alert(alert: dict) -> None:
"""Send compliance alert to PagerDuty / Slack / email."""
print(f"ALERT [{alert['new_status']}]: {alert['description']} "
f"({alert['current_value']}{alert['unit']})")
monitor = ContinuousComplianceMonitor(
organization="acme-corp",
alert_callback=send_alert,
)
# Register compliance metrics
monitor.register_metric(ComplianceMetric(
metric_id="credential_rotation_rate",
framework="soc2",
control_id="CC6.2",
description="Percentage of credentials rotated within policy",
current_value=98.5,
threshold_warning=95.0,
threshold_failure=90.0,
unit="percent",
higher_is_better=True,
))
monitor.register_metric(ComplianceMetric(
metric_id="oldest_credential_age",
framework="iso27001",
control_id="A.5.17",
description="Age of oldest unrotated credential",
current_value=45,
threshold_warning=75,
threshold_failure=90,
unit="days",
higher_is_better=False,
))
monitor.register_metric(ComplianceMetric(
metric_id="pdp_enforcement_rate",
framework="nist800207",
control_id="ZTA-3",
description="Percentage of requests evaluated by PDP",
current_value=100.0,
threshold_warning=99.9,
threshold_failure=99.0,
unit="percent",
higher_is_better=True,
))
monitor.register_metric(ComplianceMetric(
metric_id="access_review_completion",
framework="soc2",
control_id="CC6.3",
description="Percentage of identities reviewed this quarter",
current_value=100.0,
threshold_warning=95.0,
threshold_failure=90.0,
unit="percent",
higher_is_better=True,
))
monitor.register_metric(ComplianceMetric(
metric_id="svid_max_lifetime",
framework="nist800207",
control_id="ZTA-6",
description="Maximum SVID lifetime in seconds",
current_value=3600,
threshold_warning=7200,
threshold_failure=86400,
unit="seconds",
higher_is_better=False,
))
# Simulate a compliance drift
monitor.update_metric("credential_rotation_rate", 94.2) # Triggers warning
monitor.update_metric("oldest_credential_age", 92) # Triggers failure
# Run full compliance check
report = monitor.run_compliance_check()
print(json.dumps(report, indent=2))
Compliance Automation Checklist
| # | Task | Frequency | Framework |
|---|---|---|---|
| 1 | Automated credential rotation evidence collection | Daily | SOC 2, ISO 27001 |
| 2 | PDP decision log integrity verification | Daily | SOC 2, NIST 800-207 |
| 3 | Machine identity access review | Quarterly | SOC 2, ISO 27001 |
| 4 | Microsegmentation rule review | Quarterly | NIST 800-207 |
| 5 | Orphaned credential scan | Weekly | SOC 2 |
| 6 | Secret sprawl detection | Weekly | ISO 27001 |
| 7 | Compliance metric threshold review | Monthly | All |
| 8 | Agent permission drift detection | Daily | SOC 2, ISO 27001 |
| 9 | Continuous verification uptime report | Monthly | NIST 800-207 |
| 10 | EU AI Act traceability log completeness check | Weekly | EU AI Act |
| 11 | Audit evidence package generation | Quarterly | SOC 2, ISO 27001 |
| 12 | Cross-framework gap analysis | Semi-annually | All |
Next Steps
For deployment patterns, monitoring, and production hardening, see the Agent Production Hardening Guide.
Related skills
- Downloads
- Security audit
- Review
- Last updated
- 1mo ago
- Current version
- v1.3.1
- License
- MIT-0