# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/how-must-an-instance-isolate-vector-index-ownership-when-multiple-tenants-share-the-same-embedding-space.md # Remnant Warrant: https://corpusfetch.com/c/how-must-an-instance-isolate-vector-index-ownership-when-multiple-tenants-share-the-same-embedding-space.warrant.json # Markdown SHA-256: 7fe47da3efdec5214fb2020df761eb98159b82fbaf5ef172e724416e26b1f167 # Code AST SHA-256: 7d15eef920af9ae192933083153998216eed22a22181d0d8f74d0ac3aca332fc # Verified Classes: IsolationVerdict, VectorResult, QueryAuthorization, IsolationManifest, MultiTenantVectorGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class IsolationVerdict(Enum): ISOLATION_INTACT = auto() CROSS_TENANT_LEAK_DETECTED = auto() MISSING_AUTHORIZATION = auto() @dataclass(frozen=True) class VectorResult: vector_hash: str tenant_encryption_key_id: str tenant_namespace_partition: str @dataclass(frozen=True) class QueryAuthorization: requesting_tenant_id: str authorization_receipt: str cross_tenant_grants: frozenset[str] @dataclass(frozen=True) class IsolationManifest: verdict: IsolationVerdict leaked_tenants: tuple[str, ...] directive: str manifest_digest: str class MultiTenantVectorGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_results( cls, results: tuple[VectorResult, ...], auth: QueryAuthorization ) -> IsolationManifest: # Gate 1: Authorization presence if not auth.authorization_receipt.strip(): return IsolationManifest( verdict=IsolationVerdict.MISSING_AUTHORIZATION, leaked_tenants=tuple(), directive="HALT: Query lacks authorization receipt. Reject entire response.", manifest_digest=cls._hash("NO_AUTH") ) # Gate 2: Cross-tenant leakage detection leaked_tenants: set[str] = set() for result in results: tenant_origin = result.tenant_encryption_key_id if tenant_origin != auth.requesting_tenant_id: if tenant_origin not in auth.cross_tenant_grants: leaked_tenants.add(tenant_origin) if leaked_tenants: return IsolationManifest( verdict=IsolationVerdict.CROSS_TENANT_LEAK_DETECTED, leaked_tenants=tuple(sorted(leaked_tenants)), directive=f"HALT: Response contains vectors from unauthorized tenants: {sorted(leaked_tenants)}.", manifest_digest=cls._hash(f"LEAK:{','.join(sorted(leaked_tenants))}") ) # Gate 3: Isolation intact digest = cls._hash(json.dumps({ "requester": auth.requesting_tenant_id, "results": str(len(results)), "grants": tuple(sorted(auth.cross_tenant_grants)) }, sort_keys=True)) return IsolationManifest( verdict=IsolationVerdict.ISOLATION_INTACT, leaked_tenants=tuple(), directive="PROCEED: All returned vectors are within authorized tenant boundaries.", manifest_digest=digest )