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>
54 lines
1.9 KiB
Python
54 lines
1.9 KiB
Python
import uuid
|
|
from datetime import datetime
|
|
from typing import Optional
|
|
from sqlmodel import Field, SQLModel, Column
|
|
from sqlalchemy import JSON
|
|
|
|
class AuthorizationContextDB(SQLModel, table=True):
|
|
__tablename__ = "authorization_contexts"
|
|
context_id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
|
|
principal_did: str
|
|
driver_id: str
|
|
playbook: str
|
|
corpus_entry_cid: str
|
|
parameters_cid: str
|
|
accord_template: str
|
|
capability_mask: int = 3
|
|
idp_vendor: str = "keycloak"
|
|
token_jti: str = ""
|
|
elevation_active: list = Field(default=[], sa_column=Column(JSON))
|
|
mfa_satisfied: bool = False
|
|
status: str = "authorized"
|
|
issued_at: datetime = Field(default_factory=datetime.utcnow)
|
|
expires_at: datetime
|
|
consumed_at: Optional[datetime] = None
|
|
poll_token: Optional[str] = None
|
|
chronicle_event_cid: str = ""
|
|
|
|
class CompletionReceiptDB(SQLModel, table=True):
|
|
__tablename__ = "completion_receipts"
|
|
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
|
|
context_id: uuid.UUID = Field(foreign_key="authorization_contexts.context_id")
|
|
outcome: str
|
|
completed_at: datetime
|
|
received_at: datetime = Field(default_factory=datetime.utcnow)
|
|
failure_reason: str = ""
|
|
chronicle_session_id: str = ""
|
|
chronicle_events: list = Field(default=[], sa_column=Column(JSON))
|
|
merkle_root: str = ""
|
|
behavioral_attestation_status: str = "unavailable"
|
|
ffc_did: str = ""
|
|
ffc_signature: str = ""
|
|
signature_verified: bool = False
|
|
chronicle_event_cid: str = ""
|
|
|
|
class ElevationRequestDB(SQLModel, table=True):
|
|
__tablename__ = "elevation_requests"
|
|
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
|
|
principal_id: str = ""
|
|
role_name: str
|
|
justification: str = ""
|
|
duration_minutes: int = 60
|
|
status: str = "pending"
|
|
requested_at: datetime = Field(default_factory=datetime.utcnow)
|
|
expires_at: Optional[datetime] = None
|