From 1c9cd7a0302425a6a64d77a9938d6b6bf244e144 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miko=C5=82aj=20Mielczarek?= <86711023+mikomiel@users.noreply.github.com> Date: Fri, 13 Nov 2026 17:07:53 +0100 Subject: [PATCH] feat(pve): extract agent bridge, client and execution worker Implement OpenSpec task 4.1 (issue #8): the agent-side execution components move into the product as explicit-configuration libraries. - privilege_broker.pve.client: mTLS broker HTTP client; broker URL, CA, client cert/key and timeout are required constructor input (no environment defaults, no built-in addresses). Transient failures (network, timeout, HTTP 429/502/503/504) raise BrokerUnavailable; final HTTP refusals raise BrokerHTTPError. - privilege_broker.pve.bridge: agent-facing request creation with an ephemeral Ed25519 key (removed on any failure), strict create-response validation (canonical UUID, pending status, integer expiry) and pending-input publication. read_status returns result/pending/unknown. - privilege_broker.pve.worker: poll-and-dispatch worker preserving the broker wire contract (lease fetched once, one-time certificate, {action, ct_id, lease, signature} on stdin over fixed non-interactive SSH options). Local expiry plus grace bounds retries; transient pre-dispatch broker failures are retried on the next poll; a consumed certificate (HTTP 409) is a terminal error with no re-mint; wrong-node answers (exit 3 + not_on_this_node) fail over to the next configured node with the same certificate; a transport timeout is unknown and never retried; key/certificate material never appears in results or logs. - Synthetic tests cover classification, node failover, bounded retries, local expiry, terminal states and secret non-leakage. Docs updated (architecture, configuration); task 4.1 ticked with the issue link. Verification: uv sync --frozen, ruff check/format, pytest (337 passed), examples/pve_authorization.py, uv build, check_estate_identifiers.py, openspec validate --strict, git diff --check: all PASS. --- docs/architecture.md | 27 +- docs/configuration.md | 75 ++- .../changes/extract-pve-foundation/tasks.md | 2 +- privilege_broker/pve/__init__.py | 2 +- privilege_broker/pve/bridge.py | 151 +++++++ privilege_broker/pve/client.py | 142 ++++++ privilege_broker/pve/worker.py | 332 ++++++++++++++ tests/test_pve_bridge.py | 217 +++++++++ tests/test_pve_client.py | 175 +++++++ tests/test_pve_worker.py | 427 ++++++++++++++++++ 10 files changed, 1537 insertions(+), 13 deletions(-) create mode 100644 privilege_broker/pve/bridge.py create mode 100644 privilege_broker/pve/client.py create mode 100644 privilege_broker/pve/worker.py create mode 100644 tests/test_pve_bridge.py create mode 100644 tests/test_pve_client.py create mode 100644 tests/test_pve_worker.py diff --git a/docs/architecture.md b/docs/architecture.md index 1296706..abf09c0 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -10,6 +10,9 @@ Notification delivery is separate and never an authorization decision. - `privilege_broker.pve.authorization`: immutable requests, proxy identity extraction, signed leases and restricted certificate argv. - `privilege_broker.pve.service`: SQLite approval state, real WebAuthn registration/authentication, browser UI and one-time certificate fetch. - `privilege_broker.notifications`: server-owned Discord delivery with credential-safe errors and inert agent text. +- `privilege_broker.pve.client`: mTLS broker HTTP client; the endpoint and trust material are explicit constructor input and transient failures classify separately from final refusals. +- `privilege_broker.pve.bridge`: agent-side request creation with an ephemeral Ed25519 key, strict response validation and pending-input publication. +- `privilege_broker.pve.worker`: poll-and-dispatch worker that fetches the lease and the one-time certificate, tries the configured nodes in order and writes terminal result files. The approval service requires explicitly supplied store, signer and target policy. Its environment factory requires private deployment configuration; @@ -20,8 +23,19 @@ and service isolation are owned by deployment, not by request bodies. Browser approval signs a short-lived PVE lease but returns only approval status. The bound agent can fetch that lease and a restricted SSH certificate. The signer runs only local `ssh-keygen`, not an SSH connection or PVE command. -No execution worker, target dispatcher or outcome ingestion exists here yet. -An `approved` request is not an executed operation. +The bridge and worker exist; the target dispatcher, durable claim store and +outcome ingestion do not (OpenSpec tasks 4.2 and 5.1). An `approved` request +is not an executed operation. + +The bridge, client and worker take explicit configuration only: they read no +environment variable and carry no built-in address, path or identity, so a +missing value fails closed. The worker keeps the broker wire contract +unchanged: the lease is fetched once, the certificate is fetched once and the +dispatcher receives `{"action", "ct_id", "lease", "signature"}` on stdin over +fixed non-interactive SSH options. Terminal results are `succeeded`, `failed`, +`unknown` (transport timeout; never retried), `error`, `denied`, `cancelled` +and `expired`. Key and certificate material never appears in results or logs, +and a consumed certificate is never re-requested. ## Compatibility and extraction adaptations @@ -60,8 +74,8 @@ delivery is not durable. Product acceptance remains incomplete. ## Remaining product stages The active [OpenSpec change](../openspec/changes/extract-pve-foundation/proposal.md) -owns authenticated agent lifecycle/outcome interfaces, the bridge, worker, -dispatcher and end-to-end acceptance. Consumer pinning, deployment and live +owns the target dispatcher and replay store, the outcome interfaces and +end-to-end acceptance. Consumer pinning, deployment and live PVE acceptance are separate gates. No current consumer has switched. PVE is the only initial provider. There is no dynamic plugin system, arbitrary @@ -75,11 +89,14 @@ The extraction uses the owned ai-agents snapshot | Original file | Product boundary | | --- | --- | -| `MCPs/servers/_pve_common.py` | Policy and canonicalization only; no transport client | +| `MCPs/servers/_pve_common.py` | Policy, canonicalization and the mTLS broker client; no transport server | | `MCPs/servers/pve_ssh_broker.py` | Authorization and certificate argv | | `MCPs/servers/pve_ssh_broker_service.py` | Approval store, service, passkey UI and certificate signer | +| `MCPs/servers/pve_broker_mcp.py` | Agent bridge: request validation, ephemeral key, input publication | +| `MCPs/servers/pve_broker_worker.py` | Poll-and-dispatch worker: lease/certificate fetch, node failover, result files | | `MCPs/servers/_discord_notify.py` | Server-owned notifications, adapted to standalone HTTP and UTC | | `MCPs/tests/test_pve_ssh_broker*.py` | Ported synthetic contract tests, supplemented by real local crypto tests | +| `MCPs/tests/test_pve_broker_mcp.py`, `MCPs/tests/test_pve_broker_worker.py` | Ported synthetic bridge and worker tests | Private source history, deployment configuration, trust roots, credentials and runbooks are not copied. Existing consumers retain their current source diff --git a/docs/configuration.md b/docs/configuration.md index 03d390f..38848d1 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -1,8 +1,9 @@ # PVE configuration -This pre-release stage supplies authorization, approval and certificate signing, -not an executor. Deployment, credentials, TLS and consumer cutover remain separate -operator decisions. No example grants access to a real target. +This pre-release stage supplies authorization, approval, certificate signing +and the agent-side bridge and worker, but not the target dispatcher. Deployment, +credentials, TLS and consumer cutover remain separate operator decisions. No +example grants access to a real target. ## Target policy @@ -88,6 +89,68 @@ passkey; denial grants no authority. Protect these URLs against unwanted cancellation and metadata disclosure. There is no authenticated generic execution-outcome API yet. +## Agent bridge and worker + +The bridge and worker are libraries with explicit configuration only. They read +no environment variable and carry no built-in address, path or identity; +deployment adapters map their own environment onto these required arguments, +and a missing value fails closed. + +```python +from pathlib import Path +from privilege_broker.pve.bridge import ApprovalBridge, BridgeConfig +from privilege_broker.pve.client import BrokerClient, BrokerClientConfig +from privilege_broker.pve.worker import WorkerConfig, process_pending_once + +client = BrokerClient( + BrokerClientConfig( + broker_url="https://broker.example.test", # required https:// URL + cert_path="/etc/example/agent-cert.pem", # required mTLS client certificate + key_path="/etc/example/agent-key.pem", # required mTLS client key + ca_path="/etc/example/broker-ca.pem", # optional trust root + ) +) +bridge = ApprovalBridge( + BridgeConfig( + state_dir=Path("/var/lib/example-broker"), # holds input/ results/ keys/ + public_url="https://broker.example.test", # approval-page base URL + requester="example-agent", # 1-100 characters + ) +) +worker_config = WorkerConfig( + state_dir=Path("/var/lib/example-broker"), + targets=("ops@node-a", "ops@node-b"), # tried in order + ssh_port=8922, + known_hosts_path="/etc/example/known_hosts", # pinned host keys + poll_interval=5, + ssh_timeout=240, + expiry_grace=60, +) +``` + +`bridge.request_mutation(action, ct_id, reason, client)` validates the typed +action, target and reason, writes an ephemeral Ed25519 key (removed again on any +failure), creates the broker request and publishes +`input/{request_id}.json`; it returns the `approval_url`. `bridge.read_status` +returns the worker result, `pending` or `unknown`. + +The worker polls `input/` once per `process_pending_once(client, worker_config)` +call (or forever via `run`). A request past its deadline plus `expiry_grace` is +expired locally without asking the broker, which bounds all retries. Transient +broker failures (network, timeout, HTTP 429/502/503/504) before dispatch keep +the request for the next poll; any final refusal is terminal — in particular +HTTP 409 after a lost certificate reply ends the request as `error` and never +requests a second certificate. Dispatch uses fixed non-interactive SSH options +(`BatchMode`, `IdentitiesOnly`, no TTY, pinned known hosts, configured port) and +sends `{"action", "ct_id", "lease", "signature"}` on stdin, preserving the +dispatcher wire contract. A node that does not host the target answers exit 3 +with `not_on_this_node` and the next configured node is tried with the same +certificate; a transport timeout is `unknown` and is never retried. Results in +`results/{request_id}.json` are `succeeded`, `failed`, `unknown`, `error`, +`denied`, `cancelled` or `expired`; key and certificate material never appears +in results or logs. The target dispatcher and its durable claim store remain +OpenSpec task 4.2. + ## Certificate source restriction The library and service require an explicit source restriction: @@ -106,9 +169,9 @@ The builder only returns argv; the configured service signer invokes local `ssh-keygen` with that argv and a timeout. It does not connect to PVE. A certificate is safe only with the trusted typed dispatcher and lease -verification on the target. Those execution components are not extracted yet. -Do not treat certificate issuance as an operation outcome or enable a general -shell while waiting for the dispatcher. +verification on the target. The target dispatcher and its claim store are not +extracted yet (OpenSpec task 4.2). Do not treat certificate issuance as an +operation outcome or enable a general shell while waiting for the dispatcher. ## Operator notifications diff --git a/openspec/changes/extract-pve-foundation/tasks.md b/openspec/changes/extract-pve-foundation/tasks.md index 408b8aa..8158c35 100644 --- a/openspec/changes/extract-pve-foundation/tasks.md +++ b/openspec/changes/extract-pve-foundation/tasks.md @@ -20,7 +20,7 @@ ## 4. PVE execution components -- [ ] 4.1 Extract the agent bridge, client and worker using explicit endpoints, trust and state configuration; verify pre-dispatch transient retry, expiry, lost-certificate terminal failure and no secret-bearing output. +- [x] 4.1 Extract the agent bridge, client and worker using explicit endpoints, trust and state configuration; verify pre-dispatch transient retry, expiry, lost-certificate terminal failure and no secret-bearing output. [Issue #8](https://github.com/mikomiel/privilege-broker/issues/8). - [ ] 4.2 Extract the typed dispatcher and replay store; verify fixed argv, current policy, source/certificate constraints, wrong-node routing, concurrent single claim and restart replay protection. - [ ] 4.3 Preserve policy-bounded snapshot pruning and its approval display; verify synthetic old/unrelated/retained snapshot cases and update PVE operation documentation. diff --git a/privilege_broker/pve/__init__.py b/privilege_broker/pve/__init__.py index 4887421..821fab5 100644 --- a/privilege_broker/pve/__init__.py +++ b/privilege_broker/pve/__init__.py @@ -1 +1 @@ -"""PVE policy, canonical authorization and SSH certificate constraints.""" +"""PVE policy, canonical authorization, approval service and the agent-side bridge, client and worker.""" diff --git a/privilege_broker/pve/bridge.py b/privilege_broker/pve/bridge.py new file mode 100644 index 0000000..f0e8ed1 --- /dev/null +++ b/privilege_broker/pve/bridge.py @@ -0,0 +1,151 @@ +"""Agent-facing approval request bridge. + +Creates typed broker requests with an ephemeral Ed25519 key and records the +pending input file the worker consumes. This is a library: it starts no +server, reads no environment and carries no deployment default. The approval +URL it returns points at the operator-facing page of the configured public +URL. +""" + +from __future__ import annotations + +import json +import re +import uuid +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from cryptography.hazmat.primitives.serialization import Encoding, NoEncryption, PrivateFormat, PublicFormat + +from .client import BrokerClient +from .common import TYPED_ACTIONS + +_UUID_RE = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$") + + +@dataclass(frozen=True) +class BridgeConfig: + """Explicit bridge configuration; no deployment defaults exist.""" + + #: Persistent state root; holds ``input/``, ``results/`` and ``keys/``. + state_dir: Path + #: ``https://`` base URL of the broker's public approval pages. + public_url: str + #: Requester name sent in every broker request. + requester: str + + def __post_init__(self) -> None: + if isinstance(self.state_dir, str): + object.__setattr__(self, "state_dir", Path(self.state_dir)) + if not self.public_url.startswith("https://"): + raise ValueError("public_url must be an https:// URL") + if not isinstance(self.requester, str) or not 1 <= len(self.requester.strip()) <= 100: + raise ValueError("requester must be 1-100 characters") + + +class ApprovalBridge: + """Creates approval-gated mutation requests and reads their local state. + + Nothing executes here: a request only exists at the broker until a human + approves it, and the worker turns an approval into a dispatch. + """ + + def __init__(self, config: BridgeConfig) -> None: + self._config = config + + @property + def _input_dir(self) -> Path: + return self._config.state_dir / "input" + + @property + def _results_dir(self) -> Path: + return self._config.state_dir / "results" + + @property + def _keys_dir(self) -> Path: + return self._config.state_dir / "keys" + + def _validated_create_response(self, response: object) -> tuple[str, int]: + """Validate the broker create response before any filesystem publication.""" + if not isinstance(response, dict): + raise ValueError("broker create response must be an object") + request_id = response.get("request_id") + if not isinstance(request_id, str) or not _UUID_RE.fullmatch(request_id): + raise ValueError("broker create response request_id must be a canonical UUID") + if response.get("status") != "pending": + raise ValueError("broker create response status must be pending") + expires_at = response.get("expires_at") + if type(expires_at) is not int: + raise ValueError("broker create response expires_at must be an integer") + return request_id, expires_at + + def request_mutation(self, action: str, ct_id: int, reason: str, client: BrokerClient) -> dict[str, Any]: + """Request an approval-gated typed operation; nothing runs until approved. + + Returns ``request_id``, ``status``, ``expires_at`` and the + ``approval_url`` to hand to the operator. The ephemeral private key is + removed again if the broker call or the response validation fails. + """ + if action not in TYPED_ACTIONS: + raise ValueError(f"action must be one of {sorted(TYPED_ACTIONS)}") + if type(ct_id) is not int or ct_id < 1: + raise ValueError("ct_id must be a positive integer") + if not isinstance(reason, str) or not reason.strip() or len(reason) > 500: + raise ValueError("reason must be 1-500 characters") + + private_key = Ed25519PrivateKey.generate() + key_id = uuid.uuid4().hex + self._keys_dir.mkdir(mode=0o700, parents=True, exist_ok=True) + key_path = self._keys_dir / f"{key_id}.key" + key_path.write_bytes(private_key.private_bytes(Encoding.PEM, PrivateFormat.OpenSSH, NoEncryption())) + key_path.chmod(0o600) + public_key = private_key.public_key().public_bytes(Encoding.OpenSSH, PublicFormat.OpenSSH).decode("ascii") + + try: + response = client.request( + "POST", + "/v1/broker/requests", + { + "action": action, + "ct_id": ct_id, + "reason": reason, + "requester": self._config.requester, + "ssh_public_key": public_key, + }, + ) + request_id, expires_at = self._validated_create_response(response) + except Exception: + # The ephemeral private key never survives a failed request. + key_path.unlink(missing_ok=True) + raise + + self._input_dir.mkdir(mode=0o700, parents=True, exist_ok=True) + input_data = { + "request_id": request_id, + "action": action, + "ct_id": ct_id, + "reason": reason, + "public_key": public_key, + "key_id": key_id, + "expires_at": expires_at, + } + (self._input_dir / f"{request_id}.json").write_text(json.dumps(input_data), encoding="utf-8") + return { + "request_id": request_id, + "status": "pending", + "expires_at": expires_at, + "approval_url": f"{self._config.public_url.rstrip('/')}/v1/broker/requests/{request_id}/approval", + } + + def read_status(self, request_id: str) -> dict[str, Any]: + """Return the worker's terminal result, ``pending``, or ``unknown``.""" + if not isinstance(request_id, str) or not _UUID_RE.fullmatch(request_id): + raise ValueError("request_id must be a canonical UUID") + result_file = self._results_dir / f"{request_id}.json" + if result_file.exists(): + return json.loads(result_file.read_text(encoding="utf-8")) + if (self._input_dir / f"{request_id}.json").exists(): + return {"status": "pending"} + return {"status": "unknown", "error": "request not found or expired"} diff --git a/privilege_broker/pve/client.py b/privilege_broker/pve/client.py new file mode 100644 index 0000000..8b3aa9f --- /dev/null +++ b/privilege_broker/pve/client.py @@ -0,0 +1,142 @@ +"""mTLS HTTP client for the approval broker, with explicit configuration only. + +The client never reads deployment state from the environment: the broker URL, +the trust root and the client certificate pair are constructor arguments, so a +missing value fails closed instead of silently dialling a built-in address. +""" + +from __future__ import annotations + +import json +import ssl +import urllib.error +import urllib.request +from dataclasses import dataclass +from typing import Any + +#: Statuses that say "later", not "no": rate limiting and an unavailable +#: upstream behind the broker's TLS front end. +_RETRYABLE_STATUSES: frozenset[int] = frozenset({429, 502, 503, 504}) + +#: Upper bound on a response body the client reads; anything larger is refused. +_MAX_BODY_BYTES = 1024 * 1024 + +#: Error detail kept in exceptions; the broker's own message, truncated. +_DETAIL_LIMIT = 1024 + + +class BrokerUnavailable(RuntimeError): + """The broker was not reached, or asked the caller to come back later. + + A network or timeout failure, or HTTP 429, 502, 503 or 504. The same call + may be repeated later. + """ + + +class BrokerHTTPError(RuntimeError): + """The broker refused the call with any other HTTP status.""" + + def __init__(self, status: int | None, message: str) -> None: + super().__init__(message) + self.status = status + + +@dataclass(frozen=True) +class BrokerClientConfig: + """Explicit broker endpoint and mTLS trust material. + + No field carries a deployment default: the caller (bridge, worker or a + deployment adapter) supplies every value. + """ + + #: ``https://`` URL of the broker's TLS front end. + broker_url: str + #: Path to the PEM client certificate presented to the broker. + cert_path: str + #: Path to the PEM private key matching ``cert_path``. + key_path: str + #: Optional path to the PEM CA bundle that signed the broker's certificate. + ca_path: str | None = None + #: Per-request timeout in seconds. + timeout: float = 15.0 + + def __post_init__(self) -> None: + if not self.broker_url.startswith("https://"): + raise ValueError("broker_url must be an https:// URL") + if not self.cert_path or not self.key_path: + raise ValueError("mTLS client certificate and key paths are required") + if self.timeout <= 0: + raise ValueError("timeout must be positive") + + +def _truncate(text: str, limit: int = _DETAIL_LIMIT) -> str: + return text if len(text) <= limit else text[:limit] + "..." + + +class BrokerClient: + """mTLS HTTP client for the PVE approval broker. + + Failures are classified so callers can distinguish a transient + ``BrokerUnavailable`` (retry later) from a final ``BrokerHTTPError``. + """ + + def __init__( + self, + config: BrokerClientConfig, + *, + ssl_context: ssl.SSLContext | None = None, + opener: Any = None, + ) -> None: + self._config = config + if ssl_context is None: + ssl_context = ssl.create_default_context(cafile=config.ca_path or None) + ssl_context.load_cert_chain(config.cert_path, config.key_path) + ssl_context.check_hostname = True + self._context = ssl_context + if opener is None: + # ProxyHandler({}) disables environment proxies: the mTLS channel + # must reach the broker directly. + opener = urllib.request.build_opener( + urllib.request.ProxyHandler({}), + urllib.request.HTTPSHandler(context=ssl_context), + ) + self._opener = opener + self._broker_url = config.broker_url.rstrip("/") + + def request(self, method: str, path: str, payload: dict[str, Any] | None = None) -> dict[str, Any]: + """Send one JSON request and return the decoded object. + + Raises ``BrokerUnavailable`` for transient failures, ``BrokerHTTPError`` + for final HTTP refusals and ``RuntimeError`` for unusable bodies. + """ + if not path.startswith("/"): + raise ValueError("path must start with '/'") + body = json.dumps(payload).encode("utf-8") if payload is not None else None + request = urllib.request.Request( + f"{self._broker_url}{path}", + method=method, + data=body, + headers={"Content-Type": "application/json"} if body else {}, + ) + try: + with self._opener.open(request, timeout=self._config.timeout) as response: + raw = response.read(_MAX_BODY_BYTES + 1) + except urllib.error.HTTPError as exc: + try: + detail = exc.read().decode("utf-8", errors="replace") + except OSError: + detail = "" + message = f"broker HTTP {exc.status}: {_truncate(detail)}" + if exc.status in _RETRYABLE_STATUSES: + raise BrokerUnavailable(message) from exc + raise BrokerHTTPError(exc.status, message) from exc + except (urllib.error.URLError, TimeoutError, OSError) as exc: + raise BrokerUnavailable(f"broker unavailable: {type(exc).__name__}") from exc + if len(raw) > _MAX_BODY_BYTES: + raise RuntimeError("broker response too large") + try: + result = json.loads(raw.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise RuntimeError("broker returned a non-JSON body") from exc + # A body that is not a JSON object carries no usable answer. + return result if isinstance(result, dict) else {} diff --git a/privilege_broker/pve/worker.py b/privilege_broker/pve/worker.py new file mode 100644 index 0000000..c20e8a8 --- /dev/null +++ b/privilege_broker/pve/worker.py @@ -0,0 +1,332 @@ +"""Poll-and-dispatch worker for approved broker requests. + +Polls the broker for requests whose input files exist in the state directory. +On approval: fetches the lease and the one-time SSH certificate, dispatches to +the typed dispatcher over SSH, and writes a result file. On denial, expiry or +cancellation: writes a terminal status. The target container may live on any +configured node; a node that does not host it answers ``not_on_this_node`` +(exit 3) without claiming the lease, and only then is the next node tried. + +Terminal result statuses written to ``results/{request_id}.json``:: + + succeeded dispatcher exited 0 + failed dispatcher exited non-zero + unknown ssh timed out; the operation may or may not have run + error the dispatch could not be carried out at all + denied | cancelled | expired decided at the broker, never dispatched + +A request past its deadline plus a grace period is expired locally, without +asking the broker, so a forgotten request cannot be polled forever. When the +lease or certificate call finds the broker unavailable (network failure, +timeout, HTTP 429/502/503/504) the request is kept and retried on the next +poll, bounded by that local expiry; any other refusal is final. The +certificate is issued once: if its response is lost after signing, the retry +gets HTTP 409 and the request ends as ``error`` -- a second certificate is +never issued for one approval, and an ambiguous dispatch is never repeated. + +State directory layout:: + + / + input/{request_id}.json written by the bridge + results/{request_id}.json written by this worker + keys/{key_id}.key ephemeral private key + keys/{key_id}-cert.pub SSH certificate + +All configuration is explicit constructor input; the module reads no +environment and starts no loop at import. +""" + +from __future__ import annotations + +import json +import logging +import subprocess +import time +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Protocol + +from .client import BrokerClient, BrokerUnavailable + +_LOG = logging.getLogger("privilege_broker.pve.worker") + +#: The dispatcher's exit code when the container is not on the node reached. +NOT_ON_THIS_NODE_EXIT = 3 +#: Truncation applied to captured dispatcher output in a result. +_OUTPUT_LIMIT = 4096 +#: SSH connect timeout; the overall dispatch timeout is ``WorkerConfig.ssh_timeout``. +_CONNECT_TIMEOUT = 10 + +#: Broker statuses that end a request without any dispatch. +TERMINAL_BROKER_STATUSES: frozenset[str] = frozenset({"denied", "cancelled", "expired"}) + + +class SshRunner(Protocol): + """Runs one SSH invocation; injectable so tests never open a socket.""" + + def __call__( + self, + argv: list[str], + *, + input: str, + capture_output: bool, + text: bool, + timeout: float, + ) -> subprocess.CompletedProcess[str]: ... + + +@dataclass(frozen=True) +class WorkerConfig: + """Explicit worker configuration; no deployment defaults exist.""" + + #: Persistent state root holding ``input/``, ``results/`` and ``keys/``. + state_dir: Path + #: SSH targets tried in order, e.g. ``("ops@node-a", "ops@node-b")``. + targets: tuple[str, ...] + #: SSH port on the targets. + ssh_port: int + #: Pinned known-hosts file for the targets. + known_hosts_path: str + #: Seconds between polls. + poll_interval: float = 5.0 + #: Overall timeout of one SSH dispatch. + ssh_timeout: float = 240.0 + #: Slack over the broker deadline before a request expires locally. + expiry_grace: float = 60.0 + + def __post_init__(self) -> None: + if isinstance(self.state_dir, str): + object.__setattr__(self, "state_dir", Path(self.state_dir)) + if isinstance(self.targets, list): + object.__setattr__(self, "targets", tuple(self.targets)) + if not self.targets or not all(isinstance(t, str) and t.strip() for t in self.targets): + raise ValueError("targets must be a non-empty sequence of non-empty strings") + if type(self.ssh_port) is not int or not 1 <= self.ssh_port <= 65535: + raise ValueError("ssh_port must be an integer in 1-65535") + if not self.known_hosts_path: + raise ValueError("known_hosts_path is required") + if self.poll_interval <= 0 or self.ssh_timeout <= 0 or self.expiry_grace < 0: + raise ValueError("poll_interval and ssh_timeout must be positive, expiry_grace non-negative") + + +def _dispatcher_outcome(stdout: str) -> dict[str, Any]: + """Lift the typed dispatcher's own JSON result, when it produced one. + + The dispatcher prints a single JSON object. Anything else -- an sshd error, + a truncated stream -- is not fatal here, because the exit code already + classifies the run; the lifted fields are additional detail, not the verdict. + """ + try: + parsed = json.loads(stdout) + except (json.JSONDecodeError, ValueError): + return {} + if not isinstance(parsed, dict): + return {} + return {f"dispatcher_{key}": parsed[key] for key in ("lease_id", "replayed", "error") if key in parsed} + + +def dispatch( + client: BrokerClient, + config: WorkerConfig, + request_id: str, + input_data: dict[str, Any], + *, + ssh_runner: SshRunner, +) -> dict[str, Any]: + """Fetch lease and certificate, then dispatch to each configured target. + + ``BrokerUnavailable`` from the lease or certificate call propagates: it is + raised before any SSH, so nothing ran and the caller keeps the request for + the next poll. + """ + keys_dir = config.state_dir / "keys" + public_key = input_data["public_key"] + key_id = input_data["key_id"] + key_path = keys_dir / f"{key_id}.key" + cert_path = keys_dir / f"{key_id}-cert.pub" + + lease_resp = client.request("GET", f"/v1/broker/requests/{request_id}/lease") + if not isinstance(lease_resp, dict) or not isinstance(lease_resp.get("signature"), str): + raise ValueError("broker lease response is malformed") + lease = {key: value for key, value in lease_resp.items() if key != "signature"} + signature = lease_resp["signature"] + + # The certificate request carries only the public key: the serial is the + # broker's choice, never the caller's. + cert_resp = client.request("POST", f"/v1/broker/requests/{request_id}/certificate", {"ssh_public_key": public_key}) + certificate = cert_resp.get("certificate") if isinstance(cert_resp, dict) else None + if not isinstance(certificate, str) or not certificate: + raise ValueError("broker did not issue a certificate") + cert_path.write_text(certificate, encoding="ascii") + cert_path.chmod(0o600) + + payload = json.dumps( + { + "action": input_data["action"], + "ct_id": input_data["ct_id"], + "lease": lease, + "signature": signature, + } + ) + for target in config.targets: + _LOG.info( + "request %s: dispatching %s on CT %s via %s", + request_id, + input_data["action"], + input_data["ct_id"], + target, + ) + argv = [ + "ssh", + "-F", + "/dev/null", + "-i", + str(key_path), + "-o", + f"CertificateFile={cert_path}", + "-o", + "BatchMode=yes", + "-o", + "IdentitiesOnly=yes", + "-o", + "RequestTTY=no", + "-o", + f"ConnectTimeout={_CONNECT_TIMEOUT}", + "-o", + f"UserKnownHostsFile={config.known_hosts_path}", + "-o", + "StrictHostKeyChecking=yes", + "-p", + str(config.ssh_port), + target, + ] + try: + result = ssh_runner(argv, input=payload, capture_output=True, text=True, timeout=config.ssh_timeout) + except subprocess.TimeoutExpired: + # The dispatcher bounds its own work and records the lease as + # UNKNOWN on its own timeout; the certificate is single-use and + # the claim store refuses a second dispatch of the same lease, so a + # transport timeout is reported as unknown, never retried. + _LOG.warning("request %s: ssh to %s timed out after %ss", request_id, target, config.ssh_timeout) + return { + "status": "unknown", + "node": target, + "error": f"ssh dispatch exceeded {config.ssh_timeout}s; remote outcome unknown", + } + outcome = _dispatcher_outcome(result.stdout) + _LOG.info("request %s: %s exited %s", request_id, target, result.returncode) + # Exit 3 alone is not enough: the dispatcher passes pct's own exit code + # through, and only its JSON marks the node as the wrong one. + if result.returncode == NOT_ON_THIS_NODE_EXIT and outcome.get("dispatcher_error") == "not_on_this_node": + continue + return { + # Classify from the dispatcher's exit code, not from "command attempted". + "status": "succeeded" if result.returncode == 0 else "failed", + "node": target, + "exit_code": result.returncode, + "stdout": result.stdout[:_OUTPUT_LIMIT], + "stderr": result.stderr[:_OUTPUT_LIMIT], + **outcome, + } + return {"status": "error", "error": "container not found on any configured node"} + + +def _finish( + config: WorkerConfig, + request_id: str, + input_file: Path, + key_id: object, + result: dict[str, Any], + log: logging.Logger, +) -> None: + results_dir = config.state_dir / "results" + results_dir.mkdir(mode=0o700, parents=True, exist_ok=True) + (results_dir / f"{request_id}.json").write_text(json.dumps(result), encoding="utf-8") + input_file.unlink(missing_ok=True) + if isinstance(key_id, str): + keys_dir = config.state_dir / "keys" + (keys_dir / f"{key_id}.key").unlink(missing_ok=True) + (keys_dir / f"{key_id}-cert.pub").unlink(missing_ok=True) + log.info("request %s: result %s", request_id, result["status"]) + + +def process_pending_once( + client: BrokerClient, + config: WorkerConfig, + *, + now: int | None = None, + ssh_runner: SshRunner | None = None, + logger: logging.Logger | None = None, +) -> None: + """Process every pending input file once.""" + log = logger or _LOG + runner: SshRunner = ssh_runner or subprocess.run + now = int(time.time()) if now is None else now + input_dir = config.state_dir / "input" + results_dir = config.state_dir / "results" + if not input_dir.is_dir(): + return + for input_file in sorted(input_dir.iterdir()): + if input_file.suffix != ".json": + continue + request_id = input_file.stem + if (results_dir / f"{request_id}.json").exists(): + continue + try: + input_data = json.loads(input_file.read_text(encoding="utf-8")) + except (json.JSONDecodeError, UnicodeDecodeError, OSError) as exc: + log.warning("request %s: unreadable input file: %s", request_id, exc) + continue + if not isinstance(input_data, dict): + log.warning("request %s: input file is not an object", request_id) + continue + key_id = input_data.get("key_id") + expires_at = input_data.get("expires_at") + if type(expires_at) is int and now > expires_at + config.expiry_grace: + _finish(config, request_id, input_file, key_id, {"status": "expired"}, log) + continue + try: + status_resp = client.request("GET", f"/v1/broker/requests/{request_id}") + except (RuntimeError, OSError) as exc: + # Broker availability and rate limiting are transient; the local + # expiry above bounds the retries. + log.warning("request %s: broker status unavailable: %s", request_id, exc) + continue + status = status_resp.get("status") if isinstance(status_resp, dict) else None + if not isinstance(status, str): + log.warning("request %s: broker status missing", request_id) + continue + if status == "approved": + try: + result = dispatch(client, config, request_id, input_data, ssh_runner=runner) + except BrokerUnavailable as exc: + # Raised only by the lease and certificate calls, before any + # SSH: nothing ran, so the input and key stay for the next poll. + log.warning("request %s: broker unavailable before dispatch, will retry: %s", request_id, exc) + continue + except Exception as exc: + # Fail closed: a consumed certificate (HTTP 409 "already + # fetched") ends the request as an error; no replacement + # authority is minted and no operation is repeated. + log.warning("request %s: dispatch failed: %s", request_id, exc) + result = {"status": "error", "error": str(exc)} + _finish(config, request_id, input_file, key_id, result, log) + elif status in TERMINAL_BROKER_STATUSES: + _finish(config, request_id, input_file, key_id, {"status": status}, log) + + +def run( + client: BrokerClient, + config: WorkerConfig, + *, + ssh_runner: SshRunner | None = None, + logger: logging.Logger | None = None, +) -> None: + """Poll forever. Importing the module never starts this.""" + log = logger or _LOG + for name in ("input", "results", "keys"): + (config.state_dir / name).mkdir(mode=0o700, parents=True, exist_ok=True) + log.info("polling every %ss; targets %s", config.poll_interval, ", ".join(config.targets)) + while True: + process_pending_once(client, config, ssh_runner=ssh_runner, logger=log) + time.sleep(config.poll_interval) diff --git a/tests/test_pve_bridge.py b/tests/test_pve_bridge.py new file mode 100644 index 0000000..b052783 --- /dev/null +++ b/tests/test_pve_bridge.py @@ -0,0 +1,217 @@ +"""Tests for the agent-facing approval request bridge.""" + +from __future__ import annotations + +import json +import stat +from pathlib import Path + +import pytest + +from privilege_broker.pve.bridge import ApprovalBridge, BridgeConfig +from privilege_broker.pve.client import BrokerHTTPError, BrokerUnavailable + +VALID_ID = "123e4567-e89b-42d3-a456-426614174000" + + +class FakeBroker: + """Records requests and answers with a scripted create response or error.""" + + def __init__(self, response: object = None, error: Exception | None = None) -> None: + self._response = ( + {"request_id": VALID_ID, "status": "pending", "expires_at": 1_800_000_000} + if response is None + else response + ) + self._error = error + self.calls: list[tuple[str, str, object]] = [] + + def request(self, method: str, path: str, payload: dict | None = None) -> dict: + self.calls.append((method, path, payload)) + if self._error is not None: + raise self._error + if not isinstance(self._response, dict): + return self._response + return dict(self._response) + + +@pytest.fixture +def config(tmp_path: Path) -> BridgeConfig: + return BridgeConfig( + state_dir=tmp_path, + public_url="https://broker.example.test", + requester="example-agent", + ) + + +@pytest.fixture +def bridge(config: BridgeConfig) -> ApprovalBridge: + return ApprovalBridge(config) + + +def _key_files(config: BridgeConfig) -> list[Path]: + keys = config.state_dir / "keys" + return list(keys.iterdir()) if keys.is_dir() else [] + + +def _input_files(config: BridgeConfig) -> list[Path]: + inputs = config.state_dir / "input" + return list(inputs.iterdir()) if inputs.is_dir() else [] + + +# --------------------------------------------------------------------------- +# Configuration +# --------------------------------------------------------------------------- + + +def test_config_requires_https_public_url(tmp_path: Path) -> None: + with pytest.raises(ValueError, match="https://"): + BridgeConfig(state_dir=tmp_path, public_url="http://broker.example.test", requester="agent") + + +def test_config_requires_a_requester(tmp_path: Path) -> None: + with pytest.raises(ValueError, match="requester"): + BridgeConfig(state_dir=tmp_path, public_url="https://broker.example.test", requester=" ") + + +# --------------------------------------------------------------------------- +# Input validation — refusals happen before any broker call or key write +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("action", ["shell", "ct.exec", "", "CT.START"]) +def test_untyped_action_is_refused(bridge: ApprovalBridge, action: str) -> None: + broker = FakeBroker() + with pytest.raises(ValueError, match="action"): + bridge.request_mutation(action, 9001, "reason", broker) + assert broker.calls == [] + assert _key_files(bridge._config) == [] + + +@pytest.mark.parametrize("ct_id", [0, -1, 1.5, True, "9001", None]) +def test_invalid_ct_id_is_refused(bridge: ApprovalBridge, ct_id: object) -> None: + broker = FakeBroker() + with pytest.raises(ValueError, match="ct_id"): + bridge.request_mutation("ct.reboot", ct_id, "reason", broker) # type: ignore[arg-type] + assert broker.calls == [] + assert _key_files(bridge._config) == [] + + +@pytest.mark.parametrize("reason", ["", " ", "x" * 501, 42]) +def test_invalid_reason_is_refused(bridge: ApprovalBridge, reason: object) -> None: + broker = FakeBroker() + with pytest.raises(ValueError, match="reason"): + bridge.request_mutation("ct.reboot", 9001, reason, broker) # type: ignore[arg-type] + assert broker.calls == [] + assert _key_files(bridge._config) == [] + + +# --------------------------------------------------------------------------- +# Successful request +# --------------------------------------------------------------------------- + + +def test_success_writes_input_and_ephemeral_key(bridge: ApprovalBridge, config: BridgeConfig) -> None: + broker = FakeBroker() + result = bridge.request_mutation("ct.reboot", 9001, "Apply kernel update", broker) + + assert result == { + "request_id": VALID_ID, + "status": "pending", + "expires_at": 1_800_000_000, + "approval_url": f"https://broker.example.test/v1/broker/requests/{VALID_ID}/approval", + } + ((method, path, payload),) = broker.calls + assert (method, path) == ("POST", "/v1/broker/requests") + assert payload["action"] == "ct.reboot" + assert payload["ct_id"] == 9001 + assert payload["reason"] == "Apply kernel update" + assert payload["requester"] == "example-agent" + public_key = payload["ssh_public_key"] + assert public_key.startswith("ssh-ed25519 ") + + (input_file,) = _input_files(config) + input_data = json.loads(input_file.read_text(encoding="utf-8")) + assert input_data["request_id"] == VALID_ID + assert input_data["action"] == "ct.reboot" + assert input_data["public_key"] == public_key + # The input file carries the public key only, never the private key. + assert "PRIVATE KEY" not in input_file.read_text(encoding="utf-8") + + (key_file,) = _key_files(config) + assert key_file.name == f"{input_data['key_id']}.key" + assert stat.S_IMODE(key_file.stat().st_mode) == 0o600 + assert key_file.read_bytes().startswith(b"-----BEGIN OPENSSH") + + +# --------------------------------------------------------------------------- +# Broker and response failures clean up the ephemeral key +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "error", + [ + BrokerHTTPError(409, "broker HTTP 409: not permitted by the operate policy"), + BrokerUnavailable("broker unavailable: URLError"), + RuntimeError("unexpected"), + ], +) +def test_broker_failure_removes_the_private_key( + bridge: ApprovalBridge, config: BridgeConfig, error: Exception +) -> None: + with pytest.raises(type(error)): + bridge.request_mutation("ct.reboot", 9001, "reason", FakeBroker(error=error)) + assert _key_files(config) == [] + assert _input_files(config) == [] + + +@pytest.mark.parametrize( + "response", + [ + ["not", "an", "object"], + {}, + {"request_id": "not-a-uuid", "status": "pending", "expires_at": 1}, + {"request_id": VALID_ID + "\n", "status": "pending", "expires_at": 1}, + {"request_id": "../../etc/passwd", "status": "pending", "expires_at": 1}, + {"request_id": VALID_ID, "status": "approved", "expires_at": 1}, + {"request_id": VALID_ID, "status": "pending"}, + {"request_id": VALID_ID, "status": "pending", "expires_at": "1"}, + {"request_id": VALID_ID, "status": "pending", "expires_at": True}, + ], +) +def test_invalid_create_response_rejects_and_cleans_up( + bridge: ApprovalBridge, config: BridgeConfig, response: object +) -> None: + with pytest.raises(ValueError): + bridge.request_mutation("ct.reboot", 9001, "reason", FakeBroker(response=response)) + assert _key_files(config) == [] + assert _input_files(config) == [] + + +# --------------------------------------------------------------------------- +# Status reads +# --------------------------------------------------------------------------- + + +def test_read_status_returns_the_worker_result(bridge: ApprovalBridge, config: BridgeConfig) -> None: + results = config.state_dir / "results" + results.mkdir(parents=True) + (results / f"{VALID_ID}.json").write_text(json.dumps({"status": "succeeded", "node": "ops@node-a"})) + assert bridge.read_status(VALID_ID) == {"status": "succeeded", "node": "ops@node-a"} + + +def test_read_status_pending_for_active_input(bridge: ApprovalBridge, config: BridgeConfig) -> None: + broker = FakeBroker() + bridge.request_mutation("ct.reboot", 9001, "reason", broker) + assert bridge.read_status(VALID_ID) == {"status": "pending"} + + +def test_read_status_unknown_for_missing_request(bridge: ApprovalBridge) -> None: + assert bridge.read_status(VALID_ID) == {"status": "unknown", "error": "request not found or expired"} + + +@pytest.mark.parametrize("request_id", ["not-a-uuid", "../../etc/passwd", "", VALID_ID + "x"]) +def test_read_status_rejects_non_canonical_ids(bridge: ApprovalBridge, request_id: str) -> None: + with pytest.raises(ValueError, match="UUID"): + bridge.read_status(request_id) diff --git a/tests/test_pve_client.py b/tests/test_pve_client.py new file mode 100644 index 0000000..7de6e4d --- /dev/null +++ b/tests/test_pve_client.py @@ -0,0 +1,175 @@ +"""Tests for the explicit-config mTLS broker client.""" + +from __future__ import annotations + +import io +import json +import ssl +import urllib.error + +import pytest + +from privilege_broker.pve.client import BrokerClient, BrokerClientConfig, BrokerHTTPError, BrokerUnavailable + +CONFIG = BrokerClientConfig( + broker_url="https://broker.example.test", + cert_path="/etc/example/agent-cert.pem", + key_path="/etc/example/agent-key.pem", + ca_path="/etc/example/broker-ca.pem", + timeout=15.0, +) + + +class _FakeResponse: + def __init__(self, body: bytes) -> None: + self._body = body + + def __enter__(self) -> _FakeResponse: + return self + + def __exit__(self, *args: object) -> None: + return None + + def read(self, limit: int = -1) -> bytes: + return self._body[:limit] if limit >= 0 else self._body + + +class _FakeOpener: + """Stands in for urllib's opener; records requests and answers scripted.""" + + def __init__(self, outcomes: list[object]) -> None: + self._outcomes = list(outcomes) + self.requests: list[urllib.request.Request] = [] + + def open(self, request: urllib.request.Request, timeout: float = 0) -> object: + self.requests.append(request) + if not self._outcomes: + raise AssertionError("no scripted outcome left") + outcome = self._outcomes.pop(0) + if isinstance(outcome, Exception): + raise outcome + return outcome + + +def _client(opener: _FakeOpener) -> BrokerClient: + return BrokerClient(CONFIG, ssl_context=ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT), opener=opener) + + +# --------------------------------------------------------------------------- +# Configuration +# --------------------------------------------------------------------------- + + +def test_config_requires_https_broker_url() -> None: + with pytest.raises(ValueError, match="https://"): + BrokerClientConfig(broker_url="http://broker.example.test", cert_path="c", key_path="k") + + +def test_config_requires_client_certificate_and_key() -> None: + with pytest.raises(ValueError, match="certificate and key"): + BrokerClientConfig(broker_url="https://broker.example.test", cert_path="", key_path="k") + with pytest.raises(ValueError, match="certificate and key"): + BrokerClientConfig(broker_url="https://broker.example.test", cert_path="c", key_path="") + + +def test_config_requires_positive_timeout() -> None: + with pytest.raises(ValueError, match="timeout"): + BrokerClientConfig(broker_url="https://broker.example.test", cert_path="c", key_path="k", timeout=0) + + +def test_default_context_loads_the_configured_pair_and_verifies_hostname( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The client builds its mTLS context from the explicit config paths only.""" + + class _RecordingContext: + check_hostname = False + + def __init__(self) -> None: + self.loaded: list[tuple[str, str]] = [] + + def load_cert_chain(self, certfile: str, keyfile: str) -> None: + self.loaded.append((certfile, keyfile)) + + context = _RecordingContext() + monkeypatch.setattr(ssl, "create_default_context", lambda **kwargs: context) + client = BrokerClient(CONFIG) + assert context.loaded == [(CONFIG.cert_path, CONFIG.key_path)] + assert context.check_hostname is True + assert client._context is context + + +# --------------------------------------------------------------------------- +# Request handling and failure classification +# --------------------------------------------------------------------------- + + +def test_request_sends_method_path_and_json_body() -> None: + opener = _FakeOpener([_FakeResponse(b'{"status": "pending"}')]) + client = _client(opener) + assert client.request("POST", "/v1/broker/requests", {"action": "ct.reboot"}) == {"status": "pending"} + (request,) = opener.requests + assert request.full_url == "https://broker.example.test/v1/broker/requests" + assert request.get_method() == "POST" + assert request.headers["Content-type"] == "application/json" + assert json.loads(request.data.decode()) == {"action": "ct.reboot"} + + +def test_request_without_payload_sends_no_body() -> None: + opener = _FakeOpener([_FakeResponse(b'{"status": "ok"}')]) + assert _client(opener).request("GET", "/health") == {"status": "ok"} + assert opener.requests[0].data is None + + +def test_request_rejects_a_path_without_leading_slash() -> None: + with pytest.raises(ValueError, match="path"): + _client(_FakeOpener([])).request("GET", "v1/broker/requests") + + +def test_non_object_json_response_becomes_empty_dict() -> None: + opener = _FakeOpener([_FakeResponse(b'["not", "an", "object"]')]) + assert _client(opener).request("GET", "/x") == {} + + +def test_invalid_json_body_is_a_refusal() -> None: + opener = _FakeOpener([_FakeResponse(b"not json at all")]) + with pytest.raises(RuntimeError, match="non-JSON"): + _client(opener).request("GET", "/x") + + +@pytest.mark.parametrize("status", [429, 502, 503, 504]) +def test_retryable_statuses_are_broker_unavailable(status: int) -> None: + error = urllib.error.HTTPError( + "https://broker.example.test/x", status, "err", None, io.BytesIO(b'{"detail": "busy"}') + ) + with pytest.raises(BrokerUnavailable, match=f"broker HTTP {status}"): + _client(_FakeOpener([error])).request("GET", "/x") + + +def test_final_http_status_is_broker_http_error_with_status() -> None: + error = urllib.error.HTTPError( + "https://broker.example.test/x", 409, "err", None, io.BytesIO(b'{"detail": "already fetched"}') + ) + with pytest.raises(BrokerHTTPError, match="already fetched") as excinfo: + _client(_FakeOpener([error])).request("GET", "/x") + assert excinfo.value.status == 409 + + +def test_http_error_detail_is_truncated() -> None: + error = urllib.error.HTTPError("https://broker.example.test/x", 400, "err", None, io.BytesIO(b"x" * 5000)) + with pytest.raises(BrokerHTTPError) as excinfo: + _client(_FakeOpener([error])).request("GET", "/x") + assert len(str(excinfo.value)) <= len("broker HTTP 400: ") + 1024 + 3 + + +@pytest.mark.parametrize( + "failure", + [ + urllib.error.URLError("connection refused"), + TimeoutError("timed out"), + ConnectionResetError("reset"), + ], +) +def test_transport_failures_are_broker_unavailable(failure: Exception) -> None: + with pytest.raises(BrokerUnavailable, match="broker unavailable"): + _client(_FakeOpener([failure])).request("GET", "/x") diff --git a/tests/test_pve_worker.py b/tests/test_pve_worker.py new file mode 100644 index 0000000..583ca3d --- /dev/null +++ b/tests/test_pve_worker.py @@ -0,0 +1,427 @@ +"""Tests for the poll-and-dispatch worker. + +All broker and SSH effects are fakes; no test opens a socket, contacts PVE or +touches a real host. State lives in a temporary directory. +""" + +from __future__ import annotations + +import json +import logging +import subprocess +from pathlib import Path +from subprocess import CompletedProcess + +import pytest + +from privilege_broker.pve.client import BrokerHTTPError, BrokerUnavailable +from privilege_broker.pve.worker import WorkerConfig, process_pending_once + +NOW = 1_800_000_000 +PUB_KEY = "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIFakePublicKeyMaterialForTestsOnly" +FAKE_CERT = "FAKE-CERT-MATERIAL" +FAKE_KEY_CONTENT = "fake-key-content" +_NOT_HERE = json.dumps({"error": "not_on_this_node", "exit_code": 3}) + +LEASE = { + "lease_id": "lease-1", + "signature": "AAAA", + "audience": "pve-privileged-executor", + "policy_version": "pve-operate-v1", + "issued_at": 0, + "expires_at": 9_999_999_999, + "max_dispatches": 1, + "approval_id": "approval-001", + "approval_binding": {"canonicalization": "pve-approval-v1", "request_digest": "aa" * 32}, + "execution_binding": {"canonicalization": "pve-execution-v1", "arguments_digest": "bb" * 32}, +} + + +class FakeBroker: + def __init__( + self, + status: str, + *, + lease_error: Exception | None = None, + cert_error: Exception | None = None, + status_response: dict | None = None, + ) -> None: + self._status = status + self._lease_error = lease_error + self._cert_error = cert_error + self._status_response = status_response + self.calls = 0 + self.fetched_lease = 0 + self.fetched_cert = 0 + self.cert_payloads: list[dict | None] = [] + + def request(self, method: str, path: str, payload: dict | None = None) -> dict: + self.calls += 1 + if method == "GET" and path.endswith("/lease"): + self.fetched_lease += 1 + if self._lease_error is not None: + raise self._lease_error + return dict(LEASE) + if method == "POST" and path.endswith("/certificate"): + self.fetched_cert += 1 + self.cert_payloads.append(payload) + if self._cert_error is not None: + raise self._cert_error + return {"status": "issued", "certificate": FAKE_CERT} + if method == "GET": + return self._status_response if self._status_response is not None else {"status": self._status} + raise AssertionError(f"unexpected {method} {path}") + + +class UnavailableBroker: + """Every call is a transient failure.""" + + def __init__(self) -> None: + self.calls = 0 + + def request(self, method: str, path: str, payload: dict | None = None) -> dict: + self.calls += 1 + raise BrokerUnavailable("broker unavailable: URLError") + + +class ScriptedSsh: + """Answers successive SSH calls from scripted (returncode, stdout) pairs.""" + + def __init__(self, outcomes: list[tuple[int, str] | subprocess.TimeoutExpired]) -> None: + self._outcomes = list(outcomes) + self.calls: list[tuple[list[str], dict]] = [] + + def __call__(self, argv: list[str], **kwargs: object) -> CompletedProcess[str]: + self.calls.append((argv, kwargs)) + if not self._outcomes: + raise AssertionError("no scripted SSH outcome left") + outcome = self._outcomes.pop(0) + if isinstance(outcome, subprocess.TimeoutExpired): + raise outcome + code, stdout = outcome + return CompletedProcess(argv, code, stdout, "") + + +@pytest.fixture +def config(tmp_path: Path) -> WorkerConfig: + return WorkerConfig( + state_dir=tmp_path, + targets=("ops@node-a", "ops@node-b"), + ssh_port=8922, + known_hosts_path=str(tmp_path / "known_hosts"), + poll_interval=5, + ssh_timeout=240, + expiry_grace=60, + ) + + +def _write_input(config: WorkerConfig, request_id: str = "req-1", *, expires_at: int = NOW + 300) -> Path: + input_dir = config.state_dir / "input" + input_dir.mkdir(parents=True, exist_ok=True) + input_file = input_dir / f"{request_id}.json" + input_file.write_text( + json.dumps( + { + "action": "ct.start", + "ct_id": 9001, + "reason": "synthetic", + "public_key": PUB_KEY, + "key_id": "key-1", + "expires_at": expires_at, + } + ), + encoding="utf-8", + ) + keys = config.state_dir / "keys" + keys.mkdir(exist_ok=True) + key_path = keys / "key-1.key" + key_path.write_text(FAKE_KEY_CONTENT) + key_path.chmod(0o600) + return input_file + + +def _result(config: WorkerConfig, request_id: str = "req-1") -> dict: + return json.loads((config.state_dir / "results" / f"{request_id}.json").read_text(encoding="utf-8")) + + +def _key_and_cert_exist(config: WorkerConfig) -> bool: + keys = config.state_dir / "keys" + return (keys / "key-1.key").exists() or (keys / "key-1-cert.pub").exists() + + +# --------------------------------------------------------------------------- +# Dispatch classification +# --------------------------------------------------------------------------- + + +def test_approved_request_dispatches_and_succeeds(config: WorkerConfig) -> None: + input_file = _write_input(config) + ssh = ScriptedSsh([(0, json.dumps({"exit_code": 0, "lease_id": "lease-1", "replayed": False}))]) + broker = FakeBroker("approved") + + process_pending_once(broker, config, now=NOW, ssh_runner=ssh) + + result = _result(config) + assert result["status"] == "succeeded" + assert result["node"] == "ops@node-a" + assert result["exit_code"] == 0 + assert result["dispatcher_lease_id"] == "lease-1" + assert not input_file.exists() + assert not _key_and_cert_exist(config) + + # The dispatcher wire contract on stdin is unchanged: action, ct_id, the + # lease claims without the signature, and the signature separately. + ((argv, kwargs),) = ssh.calls + payload = json.loads(kwargs["input"]) + assert payload["action"] == "ct.start" + assert payload["ct_id"] == 9001 + assert payload["signature"] == "AAAA" + assert "signature" not in payload["lease"] + assert payload["lease"]["lease_id"] == "lease-1" + assert kwargs["timeout"] == 240 + + # The certificate request carries only the public key; the broker picks the serial. + assert broker.cert_payloads == [{"ssh_public_key": PUB_KEY}] + + # Fixed, non-interactive SSH options with the configured trust material. + assert argv[0] == "ssh" + assert "-F" in argv and "/dev/null" in argv + assert f"CertificateFile={config.state_dir / 'keys' / 'key-1-cert.pub'}" in argv + fixed_options = ( + "BatchMode=yes", + "IdentitiesOnly=yes", + "RequestTTY=no", + "ConnectTimeout=10", + "StrictHostKeyChecking=yes", + ) + for option in fixed_options: + assert option in argv + assert f"UserKnownHostsFile={config.known_hosts_path}" in argv + assert argv[argv.index("-p") + 1] == "8922" + assert argv[-1] == "ops@node-a" + + +def test_nonzero_dispatcher_exit_is_failed_not_dispatched(config: WorkerConfig) -> None: + _write_input(config) + ssh = ScriptedSsh([(255, "CT 9001 already running\n")]) + process_pending_once(FakeBroker("approved"), config, now=NOW, ssh_runner=ssh) + result = _result(config) + assert result["status"] == "failed" + assert result["exit_code"] == 255 + # Non-JSON stdout lifts no dispatcher fields; the exit code is the verdict. + assert not [key for key in result if key.startswith("dispatcher_")] + + +def test_ssh_timeout_is_unknown_and_never_retried(config: WorkerConfig) -> None: + _write_input(config) + ssh = ScriptedSsh([subprocess.TimeoutExpired(cmd="ssh", timeout=240)]) + process_pending_once(FakeBroker("approved"), config, now=NOW, ssh_runner=ssh) + result = _result(config) + assert result["status"] == "unknown" + assert result["node"] == "ops@node-a" + assert "240" in result["error"] + # A lost reply is not a reason to dispatch again: exactly one attempt. + assert len(ssh.calls) == 1 + + +def test_wrong_node_moves_to_the_next_target_with_the_same_certificate(config: WorkerConfig) -> None: + _write_input(config) + ssh = ScriptedSsh([(3, _NOT_HERE), (0, json.dumps({"exit_code": 0, "lease_id": "lease-1"}))]) + process_pending_once(FakeBroker("approved"), config, now=NOW, ssh_runner=ssh) + + result = _result(config) + assert result["status"] == "succeeded" + assert result["node"] == "ops@node-b" + assert [call[0][-1] for call in ssh.calls] == ["ops@node-a", "ops@node-b"] + # Same key, certificate, payload and options; only the target differs. + assert ssh.calls[0][0][:-1] == ssh.calls[1][0][:-1] + assert ssh.calls[0][1]["input"] == ssh.calls[1][1]["input"] + + +def test_exit_3_without_the_dispatcher_marker_is_final(config: WorkerConfig) -> None: + """pct exit codes pass through the dispatcher; only its JSON says wrong node.""" + _write_input(config) + ssh = ScriptedSsh([(3, json.dumps({"exit_code": 3, "stdout": "", "stderr": "x"}))]) + process_pending_once(FakeBroker("approved"), config, now=NOW, ssh_runner=ssh) + result = _result(config) + assert result["status"] == "failed" + assert result["node"] == "ops@node-a" + assert len(ssh.calls) == 1 + + +def test_no_hosting_node_is_an_error(config: WorkerConfig) -> None: + _write_input(config) + ssh = ScriptedSsh([(3, _NOT_HERE), (3, _NOT_HERE)]) + process_pending_once(FakeBroker("approved"), config, now=NOW, ssh_runner=ssh) + assert _result(config) == {"status": "error", "error": "container not found on any configured node"} + assert not _key_and_cert_exist(config) + + +# --------------------------------------------------------------------------- +# Terminal broker decisions and transient failures +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("status", ["denied", "cancelled", "expired"]) +def test_terminal_broker_status_never_dispatches(config: WorkerConfig, status: str) -> None: + input_file = _write_input(config) + ssh = ScriptedSsh([]) + process_pending_once(FakeBroker(status), config, now=NOW, ssh_runner=ssh) + assert _result(config) == {"status": status} + assert ssh.calls == [] + assert not input_file.exists() + + +def test_unavailable_status_poll_preserves_the_request(config: WorkerConfig) -> None: + input_file = _write_input(config) + ssh = ScriptedSsh([]) + broker = UnavailableBroker() + process_pending_once(broker, config, now=NOW, ssh_runner=ssh) + assert not (config.state_dir / "results" / "req-1.json").exists() + assert input_file.exists() + assert ssh.calls == [] + + +def test_broker_unavailable_before_dispatch_is_retried_on_the_next_poll(config: WorkerConfig) -> None: + """A transient lease/certificate failure keeps the request; nothing ran yet.""" + input_file = _write_input(config) + broker = FakeBroker("approved", lease_error=BrokerUnavailable("broker unavailable: URLError")) + process_pending_once(broker, config, now=NOW, ssh_runner=ScriptedSsh([])) + assert not (config.state_dir / "results" / "req-1.json").exists() + assert input_file.exists() + + # Second poll: the broker answers, dispatch proceeds. + broker2 = FakeBroker("approved") + ssh = ScriptedSsh([(0, "")]) + process_pending_once(broker2, config, now=NOW, ssh_runner=ssh) + assert _result(config)["status"] == "succeeded" + assert broker.fetched_lease == 1 and broker2.fetched_lease == 1 + + +def test_a_repeated_certificate_fetch_is_a_terminal_error(config: WorkerConfig) -> None: + """HTTP 409 after a lost certificate reply is terminal: no re-mint, no replay.""" + _write_input(config) + ssh = ScriptedSsh([]) + broker = FakeBroker( + "approved", + cert_error=BrokerHTTPError(409, "broker HTTP 409: certificate has already been fetched"), + ) + process_pending_once(broker, config, now=NOW, ssh_runner=ssh) + result = _result(config) + assert result["status"] == "error" + assert "already been fetched" in result["error"] + assert broker.fetched_cert == 1 + assert ssh.calls == [] + + +# --------------------------------------------------------------------------- +# Local expiry bounds polling without asking the broker +# --------------------------------------------------------------------------- + + +def test_past_deadline_plus_grace_expires_locally_without_a_broker_call(config: WorkerConfig) -> None: + _write_input(config, expires_at=NOW - 61) + broker = FakeBroker("approved") + process_pending_once(broker, config, now=NOW, ssh_runner=ScriptedSsh([])) + assert _result(config) == {"status": "expired"} + assert broker.calls == 0 + + +def test_within_grace_period_still_asks_the_broker(config: WorkerConfig) -> None: + _write_input(config, expires_at=NOW - 50) + broker = FakeBroker("pending") + process_pending_once(broker, config, now=NOW, ssh_runner=ScriptedSsh([])) + assert broker.calls == 1 + assert not (config.state_dir / "results" / "req-1.json").exists() + + +# --------------------------------------------------------------------------- +# Malformed or completed input +# --------------------------------------------------------------------------- + + +def test_malformed_input_file_is_preserved_and_skipped(config: WorkerConfig) -> None: + input_dir = config.state_dir / "input" + input_dir.mkdir(parents=True) + bad = input_dir / "bad.json" + bad.write_text("{not json", encoding="utf-8") + broker = FakeBroker("approved") + process_pending_once(broker, config, now=NOW, ssh_runner=ScriptedSsh([])) + assert bad.exists() + assert broker.calls == 0 + assert not (config.state_dir / "results" / "bad.json").exists() + + +def test_existing_result_skips_the_input(config: WorkerConfig) -> None: + _write_input(config) + results = config.state_dir / "results" + results.mkdir(parents=True) + (results / "req-1.json").write_text(json.dumps({"status": "succeeded"})) + broker = FakeBroker("approved") + process_pending_once(broker, config, now=NOW, ssh_runner=ScriptedSsh([])) + assert broker.calls == 0 + + +def test_missing_input_dir_is_a_noop(config: WorkerConfig) -> None: + process_pending_once(FakeBroker("approved"), config, now=NOW, ssh_runner=ScriptedSsh([])) + + +def test_status_response_without_status_field_is_preserved(config: WorkerConfig) -> None: + _write_input(config) + broker = FakeBroker("approved", status_response={}) + process_pending_once(broker, config, now=NOW, ssh_runner=ScriptedSsh([])) + assert not (config.state_dir / "results" / "req-1.json").exists() + + +# --------------------------------------------------------------------------- +# Secrets never reach results or logs +# --------------------------------------------------------------------------- + + +def test_result_and_logs_carry_no_key_or_certificate_material( + config: WorkerConfig, caplog: pytest.LogCaptureFixture +) -> None: + _write_input(config) + ssh = ScriptedSsh([(0, json.dumps({"exit_code": 0, "lease_id": "lease-1", "stdout": "ok"}))]) + with caplog.at_level(logging.INFO, logger="privilege_broker.pve.worker"): + process_pending_once(FakeBroker("approved"), config, now=NOW, ssh_runner=ssh) + result_text = json.dumps(_result(config)) + assert FAKE_KEY_CONTENT not in result_text + assert FAKE_CERT not in result_text + assert PUB_KEY not in result_text + assert FAKE_KEY_CONTENT not in caplog.text + assert FAKE_CERT not in caplog.text + + +# --------------------------------------------------------------------------- +# Configuration validation +# --------------------------------------------------------------------------- + + +def test_config_requires_targets_and_trust(tmp_path: Path) -> None: + with pytest.raises(ValueError, match="targets"): + WorkerConfig(state_dir=tmp_path, targets=(), ssh_port=22, known_hosts_path="kh") + with pytest.raises(ValueError, match="ssh_port"): + WorkerConfig(state_dir=tmp_path, targets=("ops@node-a",), ssh_port=0, known_hosts_path="kh") + with pytest.raises(ValueError, match="ssh_port"): + WorkerConfig( + state_dir=tmp_path, + targets=("ops@node-a",), + ssh_port=True, + known_hosts_path="kh", # type: ignore[arg-type] + ) + with pytest.raises(ValueError, match="known_hosts_path"): + WorkerConfig(state_dir=tmp_path, targets=("ops@node-a",), ssh_port=22, known_hosts_path="") + with pytest.raises(ValueError, match="positive"): + WorkerConfig(state_dir=tmp_path, targets=("ops@node-a",), ssh_port=22, known_hosts_path="kh", ssh_timeout=0) + + +def test_config_accepts_a_list_of_targets_and_string_state_dir(tmp_path: Path) -> None: + config = WorkerConfig( + state_dir=str(tmp_path), # type: ignore[arg-type] + targets=["ops@node-a"], + ssh_port=22, + known_hosts_path="kh", + ) + assert config.targets == ("ops@node-a",) + assert config.state_dir == tmp_path