feat(doctor): report per-database journal mode with WAL-reset exposure
hermes doctor already warns when the linked SQLite carries the WAL-reset bug, but it never said which databases are actually exposed. A database already in WAL mode on a vulnerable runtime can still corrupt; one on a rollback journal cannot. Doctor now lists each Hermes-managed database with its journal mode next to the SQLite version line and marks the WAL ones as exposed when the runtime is vulnerable. The probe reads the 20-byte file header and checks byte 18 (2 = WAL, 1 = rollback journal). It deliberately avoids the SQLite engine: even a read-only open creates -wal/-shm sidecars next to a WAL database, needs directory write access, and can wait on locks. The header read does none of that. It cannot tell delete from truncate/persist, so doctor reports 'rollback journal mode' rather than an exact mode name. Missing files are skipped; empty, unreadable, or corrupt files are reported as unreadable without failing doctor. The database list reuses backup.py's _QUICK_STATE_FILES plus per-board kanban databases. Exposure uses hermes_state.is_sqlite_wal_reset_vulnerable, so the 3.50.7 and 3.44.6 backports count as fixed.
This commit is contained in:
parent
886092bc54
commit
6583297086
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
Loading…
Reference in New Issue