diff --git a/backend/app/database.py b/backend/app/database.py index 7669f17b..c09e6119 100644 --- a/backend/app/database.py +++ b/backend/app/database.py @@ -78,6 +78,21 @@ FOREIGN KEY(capture_id) REFERENCES captures(id) ON DELETE SET NULL ); +-- `id` is deliberately a single-column, GLOBALLY unique primary key -- do NOT "fix" this into a +-- composite PRIMARY KEY(user_id, id) the way `entities` below has. SQLite requires a foreign key's +-- parent columns to be a PRIMARY KEY or UNIQUE, and five single-column child FKs point at +-- memories(id): memory_entities, memory_topics, memory_relations (source_memory_id AND +-- target_memory_id), and memory_vec_map -- whose memory_id is itself independently declared UNIQUE. +-- Making this PK composite makes every one of them fail at runtime with +-- OperationalError: foreign key mismatch - "memory_entities" referencing "memories" +-- and the only escapes are rewriting all of those child tables to carry user_id, or keeping a +-- standalone UNIQUE index on memories.id -- which re-imposes the exact global uniqueness the +-- composite key was supposed to relax. (`entities` could go composite precisely because nothing +-- FKs to entities(id).) The invariant this schema actually needs is therefore "memory/task ids are +-- globally unique", enforced on write: see the cross-tenant collision guards in storage.py +-- (_save_memory, _save_task, and both loops of rebuild_index_from_vault), which re-salt a derived +-- id with user_id when it would otherwise collide with another tenant's row. Without them +-- INSERT OR REPLACE resolves purely on this key and silently clobbers the other tenant. CREATE TABLE IF NOT EXISTS memories ( id TEXT PRIMARY KEY, capture_id TEXT, @@ -124,6 +139,9 @@ PRIMARY KEY(user_id, id) ); +-- Single-column, globally unique PK for the same reason as `memories` above (task_entities and +-- task_topics both carry single-column FKs to tasks(id)); uniqueness is enforced on write by the +-- cross-tenant collision guards in storage.py. CREATE TABLE IF NOT EXISTS tasks ( id TEXT PRIMARY KEY, capture_id TEXT, diff --git a/backend/app/storage.py b/backend/app/storage.py index 93d1bb1e..015da830 100644 --- a/backend/app/storage.py +++ b/backend/app/storage.py @@ -21563,6 +21563,15 @@ def rebuild_index_from_vault(self, user_id: str) -> dict[str, Any]: ) if principal_row is not None and str(memory.get("superseded_by") or "").strip(): rebuilt_trust = round(rebuilt_trust * 0.5, 4) + # Cross-tenant collision guard (same as _save_memory): memories.id is the sole + # primary key, and in shared-vault "bucket" deployments a vault markdown file's + # frontmatter id is not trust-boundary-checked before rebuild. Re-salt with + # user_id only if the id already belongs to a different user, so an ordinary + # single-tenant rebuild round-trips ids unchanged. + memory_id = self._tenant_unique_id( + conn, "memories", user_id, str(memory.get("id") or "") + ) + memory["id"] = memory_id conn.execute( """ INSERT OR REPLACE INTO memories @@ -21638,6 +21647,10 @@ def rebuild_index_from_vault(self, user_id: str) -> dict[str, Any]: topics = task.get("topics", []) entity_ids = task.get("entity_ids", []) captured_at = task.get("captured_at") or timestamp + # Cross-tenant collision guard (same as _save_task); see the matching comment + # on the memories rebuild above. + task_id = self._tenant_unique_id(conn, "tasks", user_id, str(task.get("id") or "")) + task["id"] = task_id conn.execute( """ INSERT OR REPLACE INTO tasks @@ -28393,6 +28406,42 @@ def _job_health_item(self, job: dict[str, Any], *, now: datetime) -> dict[str, A "age_seconds": _age_seconds(job.get("updated_at"), now=now), } + _TENANT_UNIQUE_TABLES = {"memories": "mem_", "tasks": "task_"} + + def _tenant_unique_id( + self, conn, table: str, user_id: str, candidate: str, *, max_attempts: int = 8 + ) -> str: + """Return an id for `user_id` that no OTHER tenant already owns in `table`. + + `memories.id` / `tasks.id` are single-column primary keys (see the comment on the table + definitions in database.py for why they must stay globally unique), and INSERT OR REPLACE + resolves conflicts purely on that key — so writing an id another tenant owns silently + deletes their row. Re-salting once is NOT enough: the salted id can itself be owned by a + third tenant, which would clobber *them* instead (verified: tenant C holding + stable_id("mem_", "tenant-b:collide_mem") was erased when tenant B salted into it). So + chain deterministically until the id is free. + + Deterministic and idempotent: the chain depends only on user_id and DB state, and the + `user_id != ?` predicate ignores this user's OWN row, so re-saving the same content + re-derives the same id instead of accumulating duplicates. The first hop is unchanged from + the original single-salt form, so ids already written stay stable. + """ + prefix = self._TENANT_UNIQUE_TABLES[table] # KeyError = caller bug, never user input + for _ in range(max_attempts): + taken = conn.execute( + f"SELECT 1 FROM {table} WHERE id = ? AND user_id != ? LIMIT 1", + (candidate, user_id), + ).fetchone() + if taken is None: + return candidate + candidate = stable_id(prefix, f"{user_id}:{candidate}") + # Unreachable short of an engineered pile-up of chained hash collisions each owned by a + # different tenant. Fail loudly rather than fall through and clobber someone's data. + raise ValueError( + f"could not derive a tenant-unique {table} id for user {user_id!r} " + f"after {max_attempts} attempts" + ) + def _token_hash(self, token: str, salt: str) -> str: return hashlib.sha256(f"{salt}:{token}".encode("utf-8")).hexdigest() @@ -28437,13 +28486,35 @@ def _save_memory( memory_id = stable_id("mem_", f"{user_id}:{author_principal_id}:{base_memory_id}") if source_account and external_id: memory_id = stable_id("mem_", f"{user_id}:{capture_id}:{base_memory_id}") + # Cross-tenant collision guard: memories.id is the sole primary key (not composite with + # user_id), and on the default local/CLI/vault-import path (no principal, no + # source_account+external_id) memory_id is derived from content alone, or is whatever + # explicit id the caller supplied. Two different users can land on the identical id, and + # INSERT OR REPLACE resolves conflicts purely on the primary key — an unguarded write + # here would silently delete and replace the other tenant's row. Re-salt with user_id + # only in that collision case, so the ordinary (non-colliding) path keeps its id exactly + # as derived/supplied, preserving explicit-id passthrough for existing callers. + pre_collision_salt_memory_id = memory_id + memory_id = self._tenant_unique_id(conn, "memories", user_id, memory_id) # Anti-resurrection: the memory id is deterministic (derived from user+capture+content), so # re-processing a capture whose child memory the user explicitly FORGOT would recompute the # same id and re-insert it. If that memory was tombstoned, honor the forget — skip the # re-derivation entirely. The tombstone is per-(user,memory), so a genuinely new memory (new # capture or changed content -> different id) is never suppressed, and un-forgetting isn't a # feature. Returns None; the caller drops the record. - if self._is_tombstoned_in_conn(conn, user_id, "memory", memory_id): + # + # Checked against BOTH the pre- and post-cross-tenant-salt id: a memory tombstoned before + # any other tenant ever collided with its id was tombstoned under the plain derived id, but + # a re-derivation attempted *after* a different tenant's row has since taken that id salts + # to a different string above — checking only the post-salt id would miss the tombstone + # entirely and silently resurrect content the user explicitly forgot (tenant A forgets + # "m1", tenant B later saves its own unrelated memory under literal id "m1" in the same + # shared/bucket database, and tenant A's next resync of the same source re-derives "m1", + # salts away from B's row, and — without this second check — reinserts the forgotten memory). + if self._is_tombstoned_in_conn(conn, user_id, "memory", memory_id) or ( + memory_id != pre_collision_salt_memory_id + and self._is_tombstoned_in_conn(conn, user_id, "memory", pre_collision_salt_memory_id) + ): return None kind = record.get("kind", "observation") # #17 deterministic layer: honor a valid explicit/kind layer, else classify by content via @@ -32028,6 +32099,13 @@ def _task_filters( def _save_task(self, conn, capture_id: str, user_id: str, task: dict[str, Any], captured_at: str) -> dict[str, Any]: task_id = task["id"] + # Cross-tenant collision guard: tasks.id is the sole primary key (not composite with + # user_id), and the extractor derives it from content alone (or it's whatever explicit + # id the caller supplied), so two different users can land on the identical id. INSERT + # OR REPLACE resolves conflicts purely on the primary key, so an unguarded write would + # silently delete and replace the other tenant's row. Re-salt only in that collision + # case; see the matching guard in _save_memory for the same pattern. + task_id = self._tenant_unique_id(conn, "tasks", user_id, task_id) topics = task.get("topics", []) entity_ids = task.get("entity_ids", []) conn.execute( diff --git a/backend/app/tenant_forensics.py b/backend/app/tenant_forensics.py new file mode 100644 index 00000000..f42cf6b8 --- /dev/null +++ b/backend/app/tenant_forensics.py @@ -0,0 +1,270 @@ +"""Forensics for the cross-tenant id-collision bug. + +`memories.id` and `tasks.id` are single-column primary keys (see the comment on those table +definitions in database.py for why they must stay globally unique). Before the collision guards in +storage.py existed, an `INSERT OR REPLACE` from a second tenant that happened to derive the same id +silently DELETED the first tenant's row instead of raising. The guards stop that happening again, +but they cannot un-lose data already destroyed in a database that ran the old code. + +This module answers the operational question the guards do not: *was this deployment ever hit?* + +It works because a clobber only rewrote the `memories`/`tasks` row itself. Sibling rows carry their +own `user_id`, and the `memory_events` audit log is append-only and was never rewritten — so a row +whose `user_id` disagrees with the owning memory's is physical evidence that the memory changed +hands. Findings are strongest for `memory_events` (never rewritten by any write path) and weakest +where a later write may legitimately have re-pointed a row, which is why each check is reported +separately with its own confidence rather than summed into one number. + +Read-only: opens the database in SQLite read-only URI mode and never writes. + + python -m backend.app.tenant_forensics /path/to/index.sqlite + python -m backend.app.tenant_forensics /path/to/index.sqlite --json +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path +from typing import Any + +from .sqlite_runtime import sqlite3 + + +# (label, confidence, sql). Each query returns evidence rows where a sibling row's owner disagrees +# with the owner of the memory/task it belongs to. +_CHECKS: tuple[tuple[str, str, str], ...] = ( + ( + "memory_events", + "high", + # The audit log is append-only: no write path rewrites an existing row's user_id, so a + # disagreement here is the most direct evidence that a memory changed hands. + """ + SELECT e.user_id AS other_user_id, m.user_id AS current_owner, e.object_id AS object_id, + COUNT(*) AS rows + FROM memory_events e + JOIN memories m ON m.id = e.object_id + WHERE e.object_type = 'memory' AND e.user_id != m.user_id + GROUP BY e.user_id, m.user_id, e.object_id + """, + ), + ( + "memory_vec_map", + "high", + """ + SELECT v.user_id AS other_user_id, m.user_id AS current_owner, v.memory_id AS object_id, + COUNT(*) AS rows + FROM memory_vec_map v + JOIN memories m ON m.id = v.memory_id + WHERE v.user_id != m.user_id + GROUP BY v.user_id, m.user_id, v.memory_id + """, + ), + ( + "memory_relations", + "high", + """ + SELECT r.user_id AS other_user_id, m.user_id AS current_owner, r.source_memory_id AS object_id, + COUNT(*) AS rows + FROM memory_relations r + JOIN memories m ON m.id = r.source_memory_id + WHERE r.user_id != m.user_id + GROUP BY r.user_id, m.user_id, r.source_memory_id + """, + ), + ( + "memory_entities", + "medium", + # The live save path DELETEs these by memory_id before reinserting, so a same-capture + # clobber erases the evidence; the vault-rebuild path does not, and leaves it behind. + """ + SELECT me.user_id AS other_user_id, m.user_id AS current_owner, me.memory_id AS object_id, + COUNT(*) AS rows + FROM memory_entities me + JOIN memories m ON m.id = me.memory_id + WHERE me.user_id != m.user_id + GROUP BY me.user_id, m.user_id, me.memory_id + """, + ), + ( + "memory_topics", + "medium", + """ + SELECT mt.user_id AS other_user_id, m.user_id AS current_owner, mt.memory_id AS object_id, + COUNT(*) AS rows + FROM memory_topics mt + JOIN memories m ON m.id = mt.memory_id + WHERE mt.user_id != m.user_id + GROUP BY mt.user_id, m.user_id, mt.memory_id + """, + ), + ( + "memories_vs_capture", + "medium", + # A memory whose parent capture belongs to someone else: the row was replaced without its + # capture_id being re-pointed, or the capture itself changed hands. + """ + SELECT c.user_id AS other_user_id, m.user_id AS current_owner, m.id AS object_id, + COUNT(*) AS rows + FROM memories m + JOIN captures c ON c.id = m.capture_id + WHERE m.capture_id IS NOT NULL AND c.user_id != m.user_id + GROUP BY c.user_id, m.user_id, m.id + """, + ), + ( + "task_entities", + "medium", + """ + SELECT te.user_id AS other_user_id, t.user_id AS current_owner, te.task_id AS object_id, + COUNT(*) AS rows + FROM task_entities te + JOIN tasks t ON t.id = te.task_id + WHERE te.user_id != t.user_id + GROUP BY te.user_id, t.user_id, te.task_id + """, + ), + ( + "task_topics", + "medium", + """ + SELECT tt.user_id AS other_user_id, t.user_id AS current_owner, tt.task_id AS object_id, + COUNT(*) AS rows + FROM task_topics tt + JOIN tasks t ON t.id = tt.task_id + WHERE tt.user_id != t.user_id + GROUP BY tt.user_id, t.user_id, tt.task_id + """, + ), +) + + +def _connect_readonly(db_path: Path) -> sqlite3.Connection: + conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True) + conn.row_factory = sqlite3.Row + return conn + + +def cross_tenant_integrity_report(db_path: str | Path, *, sample_limit: int = 20) -> dict[str, Any]: + """Scan a database for residue of a pre-guard cross-tenant clobber. Never writes.""" + db_path = Path(db_path) + if not db_path.exists(): + raise FileNotFoundError(f"no such database: {db_path}") + + findings: list[dict[str, Any]] = [] + skipped: list[dict[str, str]] = [] + conn = _connect_readonly(db_path) + try: + # Counted across the audit log and captures too, NOT just surviving memories: a clobber + # that erased a tenant's only memory removes them from `memories` entirely, so counting + # that table alone would report the worst case as a harmless single-tenant database. + tenant_count = conn.execute( + """ + SELECT COUNT(*) FROM ( + SELECT user_id FROM memories + UNION SELECT user_id FROM captures + UNION SELECT user_id FROM memory_events + ) + """ + ).fetchone()[0] + for label, confidence, sql in _CHECKS: + try: + rows = conn.execute(sql).fetchall() + except sqlite3.OperationalError as exc: + # A table this deployment does not have (memory_vec_map needs sqlite-vec) or a + # pre-migration schema. Report it rather than silently scoring the DB as clean. + skipped.append({"check": label, "reason": str(exc)}) + continue + if not rows: + continue + findings.append( + { + "check": label, + "confidence": confidence, + "affected_objects": len(rows), + "evidence_rows": sum(int(r["rows"]) for r in rows), + "tenants_involved": sorted( + {r["other_user_id"] for r in rows} | {r["current_owner"] for r in rows} + ), + "samples": [ + { + "object_id": r["object_id"], + "current_owner": r["current_owner"], + "other_user_id": r["other_user_id"], + "rows": int(r["rows"]), + } + for r in rows[:sample_limit] + ], + } + ) + finally: + conn.close() + + high = [f for f in findings if f["confidence"] == "high"] + return { + "database": str(db_path), + "distinct_tenants_seen": tenant_count, + "multi_tenant": tenant_count > 1, + # Every check compares two DIFFERENT user_ids, so a finding is cross-tenant evidence by + # construction -- it is deliberately NOT gated on the surviving tenant count, because the + # most severe case (a tenant wiped out entirely) leaves the fewest survivors behind. + "clobber_evidence": bool(high), + "suspicious": bool(findings), + "findings": findings, + "skipped_checks": skipped, + } + + +def format_report(report: dict[str, Any]) -> str: + lines = [f"database: {report['database']}", f"tenants seen: {report['distinct_tenants_seen']}"] + if report["clobber_evidence"]: + lines.append("") + lines.append("!! CROSS-TENANT CLOBBER EVIDENCE FOUND -- data was likely destroyed.") + elif report["suspicious"]: + lines.append("") + lines.append("Lower-confidence ownership mismatches found; review below.") + else: + lines.append("") + lines.append("No ownership mismatches found.") + + for finding in report["findings"]: + lines.append("") + lines.append( + f"[{finding['confidence'].upper()}] {finding['check']}: " + f"{finding['affected_objects']} object(s), {finding['evidence_rows']} row(s)" + ) + lines.append(f" tenants: {', '.join(finding['tenants_involved'])}") + for sample in finding["samples"]: + lines.append( + f" {sample['object_id']} now owned by {sample['current_owner']}" + f" but {sample['rows']} row(s) belong to {sample['other_user_id']}" + ) + for skip in report["skipped_checks"]: + lines.append(f" (skipped {skip['check']}: {skip['reason']})") + return "\n".join(lines) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + prog="tenant_forensics", + description="Scan a Cortex SQLite database for evidence of a cross-tenant id clobber.", + ) + parser.add_argument("database", help="path to index.sqlite (opened read-only)") + parser.add_argument("--json", action="store_true", help="emit the raw report as JSON") + parser.add_argument("--sample-limit", type=int, default=20, help="samples per finding (default 20)") + args = parser.parse_args(argv) + + try: + report = cross_tenant_integrity_report(args.database, sample_limit=args.sample_limit) + except FileNotFoundError as exc: + print(str(exc), file=sys.stderr) + return 2 + + print(json.dumps(report, indent=2) if args.json else format_report(report)) + # 1 signals high-confidence evidence so this can gate a deploy check. + return 1 if report["clobber_evidence"] else 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/backend/tests/test_import_diff.py b/backend/tests/test_import_diff.py index 8e29bb11..d4f1de0f 100644 --- a/backend/tests/test_import_diff.py +++ b/backend/tests/test_import_diff.py @@ -305,19 +305,27 @@ def test_read_only_does_not_mutate_store(self) -> None: class _EndpointHarness(unittest.TestCase): """Shared seed for the endpoint tests on both servers.""" - def _seed(self, store: CortexStore, user_id: str) -> None: + def _seed(self, store: CortexStore, user_id: str, memory_id: str = "mem_python") -> None: + # `memory_id` is parameterized (not hardcoded) because ImportDiffFastAPIEndpointTests + # shares ONE process-wide `main_module.store` across all its test methods (see that + # class's setUpClass), and unittest runs methods alphabetically within a class — so two + # tests both seeding the literal id "mem_python" under two different user_ids would + # otherwise contend for the same primary-key row on a store that isn't per-test isolated. + # Before the cross-tenant id-collision guard existed the second write silently overwrote + # the first tenant's row (masking the bug); the guard now correctly refuses that overwrite + # and re-salts instead, so a caller seeding a second, distinct user must pass its own id. store.update_settings(user_id, {"review_new_captures": False, "allow_pending_in_context": True}) store.save_capture( user_id=user_id, content="Prefers Python for backend work and TypeScript on the frontend.", source="obsidian", - source_url="local-file://mem_python", - title="mem_python", + source_url=f"local-file://{memory_id}", + title=memory_id, extracted={ "_timestamp": "2026-01-01T00:00:00Z", "summary": "Prefers Python for backend work and TypeScript on the frontend.", "records": [ - {"id": "mem_python", "kind": "fact", "layer": "preference", + {"id": memory_id, "kind": "fact", "layer": "preference", "content": "Prefers Python for backend work and TypeScript on the frontend.", "confidence": "confirmed", "importance": 4, "occurred_at": "2026-01-01T00:00:00Z", "topics": [], "entity_ids": []} @@ -443,7 +451,10 @@ def test_import_diff_route(self) -> None: def test_import_diff_pre_parsed_facts(self) -> None: user_id = "import-diff-fastapi-user2" - self._seed(self.main_module.store, user_id) + # Distinct memory_id: this class's main_module.store is shared across every test method + # (see the _seed docstring) — reusing "mem_python" here would contend with + # test_import_diff_route's own seed under a different user_id. + self._seed(self.main_module.store, user_id, memory_id="mem_python_2") response = self.client.post( "/v1/import-diff", json={"facts": [{"text": "Prefers Python for backend work and TypeScript on the frontend.", "vendor": "claude"}]}, diff --git a/backend/tests/test_sharding.py b/backend/tests/test_sharding.py index 7171e166..fcb9c879 100644 --- a/backend/tests/test_sharding.py +++ b/backend/tests/test_sharding.py @@ -493,6 +493,201 @@ def test_source_accounts_and_sync_cursors_are_user_sharded(self) -> None: self.assertEqual(registry.list_sync_cursors("alice")[0]["cursor_value"], "alice-cursor") self.assertEqual(registry.list_sync_cursors("bob"), []) + def _shared_bucket_registry(self): + """Bucket mode with a single shard routes EVERY user into the same sqlite file — the + hosted deployment shape in which a cross-tenant id collision is actually reachable.""" + registry = StoreRegistry.from_settings(self.settings(mode="bucket", shard_count=1)) + self.assertEqual( + registry.assignment_for("tenant-a").db_path, + registry.assignment_for("tenant-b").db_path, + ) + return registry + + @staticmethod + def _colliding_extraction(content: str) -> dict: + """Two tenants submitting the SAME explicit record/task id — the collision under test.""" + return { + "_timestamp": "2026-06-29T12:00:00Z", + "summary": content, + "records": [ + { + "id": "collide_mem", + "kind": "observation", + "layer": "semantic", + "content": content, + "confidence": "confirmed", + "importance": 3, + "topics": [], + "entity_ids": [], + } + ], + "tasks": [ + {"id": "collide_task", "kind": "action", "content": f"{content} task", "topics": [], "entity_ids": []} + ], + "entities": [], + } + + def _save_colliding(self, registry, user_id: str, content: str) -> None: + registry.save_capture( + user_id=user_id, + content=f"{user_id} capture", + source="unit-test", + source_url=None, + title=user_id, + extracted=self._colliding_extraction(content), + auto_approve=True, + ) + + def _rows(self, registry, table: str) -> list[tuple]: + import sqlite3 + + conn = sqlite3.connect(registry.assignment_for("tenant-a").db_path) + try: + return conn.execute(f"SELECT id, user_id, content FROM {table} ORDER BY user_id").fetchall() + finally: + conn.close() + + def test_colliding_memory_and_task_ids_do_not_clobber_another_tenant(self) -> None: + """Regression: memories.id / tasks.id are the sole PRIMARY KEY, and on the ordinary + local/CLI path an id is content-derived or caller-supplied with no user salt. INSERT OR + REPLACE resolves conflicts purely on that key, so an unguarded write silently DELETES the + first tenant's row. Both tenants must survive, each still owning its own content.""" + registry = self._shared_bucket_registry() + self._save_colliding(registry, "tenant-a", "Tenant A private content") + self._save_colliding(registry, "tenant-b", "Tenant B private content") + + memories = self._rows(registry, "memories") + self.assertEqual(len(memories), 2, f"both tenants' memories must survive, got {memories}") + self.assertEqual([row[1] for row in memories], ["tenant-a", "tenant-b"]) + self.assertEqual(memories[0][2], "Tenant A private content") + self.assertEqual(memories[1][2], "Tenant B private content") + # The colliding id was re-salted for exactly one tenant, so the rows are distinct. + self.assertNotEqual(memories[0][0], memories[1][0]) + + tasks = self._rows(registry, "tasks") + self.assertEqual(len(tasks), 2, f"both tenants' tasks must survive, got {tasks}") + self.assertEqual([row[1] for row in tasks], ["tenant-a", "tenant-b"]) + self.assertNotEqual(tasks[0][0], tasks[1][0]) + + def test_colliding_ids_keep_each_tenants_retrieval_isolated(self) -> None: + """The clobber's real damage is retrieval: after a collision each tenant must still find + its own memory and must never surface the other tenant's.""" + registry = self._shared_bucket_registry() + self._save_colliding(registry, "tenant-a", "The aardvark decision is confidential") + self._save_colliding(registry, "tenant-b", "The bobcat decision is confidential") + + a_hits = " ".join(str(hit.get("content", "")) for hit in registry.search("tenant-a", "confidential decision")) + b_hits = " ".join(str(hit.get("content", "")) for hit in registry.search("tenant-b", "confidential decision")) + + self.assertIn("aardvark", a_hits, "tenant-a must still retrieve its own memory") + self.assertNotIn("bobcat", a_hits, "tenant-a must never retrieve tenant-b's memory") + self.assertIn("bobcat", b_hits, "tenant-b must still retrieve its own memory") + self.assertNotIn("aardvark", b_hits, "tenant-b must never retrieve tenant-a's memory") + + def test_cross_tenant_salt_does_not_resurrect_a_forgotten_memory(self) -> None: + """The collision salt must not defeat anti-resurrection. A memory is tombstoned under its + PLAIN derived id; if another tenant later takes that id, the re-derivation salts to a + different string, so checking only the post-salt id would miss the tombstone and silently + restore content the user explicitly forgot.""" + registry = self._shared_bucket_registry() + + self._save_colliding(registry, "tenant-a", "Tenant A forgotten content") + a_before = self._rows(registry, "memories") + self.assertEqual(len(a_before), 1) + self.assertEqual(a_before[0][0], "collide_mem") + + # Tenant A explicitly forgets it (tombstone is recorded under the plain id). + self.assertTrue(registry.delete_memory("tenant-a", "collide_mem")) + self.assertEqual(self._rows(registry, "memories"), []) + + # Tenant B now takes that literal id for its own unrelated memory. + self._save_colliding(registry, "tenant-b", "Tenant B unrelated content") + + # Tenant A resyncs the same source: re-derives "collide_mem", which now collides with B's + # row and salts away from it. The forget must still be honored. + self._save_colliding(registry, "tenant-a", "Tenant A forgotten content") + + rows = self._rows(registry, "memories") + owners = [row[1] for row in rows] + self.assertNotIn("tenant-a", owners, f"tenant-a's forgotten memory must not be resurrected, got {rows}") + self.assertEqual(owners, ["tenant-b"], "tenant-b's memory must be untouched") + + def test_salted_id_that_a_third_tenant_already_owns_does_not_clobber_them(self) -> None: + """Second-order collision: re-salting ONCE is not enough. The salted id can itself be owned + by a third tenant, and writing it would clobber *them* instead — the same bug one level + down. The derivation must chain until the id is genuinely free.""" + from backend.app.extractor import stable_id + + registry = self._shared_bucket_registry() + # The exact id tenant-b's first salt hop lands on. + first_hop = stable_id("mem_", "tenant-b:collide_mem") + + # Tenant C already owns it, before anyone collides. + registry.save_capture( + user_id="tenant-c", + content="tenant-c capture", + source="unit-test", + source_url=None, + title="tenant-c", + extracted={ + "_timestamp": "2026-06-29T12:00:00Z", + "summary": "Tenant C precious data", + "records": [ + {"id": first_hop, "kind": "observation", "layer": "semantic", + "content": "Tenant C precious data", "confidence": "confirmed", + "importance": 3, "topics": [], "entity_ids": []} + ], + "tasks": [], + "entities": [], + }, + auto_approve=True, + ) + self._save_colliding(registry, "tenant-a", "Tenant A content") + self._save_colliding(registry, "tenant-b", "Tenant B content") + + rows = self._rows(registry, "memories") + owners = {row[1] for row in rows} + self.assertEqual( + owners, {"tenant-a", "tenant-b", "tenant-c"}, + f"all three tenants must survive a chained collision, got {rows}", + ) + by_owner = {row[1]: row[2] for row in rows} + self.assertEqual(by_owner["tenant-c"], "Tenant C precious data", "tenant-c must not be clobbered") + self.assertEqual(len({row[0] for row in rows}), 3, "every row must have a distinct id") + + def test_many_tenants_colliding_on_one_id_all_survive(self) -> None: + """Property check over the whole chain rather than one hand-picked collision: N tenants all + submit the SAME id, so each new writer must chain past every earlier one. No tenant may be + lost, every tenant must read back its own content, and ids must stay distinct.""" + registry = self._shared_bucket_registry() + tenants = [f"tenant-{i:02d}" for i in range(12)] + for tenant in tenants: + self._save_colliding(registry, tenant, f"content owned by {tenant}") + + rows = self._rows(registry, "memories") + self.assertEqual(len(rows), len(tenants), f"expected one row per tenant, got {len(rows)}") + self.assertEqual({row[1] for row in rows}, set(tenants), "no tenant may be dropped") + self.assertEqual(len({row[0] for row in rows}), len(tenants), "ids must all be distinct") + for row_id, owner, content in rows: + self.assertEqual(content, f"content owned by {owner}", f"{owner} must own its own content") + + def test_repeated_saves_under_collision_stay_idempotent(self) -> None: + """The chained derivation must be deterministic: re-saving identical content must land on + the SAME id rather than chaining further and accumulating a duplicate row per save.""" + registry = self._shared_bucket_registry() + self._save_colliding(registry, "tenant-a", "Tenant A content") + self._save_colliding(registry, "tenant-b", "Tenant B content") + after_first = self._rows(registry, "memories") + + for _ in range(4): + self._save_colliding(registry, "tenant-b", "Tenant B content") + + after_repeats = self._rows(registry, "memories") + self.assertEqual( + [row[0] for row in after_repeats], [row[0] for row in after_first], + "repeated saves must reuse the same ids, not accumulate duplicates", + ) + if __name__ == "__main__": unittest.main() diff --git a/backend/tests/test_tenant_forensics.py b/backend/tests/test_tenant_forensics.py new file mode 100644 index 00000000..6244e539 --- /dev/null +++ b/backend/tests/test_tenant_forensics.py @@ -0,0 +1,156 @@ +from __future__ import annotations + +import sqlite3 +import tempfile +import unittest +from pathlib import Path + +from backend.app.database import init_db +from backend.app.storage import CortexStore +from backend.app.tenant_forensics import cross_tenant_integrity_report, format_report, main + + +def _extraction(memory_id: str, content: str) -> dict: + return { + "_timestamp": "2026-06-29T12:00:00Z", + "summary": content, + "records": [ + {"id": memory_id, "kind": "observation", "layer": "semantic", "content": content, + "confidence": "confirmed", "importance": 3, "topics": ["alpha"], "entity_ids": []} + ], + "tasks": [], + "entities": [], + } + + +class TenantForensicsTests(unittest.TestCase): + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.root = Path(self.tmp.name) + self.db = self.root / "index.sqlite" + init_db(self.db) + self.store = CortexStore(self.db, self.root / "vault") + self.store._vector_ready = lambda conn: False + + def tearDown(self) -> None: + self.tmp.cleanup() + + def _save(self, user_id: str, memory_id: str, content: str) -> None: + self.store.save_capture( + user_id=user_id, + content=f"{content} capture", + source="unit-test", + source_url=None, + title=user_id, + extracted=_extraction(memory_id, content), + auto_approve=True, + ) + + def _simulate_pre_guard_clobber(self) -> None: + """Reproduce exactly what the old code left behind: tenant-b's INSERT OR REPLACE takes over + tenant-a's `memories` row, while tenant-a's append-only audit rows survive pointing at it.""" + conn = sqlite3.connect(self.db) + try: + row = conn.execute( + "SELECT * FROM memories WHERE user_id = 'tenant-a' LIMIT 1" + ).fetchone() + assert row is not None + columns = [d[0] for d in conn.execute("SELECT * FROM memories LIMIT 0").description] + values = dict(zip(columns, row)) + values["user_id"] = "tenant-b" + values["content"] = "Tenant B content that overwrote A" + placeholders = ", ".join("?" for _ in columns) + conn.execute( + f"INSERT OR REPLACE INTO memories ({', '.join(columns)}) VALUES ({placeholders})", + [values[c] for c in columns], + ) + conn.commit() + finally: + conn.close() + + def test_clean_multi_tenant_database_reports_no_findings(self) -> None: + self._save("tenant-a", "mem_a", "Tenant A content") + self._save("tenant-b", "mem_b", "Tenant B content") + + report = cross_tenant_integrity_report(self.db) + + self.assertFalse(report["clobber_evidence"]) + self.assertFalse(report["suspicious"]) + self.assertEqual(report["findings"], []) + self.assertTrue(report["multi_tenant"]) + + def test_guard_handled_collision_is_not_flagged(self) -> None: + """Two tenants submitting the same id is now handled by the write-side guard; that is + normal operation and must not be reported as a clobber.""" + self._save("tenant-a", "shared_id", "Tenant A content") + self._save("tenant-b", "shared_id", "Tenant B content") + + report = cross_tenant_integrity_report(self.db) + + self.assertFalse(report["clobber_evidence"], f"guard-handled collision must be clean: {report['findings']}") + self.assertFalse(report["suspicious"]) + + def test_detects_a_pre_guard_clobber(self) -> None: + self._save("tenant-a", "shared_id", "Tenant A content that gets destroyed") + self._simulate_pre_guard_clobber() + + report = cross_tenant_integrity_report(self.db) + + self.assertTrue(report["clobber_evidence"], "the clobber must be detected") + checks = {finding["check"] for finding in report["findings"]} + self.assertIn("memory_events", checks, "the append-only audit log is the primary signal") + finding = next(f for f in report["findings"] if f["check"] == "memory_events") + self.assertEqual(finding["confidence"], "high") + self.assertEqual(sorted(finding["tenants_involved"]), ["tenant-a", "tenant-b"]) + + def test_total_erasure_of_a_tenant_is_still_detected(self) -> None: + """Regression: the worst case leaves the FEWEST survivors. When the victim's only memory is + taken over, `memories` holds a single tenant — gating detection on the surviving tenant + count would report a total wipe as a harmless single-tenant database.""" + self._save("tenant-a", "shared_id", "Tenant A sole memory") + self._simulate_pre_guard_clobber() + + conn = sqlite3.connect(self.db) + try: + survivors = conn.execute("SELECT COUNT(DISTINCT user_id) FROM memories").fetchone()[0] + finally: + conn.close() + self.assertEqual(survivors, 1, "precondition: the victim is gone from memories") + + report = cross_tenant_integrity_report(self.db) + self.assertTrue(report["clobber_evidence"], "a total wipe must still be flagged") + self.assertGreater(report["distinct_tenants_seen"], 1, "the victim is still visible in the audit log") + + def test_report_is_read_only(self) -> None: + self._save("tenant-a", "mem_a", "Tenant A content") + before = self.db.read_bytes() + + cross_tenant_integrity_report(self.db) + + self.assertEqual(self.db.read_bytes(), before, "forensics must never write to the database") + + def test_cli_exit_code_signals_evidence(self) -> None: + self._save("tenant-a", "shared_id", "Tenant A content") + self.assertEqual(main([str(self.db)]), 0, "clean database exits 0") + + self._simulate_pre_guard_clobber() + self.assertEqual(main([str(self.db)]), 1, "evidence exits 1 so it can gate a deploy check") + + def test_missing_database_is_reported_not_crashed(self) -> None: + self.assertEqual(main([str(self.root / "nope.sqlite")]), 2) + with self.assertRaises(FileNotFoundError): + cross_tenant_integrity_report(self.root / "nope.sqlite") + + def test_format_report_renders_findings(self) -> None: + self._save("tenant-a", "shared_id", "Tenant A content") + self._simulate_pre_guard_clobber() + + text = format_report(cross_tenant_integrity_report(self.db)) + + self.assertIn("CROSS-TENANT CLOBBER EVIDENCE FOUND", text) + self.assertIn("memory_events", text) + self.assertIn("tenant-a", text) + + +if __name__ == "__main__": + unittest.main()