131 lines
4.5 KiB
Python
131 lines
4.5 KiB
Python
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
|