# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/what-invariants-must-survive-when-idempotency-keys-collide-across-tenants.md # Remnant Warrant: https://corpusfetch.com/c/what-invariants-must-survive-when-idempotency-keys-collide-across-tenants.warrant.json # Markdown SHA-256: c6a81bafb902f95f7b3e6f2c664a551f85b38895de0241d1c4d24fcbdd25ae8e # Code AST SHA-256: a79c050cd62c9242154a22eb27126d710d85eafaf27df4b01f1c29372ec24c87 # Verified Classes: IdempotencyVerdict, IdempotencyKey, StoredKeyState, IdempotencyPolicy, IdempotencyManifest, CrossTenantIdempotencyGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class IdempotencyVerdict(Enum): NEW_OPERATION = auto() LEGITIMATE_DEDUPLICATION = auto() CROSS_TENANT_COLLISION = auto() CROSS_TENANT_AUTHORIZED = auto() @dataclass(frozen=True) class IdempotencyKey: key_hash: str tenant_id: str operation_context_hash: str @dataclass(frozen=True) class StoredKeyState: stored_key: IdempotencyKey stored_result_hash: str | None @dataclass(frozen=True) class IdempotencyPolicy: cross_tenant_authorizations: frozenset[str] @dataclass(frozen=True) class IdempotencyManifest: verdict: IdempotencyVerdict directive: str manifest_digest: str class CrossTenantIdempotencyGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_key( cls, request: IdempotencyKey, stored: StoredKeyState | None, policy: IdempotencyPolicy ) -> IdempotencyManifest: # Gate 1: New operation if stored is None: digest = cls._hash(json.dumps({ "key": request.key_hash, "tenant": request.tenant_id, "context": request.operation_context_hash }, sort_keys=True)) return IdempotencyManifest( verdict=IdempotencyVerdict.NEW_OPERATION, directive="PROCEED: New operation, no prior key.", manifest_digest=digest ) # Gate 2: Same tenant if stored.stored_key.tenant_id == request.tenant_id: if stored.stored_key.operation_context_hash == request.operation_context_hash: digest = cls._hash(json.dumps({ "key": request.key_hash, "tenant": request.tenant_id, "result": stored.stored_result_hash or "NONE" }, sort_keys=True)) return IdempotencyManifest( verdict=IdempotencyVerdict.LEGITIMATE_DEDUPLICATION, directive="PROCEED: Same tenant, same context. Return cached result.", manifest_digest=digest ) return IdempotencyManifest( verdict=IdempotencyVerdict.CROSS_TENANT_COLLISION, directive=f"HALT: Same tenant but different context. Key reuse rejected.", manifest_digest=cls._hash(f"CONTEXT_DIFF:{request.key_hash}") ) # Gate 3: Cross-tenant authorization_key = f"{stored.stored_key.tenant_id}->{request.tenant_id}" if authorization_key in policy.cross_tenant_authorizations: digest = cls._hash(json.dumps({ "key": request.key_hash, "from_tenant": stored.stored_key.tenant_id, "to_tenant": request.tenant_id, "authorized": authorization_key }, sort_keys=True)) return IdempotencyManifest( verdict=IdempotencyVerdict.CROSS_TENANT_AUTHORIZED, directive=f"PROCEED: Cross-tenant authorized ({authorization_key}).", manifest_digest=digest ) # Gate 4: Unauthorized collision return IdempotencyManifest( verdict=IdempotencyVerdict.CROSS_TENANT_COLLISION, directive=f"HALT: Key collision across tenants {stored.stored_key.tenant_id} and {request.tenant_id}.", manifest_digest=cls._hash(f"COLLISION:{request.key_hash}") )