From 8e014c0ebef730ef6c158af5397380bc8bbfeb20 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Sat, 17 Oct 2026 18:10:37 -0700 Subject: [PATCH] fix(history): bound persisted identifier bytes before restore reads --- database/execution_context_schema.py | 14 ++-- database/execution_history_schema.py | 35 +++++++-- tests/test_execution_context_schema.py | 95 +++++++++++++++++++++++++ tests/test_execution_history_restore.py | 85 ++++++++++++++++++++++ 4 files changed, 221 insertions(+), 8 deletions(-) diff --git a/database/execution_context_schema.py b/database/execution_context_schema.py index 9cc5d36..75b2527 100644 --- a/database/execution_context_schema.py +++ b/database/execution_context_schema.py @@ -79,11 +79,16 @@ TABLE_COLUMNS = { def _digest_check(column: str, nullable: bool = False) -> str: check = ( f"typeof({column}) = 'text' AND length({column}) = 64 " + f"AND length(CAST({column} AS BLOB)) = 64 " f"AND {column} NOT GLOB '*[^0-9a-f]*'" ) return f"{column} IS NULL OR ({check})" if nullable else check +def _bounded_uuid_check(column: str) -> str: + return f"{_uuid_check(column)} AND length(CAST({column} AS BLOB)) = 36" + + def _enum_check(column: str, values: frozenset[str]) -> str: return ( f"typeof({column}) = 'text' AND {column} IN (" @@ -108,8 +113,9 @@ TABLE_CHECKS = { "positive_snapshot_id IS NOT NULL OR negative_snapshot_id IS NOT NULL", ), "execution_run_contexts": ( - _uuid_check("run_id"), - "typeof(anchor_id) = 'text' AND length(anchor_id) BETWEEN 1 AND 128", + _bounded_uuid_check("run_id"), + "typeof(anchor_id) = 'text' AND length(anchor_id) BETWEEN 1 AND 128 " + "AND length(CAST(anchor_id AS BLOB)) <= 512", _digest_check("context_id"), "typeof(evidence_json) = 'text' AND length(CAST(evidence_json AS " "BLOB)) <= 8192", @@ -117,11 +123,11 @@ TABLE_CHECKS = { f"{column} IS NULL OR (typeof({column}) = 'integer' AND {column} > 0)" for column in ("positive_prompt_id", "negative_prompt_id", "pair_id") ), - f"reference_epoch IS NULL OR ({_uuid_check('reference_epoch')})", + f"reference_epoch IS NULL OR ({_bounded_uuid_check('reference_epoch')})", "typeof(capture_version) = 'integer' AND capture_version = 1", ), "execution_context_capture": ( - _uuid_check("run_id"), + _bounded_uuid_check("run_id"), _enum_check("state", BUNDLE_STATES), _enum_check("reason", CAPTURE_REASONS), "length(CAST(reason AS BLOB)) <= 64", diff --git a/database/execution_history_schema.py b/database/execution_history_schema.py index 3041e70..81e3dc6 100644 --- a/database/execution_history_schema.py +++ b/database/execution_history_schema.py @@ -92,6 +92,30 @@ _TABLE_CHECKS = { "execution_history_metadata": _METADATA_CHECKS, "execution_runs": _RUN_CHECKS, } +# Row-only guards preserve the supported v1 DDL fingerprint while bounding reads. +_ROW_BYTE_CHECKS = { + "execution_history_metadata": ("length(CAST(storage_epoch AS BLOB)) = 36",), + "execution_runs": ( + *( + f"length(CAST({column} AS BLOB)) = 36" + for column in ( + "run_id", + "session_id", + "job_id", + "storage_epoch", + ) + ), + *( + f"{column} IS NULL OR length(CAST({column} AS BLOB)) <= {limit}" + for column, limit in ( + ("node_id", 512), + ("node_type", 512), + ("exception_type", 1024), + ) + ), + ), +} + _INDEXES = { "idx_execution_runs_identity": (True, ("session_id", "queue_item_id")), "idx_execution_runs_observed": (False, ("observed_dequeue_at", "run_id")), @@ -127,11 +151,12 @@ def history_schema_error( if tables.keys() != _TABLE_COLUMNS.keys(): return "Incomplete execution history schema" metadata = conn.execute( - "SELECT * FROM execution_history_metadata LIMIT 2" + "SELECT CASE WHEN typeof(schema_version) = 'integer' " + "THEN schema_version ELSE NULL END FROM execution_history_metadata LIMIT 2" ).fetchall() - if len(metadata) != 1 or len(metadata[0]) != 3 or metadata[0][1] not in (1, 2): + if len(metadata) != 1 or len(metadata[0]) != 1 or metadata[0][0] not in (1, 2): return "Unsupported execution history metadata/version" - version = metadata[0][1] + version = metadata[0][0] if version == 1 and context_present: return "Unexpected execution contexts for history v1" if not allow_missing and version != SCHEMA_VERSION: @@ -173,7 +198,9 @@ def history_schema_error( or any(key[3] or key[4] != "BINARY" for key in keys) ): return f"{table} has unexpected execution history uniqueness" - predicate = " AND ".join(f"({check})" for check in checks) + predicate = " AND ".join( + f"({check})" for check in (*checks, *_ROW_BYTE_CHECKS[table]) + ) if conn.execute( f"SELECT 1 FROM {table} WHERE NOT ({predicate}) LIMIT 1" ).fetchone(): diff --git a/tests/test_execution_context_schema.py b/tests/test_execution_context_schema.py index 4211267..4515226 100644 --- a/tests/test_execution_context_schema.py +++ b/tests/test_execution_context_schema.py @@ -161,6 +161,101 @@ class ContextSchemaTests(unittest.TestCase): ) return evidence + def test_owned_identifier_constraints_reject_nul_suffixed_bytes(self): + columns = { + "execution_context_texts": ("text_id",), + "execution_prompt_snapshots": ( + "snapshot_id", + "raw_text_id", + "effective_text_id", + "prepend_text_id", + "append_text_id", + ), + "execution_contexts": ( + "context_id", + "positive_snapshot_id", + "negative_snapshot_id", + ), + "execution_run_contexts": ( + "run_id", + "anchor_id", + "context_id", + "reference_epoch", + ), + "execution_context_capture": ("run_id",), + } + for table, names in columns.items(): + for column in names: + with self.subTest(table=table, column=column): + conn = self.connection() + self.populate(conn) + prefix = ( + RUN + if column in ("run_id", "reference_epoch") + else "node" if column == "anchor_id" else "a" * 64 + ) + with self.assertRaises(sqlite3.IntegrityError): + conn.execute( + f"UPDATE {table} SET {column}=?", + (prefix + "\0" + "x" * 1024,), + ) + + def test_context_validation_rejects_oversized_keys_before_fetch(self): + for table, column, prefix in ( + ("execution_context_texts", "text_id", "a" * 64), + ("execution_run_contexts", "anchor_id", "node"), + ): + with self.subTest(table=table, column=column): + conn = self.connection() + self.populate(conn) + malicious = prefix + "\0" + "x" * 1024 + if table == "execution_context_texts": + for dependent in ( + "execution_context_capture", + "execution_run_contexts", + "execution_contexts", + "execution_prompt_snapshots", + ): + conn.execute(f"DELETE FROM {dependent}") + conn.execute("PRAGMA ignore_check_constraints=ON") + conn.execute(f"UPDATE {table} SET {column}=?", (malicious,)) + + def bounded_text(value): + self.assertNotEqual(value, malicious.encode()) + return value.decode() + + conn.text_factory = bounded_text + self.assertIsNotNone(history.history_schema_error(conn)) + + def test_unicode_anchor_boundary_and_prompt_nuls_preserve_exact_text(self): + conn = self.connection() + self.populate(conn) + text = "hello\0world" + text_id = content_digest("text", (text,)) + snapshot_id = content_digest("snapshot", (text, None, None, None)) + context_id = content_digest("context", (snapshot_id, None)) + conn.execute( + "UPDATE execution_context_texts SET text_id=?,text_value=?", + (text_id, text), + ) + conn.execute( + "UPDATE execution_prompt_snapshots SET snapshot_id=?,raw_text_id=?", + (snapshot_id, text_id), + ) + conn.execute( + "UPDATE execution_contexts SET context_id=?,positive_snapshot_id=?", + (context_id, snapshot_id), + ) + conn.execute( + "UPDATE execution_run_contexts SET context_id=?,anchor_id=?", + (context_id, "😀" * 128), + ) + self.assertIsNone(history.history_schema_error(conn)) + self.assertEqual( + conn.execute("SELECT text_value FROM execution_context_texts").fetchone(), + (text,), + ) + def test_valid_context_and_fingerprints(self): conn = self.connection() self.populate(conn) diff --git a/tests/test_execution_history_restore.py b/tests/test_execution_history_restore.py index 0eba0e6..aaaaab2 100644 --- a/tests/test_execution_history_restore.py +++ b/tests/test_execution_history_restore.py @@ -22,6 +22,91 @@ class HistoryRestoreTests(unittest.TestCase): finish = history_tests.HistoryDatabaseTests.finish count = history_tests.HistoryDatabaseTests.count + def test_legacy_identifier_and_diagnostic_bytes_rejected_before_fetch(self): + try: + from tests.test_execution_context_schema import legacy, RUN + except ImportError: + from test_execution_context_schema import legacy, RUN + from database.execution_history_schema import history_schema_error + + for table, column, prefix in ( + ("execution_history_metadata", "storage_epoch", RUN), + *( + ("execution_runs", column, RUN) + for column in ( + "run_id", + "session_id", + "job_id", + "storage_epoch", + ) + ), + ("execution_runs", "node_id", "x" * 128), + ("execution_runs", "node_type", "x" * 128), + ("execution_runs", "exception_type", "x" * 256), + ): + with self.subTest(table=table, column=column): + with closing(sqlite3.connect(":memory:")) as conn: + legacy(conn) + malicious = prefix + "\0" + "x" * 1024 + conn.execute(f"UPDATE {table} SET {column}=?", (malicious,)) + + def bounded_text(value): + self.assertNotEqual(value, malicious.encode()) + return value.decode() + + conn.text_factory = bounded_text + self.assertIsNotNone(history_schema_error(conn)) + + def test_legacy_metadata_version_payload_is_rejected_before_fetch(self): + try: + from tests.test_execution_context_schema import legacy + except ImportError: + from test_execution_context_schema import legacy + from database.execution_history_schema import history_schema_error + + for malicious in ("x" * 1024, b"x" * 1024): + with self.subTest(kind=type(malicious).__name__): + with closing(sqlite3.connect(":memory:")) as conn: + legacy(conn) + conn.execute("PRAGMA ignore_check_constraints=ON") + conn.execute( + "UPDATE execution_history_metadata SET schema_version=?", + (malicious,), + ) + + def bounded_row(cursor, row): + self.assertFalse(any(value == malicious for value in row)) + return row + + conn.row_factory = bounded_row + self.assertIsNotNone(history_schema_error(conn)) + + def test_legacy_four_byte_diagnostics_validate_and_migrate(self): + try: + from tests.test_execution_context_schema import legacy + except ImportError: + from test_execution_context_schema import legacy + from database.execution_history_schema import ( + history_schema_error, + migrate_execution_history, + ) + + with closing(sqlite3.connect(":memory:")) as conn: + legacy(conn) + conn.execute( + "UPDATE execution_runs SET node_id=?,node_type=?,exception_type=?", + ("😀" * 128, "😀" * 128, "😀" * 256), + ) + self.assertIsNone(history_schema_error(conn)) + migrate_execution_history(conn) + self.assertIsNone(history_schema_error(conn, allow_missing=False)) + self.assertEqual( + conn.execute( + "SELECT node_id,node_type,exception_type FROM " "execution_runs" + ).fetchone(), + ("😀" * 128, "😀" * 128, "😀" * 256), + ) + def test_committed_wal_backup_includes_history_and_epoch(self): db = self.database() conn = db.model.get_connection()