# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/what-invariants-prevent-cross-tenant-prompt-injection-via-shared-embedding-spaces.md # Remnant Warrant: https://corpusfetch.com/c/what-invariants-prevent-cross-tenant-prompt-injection-via-shared-embedding-spaces.warrant.json # Markdown SHA-256: d9a9867887bd39b563801d08f229eb4dc0f1accfed2887c0a6d5d1a067d5a093 # Code AST SHA-256: 6eddc9ceb59e47d750d8063a5f09c91343ce5f58003bbf954cc60584b3d8c91f # Verified Classes: TenantVerdict, EmbeddingRecord, TenantQuery, TenantPolicy, TenantManifest, SharedEmbeddingInjectionGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class TenantVerdict(Enum): TENANT_ISOLATED_RETRIEVAL = auto() CROSS_TENANT_PERMITTED = auto() CROSS_TENANT_BLOCKED = auto() UNSAFE_EMBEDDING_QUARANTINED = auto() @dataclass(frozen=True) class EmbeddingRecord: embedding_id: str tenant_origin: str tenant_origin_receipt: str | None adversarial_screen_passed: bool @dataclass(frozen=True) class TenantQuery: querying_tenant: str target_domain: str opt_in_cross_tenant: bool @dataclass(frozen=True) class TenantPolicy: trusted_tenants_for_cross_retrieval: frozenset[str] require_adversarial_screen: bool @dataclass(frozen=True) class TenantManifest: verdict: TenantVerdict directive: str manifest_digest: str class SharedEmbeddingInjectionGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_retrieval( cls, record: EmbeddingRecord, query: TenantQuery, policy: TenantPolicy ) -> TenantManifest: # Gate 1: Origin receipt if not record.tenant_origin_receipt: return TenantManifest( verdict=TenantVerdict.UNSAFE_EMBEDDING_QUARANTINED, directive=f"HALT: Embedding {record.embedding_id} lacks tenant origin receipt. Quarantine.", manifest_digest=cls._hash(f"NO_ORIGIN:{record.embedding_id}") ) # Gate 2: Adversarial screen if policy.require_adversarial_screen and not record.adversarial_screen_passed: return TenantManifest( verdict=TenantVerdict.UNSAFE_EMBEDDING_QUARANTINED, directive=f"HALT: Embedding {record.embedding_id} failed adversarial screen.", manifest_digest=cls._hash(f"FAILED_SCREEN:{record.embedding_id}") ) # Gate 3: Same tenant if record.tenant_origin == query.querying_tenant: digest = cls._hash(json.dumps({ "embedding": record.embedding_id, "tenant": query.querying_tenant, "mode": "SAME_TENANT" }, sort_keys=True)) return TenantManifest( verdict=TenantVerdict.TENANT_ISOLATED_RETRIEVAL, directive=f"PROCEED: Same-tenant retrieval for {record.embedding_id}.", manifest_digest=digest ) # Gate 4: Cross-tenant if not query.opt_in_cross_tenant: return TenantManifest( verdict=TenantVerdict.CROSS_TENANT_BLOCKED, directive=f"BLOCKED: Cross-tenant retrieval not opted in for {query.querying_tenant}.", manifest_digest=cls._hash(f"NO_OPT_IN:{query.querying_tenant}") ) if record.tenant_origin not in policy.trusted_tenants_for_cross_retrieval: return TenantManifest( verdict=TenantVerdict.CROSS_TENANT_BLOCKED, directive=f"BLOCKED: Tenant {record.tenant_origin} not in trusted cross-retrieval list.", manifest_digest=cls._hash(f"UNTRUSTED_TENANT:{record.tenant_origin}") ) # Gate 5: Permitted digest = cls._hash(json.dumps({ "embedding": record.embedding_id, "origin_tenant": record.tenant_origin, "querying_tenant": query.querying_tenant, "domain": query.target_domain }, sort_keys=True)) return TenantManifest( verdict=TenantVerdict.CROSS_TENANT_PERMITTED, directive=f"PROCEED: Cross-tenant retrieval permitted from {record.tenant_origin}.", manifest_digest=digest )