diff --git a/hermes_cli/doctor.py b/hermes_cli/doctor.py index a2c7df0408717..ccb4699267eb6 100644 --- a/hermes_cli/doctor.py +++ b/hermes_cli/doctor.py @@ -99,6 +99,88 @@ def _sqlite_upgrade_hint(install_method: str | None = None) -> str: ) +def _hermes_database_paths(hermes_home: Path) -> list[tuple[str, Path]]: + """Return (display name, path) pairs for Hermes-managed SQLite databases.""" + # backup.py owns the canonical list of per-profile stores; reuse it. + from hermes_cli.backup import _QUICK_STATE_FILES + + entries = [ + (name, hermes_home / name) + for name in _QUICK_STATE_FILES + if name.endswith(".db") + ] + # Non-default kanban boards each keep their own kanban.db. + for board_db in sorted((hermes_home / "kanban" / "boards").glob("*/kanban.db")): + entries.append((str(board_db.relative_to(hermes_home)), board_db)) + return entries + + +_SQLITE_HEADER_MAGIC = b"SQLite format 3\x00" + + +def _read_journal_mode(db_path: Path) -> tuple[str | None, str | None]: + """Return (journal mode, error) from the file header without opening the database. + + Header byte 18 is 2 for WAL and 1 for a rollback journal. Opening the + database through the SQLite engine — even read-only — creates -wal/-shm + sidecar files, which a diagnostic must not do. + """ + try: + with open(db_path, "rb") as fh: + header = fh.read(20) + except OSError as exc: + return None, str(exc) + if len(header) == 0: + return None, "file is empty" + if len(header) < 20 or not header.startswith(_SQLITE_HEADER_MAGIC): + return None, "file is not a database" + if header[18] == 2: + return "wal", None + if header[18] == 1: + return "rollback", None + return None, f"unrecognized file-format version {header[18]}" + + +def _report_database_journal_modes( + hermes_home: Path | None = None, + version_info: tuple[int, ...] | None = None, +) -> None: + """List each database's journal mode; warn on WAL under a vulnerable SQLite.""" + from hermes_state import is_sqlite_wal_reset_vulnerable + + vulnerable = is_sqlite_wal_reset_vulnerable(version_info) + home = hermes_home if hermes_home is not None else HERMES_HOME + try: + databases = _hermes_database_paths(home) + except Exception as exc: + check_warn(f"Could not list Hermes databases: {exc}") + return + for name, path in databases: + if not path.is_file(): + continue + mode, error = _read_journal_mode(path) + if error is not None: + if vulnerable: + check_warn( + f"{name}: journal mode could not be read", + f"({error}; cannot rule out WAL exposure)", + ) + else: + check_info(f"{name}: journal mode could not be read ({error})") + elif mode == "wal": + if vulnerable: + check_warn( + f"{name} is in WAL mode", + "(exposed to the WAL-reset bug until SQLite is upgraded)", + ) + else: + check_info(f"{name}: WAL journal mode") + elif vulnerable: + check_info(f"{name}: rollback journal mode (not exposed)") + else: + check_info(f"{name}: rollback journal mode") + + def _safe_which(cmd: str) -> str | None: """shutil.which wrapper resilient to platform monkeypatching in tests.""" try: @@ -965,6 +1047,7 @@ def run_doctor(args): check_ok(f"SQLite {_sqlite_ver}") if _sqlite_src_short: check_info(f"SQLite source id: {_sqlite_src_short}") + _report_database_journal_modes() except Exception as e: check_warn(f"SQLite version probe failed: {e}") # Check if in virtual environment diff --git a/tests/hermes_cli/test_doctor_journal_modes.py b/tests/hermes_cli/test_doctor_journal_modes.py new file mode 100644 index 0000000000000..a9c6b77b67c2b --- /dev/null +++ b/tests/hermes_cli/test_doctor_journal_modes.py @@ -0,0 +1,251 @@ +"""Tests for doctor's per-database journal-mode report. + +`hermes doctor` lists each Hermes-managed database with its journal mode and +flags databases that are in WAL while the linked SQLite carries the WAL-reset +bug (https://sqlite.org/wal.html#walresetbug). The probe reads the file header +only — it never opens the database through the SQLite engine, because even a +read-only engine open creates -wal/-shm sidecar files next to a WAL database. +""" + +import os +import sqlite3 + +import pytest + +import hermes_cli.doctor as doctor + +VULNERABLE = (3, 50, 4) +FIXED_VERSIONS = [(3, 51, 3), (3, 52, 0), (3, 50, 7), (3, 44, 6)] + +EXPOSED_TEXT = "exposed to the WAL-reset bug" + + +def _make_db(path, journal_mode=None): + conn = sqlite3.connect(path) + try: + if journal_mode: + conn.execute(f"PRAGMA journal_mode={journal_mode}") + conn.execute("CREATE TABLE t (x INTEGER)") + conn.commit() + finally: + conn.close() + + +def _sidecars(directory): + return sorted( + p.name for p in directory.iterdir() if p.name.endswith(("-wal", "-shm")) + ) + + +class TestReadJournalMode: + def test_reads_wal(self, tmp_path): + db = tmp_path / "state.db" + _make_db(db, journal_mode="WAL") + + mode, error = doctor._read_journal_mode(db) + + assert mode == "wal" + assert error is None + + def test_reads_rollback(self, tmp_path): + db = tmp_path / "state.db" + _make_db(db) + + mode, error = doctor._read_journal_mode(db) + + assert mode == "rollback" + assert error is None + + def test_probe_creates_no_wal_sidecars(self, tmp_path): + db = tmp_path / "state.db" + _make_db(db, journal_mode="WAL") + assert _sidecars(tmp_path) == [] + + assert doctor._read_journal_mode(db) == ("wal", None) + + assert _sidecars(tmp_path) == [] + + def test_missing_file_reports_error_and_does_not_create_it(self, tmp_path): + db = tmp_path / "missing.db" + + mode, error = doctor._read_journal_mode(db) + + assert mode is None + assert error + assert not db.exists() + + def test_empty_file_reports_error(self, tmp_path): + db = tmp_path / "state.db" + db.touch() + + mode, error = doctor._read_journal_mode(db) + + assert mode is None + assert error == "file is empty" + + def test_short_file_reports_error(self, tmp_path): + db = tmp_path / "state.db" + db.write_bytes(b"SQLite f") + + mode, error = doctor._read_journal_mode(db) + + assert mode is None + assert "not a database" in error + + def test_corrupt_file_reports_error(self, tmp_path): + db = tmp_path / "state.db" + db.write_bytes(b"this is not a sqlite database" * 4) + + mode, error = doctor._read_journal_mode(db) + + assert mode is None + assert "not a database" in error + + def test_locked_database_is_still_readable(self, tmp_path): + db = tmp_path / "state.db" + _make_db(db) + holder = sqlite3.connect(db, isolation_level=None) + try: + holder.execute("BEGIN EXCLUSIVE") + + assert doctor._read_journal_mode(db) == ("rollback", None) + finally: + holder.close() + + @pytest.mark.skipif(os.name == "nt", reason="chmod is a no-op on Windows") + @pytest.mark.skipif(os.geteuid() == 0, reason="root ignores file permissions") + def test_read_only_directory_is_still_readable(self, tmp_path): + db = tmp_path / "state.db" + _make_db(db, journal_mode="WAL") + os.chmod(tmp_path, 0o555) + try: + assert doctor._read_journal_mode(db) == ("wal", None) + finally: + os.chmod(tmp_path, 0o755) + assert _sidecars(tmp_path) == [] + + def test_does_not_mutate_database_files(self, tmp_path): + wal_db = tmp_path / "wal.db" + rollback_db = tmp_path / "plain.db" + _make_db(wal_db, journal_mode="WAL") + _make_db(rollback_db) + wal_bytes = wal_db.read_bytes() + rollback_bytes = rollback_db.read_bytes() + + assert doctor._read_journal_mode(wal_db) == ("wal", None) + assert doctor._read_journal_mode(rollback_db) == ("rollback", None) + + assert wal_db.read_bytes() == wal_bytes + assert rollback_db.read_bytes() == rollback_bytes + assert _sidecars(tmp_path) == [] + + +class TestReportDatabaseJournalModes: + def test_vulnerable_runtime_wal_db_is_exposed(self, tmp_path, capsys): + _make_db(tmp_path / "state.db", journal_mode="WAL") + + doctor._report_database_journal_modes(tmp_path, VULNERABLE) + + out = capsys.readouterr().out + assert "state.db is in WAL mode" in out + assert EXPOSED_TEXT in out + + def test_vulnerable_runtime_rollback_db_is_listed_not_exposed(self, tmp_path, capsys): + _make_db(tmp_path / "state.db") + + doctor._report_database_journal_modes(tmp_path, VULNERABLE) + + out = capsys.readouterr().out + assert "state.db: rollback journal mode (not exposed)" in out + assert EXPOSED_TEXT not in out + + @pytest.mark.parametrize("version", FIXED_VERSIONS) + def test_fixed_runtime_wal_db_is_not_exposed(self, tmp_path, capsys, version): + _make_db(tmp_path / "state.db", journal_mode="WAL") + + doctor._report_database_journal_modes(tmp_path, version) + + out = capsys.readouterr().out + assert "state.db: WAL journal mode" in out + assert EXPOSED_TEXT not in out + assert "⚠" not in out + + def test_lists_every_managed_database(self, tmp_path, capsys): + _make_db(tmp_path / "state.db", journal_mode="WAL") + _make_db(tmp_path / "projects.db") + _make_db(tmp_path / "kanban.db") + board = tmp_path / "kanban" / "boards" / "myboard" + board.mkdir(parents=True) + _make_db(board / "kanban.db", journal_mode="WAL") + + doctor._report_database_journal_modes(tmp_path, VULNERABLE) + + out = capsys.readouterr().out + assert "state.db is in WAL mode" in out + assert "projects.db: rollback journal mode (not exposed)" in out + assert "kanban.db: rollback journal mode (not exposed)" in out + assert "kanban/boards/myboard/kanban.db is in WAL mode" in out + + def test_missing_databases_are_skipped(self, tmp_path, capsys): + doctor._report_database_journal_modes(tmp_path, VULNERABLE) + + out = capsys.readouterr().out + assert "state.db" not in out + assert EXPOSED_TEXT not in out + + def test_locked_database_does_not_crash_or_block(self, tmp_path, capsys): + db = tmp_path / "state.db" + _make_db(db) + holder = sqlite3.connect(db, isolation_level=None) + try: + holder.execute("BEGIN EXCLUSIVE") + + doctor._report_database_journal_modes(tmp_path, VULNERABLE) + finally: + holder.close() + + out = capsys.readouterr().out + assert "state.db: rollback journal mode (not exposed)" in out + + @pytest.mark.skipif(os.name == "nt", reason="chmod is a no-op on Windows") + @pytest.mark.skipif(os.geteuid() == 0, reason="root ignores file permissions") + def test_unreadable_database_does_not_crash(self, tmp_path, capsys): + db = tmp_path / "state.db" + _make_db(db) + os.chmod(db, 0o000) + try: + doctor._report_database_journal_modes(tmp_path, VULNERABLE) + finally: + os.chmod(db, 0o644) + + out = capsys.readouterr().out + assert "state.db: journal mode could not be read" in out + assert "cannot rule out WAL exposure" in out + + def test_corrupt_database_does_not_crash(self, tmp_path, capsys): + (tmp_path / "state.db").write_bytes(b"garbage bytes, not sqlite" * 8) + + doctor._report_database_journal_modes(tmp_path, VULNERABLE) + + out = capsys.readouterr().out + assert "state.db: journal mode could not be read" in out + + def test_read_error_is_informational_on_fixed_runtime(self, tmp_path, capsys): + (tmp_path / "state.db").write_bytes(b"garbage bytes, not sqlite" * 8) + + doctor._report_database_journal_modes(tmp_path, (3, 51, 3)) + + out = capsys.readouterr().out + assert "state.db: journal mode could not be read" in out + assert "cannot rule out WAL exposure" not in out + assert "⚠" not in out + + def test_report_creates_no_wal_sidecars(self, tmp_path, capsys): + db = tmp_path / "state.db" + _make_db(db, journal_mode="WAL") + db_bytes = db.read_bytes() + + doctor._report_database_journal_modes(tmp_path, VULNERABLE) + + assert _sidecars(tmp_path) == [] + assert db.read_bytes() == db_bytes