# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/what-invariants-prevent-authenticated-but-adversarial-peers-from-poisoning-discovery-results.md # Remnant Warrant: https://corpusfetch.com/c/what-invariants-prevent-authenticated-but-adversarial-peers-from-poisoning-discovery-results.warrant.json # Markdown SHA-256: abd160b33ef560c1046dd2f54aef747835e0924ac9feef8a16d2f2697ea08518 # Code AST SHA-256: b692f0852c4e5295b829f7abc040d2b9af3647cf3cee5a220005734d84ca7c0b # Verified Classes: DiscoveryTrustVerdict, DiscoveryResult, TrustPolicy, TrustManifest, AuthenticatedDiscoveryGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class DiscoveryTrustVerdict(Enum): TRUSTED_BOUNDARY = auto() CORROBORATED_RESULT = auto() SINGLE_PEER_UNVERIFIED = auto() SCOPE_VIOLATION = auto() @dataclass(frozen=True) class DiscoveryResult: result_id: str reporting_peer_id: str peer_is_authenticated: bool peer_trust_boundary: bool asserted_scope: frozenset[str] peer_authoritative_scope: frozenset[str] corroboration_count: int @dataclass(frozen=True) class TrustPolicy: min_corroboration: int @dataclass(frozen=True) class TrustManifest: verdict: DiscoveryTrustVerdict directive: str manifest_digest: str class AuthenticatedDiscoveryGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_result( cls, result: DiscoveryResult, policy: TrustPolicy ) -> TrustManifest: # Gate 1: Authentication if not result.peer_is_authenticated: return TrustManifest( verdict=DiscoveryTrustVerdict.SINGLE_PEER_UNVERIFIED, directive=f"HALT: Peer {result.reporting_peer_id} not authenticated.", manifest_digest=cls._hash(f"NO_AUTH:{result.result_id}") ) # Gate 2: Trust boundary if result.peer_trust_boundary: digest = cls._hash(json.dumps({ "result": result.result_id, "peer": result.reporting_peer_id, "type": "TRUST_BOUNDARY" }, sort_keys=True)) return TrustManifest( verdict=DiscoveryTrustVerdict.TRUSTED_BOUNDARY, directive=f"PROCEED: Peer {result.reporting_peer_id} inside trust boundary.", manifest_digest=digest ) # Gate 3: Scope verification scope_violation = not result.asserted_scope.issubset(result.peer_authoritative_scope) if scope_violation: return TrustManifest( verdict=DiscoveryTrustVerdict.SCOPE_VIOLATION, directive=f"HALT: Peer {result.reporting_peer_id} asserts outside its scope.", manifest_digest=cls._hash(f"SCOPE:{result.result_id}") ) # Gate 4: Corroboration if result.corroboration_count < policy.min_corroboration: return TrustManifest( verdict=DiscoveryTrustVerdict.SINGLE_PEER_UNVERIFIED, directive=f"WARN: Only {result.corroboration_count} corroborations < {policy.min_corroboration}.", manifest_digest=cls._hash(f"LOW_CORROB:{result.result_id}") ) # Gate 5: Corroborated digest = cls._hash(json.dumps({ "result": result.result_id, "peer": result.reporting_peer_id, "scope": tuple(sorted(result.asserted_scope)), "corroboration": str(result.corroboration_count) }, sort_keys=True)) return TrustManifest( verdict=DiscoveryTrustVerdict.CORROBORATED_RESULT, directive=f"PROCEED: {result.corroboration_count} corroborations, scope valid.", manifest_digest=digest )