security: harden private SQLite state boundaries (Closes #891)

tests/test_private_state.py

@@ -0,0 +1,130 @@
+import os
+import sqlite3
+import stat
+from pathlib import Path
+
+import pytest
+
+from src.available_issue_snapshot_store import AvailableIssueSnapshotStore
+from src.completed_filed_review_store import CompletedFiledReviewStore
+from src.idempotency import IdempotencyLedger
+from src.later_store import LaterStore
+from src.live_snapshot_store import LiveSnapshotStore
+from src.login_attempt_store import LoginAttemptStore
+from src.passkey_store import PasskeyStore
+from src.private_state import connect_private_sqlite
+from src.push_subscription_store import PushSubscriptionStore
+from src.saved_search_store import SavedSearchStore
+from src.security_event_store import SecurityEventStore
+from src.session_store import SessionStore
+from src.today_store import TodayStore
+from src.unfiled_draft_store import UnfiledDraftStore
+
+
+def mode(path: Path) -> int:
+    return stat.S_IMODE(path.stat().st_mode)
+
+
+def test_private_sqlite_connection_enforces_directory_database_and_sidecar_modes(tmp_path):
+    state = tmp_path / "state"
+    old_umask = os.umask(0o022)
+    try:
+        connection = connect_private_sqlite(state / "private.sqlite3")
+        connection.execute("PRAGMA journal_mode=WAL")
+        connection.execute("CREATE TABLE secrets (value TEXT)")
+        connection.execute("INSERT INTO secrets VALUES ('operator data')")
+        connection.commit()
+
+        assert mode(state) == 0o700
+        assert mode(state / "private.sqlite3") == 0o600
+        assert mode(state / "private.sqlite3-wal") == 0o600
+        assert mode(state / "private.sqlite3-shm") == 0o600
+    finally:
+        connection.close()
+        os.umask(old_umask)
+
+
+def test_private_sqlite_connection_repairs_existing_permissive_modes_without_losing_rows(tmp_path):
+    state = tmp_path / "state"
+    state.mkdir(mode=0o755)
+    database = state / "private.sqlite3"
+    with sqlite3.connect(database) as connection:
+        connection.execute("CREATE TABLE secrets (value TEXT)")
+        connection.execute("INSERT INTO secrets VALUES ('keep me')")
+    state.chmod(0o755)
+    database.chmod(0o644)
+
+    with connect_private_sqlite(database) as connection:
+        value = connection.execute("SELECT value FROM secrets").fetchone()[0]
+
+    assert value == "keep me"
+    assert mode(state) == 0o700
+    assert mode(database) == 0o600
+
+
+def test_private_sqlite_connection_rejects_a_symlink_before_opening_target(tmp_path):
+    target = tmp_path / "target.sqlite3"
+    with sqlite3.connect(target) as connection:
+        connection.execute("CREATE TABLE sentinel (value TEXT)")
+        connection.execute("INSERT INTO sentinel VALUES ('unchanged')")
+    link = tmp_path / "state" / "private.sqlite3"
+    link.parent.mkdir()
+    link.symlink_to(target)
+
+    with pytest.raises(ValueError, match="symlink"):
+        connect_private_sqlite(link)
+
+    with sqlite3.connect(target) as connection:
+        assert connection.execute("SELECT value FROM sentinel").fetchone()[0] == "unchanged"
+
+
+@pytest.mark.parametrize(
+    "build",
+    [
+        TodayStore,
+        LaterStore,
+        SavedSearchStore,
+        UnfiledDraftStore,
+        CompletedFiledReviewStore,
+        PushSubscriptionStore,
+        LiveSnapshotStore,
+        AvailableIssueSnapshotStore,
+        lambda path: IdempotencyLedger(path, ttl_seconds=60, max_entries=10),
+    ],
+)
+def test_eager_private_stores_share_the_private_filesystem_boundary(tmp_path, build):
+    state = tmp_path / "state"
+    old_umask = os.umask(0o022)
+    try:
+        build(state / "store.sqlite3")
+    finally:
+        os.umask(old_umask)
+
+    assert mode(state) == 0o700
+    assert mode(state / "store.sqlite3") == 0o600
+
+
+@pytest.mark.parametrize(
+    "exercise",
+    [
+        lambda path: SecurityEventStore(path, clock=lambda: 1).record("sign_in"),
+        lambda path: LoginAttemptStore(
+            path, clock=lambda: 1, max_failures=3, window_seconds=60
+        ).record_failure("203.0.113.10"),
+        lambda path: PasskeyStore(path, clock=lambda: 1).issue_challenge(
+            b"challenge", session_id=None, purpose="authentication", action="sign_in",
+            target="dashboard", source="browser", ttl_seconds=60,
+        ),
+        lambda path: SessionStore(path, clock=lambda: 1).activate("session", 60),
+    ],
+)
+def test_on_demand_private_stores_share_the_private_filesystem_boundary(tmp_path, exercise):
+    state = tmp_path / "state"
+    old_umask = os.umask(0o022)
+    try:
+        exercise(state / "store.sqlite3")
+    finally:
+        os.umask(old_umask)
+
+    assert mode(state) == 0o700
+    assert mode(state / "store.sqlite3") == 0o600