fastapi-gsap/gsap_broker/db_models.py
Tyler J King cbc9ad73f7 feat: fastapi-gsap — lightweight GSAP broker PoC
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>
2026-03-30 14:10:21 -04:00

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