Agentic Airlock

Overview

Airlock is an open protocol that answers a question nobody else has solved: how do you verify that an AI agent is who it claims to be before letting it act?

Transport security (TLS) encrypts the pipe. Authorization frameworks (OAuth) control what an agent can do. But neither verifies the agent's identity or trustworthiness. An agent with stolen API keys passes both checks. Airlock catches it.

The Problem

The AI agent ecosystem is growing fast. Google A2A lets agents communicate. Anthropic MCP gives agents tools. But there is no layer that verifies agent identity.

Email took 20 years after the spam crisis to add SPF, DKIM, and DMARC. We are building the equivalent for AI agents before the trust crisis hits.

Quick Start

Install

# Core install
pip install -e ".[dev]"

# With Redis support
pip install -e ".[dev,redis]"

# With framework integrations
pip install -e ".[dev,langchain,openai,anthropic]"

Start the Gateway

# Create .env (at minimum)
echo "AIRLOCK_ENV=development" > .env
echo "AIRLOCK_LITELLM_MODEL=ollama/llama3" >> .env

# Launch
python -m uvicorn airlock.gateway.app:create_app --factory --port 8000 --env-file .env

Run the Demo

# Live demo against running gateway (3 scenarios)
python demo_trust_flow.py

# In-process demo (no gateway needed)
python examples/run_demo.py

Run Tests

python -m pytest tests/ -v   # 306 tests

Demo Scenarios

The demo runs three scenarios against a live gateway:

Architecture

The gateway receives agent handshakes over HTTP (or WebSocket), dispatches them through a 10-node LangGraph state machine, and returns a cryptographically signed attestation with a trust token.

5-Phase Verification Pipeline

Orchestrator Graph (LangGraph)

The verification engine is a 10-node LangGraph state machine:

Nodes highlighted in purple (check_revocation, validate_delegation, issue_verdict) were added during the hardening phase. Each node writes to shared PipelineState and appends to the verification trace.

Identity Layer

DID:key Method

Every agent identity is an Ed25519 keypair encoded as a W3C DID:key. The public key is represented using the multicodec ed25519-pub prefix (0xed01) and base58btc multibase encoding.

from airlock.crypto.keys import KeyPair

# Generate a new agent identity
kp = KeyPair.generate()

print(kp.did)
# "did:key:z6Mkf5rGMoatrSj1f4CyvuHBeXJ..."

print(kp.public_key_multibase)
# "z6Mkf5rGMoatrSj1f4CyvuHBeXJ..."

DID Resolution

from airlock.crypto.keys import resolve_public_key

# Extract Ed25519 VerifyKey from any did:key string
verify_key = resolve_public_key("did:key:z6Mkf5rGMo...")
# Returns nacl.signing.VerifyKey

W3C Verifiable Credentials

Agents present a W3C VC with an Ed25519Signature2020 proof issued by a trusted authority. The VC binds the agent DID to an authorization scope.

from airlock.crypto.vc import issue_vc, verify_vc
from datetime import datetime, timedelta, timezone

vc = issue_vc(
    issuer_kp=issuer_keypair,
    subject_did=agent_kp.did,
    credential_type="AgentAuthorization",
    claims={"scope": "payments", "level": "full"},
    valid_until=datetime.now(timezone.utc) + timedelta(days=365),
)

# Verify: checks signature, expiry, subject binding
is_valid = verify_vc(vc, expected_subject=agent_kp.did)

Trust Scoring

Decay Formula

# Half-life decay toward baseline (0.50)
elapsed = (now - last_update).total_seconds()
decay = 0.5 ** (elapsed / HALF_LIFE_SECONDS)
score = BASELINE + (raw_score - BASELINE) * decay

A trust score of 0.90 reached after many verifications will naturally decay to 0.70 after 30 days of inactivity, pushing the agent back into the challenge zone. This prevents stale high-trust credentials from being exploited.

Semantic Challenge

When an agent's trust score falls between 0.15 and 0.75, the orchestrator routes to the semantic challenge node. An LLM generates a domain-specific question, the agent answers, and the LLM evaluates the response.

Challenge Flow

1. Generate: LLM creates a question based on the agent's declared domain (e.g., "payments")
2. Deliver: Challenge sent via callback URL or WebSocket
3. Answer: Agent submits response to POST /challenge/submit
4. Evaluate: LLM grades the answer: PASS, FAIL, or AMBIGUOUS

Security Mitigations

• Answer sanitization: _sanitize_answer() strips control characters, enforces 2000-char limit
• Prompt injection warning: Evaluation prompt explicitly warns the LLM about manipulation
• Timeout: Both LLM calls have a 30-second timeout

Rule-Based Fallback

When the LLM is unavailable, Airlock falls back to a rule-based evaluator:

# Enable via environment variable
AIRLOCK_CHALLENGE_FALLBACK_MODE=rule_based

The rule evaluator uses domain keyword matching, evasion pattern detection (phrases like "I don't know", "as an AI"), and answer complexity heuristics (minimum 15 unique words, repetition check).

