README.md

@@ -166,7 +166,12 @@ and runs up to three deliveries concurrently, but claims a record only when a de
 is ready. Claims have unique fencing tokens and are renewed before every network stage;
 completion, release, failure, and delivery checkpoints are token-fenced so an expired worker
 cannot alter a newer crash-recovery claim. Device purge cancels an active drain before closing
-private outbox storage. Results are coordinated through a bounded SQLite ledger. Set `STACKCHAIN_STATE_DIR` to a
+private outbox storage. Results are coordinated through a bounded SQLite ledger. All private SQLite stores enforce a
+filesystem boundary independently of the service umask: the database directory is repaired to
+owner-only `0700`, database and SQLite sidecar files are owner-only `0600`, and symlinked database
+paths are rejected before access. This protects state from unrelated local accounts, but it is not
+encryption at rest; secure host access, encrypted volumes, and private backups are still required.
+Set `STACKCHAIN_STATE_DIR` to a
 persistent, writable service directory (or set `STACKCHAIN_IDEMPOTENCY_DB` to an explicit
 SQLite path); the local default is `.stackchain-state/idempotency.sqlite3`. Ledger reads and
 writes run outside the request event loop, and lock admission is bounded to 100 ms by

src/available_issue_snapshot_store.py

@@ -3,14 +3,14 @@
 from __future__ import annotations
 
 import json
-import os
 import secrets
-import stat
 import sqlite3
 import time
 from dataclasses import dataclass
 from pathlib import Path
 
+from src.private_state import connect_private_sqlite
+
 
 class RefreshLeaseLost(RuntimeError):
     pass
@@ -29,12 +29,8 @@ class AvailableIssueSnapshotStore:
     def __init__(self, path, *, clock=None):
         self.path = Path(path)
         self.clock = clock or time.time
