From e144661ff7049451365f0dccc80ba840f39d3433 Mon Sep 17 00:00:00 2001 From: Omry Yadan Date: Sat, 31 Oct 2026 01:26:40 +0800 Subject: [PATCH] [arbiter-authentication/2.10.3] Prepared reuse and single activation Preserve a selected portable agent identity through target preparation and activation. The target binds approval to the exact registered issuer revision, issues its own access credentials, and commits the result and unique agent binding in one transaction. A denied or inactive target binding blocks another enrollment. Validate the selected identity before the client sends ready, and recover the exact committed target result after a lost reply or restart. Recheck portable certificate validity after account evaluation, before authorizing a new commit. Public registration and CLI source selection remain on the fresh flow until the next planned slice. Validation: 1,853 Python tests passed (43 skipped), 20 portable-focused tests passed, explicit full formatting and lint/type checks passed, and Go tests/vet passed. Real Python/Go peers cover A offline, B restart, lost replies and changed-certificate rejection. The expiry regression fails against the original activation code and passes with this fix. Native CI and remote review are separate publication gates. --- client/go-cli/internal/identity/prepared.go | 6 + .../identity/registration_peer_test.go | 126 ++++++++++ docs/enrollment.md | 7 + ...+portable-enrollment-activation.feature.md | 1 + server/src/arbiter_server/enrollment.py | 50 +++- .../arbiter_server/enrollment_activation.py | 34 ++- server/src/arbiter_server/owner_issuers.py | 46 ++++ server/src/arbiter_server/prepared.py | 40 +++- .../integration/test_enrollment_cli_peer.py | 215 ++++++++++++++++++ .../tests/unit/test_enrollment_activation.py | 127 +++++++++++ server/tests/unit/test_prepared.py | 177 ++++++++++++++ 11 files changed, 813 insertions(+), 16 deletions(-) create mode 100644 server/news/+portable-enrollment-activation.feature.md diff --git a/client/go-cli/internal/identity/prepared.go b/client/go-cli/internal/identity/prepared.go index 98315d2..3206535 100644 --- a/client/go-cli/internal/identity/prepared.go +++ b/client/go-cli/internal/identity/prepared.go @@ -114,6 +114,12 @@ func verifyPreparedPurpose(raw []byte, material *Provisional, server *PairingSer if len(data) != 0 || !bytes.Equal(fields[0], server.AuthorityDER) || !bytes.Equal(fields[1], server.CertificateDER) || !bytes.Equal(fields[5], public.RecipientKey) { return nil, errPairing } + if material.record.Version == 2 && !bytes.Equal(material.record.Source, []byte("null")) { + selected, err := decodePortableSource(material.record.Source, material.portable, nil) + if err != nil || selected == nil || ids[1] != selected.OwnerID || ids[2] != selected.AgentID || !bytes.Equal(fields[2], selected.OwnerDER) || !bytes.Equal(fields[3], selected.Certificate) { + return nil, errPairing + } + } var serverErr error if receipt || source { serverErr = verifyHistoricalServerCredential(fields[1], Identity{Role: ServerTLS, SubjectID: ids[0]}, &Issuer{DER: fields[0], ID: ids[0]}, now) diff --git a/client/go-cli/internal/identity/registration_peer_test.go b/client/go-cli/internal/identity/registration_peer_test.go index bcd001a..697cf6c 100644 --- a/client/go-cli/internal/identity/registration_peer_test.go +++ b/client/go-cli/internal/identity/registration_peer_test.go @@ -8,6 +8,7 @@ import ( "encoding/json" "fmt" "io" + "net" "net/http" "net/url" "os" @@ -72,3 +73,128 @@ func TestPythonRegistrationPeer(t *testing.T) { } assertOnlyProvisionalLock(t, root) } + +// Real A/B servers supply the production HTTP/Noise/TLS exchange. Only local +// approval and controlled shutdown use this bounded synthetic handoff socket. +func TestPythonPortableEnrollmentPeer(t *testing.T) { + endpoint := os.Getenv("ARBITER_TEST_PORTABLE_ENROLLMENT_CONTROL") + if endpoint == "" { + t.Skip("Python portable enrollment peer is not running") + } + host, _, err := net.SplitHostPort(endpoint) + if err != nil || host != "127.0.0.1" { + t.Fatal("loopback required") + } + conn, err := net.DialTimeout("tcp", endpoint, 5*time.Second) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + conn.SetDeadline(time.Now().Add(40 * time.Second)) + input, output := json.NewDecoder(io.LimitReader(conn, 32768)), json.NewEncoder(conn) + var config struct{ A, B, Mode string } + if input.Decode(&config) != nil { + t.Fatal("fixture configuration") + } + ctx := context.Background() + root := filepath.Join(custodyTestDir(t), "state") + err = WithEnrollmentOwner(ctx, root, func() error { + original, secret, e := NewProvisional(config.A, "source", time.Now()) + if e != nil { + return e + } + var first *EnrollmentReceipt + var sourceBytes []byte + for phase := 0; phase < 2; phase++ { + material := original + if phase == 1 { + source, e := LoadPortableIdentity(root, original.record.Selector, time.Now()) + if e != nil { + return e + } + material, secret, e = NewProvisionalFromIdentity(config.B, "target", time.Now(), source) + if e != nil { + return e + } + if samePublicKey(material.access.Public(), original.access.Public()) || bytes.Equal(material.recipient.Bytes(), original.recipient.Bytes()) { + return fmt.Errorf("target keys reused") + } + } + e = PairEnrollment(ctx, material, secret, 5*time.Second, filepath.Join(root, "prepared"), func(_ context.Context, selector, secret string) error { + if output.Encode(map[string]any{"phase": phase, "selector": selector, "secret": secret}) != nil { + return fmt.Errorf("handoff") + } + var accepted bool + if input.Decode(&accepted) != nil || !accepted { + return fmt.Errorf("approval") + } + return nil + }) + if phase == 0 && e != nil || phase == 1 && (e == nil) != (config.Mode == "ok") { + return fmt.Errorf("pairing outcome: %v", e) + } + if phase == 1 { + if config.Mode == "changed-portable" { + if _, err := os.Stat(filepath.Join(root, "prepared", material.record.Selector+".json")); !os.IsNotExist(err) { + return fmt.Errorf("changed portable identity reached ready") + } + if output.Encode("rejected") != nil { + return fmt.Errorf("rejection handoff") + } + var rejected bool + if input.Decode(&rejected) != nil || !rejected { + return fmt.Errorf("rejection acknowledgement") + } + return nil + } + if output.Encode("prepared") != nil { + return fmt.Errorf("restart handoff") + } + var restarted bool + if input.Decode(&restarted) != nil || !restarted { + return fmt.Errorf("restart") + } + } + receipt, e := ReconcileEnrollment(ctx, filepath.Join(root, "prepared"), material.record.Selector, 5*time.Second) + if e != nil { + return e + } + if phase == 0 { + first = receipt + } else { + if receipt.prepared.AgentID != first.prepared.AgentID || receipt.prepared.OwnerID != first.prepared.OwnerID || receipt.prepared.ServerID == first.prepared.ServerID || receipt.prepared.EnrollmentID == first.prepared.EnrollmentID { + return fmt.Errorf("portable/target IDs") + } + again, e := ReconcileEnrollment(ctx, filepath.Join(root, "prepared"), material.record.Selector, 5*time.Second) + if e != nil || !bytes.Equal(again.prepared.raw, receipt.prepared.raw) { + return fmt.Errorf("exact recovery") + } + } + if e = receipt.Promote(ctx, filepath.Join(root, "enrolled")); e != nil { + return e + } + if phase == 0 { + sourceBytes, e = os.ReadFile(filepath.Join(root, "enrolled", original.record.Selector+".json")) + if e != nil { + return e + } + if output.Encode("source-ready") != nil { + return fmt.Errorf("source handoff") + } + var offline bool + if input.Decode(&offline) != nil || !offline { + return fmt.Errorf("source shutdown") + } + } + } + after, e := os.ReadFile(filepath.Join(root, "enrolled", original.record.Selector+".json")) + if e != nil || !bytes.Equal(after, sourceBytes) { + return fmt.Errorf("source custody changed") + } + return nil + }) + if err != nil { + t.Fatal(err) + } + fmt.Println("ARBITER_PORTABLE_ENROLLMENT=pass") +} diff --git a/docs/enrollment.md b/docs/enrollment.md index aa1aa57..a831d76 100644 --- a/docs/enrollment.md +++ b/docs/enrollment.md @@ -47,3 +47,10 @@ time without contacting the source server or requiring its access certificate to remain valid. Saved receipt recovery checks portable evidence at issuance; this does not permit an expired portable identity to start another enrollment. The public CLI continues to create fresh identities at this intermediate cut. + +The internal portable enrollment path now keeps a locally selected agent identity +through target preparation and atomic activation. Target approval binds the exact +registered issuer revision; removal and re-registration require a new approval. +Existing target bindings, including denied or inactive enrollments, require +owner-approved repair. Public registration and CLI source selection remain on the +fresh enrollment path until the final portable-enrollment cutover. diff --git a/server/news/+portable-enrollment-activation.feature.md b/server/news/+portable-enrollment-activation.feature.md new file mode 100644 index 0000000..8711360 --- /dev/null +++ b/server/news/+portable-enrollment-activation.feature.md @@ -0,0 +1 @@ +Add internal portable-identity preparation and atomic activation with exact issuer-revision approval, unique target bindings and unchanged receipt recovery. Public enrollment remains on the fresh flow pending the final client/server cutover. diff --git a/server/src/arbiter_server/enrollment.py b/server/src/arbiter_server/enrollment.py index 23a3a04..5b9a7a3 100644 --- a/server/src/arbiter_server/enrollment.py +++ b/server/src/arbiter_server/enrollment.py @@ -8,6 +8,7 @@ import hashlib import hmac import json +from pathlib import Path import re import threading import time @@ -214,6 +215,7 @@ class ApprovedRequest: request: PendingRequest grants: tuple[tuple[str, str], ...] secret: bytes = field(repr=False) + portable_issuer: tuple[bytes, int] | None = None class EnrollmentRequests: @@ -333,7 +335,7 @@ def check() -> None: def _summary(self, request: PendingRequest, now: float) -> dict[str, object]: advertisement = request.advertisement approved = self._approved.get(advertisement.selector) - return { + result: dict[str, object] = { "selector": advertisement.selector, "name": advertisement.name, "revision": request.revision, @@ -350,12 +352,24 @@ def _summary(self, request: PendingRequest, now: float) -> dict[str, object]: "remaining_seconds": max(0, int(request.deadline - now)), "grant_count": len(approved.grants) if approved else 0, } + if advertisement.portable_identity is not None: + claim = advertisement.portable_identity + result["portable_identity"] = { + "owner_id": claim.owner_id, + "agent_id": claim.agent_id, + "issuer_sha256": claim.issuer_sha256, + "certificate_validated": approved is not None, + "possession_proved": False, + } + return result def control( self, command: str, arguments: dict[str, object], accounts: set[tuple[str, str]], + *, + root: Path | None = None, ) -> dict[str, object]: """Called only through the OS-authenticated local control dispatcher.""" if command == "open": @@ -449,12 +463,28 @@ def control( selector = arguments["selector"] if not isinstance(selector, str): raise ValueError("Invalid enrollment selector.") + issuer = None + selected = None + if command == "approve": + with self._lock: + selected = self._requests.get(selector) + if selected is not None and selected.advertisement.portable_identity: + from .owner_issuers import approve_portable + + if root is None: + raise ValueError( + "Portable enrollment requires target issuer validation" + ) + # No request lock spans bounded database or certificate work. + issuer = approve_portable(root, selected.advertisement) with self._lock: now = self._clock() self._prune(now) request = self._requests.get(selector) if request is None: raise ValueError("Enrollment request is missing or expired.") + if command == "approve" and request is not selected: + raise ValueError("Enrollment selection changed; list requests again.") if ( request.revision != arguments["revision"] or request.advertisement.digest != arguments["request_digest"] @@ -471,12 +501,16 @@ def control( return {"initialized": True, "rejected": selector} existing = self._approved.get(selector) if existing is not None: - if existing.grants != grants or not hmac.compare_digest( - existing.secret, secret + if ( + existing.portable_issuer != issuer + or existing.grants != grants + or not hmac.compare_digest(existing.secret, secret) ): raise ValueError("Enrollment approval is immutable; reject first.") else: - self._approved[selector] = ApprovedRequest(request, grants, secret) + self._approved[selector] = ApprovedRequest( + request, grants, secret, issuer + ) return {"initialized": True, "request": self._summary(request, now)} def open_window(self) -> None: @@ -502,9 +536,13 @@ def pending(self) -> tuple[PendingRequest, ...]: self._prune(self._clock()) return tuple(self._requests.values()) - def register(self, raw: bytes | bytearray) -> PendingRequest: + def register(self, raw: bytes | bytearray, *, version: int = 1) -> PendingRequest: epoch, window_deadline = self.require_open() - advertisement = decode_advertisement(raw) + advertisement = ( + decode_advertisement(raw) + if version == 1 + else decode_advertisement(raw, version=version) + ) with self._lock: if advertisement.selector in self._committing: raise _invalid() diff --git a/server/src/arbiter_server/enrollment_activation.py b/server/src/arbiter_server/enrollment_activation.py index d6a117b..408aea0 100644 --- a/server/src/arbiter_server/enrollment_activation.py +++ b/server/src/arbiter_server/enrollment_activation.py @@ -2,7 +2,7 @@ from collections.abc import Callable from contextlib import closing -from datetime import datetime +from datetime import datetime, timezone import json import hashlib from pathlib import Path @@ -16,6 +16,7 @@ from .enrollment_receipt import result_fields from .generations import ACTIVE, _control, _database from .prepared import PREPARED_TAG +from .owner_issuers import portable_issuer def activate( @@ -48,8 +49,18 @@ def activate( raise ValueError("Enrollment activation unavailable") path = _database(root, active["active"]) authority = _read_record(path) - if ids[:2] != [authority.server_id, authority.owner_id]: + claim = ad.portable_identity + if ids[0] != authority.server_id or ( + claim is None and ids[1] != authority.owner_id + ): raise ValueError("Enrollment activation unavailable") + if claim is not None and ( + ids[1:3] != [claim.owner_id, claim.agent_id] + or fields[3] != claim.certificate_der + or approval.portable_issuer is None + or fields[2] != approval.portable_issuer[0] + ): + raise ValueError("Portable identity differs from approval") for index, identity, issuer, issuer_id, public in ( (3, Identity(Role.PORTABLE, ids[2]), fields[2], ids[1], ad.portable_spki), ( @@ -78,7 +89,8 @@ def activate( 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] + or claim is None + and fields[2] != stored[Role.OWNER.value] ): raise ValueError("Enrollment activation unavailable") db.execute( @@ -111,6 +123,15 @@ def activate( (hashlib.sha256(fields[2]).hexdigest(), ids[1], fields[2]), ).fetchone() != (1,): raise ValueError("Enrollment issuer unavailable") + if ( + claim is not None + and db.execute( + "SELECT 1 FROM agent_enrollments WHERE agent_id=?", (ids[2],) + ).fetchone() + ): + raise ValueError( + "Agent is already enrolled; use owner-approved repair" + ) db.execute( "INSERT INTO agent_enrollments VALUES (?, ?)", (ids[2], ids[3]) ) @@ -126,6 +147,13 @@ def activate( ) if not set(approval.grants) <= accounts(): raise ValueError("An approved account is no longer configured") + if ( + claim is not None + and existing is None + and portable_issuer(db, ad, datetime.now(timezone.utc)) + != approval.portable_issuer + ): + raise ValueError("Portable issuer approval is no longer current") # 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. diff --git a/server/src/arbiter_server/owner_issuers.py b/server/src/arbiter_server/owner_issuers.py index 0886802..8723f76 100644 --- a/server/src/arbiter_server/owner_issuers.py +++ b/server/src/arbiter_server/owner_issuers.py @@ -8,11 +8,57 @@ import sqlite3 from cryptography import x509 +from cryptography.hazmat.primitives import serialization from . import authority as a from .credentials import Identity, Role, parse_certificate, verify_certificate from .enrollment_receipt import MAX_RESULT, result_fields from .generations import ACTIVE, _control, _database +from .enrollment import Advertisement + + +def portable_issuer( + db: sqlite3.Connection, advertisement: Advertisement, now: datetime +) -> tuple[bytes, int]: + """Verify a claim against independently registered trust, not supplied trust.""" + claim = advertisement.portable_identity + if claim is None: + raise ValueError("Portable identity is missing") + row = db.execute( + "SELECT substr(certificate_der,1,4097), revision FROM owner_issuers " + "WHERE issuer_sha256=? AND owner_id=? AND active=1", + (claim.issuer_sha256, claim.owner_id), + ).fetchone() + if row is None or hashlib.sha256(row[0]).hexdigest() != claim.issuer_sha256: + raise ValueError("Portable issuer is not active and registered") + issuer, revision = row + verify_certificate(issuer, Identity(Role.OWNER, claim.owner_id), now=now) + certificate = verify_certificate( + claim.certificate_der, + Identity(Role.PORTABLE, claim.agent_id), + issuer_der=issuer, + issuer_id=claim.owner_id, + now=now, + ) + if ( + certificate.public_key().public_bytes( + serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo + ) + != advertisement.portable_spki + ): + raise ValueError("Portable key differs from the request") + return issuer, revision + + +def approve_portable(root: Path, advertisement: Advertisement) -> tuple[bytes, int]: + active = _control(root / ACTIVE) + if active is None: + raise ValueError("No live enrollment authority") + path = _database(root, active["active"]) + with closing( + sqlite3.connect(path.as_uri() + "?mode=ro", uri=True, timeout=0) + ) as db: + return portable_issuer(db, advertisement, datetime.now(timezone.utc)) def create_tables(db: sqlite3.Connection, record: a.AuthorityRecord) -> None: diff --git a/server/src/arbiter_server/prepared.py b/server/src/arbiter_server/prepared.py index 1d809f1..91ca1e1 100644 --- a/server/src/arbiter_server/prepared.py +++ b/server/src/arbiter_server/prepared.py @@ -28,6 +28,10 @@ def prepare(root: Path, approval: ApprovedRequest, expires_at: int) -> bytes: raise ValueError("No live enrollment authority") path = _database(root, active["active"]) record = _read_record(path) + ad = approval.request.advertisement + claim = ad.portable_identity + now = datetime.now(timezone.utc) + source_issuer = None with closing( sqlite3.connect(path.as_uri() + "?mode=ro", uri=True, timeout=0) ) as db: @@ -40,18 +44,40 @@ def prepare(root: Path, approval: ApprovedRequest, expires_at: int) -> bytes: recovery, origin = db.execute( "SELECT verifier, origin_verifier FROM recovery_trust WHERE singleton=1" ).fetchone() - now = datetime.now(timezone.utc) + if claim is not None: + from .owner_issuers import portable_issuer + + source_issuer = portable_issuer(db, ad, now) + if ( + source_issuer != approval.portable_issuer + or db.execute( + "SELECT 1 FROM agent_enrollments WHERE agent_id=?", + (claim.agent_id,), + ).fetchone() + ): + raise ValueError("Portable identity already enrolled or issuer changed") agent, enrollment = str(uuid4()), str(uuid4()) certificates = [] - ad = approval.request.advertisement - for role, identity, spki in ( - (Role.OWNER, Identity(Role.PORTABLE, agent), ad.portable_spki), + owner_id, owner_der = record.owner_id, rows[Role.OWNER.value][0] + if claim is not None: + assert source_issuer is not None + owner_id, agent = claim.owner_id, claim.agent_id + owner_der = source_issuer[0] + certificates.append(claim.certificate_der) + if len({record.server_id, owner_id, agent, enrollment}) != 4: + raise ValueError("Enrollment identity collision") + issuance = [ ( Role.SERVER_AUTHORITY, Identity(Role.ACCESS, agent, record.server_id, enrollment), ad.access_spki, ), - ): + ] + if claim is None: + issuance.insert( + 0, (Role.OWNER, Identity(Role.PORTABLE, agent), ad.portable_spki) + ) + for role, identity, spki in issuance: certificate, private = rows[role.value] key = serialization.load_der_private_key(private, password=None) public = serialization.load_der_public_key(spki) @@ -70,7 +96,7 @@ def prepare(root: Path, approval: ApprovedRequest, expires_at: int) -> bytes: # JSON arrays of strings, never an extensible object or executable content. result = ( PREPARED_TAG - + (record.server_id + record.owner_id + agent + enrollment).encode("ascii") + + (record.server_id + owner_id + agent + enrollment).encode("ascii") + ad.selector.encode("ascii") + bytes.fromhex(ad.digest) + approval.request.revision.encode("ascii") @@ -79,7 +105,7 @@ def prepare(root: Path, approval: ApprovedRequest, expires_at: int) -> bytes: for value in ( rows[Role.SERVER_AUTHORITY.value][0], rows[Role.SERVER_TLS.value][0], - rows[Role.OWNER.value][0], + owner_der, *certificates, ad.recipient_spki, recovery, diff --git a/server/tests/integration/test_enrollment_cli_peer.py b/server/tests/integration/test_enrollment_cli_peer.py index 7b1a98b..4626e40 100644 --- a/server/tests/integration/test_enrollment_cli_peer.py +++ b/server/tests/integration/test_enrollment_cli_peer.py @@ -336,3 +336,218 @@ def operator(): assert row[0] == 1 and bytes(row[1]) == saved_result[-1] assert db.execute("SELECT COUNT(*) FROM enrollments").fetchone() == (1,) assert not requests.pending() and not responder._attempts + + +@pytest.mark.parametrize("mode", ["ok", "lost-reply", "changed-portable"]) +def test_portable_python_go_enrollment( + tmp_path, private_windows_test_parent, mode, monkeypatch +): + from contextlib import ExitStack, closing + from functools import partial + import sqlite3 + from arbiter_server.owner_issuers import control + from arbiter_server.enrollment_receipt import result_fields + from server.tests.unit.test_enrollment_activation import database + + if shutil.which("go") is None: + pytest.skip("go is not installed") + apps, roots = [], [] + for name in ("a", "b"): + base = tmp_path / name + base.mkdir(mode=0o777 if os.name == "nt" else 0o700) + root = base / "server" / "authority" + initialize_authority( + root, + recovery_file=tmp_path / "operator.json", + reuse_recovery_file=name == "b", + ) + apps.append(cast(Starlette, _client(base)._app)) + roots.append(root) + # Exercise the internal v2 prefix deliberately. The product public parser + # remains v1 until the next cut; all other carrier/approval code is real. + requests = apps[1].state.enrollment_requests + monkeypatch.setattr(requests, "register", partial(requests.register, version=2)) + saved = [] + original_prepare = apps[1].state.pairing_responder.prepare + + async def prepare(session, root): + raw = await original_prepare(session, root) + saved.append(raw) + if mode == "changed-portable": + from cryptography.hazmat.primitives import serialization + from cryptography.hazmat.primitives.asymmetric import ec + from arbiter_server.credentials import ( + Identity, + Role, + SigningCredential, + issue_certificate, + ) + + ids, fields = result_fields(raw) + with closing(sqlite3.connect(database(roots[0]))) as db: + der, key = db.execute( + "SELECT certificate_der, private_key_der FROM credentials WHERE role='owner'" + ).fetchone() + private = serialization.load_der_private_key(key, None) + public = serialization.load_der_public_key( + session.approval.request.advertisement.portable_spki + ) + assert isinstance(private, ec.EllipticCurvePrivateKey) + assert isinstance(public, ec.EllipticCurvePublicKey) + issuer = SigningCredential(Identity(Role.OWNER, ids[1]), der, private) + fields[3] = issue_certificate( + Identity(Role.PORTABLE, ids[2]), + public, + issuer, + ) + assert fields[3] != result_fields(raw)[1][3] + fixed = len(PREPARED_TAG) + 144 + 22 + 32 + 32 + 8 + raw = raw[:fixed] + b"".join(len(f).to_bytes(2, "big") + f for f in fields) + return raw + + monkeypatch.setattr(apps[1].state.pairing_responder, "prepare", prepare) + original = apps[1].state.pairing_responder.activate + + async def activate(*args): + raw = await original(*args) + if mode == "lost-reply": + raise ValueError("synthetic dropped committed reply") + return raw + + monkeypatch.setattr(apps[1].state.pairing_responder, "activate", activate) + cert, key = tmp_path / "outer.crt", tmp_path / "outer.key" + _generate_self_signed_tls_certificate( + cert_path=cert, key_path=key, names=("127.0.0.1",) + ) + + def serve(stack, app, **kwargs): + return stack.enter_context( + live_server( + app, + ws=app.state.native_websocket_protocol, + http=app.state.native_http_protocol, + ssl_certfile=str(cert), + ssl_keyfile=str(key), + **kwargs, + ) + ) + + with ( + ExitStack() as a_stack, + ExitStack() as b_stack, + socket.socket() as reserved, + socket.socket() as listener, + ): + reserved.bind(("127.0.0.1", 0)) + ports = [serve(a_stack, apps[0]), reserved.getsockname()[1]] + apps[0].state.enrollment_requests.open_window() + listener.bind(("127.0.0.1", 0)) + listener.listen(1) + listener.settimeout(30) + + def operator(): + connection, _ = listener.accept() + with connection, connection.makefile("rwb") as stream: + connection.settimeout(30) + + def send(value): + stream.write(json.dumps(value).encode() + b"\n") + stream.flush() + + send( + dict( + A=f"https://127.0.0.1:{ports[0]}", + B=f"https://127.0.0.1:{ports[1]}", + Mode=mode, + ) + ) + for phase in (0, 1): + handoff = json.loads(stream.readline(2049)) + assert handoff["phase"] == phase + requests = apps[phase].state.enrollment_requests + pending = next( + r + for r in requests.pending() + if r.advertisement.selector == handoff["selector"] + ) + requests.control( + "approve", + dict( + selector=pending.advertisement.selector, + revision=pending.revision, + request_digest=pending.advertisement.digest, + psk=handoff["secret"], + grants=[], + ), + set(), + root=roots[phase], + ) + send(True) + signal = json.loads(stream.readline(65)) + if phase == 0: + assert signal == "source-ready" + with closing(sqlite3.connect(database(roots[0]))) as db: + source = db.execute( + "SELECT prepared FROM enrollments" + ).fetchone()[0] + _, fields = result_fields(source) + # Sequence lifetimes: shared harness logger patches must + # unwind in order. B starts only after A is fully offline. + a_stack.close() + reserved.close() + assert serve(b_stack, apps[1], bind_port=ports[1]) == ports[1] + apps[1].state.enrollment_requests.open_window() + control( + roots[1], + "register", + dict(certificate_der=base64.b64encode(fields[2]).decode()), + ) + else: + if mode == "changed-portable": + assert signal == "rejected" + send(True) + return + assert signal == "prepared" + b_stack.close() + restarted = cast(Starlette, _client(tmp_path / "b")._app) + assert serve(b_stack, restarted, bind_port=ports[1]) == ports[1] + assert restarted.state.enrollment_requests.pending() == () + send(True) + + with ThreadPoolExecutor(max_workers=1) as pool: + task = pool.submit(operator) + result = subprocess.run( + [ + "go", + "test", + "-count=1", + "-run", + "^TestPythonPortableEnrollmentPeer$", + "-v", + "./internal/identity", + ], + cwd=Path(__file__).parents[3] / "client" / "go-cli", + env={ + **os.environ, + "ARBITER_TEST_PORTABLE_ENROLLMENT_CONTROL": f"127.0.0.1:{listener.getsockname()[1]}", + }, + capture_output=True, + text=True, + timeout=90, + ) + assert result.returncode == 0, result.stdout + result.stderr + task.result(timeout=5) + assert "ARBITER_PORTABLE_ENROLLMENT=pass" in result.stdout + assert len(saved) == 1 + with closing(sqlite3.connect(database(roots[1]))) as db: + if mode == "changed-portable": + assert db.execute( + "SELECT COUNT(*) FROM agent_enrollments" + ).fetchone() == (0,) + return + assert db.execute("SELECT prepared FROM enrollments").fetchall() == [ + (saved[0],) + ] + assert db.execute("SELECT COUNT(*) FROM agent_enrollments").fetchone() == ( + 1, + ) diff --git a/server/tests/unit/test_enrollment_activation.py b/server/tests/unit/test_enrollment_activation.py index 973425e..991660f 100644 --- a/server/tests/unit/test_enrollment_activation.py +++ b/server/tests/unit/test_enrollment_activation.py @@ -665,3 +665,130 @@ def check(): ) == raw ) + + +@pytest.mark.parametrize( + "failure", [None, "removed", "re-added", "rollback", "active", "denied", "inactive"] +) +def test_portable_atomic_activation(tmp_path, private_windows_test_parent, failure): + from arbiter_server.owner_issuers import control + from server.tests.unit.test_prepared import portable_approval + + source, root, requests, approval, keys, now, old, _ = portable_approval(tmp_path) + raw = prepare(root, approval, int(now.timestamp()) + 300) + claim = approval.request.advertisement.portable_identity + assert claim is not None and approval.portable_issuer is not None + # Both valid results are prepared before either transaction owns the binding. + from uuid import uuid4 + from arbiter_server.enrollment import ApprovedRequest + + other_ad = replace( + approval.request.advertisement, + selector=base64.urlsafe_b64encode(os.urandom(16)).decode().rstrip("="), + digest=hashlib.sha256(os.urandom(32)).hexdigest(), + ) + other = ApprovedRequest( + replace(approval.request, advertisement=other_ad, revision=uuid4().hex), + (), + approval.secret, + approval.portable_issuer, + ) + competing = prepare(root, other, int(now.timestamp()) + 300) + if failure in {"removed", "re-added"}: + control(root, "remove", {"issuer_sha256": claim.issuer_sha256}) + if failure == "re-added": + control( + root, + "register", + { + "certificate_der": base64.b64encode( + approval.portable_issuer[0] + ).decode() + }, + ) + + def check(): + if failure == "rollback": + raise ValueError("synthetic live commit failure") + + if failure in {"removed", "re-added", "rollback"}: + with pytest.raises(ValueError): + activate(root, approval, raw, set, check, now) + with sqlite3.connect(database(root)) as db: + assert db.execute("SELECT COUNT(*) FROM agent_enrollments").fetchone() == ( + 0, + ) + return + assert activate(root, approval, raw, set, check, now) == raw + assert activate(root, approval, raw, set, check, now) == raw + if failure in {"active", "denied", "inactive"}: + with sqlite3.connect(database(root)) as db: + if failure == "denied": + db.execute("UPDATE enrollments SET denied=1") + if failure == "inactive": + db.execute("UPDATE enrollments SET inactive_after=0") + db.commit() + with pytest.raises(ValueError): + prepare(root, approval, int(now.timestamp()) + 300) + with pytest.raises(ValueError): + activate(root, other, competing, set, check, now) + else: + requests.reset() + exporter = os.urandom(32) + assert ( + retrieve( + root, + request_for(raw, exporter, keys[0]), + exporter, + now + timedelta(minutes=10), + ) + == raw + ) + with sqlite3.connect(database(root)) as db: + assert db.execute("SELECT COUNT(*) FROM enrollments").fetchone() == (1,) + assert db.execute("SELECT COUNT(*) FROM agent_enrollments").fetchone() == (1,) + + +@pytest.mark.parametrize("expired", [False, True]) +def test_portable_expiry_during_accounts_rolls_back( + tmp_path, private_windows_test_parent, monkeypatch, expired +): + from cryptography import x509 + import arbiter_server.enrollment_activation as activation + from server.tests.unit.test_prepared import portable_approval + + _, root, _, approval, _, now, _, _ = portable_approval(tmp_path) + raw = prepare(root, approval, int(now.timestamp()) + 300) + claim = approval.request.advertisement.portable_identity + assert claim is not None + expiry = x509.load_der_x509_certificate(claim.certificate_der).not_valid_after_utc + evaluated = False + + class Clock: + @staticmethod + def now(zone): + assert evaluated and zone is timezone.utc + return (expiry if expired else now) + timedelta(seconds=1) + + def accounts(): + nonlocal evaluated + evaluated = True + return set() + + monkeypatch.setattr(activation, "datetime", Clock) + if expired: + with pytest.raises(ValueError): + activate(root, approval, raw, accounts, lambda: None, now) + else: + assert activate(root, approval, raw, accounts, lambda: None, now) == raw + assert evaluated + with sqlite3.connect(database(root)) as db: + assert db.execute("SELECT COUNT(*) FROM agent_enrollments").fetchone() == ( + 0 if expired else 1, + ) + if expired: + assert not db.execute( + "SELECT name FROM sqlite_master WHERE name='enrollments'" + ).fetchall() + else: + assert db.execute("SELECT prepared FROM enrollments").fetchall() == [(raw,)] diff --git a/server/tests/unit/test_prepared.py b/server/tests/unit/test_prepared.py index 07dbada..68d3bfb 100644 --- a/server/tests/unit/test_prepared.py +++ b/server/tests/unit/test_prepared.py @@ -5,6 +5,7 @@ import hashlib import sqlite3 import threading +from typing import Any, cast import pytest from cryptography.exceptions import InvalidSignature @@ -221,3 +222,179 @@ def delayed_read(path): responder.close() asyncio.run(run()) + + +def portable_approval(tmp_path): + """Real fresh-v1 activation at A, then an independently approved B request.""" + import base64 + import json + from arbiter_server.enrollment import EnrollmentRequests + from arbiter_server.enrollment_activation import activate + from arbiter_server.enrollment_receipt import result_fields + from arbiter_server.owner_issuers import control + from arbiter_server.prepared import prepare + from server.tests.unit.test_enrollment_requests import advertisement + + a, b = tmp_path / "a", tmp_path / "b" + for root in (a, b): + initialize_authority( + root, + recovery_file=tmp_path / "operator.json", + reuse_recovery_file=root == b, + ) + _, fresh, old_keys, _, _, now = proof_setup() + raw = prepare(a, fresh, int(now.timestamp()) + 300) + activate(a, fresh, raw, set, lambda: None, now) + ids, fields = result_fields(raw) + fingerprint = hashlib.sha256(fields[2]).hexdigest() + control(b, "register", {"certificate_der": base64.b64encode(fields[2]).decode()}) + data = advertisement() + _, fresh_b, keys, _, _, _ = proof_setup() + keys[1] = old_keys[1] + data.update( + version=2, + portable_identity=dict( + owner_id=ids[1], + agent_id=ids[2], + issuer_sha256=fingerprint, + certificate_der=base64.b64encode(fields[3]).decode(), + ), + ) + from cryptography.hazmat.primitives import serialization + + for name, key in zip(("access_spki", "portable_spki"), keys): + data[name] = base64.b64encode( + key.public_key().public_bytes( + serialization.Encoding.DER, + serialization.PublicFormat.SubjectPublicKeyInfo, + ) + ).decode() + requests = EnrollmentRequests() + requests.open_window() + pending = requests.register(json.dumps(data).encode(), version=2) + arguments: dict[str, object] = dict( + selector=pending.advertisement.selector, + revision=pending.revision, + request_digest=pending.advertisement.digest, + psk=base64.urlsafe_b64encode(fresh_b.secret).decode().rstrip("="), + grants=[], + ) + requests.control("approve", arguments, set(), root=b) + approval = requests.approved(pending.advertisement.selector) + assert approval is not None + return a, b, requests, approval, keys, now, raw, arguments + + +@pytest.mark.parametrize( + "failure", + [None, "unknown", "inactive", "re-added", "wrong-key", "expired", "binding"], +) +def test_portable_prepare_rechecks_exact_trust( + tmp_path, private_windows_test_parent, failure, monkeypatch +): + from dataclasses import replace + from arbiter_server.enrollment_receipt import result_fields + from arbiter_server.owner_issuers import control + from arbiter_server.prepared import prepare + from server.tests.unit.test_enrollment_activation import database + + _, root, requests, approval, _, now, source, arguments = portable_approval(tmp_path) + claim = approval.request.advertisement.portable_identity + assert claim is not None and approval.portable_issuer is not None + if failure in {"unknown", "inactive", "re-added"}: + with sqlite3.connect(database(root)) as db: + if failure == "unknown": + db.execute( + "DELETE FROM owner_issuers WHERE issuer_sha256=?", + (claim.issuer_sha256,), + ) + else: + db.execute( + "UPDATE owner_issuers SET active=0,revision=revision+1 WHERE issuer_sha256=?", + (claim.issuer_sha256,), + ) + db.commit() + if failure == "re-added": + import base64 + + control( + root, + "register", + { + "certificate_der": base64.b64encode( + approval.portable_issuer[0] + ).decode() + }, + ) + if failure == "wrong-key": + approval = replace( + approval, + request=replace( + approval.request, + advertisement=replace( + approval.request.advertisement, + portable_spki=approval.request.advertisement.access_spki, + ), + ), + ) + if failure == "expired": + from arbiter_server import prepared + + class Clock: + @staticmethod + def now(tz): + return now + timedelta(days=366) + + monkeypatch.setattr(prepared, "datetime", Clock) + if failure == "binding": + with sqlite3.connect(database(root)) as db: + db.execute( + "INSERT INTO agent_enrollments VALUES (?, ?)", + (claim.agent_id, "occupied"), + ) + db.commit() + if failure: + with pytest.raises(ValueError): + prepare(root, approval, int(now.timestamp()) + 366 * 86400 + 300) + return + raw = prepare(root, approval, int(now.timestamp()) + 300) + ids, fields = result_fields(raw) + old_ids, old_fields = result_fields(source) + assert ids[1:3] == old_ids[1:3] and ids[0] != old_ids[0] and ids[3] != old_ids[3] + assert fields[2:4] == old_fields[2:4] and fields[4] != old_fields[4] + assert cast(Any, requests.control("list", {}, set())["requests"])[0][ + "portable_identity" + ] == dict( + owner_id=ids[1], + agent_id=ids[2], + issuer_sha256=claim.issuer_sha256, + certificate_validated=True, + possession_proved=False, + ) + assert ( + cast(Any, requests.control("approve", arguments, set(), root=root)["request"])[ + "status" + ] + == "approved" + ) + with pytest.raises(ValueError): + requests.control("approve", arguments, set()) + + +def test_portable_approval_reset_during_issuer_validation( + tmp_path, private_windows_test_parent, monkeypatch +): + from arbiter_server import owner_issuers + + _, root, requests, _, _, _, _, arguments = portable_approval(tmp_path) + original = owner_issuers.approve_portable + + def reset(*args): + issuer = original(*args) + requests.reset() + return issuer + + monkeypatch.setattr(owner_issuers, "approve_portable", reset) + with pytest.raises(ValueError, match="missing or expired"): + requests.control("approve", arguments, set(), root=root) + assert not requests.pending()