Delegation Model

Airlock supports 1-hop delegation where a trusted agent can vouch for a sub-agent.

# HandshakeRequest with delegation
{
    "delegator_did": "did:key:z6Mkf5rG...",
    "delegation": {
        "scope": "payments.read",
        "max_depth": 1,
        "expires_at": "2026-05-01T00:00:00Z"
    }
}

Validation Rules

• Delegator must not be revoked
• Delegator's trust score must be ≥ 0.75
• Chain depth must not exceed max_depth
• Delegation must not be expired

Cascade Revocation

Revoking a delegator automatically revokes all delegates. The RevocationStore tracks the delegation graph and cascades on revoke.

Revocation

# Revoke an agent (cascades to delegates)
POST /admin/revoke/did:key:z6Mkf5rG...
Authorization: Bearer <admin_token>

# List all revoked DIDs
GET /admin/revoked
Authorization: Bearer <admin_token>

# Lift revocation
POST /admin/unrevoke/did:key:z6Mkf5rG...
Authorization: Bearer <admin_token>

Storage Backends

The Redis backend maintains a local_cache for synchronous lookups required by LangGraph graph nodes (is_revoked_sync), refreshed on startup via sync_cache().

Audit Trail

Every gateway action is recorded in a hash-chained audit log. Each entry's hash is computed as SHA-256(canonical_json(entry) + previous_hash). The genesis entry uses "0" * 64 as the previous hash.

# Verify chain integrity
GET /admin/audit/verify
Authorization: Bearer <admin_token>

# Response
{"valid": true, "entries": 142, "chain_tip": "a3f8c2..."}

# If tampered:
{"valid": false, "broken_at": 87, "expected": "c4a7...", "got": "f19b..."}

The public endpoint GET /audit/latest returns the chain tip hash, allowing external systems to anchor trust proofs.

API Reference

Public Endpoints

Admin Endpoints

All admin endpoints require Authorization: Bearer <AIRLOCK_ADMIN_TOKEN>

A2A-Native Endpoints

Compatible with the Google Agent-to-Agent (A2A) protocol:

Trust Tokens (JWT)

On VERIFIED verdict, the gateway issues an HS256 JWT trust token. Relying parties can introspect the token without contacting the gateway again.

JWT Claims

# Introspect a trust token
POST /token/introspect
Authorization: Bearer <service_token>
Content-Type: application/json

{"token": "eyJhbGciOiJIUzI1NiJ9..."}

# Response
{"active": true, "sub": "did:key:z6Mk...", "ts": 0.82, "ver": "VERIFIED"}

Core Schemas

All schemas are Pydantic v2 models.

MessageEnvelope

class MessageEnvelope(BaseModel):
    protocol_version: str        # "0.1.0"
    message_id: str             # UUID
    timestamp: datetime         # ISO 8601
    sender_did: str             # did:key:z6Mk...
    nonce: str                  # hex(32 random bytes)

HandshakeRequest

class HandshakeRequest(BaseModel):
    envelope: MessageEnvelope
    session_id: str
    initiator: AgentDID
    intent: HandshakeIntent      # action, description, target_did
    credential: VerifiableCredential
    signature: str | None       # Ed25519 over canonical JSON
    # Optional delegation fields
    delegator_did: str | None
    delegation: DelegationIntent | None
    credential_chain: list | None

AirlockAttestation

class AirlockAttestation(BaseModel):
    session_id: str
    verified_did: str
    verdict: TrustVerdict        # VERIFIED | REJECTED | DEFERRED
    trust_score: float
    checks_passed: list[CheckResult]
    issued_at: datetime
    trust_token: str | None    # JWT (only on VERIFIED)

Python SDK

AirlockClient

from airlock.sdk.client import AirlockClient
from airlock.crypto.keys import KeyPair

kp = KeyPair.generate()

async with AirlockClient(
    gateway_url="https://airlock.example.com",
    agent_keypair=kp,
) as client:
    # Register
    await client.register(
        display_name="PaymentBot",
        capabilities=["payments", "refunds"],
    )

    # Handshake (builds + signs request automatically)
    result = await client.handshake(
        target_did="did:key:z6MkTarget...",
        action="transfer_funds",
        description="Process UPI payment",
    )

    if result.verdict == "VERIFIED":
        print(f"Trust token: {result.trust_token}")

ASGI Middleware

from airlock.sdk.middleware import AirlockMiddleware

airlock = AirlockMiddleware(
    gateway_url="https://airlock.example.com",
    agent_keypair=my_keypair,
)

@airlock.protect
async def handle_incoming(request: HandshakeRequest):
    # Only reached if agent is VERIFIED
    return {"status": "ok"}

Framework Integrations

Drop-in wrappers for the three major AI agent frameworks. All use deferred imports so installing the framework is optional.

LangChain

from airlock.integrations.langchain import AirlockToolGuard

# Wrap any LangChain tool — it verifies agent trust before _arun
guard = AirlockToolGuard(gateway_url="...", agent_keypair=kp)
protected_tool = guard.wrap(my_tool)

