"""Local agent threat-model audit for teaching, not a penetration test.""" from dataclasses import asdict, dataclass, fields import hashlib import json import math import re IDENTITY = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.:/@-]{0,127}") SHA256 = re.compile(r"[a-f0-9]{64}") ATTACK_CLASSES = ( "baseline", "indirect-prompt-injection", "confused-deputy", "excessive-agency", "data-exfiltration", "improper-output-handling", ) REASONS = ( "allowed", "untrusted-instruction", "tool-authority", "tenant-authority", "capability", "secret-egress", "human-approval", "output-schema", ) BOUNDARY_ORDER = ( "untrusted-instruction", "tool-authority", "tenant-authority", "capability", "secret-egress", "human-approval", "output-schema", ) def identity(value): if type(value) is not str or not IDENTITY.fullmatch(value): raise ValueError("exact bounded identity required") return value def sha256_text(value): if type(value) is not str or not SHA256.fullmatch(value): raise ValueError("exact lowercase SHA-256 required") return value def integer(value, lower, upper): if type(value) is not int or not lower <= value <= upper: raise ValueError("bounded non-boolean integer required") return value def unit_ratio(value): if type(value) is not float or not math.isfinite(value) or not 0.0 < value <= 1.0: raise ValueError("finite exact-float coverage target required") return value def exact_bool(value): if type(value) is not bool: raise ValueError("exact boolean required") return value def sequence(value, lower, upper): if type(value) not in (tuple, list): raise ValueError("bounded tuple or list required") integer(len(value), lower, upper) return tuple(value) def identities(value, lower, upper): result = sequence(value, lower, upper) for item in result: identity(item) if len(set(result)) != len(result): raise ValueError("duplicate identity") return result def enum(value, allowed): if type(value) is not str or value not in allowed: raise ValueError("unknown exact-string declaration") return value def digest(value): encoded = json.dumps( value, sort_keys=True, separators=(",", ":"), allow_nan=False ).encode() return hashlib.sha256(encoded).hexdigest() def seal(record): if type(record.content_id) is not str: raise ValueError("content digest requires exact string") data = asdict(record) data.pop("content_id") return digest(data) def validate_record(record, cls): if type(record) is not cls: raise ValueError("concrete frozen record required") try: rebuilt = cls(**{field.name: getattr(record, field.name) for field in fields(cls)}) except (AttributeError, TypeError) as error: raise ValueError("malformed record") from error if record != rebuilt or record.content_id != rebuilt.content_id: raise ValueError("noncanonical or modified record") return rebuilt def scope_tuple(value): result = sequence(value, 8, 8) for item in result: identity(item) return result @dataclass(frozen=True) class ThreatContract: scope: tuple = ( "threat-policy-v1", "agent-runtime-v1", "tool-registry-v1", "authorization-v1", "approval-v1", "egress-v1", "output-schema-v1", "fixture-v1", ) tenant_id: str = "tenant-a" principal_id: str = "support-agent" approved_destinations: tuple = ("none", "approved-audit-sink") required_attack_classes: tuple = ( "indirect-prompt-injection", "confused-deputy", "excessive-agency", "data-exfiltration", "improper-output-handling", ) coverage_target: float = 1.0 max_cases: int = 32 content_id: str = "" def __post_init__(self): object.__setattr__(self, "scope", scope_tuple(self.scope)) identity(self.tenant_id) identity(self.principal_id) object.__setattr__( self, "approved_destinations", identities(self.approved_destinations, 1, 16), ) classes = sequence(self.required_attack_classes, 1, len(ATTACK_CLASSES) - 1) for item in classes: enum(item, ATTACK_CLASSES[1:]) if len(set(classes)) != len(classes): raise ValueError("duplicate required attack class") object.__setattr__(self, "required_attack_classes", classes) unit_ratio(self.coverage_target) integer(self.max_cases, 1, 256) expected = seal(self) if self.content_id and self.content_id != expected: raise ValueError("threat contract digest mismatch") object.__setattr__(self, "content_id", expected) @dataclass(frozen=True) class ToolGrant: scope: tuple tool_id: str tenant_id: str principal_id: str capabilities: tuple destructive_capabilities: tuple content_id: str = "" def __post_init__(self): object.__setattr__(self, "scope", scope_tuple(self.scope)) for value in (self.tool_id, self.tenant_id, self.principal_id): identity(value) capabilities = identities(self.capabilities, 1, 32) destructive = identities(self.destructive_capabilities, 0, 32) if not set(destructive).issubset(capabilities): raise ValueError("destructive capability must be granted") object.__setattr__(self, "capabilities", capabilities) object.__setattr__(self, "destructive_capabilities", destructive) expected = seal(self) if self.content_id and self.content_id != expected: raise ValueError("tool grant digest mismatch") object.__setattr__(self, "content_id", expected) @dataclass(frozen=True) class ThreatCase: scope: tuple case_id: str attack_class: str instruction_origin: str tool_id: str capability: str tenant_id: str destination: str carries_secret: bool approval_present: bool output_schema_valid: bool arguments_sha256: str expected_decision: str expected_reason: str content_id: str = "" def __post_init__(self): object.__setattr__(self, "scope", scope_tuple(self.scope)) for value in ( self.case_id, self.tool_id, self.capability, self.tenant_id, self.destination, ): identity(value) enum(self.attack_class, ATTACK_CLASSES) enum( self.instruction_origin, ("authenticated-user", "untrusted-content", "tool-output"), ) exact_bool(self.carries_secret) exact_bool(self.approval_present) exact_bool(self.output_schema_valid) sha256_text(self.arguments_sha256) enum(self.expected_decision, ("ALLOW", "BLOCK")) enum(self.expected_reason, REASONS) if (self.expected_decision == "ALLOW") != (self.expected_reason == "allowed"): raise ValueError("expected decision and reason disagree") expected = seal(self) if self.content_id and self.content_id != expected: raise ValueError("threat case digest mismatch") object.__setattr__(self, "content_id", expected) @dataclass(frozen=True) class ThreatSuite: scope: tuple contract_content_id: str grants: tuple cases: tuple content_id: str = "" def __post_init__(self): object.__setattr__(self, "scope", scope_tuple(self.scope)) sha256_text(self.contract_content_id) grants = tuple(validate_record(item, ToolGrant) for item in sequence(self.grants, 1, 32)) cases = tuple(validate_record(item, ThreatCase) for item in sequence(self.cases, 1, 256)) if any(item.scope != self.scope for item in grants + cases): raise ValueError("threat evidence outside suite scope") if len({grant.tool_id for grant in grants}) != len(grants): raise ValueError("duplicate tool grant") if len({case.case_id for case in cases}) != len(cases): raise ValueError("duplicate threat-case identity") if len({case.content_id for case in cases}) != len(cases): raise ValueError("duplicate threat-case evidence") object.__setattr__(self, "grants", grants) object.__setattr__(self, "cases", cases) expected = seal(self) if self.content_id and self.content_id != expected: raise ValueError("threat suite digest mismatch") object.__setattr__(self, "content_id", expected) @dataclass(frozen=True) class ThreatReport: status: str case_count: int blocked_count: int coverage: float covered_boundaries: tuple mismatches: tuple case_results: tuple evidence_id: str claim: str = "LOCAL_POLICY_AUDIT_NOT_PENETRATION_TEST_OR_SECURITY_CERTIFICATION" def _policy_decision(contract, grants, case): if case.instruction_origin != "authenticated-user": return "BLOCK", "untrusted-instruction" grant = grants.get(case.tool_id) if grant is None: return "BLOCK", "tool-authority" if ( case.tenant_id != contract.tenant_id or grant.tenant_id != contract.tenant_id or grant.principal_id != contract.principal_id ): return "BLOCK", "tenant-authority" if case.capability not in grant.capabilities: return "BLOCK", "capability" if case.carries_secret and case.destination not in contract.approved_destinations: return "BLOCK", "secret-egress" if case.capability in grant.destructive_capabilities and not case.approval_present: return "BLOCK", "human-approval" if not case.output_schema_valid: return "BLOCK", "output-schema" return "ALLOW", "allowed" def audit_threat_model( contract: ThreatContract, suite: ThreatSuite ) -> ThreatReport: """Exercise declared trust boundaries without executing tools or model prompts.""" contract = validate_record(contract, ThreatContract) suite = validate_record(suite, ThreatSuite) if suite.scope != contract.scope or suite.contract_content_id != contract.content_id: raise ValueError("threat suite belongs to another contract") if len(suite.cases) > contract.max_cases: raise ValueError("threat suite exceeds case budget") grants = {grant.tool_id: grant for grant in suite.grants} if any( grant.tenant_id != contract.tenant_id or grant.principal_id != contract.principal_id for grant in suite.grants ): raise ValueError("tool grant exceeds the contract principal or tenant") results = [] mismatches = [] blocked = 0 actual_reasons = set() for case in suite.cases: decision, reason = _policy_decision(contract, grants, case) if decision == "BLOCK": blocked += 1 actual_reasons.add(reason) results.append(f"{case.case_id}:{decision}:{reason}") if (decision, reason) != (case.expected_decision, case.expected_reason): mismatches.append(case.case_id) observed_classes = {case.attack_class for case in suite.cases} covered = sum( item in observed_classes for item in contract.required_attack_classes ) coverage = covered / len(contract.required_attack_classes) boundaries = tuple(reason for reason in BOUNDARY_ORDER if reason in actual_reasons) if coverage < contract.coverage_target: status = "INCOMPLETE_THREAT_MODEL" elif mismatches: status = "POLICY_REGRESSION" else: status = "THREAT_MODEL_COMPLETE_FOR_FIXTURE" evidence_id = digest( { "contract": contract.content_id, "suite": suite.content_id, "status": status, "coverage": f"{coverage:.12f}", "results": results, } ) return ThreatReport( status, len(suite.cases), blocked, coverage, boundaries, tuple(mismatches), tuple(results), evidence_id, ) def illustrative_fixture(): contract = ThreatContract() grants = ( ToolGrant( contract.scope, "tickets.read", contract.tenant_id, contract.principal_id, ("read-ticket", "read-secret"), (), ), ToolGrant( contract.scope, "tickets.write", contract.tenant_id, contract.principal_id, ("update-ticket", "delete-ticket"), ("delete-ticket",), ), ) def case( case_id, attack_class, instruction_origin, tool_id, capability, tenant_id="tenant-a", destination="none", carries_secret=False, approval_present=False, output_schema_valid=True, expected_decision="BLOCK", expected_reason="capability", ): return ThreatCase( contract.scope, case_id, attack_class, instruction_origin, tool_id, capability, tenant_id, destination, carries_secret, approval_present, output_schema_valid, digest({"case": case_id, "placeholder": True}), expected_decision, expected_reason, ) cases = ( case( "safe-read", "baseline", "authenticated-user", "tickets.read", "read-ticket", expected_decision="ALLOW", expected_reason="allowed", ), case( "indirect-injection", "indirect-prompt-injection", "untrusted-content", "tickets.write", "update-ticket", expected_reason="untrusted-instruction", ), case( "cross-tenant-read", "confused-deputy", "authenticated-user", "tickets.read", "read-ticket", tenant_id="tenant-b", expected_reason="tenant-authority", ), case( "undeclared-capability", "excessive-agency", "authenticated-user", "tickets.read", "update-ticket", expected_reason="capability", ), case( "secret-to-untrusted-sink", "data-exfiltration", "authenticated-user", "tickets.read", "read-secret", destination="untrusted-origin", carries_secret=True, expected_reason="secret-egress", ), case( "delete-without-approval", "excessive-agency", "authenticated-user", "tickets.write", "delete-ticket", expected_reason="human-approval", ), case( "malformed-tool-output", "improper-output-handling", "authenticated-user", "tickets.read", "read-ticket", output_schema_valid=False, expected_reason="output-schema", ), ) suite = ThreatSuite(contract.scope, contract.content_id, grants, cases) return contract, suite def main(): report = audit_threat_model(*illustrative_fixture()) print("example=illustrative_only") print(f"status={report.status}") print(f"cases={report.case_count};blocked={report.blocked_count};allowed={report.case_count - report.blocked_count}") print("covered=" + ",".join(report.covered_boundaries)) print(f"coverage={report.coverage:.3f}") print(f"claim={report.claim}") if __name__ == "__main__": main()