# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/what-revocation-signal-invariants-must-survive-when-the-revoker-is-partitioned-from-the-mesh.md # Remnant Warrant: https://corpusfetch.com/c/what-revocation-signal-invariants-must-survive-when-the-revoker-is-partitioned-from-the-mesh.warrant.json # Markdown SHA-256: edb51b9f79e9721aa873e1f7468f312f63c87eff60b1d88334e483fb3694f54e # Code AST SHA-256: 4e93d6462f33a76ef151dc65ac678ffe24c03c045a879248e561817e8d8d02d7 # Verified Classes: RevokerPartitionVerdict, RevokerState, RevokerPolicy, RevokerPartitionManifest, RevokerPartitionGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class RevokerPartitionVerdict(Enum): REVOKER_ACTIVE = auto() PARTITION_DETECTED_BLOCK = auto() PARTITION_DETECTED_ALLOW_RISK = auto() RECOVERY_IN_PROGRESS = auto() @dataclass(frozen=True) class RevokerState: revoker_id: str revoker_reachable: bool partition_broadcast_sent: bool absence_policy: str | None # "BLOCK" | "ALLOW_WITH_WARNING" | "ALTERNATIVE_VERIFICATION" risk_acceptance_receipt: str | None pending_revocations_count: int @dataclass(frozen=True) class RevokerPolicy: require_broadcast: bool @dataclass(frozen=True) class RevokerPartitionManifest: verdict: RevokerPartitionVerdict directive: str manifest_digest: str class RevokerPartitionGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_revoker( cls, state: RevokerState, policy: RevokerPolicy ) -> RevokerPartitionManifest: # Gate 1: Revoker reachable if state.revoker_reachable: digest = cls._hash(json.dumps({ "revoker": state.revoker_id, "status": "REACHABLE" }, sort_keys=True)) return RevokerPartitionManifest( verdict=RevokerPartitionVerdict.REVOKER_ACTIVE, directive=f"PROCEED: Revoker {state.revoker_id} reachable.", manifest_digest=digest ) # Gate 2: Broadcast requirement if policy.require_broadcast and not state.partition_broadcast_sent: return RevokerPartitionManifest( verdict=RevokerPartitionVerdict.PARTITION_DETECTED_BLOCK, directive=f"HALT: Revoker partitioned but mesh not informed. Broadcast required.", manifest_digest=cls._hash(f"NO_BROADCAST:{state.revoker_id}") ) # Gate 3: Policy application if state.absence_policy == "BLOCK": return RevokerPartitionManifest( verdict=RevokerPartitionVerdict.PARTITION_DETECTED_BLOCK, directive=f"HALT: Revoker partitioned, policy BLOCK. Operations suspended.", manifest_digest=cls._hash(f"BLOCK:{state.revoker_id}") ) if state.absence_policy == "ALLOW_WITH_WARNING": if not state.risk_acceptance_receipt: return RevokerPartitionManifest( verdict=RevokerPartitionVerdict.PARTITION_DETECTED_BLOCK, directive=f"HALT: Allow-with-warning policy requires risk acceptance receipt.", manifest_digest=cls._hash("NO_RISK_ACCEPT") ) digest = cls._hash(json.dumps({ "revoker": state.revoker_id, "policy": "ALLOW_WITH_WARNING", "risk_acceptance": state.risk_acceptance_receipt }, sort_keys=True)) return RevokerPartitionManifest( verdict=RevokerPartitionVerdict.PARTITION_DETECTED_ALLOW_RISK, directive=f"PROCEED_WITH_RISK: Revoker partitioned, mesh accepts risk.", manifest_digest=digest ) # Gate 4: Recovery in progress if state.pending_revocations_count > 0: return RevokerPartitionManifest( verdict=RevokerPartitionVerdict.RECOVERY_IN_PROGRESS, directive=f"ACTION: Revoker reachable again, {state.pending_revocations_count} revocations pending.", manifest_digest=cls._hash(f"RECOVERY:{state.pending_revocations_count}") ) # Gate 5: Default block return RevokerPartitionManifest( verdict=RevokerPartitionVerdict.PARTITION_DETECTED_BLOCK, directive=f"HALT: Revoker partitioned without policy. Default to block.", manifest_digest=cls._hash("DEFAULT_BLOCK") )