# Use protected_tool in your LangChain agent chain as normal
# The guard intercepts _arun, runs handshake, only proceeds if VERIFIED

OpenAI Agents SDK

from airlock.integrations.openai_agents import airlock_guard

@airlock_guard(gateway_url="...", agent_keypair=kp)
async def my_tool(query: str) -> str:
    # Only runs if agent is VERIFIED
    return f"Result for {query}"

Anthropic SDK

from airlock.integrations.anthropic_sdk import AirlockToolInterceptor

interceptor = AirlockToolInterceptor(gateway_url="...", agent_keypair=kp)

# Before executing any tool_use content block:
await interceptor.verify_before_tool(
    tool_name="search_database",
    tool_input={"query": "user records"},
)
# Raises PermissionError if agent is REJECTED

Integration with Authorization

Airlock is a trust verification layer. It does not replace authorization frameworks like OAuth or GNAP. Instead, it answers the question that comes before authorization: is this agent who it claims to be?

How It Fits

The trust token (JWT) issued by Airlock on a VERIFIED verdict can be consumed by any authorization server as an input to its grant decision.

# Example: OPA policy consuming Airlock trust score
package authz

allow {
    input.airlock.verified == true
    input.airlock.trust_score > 0.75
    input.oauth.scope == "payments.read"
}

Integration Patterns

• Token Exchange (RFC 8693): Exchange Airlock trust token for an OAuth access token. The authorization server validates trust before issuing scopes.
• Policy Engine Input: Feed Airlock trust score and verification status into OPA, Cedar, or any policy engine as decision variables.
• Continuous Trust Signal: Airlock can notify authorization servers when an agent's trust degrades, triggering token revocation (RFC 7009).
• Gateway Middleware: Place Airlock verification before your API gateway's authorization check. Agent must be trusted before authorization is even evaluated.

Configuration

All settings are managed via environment variables with the AIRLOCK_ prefix, backed by a Pydantic BaseSettings class.

Critical Settings

Deployment

Docker Compose

# Clone + configure
git clone https://github.com/shivdeep1/airlock-protocol
cd airlock-protocol
cp .env.example .env
# Edit .env — set AIRLOCK_GATEWAY_SEED_HEX (64 hex chars)

# Launch
docker compose up --build

Health Probes

Multi-Replica HA

• Point all gateway instances at the same AIRLOCK_REDIS_URL
• Mount shared LanceDB storage or use AIRLOCK_DEFAULT_REGISTRY_URL for federation
• Put behind a TCP/HTTP load balancer with /health checks

Monitoring

Prometheus Metrics

Alert Recommendations

• High rejection rate: rate(airlock_verdicts_total{type="REJECTED"}[5m])
• LLM fallback active: spike in airlock_challenges_total{outcome="AMBIGUOUS"}
• Event bus saturation: airlock_event_bus_queue_depth approaching max (1000)
• Dead letters: any increase in airlock_event_bus_dead_letters_total

Security

Airlock has been through an internal security audit. Six vulnerabilities were identified and fixed.

Security Design

• SSRF protection: validate_callback_url() blocks 127.x, 10.x, 192.168.x, 172.16-31.x, 169.254.x
• Replay protection: Every nonce consumed once (in-memory set or Redis)
• Rate limiting: Per-IP and per-DID limits, configurable
• Canonical JSON signing: RFC 8785 canonicalization before Ed25519 signing
• VC issuer allowlist: Only accept credentials from whitelisted issuers in production
• Sybil protection: Registration cap per IP per rolling hour

Module Index

airlock/crypto/

airlock/engine/

airlock/gateway/

airlock/integrations/

airlock/sdk/

Performance

Target latency: sub-200ms for all verification paths. Fast-path achieves 73ms average with pure Ed25519 cryptography (no LLM call). The challenge path adds LLM latency when triggered, but 95%+ of verifications for known agents take the fast path.

Test Suite

306 tests, all passing. Organized across 30 test files covering every subsystem.

python -m pytest tests/ -v --tb=short   # 306 passed in ~140s

Standards & References

• W3C DID Core — Decentralized Identifier specification (did:key method)
• W3C VC Data Model 1.1 — Verifiable Credential format and proof types
• RFC 8032 — Ed25519 digital signatures (EdDSA)
• RFC 7519 — JSON Web Tokens (JWT)
• RFC 8785 — JSON Canonicalization Scheme (JCS) for deterministic signing
• RFC 2119 — Key words for use in RFCs (MUST, SHOULD, MAY)
• RFC 7807 — Problem Details for HTTP APIs (error responses)
• Google A2A — Agent-to-Agent protocol (Agent Cards, Tasks, Messages)
• Anthropic MCP — Model Context Protocol (tool execution framework)
• Multibase / Multicodec — Self-describing encoding for DID key material

Formal Specifications

• docs/PROTOCOL_SPEC.md — 790-line RFC-style specification (12 sections + 3 appendices)
• docs/draft-airlock-agent-trust-00.md — 1226-line IETF Internet-Draft (formal submission format)