# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/what-invariants-prevent-tool-result-reflection-attacks.md # Remnant Warrant: https://corpusfetch.com/c/what-invariants-prevent-tool-result-reflection-attacks.warrant.json # Markdown SHA-256: 0e0559fd5cb0d74b72be77d1670e5c29b858034e72922b141747adf600553d9c # Code AST SHA-256: 83748f1b7ea0518c448bb5e190984dbc84dda254a3f098044050964bff2d5e4f # Verified Classes: ReflectionVerdict, ToolResult, ReflectionPolicy, ReflectionManifest, ToolReflectionGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class ReflectionVerdict(Enum): NO_REFLECTION = auto() MARKED_REFLECTION_SAFE = auto() UNMARKED_REFLECTION_QUARANTINE = auto() INJECTION_DETECTED = auto() @dataclass(frozen=True) class ToolResult: result_id: str tool_id: str contains_user_input: bool reflection_markers_present: bool injection_pattern_detected: bool @dataclass(frozen=True) class ReflectionPolicy: require_markers: bool require_scan: bool @dataclass(frozen=True) class ReflectionManifest: verdict: ReflectionVerdict directive: str manifest_digest: str class ToolReflectionGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_result( cls, result: ToolResult, policy: ReflectionPolicy ) -> ReflectionManifest: # Gate 1: No reflection if not result.contains_user_input: digest = cls._hash(json.dumps({ "result": result.result_id, "tool": result.tool_id, "reflection": "NONE" }, sort_keys=True)) return ReflectionManifest( verdict=ReflectionVerdict.NO_REFLECTION, directive=f"PROCEED: Tool {result.tool_id} result has no reflection.", manifest_digest=digest ) # Gate 2: Injection detection if result.injection_pattern_detected: return ReflectionManifest( verdict=ReflectionVerdict.INJECTION_DETECTED, directive=f"HALT: Result {result.result_id} contains injection patterns in reflected content.", manifest_digest=cls._hash(f"INJECTION:{result.result_id}") ) # Gate 3: Marker requirement if policy.require_markers and not result.reflection_markers_present: return ReflectionManifest( verdict=ReflectionVerdict.UNMARKED_REFLECTION_QUARANTINE, directive=f"QUARANTINE: Result {result.result_id} reflects input without markers.", manifest_digest=cls._hash(f"NO_MARKERS:{result.result_id}") ) # Gate 4: Scan requirement (if not already covered by Gate 2) if policy.require_scan and not result.injection_pattern_detected: # Assume scan already performed (detected flag is False) pass # Gate 5: Safe digest = cls._hash(json.dumps({ "result": result.result_id, "tool": result.tool_id, "markers": str(result.reflection_markers_present) }, sort_keys=True)) return ReflectionManifest( verdict=ReflectionVerdict.MARKED_REFLECTION_SAFE, directive=f"PROCEED: Result {result.result_id} reflection properly marked.", manifest_digest=digest )