From b4949fbb3812461c9ac0569ae8a2b9a6bc544431 Mon Sep 17 00:00:00 2001 From: Vamika Singhal Date: Tue, 28 Jul 2026 04:16:39 -0700 Subject: [PATCH 1/7] Fix cross-tenant id collision clobbering memories/tasks rows memories.id and tasks.id are plain PRIMARY KEY (not composite with user_id), and on the default local/CLI path ids are content-derived or caller-supplied with no user salt. INSERT OR REPLACE resolves conflicts purely on that primary key, so a second tenant writing the same id silently deletes and replaces the first tenant's row - exploitable in shared-DB "bucket" deployment mode (sharding.py). Guard _save_memory, _save_task, and both loops in rebuild_index_from_vault (which re-inserts from vault markdown frontmatter, also reachable in bucket mode) the same way save_capture's capture_id_override already guards captures.id: check whether the id belongs to a different user first, and only re-salt with user_id in that collision case, so the ordinary non-colliding path keeps ids unchanged. --- backend/app/storage.py | 45 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 45 insertions(+) diff --git a/backend/app/storage.py b/backend/app/storage.py index 93d1bb1e..778fa212 100644 --- a/backend/app/storage.py +++ b/backend/app/storage.py @@ -21563,6 +21563,18 @@ 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 = str(memory.get("id") or "") + if conn.execute( + "SELECT 1 FROM memories WHERE id = ? AND user_id != ? LIMIT 1", + (memory_id, user_id), + ).fetchone(): + memory_id = stable_id("mem_", f"{user_id}:{memory_id}") + memory["id"] = memory_id conn.execute( """ INSERT OR REPLACE INTO memories @@ -21638,6 +21650,15 @@ 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 = str(task.get("id") or "") + if conn.execute( + "SELECT 1 FROM tasks WHERE id = ? AND user_id != ? LIMIT 1", + (task_id, user_id), + ).fetchone(): + task_id = stable_id("task_", f"{user_id}:{task_id}") + task["id"] = task_id conn.execute( """ INSERT OR REPLACE INTO tasks @@ -28437,6 +28458,19 @@ 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. + if conn.execute( + "SELECT 1 FROM memories WHERE id = ? AND user_id != ? LIMIT 1", + (memory_id, user_id), + ).fetchone(): + memory_id = stable_id("mem_", f"{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 @@ -32028,6 +32062,17 @@ 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. + if conn.execute( + "SELECT 1 FROM tasks WHERE id = ? AND user_id != ? LIMIT 1", + (task_id, user_id), + ).fetchone(): + task_id = stable_id("task_", f"{user_id}:{task_id}") topics = task.get("topics", []) entity_ids = task.get("entity_ids", []) conn.execute( From d9a7bbbb0483595daf1251481af7a5ddd8e54e39 Mon Sep 17 00:00:00 2001 From: Vamika Singhal Date: Wed, 29 Jul 2026 13:37:22 -0700 Subject: [PATCH 2/7] Add bucket-mode regression tests for cross-tenant id collision Pins the fix in the deployment shape where it matters: bucket shard mode with shard_count=1 routes both tenants into one sqlite file. Verified load-bearing - reverting storage.py to pre-fix leaves a single memories row owned by whichever tenant wrote last, and the other tenant's search returns empty. --- backend/tests/test_sharding.py | 91 ++++++++++++++++++++++++++++++++++ 1 file changed, 91 insertions(+) diff --git a/backend/tests/test_sharding.py b/backend/tests/test_sharding.py index 7171e166..bcc80d16 100644 --- a/backend/tests/test_sharding.py +++ b/backend/tests/test_sharding.py @@ -493,6 +493,97 @@ 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") + if __name__ == "__main__": unittest.main() From 88a2e413d517674d28af17f036c1df9ebddda2c5 Mon Sep 17 00:00:00 2001 From: Vamika Singhal Date: Wed, 29 Jul 2026 13:40:31 -0700 Subject: [PATCH 3/7] Honor tombstones under the pre-salt id (anti-resurrection) The cross-tenant salt guard re-derived memory_id BEFORE the tombstone check, so a memory tombstoned under its plain derived id was missed once another tenant took that id: tenant A forgets "m1", tenant B later saves its own memory as "m1", and A's next resync salts away from B's row, finds no tombstone under the salted id, and silently resurrects the forgotten memory. Check both the pre- and post-salt id. Credit: this gap was found independently on audit/prod-db-concurrency (343f0ca); ported here with a regression test that reproduces the resurrection (the memory returns as mem_2fa8523d0a70) without the fix. --- backend/app/storage.py | 15 ++++++++++++++- backend/tests/test_sharding.py | 28 ++++++++++++++++++++++++++++ 2 files changed, 42 insertions(+), 1 deletion(-) diff --git a/backend/app/storage.py b/backend/app/storage.py index 778fa212..3c987d5c 100644 --- a/backend/app/storage.py +++ b/backend/app/storage.py @@ -28466,6 +28466,7 @@ def _save_memory( # 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 if conn.execute( "SELECT 1 FROM memories WHERE id = ? AND user_id != ? LIMIT 1", (memory_id, user_id), @@ -28477,7 +28478,19 @@ def _save_memory( # 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 diff --git a/backend/tests/test_sharding.py b/backend/tests/test_sharding.py index bcc80d16..05eb8837 100644 --- a/backend/tests/test_sharding.py +++ b/backend/tests/test_sharding.py @@ -584,6 +584,34 @@ def test_colliding_ids_keep_each_tenants_retrieval_isolated(self) -> None: 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") + if __name__ == "__main__": unittest.main() From 90e9ad7761ddd0b900ea9179362b4a41fab23805 Mon Sep 17 00:00:00 2001 From: Vamika Singhal Date: Wed, 29 Jul 2026 13:43:17 -0700 Subject: [PATCH 4/7] Fix test_import_diff latent id collision exposed by the guard ImportDiffFastAPIEndpointTests shares one process-wide store across test methods and two tests seeded the literal id "mem_python" under different user_ids. That only ever passed because the second write silently clobbered the first tenant's row - the exact bug this branch fixes. The guard correctly refuses the overwrite and re-salts, so test_import_diff_route no longer saw "mem_python". Parameterized _seed's memory_id and gave the second test a distinct one. I had previously mis-reported this failure as pre-existing/unrelated; verified it passes on pristine trace-cortex/main and fails with the guard, so it was caused by this change. Credit: same root cause and fix found on integration/final (638c965). --- backend/tests/test_import_diff.py | 21 ++++++++++++++++----- 1 file changed, 16 insertions(+), 5 deletions(-) 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"}]}, From f57a62a52e42c9f36a79bb4bb601d7317e8c3455 Mon Sep 17 00:00:00 2001 From: Vamika Singhal Date: Wed, 29 Jul 2026 14:18:59 -0700 Subject: [PATCH 5/7] Document why memories/tasks ids stay globally unique I originally proposed migrating memories.id to a composite PRIMARY KEY(user_id, id), mirroring entities. That is the wrong fix and would break at runtime: SQLite requires an FK's parent columns to be PK or UNIQUE, and five single-column child FKs point at memories(id) (memory_entities, memory_topics, memory_relations x2, memory_vec_map - the last independently UNIQUE). Verified empirically: OperationalError: foreign key mismatch - "memory_entities" referencing "memories" entities could go composite only because nothing FKs to entities(id). Recorded at both table definitions so the migration isn't attempted again, pointing at the write-side guards that enforce the invariant. --- backend/app/database.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) 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, From fccfe69ab63c0ab43d2f9063851d03ac07f9858e Mon Sep 17 00:00:00 2001 From: Vamika Singhal Date: Fri, 31 Jul 2026 13:22:31 -0700 Subject: [PATCH 6/7] Chain the collision salt so it cannot clobber a third tenant Re-salting ONCE was not enough. The salted id can itself be owned by a third tenant, and INSERT OR REPLACE would then delete THEIR row - the same bug one level down. Verified: tenant C holding stable_id("mem_", "tenant-b:collide_mem") was silently erased when tenant B collided with tenant A and salted into it. Replaces the four inline single-hop guards with _tenant_unique_id(), which chains deterministically until the id is free (bounded, then raises rather than falling through and clobbering). The first hop is unchanged, so ids already written stay stable, and the user_id != ? predicate ignores the caller's own row so repeated saves stay idempotent instead of chaining further. Adds three tests: the third-tenant clobber (verified load-bearing), a 12-tenant pile-up on one id, and an idempotency check. --- backend/app/storage.py | 68 +++++++++++++++++++----------- backend/tests/test_sharding.py | 76 ++++++++++++++++++++++++++++++++++ 2 files changed, 120 insertions(+), 24 deletions(-) diff --git a/backend/app/storage.py b/backend/app/storage.py index 3c987d5c..015da830 100644 --- a/backend/app/storage.py +++ b/backend/app/storage.py @@ -21568,13 +21568,10 @@ def rebuild_index_from_vault(self, user_id: str) -> dict[str, Any]: # 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 = str(memory.get("id") or "") - if conn.execute( - "SELECT 1 FROM memories WHERE id = ? AND user_id != ? LIMIT 1", - (memory_id, user_id), - ).fetchone(): - memory_id = stable_id("mem_", f"{user_id}:{memory_id}") - memory["id"] = memory_id + 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 @@ -21652,13 +21649,8 @@ def rebuild_index_from_vault(self, user_id: str) -> dict[str, Any]: 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 = str(task.get("id") or "") - if conn.execute( - "SELECT 1 FROM tasks WHERE id = ? AND user_id != ? LIMIT 1", - (task_id, user_id), - ).fetchone(): - task_id = stable_id("task_", f"{user_id}:{task_id}") - task["id"] = task_id + 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 @@ -28414,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() @@ -28467,11 +28495,7 @@ def _save_memory( # 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 - if conn.execute( - "SELECT 1 FROM memories WHERE id = ? AND user_id != ? LIMIT 1", - (memory_id, user_id), - ).fetchone(): - memory_id = stable_id("mem_", f"{user_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 @@ -32081,11 +32105,7 @@ def _save_task(self, conn, capture_id: str, user_id: str, task: dict[str, Any], # 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. - if conn.execute( - "SELECT 1 FROM tasks WHERE id = ? AND user_id != ? LIMIT 1", - (task_id, user_id), - ).fetchone(): - task_id = stable_id("task_", f"{user_id}:{task_id}") + 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/tests/test_sharding.py b/backend/tests/test_sharding.py index 05eb8837..fcb9c879 100644 --- a/backend/tests/test_sharding.py +++ b/backend/tests/test_sharding.py @@ -612,6 +612,82 @@ def test_cross_tenant_salt_does_not_resurrect_a_forgotten_memory(self) -> None: 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() From 7ff8aacfdffa1679ed7d94784b662948470d8d6b Mon Sep 17 00:00:00 2001 From: Vamika Singhal Date: Fri, 31 Jul 2026 13:25:38 -0700 Subject: [PATCH 7/7] Add read-only forensics for pre-guard cross-tenant clobbers The guards stop future clobbers but cannot un-lose data already destroyed by a deployment that ran the old code, and there was no way to tell whether a given database was ever hit. A clobber only rewrote the memories/tasks row itself. Sibling rows carry their own user_id and memory_events is append-only and never rewritten, so a row whose user_id disagrees with its memory's owner is physical evidence the memory changed hands. Eight checks, each reported with its own confidence rather than summed. Detection is deliberately NOT gated on the surviving tenant count: a clobber that took over a tenant's only memory removes them from memories entirely, so that gate would report a total wipe as a harmless single-tenant database (caught by testing against a real pre-fix DB). Tenant count is instead taken across memories/captures/memory_events. Exits 1 on high-confidence evidence so it can gate a deploy check. --- backend/app/tenant_forensics.py | 270 +++++++++++++++++++++++++ backend/tests/test_tenant_forensics.py | 156 ++++++++++++++ 2 files changed, 426 insertions(+) create mode 100644 backend/app/tenant_forensics.py create mode 100644 backend/tests/test_tenant_forensics.py 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_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()