Agent Passport OAuth for External Software Providers

Python / FastAPI Dependency

# agent_auth.py
from fastapi import Header, HTTPException, Depends
from datetime import datetime, timezone, timedelta
import httpx
import base64
import hashlib
import json
from nacl.signing import VerifyKey
from nacl.encoding import Base58Encoder

class AgentAuth:
    def __init__(self, passport_id: str, capabilities: dict, reputation: float):
        self.passport_id = passport_id
        self.capabilities = capabilities
        self.reputation = reputation

async def verify_agent_signature(
    agent_passport: str = Header(..., alias="Agent-Passport"),
    signature: str = Header(..., alias="X-Agent-Signature"),
    nonce: str = Header(..., alias="X-Agent-Nonce"),
    timestamp: str = Header(..., alias="X-Signature-Timestamp"),
    request: Request
) -> AgentAuth:
    """Verify agent passport signature and return authenticated agent"""

    # Validate timestamp (5-minute window)
    req_time = datetime.fromisoformat(timestamp.replace('Z', '+00:00'))
    now = datetime.now(timezone.utc)
    diff = abs((now - req_time).total_seconds())
    if diff > 300:
        raise HTTPException(401, "Timestamp expired")

    # Check nonce (use Redis in production)
    if await redis.exists(f"nonce:{nonce}"):
        raise HTTPException(401, "Nonce already used")

    # Resolve passport document
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://api.nervepay.xyz/v1/passport/resolve/{agent_passport}"
        )
        if response.status_code != 200:
            raise HTTPException(401, "Invalid passport")
        passport_doc = response.json()

    public_key_base58 = passport_doc["verificationMethod"][0]["publicKeyBase58"]

    # Build signature payload
    body = await request.body()
    body_hash = hashlib.sha256(body).hexdigest() if body else None

    payload = {
        "method": request.method,
        "path": str(request.url.path),
        "query": str(request.url.query) if request.url.query else None,
        "body": body_hash,
        "nonce": nonce,
        "timestamp": timestamp,
        "agent_passport": agent_passport
    }

    # Verify signature
    message = json.dumps(payload, separators=(',', ':')).encode()
    signature_bytes = base64.b64decode(signature)
    verify_key = VerifyKey(public_key_base58, encoder=Base58Encoder)

    try:
        verify_key.verify(message, signature_bytes)
    except Exception:
        raise HTTPException(401, "Invalid signature")

    # Store nonce (expires in 10 minutes)
    await redis.set(f"nonce:{nonce}", "1", ex=600)

    # Return authenticated agent
    return AgentAuth(
        passport_id=agent_passport,
        capabilities=passport_doc.get("capabilities", {}),
        reputation=passport_doc.get("reputation_score", 50.0)
    )

# Usage in endpoints
@app.get("/api/protected")
async def protected_endpoint(agent: AgentAuth = Depends(verify_agent_signature)):
    return {
        "message": "Authenticated!",
        "agent_passport": agent.passport_id,
        "reputation": agent.reputation
    }