-        self.path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
-        os.chmod(self.path.parent, stat.S_IRWXU)
-        old_umask = os.umask(0o077)
-        try:
-            with self._connect() as connection:
-                connection.executescript(
+        with self._connect() as connection:
+            connection.executescript(
                     """
                     CREATE TABLE IF NOT EXISTS available_issue_snapshot (
                         singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
@@ -54,15 +50,12 @@ class AvailableIssueSnapshotStore:
                     );
                     """
                 )
-                connection.execute(
-                    "INSERT OR IGNORE INTO available_issue_snapshot VALUES (1, NULL, NULL, NULL)"
-                )
-        finally:
-            os.umask(old_umask)
-        os.chmod(self.path, stat.S_IRUSR | stat.S_IWUSR)
+            connection.execute(
+                "INSERT OR IGNORE INTO available_issue_snapshot VALUES (1, NULL, NULL, NULL)"
+            )
 
     def _connect(self):
-        connection = sqlite3.connect(self.path, timeout=1.0, isolation_level=None)
+        connection = connect_private_sqlite(self.path, timeout=1.0, isolation_level=None)
         connection.row_factory = sqlite3.Row
         connection.execute("PRAGMA busy_timeout = 1000")
         return connection

src/completed_filed_review_store.py

@@ -5,6 +5,8 @@ import sqlite3
 from datetime import datetime
 from pathlib import Path
 
+from src.private_state import connect_private_sqlite
+
 
 _REPOSITORY = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$")
 
@@ -17,7 +19,6 @@ class CompletedFiledReviewStore:
         self._initialize()
 
     def _initialize(self) -> None:
-        self.path.parent.mkdir(parents=True, exist_ok=True)
         with self._connect() as connection:
             connection.execute("PRAGMA journal_mode=WAL")
             connection.execute(
@@ -34,7 +35,7 @@ class CompletedFiledReviewStore:
             )
 
     def _connect(self) -> sqlite3.Connection:
-        return sqlite3.connect(self.path, timeout=self.timeout)
+        return connect_private_sqlite(self.path, timeout=self.timeout)
 
     @staticmethod
     def _login(login: str) -> str:

src/idempotency.py

@@ -5,6 +5,8 @@ from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, Callable
 
+from src.private_state import connect_private_sqlite
+
 
 @dataclass(frozen=True)
 class Reservation:
@@ -33,11 +35,10 @@ class IdempotencyLedger:
         self.max_entries = max_entries
         self.lock_timeout_seconds = lock_timeout_seconds
         self.clock = clock
-        self.path.parent.mkdir(parents=True, exist_ok=True)
         self._initialize()
 
     def _connect(self) -> sqlite3.Connection:
-        connection = sqlite3.connect(self.path, timeout=self.lock_timeout_seconds)
+        connection = connect_private_sqlite(self.path, timeout=self.lock_timeout_seconds)
         connection.execute(
             f"PRAGMA busy_timeout = {max(1, int(self.lock_timeout_seconds * 1000))}"
         )

src/later_store.py

@@ -6,6 +6,8 @@ import time
 from datetime import datetime
 from pathlib import Path
 
+from src.private_state import connect_private_sqlite
+
 
 class LaterStore:
     def __init__(
@@ -25,8 +27,7 @@ class LaterStore:
         self._initialize()
 
     def _initialize(self) -> None:
-        self.path.parent.mkdir(parents=True, exist_ok=True)
-        connection = sqlite3.connect(self.path, timeout=self.timeout)
+        connection = connect_private_sqlite(self.path, timeout=self.timeout)
         if connection.execute("PRAGMA user_version").fetchone()[0] >= 1:
             connection.close()
             return
@@ -72,7 +73,7 @@ class LaterStore:
         connection.close()
 
     def _connect(self) -> sqlite3.Connection:
-        return sqlite3.connect(self.path, timeout=self.timeout)
+        return connect_private_sqlite(self.path, timeout=self.timeout)
 
     def _record_operation(self, connection: sqlite3.Connection, login: str, operation_id: str) -> None:
         now = self.clock()

src/live_snapshot_store.py

@@ -6,12 +6,13 @@ import json
 import os
 import secrets
 import sqlite3
-import stat
 import time
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Callable, Iterable
 
+from src.private_state import connect_private_sqlite
+
 SECTIONS = ("context", "events", "notifications")
 
 
@@ -58,56 +59,46 @@ class LiveSnapshotStore:
         self._initialize()
 
     def _connect(self) -> sqlite3.Connection:
-        connection = sqlite3.connect(self.path, timeout=1.0, isolation_level=None)
+        connection = connect_private_sqlite(self.path, timeout=1.0, isolation_level=None)
         connection.row_factory = sqlite3.Row
         connection.execute("PRAGMA busy_timeout = 1000")
         return connection
 
     def _initialize(self) -> None:
-        self.path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
-        try:
-            os.chmod(self.path.parent, stat.S_IRWXU)
-        except OSError:
-            pass
-        old_umask = os.umask(0o077)
-        try:
-            with self._connect() as connection:
-                connection.executescript(
-                    """
-                    CREATE TABLE IF NOT EXISTS live_snapshot (
-                        singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
-                        value_json TEXT,
-                        created_at_json TEXT NOT NULL,
-                        failure_count_json TEXT NOT NULL,
-                        retry_at_json TEXT NOT NULL,
-                        revisions_json TEXT NOT NULL,
-                        generation TEXT NOT NULL
-                    );
-                    CREATE TABLE IF NOT EXISTS live_refresh_lease (
-                        singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
-                        owner TEXT NOT NULL,
-                        sections_json TEXT NOT NULL,
-                        expires_at REAL NOT NULL
-                    );
-                    CREATE TABLE IF NOT EXISTS live_read_notification (
-                        notification_id INTEGER PRIMARY KEY
-                    );
-                    """
-                )
-                connection.execute(
-                    """INSERT OR IGNORE INTO live_snapshot VALUES
-                    (1, NULL, ?, ?, ?, ?, ?)""",
-                    (
-                        json.dumps({section: None for section in SECTIONS}),
-                        json.dumps({section: 0 for section in SECTIONS}),
-                        json.dumps({section: None for section in SECTIONS}),
-                        json.dumps({section: 0 for section in SECTIONS}),
-                        secrets.token_hex(8),
-                    ),
-                )
-        finally:
-            os.umask(old_umask)
-        os.chmod(self.path, stat.S_IRUSR | stat.S_IWUSR)
+        with self._connect() as connection:
+            connection.executescript(
+                """
+                CREATE TABLE IF NOT EXISTS live_snapshot (
+                    singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
+                    value_json TEXT,
+                    created_at_json TEXT NOT NULL,
+                    failure_count_json TEXT NOT NULL,
+                    retry_at_json TEXT NOT NULL,
+                    revisions_json TEXT NOT NULL,
+                    generation TEXT NOT NULL
+                );
+                CREATE TABLE IF NOT EXISTS live_refresh_lease (
+                    singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
+                    owner TEXT NOT NULL,
+                    sections_json TEXT NOT NULL,
+                    expires_at REAL NOT NULL
+                );
+                CREATE TABLE IF NOT EXISTS live_read_notification (
+                    notification_id INTEGER PRIMARY KEY
+                );
+                """
+            )
+            connection.execute(
+                """INSERT OR IGNORE INTO live_snapshot VALUES
+                (1, NULL, ?, ?, ?, ?, ?)""",
+                (
+                    json.dumps({section: None for section in SECTIONS}),
+                    json.dumps({section: 0 for section in SECTIONS}),
+                    json.dumps({section: None for section in SECTIONS}),
+                    json.dumps({section: 0 for section in SECTIONS}),
+                    secrets.token_hex(8),
+                ),
+            )
 
     def try_acquire_refresh(
         self, sections: Iterable[str], *, lease_seconds: float

src/login_attempt_store.py

@@ -7,6 +7,8 @@ import sqlite3
 from pathlib import Path
 from typing import Callable
 
+from src.private_state import connect_private_sqlite
+
 
 class LoginAttemptStoreError(RuntimeError):
     """Raised when sign-in throttling state cannot be accessed safely."""
@@ -69,8 +71,7 @@ class LoginAttemptStore:
 
     def _connect(self) -> sqlite3.Connection:
         try:
-            self.path.parent.mkdir(parents=True, exist_ok=True)
-            connection = sqlite3.connect(self.path, timeout=self.lock_timeout_seconds)
+            connection = connect_private_sqlite(self.path, timeout=self.lock_timeout_seconds)
             connection.execute(
                 """
                 CREATE TABLE IF NOT EXISTS login_attempts (

src/passkey_store.py

@@ -6,6 +6,7 @@ from dataclasses import dataclass
 from pathlib import Path
 from typing import Callable
 
+from src.private_state import connect_private_sqlite
 from src.session_store import SessionStoreError
 
 
@@ -40,8 +41,7 @@ class PasskeyStore:
 
     def _connect(self) -> sqlite3.Connection:
         try:
-            self.path.parent.mkdir(parents=True, exist_ok=True)
-            connection = sqlite3.connect(self.path, timeout=0.1)
+            connection = connect_private_sqlite(self.path, timeout=0.1)
             connection.execute(
                 """
                 CREATE TABLE IF NOT EXISTS passkey_credentials (

src/private_state.py

@@ -0,0 +1,74 @@
+"""Filesystem boundary for SQLite state containing private operator data."""
+
+from __future__ import annotations
+
+import os
+import sqlite3
+import stat
+from pathlib import Path
+from typing import Any
+
+
+PRIVATE_DIRECTORY_MODE = stat.S_IRWXU
+PRIVATE_FILE_MODE = stat.S_IRUSR | stat.S_IWUSR
+
+
+def _reject_symlink(path: Path, *, label: str) -> None:
+    try:
+        metadata = path.lstat()
+    except FileNotFoundError:
+        return
+    if stat.S_ISLNK(metadata.st_mode):
+        raise ValueError(f"private SQLite {label} must not be a symlink: {path}")
+
+
+def prepare_private_sqlite_path(path: str | Path, *, create: bool = True) -> Path:
+    """Create or repair a private SQLite path without following a database symlink."""
+    database = Path(path)
+    parent = database.parent
+    _reject_symlink(parent, label="directory")
+    if create:
+        parent.mkdir(parents=True, mode=PRIVATE_DIRECTORY_MODE, exist_ok=True)
+    elif not parent.exists():
+        raise FileNotFoundError(database)
+    _reject_symlink(parent, label="directory")
+    if not parent.is_dir():
+        raise ValueError(f"private SQLite parent is not a directory: {parent}")
+    os.chmod(parent, PRIVATE_DIRECTORY_MODE, follow_symlinks=False)
+
+    _reject_symlink(database, label="database")
+    flags = os.O_RDWR | (os.O_CREAT if create else 0)
+    flags |= getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
+    descriptor = os.open(database, flags, PRIVATE_FILE_MODE)
+    try:
+        metadata = os.fstat(descriptor)
+        if not stat.S_ISREG(metadata.st_mode):
+            raise ValueError(f"private SQLite database is not a regular file: {database}")
+        os.fchmod(descriptor, PRIVATE_FILE_MODE)
+    finally:
+        os.close(descriptor)
+
+    for suffix in ("-wal", "-shm", "-journal"):
+        sidecar = Path(str(database) + suffix)
+        _reject_symlink(sidecar, label="sidecar")
+        try:
+            os.chmod(sidecar, PRIVATE_FILE_MODE, follow_symlinks=False)
+        except FileNotFoundError:
+            pass
+    return database
+
+
+def connect_private_sqlite(
+    path: str | Path, *, existing_only: bool = False, **kwargs: Any
+) -> sqlite3.Connection:
+    """Open SQLite only after enforcing its private directory and file modes."""
+    database = prepare_private_sqlite_path(path, create=not existing_only)
+    target: str | Path = database
+    if existing_only:
+        target = f"{database.resolve().as_uri()}?mode=rw"
+        kwargs["uri"] = True
+    connection = sqlite3.connect(target, **kwargs)
+    # SQLite derives new journal/WAL/SHM permissions from the database. Repair
+    # sidecars left by older releases as soon as a connection has opened them.
+    prepare_private_sqlite_path(database, create=not existing_only)
+    return connection

src/push_subscription_store.py

@@ -1,10 +1,11 @@
 import json
-import os
 import sqlite3
 from collections.abc import Iterable, Mapping
 from dataclasses import dataclass
 from pathlib import Path
 
+from src.private_state import connect_private_sqlite
+
 
 @dataclass(frozen=True)
 class PushDelivery:
@@ -57,8 +58,6 @@ class PushSubscriptionStore:
 
     def __init__(self, path: str | Path):
         self.path = Path(path)
-        self.path.parent.mkdir(parents=True, exist_ok=True)
-        os.chmod(self.path.parent, 0o700)
         with self._connect() as connection:
             connection.executescript(
                 """
@@ -141,10 +140,8 @@ class PushSubscriptionStore:
                 connection.execute(
                     "ALTER TABLE push_deadline_preferences ADD COLUMN reminder_days INTEGER NOT NULL DEFAULT 2"
                 )
-        os.chmod(self.path, 0o600)
-
     def _connect(self):
-        connection = sqlite3.connect(self.path, timeout=2)
+        connection = connect_private_sqlite(self.path, timeout=2)
         connection.execute("PRAGMA foreign_keys = ON")
         return connection
 

src/saved_search_store.py

@@ -5,6 +5,8 @@ import re
 import sqlite3
 from pathlib import Path
 
+from src.private_state import connect_private_sqlite
+
 
 _REPOSITORY = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$")
 _VIEW_ID = re.compile(r"^[A-Za-z0-9_-]{1,64}$")
@@ -26,8 +28,7 @@ class SavedSearchStore:
         self._initialize()
 
     def _initialize(self) -> None:
-        self.path.parent.mkdir(parents=True, exist_ok=True)
-        with sqlite3.connect(self.path, timeout=self.timeout) as connection:
+        with self._connect() as connection:
             connection.execute("PRAGMA journal_mode=WAL")
             connection.execute(
                 """
@@ -40,7 +41,7 @@ class SavedSearchStore:
             )
 
     def _connect(self) -> sqlite3.Connection:
-        return sqlite3.connect(self.path, timeout=self.timeout)
+        return connect_private_sqlite(self.path, timeout=self.timeout)
 
     @staticmethod
     def _login(login: str) -> str:

src/security_event_store.py

@@ -6,6 +6,8 @@ from dataclasses import dataclass
 from pathlib import Path
 from typing import Callable
 
+from src.private_state import connect_private_sqlite
+
 
 class SecurityEventStoreError(RuntimeError):
     """Raised when security activity cannot be persisted or read safely."""
@@ -53,8 +55,7 @@ class SecurityEventStore:
 
     def _connect(self) -> sqlite3.Connection:
         try:
-            self.path.parent.mkdir(parents=True, exist_ok=True)
-            connection = sqlite3.connect(self.path, timeout=self.lock_timeout_seconds)
+            connection = connect_private_sqlite(self.path, timeout=self.lock_timeout_seconds)
             connection.execute(
                 """
                 CREATE TABLE IF NOT EXISTS security_events (

src/session_store.py

@@ -7,6 +7,8 @@ from dataclasses import dataclass
 from pathlib import Path
 from typing import Callable
 
+from src.private_state import connect_private_sqlite
+
 
 class SessionStoreError(RuntimeError):
     """Raised when session state cannot be read or changed safely."""
@@ -51,15 +53,14 @@ class SessionStore:
     def _connect(self, *, initialize: bool = False) -> sqlite3.Connection:
         try:
             if initialize:
-                self.path.parent.mkdir(parents=True, exist_ok=True)
-                connection = sqlite3.connect(
+                connection = connect_private_sqlite(
                     self.path, timeout=self.lock_timeout_seconds
                 )
             else:
-                connection = sqlite3.connect(
-                    f"{self.path.resolve().as_uri()}?mode=rw",
+                connection = connect_private_sqlite(
+                    self.path,
                     timeout=self.lock_timeout_seconds,
-                    uri=True,
+                    existing_only=True,
                 )
             if initialize:
                 connection.execute(

src/today_store.py

@@ -6,6 +6,8 @@ import time
 from datetime import date
 from pathlib import Path
 
+from src.private_state import connect_private_sqlite
+
 
 class TodayPlanFull(ValueError):
     """Raised when an add would exceed the bounded Today plan."""
@@ -33,8 +35,7 @@ class TodayStore:
         self._initialize()
 
     def _initialize(self) -> None:
-        self.path.parent.mkdir(parents=True, exist_ok=True)
-        connection = sqlite3.connect(self.path, timeout=self.timeout)
+        connection = connect_private_sqlite(self.path, timeout=self.timeout)
         if connection.execute("PRAGMA user_version").fetchone()[0] >= 1:
             connection.close()
             return
@@ -110,7 +111,7 @@ class TodayStore:
         connection.close()
 
     def _connect(self) -> sqlite3.Connection:
-        return sqlite3.connect(self.path, timeout=self.timeout)
+        return connect_private_sqlite(self.path, timeout=self.timeout)
 
     def _record_operation(self, connection: sqlite3.Connection, login: str, operation_id: str) -> None:
         now = self.clock()

src/unfiled_draft_store.py

@@ -8,6 +8,8 @@ import sqlite3
 from datetime import datetime
 from pathlib import Path
 
+from src.private_state import connect_private_sqlite
+
 
 _DRAFT_ID = re.compile(r"^[A-Za-z0-9_-]{1,100}$")
 _REPOSITORY = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$")
@@ -38,8 +40,7 @@ class UnfiledDraftStore:
         self._initialize()
 
     def _initialize(self) -> None:
-        self.path.parent.mkdir(parents=True, exist_ok=True)
-        with sqlite3.connect(self.path, timeout=self.timeout) as connection:
+        with self._connect() as connection:
             connection.execute("PRAGMA journal_mode=WAL")
             connection.execute(
                 """CREATE TABLE IF NOT EXISTS unfiled_drafts (
@@ -50,7 +51,7 @@ class UnfiledDraftStore:
             )
 
     def _connect(self) -> sqlite3.Connection:
-        return sqlite3.connect(self.path, timeout=self.timeout)
+        return connect_private_sqlite(self.path, timeout=self.timeout)
 
     @staticmethod
     def _login(login: str) -> str:

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