From f4865575f416111e84dc279618d66f1aeba83d9d Mon Sep 17 00:00:00 2001 From: abrichr Date: Mon, 19 Oct 2026 13:06:12 -0400 Subject: [PATCH] fix(flow): create the durable authority database owner-only before SQLite opens it SQLite created a missing authority database with the process umask, and _connect tightened it to 0600 only after the schema was set up. A second durable run starting at the same moment (two --workers, or two processes on a fresh machine) saw a world-readable file and refused with "the external durable authority database permissions are too broad", which stopped the pool. The file is now created 0600 with O_EXCL (and O_NOFOLLOW) first. The permission check is unchanged. Co-Authored-By: Claude Opus 5.5 --- .../runtime/durable/authority.py | 19 +++++ .../tests/test_durable_authority_create.py | 69 +++++++++++++++++++ 2 files changed, 88 insertions(+) create mode 100644 packages/flow/tests/test_durable_authority_create.py diff --git a/packages/flow/openadapt_flow/runtime/durable/authority.py b/packages/flow/openadapt_flow/runtime/durable/authority.py index 03e3fe649..c5183fb44 100644 --- a/packages/flow/openadapt_flow/runtime/durable/authority.py +++ b/packages/flow/openadapt_flow/runtime/durable/authority.py @@ -655,6 +655,25 @@ def _connect(self) -> sqlite3.Connection: raise DurableAuthorityBusy( "the external durable authority database permissions are too broad" ) + elif os.name != "nt": + # Create the database owner-only before SQLite opens it. SQLite + # would create it with the process umask and the chmod below comes + # later, so a second durable run starting at the same moment (two + # workers, two processes) saw a world-readable file and refused. + try: + descriptor = os.open( + self.db_path, + os.O_CREAT | os.O_EXCL | os.O_WRONLY | getattr(os, "O_NOFOLLOW", 0), + 0o600, + ) + except FileExistsError: + pass + except OSError as exc: + raise DurableAuthorityBusy( + "the external durable authority database can't be created" + ) from exc + else: + os.close(descriptor) connection: Optional[sqlite3.Connection] = None try: connection = sqlite3.connect( diff --git a/packages/flow/tests/test_durable_authority_create.py b/packages/flow/tests/test_durable_authority_create.py new file mode 100644 index 000000000..70588a9ee --- /dev/null +++ b/packages/flow/tests/test_durable_authority_create.py @@ -0,0 +1,69 @@ +"""The external durable authority database is created owner-only. + +Two durable runs that start at the same moment (two workers, or two +processes) both open the authority database. SQLite creates a missing file +with the process umask; if the first run's later chmod hadn't happened yet, +the second run saw a world-readable file and refused. The file now exists, +owner-only, before SQLite opens it. +""" + +from __future__ import annotations + +import os +import sqlite3 +import stat +import threading + +import pytest + +from openadapt_flow.runtime.durable import authority as authority_module +from tests.test_durable_authority_v13 import _fresh_run + + +def _store(path): + """Start one fresh durable run (its own run id and namespace).""" + + import uuid + + return _fresh_run(path, run_id=str(uuid.uuid4()), namespace_id=uuid.uuid4().hex) + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX permission bits") +def test_a_fresh_database_is_owner_only_before_sqlite_opens_it(tmp_path, monkeypatch): + db_path = tmp_path / "authority" / "authority.sqlite3" + monkeypatch.setenv(authority_module.AUTHORITY_DB_ENV, str(db_path)) + seen: list[int] = [] + real_connect = sqlite3.connect + + def connect(*args, **kwargs): + if not seen: + seen.append(stat.S_IMODE(os.lstat(db_path).st_mode)) + return real_connect(*args, **kwargs) + + monkeypatch.setattr(authority_module.sqlite3, "connect", connect) + os.umask(0o022) + _store(tmp_path / "a") + assert seen and seen[0] & 0o077 == 0, oct(seen[0]) + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX permission bits") +def test_two_durable_runs_starting_together_both_start(tmp_path, monkeypatch): + db_path = tmp_path / "authority" / "authority.sqlite3" + monkeypatch.setenv(authority_module.AUTHORITY_DB_ENV, str(db_path)) + errors: list[BaseException] = [] + barrier = threading.Barrier(4) + + def start(name: str) -> None: + try: + barrier.wait() + _store(tmp_path / name) + except BaseException as exc: # noqa: BLE001 - collected for the assert + errors.append(exc) + + threads = [threading.Thread(target=start, args=(f"run-{i}",)) for i in range(4)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + assert errors == [] + assert stat.S_IMODE(os.lstat(db_path).st_mode) & 0o077 == 0