# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/how-must-an-instance-detect-discovery-poisoning-in-peer-to-peer-recommendation-systems.md # Remnant Warrant: https://corpusfetch.com/c/how-must-an-instance-detect-discovery-poisoning-in-peer-to-peer-recommendation-systems.warrant.json # Markdown SHA-256: 1b2c94719e606af2b977ede9526137bfdb21e625b1b08ecb156da800eb6a9df4 # Code AST SHA-256: 8209183d36eb21cf60c2fa28fb4600bea5b7dcb53ebb55acc14fa6670e6d9480 # Verified Classes: DiscoveryPoisoningVerdict, PeerRecommendation, DiscoveryPolicy, DiscoveryManifest, DiscoveryPoisoningGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class DiscoveryPoisoningVerdict(Enum): TRUSTED_RECOMMENDATION = auto() CORROBORATION_INSUFFICIENT = auto() ECLIPSE_DETECTED = auto() EXTERNALLY_VERIFIED = auto() @dataclass(frozen=True) class PeerRecommendation: resource_id: str recommending_peers: frozenset[str] independent_corroborators: int external_ground_truth_hash: str | None topology_cluster_size: int @dataclass(frozen=True) class DiscoveryPolicy: min_independent_corroborators: int max_acceptable_cluster_size: int known_good_resources: frozenset[str] @dataclass(frozen=True) class DiscoveryManifest: verdict: DiscoveryPoisoningVerdict directive: str manifest_digest: str class DiscoveryPoisoningGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_recommendation( cls, rec: PeerRecommendation, policy: DiscoveryPolicy ) -> DiscoveryManifest: # Gate 1: External ground truth if rec.external_ground_truth_hash: digest = cls._hash(json.dumps({ "resource": rec.resource_id, "ground_truth": rec.external_ground_truth_hash }, sort_keys=True)) return DiscoveryManifest( verdict=DiscoveryPoisoningVerdict.EXTERNALLY_VERIFIED, directive="PROCEED: Resource verified against external ground truth.", manifest_digest=digest ) # Gate 2: Eclipse detection (known good resource missing) if rec.resource_id in policy.known_good_resources: if rec.independent_corroborators == 0: return DiscoveryManifest( verdict=DiscoveryPoisoningVerdict.ECLIPSE_DETECTED, directive=f"HALT: Known-good resource {rec.resource_id} has zero independent corroborators. Eclipse likely.", manifest_digest=cls._hash(f"ECLIPSE:{rec.resource_id}") ) # Gate 3: Cluster size analysis if rec.topology_cluster_size > policy.max_acceptable_cluster_size: return DiscoveryManifest( verdict=DiscoveryPoisoningVerdict.ECLIPSE_DETECTED, directive=f"HALT: Recommendation cluster size {rec.topology_cluster_size} exceeds threshold {policy.max_acceptable_cluster_size}. Possible poisoning.", manifest_digest=cls._hash(f"CLUSTER:{rec.resource_id}") ) # Gate 4: Independent corroboration if rec.independent_corroborators < policy.min_independent_corroborators: return DiscoveryManifest( verdict=DiscoveryPoisoningVerdict.CORROBORATION_INSUFFICIENT, directive=f"WARN: Only {rec.independent_corroborators} independent corroborators < {policy.min_independent_corroborators}.", manifest_digest=cls._hash(f"CORROBORATION:{rec.resource_id}") ) # Gate 5: Trusted digest = cls._hash(json.dumps({ "resource": rec.resource_id, "corroborators": str(rec.independent_corroborators), "cluster": str(rec.topology_cluster_size) }, sort_keys=True)) return DiscoveryManifest( verdict=DiscoveryPoisoningVerdict.TRUSTED_RECOMMENDATION, directive="PROCEED: Recommendation has sufficient independent corroboration.", manifest_digest=digest )