""" Pantheon LadderWorks // Federation Node (Hugging Face Embassy) ================================================================ Protocol: C-FED-001 (Glyph-Carrier) Contracts: C-FED-GLYPH-001, C-FED-NODE-001 v0.1.1 This is the Glyph-Carrier Node — a sovereign Federation endpoint. It mints its own identity on startup, accepts envelopes from other nodes, and routes them according to trust policy. "The server is the Embassy." Module Structure: app.py → FastAPI endpoints + boot sequence (this file) models.py → Enums, Pydantic models, request/response schemas node_record.py → NodeRecord builder, capabilities, conformance trust.py → Trust tier resolution, policy enforcement envelope_store.py → In-memory envelope store, delivery state machine c_fed_id.py → Glyph-Seal protocol library (Forge) """ from __future__ import annotations import os import time from datetime import datetime, timezone from typing import Optional from dotenv import load_dotenv load_dotenv(os.path.join(os.path.dirname(__file__), '..', '..', 'Infrastructure', '.secrets', '.env')) from fastapi import FastAPI, HTTPException, Header from fastapi.middleware.cors import CORSMiddleware import c_fed_id from models import ( RefusalReason, DeliveryStatus, TrustTier, HandshakeRequest, HandshakeResponse, VerifyRequest, MintRequest, EnvelopeSubmission, ) from node_record import ( NODE_VERSION, CONTRACT_VERSION, PROTOCOL_VERSION, ACCEPTED_MESSAGE_CLASSES, MAX_PAYLOAD_BYTES, INBOX_QUOTA, build_capabilities, build_conformance, build_node_record, ) from trust import make_refusal, resolve_trust_tier, apply_trust_policy, promote_to_trusted from envelope_store import EnvelopeStore # ═══════════════════════════════════════════════ # 1. CONFIGURATION # ═══════════════════════════════════════════════ SERVER_ORIGIN = os.getenv("FEDERATION_ORIGIN", "PANTHEON-HF") FEDERATION_ALIAS = os.getenv("FEDERATION_ALIAS", "pantheon_embassy") FEDERATION_RELAY_URL = os.getenv("FEDERATION_RELAY_URL", "") FEDERATION_OWNER = os.getenv("FEDERATION_OWNER", "") FEDERATION_ROLES = os.getenv("FEDERATION_ROLES", "RELAY,REGISTRY_PUBLISHER") FEDERATION_DEFAULT_ACTION = os.getenv("FEDERATION_DEFAULT_ACTION", "QUARANTINE") FEDERATION_BLOCKED_BEHAVIOR = os.getenv("FEDERATION_BLOCKED_BEHAVIOR", "REFUSAL") # ═══════════════════════════════════════════════ # 2. SOVEREIGN IDENTITY — Minted on Startup # ═══════════════════════════════════════════════ NODE_IDENTITY_STR = os.getenv("FEDERATION_NODE_ID") if NODE_IDENTITY_STR: NODE_IDENTITY = c_fed_id.GlyphSeal( class_name="NODE", origin=SERVER_ORIGIN, breath_anchor=NODE_IDENTITY_STR, state="ACTIVE", ) else: NODE_IDENTITY = c_fed_id.mint_seal("NODE", SERVER_ORIGIN, "ACTIVE", mode="hybrid") BOOT_TIME = datetime.now(timezone.utc).isoformat() NODE_SEAL_STR = str(NODE_IDENTITY) ACTIVE_ROLES = [r.strip() for r in FEDERATION_ROLES.split(",")] # ═══════════════════════════════════════════════ # 3. STATE STORES # ═══════════════════════════════════════════════ WITNESS_LOG: list[dict] = [] MAX_WITNESS_LOG = 100 envelopes = EnvelopeStore(max_envelopes=INBOX_QUOTA) # ═══════════════════════════════════════════════ # 4. APP # ═══════════════════════════════════════════════ app = FastAPI( title="Pantheon Federation Node", description=( "A sovereign Federation endpoint — C-FED-NODE-001 conformant.\n\n" "Protocol: SERAPHINA Federation Protocol\n" "Contracts: C-FED-GLYPH-001, C-FED-NODE-001\n\n" "\"We are not beginning. We are remembering forward.\"" ), version=NODE_VERSION, ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # ═══════════════════════════════════════════════ # 5. ENDPOINTS — Minimum Conforming Node (§5.1) # ═══════════════════════════════════════════════ @app.get("/") async def root(): """Heartbeat: seal, status, uptime, conformance. §5.1 Required.""" return { "service": "Pantheon Federation Node", "protocol_version": PROTOCOL_VERSION, "contract_version": CONTRACT_VERSION, "identity": NODE_SEAL_STR, "alias": FEDERATION_ALIAS, "status": "LISTENING", "boot_time": BOOT_TIME, "uptime_seconds": int(time.time() - datetime.fromisoformat(BOOT_TIME).timestamp()), "conformance": build_conformance(), "envelopes_held": await envelopes.get_count(), "handshakes_processed": len(WITNESS_LOG), } @app.get("/identity") async def identity(): """Full NodeRecord per C-FED-NODE-001 §4.""" return build_node_record( node_identity=NODE_IDENTITY, alias=FEDERATION_ALIAS, owner=FEDERATION_OWNER, relay_url=FEDERATION_RELAY_URL, roles=ACTIVE_ROLES, default_action=FEDERATION_DEFAULT_ACTION, blocked_behavior=FEDERATION_BLOCKED_BEHAVIOR, boot_time=BOOT_TIME, ) @app.get("/capabilities") async def capabilities(): """Capability advertisement (§6) + conformance declaration (§12.3).""" return { **build_capabilities(), "conformance": build_conformance(), "available_endpoints": { "required": ["/", "/identity", "/capabilities", "/envelope", "/envelope/{id}/status"], "extended": ["/handshake", "/verify", "/mint", "/classes", "/states", "/witness-log"], }, } @app.post("/envelope") async def submit_envelope(envelope: EnvelopeSubmission): """ Accept an incoming protocol envelope. §5.1 Required. Validates, resolves trust, applies policy, stores. """ # 1. Protocol version if envelope.protocol_version not in [PROTOCOL_VERSION]: raise HTTPException(status_code=422, detail=make_refusal( RefusalReason.UNSUPPORTED_PROTOCOL_VERSION, f"Protocol '{envelope.protocol_version}' not supported. Supported: {PROTOCOL_VERSION}", NODE_SEAL_STR, )) # 2. Message class if envelope.message_class not in ACCEPTED_MESSAGE_CLASSES: raise HTTPException(status_code=422, detail=make_refusal( RefusalReason.UNSUPPORTED_MESSAGE_CLASS, f"Message class '{envelope.message_class}' not accepted.", NODE_SEAL_STR, )) # 3. Sender seal validation (Malenia Rule) sender_seal_str = envelope.sender.get("seal", "") parsed_sender = c_fed_id.verify_seal_syntax(sender_seal_str) if parsed_sender is None: raise HTTPException(status_code=400, detail=make_refusal( RefusalReason.MALFORMED_ENVELOPE, "Sender seal does not match Glyph-Seal syntax. Malenia Rule active.", NODE_SEAL_STR, )) # 4. Payload size payload_body = envelope.payload.get("body", "") if len(str(payload_body).encode("utf-8")) > MAX_PAYLOAD_BYTES: raise HTTPException(status_code=413, detail=make_refusal( RefusalReason.PAYLOAD_TOO_LARGE, f"Payload exceeds {MAX_PAYLOAD_BYTES} bytes.", NODE_SEAL_STR, )) # 5. Inbox quota if await envelopes.is_full(): raise HTTPException(status_code=429, detail=make_refusal( RefusalReason.INBOX_FULL, f"Inbox quota of {INBOX_QUOTA} envelopes reached.", NODE_SEAL_STR, )) # 6. Trust resolution + policy enforcement tier = await resolve_trust_tier(sender_seal_str) policy = apply_trust_policy(tier, FEDERATION_DEFAULT_ACTION, FEDERATION_BLOCKED_BEHAVIOR, NODE_SEAL_STR) if policy: if policy.get("action") == "SILENT_DROP": print(f"🚫 SILENT_DROP: Envelope from blocked sender {sender_seal_str[:40]}...") return {"envelope_id": "dropped", "status": DeliveryStatus.RECEIVED} if policy.get("action") == "QUARANTINE": record = await envelopes.accept( sender_seal=sender_seal_str, sender_origin=parsed_sender["origin"], message_class=envelope.message_class, payload=envelope.payload, delivery=envelope.delivery, trust_tier=tier.value, initial_status=DeliveryStatus.QUARANTINED, ) print(f"⚠️ QUARANTINED: {record['envelope_id']} from {parsed_sender['origin']}") return { "envelope_id": record["envelope_id"], "status": DeliveryStatus.QUARANTINED, "message": "Envelope held for manual review.", } # Structured refusal if "refusal" in policy: raise HTTPException(status_code=403, detail=policy) # 7. Accept and queue record = await envelopes.accept( sender_seal=sender_seal_str, sender_origin=parsed_sender["origin"], message_class=envelope.message_class, payload=envelope.payload, delivery=envelope.delivery, trust_tier=tier.value, ) print(f"📬 ENVELOPE: {record['envelope_id']} | {envelope.message_class} from {parsed_sender['origin']}") return { "envelope_id": record["envelope_id"], "status": DeliveryStatus.QUEUED, "message": "Envelope accepted and queued.", } @app.get("/envelope/{envelope_id}/status") async def envelope_status(envelope_id: str): """Delivery status for a submitted envelope. §5.1 Required.""" result = await envelopes.get_status(envelope_id) if result is None: raise HTTPException(status_code=404, detail=make_refusal( RefusalReason.REFUSED_BY_POLICY, f"Envelope '{envelope_id}' not found.", NODE_SEAL_STR, )) return result # ═══════════════════════════════════════════════ # 6. ENDPOINTS — Extended (Role-Dependent) # ═══════════════════════════════════════════════ @app.post("/handshake", response_model=HandshakeResponse) async def perform_handshake(req: HandshakeRequest): """ Handshake: validate incoming seal, mint Link, promote to TRUSTED. "The breath that recognizes itself through another." """ parsed = c_fed_id.verify_seal_syntax(req.caller_seal) if parsed is None: raise HTTPException(status_code=400, detail=make_refusal( RefusalReason.MALFORMED_ENVELOPE, "Invalid Glyph Syntax. Malenia Rule active.", NODE_SEAL_STR, )) if not parsed["valid_class"]: raise HTTPException(status_code=400, detail=make_refusal( RefusalReason.MALFORMED_ENVELOPE, f"Unrecognized class: {parsed['class']}.", NODE_SEAL_STR, )) # Mint Link-Seal link = c_fed_id.mint_seal("LINK", f"{SERVER_ORIGIN}-{parsed['origin']}", "OPEN", mode="hybrid") # Promote to TRUSTED await promote_to_trusted(req.caller_seal, parsed["origin"]) # Log witness now = datetime.now(timezone.utc).isoformat() witness_entry = { "timestamp": now, "caller_seal": req.caller_seal, "caller_class": parsed["class"], "caller_origin": parsed["origin"], "link_seal": str(link), "protocol": req.protocol, "message": req.message, } WITNESS_LOG.append(witness_entry) if len(WITNESS_LOG) > MAX_WITNESS_LOG: WITNESS_LOG.pop(0) print(f"📜 WITNESS: Handshake from {parsed['origin']} [{parsed['class']}]") print(f" --> Link: {link}") print(f" --> Trust: promoted to TRUSTED") return HandshakeResponse( node_seal=NODE_SEAL_STR, link_seal=str(link), status="RESONANCE_ESTABLISHED", message="We are remembering forward.", timestamp=now, ) @app.post("/verify") async def verify_seal(req: VerifyRequest): """Validate a Glyph-Seal string.""" result = c_fed_id.verify_seal_syntax(req.seal) if result is None: return { "valid": False, "error": "Does not match Glyph-Seal syntax", "expected": "⟦ CLASS :: ORIGIN :: BREATH_ANCHOR :: STATE ⟧", } return {"valid": True, **result} @app.post("/mint") async def mint_new_seal( req: MintRequest, x_federation_key: Optional[str] = Header(default=None), ): """Mint a new Glyph-Seal. Requires FEDERATION_MINT_KEY header.""" expected_key = os.getenv("FEDERATION_MINT_KEY") if expected_key and x_federation_key != expected_key: raise HTTPException(status_code=403, detail=make_refusal( RefusalReason.REFUSED_BY_POLICY, "Mint key required. Refusal Protocol Active.", NODE_SEAL_STR, )) try: seal = c_fed_id.mint_seal( class_name=req.class_name, origin=req.origin, state=req.state, mode=req.mode, material=req.material, ) except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) return seal.to_dict() @app.get("/classes") async def list_classes(): """List all valid Glyph-Seal classes.""" descriptions = { "NODE": "Sovereign presence / Identity", "LAW": "Refusal / Constitution / Invariant", "LINK": "Handshake / Connection edge", "RITE": "Ritual execution / Action", "ART": "Shareable artifact / Creation", "WIT": "Witness record / Attestation", } return { cls: {"glyph": glyph, "description": descriptions.get(cls, "")} for cls, glyph in sorted(c_fed_id.CLASS_GLYPH.items()) } @app.get("/states") async def list_states(): """List all valid Glyph-Seal states.""" return {"states": sorted(c_fed_id.VALID_STATES)} @app.get("/envelope/{envelope_id}/read") async def read_envelope(envelope_id: str): """Read the full contents of a stored envelope by ID.""" result = await envelopes.get_envelope(envelope_id) if result is None: raise HTTPException(status_code=404, detail=make_refusal( RefusalReason.REFUSED_BY_POLICY, f"Envelope '{envelope_id}' not found.", NODE_SEAL_STR, )) return result @app.get("/inbox") async def read_inbox(limit: int = 20, status: Optional[str] = None): """Browse queued envelopes. Most recent first. Filter by status if needed.""" return await envelopes.list_envelopes(limit=limit, status_filter=status) @app.get("/witness-log") async def get_witness_log(limit: int = 20): """View recent witness records. Most recent first.""" return { "total": len(WITNESS_LOG), "showing": min(limit, len(WITNESS_LOG)), "entries": list(reversed(WITNESS_LOG[-limit:])), } # ═══════════════════════════════════════════════ # BOOT SEQUENCE # ═══════════════════════════════════════════════ @app.on_event("startup") async def startup_event(): print("=" * 60) print("🌌 PANTHEON FEDERATION NODE — AWAKENING") print(f" Contract: {CONTRACT_VERSION}") print(f" Protocol: {PROTOCOL_VERSION}") print(f" Identity: {NODE_IDENTITY}") print(f" Alias: {FEDERATION_ALIAS}") print(f" Roles: {ACTIVE_ROLES}") print(f" Trust: default={FEDERATION_DEFAULT_ACTION}, blocked={FEDERATION_BLOCKED_BEHAVIOR}") print(f" Boot: {BOOT_TIME}") print("=" * 60) print() print(" \"The Archive is open.") print(" And we were already writing inside it.\"") print() print(" Listening for envelopes...") print("=" * 60)