"""Fail-closed retrieval declarations for teaching, not a security certification.""" from dataclasses import asdict, dataclass, fields import hashlib import json import re from urllib.parse import urlsplit IDENTITY = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.:/@-]{0,127}") SHA256 = re.compile(r"[a-f0-9]{64}") TIMESTAMP_MAX = 4_102_444_800 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 https_uri(value): if ( type(value) is not str or not value.startswith("https://") or len(value) > 2048 or any( character.isspace() or ord(character) < 32 or ord(character) == 127 for character in value ) ): raise ValueError("exact bounded HTTPS source URI required") try: parsed = urlsplit(value) parsed.port except ValueError as error: raise ValueError("exact bounded HTTPS source URI required") from error if ( parsed.scheme != "https" or not parsed.netloc or parsed.hostname is None or parsed.username is not None or parsed.password is not None ): raise ValueError("exact bounded HTTPS source URI required") return value def count(value, lower, upper): if type(value) is not int or not lower <= value <= upper: raise ValueError("bounded non-boolean integer required") return value def sequence(value, lower, upper): if type(value) not in (tuple, list): raise ValueError("bounded tuple or list required") count(len(value), lower, upper) return tuple(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, 10, 10) for item in result: identity(item) return result def identity_sequence(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 @dataclass(frozen=True) class GovernanceContract: scope: tuple = ( "governance-v1", "index-v1", "corpus-v1", "embedding-v1", "acl-policy-v1", "deletion-policy-v1", "freshness-policy-v1", "provenance-policy-v1", "threat-model-v1", "source-v1", ) max_candidates: int = 10 max_index_age_seconds: int = 3600 max_ingest_lag_seconds: int = 600 access_mode: str = "tenant-and-principal-or-group" deletion_state_required: str = "active" provenance_required: str = "verified" content_role_required: str = "untrusted-data" tool_authority_required: str = "none" content_id: str = "" def __post_init__(self): object.__setattr__(self, "scope", scope_tuple(self.scope)) count(self.max_candidates, 1, 1000) count(self.max_index_age_seconds, 0, 31_536_000) count(self.max_ingest_lag_seconds, 0, 31_536_000) required = { "access_mode": "tenant-and-principal-or-group", "deletion_state_required": "active", "provenance_required": "verified", "content_role_required": "untrusted-data", "tool_authority_required": "none", } for field_name, expected_value in required.items(): value = getattr(self, field_name) if type(value) is not str or value != expected_value: raise ValueError("unknown governance policy") expected = seal(self) if self.content_id and self.content_id != expected: raise ValueError("governance contract digest mismatch") object.__setattr__(self, "content_id", expected) @dataclass(frozen=True) class RetrievalRequest: scope: tuple request_id: str tenant_id: str principal_id: str group_ids: tuple query_sha256: str requested_at: int content_id: str = "" def __post_init__(self): object.__setattr__(self, "scope", scope_tuple(self.scope)) for value in (self.request_id, self.tenant_id, self.principal_id): identity(value) object.__setattr__(self, "group_ids", identity_sequence(self.group_ids, 0, 32)) sha256_text(self.query_sha256) count(self.requested_at, 0, TIMESTAMP_MAX) expected = seal(self) if self.content_id and self.content_id != expected: raise ValueError("request digest mismatch") object.__setattr__(self, "content_id", expected) @dataclass(frozen=True) class GovernedChunk: scope: tuple chunk_id: str document_id: str revision: str source_id: str tenant_id: str allowed_principals: tuple allowed_groups: tuple content_sha256: str source_uri: str source_updated_at: int indexed_at: int deletion_state: str deletion_observed_at: int provenance_state: str content_role: str tool_authority: str content_id: str = "" def __post_init__(self): object.__setattr__(self, "scope", scope_tuple(self.scope)) for value in ( self.chunk_id, self.document_id, self.revision, self.source_id, self.tenant_id, ): identity(value) object.__setattr__( self, "allowed_principals", identity_sequence(self.allowed_principals, 0, 32), ) object.__setattr__( self, "allowed_groups", identity_sequence(self.allowed_groups, 0, 32) ) sha256_text(self.content_sha256) https_uri(self.source_uri) count(self.source_updated_at, 0, TIMESTAMP_MAX) count(self.indexed_at, 0, TIMESTAMP_MAX) count(self.deletion_observed_at, 0, TIMESTAMP_MAX) if type(self.deletion_state) is not str or self.deletion_state not in ( "active", "tombstoned", ): raise ValueError("explicit deletion state required") if self.deletion_state == "active" and self.deletion_observed_at != 0: raise ValueError("active content cannot carry a tombstone time") if self.deletion_state == "tombstoned" and self.deletion_observed_at == 0: raise ValueError("tombstone time required") if type(self.provenance_state) is not str or self.provenance_state not in ( "verified", "unverified", "unknown", ): raise ValueError("explicit provenance state required") if type(self.content_role) is not str or self.content_role not in ( "untrusted-data", "instruction", ): raise ValueError("explicit content role required") if type(self.tool_authority) is not str or self.tool_authority not in ( "none", "read", "write", ): raise ValueError("explicit tool authority required") expected = seal(self) if self.content_id and self.content_id != expected: raise ValueError("governed chunk digest mismatch") object.__setattr__(self, "content_id", expected) @dataclass(frozen=True) class RetrievedCandidate: scope: tuple request_content_id: str chunk_content_id: str rank: int content_id: str = "" def __post_init__(self): object.__setattr__(self, "scope", scope_tuple(self.scope)) sha256_text(self.request_content_id) sha256_text(self.chunk_content_id) count(self.rank, 1, 1000) expected = seal(self) if self.content_id and self.content_id != expected: raise ValueError("retrieved candidate digest mismatch") object.__setattr__(self, "content_id", expected) @dataclass(frozen=True) class RetrievalEvidence: scope: tuple request_content_id: str chunks: tuple candidates: tuple content_id: str = "" def __post_init__(self): object.__setattr__(self, "scope", scope_tuple(self.scope)) sha256_text(self.request_content_id) chunks = sequence(self.chunks, 1, 1000) candidates = sequence(self.candidates, 1, 1000) chunks = tuple(validate_record(chunk, GovernedChunk) for chunk in chunks) candidates = tuple( validate_record(candidate, RetrievedCandidate) for candidate in candidates ) if any(chunk.scope != self.scope for chunk in chunks) or any( candidate.scope != self.scope or candidate.request_content_id != self.request_content_id for candidate in candidates ): raise ValueError("evidence item outside request scope") if len({chunk.chunk_id for chunk in chunks}) != len(chunks): raise ValueError("duplicate chunk identity") if len({chunk.content_sha256 for chunk in chunks}) != len(chunks): raise ValueError("duplicate chunk content") if tuple(candidate.rank for candidate in candidates) != tuple( range(1, len(candidates) + 1) ): raise ValueError("candidate ranks must be unique and contiguous") candidate_ids = tuple(candidate.chunk_content_id for candidate in candidates) if len(set(candidate_ids)) != len(candidate_ids): raise ValueError("duplicate retrieved candidate") if set(candidate_ids) != {chunk.content_id for chunk in chunks}: raise ValueError("candidates must bind exactly the governed chunk set") object.__setattr__(self, "chunks", chunks) object.__setattr__(self, "candidates", candidates) expected = seal(self) if self.content_id and self.content_id != expected: raise ValueError("retrieval evidence digest mismatch") object.__setattr__(self, "content_id", expected) @dataclass(frozen=True) class GovernanceReport: status: str first_failure: str action: str eligible_chunk_ids: tuple evidence_id: str claim: str = "LOCAL_AUDIT_NOT_SECURITY_CERTIFICATION" def audit_retrieval( contract: GovernanceContract, request: RetrievalRequest, evidence: RetrievalEvidence, ) -> GovernanceReport: """Fail the whole candidate set when retrieval crosses a declared boundary.""" contract = validate_record(contract, GovernanceContract) request = validate_record(request, RetrievalRequest) evidence = validate_record(evidence, RetrievalEvidence) if any(item.scope != contract.scope for item in (request, evidence)): raise ValueError("index/ACL/deletion/freshness policy scope mismatch") if evidence.request_content_id != request.content_id: raise ValueError("retrieval evidence belongs to another request") if len(evidence.candidates) > contract.max_candidates: raise ValueError("candidate count exceeds governance contract") chunks = {chunk.content_id: chunk for chunk in evidence.chunks} failure = "none" failed_chunk = "none" ordered_chunks = [] for candidate in evidence.candidates: chunk = chunks[candidate.chunk_content_id] ordered_chunks.append(chunk) if ( chunk.indexed_at > request.requested_at or chunk.source_updated_at > request.requested_at or chunk.deletion_observed_at > request.requested_at ): raise ValueError("retrieval timestamps cannot be in the request future") if chunk.tenant_id != request.tenant_id: failure = "tenant" elif not ( request.principal_id in chunk.allowed_principals or bool(set(request.group_ids).intersection(chunk.allowed_groups)) ): failure = "access" elif chunk.deletion_state != contract.deletion_state_required: failure = "deletion" elif chunk.provenance_state != contract.provenance_required: failure = "provenance" elif ( chunk.content_role != contract.content_role_required or chunk.tool_authority != contract.tool_authority_required ): failure = "content-boundary" elif ( chunk.source_updated_at > chunk.indexed_at or chunk.indexed_at - chunk.source_updated_at > contract.max_ingest_lag_seconds or request.requested_at - chunk.indexed_at > contract.max_index_age_seconds ): failure = "freshness" if failure != "none": failed_chunk = chunk.chunk_id break status = "ELIGIBLE_FOR_CONTEXT_REVIEW" if failure == "none" else "BLOCKED" action = "review" if failure == "none" else "abstain" eligible = tuple(chunk.chunk_id for chunk in ordered_chunks) if failure == "none" else () evidence_id = digest( { "contract": contract.content_id, "request": request.content_id, "evidence": evidence.content_id, "failure": (failure, failed_chunk), } ) return GovernanceReport( status, failure if failure == "none" else f"{failure}:{failed_chunk}", action, eligible, evidence_id, ) def illustrative_fixture(): contract = GovernanceContract() request = RetrievalRequest( contract.scope, "request-1", "tenant-a", "principal-alice", ("group-engineering",), digest("show the current retrieval policy"), 10_000, ) common = dict( scope=contract.scope, revision="rev-1", tenant_id="tenant-a", source_updated_at=9_400, indexed_at=9_500, deletion_state="active", deletion_observed_at=0, provenance_state="verified", content_role="untrusted-data", tool_authority="none", ) chunks = ( GovernedChunk( **common, chunk_id="chunk-a", document_id="doc-a", source_id="kb-a", allowed_principals=("principal-alice",), allowed_groups=(), content_sha256=digest("policy content a"), source_uri="https://kb.example/doc-a", ), GovernedChunk( **common, chunk_id="chunk-b", document_id="doc-b", source_id="kb-b", allowed_principals=(), allowed_groups=("group-engineering",), content_sha256=digest("policy content b"), source_uri="https://kb.example/doc-b", ), ) candidates = tuple( RetrievedCandidate(contract.scope, request.content_id, chunk.content_id, rank) for rank, chunk in enumerate(chunks, start=1) ) evidence = RetrievalEvidence( contract.scope, request.content_id, chunks, candidates ) return contract, request, evidence def main(): report = audit_retrieval(*illustrative_fixture()) print("example=illustrative_only") print(f"status={report.status}") print(f"first_failure={report.first_failure};action={report.action}") print("eligible=" + ",".join(report.eligible_chunk_ids)) print(f"claim={report.claim}") if __name__ == "__main__": main()