Tyler King
944b3fde19
FastAPI companion service to the GSAP broker for AI agent
identity delegation in governed shell sessions.
Implements GCAP-SPEC-LLM-PRINCIPAL-BROKER-0001:
POST /delegate — request delegation (human → AI agent)
POST /delegate/{id}/revoke — revoke delegation
GET /delegate/{id} — delegation status
GET /agents — list active delegations
GET /health — health check
Delegation lifecycle:
REQUESTED → ACTIVE → EXPIRED | REVOKED
Cascading revocation on delegator AC revocation
Background cleanup of expired delegations (30s interval)
Keycloak integration:
Registers ephemeral agent clients per delegation
Deletes clients on revocation/expiry
Dev mode: stubs when no client_secret configured
GSAP broker integration:
Requests delegated ACs via on_behalf_of pattern
Scope narrowing: agent ceiling ≤ delegator ceiling
Dev mode: stubs when no bearer_token configured
Chronicle integration:
DELEGATION_CREATED (0x3001)
DELEGATION_REVOKED (0x3003)
DELEGATION_EXPIRED (0x3004)
All 7 smoke tests pass (health, create, list, query, revoke, verify, empty).
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
65 lines
2.1 KiB
Python
65 lines
2.1 KiB
Python
"""Chronicle CloudEvents client — matches fastapi-gsap chronicle pattern."""
|
|
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
from datetime import datetime, UTC
|
|
|
|
import httpx
|
|
|
|
from llm_broker.settings import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
async def emit(kind: str, payload: dict) -> str:
|
|
"""Post a delegation event to Chronicle. Returns CID or empty string."""
|
|
url = settings.chronicle_webhook_url
|
|
if not url:
|
|
return ""
|
|
try:
|
|
event_json = json.dumps({"kind": kind, **payload}, sort_keys=True, default=str)
|
|
cid = "sha256:" + hashlib.sha256(event_json.encode()).hexdigest()
|
|
async with httpx.AsyncClient(timeout=5.0) as client:
|
|
await client.post(url, json={
|
|
"pusher": {"login": payload.get("delegator_did", settings.broker_did)},
|
|
"ref": f"refs/llm-broker/{kind}",
|
|
"repository": {"full_name": "llm-broker/delegation"},
|
|
"commits": [{"message": f"{kind}: {json.dumps(payload, default=str)}"}],
|
|
}, headers={"X-Forgejo-Event": "push"})
|
|
return cid
|
|
except Exception as e:
|
|
logger.warning("Chronicle emit failed: %s: %s", kind, e)
|
|
return ""
|
|
|
|
|
|
async def delegation_created(
|
|
delegation_id: str, delegator_did: str, agent_did: str,
|
|
agent_type: str, scope: dict,
|
|
) -> str:
|
|
return await emit("DELEGATION_CREATED", {
|
|
"event_code": "0x3001",
|
|
"delegation_id": delegation_id,
|
|
"delegator_did": delegator_did,
|
|
"agent_did": agent_did,
|
|
"agent_type": agent_type,
|
|
"scope": scope,
|
|
"timestamp": datetime.now(UTC).isoformat(),
|
|
})
|
|
|
|
|
|
async def delegation_revoked(delegation_id: str, reason: str) -> str:
|
|
return await emit("DELEGATION_REVOKED", {
|
|
"event_code": "0x3003",
|
|
"delegation_id": delegation_id,
|
|
"reason": reason,
|
|
"timestamp": datetime.now(UTC).isoformat(),
|
|
})
|
|
|
|
|
|
async def delegation_expired(delegation_id: str) -> str:
|
|
return await emit("DELEGATION_EXPIRED", {
|
|
"event_code": "0x3004",
|
|
"delegation_id": delegation_id,
|
|
"timestamp": datetime.now(UTC).isoformat(),
|
|
})
|