From 6b7a75838a3fc92378809017bd492a4f5aef21fc Mon Sep 17 00:00:00 2001 From: Omry Yadan Date: Wed, 21 Oct 2026 10:37:49 +0800 Subject: [PATCH] [authentication/1.5] Serve authenticated POSIX local control Allow the existing authority init and status commands to reach the running POSIX server through a private Unix socket. Authenticate both peers with kernel credentials, bind the endpoint to the authority directory, and keep live mutations in the process that owns the instance lock. Support the same commands through verified SSH as the service account. Bound control frames, connections and administrative work. Retain accepted descriptors, transports and actual authority work through disconnects and cancellation; join them before releasing the instance lock. Retry established transient accept errors without a work queue and report unexpected listener failure. Document setup, recovery custody, endpoint failures and interrupted replies. Validation: 1304 unit tests passed on native Python and Python 3.10 with 14 platform skips each; Black/Pyrefly, Go tests/vet and website build passed. Real CLI, malformed and oversized frames, slow clients, endpoint replacement, cancellation and injected accept failures are covered. Native Linux and macOS CI must run the real different-user rejection test on the published head. --- .github/workflows/ci.yml | 4 + server/news/+posix-local-control.feature.md | 7 + server/src/arbiter_server/authority.py | 86 ++-- server/src/arbiter_server/local_control.py | 394 ++++++++++++++++ server/src/arbiter_server/main.py | 10 +- server/tests/unit/conftest.py | 13 + server/tests/unit/test_local_control.py | 493 ++++++++++++++++++++ website/docs/operate/credentials.md | 21 + website/docs/operate/security.md | 30 +- 9 files changed, 1016 insertions(+), 42 deletions(-) create mode 100644 server/news/+posix-local-control.feature.md create mode 100644 server/src/arbiter_server/local_control.py create mode 100644 server/tests/unit/test_local_control.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1d062a0..6014b62 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -154,6 +154,10 @@ jobs: GOCACHE: ${{ runner.temp }}/go-build run: nox -s ${{ matrix.nox_session }} + - name: Verify native local-control rejection across OS users + if: ${{ startsWith(matrix.platform_id, 'linux-') || startsWith(matrix.platform_id, 'macos-') }} + run: sudo .nox/unit/bin/python -m pytest server/tests/unit/test_local_control.py::test_real_different_os_user_is_rejected -q + website: name: Website runs-on: ubuntu-latest diff --git a/server/news/+posix-local-control.feature.md b/server/news/+posix-local-control.feature.md new file mode 100644 index 0000000..36431b6 --- /dev/null +++ b/server/news/+posix-local-control.feature.md @@ -0,0 +1,7 @@ +On Linux and macOS, authority initialization and status can reach the running +server through a Unix socket inside the private authority directory, authenticated by +kernel peer credentials in both directions. Temporary contention without a live +endpoint retains the busy diagnostic; unsafe endpoints still fail explicitly. Existing commands retain stopped-instance operation. Control messages, +connections and work are bounded, and shutdown keeps the instance lock until +actual administrative work finishes. Run commands as the server's OS account; +loopback HTTP and socket-path knowledge do not grant administrative authority. diff --git a/server/src/arbiter_server/authority.py b/server/src/arbiter_server/authority.py index 9ec12fb..92e51f3 100644 --- a/server/src/arbiter_server/authority.py +++ b/server/src/arbiter_server/authority.py @@ -26,6 +26,10 @@ class AuthorityError(ValueError): """An offline authority operation could not complete safely.""" +class AuthorityBusy(AuthorityError): + """The running instance must handle this administrative operation.""" + + @dataclass(frozen=True) class AuthorityRecord: owner_id: str @@ -233,7 +237,7 @@ def unlock() -> None: try: lock() except OSError as exc: - raise AuthorityError("authority state is busy; try again later") from exc + raise AuthorityBusy("authority state is busy; try again later") from exc try: yield finally: @@ -311,31 +315,33 @@ def initialize_authority( """Create one private set of IDs, refusing to replace any existing database.""" root = _root(state_dir, create=True) with _locked(root): - from . import generations + return _initialize_locked(root, recovery_file, reuse_recovery_file) - if generations.reconcile(root) is not None: - raise AuthorityError( - "authority database already exists; inspect authority status" - ) - if recovery_file is None: - raise AuthorityError( - "select an external private --recovery-file destination" - ) - from .recovery import register_and_publish, setup_file - with setup_file(recovery_file, root, reuse=reuse_recovery_file) as ( - path, - material, - ): - record = AuthorityRecord( - material.owner_id, str(uuid4()), datetime.now(timezone.utc).isoformat() - ) - with generations.staging(root) as staged: - _write_database(staged / _PENDING, record) - register_and_publish(path, record, material, staged / _PENDING) - _publish(staged / _PENDING, staged / DATABASE) - generations.activate(root, staged) - return record +def _initialize_locked( + root: Path, recovery_file: Path | None, reuse_recovery_file: bool +) -> AuthorityRecord: + """Run only while this process owns the persistent instance lock.""" + from . import generations + + if generations.reconcile(root) is not None: + raise AuthorityError( + "authority database already exists; inspect authority status" + ) + if recovery_file is None: + raise AuthorityError("select an external private --recovery-file destination") + from .recovery import register_and_publish, setup_file + + with setup_file(recovery_file, root, reuse=reuse_recovery_file) as (path, material): + record = AuthorityRecord( + material.owner_id, str(uuid4()), datetime.now(timezone.utc).isoformat() + ) + with generations.staging(root) as staged: + _write_database(staged / _PENDING, record) + register_and_publish(path, record, material, staged / _PENDING) + _publish(staged / _PENDING, staged / DATABASE) + generations.activate(root, staged) + return record def authority_status(state_dir: Path) -> AuthorityRecord | None: @@ -422,25 +428,35 @@ def authority_cli( import json try: - record = ( - initialize_authority( - state_dir, - recovery_file=recovery_file, - reuse_recovery_file=reuse_recovery_file, + try: + record = ( + initialize_authority( + state_dir, + recovery_file=recovery_file, + reuse_recovery_file=reuse_recovery_file, + ) + if action == "init" + else authority_status(state_dir) ) - if action == "init" - else authority_status(state_dir) - ) + payload = record.to_dict() if record is not None else {"initialized": False} + except AuthorityBusy as busy: + if os.name == "nt": + raise + from .local_control import request + + try: + payload = request(state_dir, action, recovery_file, reuse_recovery_file) + except (FileNotFoundError, ConnectionRefusedError) as exc: + raise busy from exc except (OSError, ValueError, sqlite3.Error) as exc: print_cli_error(str(exc), area="authority") return 1 - payload = record.to_dict() if record is not None else {"initialized": False} if as_json: print(json.dumps(payload, sort_keys=True)) - elif record is None: + elif not payload["initialized"]: print("authority: not initialized") else: print( - f"authority: initialized\nowner: {record.owner_id}\nserver: {record.server_id}" + f"authority: initialized\nowner: {payload['owner_id']}\nserver: {payload['server_id']}" ) return 0 diff --git a/server/src/arbiter_server/local_control.py b/server/src/arbiter_server/local_control.py new file mode 100644 index 0000000..413daff --- /dev/null +++ b/server/src/arbiter_server/local_control.py @@ -0,0 +1,394 @@ +"""One bounded administrative exchange, authenticated by the local OS.""" + +from __future__ import annotations + +import asyncio +from contextlib import asynccontextmanager +import ctypes +from errno import ECONNABORTED, EMFILE, ENFILE, ENOBUFS, ENOMEM +import json +import os +from pathlib import Path +import socket +import sqlite3 +import stat +import struct +import sys +import time +from typing import Any, AsyncGenerator, cast + +from . import authority as a + +MAX_MESSAGE = 64 * 1024 +MAX_CONNECTIONS = 16 +EXCHANGE_TIMEOUT = 10 +_TRANSIENT = {ECONNABORTED, EMFILE, ENFILE, ENOBUFS, ENOMEM} + + +def _socket_identity(path: Path) -> tuple[int, int]: + a._root(path.parent, create=False) + info = path.lstat() + if ( + not stat.S_ISSOCK(info.st_mode) + or info.st_uid != os.geteuid() + or info.st_mode & 0o077 + or info.st_nlink != 1 + ): + raise a.AuthorityError("unsafe local control endpoint") + return info.st_dev, info.st_ino + + +def _peer_uid(peer: Any) -> int: + if sys.platform.startswith("linux"): + credentials = peer.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12) + return struct.unpack("iII", credentials)[1] + if sys.platform.startswith(("darwin", "freebsd", "openbsd", "netbsd")): + libc = ctypes.CDLL(None, use_errno=True) + getpeereid = libc.getpeereid + getpeereid.argtypes = [ + ctypes.c_int, + ctypes.POINTER(ctypes.c_uint), + ctypes.POINTER(ctypes.c_uint), + ] + getpeereid.restype = ctypes.c_int + uid, gid = ctypes.c_uint(), ctypes.c_uint() + if getpeereid(peer.fileno(), ctypes.byref(uid), ctypes.byref(gid)): + raise OSError(ctypes.get_errno(), "getpeereid failed") + return uid.value + raise a.AuthorityError("native local peer verification is unavailable") + + +def _frame(value: dict[str, object]) -> bytes: + data = json.dumps(value, separators=(",", ":")).encode() + if len(data) > MAX_MESSAGE: + raise a.AuthorityError("local control message is too large") + return struct.pack("!I", len(data)) + data + + +def _unique(pairs: list[tuple[str, object]]) -> dict[str, object]: + result: dict[str, object] = {} + for key, value in pairs: + if key in result: + raise a.AuthorityError("duplicate local control field") + result[key] = value + return result + + +def _decode(data: bytes) -> dict[str, Any]: + try: + value = json.loads(data, object_pairs_hook=_unique) + if not isinstance(value, dict): + raise ValueError + return value + except (ValueError, UnicodeError, RecursionError) as exc: + raise a.AuthorityError("invalid local control message") from exc + + +def _receive(peer: socket.socket, count: int, deadline: float) -> bytes: + data = bytearray() + while len(data) < count: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError("local control exchange timed out") + peer.settimeout(remaining) + chunk = peer.recv(count - len(data)) + if not chunk: + raise a.AuthorityError("local control connection closed") + data.extend(chunk) + return bytes(data) + + +def request( + state_dir: Path, action: str, recovery_file: Path | None, reuse: bool +) -> dict[str, Any]: + root = a._root(state_dir, create=False) + endpoint = root / "control.sock" + identity = _socket_identity(endpoint) + value: dict[str, object] = {"root": str(root), "action": action} + if action == "init": + if recovery_file is None: + raise a.AuthorityError( + "select an external private --recovery-file destination" + ) + value.update( + recovery_file=str(recovery_file.expanduser().absolute()), reuse=reuse + ) + with socket.socket(socket.AF_UNIX) as peer: + deadline = time.monotonic() + EXCHANGE_TIMEOUT + peer.settimeout(EXCHANGE_TIMEOUT) + peer.connect(str(endpoint)) + if _peer_uid(peer) != os.geteuid() or _socket_identity(endpoint) != identity: + raise a.AuthorityError("local control server identity changed") + peer.sendall(_frame(value)) + size = struct.unpack("!I", _receive(peer, 4, deadline))[0] + if size > MAX_MESSAGE: + raise a.AuthorityError("local control message is too large") + reply = _decode(_receive(peer, size, deadline)) + if reply.get("ok") is not True: + raise a.AuthorityError(str(reply.get("error", "local control failed"))) + result = reply.get("result") + if not isinstance(result, dict) or type(result.get("initialized")) is not bool: + raise a.AuthorityError("invalid local control result") + return result + + +class _Listener(socket.socket): + def __init__(self) -> None: + super().__init__(socket.AF_UNIX, socket.SOCK_STREAM) + self.pending: set[socket.socket] = set() + + def accept(self) -> tuple[socket.socket, Any]: + peer, address = super().accept() + # Own the descriptor before sock_accept's future can be cancelled. + self.pending.add(peer) + return peer, address + + +class _Protocol(asyncio.StreamReaderProtocol): + """Keep even admission-rejected transports alive until connection_lost.""" + + def __init__(self, control: LocalControl) -> None: + self.control = control + self.transport: asyncio.Transport | None = None + super().__init__(asyncio.StreamReader(limit=MAX_MESSAGE), control.accept) + + def connection_made(self, transport: asyncio.BaseTransport) -> None: + self.transport = cast(asyncio.Transport, transport) + self.control.transports.add(self.transport) + self.control.drained.clear() + super().connection_made(transport) + + def connection_lost(self, exc: Exception | None) -> None: + try: + super().connection_lost(exc) + finally: + if self.transport is not None: + self.control.transports.discard(self.transport) + self.transport = None + if not self.control.transports: + self.control.drained.set() + + +class LocalControl: + def __init__(self, root: Path) -> None: + self.root = a._root(root, create=False) + self.path = self.root / "control.sock" + self.identity: tuple[int, int] | None = None + self.listener: _Listener | None = None + self.acceptor: asyncio.Task[None] | None = None + self.connections: set[asyncio.StreamWriter] = set() + self.handlers: set[asyncio.Task[Any]] = set() + self.transports: set[asyncio.Transport] = set() + self.drained = asyncio.Event() + self.drained.set() + self.work: asyncio.Task[dict[str, Any]] | None = None + self.closed = False + + async def start(self) -> None: + # The caller already owns the persistent authority lock, so no other + # legitimate instance for this exact root can own this stale socket. + if self.path.exists() or self.path.is_symlink(): + _socket_identity(self.path) + self.path.unlink() + self.listener = _Listener() + try: + self.listener.bind(str(self.path)) + except OSError as exc: + raise a.AuthorityError( + f"cannot bind private control socket {self.path}; use a short local " + f"server-data path on a Unix-socket-capable filesystem: {exc}" + ) from exc + self.path.chmod(0o600) + self.identity = _socket_identity(self.path) + self.listener.listen(MAX_CONNECTIONS) + self.listener.setblocking(False) + self.acceptor = asyncio.create_task(self.listen()) + self.acceptor.add_done_callback(self.acceptor_finished) + + def acceptor_finished(self, task: asyncio.Task[None]) -> None: + if not task.cancelled() and (error := task.exception()) is not None: + task.get_loop().call_exception_handler( + {"message": "Arbiter local control listener failed", "exception": error} + ) + + async def listen(self) -> None: + assert self.listener is not None + loop = asyncio.get_running_loop() + while not self.closed: + peer = None + try: + peer, _ = await loop.sock_accept(self.listener) + if self.closed or len(self.connections) >= MAX_CONNECTIONS: + peer.close() + continue + await loop.connect_accepted_socket(lambda: _Protocol(self), peer) + except OSError as exc: + if peer is not None: + peer.close() + if exc.errno not in _TRANSIENT: + raise + await asyncio.sleep(1) + except BaseException: + if peer is not None: + peer.close() + raise + finally: + if peer is not None: + self.listener.pending.discard(peer) + + def accept( + self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter + ) -> None: + if self.closed or len(self.connections) >= MAX_CONNECTIONS: + # No request task or reply queue for rejected connections. The + # protocol retains the transport until its close callback runs. + writer.transport.abort() + return + self.connections.add(writer) + task = asyncio.create_task(self.handle(reader, writer)) + self.handlers.add(task) + + def finished(done: asyncio.Task[Any]) -> None: + self.handlers.discard(done) + # Cancellation may arrive before handle() executes its first line. + self.connections.discard(writer) + + task.add_done_callback(finished) + + def dispatch(self, value: dict[str, Any]) -> dict[str, Any]: + action = value.get("action") + keys = ( + {"root", "action", "recovery_file", "reuse"} + if action == "init" + else {"root", "action"} + ) + if value.get("root") != str(self.root) or set(value) != keys: + raise a.AuthorityError("invalid local control request") + if action == "init": + if ( + not isinstance(value["recovery_file"], str) + or type(value["reuse"]) is not bool + ): + raise a.AuthorityError("invalid local setup arguments") + record = a._initialize_locked( + self.root, Path(value["recovery_file"]), value["reuse"] + ) + elif action == "status": + record = a._current_record(self.root) + else: + raise a.AuthorityError("unsupported local control action") + return record.to_dict() if record else {"initialized": False} + + async def exchange( + self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter + ) -> None: + if _peer_uid(writer.get_extra_info("socket")) != os.geteuid(): + raise a.AuthorityError("local control caller is not the instance user") + if _socket_identity(self.path) != self.identity: + raise a.AuthorityError("local control endpoint was replaced") + size = struct.unpack("!I", await reader.readexactly(4))[0] + if size > MAX_MESSAGE: + raise a.AuthorityError("local control message is too large") + value = _decode(await reader.readexactly(size)) + if self.work is not None and not self.work.done(): + writer.write( + _frame( + { + "ok": False, + "error": "local authority work is busy; try again later", + } + ) + ) + await writer.drain() + return + # Shield the actual work: a disconnect/deadline must not free capacity + # or release the instance lock while file/SQLite work is still running. + self.work = asyncio.create_task(asyncio.to_thread(self.dispatch, value)) + self.work.add_done_callback( + lambda done: done.exception() if not done.cancelled() else None + ) + try: + result = await asyncio.shield(self.work) + response: dict[str, object] = {"ok": True, "result": result} + except (OSError, ValueError, sqlite3.Error) as exc: + response = {"ok": False, "error": str(exc)} + writer.write(_frame(response)) + await writer.drain() + + async def handle( + self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter + ) -> None: + try: + await asyncio.wait_for(self.exchange(reader, writer), EXCHANGE_TIMEOUT) + except ( + OSError, + ValueError, + sqlite3.Error, + asyncio.IncompleteReadError, + asyncio.TimeoutError, + ): + # Invalid or unauthenticated callers receive no state or error text. + pass + finally: + writer.close() + try: + await writer.wait_closed() + except (OSError, asyncio.CancelledError): + pass + + async def close(self) -> None: + self.closed = True + # Own and join the accept operation as well as established transports. + # asyncio.Server can leave an already-scheduled accept task creating a + # transport after Server.close(), before connection_made can track it. + if self.listener is not None: + self.listener.close() + if self.acceptor is not None: + self.acceptor.cancel() + await asyncio.gather(self.acceptor, return_exceptions=True) + if self.listener is not None: + for peer in self.listener.pending: + peer.close() + self.listener.pending.clear() + for transport in tuple(self.transports): + transport.abort() + for task in tuple(self.handlers): + task.cancel() + if self.handlers: + await asyncio.gather(*self.handlers, return_exceptions=True) + await self.drained.wait() + cancelled = False + while self.work is not None and not self.work.done(): + try: + await asyncio.shield(self.work) + except asyncio.CancelledError: + cancelled = True + except (OSError, ValueError, sqlite3.Error): + pass + if self.identity is not None and self.path.exists(): + if _socket_identity(self.path) == self.identity: + self.path.unlink() + if cancelled: + raise asyncio.CancelledError + + +@asynccontextmanager +async def running(root: Path) -> AsyncGenerator[None, None]: + if os.name == "nt": + yield + return + control = LocalControl(root) + try: + await control.start() + yield + finally: + closing = asyncio.create_task(control.close()) + cancelled = False + while not closing.done(): + try: + await asyncio.shield(closing) + except asyncio.CancelledError: + cancelled = True + await closing + if cancelled: + raise asyncio.CancelledError diff --git a/server/src/arbiter_server/main.py b/server/src/arbiter_server/main.py index 2a9fbb0..e4e19e1 100644 --- a/server/src/arbiter_server/main.py +++ b/server/src/arbiter_server/main.py @@ -2273,10 +2273,13 @@ def _create_native_http_app( @asynccontextmanager async def lifespan(_app: object) -> AsyncGenerator[None, None]: from .authority import running_instance + from .local_control import running - with running_instance(_server_data_root(_storage_config(cfg)) / "authority"): + root = _server_data_root(_storage_config(cfg)) / "authority" + with running_instance(root): try: - yield + async with running(root): + yield finally: http_work.close() @@ -4421,7 +4424,8 @@ def _build_parser() -> argparse.ArgumentParser: subcommands = parser.add_subparsers(dest="command", required=True) authority = subcommands.add_parser( - "authority", help="initialize or inspect offline authority identity records" + "authority", + help="initialize or inspect authority through local OS authentication", ) authority_commands = authority.add_subparsers( dest="authority_command", required=True diff --git a/server/tests/unit/conftest.py b/server/tests/unit/conftest.py index 866400a..0befe19 100644 --- a/server/tests/unit/conftest.py +++ b/server/tests/unit/conftest.py @@ -1,8 +1,10 @@ from __future__ import annotations +import atexit import os from pathlib import Path import subprocess +import tempfile import pytest @@ -42,3 +44,14 @@ def private_windows_test_parent(tmp_path: Path) -> None: text=True, ) ensure_runtime_config_permissions(config_dir=tmp_path, env_file=None) + + +@pytest.fixture +def tmp_path(tmp_path_factory): + # Unix socket addresses include the full path; keep native tests portable. + if os.name == "nt": + yield tmp_path_factory.mktemp("test") + else: + directory = tempfile.TemporaryDirectory(prefix="arbiter-", dir="/tmp") + atexit.register(directory.cleanup) + yield Path(directory.name).resolve() diff --git a/server/tests/unit/test_local_control.py b/server/tests/unit/test_local_control.py new file mode 100644 index 0000000..3530d5f --- /dev/null +++ b/server/tests/unit/test_local_control.py @@ -0,0 +1,493 @@ +from __future__ import annotations + +import asyncio +from contextlib import asynccontextmanager, suppress +import gc +import json +import os +from pathlib import Path +import socket +import struct +import subprocess +import sys +import tempfile +import threading +from typing import Any, cast + +import pytest + +from arbiter_server import authority as a, local_control as c +from arbiter_server.config import AppConfig +from arbiter_server.main import build_server +from server.tests.unit.test_native_http_protocol import _asgi_request + +pytestmark = pytest.mark.skipif(os.name == "nt", reason="native POSIX control") +CONTROL_FAILURES: tuple[type[Exception], ...] = (a.AuthorityError, OSError) + + +@asynccontextmanager +async def serve(root): + with a.running_instance(root): + control = c.LocalControl(root) + try: + await control.start() + yield control + finally: + await control.close() + + +async def call(root, action="status", recovery=None, reuse=False): + return await asyncio.to_thread(c.request, root, action, recovery, reuse) + + +async def wire(control, data, size=None): + reader, writer = await asyncio.open_unix_connection(str(control.path)) + writer.write(struct.pack("!I", len(data) if size is None else size) + data) + await writer.drain() + try: + return await asyncio.wait_for(reader.read(), 2) + finally: + writer.close() + await writer.wait_closed() + + +def test_real_asgi_cli_setup_status_and_restart(tmp_path): + cfg = AppConfig() + cfg.arbiter.storage.server_data_dir = str(tmp_path / "server") + cfg.arbiter.storage.plugin_data_dir = str(tmp_path / "plugins") + app = cast(Any, build_server(cfg, service_plugins=[], prepare_tls=False).app) + root = tmp_path / "server" / "authority" + recovery = tmp_path / "operator.json" + + async def command(action): + argv = [sys.executable, "-m", "arbiter_server", "authority", action] + argv += ["--state-dir", str(root), "--json"] + if action == "init": + argv += ["--recovery-file", recovery.name] + return await asyncio.to_thread( + subprocess.run, + argv, + text=True, + capture_output=True, + timeout=15, + cwd=tmp_path, + ) + + async def exercise(): + for stale in (False, True): + with a._locked(a._root(root, create=True)): + if stale: + with socket.socket(socket.AF_UNIX) as peer: + peer.bind(str((root / "control.sock"))) + (root / "control.sock").chmod(0o600) + busy = await command("status") + assert busy.returncode == 1 and "authority state is busy" in busy.stderr + async with app.router.lifespan_context(app): + assert (await command("status")).stdout.strip() == '{"initialized": false}' + result = await command("init") + assert result.returncode == 0, result.stderr + initial = json.loads(result.stdout) + assert initial["initialized"] is True + assert json.loads((await command("status")).stdout) == initial + denied = await command("init") + assert denied.returncode == 1 and "already exists" in denied.stderr + assert (await _asgi_request(app, "GET", "/_health_")).status_code == 200 + custody = recovery.read_bytes() + assert b"PRIVATE" not in result.stdout.encode() + assert not (root / "control.sock").exists() + restored = a.authority_status(root) + assert restored is not None and restored.to_dict() == initial + async with app.router.lifespan_context(app): + assert json.loads((await command("status")).stdout) == initial + assert recovery.read_bytes() == custody + + asyncio.run(exercise()) + + +@pytest.mark.skipif( + getattr(os, "geteuid", lambda: 1)() != 0, + reason="requires native root to drop to a distinct UID", +) +def test_real_different_os_user_is_rejected(monkeypatch): + import pwd + + account = pwd.getpwnam("nobody") + seen = [] + real = c._peer_uid + + def observe(peer): + uid = real(peer) + seen.append(uid) + return uid + + monkeypatch.setattr(c, "_peer_uid", observe) + with tempfile.TemporaryDirectory(prefix="arbiter-peer-", dir="/tmp") as directory: + parent = Path(directory) + root = parent / "authority" + + async def exercise(): + async with serve(root) as control: + # Test-only access through the filesystem lets the other UID + # reach the kernel peer check. Production never broadens modes. + parent.chmod(0o711) + root.chmod(0o711) + control.path.chmod(0o666) + script = """ +import os, socket, sys +os.setgroups([]) +os.setgid(int(sys.argv[2])) +os.setuid(int(sys.argv[1])) +with socket.socket(socket.AF_UNIX) as peer: + peer.settimeout(2) + peer.connect(sys.argv[3]) + try: + assert not peer.recv(1) + except ConnectionResetError: + pass +print(os.geteuid()) +""" + try: + argv = [sys.executable, "-I", "-c", script] + argv.extend( + map(str, (account.pw_uid, account.pw_gid, control.path)) + ) + result = await asyncio.to_thread( + subprocess.run, + argv, + text=True, + capture_output=True, + timeout=5, + ) + assert result.returncode == 0, result.stderr + assert int(result.stdout) == account.pw_uid != os.geteuid() + assert seen == [account.pw_uid] + assert control.work is None + finally: + control.path.chmod(0o600) + root.chmod(0o700) + parent.chmod(0o700) + + asyncio.run(exercise()) + + +def test_kernel_peer_uid_is_checked_in_both_directions(tmp_path, monkeypatch): + root = tmp_path / "authority" + real = c._peer_uid + observed = [] + + def peer(sock): + uid = real(sock) + observed.append(uid) + return uid + + monkeypatch.setattr(c, "_peer_uid", peer) + + async def exercise(): + async with serve(root) as control: + assert control.path.parent == root.resolve() + assert await call(root) == {"initialized": False} + assert len(observed) == 2 and set(observed) == {os.geteuid()} + + asyncio.run(exercise()) + + +@pytest.mark.parametrize("side", ["client", "server"]) +def test_wrong_peer_uid_is_rejected_before_dispatch(tmp_path, monkeypatch, side): + root = tmp_path / "authority" + real = c._peer_uid + dispatched = [] + + def peer(sock): + # The synchronous client runs in a worker; server checks run on the loop. + client = threading.current_thread() is not threading.main_thread() + return os.geteuid() + 1 if client == (side == "client") else real(sock) + + monkeypatch.setattr(c, "_peer_uid", peer) + monkeypatch.setattr( + c.LocalControl, "dispatch", lambda *args: dispatched.append(True) + ) + + async def exercise(): + async with serve(root): + with pytest.raises(CONTROL_FAILURES): + await call(root) + assert not dispatched + + asyncio.run(exercise()) + + +@pytest.mark.parametrize("damage", ["mode", "file", "symlink", "replacement"]) +def test_endpoint_damage_fails_closed_and_is_not_deleted(tmp_path, damage): + root = tmp_path / "authority" + + async def exercise(): + with a.running_instance(root): + control = c.LocalControl(root) + await control.start() + path = control.path + replacement = None + if damage == "mode": + path.chmod(0o666) + else: + path.unlink() + if damage == "symlink": + path.symlink_to(tmp_path / "absent") + elif damage == "file": + path.write_text("preserve") + else: + replacement = socket.socket(socket.AF_UNIX) + replacement.bind(str(path)) + path.chmod(0o600) + try: + with pytest.raises(CONTROL_FAILURES): + await call(root) + try: + await control.close() + except a.AuthorityError: + pass + assert path.exists() or path.is_symlink() + finally: + if replacement is not None: + replacement.close() + path.unlink() + + asyncio.run(exercise()) + + +def test_stale_private_socket_reconciles_but_unknown_file_is_preserved(tmp_path): + root = a._root(tmp_path / "authority", create=True) + path = root / "control.sock" + with socket.socket(socket.AF_UNIX) as stale: + stale.bind(str(path)) + path.chmod(0o600) + + async def exercise(): + async with serve(root): + assert await call(root) == {"initialized": False} + path.write_text("unrelated") + path.chmod(0o600) + try: + with pytest.raises(a.AuthorityError): + async with serve(root): + pytest.fail("accepted an unrelated endpoint file") + assert path.read_text() == "unrelated" + finally: + path.unlink() + + asyncio.run(exercise()) + + +@pytest.mark.parametrize( + "data", + [ + b"", + b"{", + b"[]", + b'{"action":"status","action":"init"}', + b'{"root":"other","action":"status"}', + ], +) +def test_malformed_requests_close_and_capacity_recovers(tmp_path, data): + root = tmp_path / "authority" + + async def exercise(): + async with serve(root) as control: + response = await wire(control, data) + if response: + assert struct.unpack("!I", response[:4])[0] == len(response) - 4 + reply = json.loads(response[4:]) + assert reply["ok"] is False and "result" not in reply + assert await call(root) == {"initialized": False} + + asyncio.run(exercise()) + + +def test_oversize_prefix_is_rejected_without_body(tmp_path): + root = tmp_path / "authority" + + async def exercise(): + async with serve(root) as control: + assert await wire(control, b"", size=c.MAX_MESSAGE + 1) == b"" + assert await call(root) == {"initialized": False} + + asyncio.run(exercise()) + + +def test_exact_message_boundary_is_accepted(tmp_path): + root = tmp_path / "authority" + + async def exercise(): + async with serve(root) as control: + value = json.dumps({"root": str(root), "action": "status"}).encode() + value += b" " * (c.MAX_MESSAGE - len(value)) + response = await wire(control, value) + assert struct.unpack("!I", response[:4])[0] == len(response) - 4 + assert json.loads(response[4:]) == { + "ok": True, + "result": {"initialized": False}, + } + + asyncio.run(exercise()) + + +def test_slow_connections_are_bounded_and_timeout_recovers_capacity( + tmp_path, monkeypatch +): + root = tmp_path / "authority" + monkeypatch.setattr(c, "EXCHANGE_TIMEOUT", 0.15) + + async def exercise(): + async with serve(root) as control: + clients = [ + await asyncio.open_unix_connection(str(control.path)) + for _ in range(c.MAX_CONNECTIONS) + ] + extra = await asyncio.open_unix_connection(str(control.path)) + assert await asyncio.wait_for(extra[0].read(1), 2) == b"" + assert len(control.connections) <= c.MAX_CONNECTIONS + for reader, writer in [*clients, extra]: + assert await asyncio.wait_for(reader.read(1), 2) == b"" + writer.close() + await writer.wait_closed() + assert await call(root) == {"initialized": False} + + asyncio.run(exercise()) + + +@pytest.mark.filterwarnings("error::ResourceWarning") +@pytest.mark.filterwarnings("error::pytest.PytestUnraisableExceptionWarning") +@pytest.mark.parametrize("ending", ["malformed", "timeout", "overload", "cancel"]) +def test_all_transports_close_before_control_returns(tmp_path, monkeypatch, ending): + root = tmp_path / "authority" + monkeypatch.setattr(c, "EXCHANGE_TIMEOUT", 0.1) + monkeypatch.setattr(c, "MAX_CONNECTIONS", 1) + if ending == "cancel": + # A cancelled handler may never execute its own cleanup body. + monkeypatch.setattr( + c.LocalControl, "handle", lambda *args: asyncio.Event().wait() + ) + transports = [] + connected = c._Protocol.connection_made + + def observe(protocol, transport): + transports.append(transport) + connected(protocol, transport) + + monkeypatch.setattr(c._Protocol, "connection_made", observe) + + async def exercise(): + clients = [] + async with serve(root) as control: + reader, writer = await asyncio.open_unix_connection(str(control.path)) + clients.append(writer) + if ending == "overload": + extra, rejected = await asyncio.open_unix_connection(str(control.path)) + clients.append(rejected) + assert await asyncio.wait_for(extra.read(1), 2) == b"" + assert len(control.handlers) == 1 + elif ending == "malformed": + writer.write(struct.pack("!I", 1) + b"{") + await writer.drain() + assert await asyncio.wait_for(reader.read(1), 2) == b"" + elif ending == "timeout": + assert await asyncio.wait_for(reader.read(1), 2) == b"" + for writer in clients: + writer.close() + await writer.wait_closed() + assert not control.transports and cast(Any, control.acceptor).done() + assert (listener := control.listener) is not None + assert not listener.pending and listener.fileno() == -1 + assert not control.handlers and not control.connections + # Excess sockets close before a stream transport or handler is created. + assert len(transports) == 1 + assert all(t.get_extra_info("socket").fileno() == -1 for t in transports) + transports.clear() + gc.collect() + + asyncio.run(exercise()) + gc.collect() + + +@pytest.mark.parametrize("method", ["sock_accept", "connect_accepted_socket"]) +def test_transient_failure_recovers_and_shutdown_joins_backoff( + tmp_path, monkeypatch, method +): + async def exercise(): + loop = asyncio.get_running_loop() + original = getattr(loop, method) + failed = asyncio.Event() + + async def transient(*args): + if not failed.is_set(): + failed.set() + raise OSError(c.EMFILE, "injected resource pressure") + return await original(*args) + + monkeypatch.setattr(loop, method, transient) + async with serve(tmp_path / "authority") as control: + with suppress(*CONTROL_FAILURES): + await call(control.root) + assert await call(control.root) == {"initialized": False} + failed.clear() + pending = asyncio.create_task(call(control.root)) + await asyncio.wait_for(failed.wait(), 2) + await control.close() + with suppress(*CONTROL_FAILURES): + await pending + assert cast(Any, control.acceptor).done() and not control.path.exists() + assert not control.transports and not cast(Any, control.listener).pending + + asyncio.run(exercise()) + + +def test_disconnect_and_cancelled_shutdown_retain_actual_work_and_instance_lock( + tmp_path, monkeypatch +): + cfg = AppConfig() + cfg.arbiter.storage.server_data_dir = str(tmp_path / "server") + cfg.arbiter.storage.plugin_data_dir = str(tmp_path / "plugins") + app = cast(Any, build_server(cfg, service_plugins=[], prepare_tls=False).app) + root = tmp_path / "server" / "authority" + entered, release = threading.Event(), threading.Event() + ready = asyncio.Event() + real = c.LocalControl.dispatch + + def dispatch(control, value): + entered.set() + assert release.wait(5) + return real(control, value) + + monkeypatch.setattr(c.LocalControl, "dispatch", dispatch) + + async def exercise(): + async def owner(): + async with app.router.lifespan_context(app): + ready.set() + await asyncio.Event().wait() + + task = asyncio.create_task(owner()) + await asyncio.wait_for(ready.wait(), 2) + _, writer = await asyncio.open_unix_connection(str((root / "control.sock"))) + writer.write(c._frame({"root": str(root), "action": "status"})) + await writer.drain() + assert await asyncio.to_thread(entered.wait, 2) + assert (await _asgi_request(app, "GET", "/_health_")).status_code == 200 + with pytest.raises(a.AuthorityError, match="busy"): + await call(root) + writer.close() + await writer.wait_closed() + task.cancel() + await asyncio.sleep(0.02) + task.cancel() + await asyncio.sleep(0.02) + try: + assert not task.done() + with pytest.raises(a.AuthorityBusy): + a.authority_status(root) + finally: + release.set() + with pytest.raises(asyncio.CancelledError): + await task + assert a.authority_status(root) is None + + asyncio.run(exercise()) diff --git a/website/docs/operate/credentials.md b/website/docs/operate/credentials.md index ba212e6..2e4930d 100644 --- a/website/docs/operate/credentials.md +++ b/website/docs/operate/credentials.md @@ -83,6 +83,27 @@ POSIX publication flushes files and parent directories. Windows uses same-volume [write-through moves](https://learn.microsoft.com/en-us/windows/win32/fileio/moving-directories) for files and generation directories; filesystem errors are reported as failures. +On Linux and macOS, `authority init` and `authority status` also work while the +server is running, when invoked by the server's OS user. They use a private Unix socket +and authenticate both peers through kernel credentials. Only the running server +writes its live authority state. Other users must use trusted OS tooling to run +the command as the service account; an HTTP connection cannot administer it. +Repeated initialization still refuses to replace IDs, credentials or recovery +trust. Keep the recovery file outside the server's state and agent storage. + +Live control on other POSIX systems is unsupported in this slice. An unavailable +native peer-credential backend fails closed; stop the server for offline setup +or status on those systems. Windows live control follows in a separate slice. + +After an interrupted control exchange, query status before retrying setup: a +missing reply does not prove the server stopped writing. A later startup removes +only a validated stale private socket while holding the instance lock. Unexpected +endpoint files or permissions require investigation and are preserved. If no live +endpoint exists during startup, shutdown or another offline command, the CLI +retains the authority-busy diagnostic. Unsafe endpoints still report their safety +error. The socket resides in the private authority directory; select a short +local server-data path within the OS Unix-socket path limit. + Shared public-only verification vectors are in `tests/fixtures/authentication/credential-profiles-v1.json`. They are synthetic test trust, consumed by Python now and intended for Go's slice 2.1 verifier. diff --git a/website/docs/operate/security.md b/website/docs/operate/security.md index 995a634..179d337 100644 --- a/website/docs/operate/security.md +++ b/website/docs/operate/security.md @@ -183,8 +183,28 @@ preparing state for the running server. With no override, this is beneath Arbiter's existing platform-specific server data directory. Startup takes the same persistent instance lock as offline setup and holds it through HTTP cleanup; a second start, offline initialization or offline status fails while it is held. -Stop the instance before using these offline commands. This enforces state -ownership; runtime caller authentication remains a later milestone. +On Linux and macOS, these commands contact the lock-owning running instance +through authenticated local control when offline access reports busy. Run them as the +same OS account as the server, directly or through verified SSH to that account. +With the server stopped, the commands keep using the private offline state. +This enforces administrative state ownership; runtime API caller authentication +remains a later milestone. + +The Unix socket is private and checks kernel peer credentials in both directions. +Knowing its name or connecting to a loopback TCP port does not grant local control. +It is `control.sock` inside the validated private authority directory. Another +OS user cannot pre-create its parent or replace its endpoint. Service, login and +SSH runtime-directory environment settings do not affect its location. Unix +socket paths have a platform length limit: use a short local server-data path on +a filesystem that supports Unix sockets. Startup reports binding errors; it does +not fall back to a shared temporary directory or start without local control. +Changed ownership, permissions, file type or endpoint identity fail closed. + +Control accepts at most 16 connections, a 64 KiB message and a ten-second +exchange. One authority job runs at a time, outside the HTTP event loop. A +disconnect or deadline can close the reply while that job continues; inspect +status before retrying initialization. Shutdown retains the instance lock until +the actual authority job finishes. Windows live control follows in its own slice. State is selected from a private generation. Stopped-instance status and startup can finish a valid interrupted initial publication, preserving its IDs and keys. @@ -197,8 +217,10 @@ It holds both backup-decryption and live-recovery authority; theft compromises both. See [operator recovery custody](credentials.md#operator-recovery-custody) for reuse and interrupted setup. -Stop the instance before initializing. Concurrent initialization attempts use -one persistent process lock. A failed unpublished initialization can be retried; +Stopped-instance initialization takes the same persistent process lock. On Linux +and macOS, the running instance handles authenticated local initialization instead. Windows +currently requires the instance to be stopped. A failed unpublished initialization +can be retried; a database already published must be inspected rather than overwritten. Keep the directory private; the command rejects unsafe permissions and linked files.