[authentication/2.6.1] Commit enrollment decisions transactionally

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.

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
 
 

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

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

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)