From a255811c1aa215fba7cca7592f44b3d57c2e9476 Mon Sep 17 00:00:00 2001 From: Omry Yadan Date: Tue, 27 Oct 2026 19:32:11 +0800 Subject: [PATCH] [authentication/2.9] Pairing installation and overload closure Exercise the installed native client through its real terminal handoff and authenticated same-selector recovery. Cover lost activation replies, local credential promotion failure, opaque proxy split writes, and concurrent expiry/shutdown with two real workers. Include the installed journeys in native interoperability CI with the existing smoke output directory and fresh binary builds. Correct the guide to describe ready, conditional activation and receipt while preserving the ordinary-API milestone boundary. Local validation: type checks, Black, 1,611 unit tests, uncached Go tests and vet, website build, and 52 Python/Go integration tests passed. Effective-CI installed acceptance executed all three cases with zero skips. Native six-platform CI remains required before approval. --- .github/workflows/ci.yml | 3 +- server/news/74.misc.md | 2 + .../integration/enrollment_terminal_driver.py | 163 +++++++++++++ .../integration/test_installed_enrollment.py | 225 ++++++++++++++++++ server/tests/unit/test_pairing_carrier.py | 83 +++++++ website/docs/operate/credentials.md | 18 +- 6 files changed, 488 insertions(+), 6 deletions(-) create mode 100644 server/news/74.misc.md create mode 100644 server/tests/integration/enrollment_terminal_driver.py create mode 100644 server/tests/integration/test_installed_enrollment.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f35b52c..d577182 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -159,8 +159,9 @@ jobs: - name: Verify fresh Python-Go credential interoperability if: ${{ matrix.go_tests == true }} env: + ARBITER_GO_CLIENT_SMOKE_OUTDIR: .ci/go-client-smoke GOCACHE: ${{ runner.temp }}/go-build - run: nox -s unit -- server/tests/integration/test_credential_peer.py server/tests/integration/test_provisional_peer.py server/tests/integration/test_pairing_peer.py server/tests/integration/test_enrollment_cli_peer.py server/tests/integration/test_control_tls_peer.py server/tests/integration/test_enrollment_receipt_peer.py server/tests/integration/test_activation_restart_peer.py + run: nox -s unit -- server/tests/integration/test_credential_peer.py server/tests/integration/test_provisional_peer.py server/tests/integration/test_pairing_peer.py server/tests/integration/test_enrollment_cli_peer.py server/tests/integration/test_control_tls_peer.py server/tests/integration/test_enrollment_receipt_peer.py server/tests/integration/test_activation_restart_peer.py server/tests/integration/test_installed_enrollment.py - name: Verify native local-control rejection across OS users if: ${{ startsWith(matrix.platform_id, 'linux-') || startsWith(matrix.platform_id, 'macos-') }} diff --git a/server/news/74.misc.md b/server/news/74.misc.md new file mode 100644 index 0000000..859c635 --- /dev/null +++ b/server/news/74.misc.md @@ -0,0 +1,2 @@ +Add installed native-client pairing and receipt-recovery acceptance coverage, +including fragmented proxy writes and concurrent expiry/shutdown ownership. diff --git a/server/tests/integration/enrollment_terminal_driver.py b/server/tests/integration/enrollment_terminal_driver.py new file mode 100644 index 0000000..a75521e --- /dev/null +++ b/server/tests/integration/enrollment_terminal_driver.py @@ -0,0 +1,163 @@ +"""Test-only human terminal driver for an unmodified installed executable.""" + +import json +import os +import re +import select +import socket +import struct +import subprocess +import sys +import time +from typing import Any + + +def run(): + config = json.loads(sys.stdin.readline()) + master = slave = input_fd = output_fd = -1 + terminal_options: dict[str, Any] + if sys.platform == "win32": + import ctypes + from ctypes import wintypes + import msvcrt + + kernel = ctypes.WinDLL("kernel32", use_last_error=True) + read = kernel.ReadConsoleOutputCharacterW + read.argtypes = [ + wintypes.HANDLE, + wintypes.LPWSTR, + wintypes.DWORD, + wintypes.DWORD, + ctypes.POINTER(wintypes.DWORD), + ] + write = kernel.WriteConsoleInputW + write.argtypes = [ + wintypes.HANDLE, + wintypes.LPVOID, + wintypes.DWORD, + ctypes.POINTER(wintypes.DWORD), + ] + input_fd = os.open("CONIN$", os.O_RDWR) + output_fd = os.open("CONOUT$", os.O_RDWR) + + def terminal_read(): + buffer = ctypes.create_unicode_buffer(8192) + size = wintypes.DWORD() + if not read( + msvcrt.get_osfhandle(output_fd), buffer, 8192, 0, ctypes.byref(size) + ): + raise OSError("could not inspect test console") + return buffer.value.encode() + + def enter(): + # INPUT_RECORD(KEY_EVENT): down/up, one RETURN, no secret input. + records = b"".join( + struct.pack(" 32768: + raise RuntimeError("unexpected terminal output volume") + match = re.search( + rb"Request: ([A-Za-z0-9_-]{22})\s+Secret[^:]*: ([A-Za-z0-9_-]{43})", + terminal, + ) + if ( + not handed_off + and match + and b"press Enter to continue: " in terminal + ): + selector, secret = (part.decode() for part in match.groups()) + stream.write( + json.dumps(dict(selector=selector, secret=secret)).encode() + + b"\n" + ) + stream.flush() + if stream.readline(16) != b"true\n": + raise RuntimeError("operator did not approve") + enter() + handed_off = True + time.sleep(0.01) + if process.poll() is None: + raise TimeoutError("installed enrollment did not settle") + stdout, stderr = process.communicate(timeout=5) + if not handed_off or secret.encode() in stdout + stderr: + raise RuntimeError("invalid protected terminal handoff") + confirmed = ( + process.returncode == 0 and b"Enrollment activated and saved" in stdout + ) + if not confirmed and not ( + config["recover"] + and process.returncode == 1 + and ("resume=" + selector).encode() in stderr + ): + raise RuntimeError( + "installed enrollment did not confirm or retain recovery" + ) + # Only public completion metadata leaves the synthetic human driver. + print(json.dumps(dict(selector=selector, activated=confirmed))) + finally: + if process is not None and process.poll() is None: + process.kill() + process.wait(timeout=5) + for fd in (input_fd, output_fd, master, slave): + if fd >= 0: + os.close(fd) + + +if __name__ == "__main__": + run() diff --git a/server/tests/integration/test_installed_enrollment.py b/server/tests/integration/test_installed_enrollment.py new file mode 100644 index 0000000..0e2b5a7 --- /dev/null +++ b/server/tests/integration/test_installed_enrollment.py @@ -0,0 +1,225 @@ +"""Installed platform wheel, real terminal handoff and opaque proxy writes.""" + +import base64 +from concurrent.futures import ThreadPoolExecutor +from contextlib import closing, contextmanager +import json +from pathlib import Path +import select +import socket +import socketserver +import sqlite3 +import ssl +import subprocess +import sys +import threading +import urllib.request +from typing import cast + +import pytest +from starlette.applications import Starlette + +from arbiter_server.authority import initialize_authority +from arbiter_server.main import _generate_self_signed_tls_certificate +from server.tests.integration.test_cli_entrypoint import ( + ClientCommand, + current_go_client_binary, + packaged_arbiter_client_command, +) +from server.tests.unit.conftest import private_windows_test_parent, tmp_path +from server.tests.unit.test_artifact_http_admission import live_server +from server.tests.unit.test_enrollment_activation import database +from server.tests.unit.test_native_http_protocol import _client + + +@contextmanager +def fragmented_passthrough(port): + stopped = threading.Event() + writes = [] + + class Relay(socketserver.BaseRequestHandler): + def handle(self): + self.request.settimeout(5) + with socket.create_connection(("127.0.0.1", port), timeout=5) as upstream: + peers = (self.request, upstream) + try: + while not stopped.is_set(): + ready, _, _ = select.select(peers, [], [], 0.1) + for source in ready: + data = source.recv(2048) + if not data: + return + destination = peers[1] if source is peers[0] else peers[0] + # The relay sees outer TLS ciphertext only. These + # split writes do not claim fixed TCP packet sizes. + for offset in range(0, len(data), 97): + piece = data[offset : offset + 97] + destination.sendall(piece) + writes.append((source is peers[0], len(piece))) + except (ConnectionError, OSError): + return # Peer cancellation is normal during lost replies. + + with socketserver.ThreadingTCPServer(("127.0.0.1", 0), Relay) as proxy: + proxy.daemon_threads = False + thread = threading.Thread(target=proxy.serve_forever) + thread.start() + try: + yield proxy.server_address[1], writes + finally: + stopped.set() + proxy.shutdown() + thread.join(timeout=5) + assert not thread.is_alive() + + +@pytest.mark.parametrize("mode", ["ok", "lost-reply", "promotion-failure"]) +def test_installed_enrollment_and_resume( + tmp_path, + private_windows_test_parent, + packaged_arbiter_client_command: ClientCommand, + monkeypatch, + mode, +): + root = tmp_path / "server" / "authority" + initialize_authority(root, recovery_file=tmp_path / "operator.json") + app = cast(Starlette, _client(tmp_path)._app) + responder, requests = app.state.pairing_responder, app.state.enrollment_requests + original_activate = responder.activate + committed = [] + + async def activate(*args): + receipt = await original_activate(*args) + committed.append(True) + if mode == "lost-reply": + raise ValueError("synthetic loss after actual activation commit") + return receipt + + monkeypatch.setattr(responder, "activate", activate) + certificate, key = tmp_path / "outer.crt", tmp_path / "outer.key" + _generate_self_signed_tls_certificate( + cert_path=certificate, + key_path=key, + names=("127.0.0.1",), + ) + directory = tmp_path / "installed-custody" + directory.mkdir(mode=0o700) + marker = directory / "enrolled" + if mode == "promotion-failure": + marker.write_text("preserve unrelated existing file") + with ( + live_server( + app, + ws=app.state.native_websocket_protocol, + http=app.state.native_http_protocol, + ssl_certfile=str(certificate), + ssl_keyfile=str(key), + ) as port, + fragmented_passthrough(port) as (proxy_port, writes), + ): + requests.open_window() + with socket.socket() as listener: + listener.bind(("127.0.0.1", 0)) + listener.listen(1) + listener.settimeout(30) + + def approve(): + connection, _ = listener.accept() + with connection, connection.makefile("rwb") as stream: + connection.settimeout(20) + handoff = json.loads(stream.readline(1025)) + pending = next( + p + for p in requests.pending() + if p.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(), + ) + stream.write(b"true\n") + stream.flush() + + config = dict( + command=str(packaged_arbiter_client_command.command), + directory=str(directory), + url=f"https://127.0.0.1:{proxy_port}", + control=listener.getsockname()[1], + recover=mode == "promotion-failure", + ) + with ThreadPoolExecutor(max_workers=1) as pool: + owner = pool.submit(approve) + creationflags = 0 + if sys.platform == "win32": + creationflags = subprocess.CREATE_NEW_CONSOLE + result = subprocess.run( + [ + sys.executable, + str(Path(__file__).with_name("enrollment_terminal_driver.py")), + ], + input=json.dumps(config) + "\n", + text=True, + capture_output=True, + timeout=65, + creationflags=creationflags, + ) + assert ( + result.returncode == 0 + ), "installed native terminal journey failed" + owner.result(timeout=5) + completion = json.loads(result.stdout) + selector = completion["selector"] + assert completion["activated"] == (mode != "promotion-failure") + assert committed == [True] + if mode == "promotion-failure": + assert marker.read_text() == "preserve unrelated existing file" + before = (directory / "prepared" / (selector + ".json")).read_bytes() + marker.unlink() # Explicit operator repair of this test-owned collision. + else: + before = (directory / "enrolled" / (selector + ".json")).read_bytes() + for _ in range(2): + resumed = subprocess.run( + [ + config["command"], + "--config-dir", + str(directory), + "arbiter.connect_timeout=15s", + "enrol", + "resume=" + selector, + ], + stdin=subprocess.DEVNULL, + capture_output=True, + text=True, + timeout=20, + ) + assert resumed.returncode == 0, "installed same-selector recovery failed" + assert "Enrollment activated and saved" in resumed.stdout + enrolled = directory / "enrolled" / (selector + ".json") + assert enrolled.read_bytes() == before + assert list((directory / "enrolled").glob("*.json")) == [enrolled] + assert not list((directory / "prepared").glob("*.json")) + saved = json.loads(before) + assert set(saved["provisional"]) >= { + "access_key_pkcs8", + "portable_key_pkcs8", + "recipient_key_pkcs8", + } + with closing(sqlite3.connect(database(root))) as db: + rows = db.execute("SELECT prepared FROM enrollments").fetchall() + assert len(rows) == 1 + assert bytes(rows[0][0]) == base64.b64decode(saved["result"], validate=True) + assert not requests.pending() and not responder._attempts + assert writes and {direction for direction, _ in writes} == {False, True} + assert all(0 < size <= 97 for _, size in writes) + with urllib.request.urlopen( + f"https://127.0.0.1:{port}/_health_", + context=ssl._create_unverified_context(), + timeout=5, + ) as health: + assert health.status == 200 and json.load(health) == {"status": "ok"} diff --git a/server/tests/unit/test_pairing_carrier.py b/server/tests/unit/test_pairing_carrier.py index 77d0fa1..9727645 100644 --- a/server/tests/unit/test_pairing_carrier.py +++ b/server/tests/unit/test_pairing_carrier.py @@ -473,6 +473,89 @@ def test_real_server_shutdown_closes_idle_native_carrier(server): peer.close() +def test_simultaneous_expiry_shutdown_retains_two_real_workers(server, monkeypatch): + from noise.connection import NoiseConnection + + entered = threading.Barrier(3) + release, stopped = threading.Event(), threading.Event() + original_read = NoiseConnection.read_message + original_connection = PairingProtocol.connection_made + carriers = [] + + def blocked(noise, data): + if threading.current_thread().name == "arbiter-operation": + entered.wait(timeout=10) + assert release.wait(20) + return original_read(noise, data) + + def connected(protocol, transport): + original_connection(protocol, transport) + if not protocol.closed: + carriers.append(protocol) + + monkeypatch.setattr(NoiseConnection, "read_message", blocked) + monkeypatch.setattr(PairingProtocol, "connection_made", connected) + peers = [] + with server(max_connections=2, max_handshakes=2) as (app, port, first, path): + requests, responder = app.state.enrollment_requests, app.state.pairing_responder + pending = requests.register(json.dumps(advertisement()).encode()) + requests.control( + "approve", + dict( + selector=pending.advertisement.selector, + revision=pending.revision, + request_digest=pending.advertisement.digest, + psk=base64.urlsafe_b64encode(os.urandom(32)).decode().rstrip("="), + grants=[], + ), + set(), + ) + second = requests.approved(pending.advertisement.selector) + try: + for context in (first, second): + peer = Peer( + port, + PAIRING_PATH + context.request.advertisement.selector + "/pair", + ) + peers.append(peer) + assert isinstance(peer.next()[0], events.AcceptConnection) + _, flight = initiator(context) + peer.send(events.BytesMessage(data=flight)) + entered.wait(timeout=10) + with Peer(port, path) as excess: + rejected = excess.next()[0] + assert isinstance(rejected, events.RejectConnection) + assert rejected.status_code == 503 + assert wire_request(port, "GET", "/_health_")[0] == 200 + + def expire_and_stop(): + requests._clock = lambda: pending.deadline + 1 + assert not requests.pending() + for carrier in carriers: + carrier.shutdown() + responder.close() + stopped.set() + + carriers[0].loop.call_soon_threadsafe(expire_and_stop) + assert stopped.wait(5) + assert len(responder._leases) == len(responder._handshakes) == 2 + assert len(responder._pool._pending) == 2 + assert all(lease._busy for lease in responder._leases) + assert wire_request(port, "GET", "/_health_")[0] == 200 + release.set() + deadline = time.monotonic() + 5 + while responder._leases or responder._handshakes: + assert time.monotonic() < deadline + time.sleep(0.001) + assert not responder._attempts and not requests.pending() + assert not responder._pool._pending + assert wire_request(port, "GET", "/_health_")[0] == 200 + finally: + release.set() + for peer in peers: + peer.close() + + def test_negative_native_framing_settings_are_rejected(): for name in ( "max_headers_bytes", diff --git a/website/docs/operate/credentials.md b/website/docs/operate/credentials.md index 09a71f6..9a7ed79 100644 --- a/website/docs/operate/credentials.md +++ b/website/docs/operate/credentials.md @@ -273,6 +273,12 @@ commit prevents confirmation. If the server never committed, ask the operator an explicitly start a new pairing. Do not restart pairing merely to recover a lost receipt: new pairing discards abandoned prepared records. +Use the installed `arbiter` command for both enrollment and recovery. New enrollment +opens the native terminal even when ordinary output is redirected. Run it from a +terminal you control; `resume=SELECTOR` can run without a terminal. A network proxy +must preserve the binary WebSocket stream and permit the separate receipt-control +connection. Interrupted connections do not justify deleting saved private state. + ### Public enrollment request boundary `POST /api/v1/enrollment/requests` accepts version-one provisional public @@ -415,15 +421,17 @@ fresh HTTP connection; an upgrade that bypasses this guard is rejected. Ordinary HTTP body, operation and download budgets remain separate. This endpoint completes the encrypted handshake, mutual signing-key proofs, -prepared-result delivery and ready verification before closing. Fresh Python/Go -peers exercise the public CLI/carrier and private prepared persistence. -Activation remains a subsequent delivery step. Local approval, prepared -certificates and ready verification still grant no ordinary API access. +prepared-result delivery and ready verification. After the client durably saves +the prepared result, ready verification permits the server to recheck approval +and commit activation. The endpoint returns the recorded receipt before closing. +Fresh Python/Go peers exercise the installed CLI/carrier and private prepared +persistence. Local approval or prepared certificates alone do not activate an +enrollment, and this milestone provides no ordinary authenticated API access. Registration uses the existing bounded worker admission before reading body bytes. The body limit is the smaller of the core request-body limit and 64 KiB, with a ten-second total body-read budget. Full capacity returns a retryable -`resource_exhausted` error. None of these public requests activate an identity. +`resource_exhausted` error. Registration alone does not activate an identity. ### Pairing possession proof profile