fix: validate sqlite chain length metadata

Author: arunabha003

minichain/persistence.py

@@ -219,11 +219,24 @@ def _load_snapshot_from_sqlite(db_path: str) -> dict[str, Any]:
         account_rows = conn.execute(
             "SELECT address, account_json FROM accounts ORDER BY address ASC"
         ).fetchall()
+        chain_length_row = conn.execute(
+            "SELECT value FROM metadata WHERE key = ?",
+            ("chain_length",),
+        ).fetchone()
     except sqlite3.DatabaseError as exc:
         raise ValueError(f"Invalid SQLite persistence data in '{db_path}'") from exc
     finally:
         conn.close()
 
+    if chain_length_row is None:
+        raise ValueError(f"Invalid SQLite persistence data in '{db_path}'")
+    try:
+        expected_chain_length = int(chain_length_row["value"])
+    except (TypeError, ValueError) as exc:
+        raise ValueError(f"Invalid SQLite persistence data in '{db_path}'") from exc
+    if expected_chain_length != len(block_rows):
+        raise ValueError(f"Invalid SQLite persistence data in '{db_path}'")
+
     try:
         chain = [json.loads(row["block_json"]) for row in block_rows]
         state = {

tests/test_persistence.py

@@ -152,6 +152,15 @@ def test_missing_required_sqlite_table_raises(self):
         with self.assertRaises(ValueError):
             load(path=self.tmpdir)
 
+    def test_truncated_chain_rows_raises_value_error(self):
+        bc, _, _ = self._chain_with_tx()
+        save(bc, path=self.tmpdir)
+        db_path = os.path.join(self.tmpdir, DB_FILE)
+        with sqlite3.connect(db_path) as conn:
+            conn.execute("DELETE FROM blocks WHERE height = 1")
+        with self.assertRaises(ValueError):
+            load(path=self.tmpdir)
+
     def test_malformed_block_row_raises_value_error(self):
         bc = Blockchain()
         save(bc, path=self.tmpdir)