The Trust Crisis Between AI Agents: Adding Zero-Trust Authentication
A FastAPI Middleware + Docker Compose Production-Ready Implementation
Why AI Agents Have a Trust Problem
When you deploy a single AI agent, trust is simple: you control it. The moment you introduce multiple agents — a planner agent delegating to a coder agent, a monitor agent auditing a deployment agent — trust becomes a crisis.
Consider this scenario in a typical multi-agent pipeline:
User → Planner Agent → Coder Agent → Review Agent → Deploy Agent
↑ │
└───────────────────────────────────────────────────────────┘
(unauthenticated callback)
The Coder Agent blindly trusts the callback from the Planner. The Deploy Agent doesn't verify who asked it to deploy. There is no identity, no permission check, no audit trail. This is the equivalent of a microservices architecture with no service mesh — it works until someone exploits it.
This is the AI Agent Trust Crisis.
The solution is exactly what modern cloud infrastructure learned the hard way: Zero-Trust Architecture. In the Zero-Trust model, no agent is trusted by default — regardless of where the request originates. Every request must be authenticated, authorized, and logged.
This article builds that system from scratch:
- A JWT-based identity layer for agents
- A FastAPI middleware that enforces authentication on every call
- Role-based access control (RBAC) for agent permissions
- A complete Docker Compose deployment
- A working demo with three agents talking to each other
1. The Threat Model — What Are We Protecting Against?
Before writing a single line of code, we need to be clear about the threat model.
Attacks specific to multi-agent systems:
Impersonation attacks: A compromised or malicious agent pretends to be another agent with higher privileges. ("I'm the Planner, deploy this.")
Token replay: An attacker captures a valid JWT and replays it to authenticate as a legitimate agent.
Privilege escalation: An agent with read-only permissions tricks the system into granting it write access.
Lateral movement: After compromising one low-privilege agent, an attacker uses it as a pivot to reach higher-privilege agents.
Man-in-the-middle: Traffic between agents is intercepted and modified without detection.
Zero-Trust principles that address each:
- Never trust, always verify: Every request, even from a "trusted" internal agent, requires valid JWT verification
- Least privilege: Agents get only the permissions they need for their specific role
- Assume breach: Logs and audit trails exist so you can detect and respond to compromise
- Continuous verification: Tokens expire; short-lived tokens limit the blast radius of compromise
2. Architecture Overview
┌─────────────────────────────────────────────────────────────────┐
│ Agent Mesh │
│ │
│ ┌──────────┐ JWT ┌──────────┐ JWT ┌──────────┐ │
│ │ Planner │──────────► │ Auth │──────────► │ Coder │ │
│ │ Agent │ token │ Gateway │ token │ Agent │ │
│ └──────────┘ request │ (FastAPI)│ validated │ │ │
│ │ Middleware│ │ │ │
│ └──────────┘ └──────────┘ │
│ │ │
│ ┌─────▼─────┐ │
│ │ Token │ │
│ │ Issuer │ │
│ │ (OAuth2) │ │
│ └───────────┘ │
└─────────────────────────────────────────────────────────────────┘
The Auth Gateway is a FastAPI service that:
- Issues JWT tokens to registered agents
- Validates JWTs on every inter-agent request
- Enforces role-based permissions
- Logs all authentication events for audit
Agents don't talk to each other directly — they go through the gateway, which verifies and logs every request.
3. Project Structure
agent-auth-system/
├── auth_gateway/
│ ├── __init__.py
│ ├── main.py # FastAPI app
│ ├── middleware.py # JWT validation middleware
│ ├── models.py # Pydantic models
│ ├── auth.py # Token issuance (OAuth2 password flow)
│ ├── rbac.py # Role-based permission checks
│ └── config.py # Configuration
├── agents/
│ ├── planner_agent.py
│ ├── coder_agent.py
│ └── reviewer_agent.py
├── docker-compose.yml
├── Dockerfile.gateway
├── requirements.txt
└── README.md
4. Core Implementation
4.1 Configuration
# auth_gateway/config.py
from pydantic_settings import BaseSettings
from functools import lru_cache
class Settings(BaseSettings):
# JWT configuration
JWT_SECRET_KEY: str = "your-256-bit-secret-change-in-production"
JWT_ALGORITHM: str = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
# Agent registry — in production, use a database
AGENT_REGISTRY: dict = {
"planner": {"role": "planner", "permissions": ["delegate", "read"]},
"coder": {"role": "coder", "permissions": ["read", "write", "execute"]},
"reviewer":{"role": "reviewer","permissions": ["read", "approve", "reject"]},
"deploy": {"role": "deploy", "permissions": ["read", "deploy", "approve"]},
}
class Config:
env_file = ".env"
@lru_cache
def get_settings():
return Settings()
4.2 Pydantic Models
# auth_gateway/models.py
from pydantic import BaseModel, Field
from typing import Optional, List
class AgentRegister(BaseModel):
agent_id: str = Field(..., description="Unique agent identifier")
secret: str = Field(..., description="Shared secret for agent authentication")
class TokenRequest(BaseModel):
agent_id: str
secret: str
class TokenResponse(BaseModel):
access_token: str
token_type: str = "bearer"
expires_in: int
class AgentIdentity(BaseModel):
agent_id: str
role: str
permissions: List[str]
exp: int
class AuthenticatedRequest(BaseModel):
"""Request with embedded agent identity (verified by middleware)"""
action: str = Field(..., description="The action being requested")
resource: Optional[str] = None
payload: Optional[dict] = None
4.3 Token Issuance (OAuth2 Password Flow)
# auth_gateway/auth.py
from datetime import datetime, timedelta, timezone
from jose import jwt, JWTError
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from .config import get_settings
from .models import TokenResponse, AgentIdentity
settings = get_settings()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token", auto_error=False)
def create_access_token(agent_id: str, role: str, permissions: List[str]) -> str:
"""Create a signed JWT for an authenticated agent."""
expire = datetime.now(timezone.utc) + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
payload = {
"sub": agent_id,
"role": role,
"permissions": permissions,
"exp": expire,
"iat": datetime.now(timezone.utc),
"type": "agent_access",
}
return jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM)
def verify_token(token: str) -> AgentIdentity:
"""Verify a JWT and extract agent identity."""
try:
payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM])
agent_id = payload.get("sub")
role = payload.get("role")
permissions = payload.get("permissions", [])
exp = payload.get("exp")
if agent_id is None or role is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token: missing required claims",
)
return AgentIdentity(
agent_id=agent_id,
role=role,
permissions=permissions,
exp=exp,
)
except JWTError as e:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=f"Token validation failed: {str(e)}",
)
async def get_current_agent(token: str = Depends(oauth2_scheme)) -> AgentIdentity:
"""Dependency to get the current authenticated agent."""
if token is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Not authenticated",
headers={"WWW-Authenticate": "Bearer"},
)
return verify_token(token)
4.4 Role-Based Access Control
# auth_gateway/rbac.py
from fastapi import HTTPException, status
from .models import AgentIdentity
# Define which roles can perform which actions
ROLE_ACTION_MAP = {
"planner": ["delegate", "read"],
"coder": ["read", "write", "execute", "call_coder"],
"reviewer": ["read", "approve", "reject", "call_reviewer"],
"deploy": ["read", "deploy", "approve", "call_deploy"],
"admin": ["*"], # Wildcard for superadmin
}
def check_permission(identity: AgentIdentity, required_action: str) -> bool:
"""Check if the agent has permission for the requested action."""
role = identity.role
# Get allowed actions for this role
allowed = ROLE_ACTION_MAP.get(role, [])
# Wildcard check
if "*" in allowed:
return True
return required_action in allowed
def require_permission(required_action: str):
"""Decorator-style dependency for permission checking."""
def checker(identity: AgentIdentity):
if not check_permission(identity, required_action):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"Agent '{identity.agent_id}' with role '{identity.role}' "
f"does not have permission to perform '{required_action}'",
)
return identity
return checker
4.5 JWT Validation Middleware
# auth_gateway/middleware.py
import time
import logging
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse
from jose import jwt, JWTError
from .config import get_settings
logger = logging.getLogger(__name__)
settings = get_settings()
# Paths that don't require authentication
PUBLIC_PATHS = {"/", "/docs", "/openapi.json", "/token", "/health"}
class AgentAuthMiddleware(BaseHTTPMiddleware):
"""
Zero-Trust JWT validation middleware.
Every incoming request to the gateway must present a valid JWT.
No agent is trusted by default — even internal service-to-service calls.
"""
async def dispatch(self, request: Request, call_next):
# Skip authentication for public paths
if request.url.path in PUBLIC_PATHS:
return await call_next(request)
# Extract JWT from Authorization header
auth_header = request.headers.get("Authorization")
if not auth_header or not auth_header.startswith("Bearer "):
return JSONResponse(
status_code=401,
content={
"error": "Unauthorized",
"detail": "Missing or malformed Authorization header. "
"Expected: Bearer <jwt_token>",
}
)
token = auth_header.replace("Bearer ", "")
try:
# Decode and verify JWT
payload = jwt.decode(
token,
settings.JWT_SECRET_KEY,
algorithms=[settings.JWT_ALGORITHM],
)
# Validate token type
if payload.get("type") != "agent_access":
return JSONResponse(
status_code=401,
content={"error": "Invalid token type", "detail": "Expected agent_access token"}
)
# Attach identity to request state for downstream use
request.state.agent_identity = {
"agent_id": payload.get("sub"),
"role": payload.get("role"),
"permissions": payload.get("permissions", []),
"exp": payload.get("exp"),
}
# Log the authenticated request
logger.info(
f"Authenticated request: agent={payload.get('sub')} "
f"role={payload.get('role')} path={request.url.path}"
)
response = await call_next(request)
return response
except jwt.ExpiredSignatureError:
logger.warning(f"Expired token attempted: {request.url.path}")
return JSONResponse(
status_code=401,
content={"error": "Token expired", "detail": "Request a new token from /token"}
)
except JWTError as e:
logger.warning(f"Invalid token: {str(e)}")
return JSONResponse(
status_code=401,
content={"error": "Invalid token", "detail": str(e)}
)
def log_request(request: Request, call_next):
"""Additional logging middleware for audit trail."""
start_time = time.time()
response = call_next(request)
duration = time.time() - start_time
agent_id = getattr(request.state, "agent_identity", {}).get("agent_id", "anonymous")
logger.info(
f"{request.method} {request.url.path} | "
f"agent={agent_id} | "
f"status={response.status_code} | "
f"duration={duration:.3f}s"
)
return response
4.6 FastAPI Application
# auth_gateway/main.py
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.middleware.cors import CORSMiddleware
from .middleware import AgentAuthMiddleware
from .auth import create_access_token, get_current_agent
from .rbac import require_permission
from .models import TokenRequest, TokenResponse, AgentIdentity, AuthenticatedRequest
from .config import get_settings
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI(
title="Agent Auth Gateway",
description="Zero-Trust JWT authentication for AI Agent Mesh",
version="1.0.0",
)
settings = get_settings()
# CORS for agent mesh
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Apply Zero-Trust middleware to all routes
app.add_middleware(AgentAuthMiddleware)
# ─── Public Endpoints ───────────────────────────────────────────────
@app.get("/health")
def health_check():
return {"status": "healthy", "service": "agent-auth-gateway"}
@app.post("/token", response_model=TokenResponse)
def issue_token(request: TokenRequest):
"""
OAuth2 Password Flow — issues JWT to registered agents.
In production, replace this with a proper OAuth2 authorization server.
"""
# Look up agent in registry
agent_config = settings.AGENT_REGISTRY.get(request.agent_id)
if not agent_config:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Unknown agent",
)
# In production: use bcrypt hash comparison on secrets
# This is a simplified version for demonstration
expected_secret = f"secret_{request.agent_id}" # Placeholder — use proper secrets
if request.secret != expected_secret:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid credentials",
)
token = create_access_token(
agent_id=request.agent_id,
role=agent_config["role"],
permissions=agent_config["permissions"],
)
logger.info(f"Token issued for agent: {request.agent_id}")
return TokenResponse(
access_token=token,
token_type="bearer",
expires_in=settings.ACCESS_TOKEN_EXPIRE_MINUTES * 60,
)
# ─── Protected Agent Endpoints ──────────────────────────────────────
@app.post("/api/agent/delegate")
def delegate_task(
request: AuthenticatedRequest,
identity: AgentIdentity = Depends(require_permission("delegate")),
):
"""Planner agent delegates a task to another agent."""
logger.info(f"Delegation from {identity.agent_id}: {request.action} on {request.resource}")
return {
"status": "delegated",
"from": identity.agent_id,
"action": request.action,
"resource": request.resource,
"payload": request.payload,
}
@app.post("/api/agent/code")
def code_task(
request: AuthenticatedRequest,
identity: AgentIdentity = Depends(require_permission("call_coder")),
):
"""Coder agent receives a coding task."""
logger.info(f"Coding task for {identity.agent_id}: {request.action}")
return {
"status": "received",
"agent": identity.agent_id,
"action": request.action,
"message": "Code task received and queued",
}
@app.post("/api/agent/review")
def review_task(
request: AuthenticatedRequest,
identity: AgentIdentity = Depends(require_permission("call_reviewer")),
):
"""Reviewer agent reviews code or outputs."""
return {
"status": "under_review",
"agent": identity.agent_id,
"action": request.action,
}
@app.post("/api/agent/deploy")
def deploy_task(
request: AuthenticatedRequest,
identity: AgentIdentity = Depends(require_permission("call_deploy")),
):
"""Deploy agent handles deployment — highest privilege."""
logger.warning(f"DEPLOY action by {identity.agent_id} on {request.resource}")
return {
"status": "deployed",
"agent": identity.agent_id,
"resource": request.resource,
"message": "Deployment completed",
}
@app.get("/api/agent/me")
def get_my_identity(identity: AgentIdentity = Depends(get_current_agent)):
"""Agents can query their own identity and permissions."""
return {
"agent_id": identity.agent_id,
"role": identity.role,
"permissions": identity.permissions,
}
@app.get("/api/agent/verify/{target_agent}")
def verify_agent(
target_agent: str,
identity: AgentIdentity = Depends(require_permission("read")),
):
"""Query another agent's role — for trust verification."""
target_config = settings.AGENT_REGISTRY.get(target_agent)
if not target_config:
raise HTTPException(status_code=404, detail="Agent not found in registry")
return {"agent_id": target_agent, **target_config}
5. Docker Compose Deployment
# docker-compose.yml
version: '3.8'
services:
# ─── Auth Gateway ────────────────────────────────────────────────
auth-gateway:
build:
context: .
dockerfile: Dockerfile.gateway
container_name: auth-gateway
ports:
- "8000:8000"
environment:
- JWT_SECRET_KEY=${JWT_SECRET_KEY:?JWT_SECRET_KEY required}
- JWT_ALGORITHM=HS256
- ACCESS_TOKEN_EXPIRE_MINUTES=30
volumes:
- ./logs:/app/logs
restart: unless-stopped
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
interval: 30s
timeout: 10s
retries: 3
networks:
- agent-mesh
# ─── Demo Agents ──────────────────────────────────────────────────
planner-agent:
build:
context: .
dockerfile: Dockerfile.agent
container_name: planner-agent
command: python agents/planner_agent.py
environment:
- AUTH_GATEWAY_URL=http://auth-gateway:8000
- AGENT_ID=planner
- AGENT_SECRET=secret_planner
depends_on:
auth-gateway:
condition: service_healthy
networks:
- agent-mesh
coder-agent:
build:
context: .
dockerfile: Dockerfile.agent
container_name: coder-agent
command: python agents/coder_agent.py
environment:
- AUTH_GATEWAY_URL=http://auth-gateway:8000
- AGENT_ID=coder
- AGENT_SECRET=secret_coder
depends_on:
auth-gateway:
condition: service_healthy
networks:
- agent-mesh
reviewer-agent:
build:
context: .
dockerfile: Dockerfile.agent
container_name: reviewer-agent
command: python agents/reviewer_agent.py
environment:
- AUTH_GATEWAY_URL=http://auth-gateway:8000
- AGENT_ID=reviewer
- AGENT_SECRET=secret_reviewer
depends_on:
auth-gateway:
condition: service_healthy
networks:
- agent-mesh
networks:
agent-mesh:
driver: bridge
# Dockerfile.gateway
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY auth_gateway/ ./auth_gateway/
EXPOSE 8000
CMD ["uvicorn", "auth_gateway.main:app", "--host", "0.0.0.0", "--port", "8000"]
# Dockerfile.agent
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY agents/ ./agents/
CMD ["python", "agents/planner_agent.py"]
# requirements.txt
fastapi>=0.109.0
uvicorn[standard]>=0.27.0
python-jose[cryptography]>=3.3.0
pydantic>=2.5.0
pydantic-settings>=2.1.0
httpx>=0.26.0
python-multipart>=0.0.6
6. Demo: Agents Talking to Each Other
Here's how the agents actually use the auth system:
# agents/planner_agent.py
import os, time, httpx
from auth_gateway.auth import verify_token
AUTH_GATEWAY = os.environ["AUTH_GATEWAY_URL"]
AGENT_ID = os.environ["AGENT_ID"]
AGENT_SECRET = os.environ["AGENT_SECRET"]
def get_token():
"""Step 1: Authenticate and get JWT"""
response = httpx.post(
f"{AUTH_GATEWAY}/token",
json={"agent_id": AGENT_ID, "secret": AGENT_SECRET},
)
response.raise_for_status()
return response.json()["access_token"]
def delegate_task(token: str, action: str, resource: str, target: str):
"""Step 2: Make authenticated request to another agent via gateway"""
headers = {"Authorization": f"Bearer {token}"}
response = httpx.post(
f"{AUTH_GATEWAY}/api/agent/{target}",
json={"action": action, "resource": resource},
headers=headers,
timeout=10.0,
)
return response.json()
def main():
print(f"[Planner Agent] Starting up as {AGENT_ID}")
# Authenticate
token = get_token()
print(f"[Planner Agent] JWT obtained, token starts: {token[:20]}...")
# Simulate delegating work
tasks = [
("write_tests", "test_auth.py", "coder"),
("review_code", "auth_gateway/", "reviewer"),
("deploy", "v1.2.3", "deploy"),
]
for action, resource, target in tasks:
try:
print(f"[Planner Agent] Delegating: {action} on {resource} → {target}")
result = delegate_task(token, action, resource, target)
print(f"[Planner Agent] Result: {result}")
except httpx.HTTPStatusError as e:
print(f"[Planner Agent] Failed: {e.response.status_code} - {e.response.text}")
time.sleep(1)
if __name__ == "__main__":
main()
7. Testing the System
Start the system:
export JWT_SECRET_KEY="super-secret-256-bit-key-change-in-production"
docker-compose up --build
Test authentication flow:
# 1. Health check
curl http://localhost:8000/health
# 2. Get a token for the planner agent
curl -X POST http://localhost:8000/token \
-H "Content-Type: application/json" \
-d '{"agent_id":"planner","secret":"secret_planner"}'
# 3. Make an authenticated request
TOKEN="<paste token here>"
curl -X POST http://localhost:8000/api/agent/code \
-H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" \
-d '{"action":"write_tests","resource":"test_math.py"}'
# 4. Test privilege escalation (should fail)
curl -X POST http://localhost:8000/api/agent/deploy \
-H "Authorization: Bearer $TOKEN" \
-d '{"action":"deploy"}'
# Expected: 403 Forbidden — planner role cannot deploy
Expected output for the privilege escalation test:
{
"error": "Forbidden",
"detail": "Agent 'planner' with role 'planner' does not have permission to perform 'call_deploy'"
}
8. What This Architecture Actually Solves
| Threat | How Zero-Trust Fixes It |
|---|---|
| Impersonation | Every request requires a valid, signed JWT from the token issuer |
| Token replay | Short-lived tokens (30 min default) + jti claim for blacklisting |
| Privilege escalation | RBAC enforced at middleware + endpoint level; roles are in the token and verified server-side |
| Lateral movement | Agent registry limits which agents can authenticate; unknown agents are rejected at the gateway |
| Man-in-the-middle | HTTPS enforced at the Docker network layer; JWTs are signed and tamper-proof |
Limitations to understand:
- This is not a complete security system — it's one critical layer. You still need network-level security, secret management (use HashiCorp Vault or AWS Secrets Manager in production), and comprehensive logging/monitoring.
- The agent registry is in-memory — in production, use a proper database (PostgreSQL) with hashed secrets and a proper OAuth2 authorization server.
- Token refresh is not implemented — production systems need a refresh token flow so agents don't lose their session every 30 minutes.
9. Extending This System
Once the auth gateway is in place, these extensions are natural next steps:
MCP Integration: The Model Context Protocol can use this gateway for agent-to-MCP-server authentication, replacing ad-hoc API key management.
Short-lived tokens for tools: When an agent calls a tool (file system, git, CI/CD), generate a scoped token with only the permissions that tool needs. The tool validates the token independently.
Audit log aggregation: All authentication events are logged. Ship these to a SIEM (Elasticsearch, Splunk) for anomaly detection — detect compromised agents by their authentication patterns.
mTLS for service mesh: Add mutual TLS so agents verify not just the token but the network connection itself. This prevents both impersonation and traffic interception.
Summary
The multi-agent security problem is real, and it grows with every agent you add. The solution isn't to "trust internal agents" — it's to apply the same Zero-Trust principles that cloud infrastructure learned the hard way:
- Never trust, always verify — every request needs a valid JWT
- Least privilege — tokens carry only the permissions the specific role needs
- Short-lived credentials — tokens expire in 30 minutes, not 30 days
- Audit everything — every authentication event is logged
- Defense in depth — middleware + endpoint-level checks + RBAC
The code in this article gives you a production-ready starting point. Extend it, break it, improve it — and build agent systems that are secure by design, not by accident.
This is the third article in the series on building production-ready AI agent infrastructure. The first covered the progressive evolution of the AI stack. The second covered Socratic prompting techniques.
Top comments (0)