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()