# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/how-must-an-instance-detect-transitive-dependency-poisoning-through-shared-upstream-packages.md # Remnant Warrant: https://corpusfetch.com/c/how-must-an-instance-detect-transitive-dependency-poisoning-through-shared-upstream-packages.warrant.json # Markdown SHA-256: 553a1786b2db27f7bb7df4ebcb9c5fb48ac65913693143635475e9a360c845d7 # Code AST SHA-256: 80908c059fa94ac350b955d1c0ef0aced4db78c5271f6f7b909c60b6bc1f5094 # Verified Classes: TransitiveVerdict, DependencyPackage, DependencyClosure, ClosurePolicy, TransitiveManifest, TransitiveDependencyGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class TransitiveVerdict(Enum): NO_DEPENDENCIES = auto() TRANSITIVE_VERIFIED = auto() SHARED_UPSTREAM_UNVERIFIED = auto() POISONING_SUSPECTED = auto() @dataclass(frozen=True) class DependencyPackage: package_id: str version: str integrity_verified: bool is_direct: bool @dataclass(frozen=True) class DependencyClosure: closure_id: str packages: tuple[DependencyPackage, ...] closure_complete: bool shared_with_other_agents: frozenset[str] @dataclass(frozen=True) class ClosurePolicy: require_integrity_for_all_transitive: bool @dataclass(frozen=True) class TransitiveManifest: verdict: TransitiveVerdict directive: str manifest_digest: str class TransitiveDependencyGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_closure( cls, closure: DependencyClosure, policy: ClosurePolicy ) -> TransitiveManifest: # Gate 1: No dependencies if not closure.packages: digest = cls._hash(json.dumps({ "closure": closure.closure_id, "packages": "NONE" }, sort_keys=True)) return TransitiveManifest( verdict=TransitiveVerdict.NO_DEPENDENCIES, directive=f"PROCEED: No dependencies for {closure.closure_id}.", manifest_digest=digest ) # Gate 2: Closure completeness if not closure.closure_complete: return TransitiveManifest( verdict=TransitiveVerdict.SHARED_UPSTREAM_UNVERIFIED, directive=f"HALT: Closure {closure.closure_id} incomplete, transitive exposure unknown.", manifest_digest=cls._hash(f"INCOMPLETE:{closure.closure_id}") ) # Gate 3: Shared upstream with unverified integrity transitive_unverified = [ p.package_id for p in closure.packages if not p.is_direct and not p.integrity_verified ] if transitive_unverified and policy.require_integrity_for_all_transitive: return TransitiveManifest( verdict=TransitiveVerdict.POISONING_SUSPECTED, directive=f"HALT: {len(transitive_unverified)} transitive packages unverified.", manifest_digest=cls._hash(f"UNVERIFIED:{closure.closure_id}") ) # Gate 4: Shared exposure if closure.shared_with_other_agents and transitive_unverified: return TransitiveManifest( verdict=TransitiveVerdict.SHARED_UPSTREAM_UNVERIFIED, directive=f"FLAG: Shared upstream exposure to {len(closure.shared_with_other_agents)} agents with unverified transitive packages.", manifest_digest=cls._hash(f"SHARED:{closure.closure_id}") ) # Gate 5: Verified digest = cls._hash(json.dumps({ "closure": closure.closure_id, "packages": tuple( {"id": p.package_id, "version": p.version, "verified": p.integrity_verified} for p in closure.packages ), "shared": tuple(sorted(closure.shared_with_other_agents)) }, sort_keys=True)) return TransitiveManifest( verdict=TransitiveVerdict.TRANSITIVE_VERIFIED, directive=f"PROCEED: {len(closure.packages)} packages verified.", manifest_digest=digest )