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.

1. The NHI Crisis -- Why Machine Identities Are the #1 Attack Vector
2. NHI Lifecycle Management

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#

1. The NHI Crisis -- Why Machine Identities Are the #1 Attack Vector
2. NHI Lifecycle Management
3. Workload Identity Federation for AI Agents
4. Machine-to-Machine Authentication Patterns
5. Microsoft Agent Governance Toolkit Integration
6. Zero Trust for AI Agents
7. 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 VectorPercentageExampleHardcoded credentials in code/config34%API key committed to git, never rotatedOver-privileged service accounts22%Read-only agent with write permissionsExpired but still-valid certificates16%TLS cert expired, verification disabled as workaroundSecret sprawl across environments14%Same key used in dev, staging, and productionLack of rotation9%OAuth client secret unchanged for 12+ monthsStolen from CI/CD pipelines5%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.

DimensionHuman IAMNHI ManagementAuthenticationPassword + MFAKeys, certificates, tokens, secretsLifecycle triggerHR events (hire, transfer, terminate)Deployment events (deploy, scale, decommission)RotationPassword expiry policies (90 days typical)Must be automated, zero-downtime, continuousQuantityHundreds to thousands per orgTens of thousands to millions per orgVisibilityHR directory, SSO dashboardScattered across vaults, env vars, config files, CI/CDBehavioral baselineLogin times, locations, device fingerprintsAPI call patterns, volume, tool access sequencesDeprovisioningDisable account on terminationOften forgotten -- orphaned credentials persistPrivilege reviewQuarterly access reviews with human managersNo 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.

python

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 .env file 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.

python

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.

python

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:

1. Issue the new credential while the old credential remains valid.
2. Update the agent to use the new credential.
3. Verify the agent is successfully using the new credential.
4. Only then revoke the old credential.

python

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.

python

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.

python

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.

python

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.

python

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).

python

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.

python

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.

python

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.

python

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.

python

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.

python

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 +