From deb11df7831688f9dd677fc51c49fe868892f879 Mon Sep 17 00:00:00 2001 From: ms-lead-reviewer Date: Fri, 17 Jul 2026 20:45:00 -0500 Subject: [PATCH] feat(#828): add authenticated lease broker Bind broker sessions to Linux SO_PEERCRED and /proc starttime ancestry, fence runtime generations, persist cryptographic single-use cycle tokens, and enforce protected Unix socket/state posture.\n\ncloses #828 --- docs/SITEMAP.md | 6 + docs/architecture/lease-broker-protocol.md | 18 + docs/architecture/lease-broker-security.md | 11 + docs/guides/lease-broker-operations.md | 19 + docs/scratchpads/828-lease-broker.md | 66 +++ .../framework/tools/lease-broker/daemon.py | 446 ++++++++++++++++++ 6 files changed, 566 insertions(+) create mode 100644 docs/architecture/lease-broker-protocol.md create mode 100644 docs/architecture/lease-broker-security.md create mode 100644 docs/guides/lease-broker-operations.md create mode 100644 docs/scratchpads/828-lease-broker.md create mode 100644 packages/mosaic/framework/tools/lease-broker/daemon.py diff --git a/docs/SITEMAP.md b/docs/SITEMAP.md index f53a46e..52083a5 100644 --- a/docs/SITEMAP.md +++ b/docs/SITEMAP.md @@ -1,5 +1,11 @@ # Documentation Sitemap +## Compaction refresh lease broker + +- [Internal broker protocol](architecture/lease-broker-protocol.md) — kernel identity, ancestry and generation invariants, framed requests, responses, and persisted cycle bindings. +- [Broker operations](guides/lease-broker-operations.md) — protected paths, startup, fail-closed recovery posture, distinct-principal deployment, and residual risk. +- [WI-1 security notes](architecture/lease-broker-security.md) — threat-boundary summary and coordinator review requirements. + ## CLI and skill management - [Skill registration user guide](guides/user-guide.md#claude-code-skill-registration) — register, unregister, list statuses, automatic install/update reconciliation, and Claude reload behavior. diff --git a/docs/architecture/lease-broker-protocol.md b/docs/architecture/lease-broker-protocol.md new file mode 100644 index 0000000..468871d --- /dev/null +++ b/docs/architecture/lease-broker-protocol.md @@ -0,0 +1,18 @@ +# Authenticated external lease broker protocol + +The compaction-refresh lease broker is a Linux-only, newline-framed JSON protocol over a Unix stream socket. It is runtime-neutral; M1 consumers are limited to Claude and Pi. This is an internal process boundary, not an HTTP API, so it is intentionally absent from OpenAPI. + +The broker, never the caller, obtains `(pid, uid, gid)` from kernel `SO_PEERCRED`. It correlates the PID with `/proc//stat` field 22 (`starttime`) and mints `session_id` on `register_anchor`. Presence of `session_id` in that request is refused even when its value is `null` or empty. Later requests must originate from the anchor or a descendant. The broker walks parent PIDs to the `(pid,starttime)` anchor and then rereads every walked PID's starttime before accepting the chain. + +## Request and response boundary + +Each connection carries exactly one UTF-8 JSON object followed by one newline, capped at 64 KiB. The protocol deliberately uses EOF to prove that there is exactly one frame: immediately after writing the newline, the client **MUST half-close its write side** with `shutdown(SHUT_WR)` (or Node `socket.end()`) before awaiting the response. A client that writes a newline but leaves its write side open receives no successful response; the broker's one-second connection deadline fails closed. Malformed, unterminated, multiple (including a delayed second frame), or oversized frames fail closed. Responses are one JSON object and one newline. Success has `{"ok":true,...}`; refusal has `{"ok":false,"code":"TYPED_CODE"}`. Requests are: + +- `register_anchor`: `action`, non-negative `runtime_generation`; no `session_id` field. +- `authenticate`: `action`, broker-minted `session_id`, non-negative `runtime_generation`. +- `mint_token`: authenticated identity plus `binding` containing exactly `compaction_epoch`, `request_epoch`, `h_source`, `h_payload`, and `schema_version`. +- `consume_token`: authenticated identity plus `token`. + +A higher generation for the same anchor atomically replaces the stored incarnation and consumes all prior tokens for that session. A lower generation is stale. Tokens are 256-bit values from the operating-system cryptographic RNG and are single use. Their persisted binding is the WI-1 substrate for later receipt work; WI-1 does not implement receipts, promotion, hooks, payload builders, mutator gates, or recovery. + +State replacement uses a mode-`0600` temporary file, `fsync`, atomic rename, and parent-directory `fsync`. Existing state is opened without following symlinks, must be a bounded regular file at mode `0600`, and is fully schema- and invariant-validated before use. Session identity is uniquely keyed by `(anchor_pid,anchor_starttime)`; duplicate logical sessions for one anchor refuse startup. State integrity or mode failures refuse startup. The daemon does not log session IDs or tokens. diff --git a/docs/architecture/lease-broker-security.md b/docs/architecture/lease-broker-security.md new file mode 100644 index 0000000..9c242e6 --- /dev/null +++ b/docs/architecture/lease-broker-security.md @@ -0,0 +1,11 @@ +# WI-1 lease broker security notes + +- Trusted identity comes only from Linux `SO_PEERCRED` plus `/proc` starttime, never request identity fields. +- Descendant authorization is anchored to `(pid,starttime)` and uses a complete second starttime pass to fail closed on disappearance or PID-reuse races. +- Runtime generations are monotonic per anchor; a bump revokes prior-incarnation tokens before persistence commits. +- Session IDs and cycle tokens use the OS cryptographic RNG. `Math.random` and model output are not token sources. +- Framing and persistence failures fail closed. Sensitive tokens are not logged. +- Same-principal filesystem modes are minimum hardening, not socket authenticity against the same UID. Distinct-principal service isolation is required for that stronger claim. +- WI-2+ security surfaces—receipts, promotion transactions, mutator gates, hooks, payload construction, and recovery—are explicitly out of scope. + +Coordinator security review must rerun the real socket/peercred acceptance suite on an unrestricted Linux runner and obtain the mandated independent Opus-SECREV review before integration. diff --git a/docs/guides/lease-broker-operations.md b/docs/guides/lease-broker-operations.md new file mode 100644 index 0000000..fd4da20 --- /dev/null +++ b/docs/guides/lease-broker-operations.md @@ -0,0 +1,19 @@ +# Lease broker operations + +Place the socket and state file in a dedicated directory with mode `0700`. Start the packaged daemon with: + +```bash +python3 "$MOSAIC_HOME/tools/lease-broker/daemon.py" \ + --socket /run/user/1000/mosaic-lease/broker.sock \ + --state /run/user/1000/mosaic-lease/state.json +``` + +The broker refuses an existing parent directory whose mode is not exactly `0700`, an existing state file not at `0600`, corrupt/incompatible state, or an already-existing socket path. After bind it sets the socket to `0600`. It never silently unlinks a pre-existing socket. On normal termination it unlinks only the socket inode it created, so it does not remove a replacement path. + +Clients must complete the request boundary before waiting for a reply. After sending the single JSON object and its terminating newline, the client **MUST half-close the socket's write side** (`shutdown(SHUT_WR)` in POSIX clients; `socket.end()` in Node) and only then await the response. Merely calling `write()` and waiting is invalid: the broker waits for EOF to enforce the exact-one-frame contract and fails closed at its one-second deadline. Do not replace `end()` with `write()` in client helpers. A delayed second frame remains malformed and is rejected. + +There is no automated recovery workflow in WI-1. After a crash, preserve the protected state file and restart only after verifying that no broker owns the socket. A leftover socket requires an operator to verify the owning service is stopped and remove that exact socket deliberately. Corrupt, oversized, symlinked, or non-regular state fails closed; do not overwrite it. Preserve it for incident review and establish new state only through an explicit operational decision, which invalidates prior sessions and tokens. + +## Security posture + +Directory `0700` plus socket/state `0600` is minimum same-principal hardening: it excludes other UIDs but does **not** stop the same UID from unlinking and counterfeiting the socket. It therefore does not close T-C same-UID replacement. The stronger deployment runs the broker as a distinct principal behind a protected service boundary whose clients cannot unlink or rebind the socket; that is the T-C-closing option. Server-side branch protection remains the irreducible backstop. diff --git a/docs/scratchpads/828-lease-broker.md b/docs/scratchpads/828-lease-broker.md new file mode 100644 index 0000000..ae19f24 --- /dev/null +++ b/docs/scratchpads/828-lease-broker.md @@ -0,0 +1,66 @@ +# WI-1 Scratchpad — Authenticated external lease broker + +- **Issue:** Gitea #828 +- **Milestone:** 188 — Compaction-Refresh Mechanism (M1: Claude + Pi) +- **Branch:** `feat/828-lease-broker` +- **Starting HEAD:** `d801d6c4c8a984d6a95033c49714210018d3d9a8` +- **Session role:** Orchestrator coordinating implementation; Mos retains merge authority. + +## Objective + +Implement the ratified WI-1 product lease broker under `packages/mosaic/`: Linux `SO_PEERCRED` identity, broker-minted logical session IDs, `(pid,starttime)` launcher anchors with per-hop `/proc` starttime revalidation, sibling-substitution rejection, same-PID runtime-generation revocation, crypto-RNG single-use token persistence, and protected Unix-socket posture. + +## Authority verification + +Verified before code on session start; all exact SHA-256 values matched: + +- BUILD-BRIEF: `89fdbc27ed0e5050dc7b52f3ef2ddaea691edf17fd89d51b15e26fb5ed47171b` +- SPEC-v5: `a6d07ade835758e8488ca10d3b0631caf0beb93ea3a6733631f151b0c2f01433` +- Ratification: `bac58319c9c4028b5b40e1129e0033cdb5a6b7b02033c25f06f4cb77d7779c67` +- P6 planner ruling: `b7bbb6ea6e8d9a5c3366993642ab4e4f65b961af04936dcac20bfbcdcbaf1a09` +- WI-0 Gate0 evidence: `5d418306fcc597fd514e500bee40d1509f0bf467e46ee13fc5c280ed8274759d` + +## Locked constraints + +- Build against the ratified design; do not re-derive it. +- Product code only in `packages/mosaic`; Gate0 Python probes are reference prototypes and are not shipped. +- Caller-supplied/asserted `session_id` is refused. +- Tokens use the operating-system CSPRNG via Python `secrets`; never `Math.random` or model output. +- Socket parent directory mode `0700`, socket mode `0600` minimum; document distinct-principal deployment as the stronger T-C-closing posture. +- Red-first TDD for six named cases; new-code coverage >=85%. +- No merge. PR must say `closes #828`; exact 40-character head handed to Mos for Opus-SECREV and independent review. + +## Plan + +1. Load security/testing/docs guidance and inspect existing `packages/mosaic` architecture. +2. Write the six required tests first and capture RED evidence. +3. Implement minimal broker modules and CLI/runtime integration necessary for product use. +4. Run focused tests with coverage, package gates, then full repository gates/suite. +5. Run author-side review/remediation, commit `closes #828`, queue guard, push, and open PR through Mosaic wrappers. +6. Send PR number + exact head SHA to `web1:mosaic-100`; stop without merging. + +## Risks / boundaries + +- Same-UID counterfeit socket replacement remains the disclosed T-C residual unless broker runs under a distinct principal; filesystem modes alone are minimum hardening, not a complete authenticity proof. +- `.mosaic/orchestrator/mission.json` and `.mosaic/orchestrator/session.lock` were already modified at session start and must not be included in this PR. +- Repository Woodpecker pipelines exist; CI is the canonical build path. No manual image build/deploy is in scope. + +## Progress / evidence + +- 2026-07-18 session start: mandatory mission files and orchestration guides loaded. +- STEP 0: all four authority hashes matched; artifacts read in full. +- Branch/HEAD confirmed; issue #828 open; Gate0 evidence hash confirmed. +- Initial RED: focused Vitest acceptance suite failed 11/11 because the product daemon did not exist; the expected missing-product failure was observed before implementation. +- Review-remediation RED: partial/zero-progress state writes, nested corrupt state, symlink state, canonical starttime, and duplicate-anchor generation behavior failed before their fixes. Real socket RED/GREEN runs were executed by the unrestricted parent harness because the delegated worker sandbox denies `AF_UNIX.bind()`. +- Product implementation added at `packages/mosaic/framework/tools/lease-broker/daemon.py`; Gate0 probe scripts were read as references but not copied or shipped. +- Independent Codex code review round 1 found 2 blockers + 1 should-fix (connection stall/crash, partial writes, packet-dependent framing); all were remediated with tests. +- Independent Codex code review round 2 found 2 blockers + 1 relevant should-fix (half-close contract ambiguity, incomplete persisted-state validation, symlink/non-regular state); all were remediated with tests and documentation. Pre-existing `.mosaic/*` session dirt remains excluded from the PR. +- Unrestricted focused situational suite: `35/35` GREEN. +- New Python product module coverage: `90%` (`356` statements, `36` missed), above the user-required 85%. +- Root typecheck and lint gates passed after remediation; final package/full-suite/format evidence is recorded below before push. + +## Coordinator handoff requirements + +1. Mandatory Opus-SECREV on the exact PR head; no GPT/terra substitute. +2. Independent exact-head code review and exact-head RoR before Mos-authorized merge. +3. Mos retains merge authority; this WI author stops after PR + full 40-character head handoff. diff --git a/packages/mosaic/framework/tools/lease-broker/daemon.py b/packages/mosaic/framework/tools/lease-broker/daemon.py new file mode 100644 index 0000000..b909689 --- /dev/null +++ b/packages/mosaic/framework/tools/lease-broker/daemon.py @@ -0,0 +1,446 @@ +#!/usr/bin/env python3 +"""Mosaic external lease broker for Linux SO_PEERCRED authenticated clients.""" + +from __future__ import annotations + +import argparse +import errno +import json +import os +import secrets +import signal +import socket +import stat +import struct +import sys +import time +from pathlib import Path +from typing import Final + +MAX_FRAME: Final = 64 * 1024 +MAX_STATE: Final = 4 * 1024 * 1024 +STATE_VERSION: Final = 1 +CONNECTION_DEADLINE_SECONDS: Final = 1.0 +HEX_256_LENGTH: Final = 64 + + +class BrokerFailure(Exception): + def __init__(self, code: str) -> None: + super().__init__(code) + self.code = code + + +def is_non_negative_integer(value: object) -> bool: + return type(value) is int and value >= 0 + + +def is_hex_256(value: object) -> bool: + return ( + isinstance(value, str) + and len(value) == HEX_256_LENGTH + and all(character in "0123456789abcdef" for character in value) + ) + + +def is_positive_decimal(value: object) -> bool: + return ( + isinstance(value, str) + and len(value) > 0 + and value[0] in "123456789" + and all(character in "0123456789" for character in value) + ) + + +def valid_binding(binding: object) -> bool: + if not isinstance(binding, dict): + return False + required = {"compaction_epoch", "request_epoch", "h_source", "h_payload", "schema_version"} + if set(binding) != required: + return False + if not all( + is_non_negative_integer(binding[field]) + for field in ("compaction_epoch", "request_epoch", "schema_version") + ): + return False + return all( + is_hex_256(binding[field]) + for field in ("h_source", "h_payload") + ) + + +def validate_state(value: object) -> dict[str, object]: + if not isinstance(value, dict) or set(value) != {"version", "sessions", "tokens"}: + raise BrokerFailure("STATE_INTEGRITY") + if type(value["version"]) is not int or value["version"] != STATE_VERSION: + raise BrokerFailure("STATE_INTEGRITY") + sessions = value["sessions"] + tokens = value["tokens"] + if not isinstance(sessions, dict) or not isinstance(tokens, dict): + raise BrokerFailure("STATE_INTEGRITY") + + anchors: set[tuple[int, str]] = set() + for session_id, session in sessions.items(): + if not is_hex_256(session_id) or not isinstance(session, dict): + raise BrokerFailure("STATE_INTEGRITY") + if set(session) != {"anchor_pid", "anchor_starttime", "runtime_generation"}: + raise BrokerFailure("STATE_INTEGRITY") + anchor_pid = session["anchor_pid"] + anchor_starttime = session["anchor_starttime"] + if type(anchor_pid) is not int or anchor_pid <= 0: + raise BrokerFailure("STATE_INTEGRITY") + if not is_positive_decimal(anchor_starttime): + raise BrokerFailure("STATE_INTEGRITY") + if not is_non_negative_integer(session["runtime_generation"]): + raise BrokerFailure("STATE_INTEGRITY") + anchor = (anchor_pid, anchor_starttime) + if anchor in anchors: + raise BrokerFailure("STATE_INTEGRITY") + anchors.add(anchor) + + for token_value, token in tokens.items(): + if not is_hex_256(token_value) or not isinstance(token, dict): + raise BrokerFailure("STATE_INTEGRITY") + if set(token) != {"session_id", "runtime_generation", "binding", "consumed"}: + raise BrokerFailure("STATE_INTEGRITY") + session_id = token["session_id"] + generation = token["runtime_generation"] + session = sessions.get(session_id) if isinstance(session_id, str) else None + if not is_hex_256(session_id) or not isinstance(session, dict): + raise BrokerFailure("STATE_INTEGRITY") + if not is_non_negative_integer(generation): + raise BrokerFailure("STATE_INTEGRITY") + if generation > session["runtime_generation"]: + raise BrokerFailure("STATE_INTEGRITY") + if not valid_binding(token["binding"]) or type(token["consumed"]) is not bool: + raise BrokerFailure("STATE_INTEGRITY") + return value + + +def proc_node(pid: int) -> dict[str, int | str]: + try: + text = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8") + except (FileNotFoundError, PermissionError, ProcessLookupError) as exc: + raise BrokerFailure("PID_UNAVAILABLE") from exc + close = text.rfind(")") + fields = text[close + 2 :].split() + if close < 0 or len(fields) < 20: + raise BrokerFailure("PROC_STAT_INVALID") + return {"pid": pid, "ppid": int(fields[1]), "starttime": fields[19]} + + +def verified_ancestry(peer_pid: int, anchor_pid: int, anchor_starttime: str) -> bool: + chain: list[dict[str, int | str]] = [] + seen: set[int] = set() + current = peer_pid + while current > 0 and current not in seen: + seen.add(current) + node = proc_node(current) + chain.append(node) + if current == anchor_pid: + if node["starttime"] != anchor_starttime: + return False + break + current = int(node["ppid"]) + else: + return False + if int(chain[-1]["pid"]) != anchor_pid: + return False + for original in chain: + repeated = proc_node(int(original["pid"])) + if repeated["starttime"] != original["starttime"]: + raise BrokerFailure("PID_STARTTIME_RACE") + return True + + +def secure_parent(path: Path) -> None: + parent = path.parent + if not parent.is_dir() or stat.S_IMODE(parent.stat().st_mode) != 0o700: + raise BrokerFailure("INSECURE_PARENT_MODE") + + +class StateStore: + def __init__(self, path: Path) -> None: + self.path = path + secure_parent(path) + self.value: dict[str, object] = {"version": STATE_VERSION, "sessions": {}, "tokens": {}} + flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0) + try: + descriptor = os.open(path, flags) + except FileNotFoundError: + return + except OSError as exc: + raise BrokerFailure("STATE_INTEGRITY") from exc + try: + metadata = os.fstat(descriptor) + if not stat.S_ISREG(metadata.st_mode): + raise BrokerFailure("STATE_INTEGRITY") + if stat.S_IMODE(metadata.st_mode) != 0o600: + raise BrokerFailure("INSECURE_STATE_MODE") + if metadata.st_size > MAX_STATE: + raise BrokerFailure("STATE_INTEGRITY") + chunks = bytearray() + while len(chunks) <= MAX_STATE: + chunk = os.read(descriptor, min(64 * 1024, MAX_STATE + 1 - len(chunks))) + if not chunk: + break + chunks.extend(chunk) + if len(chunks) > MAX_STATE: + raise BrokerFailure("STATE_INTEGRITY") + try: + loaded = json.loads(chunks) + except (json.JSONDecodeError, UnicodeDecodeError) as exc: + raise BrokerFailure("STATE_INTEGRITY") from exc + self.value = validate_state(loaded) + except OSError as exc: + raise BrokerFailure("STATE_INTEGRITY") from exc + finally: + os.close(descriptor) + + def sessions(self) -> dict[str, dict[str, object]]: + sessions = self.value.get("sessions") + if not isinstance(sessions, dict): + raise BrokerFailure("STATE_INTEGRITY") + return sessions + + def tokens(self) -> dict[str, dict[str, object]]: + tokens = self.value.get("tokens") + if not isinstance(tokens, dict): + raise BrokerFailure("STATE_INTEGRITY") + return tokens + + def commit(self) -> None: + temporary = self.path.with_name(f".{self.path.name}.{os.getpid()}.tmp") + descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + replaced = False + try: + try: + payload = ( + json.dumps(self.value, sort_keys=True, separators=(",", ":")) + "\n" + ).encode() + remaining = memoryview(payload) + while remaining: + written = os.write(descriptor, remaining) + if written == 0: + raise OSError(errno.EIO, "state write made no progress") + remaining = remaining[written:] + os.fsync(descriptor) + finally: + os.close(descriptor) + os.replace(temporary, self.path) + replaced = True + directory = os.open(self.path.parent, os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(directory) + finally: + os.close(directory) + finally: + if not replaced: + try: + temporary.unlink() + except FileNotFoundError: + pass + + +class Broker: + def __init__(self, store: StateStore) -> None: + self.store = store + + def authenticate(self, peer_pid: int, request: dict[str, object]) -> tuple[str, dict[str, object]]: + session_id = request.get("session_id") + generation = request.get("runtime_generation") + if not isinstance(session_id, str) or not is_non_negative_integer(generation): + raise BrokerFailure("INVALID_IDENTITY") + session = self.store.sessions().get(session_id) + if not isinstance(session, dict): + raise BrokerFailure("UNKNOWN_SESSION") + anchor_pid = session.get("anchor_pid") + anchor_starttime = session.get("anchor_starttime") + current_generation = session.get("runtime_generation") + if ( + type(anchor_pid) is not int + or not isinstance(anchor_starttime, str) + or not is_non_negative_integer(current_generation) + ): + raise BrokerFailure("STATE_INTEGRITY") + if not verified_ancestry(peer_pid, anchor_pid, anchor_starttime): + raise BrokerFailure("ANCESTRY_MISMATCH") + if generation < current_generation: + raise BrokerFailure("STALE_GENERATION") + if generation > current_generation: + session["runtime_generation"] = generation + self.revoke_session_tokens(session_id) + self.store.commit() + return session_id, session + + def session_for_anchor( + self, anchor_pid: int, anchor_starttime: str + ) -> tuple[str, dict[str, object]] | None: + for session_id, session in self.store.sessions().items(): + if ( + session["anchor_pid"] == anchor_pid + and session["anchor_starttime"] == anchor_starttime + ): + return session_id, session + return None + + def revoke_session_tokens(self, session_id: str) -> None: + for token in self.store.tokens().values(): + if token["session_id"] == session_id: + token["consumed"] = True + + def handle(self, peer: tuple[int, int, int], request: dict[str, object]) -> dict[str, object]: + peer_pid, peer_uid, peer_gid = peer + action = request.get("action") + if action == "register_anchor": + if "session_id" in request: + raise BrokerFailure("CALLER_SESSION_ID_REFUSED") + generation = request.get("runtime_generation") + if not is_non_negative_integer(generation): + raise BrokerFailure("INVALID_GENERATION") + anchor = proc_node(peer_pid) + anchor_starttime = str(anchor["starttime"]) + existing = self.session_for_anchor(peer_pid, anchor_starttime) + if existing is None: + session_id = secrets.token_hex(32) + self.store.sessions()[session_id] = { + "anchor_pid": peer_pid, + "anchor_starttime": anchor_starttime, + "runtime_generation": generation, + } + self.store.commit() + else: + session_id, session = existing + current_generation = session["runtime_generation"] + if generation < current_generation: + raise BrokerFailure("STALE_GENERATION") + if generation > current_generation: + session["runtime_generation"] = generation + self.revoke_session_tokens(session_id) + self.store.commit() + return {"ok": True, "session_id": session_id, "peer": {"pid": peer_pid, "uid": peer_uid, "gid": peer_gid, "starttime": anchor["starttime"]}} + if action == "authenticate": + self.authenticate(peer_pid, request) + return {"ok": True} + if action == "mint_token": + session_id, _ = self.authenticate(peer_pid, request) + binding = request.get("binding") + if not valid_binding(binding): + raise BrokerFailure("INVALID_BINDING") + token = secrets.token_hex(32) + self.store.tokens()[token] = { + "session_id": session_id, + "runtime_generation": request["runtime_generation"], + "binding": binding, + "consumed": False, + } + self.store.commit() + return {"ok": True, "token": token} + if action == "consume_token": + session_id, _ = self.authenticate(peer_pid, request) + token_value = request.get("token") + token = self.store.tokens().get(token_value) if isinstance(token_value, str) else None + if not isinstance(token, dict) or token.get("session_id") != session_id or token.get("runtime_generation") != request.get("runtime_generation") or token.get("consumed") is not False: + raise BrokerFailure("TOKEN_REPLAY") + token["consumed"] = True + self.store.commit() + return {"ok": True} + raise BrokerFailure("UNKNOWN_ACTION") + + +def read_frame(connection: socket.socket, deadline: float) -> dict[str, object]: + data = bytearray() + while len(data) <= MAX_FRAME: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise BrokerFailure("MALFORMED_REQUEST") + connection.settimeout(remaining) + chunk = connection.recv(min(4096, MAX_FRAME + 1 - len(data))) + if not chunk: + break + data.extend(chunk) + if len(data) > MAX_FRAME or not data.endswith(b"\n") or data.count(b"\n") != 1: + raise BrokerFailure("MALFORMED_REQUEST") + try: + value = json.loads(data) + except (json.JSONDecodeError, UnicodeDecodeError) as exc: + raise BrokerFailure("MALFORMED_REQUEST") from exc + if not isinstance(value, dict): + raise BrokerFailure("MALFORMED_REQUEST") + return value + + +def serve(socket_path: Path, state_path: Path) -> None: + secure_parent(socket_path) + if socket_path.exists() or socket_path.is_symlink(): + raise BrokerFailure("SOCKET_ALREADY_EXISTS") + store = StateStore(state_path) + broker = Broker(store) + server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + server.bind(str(socket_path)) + os.chmod(socket_path, 0o600) + owned = (socket_path.stat().st_dev, socket_path.stat().st_ino) + stopping = False + + def stop(_signum: int, _frame: object) -> None: + nonlocal stopping + stopping = True + server.close() + + signal.signal(signal.SIGTERM, stop) + signal.signal(signal.SIGINT, stop) + server.listen(16) + print("READY", flush=True) + try: + while not stopping: + try: + connection, _ = server.accept() + except OSError: + if stopping: + break + raise + with connection: + deadline = time.monotonic() + CONNECTION_DEADLINE_SECONDS + try: + raw = connection.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12) + peer = struct.unpack("3i", raw) + request = read_frame(connection, deadline) + except BrokerFailure as exc: + reply = {"ok": False, "code": exc.code} + except OSError: + continue + else: + try: + reply = broker.handle(peer, request) + except BrokerFailure as exc: + reply = {"ok": False, "code": exc.code} + try: + remaining = deadline - time.monotonic() + if remaining <= 0: + continue + connection.settimeout(remaining) + connection.sendall((json.dumps(reply, separators=(",", ":")) + "\n").encode()) + except OSError: + continue + finally: + try: + current = socket_path.stat() + if (current.st_dev, current.st_ino) == owned: + socket_path.unlink() + except FileNotFoundError: + pass + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--socket", required=True, type=Path) + parser.add_argument("--state", required=True, type=Path) + arguments = parser.parse_args() + serve(arguments.socket, arguments.state) + + +if __name__ == "__main__": + try: + main() + except BrokerFailure as failure: + print(failure.code, file=sys.stderr) + raise SystemExit(1)