From 04d468578c9219e87546944ca5bc9c64dc0a9ef8 Mon Sep 17 00:00:00 2001 From: Omry Yadan Date: Mon, 26 Oct 2026 05:56:18 +0800 Subject: [PATCH] [authentication/2.6.1] Commit enrollment decisions transactionally Persist exact approved enrollment results in one SQLite transaction. Recheck live approval and account state at the commit decision, preserve cancellation and rejection ordering, and recover private rollback journals under exclusive instance ownership. Retain that ownership until real workers settle during shutdown. Add transaction, crash-recovery and lifespan regressions. Public enrollment remains at the prepared-only boundary for the following client-promotion slice. Validation: 1,609 Python unit tests passed (43 skipped), 43 Python/Go integration tests passed, Go test/vet, Pyrefly, Black and website build passed. DBB and fresh independent Sol review completed for this exact candidate. --- ...rollment-activation-transaction.feature.md | 3 + server/src/arbiter_server/authority.py | 22 + server/src/arbiter_server/enrollment.py | 68 ++ .../arbiter_server/enrollment_activation.py | 123 ++++ server/src/arbiter_server/main.py | 23 +- server/src/arbiter_server/pairing.py | 84 ++- .../tests/unit/test_enrollment_activation.py | 663 ++++++++++++++++++ server/tests/unit/test_enrollment_admin.py | 3 +- server/tests/unit/test_local_control.py | 5 +- server/tests/unit/test_windows_control.py | 5 +- 10 files changed, 985 insertions(+), 14 deletions(-) create mode 100644 server/news/+enrollment-activation-transaction.feature.md create mode 100644 server/src/arbiter_server/enrollment_activation.py create mode 100644 server/tests/unit/test_enrollment_activation.py diff --git a/server/news/+enrollment-activation-transaction.feature.md b/server/news/+enrollment-activation-transaction.feature.md new file mode 100644 index 0000000..ead7c34 --- /dev/null +++ b/server/news/+enrollment-activation-transaction.feature.md @@ -0,0 +1,3 @@ +Add the internal transactional enrollment writer with current approval checks, +atomic completion records, and SQLite crash recovery. Public enrollment still +stops at private prepared state until client activation support is connected. diff --git a/server/src/arbiter_server/authority.py b/server/src/arbiter_server/authority.py index ec50e3d..11f1734 100644 --- a/server/src/arbiter_server/authority.py +++ b/server/src/arbiter_server/authority.py @@ -357,6 +357,28 @@ def _current_record(root: Path) -> AuthorityRecord | None: from .generations import reconcile path = reconcile(root) + if path is not None: + # The instance owner alone may recover an interrupted live SQLite + # transaction. Network readers remain read-only and never recover it. + journal = path.with_name(path.name + "-journal") + if journal.exists() or journal.is_symlink(): + _check_path(journal) + for suffix in ("-wal", "-shm"): + sidecar = path.with_name(path.name + suffix) + if sidecar.exists() or sidecar.is_symlink(): + raise AuthorityError( + "unexpected authority SQLite sidecar; preserve state" + ) + if journal.exists(): + try: + with closing( + sqlite3.connect(path.as_uri() + "?mode=rw", uri=True, timeout=0) + ) as db: + db.execute("PRAGMA schema_version").fetchone() + except sqlite3.Error as exc: + raise AuthorityError( + "authority journal recovery failed; preserve state" + ) from exc return _read_record(path) if path is not None else None diff --git a/server/src/arbiter_server/enrollment.py b/server/src/arbiter_server/enrollment.py index f51f094..80999c9 100644 --- a/server/src/arbiter_server/enrollment.py +++ b/server/src/arbiter_server/enrollment.py @@ -173,6 +173,7 @@ def __init__( self._epoch = 0 self._requests: dict[str, PendingRequest] = {} self._approved: dict[str, ApprovedRequest] = {} + self._committing: dict[str, ApprovedRequest] = {} def reset(self) -> None: with self._lock: @@ -209,6 +210,66 @@ def exhaust_approval(self, approval: ApprovedRequest) -> None: self._approved.pop(selector) self._requests.pop(selector, None) + def complete( + self, + approval: ApprovedRequest, + commit: Callable[[Callable[[], None]], bytes], + ) -> bytes: + """Reserve an approval, then consume it at one live commit decision. + + That decision orders reset/expiry against activation without holding a + registry lock during SQLite I/O. Later invalidation cannot undo an + already authorized transaction. A failed write restores only the same + still-live approval, never a request invalidated in the meantime. + """ + decided = False + with self._lock: + selector = approval.request.advertisement.selector + epoch = self._epoch + + def check() -> None: + nonlocal decided + with self._lock: + if ( + decided + or self._approved.get(selector) is not approval + or self._requests.get(selector) is not approval.request + or self._clock() >= approval.request.deadline + ): + raise ValueError("Enrollment approval is no longer current") + # Only this transaction can consume the approval. Keep the + # request counted until the write settles or is invalidated. + self._approved.pop(selector) + decided = True + + if selector in self._committing: + raise ValueError("Enrollment activation is in progress; try again") + if ( + self._approved.get(selector) is not approval + or self._clock() >= approval.request.deadline + ): + raise ValueError("Enrollment approval is no longer current") + self._committing[selector] = approval + try: + result = commit(check) + with self._lock: + if self._requests.get(selector) is approval.request: + self._requests.pop(selector) + return result + except BaseException: + with self._lock: + if ( + decided + and self._epoch == epoch + and self._requests.get(selector) is approval.request + and self._clock() < approval.request.deadline + ): + self._approved[selector] = approval + raise + finally: + with self._lock: + self._committing.pop(selector, None) + def _summary(self, request: PendingRequest, now: float) -> dict[str, object]: advertisement = request.advertisement approved = self._approved.get(advertisement.selector) @@ -339,6 +400,11 @@ def control( or request.advertisement.digest != arguments["request_digest"] ): raise ValueError("Enrollment selection changed; list requests again.") + committing = self._committing.get(selector) + if committing is not None and ( + command != "reject" or self._approved.get(selector) is not committing + ): + raise ValueError("Enrollment activation is in progress; try again") if command == "reject": self._approved.pop(selector, None) del self._requests[selector] @@ -380,6 +446,8 @@ def register(self, raw: bytes | bytearray) -> PendingRequest: epoch, window_deadline = self.require_open() advertisement = decode_advertisement(raw) with self._lock: + if advertisement.selector in self._committing: + raise _invalid() now = self._clock() if epoch != self._epoch or now >= min( window_deadline, self._window_deadline diff --git a/server/src/arbiter_server/enrollment_activation.py b/server/src/arbiter_server/enrollment_activation.py new file mode 100644 index 0000000..530b527 --- /dev/null +++ b/server/src/arbiter_server/enrollment_activation.py @@ -0,0 +1,123 @@ +"""One durable enrollment result, committed only after private client readiness.""" + +from collections.abc import Callable +from contextlib import closing +from datetime import datetime +import json +from pathlib import Path +import sqlite3 + +from cryptography.hazmat.primitives import serialization + +from .authority import _read_record +from .credentials import Identity, Role, verify_certificate +from .enrollment import ApprovedRequest +from .enrollment_receipt import result_fields +from .generations import ACTIVE, _control, _database +from .prepared import PREPARED_TAG + + +def activate( + root: Path, + approval: ApprovedRequest, + raw: bytes, + accounts: Callable[[], set[tuple[str, str]]], + check: Callable[[], None], + now: datetime, +) -> bytes: + """Instance, request and responder ownership span this bounded worker call. + + The exact result contains the identity, all certificates, recipient key and + approved grants. No second representation of those authorities is published. + """ + ids, fields = result_fields(raw) + ad = approval.request.advertisement + offset = len(PREPARED_TAG) + 144 + if ( + raw[offset : offset + 22] != ad.selector.encode() + or raw[offset + 22 : offset + 54] != bytes.fromhex(ad.digest) + or raw[offset + 54 : offset + 86] != approval.request.revision.encode() + or int.from_bytes(raw[offset + 86 : offset + 94], "big") <= now.timestamp() + or fields[5] != ad.recipient_spki + or tuple(tuple(pair) for pair in json.loads(fields[8])) != approval.grants + ): + raise ValueError("Enrollment activation unavailable") + active = _control(root / ACTIVE) + if active is None: + raise ValueError("Enrollment activation unavailable") + path = _database(root, active["active"]) + authority = _read_record(path) + if ids[:2] != [authority.server_id, authority.owner_id]: + raise ValueError("Enrollment activation unavailable") + for index, identity, issuer, issuer_id, public in ( + (3, Identity(Role.PORTABLE, ids[2]), fields[2], ids[1], ad.portable_spki), + ( + 4, + Identity(Role.ACCESS, ids[2], ids[0], ids[3]), + fields[0], + ids[0], + ad.access_spki, + ), + ): + cert = verify_certificate( + fields[index], identity, issuer_der=issuer, issuer_id=issuer_id, now=now + ) + if ( + cert.public_key().public_bytes( + serialization.Encoding.DER, + serialization.PublicFormat.SubjectPublicKeyInfo, + ) + != public + ): + raise ValueError("Enrollment activation unavailable") + with closing(sqlite3.connect(path, timeout=0, isolation_level=None)) as db: + db.execute("PRAGMA synchronous=FULL") + db.execute("BEGIN IMMEDIATE") + try: + stored = dict(db.execute("SELECT role, certificate_der FROM credentials")) + if ( + fields[0] != stored[Role.SERVER_AUTHORITY.value] + or fields[2] != stored[Role.OWNER.value] + ): + raise ValueError("Enrollment activation unavailable") + db.execute( + "CREATE TABLE IF NOT EXISTS enrollments (enrollment_id TEXT PRIMARY KEY, " + "prepared BLOB NOT NULL CHECK(length(prepared)<=65519), " + "denied INTEGER NOT NULL CHECK(denied IN (0,1)), " + "inactive_after INTEGER NOT NULL, selector TEXT NOT NULL UNIQUE, " + "revision TEXT NOT NULL)" + ) + existing = db.execute( + "SELECT prepared, denied, inactive_after FROM enrollments " + "WHERE enrollment_id=? OR selector=?", + (ids[3], ad.selector), + ).fetchone() + if existing is not None: + if ( + bytes(existing[0]) != raw + or existing[1] + or existing[2] <= now.timestamp() + ): + raise ValueError("Enrollment activation unavailable") + else: + db.execute( + "INSERT INTO enrollments VALUES (?, ?, 0, ?, ?, ?)", + ( + ids[3], + raw, + int(now.timestamp()) + 180 * 86400, + ad.selector, + approval.request.revision, + ), + ) + if not set(approval.grants) <= accounts(): + raise ValueError("An approved account is no longer configured") + # Account evaluation can invalidate or outlive the original request. + # Consume its authorization only after that work, under the live + # caller's attempt/approval locks. No such lock spans SQLite I/O. + check() + db.commit() + except BaseException: + db.rollback() + raise + return raw diff --git a/server/src/arbiter_server/main.py b/server/src/arbiter_server/main.py index f21d3fa..2f6c1a7 100644 --- a/server/src/arbiter_server/main.py +++ b/server/src/arbiter_server/main.py @@ -24,7 +24,7 @@ Sequence, ) from dataclasses import asdict, dataclass, is_dataclass -from contextlib import asynccontextmanager +from contextlib import ExitStack, asynccontextmanager from ipaddress import ip_address from pathlib import Path, PurePath from tempfile import mkstemp @@ -2321,15 +2321,26 @@ async def lifespan(_app: object) -> AsyncGenerator[None, None]: from .local_control import running root = _server_data_root(_storage_config(cfg)) / "authority" - with running_instance(root): - enrollment_requests.reset() + with ExitStack() as instance: + instance.enter_context(running_instance(root)) + release_instance = instance.pop_all().close try: + pairing_responder.handoff_instance(release_instance) + except BaseException: + release_instance() + raise + try: + enrollment_requests.reset() async with running(root, enrollment_control): yield finally: - pairing_responder.close() - enrollment_requests.reset() - http_work.close() + try: + enrollment_requests.reset() + finally: + try: + http_work.close() + finally: + pairing_responder.close() async def register_enrollment(request: Request) -> object: limit = min(MAX_ADVERTISEMENT_BYTES, max_request_body_bytes) diff --git a/server/src/arbiter_server/pairing.py b/server/src/arbiter_server/pairing.py index 626f1f5..5f1b6f5 100644 --- a/server/src/arbiter_server/pairing.py +++ b/server/src/arbiter_server/pairing.py @@ -1,10 +1,9 @@ -"""Bounded Noise channel foundation; this module cannot activate an enrollment. +"""Bounded Noise pairing, key proofs and conditional durable activation. The prologue joins the fixed prefix, ASCII selector (22 bytes), exact request SHA-256 (32 bytes), and access/portable/recipient DER SPKIs (91/91/44 bytes). The first payload on each side binds its role and SHA-256 of that prologue. -Signing-key proofs run inside this channel; durable trust and activation remain -later protocol steps. +Signing-key proofs and private client readiness precede durable activation. """ from __future__ import annotations @@ -132,6 +131,19 @@ def __init__( self._handshakes: set[PairingLease] = set() self._attempts: dict[str, _Attempts] = {} self._closed = False + self._instance_release: Callable[[], None] | None = None + + def handoff_instance(self, release: Callable[[], None]) -> None: + """Release the instance context only after every lease has settled.""" + with self._lock: + if self._closed: + raise RuntimeError("Pairing responder is closed") + if self._instance_release is not None: + raise RuntimeError("Pairing instance ownership was already handed off") + self._instance_release = release + finished = self._take_instance_release() + if finished is not None: + finished() def reserve(self) -> PairingLease: """Reserve before reading/parsing a carrier; there is no admission queue.""" @@ -162,6 +174,13 @@ def _release(self, lease: PairingLease) -> None: self._leases.discard(lease) self._handshakes.discard(lease) + def _take_instance_release(self) -> Callable[[], None] | None: + if not self._closed or self._leases or self._instance_release is None: + return None + release = self._instance_release + self._instance_release = None + return release + def _exhaust(self, state: _Attempts) -> None: self._requests.exhaust_approval(state.approval) selector = state.approval.request.advertisement.selector @@ -172,11 +191,17 @@ def _close_lease(self, lease: PairingLease) -> None: with self._lock: lease._closed = True self._release(lease) + finished = self._take_instance_release() + if finished is not None: + finished() def _finished(self, lease: PairingLease) -> None: with self._lock: lease._busy = False self._release(lease) + finished = self._take_instance_release() + if finished is not None: + finished() async def _run( self, lease: PairingLease, operation: Callable[[], Result] @@ -228,6 +253,8 @@ def _state(self, selector: str) -> _Attempts: # Reap before inserting a new state. Rejected/reset/expired contexts # cannot accumulate outside the existing pending-request capacity. for key, state in tuple(self._attempts.items()): + if state.active is not None and state.active.lease._busy: + continue if ( state.active is not None and self._clock() >= state.active.lease.deadline @@ -242,7 +269,10 @@ def _state(self, selector: str) -> _Attempts: raise _failed() if selector not in self._attempts: self._attempts[selector] = _Attempts(approved) - return self._attempts[selector] + state = self._attempts[selector] + if state.approval is not approved: + raise _failed() + return state def _live(self, session: PairingSession) -> None: self._check(session.lease) @@ -467,14 +497,58 @@ def verify() -> None: with self._lock: self._live(session) session.ready_complete = True - # No identity/access/grant write or activation is reachable here. await self._run(session.lease, verify) + async def activate( + self, + session: PairingSession, + root: Path, + accounts: Callable[[], set[tuple[str, str]]], + ) -> bytes: + from .enrollment_activation import activate + + def commit() -> bytes: + with self._lock: + self._live(session) + state = self._state(session.approval.request.advertisement.selector) + if not session.ready_complete or not state.prepared: + raise _failed() + # Build the encrypted receipt before retiring the Noise owner; + # it cannot leave this worker until SQLite commit succeeds. + wire = self._message(session, state.prepared, encrypt=True) + + def transaction(current: Callable[[], None]) -> bytes: + def check() -> None: + with self._lock: + self._check(session.lease) + current() + + return activate( + root, + session.approval, + state.prepared, + accounts, + check, + datetime.now(timezone.utc), + ) + + self._requests.complete(session.approval, transaction) + with self._lock: + self._attempts.pop( + session.approval.request.advertisement.selector, None + ) + return wire + + return await self._run(session.lease, commit) + def close(self) -> None: with self._lock: self._closed = True for lease in tuple(self._leases): lease.close() self._attempts.clear() + finished = self._take_instance_release() self._pool.close() + if finished is not None: + finished() diff --git a/server/tests/unit/test_enrollment_activation.py b/server/tests/unit/test_enrollment_activation.py new file mode 100644 index 0000000..f738143 --- /dev/null +++ b/server/tests/unit/test_enrollment_activation.py @@ -0,0 +1,663 @@ +"""Activation is one durable result under the live ready/approval owner.""" + +import asyncio +from contextlib import closing +from dataclasses import replace +import base64 +from datetime import datetime, timedelta, timezone +import hashlib +import json +import os +from pathlib import Path +import sqlite3 +import subprocess +import sys +import threading +import time +from typing import Any + +import pytest +from starlette.applications import Starlette + +from arbiter_server.authority import ( + AuthorityBusy, + initialize_authority, + running_instance, +) +from arbiter_server.config import AppConfig +from arbiter_server.enrollment_activation import activate +from arbiter_server.enrollment_receipt import result_fields, retrieve +from arbiter_server.generations import ACTIVE, _control, _database +from arbiter_server.main import build_server +from arbiter_server.pairing import prologue +from arbiter_server.pairing_identity import load_pairing_identity +from arbiter_server.pairing_proofs import _digest, _signature +from arbiter_server.prepared import READY_TAG, prepare, ready_context +from server.tests.unit.test_enrollment_receipt import request_for +from server.tests.unit.test_pairing import connected +from server.tests.unit.test_pairing_proofs import client_proof, proof_setup +from server.tests.unit.test_prepared import session_fixture + + +def database(root): + active = _control(root / ACTIVE) + assert active is not None + return _database(root, active["active"]) + + +def approve_advertisement(requests, ad, secret): + data = dict(version=1, selector=ad.selector, name=ad.name) + for name in ("access_spki", "portable_spki", "recipient_spki"): + data[name] = base64.b64encode(getattr(ad, name)).decode() + pending = requests.register(json.dumps(data).encode()) + requests.control( + "approve", + { + "selector": ad.selector, + "revision": pending.revision, + "request_digest": ad.digest, + "psk": base64.urlsafe_b64encode(secret).decode().rstrip("="), + "grants": [], + }, + set(), + ) + approval = requests.approved(ad.selector) + assert approval is not None + return approval + + +async def mark_ready(responder, session, raw, keys): + context = ready_context( + session.handshake_hash, prologue(session.approval.request.advertisement), raw + ) + await responder.ready( + session, + READY_TAG + + hashlib.sha256(raw).digest() + + _signature(keys[0], _digest(context, b"ready")), + ) + + +def enrollment_table(db): + return db.execute( + "SELECT name FROM sqlite_master WHERE name='enrollments'" + ).fetchone() + + +@pytest.mark.parametrize( + "failure", [None, "account", "check", "binding", "expired", "lock"] +) +def test_transaction_and_exact_idempotence( + tmp_path, private_windows_test_parent, failure +): + root = tmp_path / "authority" + initialize_authority(root, recovery_file=tmp_path / "operator.json") + requests, approval, keys, _, _, now = proof_setup() + approval = replace(approval, grants=(("imap", "personal"),)) + raw = prepare(root, approval, int(now.timestamp()) + 300) + if failure == "binding": + raw = raw[:-1] + b"x" + path = database(root) + blocker = sqlite3.connect(path, timeout=0) + if failure == "lock": + blocker.execute("BEGIN IMMEDIATE") + checks = [] + + def check(): + checks.append(True) + if failure == "check": + raise ValueError("synthetic deadline before commit") + + try: + + def commit(): + return activate( + root, + approval, + raw, + lambda: set() if failure == "account" else {("imap", "personal")}, + check, + now + timedelta(minutes=5) if failure == "expired" else now, + ) + + if failure: + expected: tuple[type[Exception], ...] = ( + ValueError, + sqlite3.OperationalError, + ) + with pytest.raises(expected): + commit() + with closing(sqlite3.connect(path)) as db: + assert enrollment_table(db) is None + else: + assert commit() == raw + before = path.read_bytes() + assert commit() == raw + assert path.read_bytes() == before + ids, fields = result_fields(raw) + with closing(sqlite3.connect(path)) as db: + row = db.execute("SELECT * FROM enrollments").fetchall() + assert len(row) == 1 and bytes(row[0][1]) == raw + assert row[0][3] == int(now.timestamp()) + 180 * 86400 + assert json.loads(fields[8]) == [["imap", "personal"]] + db.execute("UPDATE enrollments SET denied=1") + db.commit() + with pytest.raises(ValueError): + commit() + with pytest.raises(ValueError): + exporter = os.urandom(32) + retrieve(root, request_for(raw, exporter, keys[0]), exporter, now) + finally: + blocker.close() + requests.reset() + + +@pytest.mark.parametrize( + "failure", + [ + None, + "not-ready", + "reject", + "reset", + "expiry", + "shutdown", + "account-reset", + "account-expiry", + "account-shutdown", + "account-reject", + ], +) +def test_ready_commit_and_pending_retirement( + tmp_path, private_windows_test_parent, failure +): + async def run(): + root, requests, responder, lease, session, peer, keys = await session_fixture( + tmp_path + ) + try: + raw = await responder.prepare(session, root) + if failure != "not-ready": + await mark_ready(responder, session, raw, keys) + if failure == "reject": + requests.exhaust_approval(session.approval) + if failure == "reset": + requests.reset() + if failure == "expiry": + requests._clock = lambda: session.approval.request.deadline + if failure == "shutdown": + responder.close() + + def accounts(): + if failure == "account-reset": + requests.reset() + if failure == "account-expiry": + requests._clock = lambda: session.approval.request.deadline + if failure == "account-shutdown": + responder.close() + if failure == "account-reject": + requests.control( + "reject", + dict( + selector=session.approval.request.advertisement.selector, + revision=session.approval.request.revision, + request_digest=session.approval.request.advertisement.digest, + ), + set(), + ) + return set() + + if failure: + with pytest.raises(ValueError): + await responder.activate(session, root, accounts) + with closing(sqlite3.connect(database(root))) as db: + assert enrollment_table(db) is None + else: + assert peer.decrypt(await responder.activate(session, root, set)) == raw + assert not requests.pending() and not responder._attempts + assert ( + requests.approved(session.approval.request.advertisement.selector) + is None + ) + with pytest.raises(ValueError): + await responder.activate(session, root, set) + finally: + responder.close() + + asyncio.run(run()) + + +def test_replacement_approval_rejects_old_noise_session( + tmp_path, private_windows_test_parent +): + async def run(): + root, requests, responder, lease, session, peer, keys = await session_fixture( + tmp_path + ) + try: + old = session.approval + ad = old.request.advertisement + selection: dict[str, object] = dict( + selector=ad.selector, + revision=old.request.revision, + request_digest=ad.digest, + ) + requests.control("reject", selection, set()) + replacement = approve_advertisement(requests, ad, os.urandom(32)) + assert replacement is not None and replacement is not old + with pytest.raises(ValueError): + requests.complete(old, lambda _: b"") + with pytest.raises(ValueError): + await responder.prepare(session, root) + assert lease._closed and requests.approval_is_current(replacement) + fresh, current, current_peer, _ = await connected(responder, replacement) + assert ( + current_peer.decrypt( + await responder.message(current, b"replacement", encrypt=True) + ) + == b"replacement" + ) + fresh.close() + finally: + responder.close() + + asyncio.run(run()) + + +@pytest.mark.parametrize("invalidate", [None, "reset", "expiry"]) +def test_failed_authorized_write_restores_only_current_approval( + tmp_path, private_windows_test_parent, monkeypatch, invalidate +): + root = tmp_path / "authority" + initialize_authority(root, recovery_file=tmp_path / "operator.json") + requests, approval, _, _, _, now = proof_setup() + raw = prepare(root, approval, int(now.timestamp()) + 300) + connect = sqlite3.connect + + class FailedCommit(sqlite3.Connection): + def commit(self): + assert not requests.approval_is_current(approval) + if invalidate == "reset": + requests.reset() + if invalidate == "expiry": + requests._clock = lambda: approval.request.deadline + raise sqlite3.OperationalError("synthetic commit failure") + + monkeypatch.setattr( + sqlite3, "connect", lambda *a, **k: connect(*a, **k, factory=FailedCommit) + ) + with pytest.raises(sqlite3.OperationalError, match="synthetic commit failure"): + requests.complete( + approval, lambda check: activate(root, approval, raw, set, check, now) + ) + assert requests.approval_is_current(approval) == (invalidate is None) + assert not requests._committing + with closing(connect(database(root))) as db: + assert enrollment_table(db) is None + + +def test_decided_commit_outlives_reset_without_blocking_admission( + tmp_path, private_windows_test_parent, monkeypatch +): + entered, release = threading.Event(), threading.Event() + connect = sqlite3.connect + + class PausedCommit(sqlite3.Connection): + def commit(self): + entered.set() + assert release.wait(4) + super().commit() + + async def run(): + root, requests, responder, lease, session, peer, keys = await session_fixture( + tmp_path + ) + task = None + try: + raw = await responder.prepare(session, root) + await mark_ready(responder, session, raw, keys) + monkeypatch.setattr( + sqlite3, + "connect", + lambda *a, **k: connect(*a, **k, factory=PausedCommit), + ) + task = asyncio.create_task(responder.activate(session, root, set)) + assert await asyncio.to_thread(entered.wait, 2) + assert not requests.approval_is_current(session.approval) + assert requests.pending() == (session.approval.request,) + with pytest.raises(ValueError, match="in progress"): + requests.control( + "reject", + dict( + selector=session.approval.request.advertisement.selector, + revision=session.approval.request.revision, + request_digest=session.approval.request.advertisement.digest, + ), + set(), + ) + extra = responder.reserve() + with pytest.raises(ValueError): + await responder.start(extra, "unapproved-selector", b"x") + assert not lease._closed + extra.close() + requests.reset() + responder.close() + assert not release.is_set() and lease._busy + release.set() + with pytest.raises(ValueError): + await task + with closing(connect(database(root))) as db: + assert db.execute("SELECT prepared FROM enrollments").fetchall() == [ + (raw,) + ] + assert not requests.pending() and not requests._committing + assert not responder._leases + finally: + release.set() + if task is not None: + await asyncio.gather(task, return_exceptions=True) + responder.close() + + watchdog = threading.Timer(3, release.set) + watchdog.start() + try: + asyncio.run(run()) + finally: + watchdog.cancel() + + +@pytest.mark.parametrize("phase", ["before-decision", "after-decision"]) +def test_lifespan_holds_instance_lock_until_cancelled_writer_settles( + tmp_path, private_windows_test_parent, phase, monkeypatch +): + entered, release = threading.Event(), threading.Event() + + async def run(selected_phase: str): + root = tmp_path / "authority" + initialize_authority(root, recovery_file=tmp_path / "operator.json") + cfg = AppConfig() + cfg.arbiter.storage.server_data_dir = str(tmp_path) + cfg.arbiter.storage.plugin_data_dir = str(tmp_path / "plugins") + app = build_server(cfg, service_plugins=[], prepare_tls=False).app + assert isinstance(app, Starlette) + requests, responder = app.state.enrollment_requests, app.state.pairing_responder + original_connect = sqlite3.connect + task = shutdown = None + entered_lifespan = False + lifespan = app.router.lifespan_context(app) + try: + await lifespan.__aenter__() + entered_lifespan = True + _, fixture_approval, keys, _, _, now = proof_setup() + advertisement = fixture_approval.request.advertisement + requests.open_window() + approval = approve_advertisement( + requests, advertisement, fixture_approval.secret + ) + lease, session, peer, _ = await connected(responder, approval) + identity = load_pairing_identity(root) + assert identity is not None + leaf, authority = identity + peer.decrypt(await responder.prove_server(session, leaf, authority, now)) + await responder.prove_client( + session, peer.encrypt(client_proof(keys, session.proof_context)) + ) + raw = await responder.prepare(session, root) + await mark_ready(responder, session, raw, keys) + if selected_phase == "before-decision": + + def accounts(): + entered.set() + if not release.wait(5): + raise RuntimeError("test watchdog expired") + return set() + + else: + + class PausedCommit(sqlite3.Connection): + def commit(self): + entered.set() + if not release.wait(5): + raise RuntimeError("test watchdog expired") + return super().commit() + + def paused_connect(*args: Any, **kwargs: Any) -> sqlite3.Connection: + return original_connect(*args, **kwargs, factory=PausedCommit) + + monkeypatch.setattr(sqlite3, "connect", paused_connect) + accounts = set + task = asyncio.create_task(responder.activate(session, root, accounts)) + assert await asyncio.to_thread(entered.wait, 2) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert requests.approval_is_current(approval) is ( + selected_phase == "before-decision" + ) + + shutdown_started, shutdown_release = asyncio.Event(), asyncio.Event() + from arbiter_server.local_control import LocalControl + from arbiter_server.windows_control import WindowsControl + + control_type = WindowsControl if os.name == "nt" else LocalControl + original_local_close = control_type.close + + async def paused_close(control: Any) -> None: + shutdown_started.set() + await asyncio.wait_for(shutdown_release.wait(), 5) + await original_local_close(control) + + monkeypatch.setattr(control_type, "close", paused_close) + shutdown = asyncio.create_task(lifespan.__aexit__(None, None, None)) + await asyncio.wait_for(shutdown_started.wait(), 2) + shutdown.cancel() + shutdown_release.set() + with pytest.raises(asyncio.CancelledError): + await shutdown + + def enter_second_instance(): + with running_instance(root): + return True + + second = asyncio.create_task(asyncio.to_thread(enter_second_instance)) + with pytest.raises(AuthorityBusy): + await asyncio.wait_for(second, 2) + assert lease._busy + release.set() + for _ in range(500): + if not lease._busy: + break + await asyncio.sleep(0.01) + assert not lease._busy and not responder._leases + assert await asyncio.wait_for(asyncio.to_thread(enter_second_instance), 5) + entered_lifespan = False + monkeypatch.setattr(sqlite3, "connect", original_connect) + with running_instance(root): + with closing(original_connect(database(root))) as db: + tables = db.execute( + "SELECT name FROM sqlite_master WHERE name='enrollments'" + ).fetchall() + if selected_phase == "after-decision": + assert db.execute( + "SELECT prepared FROM enrollments" + ).fetchall() == [(raw,)] + else: + assert not tables + finally: + release.set() + monkeypatch.setattr(sqlite3, "connect", original_connect) + if task is not None: + await asyncio.gather(task, return_exceptions=True) + if shutdown is not None: + await asyncio.gather(shutdown, return_exceptions=True) + elif entered_lifespan: + await lifespan.__aexit__(None, None, None) + responder.close() + + watchdog = threading.Timer(8, release.set) + watchdog.start() + try: + asyncio.run(run(phase)) + finally: + watchdog.cancel() + + +def test_lifespan_reentry_rejects_closed_responder( + tmp_path, private_windows_test_parent +): + root = tmp_path / "authority" + initialize_authority(root, recovery_file=tmp_path / "operator.json") + cfg = AppConfig() + cfg.arbiter.storage.server_data_dir = str(tmp_path) + cfg.arbiter.storage.plugin_data_dir = str(tmp_path / "plugins") + app = build_server(cfg, service_plugins=[], prepare_tls=False).app + assert isinstance(app, Starlette) + first = app.router.lifespan_context(app) + second = app.router.lifespan_context(app) + + async def run(): + await first.__aenter__() + await first.__aexit__(None, None, None) + with pytest.raises(RuntimeError, match="closed"): + await second.__aenter__() + assert not (root / "control.sock").exists() + with running_instance(root): + pass + + asyncio.run(run()) + + +@pytest.mark.parametrize("shutdown", [False, True]) +def test_reject_during_slow_accounts_keeps_admission_and_worker_owned( + tmp_path, private_windows_test_parent, shutdown +): + entered, release = threading.Event(), threading.Event() + + def accounts(): + entered.set() + assert release.wait(4) + return set() + + async def run(): + root, requests, responder, lease, session, peer, keys = await session_fixture( + tmp_path + ) + task = None + try: + raw = await responder.prepare(session, root) + await mark_ready(responder, session, raw, keys) + task = asyncio.create_task(responder.activate(session, root, accounts)) + assert await asyncio.to_thread(entered.wait, 2) + started = time.monotonic() + extra = responder.reserve() + extra.close() + assert time.monotonic() - started < 0.5 and not release.is_set() + result = requests.control( + "reject", + dict( + selector=session.approval.request.advertisement.selector, + revision=session.approval.request.revision, + request_digest=session.approval.request.advertisement.digest, + ), + set(), + ) + assert result["rejected"] == session.approval.request.advertisement.selector + if shutdown: + requests.reset() + responder.close() + assert lease in responder._leases and lease._busy + release.set() + with pytest.raises(ValueError): + await task + assert not responder._leases and not requests._committing + with closing(sqlite3.connect(database(root))) as db: + assert enrollment_table(db) is None + finally: + release.set() + if task is not None: + await asyncio.gather(task, return_exceptions=True) + responder.close() + + watchdog = threading.Timer(3, release.set) + watchdog.start() + try: + asyncio.run(run()) + finally: + watchdog.cancel() + + +@pytest.mark.parametrize("phase", ["before-commit", "after-commit"]) +def test_real_process_crash_and_restart(tmp_path, private_windows_test_parent, phase): + root = tmp_path / "authority" + initialize_authority(root, recovery_file=tmp_path / "operator.json") + _, approval, keys, _, _, now = proof_setup() + raw = prepare(root, approval, int(now.timestamp()) + 300) + ad = approval.request.advertisement + data = dict( + root=str(root), + phase=phase, + raw=base64.b64encode(raw).decode(), + advertisement=dict( + selector=ad.selector, + name=ad.name, + digest=ad.digest, + **{ + k: base64.b64encode(getattr(ad, k)).decode() + for k in ("access_spki", "portable_spki", "recipient_spki") + }, + ), + revision=approval.request.revision, + ) + # No keys or PSK in argv/env. Exit without finally/close/rollback. A one-page + # cache forces actual dirty-page spill, not merely a zero-header journal. + script = """ +import sys,json,base64,os,time,sqlite3 +from pathlib import Path +from datetime import datetime,timezone +from arbiter_server.authority import running_instance +from arbiter_server.enrollment import Advertisement, PendingRequest, ApprovedRequest +from arbiter_server.enrollment_activation import activate +d=json.load(sys.stdin) +ad=d['advertisement'] +for k in ('access_spki','portable_spki','recipient_spki'): ad[k]=base64.b64decode(ad[k]) +a=ApprovedRequest(PendingRequest(Advertisement(**ad),d['revision'],time.monotonic(),time.monotonic()+300),(),b'') +original=sqlite3.connect +def connect(*args,**kwargs): + db=original(*args,**kwargs); db.execute('PRAGMA cache_size=1'); return db +with running_instance(Path(d['root'])): + sqlite3.connect=connect + def check(): + if d['phase']=='before-commit': os._exit(61) + activate(Path(d['root']),a,base64.b64decode(d['raw']),set,check,datetime.now(timezone.utc)) + os._exit(62) +""" + result = subprocess.run( + [sys.executable, "-c", script], + input=json.dumps(data), + capture_output=True, + text=True, + timeout=15, + env={**os.environ, "PYTHONPATH": str(Path(__file__).parents[2] / "src")}, + ) + assert result.returncode == (61 if phase == "before-commit" else 62), result.stderr + if phase == "before-commit": + journal = database(root).with_name(database(root).name + "-journal") + assert journal.exists() and journal.read_bytes()[:8] != bytes(8) + with running_instance(root): + with closing(sqlite3.connect(database(root))) as db: + tables = db.execute( + "SELECT name FROM sqlite_master WHERE name='enrollments'" + ).fetchall() + assert bool(tables) == (phase == "after-commit") + if tables: + assert db.execute("SELECT COUNT(*) FROM enrollments").fetchone() == (1,) + if phase == "after-commit": + exporter = os.urandom(32) + assert ( + retrieve( + root, + request_for(raw, exporter, keys[0]), + exporter, + datetime.now(timezone.utc), + ) + == raw + ) diff --git a/server/tests/unit/test_enrollment_admin.py b/server/tests/unit/test_enrollment_admin.py index 83c2412..4a2a847 100644 --- a/server/tests/unit/test_enrollment_admin.py +++ b/server/tests/unit/test_enrollment_admin.py @@ -395,7 +395,8 @@ async def exercise(): manager.pending() == () and manager.approved(pending.advertisement.selector) is None ) - async with app.router.lifespan_context(app): + restart_app = cast(Any, _client(tmp_path)._app) + async with restart_app.router.lifespan_context(restart_app): listing = await command("list") assert listing["requests"] == [] and listing["open"] is False for path in root.rglob("*"): diff --git a/server/tests/unit/test_local_control.py b/server/tests/unit/test_local_control.py index 3530d5f..7105158 100644 --- a/server/tests/unit/test_local_control.py +++ b/server/tests/unit/test_local_control.py @@ -97,7 +97,10 @@ async def exercise(): assert not (root / "control.sock").exists() restored = a.authority_status(root) assert restored is not None and restored.to_dict() == initial - async with app.router.lifespan_context(app): + restart_app = cast( + Any, build_server(cfg, service_plugins=[], prepare_tls=False).app + ) + async with restart_app.router.lifespan_context(restart_app): assert json.loads((await command("status")).stdout) == initial assert recovery.read_bytes() == custody diff --git a/server/tests/unit/test_windows_control.py b/server/tests/unit/test_windows_control.py index 52b1ebe..0dfdf32 100644 --- a/server/tests/unit/test_windows_control.py +++ b/server/tests/unit/test_windows_control.py @@ -164,7 +164,10 @@ async def exercise(): assert not (root / wc._LOCATOR).exists() with pytest.raises(OSError): wc.w.Native().open(name) - async with app.router.lifespan_context(app): + fresh = cast( + Any, build_server(cfg, service_plugins=[], prepare_tls=False).app + ) + async with fresh.router.lifespan_context(fresh): assert wc._locator(root)[0] != name assert json.loads((await command("status")).stdout) == expected assert recovery.read_bytes() == custody