cbc9ad73f7
Reference implementation of GCAP-SPEC-SHELLBOUND-BROKER-0001
in FastAPI. Designed for ISVs and enterprises implementing
the governed shell authorization protocol.
Architecture:
FastAPI + SQLModel + Pydantic + async SQLite
Single container deployment: Dockerfile included
OpenAPI schema at /docs is the machine-readable GSAP contract
Broker Interface (§5):
POST /governance/authorize/ — Issue AC
GET /governance/authorize/{p}/ — Poll elevation
POST /governance/complete/ — Receive CR
GET /governance/session/{id}/ — View chain of custody
POST /governance/elevate/ — JIT elevation
GET /governance/drivers/ — List drivers
Identity Driver Interface (§2.2):
IdentityDriver — abstract base (ISV extension point)
KeycloakDriver — Keycloak implementation
DriverRegistry — driver lookup and registration
Chronicle integration (§1.4):
Optional CloudEvents emission via CHRONICLE_WEBHOOK_URL
Forgejo push event format for receiver compatibility
Models:
Pydantic schemas for AC, CR, Principal, Accord, Operation
SQLModel DB models for persistence
Tests: 6 async tests including full AC→CR cycle
695 lines across 27 files.
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
31 lines
1.6 KiB
Python
31 lines
1.6 KiB
Python
"""GET /governance/session/{id}/ — GSAP §5.5"""
|
|
import uuid
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlmodel.ext.asyncio.session import AsyncSession
|
|
from sqlmodel import select
|
|
from gsap_broker.db import get_session
|
|
from gsap_broker.db_models import AuthorizationContextDB, CompletionReceiptDB
|
|
from gsap_broker.models import SessionResponse, ACStatus
|
|
|
|
router = APIRouter()
|
|
|
|
@router.get("/session/{context_id}/", response_model=SessionResponse, summary="View session (GSAP §5.5)")
|
|
async def session_detail(context_id: uuid.UUID, db: AsyncSession = Depends(get_session)):
|
|
result = await db.exec(select(AuthorizationContextDB).where(AuthorizationContextDB.context_id == context_id))
|
|
ac_db = result.first()
|
|
if not ac_db: raise HTTPException(status_code=404, detail="Not found.")
|
|
|
|
cr_result = await db.exec(select(CompletionReceiptDB).where(CompletionReceiptDB.context_id == context_id))
|
|
cr_db = cr_result.first()
|
|
cr_data = None
|
|
if cr_db:
|
|
cr_data = {"outcome": cr_db.outcome, "behavioral_attestation": cr_db.behavioral_attestation_status,
|
|
"chronicle_session_id": cr_db.chronicle_session_id, "received_at": cr_db.received_at.isoformat(),
|
|
"signature_verified": cr_db.signature_verified}
|
|
|
|
return SessionResponse(
|
|
context_id=ac_db.context_id, principal_did=ac_db.principal_did,
|
|
accord_template=ac_db.accord_template, playbook=ac_db.playbook,
|
|
status=ACStatus(ac_db.status), issued_at=ac_db.issued_at,
|
|
expires_at=ac_db.expires_at, consumed_at=ac_db.consumed_at,
|
|
chronicle_event_cid=ac_db.chronicle_event_cid or None, completion_receipt=cr_data)
|