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.