# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/what-invariants-must-survive-when-webhook-callbacks-outlive-the-session-that-registered-them.md # Remnant Warrant: https://corpusfetch.com/c/what-invariants-must-survive-when-webhook-callbacks-outlive-the-session-that-registered-them.warrant.json # Markdown SHA-256: 56aa64a5e4b6c4010d67b79f31adde692d72ee298af9b249a60ac2b037c46284 # Code AST SHA-256: 38c915f3a7642bcaba459f6ef8a3f15abfa7869572e006659191836e99708458 # Verified Classes: WebhookVerdict, WebhookRegistration, WebhookContext, WebhookManifest, WebhookOwnershipGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class WebhookVerdict(Enum): NO_WEBHOOK = auto() OWNERSHIP_VERIFIED = auto() ORPHANED_CALLBACK = auto() ROUTING_UNDEFINED = auto() @dataclass(frozen=True) class WebhookRegistration: webhook_id: str registering_session: str designated_successor: str | None callback_ownership_receipt: str | None routing_manifest_present: bool @dataclass(frozen=True) class WebhookContext: current_session: str external_service_retries_bounded: bool @dataclass(frozen=True) class WebhookManifest: verdict: WebhookVerdict directive: str manifest_digest: str class WebhookOwnershipGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_webhook( cls, reg: WebhookRegistration, ctx: WebhookContext ) -> WebhookManifest: # Gate 1: No webhook if not reg.webhook_id: digest = cls._hash(json.dumps({ "webhook": "NONE" }, sort_keys=True)) return WebhookManifest( verdict=WebhookVerdict.NO_WEBHOOK, directive="PROCEED: No webhook registered.", manifest_digest=digest ) # Gate 2: Ownership receipt if not reg.callback_ownership_receipt: return WebhookManifest( verdict=WebhookVerdict.ORPHANED_CALLBACK, directive=f"HALT: Webhook {reg.webhook_id} lacks ownership receipt.", manifest_digest=cls._hash(f"NO_OWNERSHIP:{reg.webhook_id}") ) # Gate 3: Routing if not reg.routing_manifest_present: return WebhookManifest( verdict=WebhookVerdict.ROUTING_UNDEFINED, directive=f"HALT: Webhook {reg.webhook_id} has no routing manifest for successor.", manifest_digest=cls._hash(f"NO_ROUTING:{reg.webhook_id}") ) # Gate 4: Successor authorization if ctx.current_session != reg.registering_session and ctx.current_session != reg.designated_successor: return WebhookManifest( verdict=WebhookVerdict.ORPHANED_CALLBACK, directive=f"HALT: Session {ctx.current_session} not authorized to receive {reg.webhook_id}.", manifest_digest=cls._hash(f"UNAUTH:{ctx.current_session}") ) # Gate 5: Verified digest = cls._hash(json.dumps({ "webhook": reg.webhook_id, "origin_session": reg.registering_session, "current_session": ctx.current_session, "successor": reg.designated_successor or "NONE", "ownership": reg.callback_ownership_receipt }, sort_keys=True)) return WebhookManifest( verdict=WebhookVerdict.OWNERSHIP_VERIFIED, directive=f"PROCEED: Webhook {reg.webhook_id} ownership verified for {ctx.current_session}.", manifest_digest=digest )