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