From 0f5c1e3408bd27483f8149fce43bce0263f57735 Mon Sep 17 00:00:00 2001 From: umbra Date: Wed, 30 Sep 2026 06:18:26 -0700 Subject: [PATCH] Run the agent as the service account it claims to be The compute proof of concept had never been run with the agent as umbra-poc. Running it that way found three defects, all the same shape: a check that passed because its mechanism did nothing. The demonstration never reached the agent at all. It held the pid it spawned, which under `sudo -n -u` is the sudo wrapper - sudo forks the agent before exec'ing. Signalling that killed the wrapper and left the worker running, renewing its lease and holding the job indefinitely, under a printed note saying "exactly like a power cut". An unprivileged process cannot signal another account's process at all, and kill(2) on a process group succeeds as soon as one member is signalled, so the wrapper's death read as the agent's. Every run leaked an agent. Lease recovery was a diagnosis rather than a mechanism. sweep_expired_leases documented itself as "called by the controller's own timer" and there was no timer; a killed node's job came back only when somebody noticed a stalled job and posted to an administrative route by hand. The proof of concept swept by hand and recorded the result as automatic. app.py now runs the loop for the life of the application, building the compute repository at startup so a restart mid-job still recovers it. The power cut is now the agent's own: os._exit(137) the instant it claims a job, before running it, because a process that does not own the agent cannot signal it. A clean stop goes through a stop_file the agent polls, for the same reason. Liveness is measured by finding the worker's own pid in /proc matched on its config path, sampled repeatedly and accumulated - the wrapper and the worker are not simultaneous, so a single sample finds only the wrapper. The node key travels by `sudo --preserve-env` rather than as VAR=value in argv, because argv is world-readable through /proc//cmdline and the environment is not. The demonstration checks whether the key will survive before starting and falls back to a pipe, which node_agent.py --node-key-stdin accepts. Neither a file nor argv ever holds it. Eight tests, each verified to fail against a mutation of the code it covers: three for the sweeper (it must return an expired lease, must survive a failing round, must not invent a repository it has no database for) and five driving a real agent process against a stub controller (stops on the stop file, exits 137 holding an uncompleted job, takes the key from both routes and neither, reports an empty pipe as an empty pipe). 448 pass with a database, 421 without. The deployment database is untouched: users=2 orgs=3 projects=0 audit=43, no compute tables. --- docs/umbratrans-deployment.md | 43 ++ poc/poc_node_user.sh | 218 +++++++- poc/run_poc.py | 653 +++++++++++++++++++++--- umbratrans_web/app.py | 87 ++++ umbratrans_web/compute.py | 17 + umbratrans_web/node_agent.py | 91 +++- umbratrans_web/routes_compute.py | 14 +- umbratrans_web/tests/test_compute.py | 142 ++++++ umbratrans_web/tests/test_node_agent.py | 303 +++++++++++ 9 files changed, 1466 insertions(+), 102 deletions(-) create mode 100644 umbratrans_web/tests/test_node_agent.py diff --git a/docs/umbratrans-deployment.md b/docs/umbratrans-deployment.md index 57d206f..30bf1c7 100644 --- a/docs/umbratrans-deployment.md +++ b/docs/umbratrans-deployment.md @@ -126,6 +126,49 @@ carries `TENANT_ROW_PREDICATE`, and why the review gate's own SQL in before approving the migration; the failure mode is an assigned member who cannot sign in, and it was found by the integration suite and not by any unit test. +### Lease recovery runs itself now, and that changed what a deploy must include + +A killed node's job is returned by the controller's own sweeper, started with the +application lifespan (`LEASE_SWEEP_SECONDS`, 10 s). It did not used to, and that +was not a design choice: `sweep_expired_leases` carried a docstring saying +"Called by the controller's own timer", and there was no timer. Recovery depended +on a human noticing a stalled job and posting to `POST /api/compute/leases/sweep` +by hand. The route is still there — driving it by hand is how you watch the +mechanism work the first time — but it is a diagnostic now, not the mechanism. + +Two consequences for an operator: + +- **A restart mid-job no longer strands it.** The compute repository is built at + startup rather than on first use, specifically so the sweeper has something to + sweep. A deployment with no database configured gets no repository and no + sweeper, which is not a degraded compute tier but the absence of one. +- **Watch for `returned N job(s) whose lease expired` in the log.** The sweeper + logs a warning per recovery round, because a sweeper that dies quietly is worse + than no sweeper: the queue simply stops returning work. + +The sweeper survives a failing round and logs the traceback. A loop that stopped +on the first error would look identical to a queue with nothing in it. + +### The agent cannot be signalled, by anyone who does not own it + +A node agent runs as its own unix account, so its supervisor cannot `kill(2)` it — +that fails across accounts. Stopping one therefore goes through a `stop_file` in +its config, which the agent polls each iteration. Fault injection +(`--fault-exit-after-claim N`) makes it `os._exit` instead, for exercising lease +recovery against a process that really vanished. + +Two sudo details, both learned by running it rather than by reading about it: + +- `sudo` **resets the environment**, so `UMBRA_NODE_KEY` does not reach the agent + unless the sudoers policy preserves it. `sudo -n -u … --preserve-env=UMBRA_NODE_KEY` + is what the proof of concept uses, because the alternative (`VAR=value` before + the command) puts the key in argv, which every user on the host can read from + `/proc//cmdline`. Where the policy refuses, the agent takes the key on + stdin instead. +- `sudo` **forks** before exec'ing, so the process you spawned is the wrapper and + the agent is its child. Signalling the wrapper kills the wrapper, leaves the + agent running, and renews its lease for ever. + ### Deploying this needs no migration The `projects` table, its `(id, organization_id)` unique constraint, the diff --git a/poc/poc_node_user.sh b/poc/poc_node_user.sh index be777a8..6e4a78a 100755 --- a/poc/poc_node_user.sh +++ b/poc/poc_node_user.sh @@ -6,6 +6,7 @@ # poc/poc_node_user.sh verify assert the boundary holds (needs sudo -u) # poc/poc_node_user.sh run verify, then run the whole PoC # poc/poc_node_user.sh status what exists right now +# poc/poc_node_user.sh wait [url] [n] wait for a controller, then stop # # Why an account at all, when the agent is one file and no dependencies: because # the boundary worth proving is that a worker on a node has *no* database @@ -17,6 +18,16 @@ # /home/umbra. That is the whole reason the setup is this short: no group # memberships, no path grants, nothing to widen later. # +# Two things `run` needs from the host, both about privilege and neither obvious: +# +# 1. `sudo -n -u umbra-poc` must work without a password. It is run from inside +# a loop, so a prompt would hang rather than fail. +# 2. sudo must carry the node key through to the agent. sudo resets the +# environment by policy, so `UMBRA_NODE_KEY` does not arrive unless the +# sudoers policy allows it. The demonstration checks this before it starts +# and, if the key will not survive, sends it down a pipe instead - so the +# run succeeds either way, and says which way it used. `verify` reports it. +# # Nothing here touches the deployment database or the deployment service. set -euo pipefail @@ -34,12 +45,39 @@ CONFIG="$AGENT_HOME/node.json" # Facts about this host, discovered rather than assumed. Overridable so the # script is not welded to one machine. +# +# Two interpreters, and they are not interchangeable: +# +# PYTHON_BIN the *agent's* interpreter. /usr/bin/python3, outside any +# home directory, because umbra-poc cannot reach umbra's. +# CONTROLLER_PY the *controller's*. Inside umbra's home and inside the +# project venv, which is fine: the controller runs as umbra. +# Resolved explicitly rather than shelling out to `uv`, which +# is not on PATH in a non-login shell and produced +# "exec: uv: not found" the first time this ran unattended. CONTROLLER_URL="${UMBRA_CONTROLLER_URL:-http://127.0.0.1:8543}" PYTHON_BIN="${UMBRA_PYTHON_BIN:-/usr/bin/python3}" pass_count=0 fail_count=0 +# The controller's interpreter: the project venv if it is there, otherwise +# whatever python3 is on PATH. The controller runs as umbra, so reading the venv +# is expected; it is only the *agent* that must not need it. +resolve_controller_python() { + local repo venv + repo="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" + venv="$repo/.venv/bin/python" + if [[ -x "$venv" ]]; then + printf '%s' "$venv" + elif command -v python3 >/dev/null 2>&1; then + command -v python3 + else + printf '' + fi +} +CONTROLLER_PY="$(resolve_controller_python)" + bold() { printf '\033[1m%s\033[0m\n' "$*"; } ok() { printf ' \033[32mok\033[0m %s\n' "$*"; pass_count=$((pass_count + 1)); } bad() { printf ' \033[31mFAIL\033[0m %s\n' "$*"; fail_count=$((fail_count + 1)); } @@ -84,6 +122,151 @@ require_agent_runner() { return 0 } +# --------------------------------------------------------------------------- +# wait for the controller +# --------------------------------------------------------------------------- +# +# Bounded retry with visible progress, because "no controller" has two very +# different causes and only one of them is fixed by waiting: the process is +# still starting, or it is not there and nothing is going to bring it. An +# immediate failure is unhelpful for the first; an unbounded wait is harmful for +# the second. So: poll, log every attempt, back off to a cap, and stop. +# +# Each attempt records *why* it failed, because the three reasons need different +# responses and a bare "unreachable" does not distinguish them: +# +# connection refused the process is not listening; usually still starting +# name resolution a typo in the URL, or no DNS for that host +# HTTP 500 the controller is up and broken; waiting may not help, +# and the log is where the answer is + +WAIT_ATTEMPTS="${UMBRA_WAIT_ATTEMPTS:-12}" +WAIT_INTERVAL="${UMBRA_WAIT_INTERVAL:-1}" +WAIT_MAX_INTERVAL="${UMBRA_WAIT_MAX_INTERVAL:-10}" +WAIT_LOG_FILE="${UMBRA_WAIT_LOG:-}" + +# Every line of the wait goes to stderr, so a caller running this in a command +# substitution does not capture the progress as if it were a result. When a log +# file is set it gets a copy. Written with a function rather than a duplicated +# file descriptor, because juggling fd 3 across a loop and two early returns is +# how a diagnostic ends up printing "Bad file descriptor" when it is the one thing +# you most need to read. +wait_say() { + if [[ -n "$WAIT_LOG_FILE" ]]; then + printf '%s\n' "$*" | tee -a "$WAIT_LOG_FILE" >&2 + else + printf '%s\n' "$*" >&2 + fi +} + +# The host part of a URL, without the port. `getent hosts 127.0.0.1:8543` fails +# on the port, and a literal address reported as "does not resolve" sends whoever +# is reading it to look at DNS when the actual problem is nothing listening. +url_host() { + local rest="${1#*://}" + rest="${rest%%/*}" + printf '%s' "${rest%%:*}" +} + +# Accept either a base URL or a health URL. Callers reasonably pass +# http://127.0.0.1:8543 and reasonably expect that to work, and probing it +# verbatim 404s and reports a healthy controller as broken. +health_url() { + case "$1" in + */api/health) printf '%s' "$1" ;; + */) printf '%s' "${1}api/health" ;; + */*) printf '%s/api/health' "$1" ;; + *) printf '%s/api/health' "$1" ;; + esac +} + +# Ask the controller's own health endpoint and classify the answer. +probe_controller() { + local url host code + url="$(health_url "$1")" + host="$(url_host "$url")" + host="$(url_host "$url")" + if curl -fsS --max-time 4 "$url" >/dev/null 2>&1; then + printf 'up' + return 0 + fi + code="$(curl -sS -o /dev/null -w '%{http_code}' --max-time 4 "$url" 2>/dev/null || true)" + if [[ "$code" == "404" ]]; then + # Something is listening and it is not the health endpoint. Usually the + # wrong path, and "connection refused" would send the reader to the wrong + # place entirely. + printf 'answering 404 (wrong path? %s)' "$url" + return 1 + fi + if [[ "$code" != "000" ]]; then + # Up and unhappy. The controller's own log is where the reason is, and + # waiting may not be the fix. + printf 'answering HTTP %s' "$code" + return 1 + fi + if [[ "$host" =~ ^[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+$ ]] || [[ "$host" == "localhost" ]]; then + # A literal address or localhost: nothing is listening, or a firewall is + # dropping it. "Refused" is the far more common of the two. + printf 'connection refused' + elif getent hosts "$host" >/dev/null 2>&1; then + printf 'connection refused' + else + printf 'host does not resolve' + fi + return 1 +} + +do_wait() { + local url="${1:-$CONTROLLER_URL}" + local attempts="${2:-$WAIT_ATTEMPTS}" + + wait_say "" + wait_say "Waiting for the controller at $(health_url "$url")" + local started elapsed interval attempt reason + started=$(date +%s) + interval="$WAIT_INTERVAL" + for (( attempt = 1; attempt <= attempts; attempt++ )); do + elapsed=$(( $(date +%s) - started )) + reason="$(probe_controller "$url" || true)" + if [[ "$reason" == "up" ]]; then + wait_say " ok controller is up after ${attempt} attempt(s), ${elapsed}s" + return 0 + fi + if (( attempt < attempts )); then + wait_say " .. attempt ${attempt}/${attempts} at +${elapsed}s: ${reason}; retrying in ${interval}s" + else + wait_say " .. attempt ${attempt}/${attempts} at +${elapsed}s: ${reason}" + fi + sleep "$interval" + if (( interval < WAIT_MAX_INTERVAL )); then + interval=$(( interval * 2 )) + (( interval > WAIT_MAX_INTERVAL )) && interval="$WAIT_MAX_INTERVAL" + fi + done + elapsed=$(( $(date +%s) - started )) + + wait_say "" + wait_say " FAIL no controller after ${attempts} attempt(s) over ${elapsed}s" + # A likely cause: the controller is up somewhere else. The PoC runs on 8543 + # and the deployment on 8442, and asking for the wrong one is the single most + # likely reason for being here. + local other + for other in ${UMBRA_ALT_CONTROLLERS:-http://127.0.0.1:8543 http://127.0.0.1:8442}; do + if [[ "$(health_url "$other")" != "$(health_url "$url")" ]] && [[ "$(probe_controller "$other" || true)" == "up" ]]; then + wait_say " note: a controller IS answering at ${other}" + wait_say " you asked for ${url}. Set UMBRA_CONTROLLER_URL to use it." + fi + done + wait_say "" + wait_say " start one with:" + wait_say " UMBRATRANS_DATABASE_DSN=postgresql:///umbratrans_scratch \\" + wait_say " UMBRA_ENV=test uv run python -m umbratrans_web serve \\" + wait_say " --host 127.0.0.1 --port ${url##*:} --repository postgres" + wait_say "" + wait_say " or wait longer: UMBRA_WAIT_ATTEMPTS=40 $0 wait" + return 1 +} + # --------------------------------------------------------------------------- # setup # --------------------------------------------------------------------------- @@ -277,20 +460,23 @@ do_verify() { do_run() { do_verify || true head_ "Running the proof of concept with the agent as $AGENT_USER" - if ! curl -fsS --max-time 5 "$CONTROLLER_URL/api/health" >/dev/null 2>&1; then - bad "no controller at $CONTROLLER_URL" - note "start it with: UMBRATRANS_DATABASE_DSN=\$DSN UMBRA_ENV=test \\" - note " uv run python -m umbratrans_web serve --host 127.0.0.1 --port 8543 --repository postgres" + # Wait rather than failing on the first refusal: the controller may be a few + # seconds behind a restart. Bounded, and every attempt logged, so a genuine + # absence says so instead of hanging. + if ! do_wait "$CONTROLLER_URL" "$WAIT_ATTEMPTS"; then return 2 fi - ok "controller is up at $CONTROLLER_URL" if [[ $fail_count -gt 0 ]]; then printf '\n\033[33mRunning anyway; the boundary problems above are the interesting part.\033[0m\n' fi + if [[ -z "$CONTROLLER_PY" ]]; then + bad "no python for the controller; set one in the venv or on PATH" + return 2 + fi local repo_dir; repo_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" - ( cd "$repo_dir" && exec uv run python poc/run_poc.py \ + ( cd "$repo_dir" && exec "$CONTROLLER_PY" poc/run_poc.py \ --controller "$CONTROLLER_URL" \ --agent-user "$AGENT_USER" \ "$@" ) @@ -307,12 +493,26 @@ do_status() { note "home : $(getent passwd "$AGENT_USER" | cut -d: -f6)" note "shell: $(getent passwd "$AGENT_USER" | cut -d: -f7)" for path in "$AGENT_DIR/node_agent.py" "$CONFIG" "$ENV_FILE" "$REPO_DIR"; do - if [[ -e "$path" ]]; then + if [[ -r "$path" ]]; then note "$(stat -c '%A %U:%G %n' "$path")" + elif [[ -e "$path" ]]; then + # Exists, but this user cannot stat it. Reporting "missing" here + # would be a confident answer to a question nothing was consulted + # for, which is the failure this script has now hit twice. + note "exists, not readable by you $path" else note "missing $path" fi done + # The agent's own view of its home, which is the only way to learn what is + # in a directory this user cannot enter. + if as_agent true 2>/dev/null; then + for entry in "$AGENT_HOME" "$CONFIG"; do + note "as $AGENT_USER: $(as_agent sh -c "test -e '$entry' && echo present || echo absent" 2>/dev/null) $entry" + done + else + note "as $AGENT_USER: unknown (no privilege drop)" + fi printf '\n' note "can read agent : $(agent_can_read "$AGENT_DIR/node_agent.py")" note "can read env file: $(agent_can_read "$ENV_FILE")" @@ -327,6 +527,10 @@ case "${1:-}" in setup) shift; do_setup "$@" ;; verify) shift; do_verify "$@" ;; run) shift; do_run "$@" ;; + wait) shift + controller="${1:-$CONTROLLER_URL}" + attempts="${2:-$WAIT_ATTEMPTS}" + do_wait "$controller" "$attempts" "${UMBRA_WAIT_LOG:-}" ;; status) shift; do_status "$@" ;; *) sed -n '3,20p' "${BASH_SOURCE[0]}" | sed 's/^# \{0,1\}//' diff --git a/poc/run_poc.py b/poc/run_poc.py index 9cd2c5e..99df687 100644 --- a/poc/run_poc.py +++ b/poc/run_poc.py @@ -33,6 +33,7 @@ import argparse import json import os import pwd +import shutil import signal import subprocess import sys @@ -205,11 +206,355 @@ class AgentHandle: user: str config_path: Path process: subprocess.Popen | None = None + #: How the node key reached this agent. Empty for the ordinary case. + key_transport: str = "" + #: Where to ask this agent to stop. Defaults to ``.stop``. + stop_file: Path | None = None def describe(self) -> str: return f"{self.name} (uid {self.user})" +#: Where `poc_node_user.sh setup` installs the agent, and the interpreter it may +#: use. Both are outside umbra's home directory, which matters: umbra's home is +#: 0750, so an agent running as another user cannot execute anything inside it - +#: including the project venv. Launching the agent with umbra's interpreter +#: fails with "Permission denied" and proves nothing about the boundary. +INSTALLED_AGENT = Path("/opt/umbra/node-agent/node_agent.py") +INSTALLED_PYTHON = "/usr/bin/python3" + + +def resolve_agent_command(run_as: str) -> tuple[str, str, str]: + """Return ``(python, script, description)`` for the agent to run under. + + Prefers the installed copy when running as a different user, because that is + the arrangement a real node has: a self-contained file on a system + interpreter, with no access to the controller's home directory. The + description is returned so the caller can say which one it used - a proof of + concept that quietly tested the repository copy instead of the installed one + would not be testing what it claims to. + """ + + me = pwd.getpwuid(os.getuid()).pw_name + repo_script = Path(__file__).resolve().parents[1] / "umbratrans_web" / "node_agent.py" + if run_as and run_as != me and INSTALLED_AGENT.exists(): + return ( + INSTALLED_PYTHON, + str(INSTALLED_AGENT), + f"installed copy at {INSTALLED_AGENT} on {INSTALLED_PYTHON}", + ) + return sys.executable, str(repo_script), f"repository copy at {repo_script} on {sys.executable}" + + +def _spawn_prefix(run_as: str) -> tuple[list[str], str | None]: + """How to become ``run_as``, and an explanation if we cannot. + + A process that is not root **cannot** change to another user: ``setresuid`` + requires privilege. So ``setpriv`` is correct only when this controller is + already root, and the first version of this script used it unconditionally - + which failed with ``setresuid: Operation not permitted`` for every run as an + ordinary user, and the agent died before it wrote a line of log. + + ``sudo -n -u`` is the mechanism that works from an unprivileged controller, + and ``-n`` keeps it from blocking on a password prompt inside a loop. + + Two sudo options are not interchangeable here, and the difference is where a + secret ends up: + + sudo -n -u agent UMBRA_NODE_KEY=... the key is in sudo's argv, and + argv is world-readable through + /proc//cmdline for as long as + the process lives + sudo -n -u agent --preserve-env=... the key is in the environment, + which /proc//environ exposes + only to that user and to root + + So the key goes in the environment, and ``env_reset`` - which is usually the + thing that breaks environment passing - is doing us a favour here by stripping + everything else: the controller's ``PYTHONPATH``, ``VIRTUAL_ENV`` and + ``LD_LIBRARY_PATH`` all point into umbra's home, which umbra-poc cannot read + and has no business seeing. + + ``-H`` gives the agent the target user's own home rather than inheriting + umbra's, which is both more correct and the difference between a writable + directory and a 0750 one the agent cannot enter. + """ + + me = pwd.getpwuid(os.getuid()).pw_name + if not run_as or run_as == me: + return [], None + if os.geteuid() == 0: + # Already root: no need for sudo, and the environment passes through. + return ["setpriv", "--reuid", run_as, "--regid", run_as, "--init-groups", "--"], None + if shutil.which("sudo"): + return ["sudo", "-n", "-H", "-u", run_as, "--preserve-env=UMBRA_NODE_KEY", "--"], None + if shutil.which("setpriv"): + return ( + [], + f"this process is not root, so it cannot change to {run_as} itself. Run under " + f"sudo, or install a NOPASSWD rule for 'sudo -n -u {run_as}', or start the " + "agent yourself and pass --node-key-stdin.", + ) + return ( + [], + f"this process is not root and has no sudo, so it cannot become {run_as}. Start " + "the agent yourself and pass --node-key-stdin.", + ) + + +def _env_survives_sudo(run_as: str, name: str) -> tuple[bool, str]: + """Ask sudo whether ``name`` survives the trip to ``run_as``. + + Worth asking rather than assuming, because the answer is a property of the + sudoers policy on the host and not of this code. A sudoers file without the + SETENV tag refuses ``--preserve-env``, and the symptom otherwise arrives 400 + lines away as the agent saying ``UMBRA_NODE_KEY is not set`` - which reads as + a bug in the agent and is really a policy on the machine. + + A throwaway value is used: the point is whether the name arrives at all. + """ + + if os.geteuid() == 0 or not run_as or run_as == pwd.getpwuid(os.getuid()).pw_name: + return True, "" + prefix, problem = _spawn_prefix(run_as) + if problem: + return False, problem + probe = subprocess.run( + prefix + ["/bin/sh", "-c", 'test -n "$' + name + '" && printf pass || printf fail'], + capture_output=True, + text=True, + env={**os.environ, name: "probe-value"}, + timeout=20, + ) + if probe.stdout.strip() == "pass": + return True, "" + # Only sudo's own complaint is worth quoting. The probe's stdout carries our + # own "fail" token, and stitching that into the sentence produced + # "sudo will not pass UMBRA_NODE_KEY through to umbra-poc: fail", which tells + # the reader nothing they did not already know. + detail = (probe.stderr or "").strip().splitlines() + return False, ( + f"sudo will not pass {name} through to {run_as}." + + (f" sudo said: {detail[-1]}" if detail else "") + + f"\n Either add 'Defaults:{pwd.getpwuid(os.getuid()).pw_name} env_keep += \"{name}\"'" + + " to /etc/sudoers,\n or run the controller as root, or start the agent yourself" + + " and pass the key on stdin.\n Nothing secret was sent: the probe used a throwaway value." + ) + + +def agent_command( + *, + config_path: Path, + key: str, + run_as: str, + extra_env: dict[str, str] | None = None, + extra_args: list[str] | None = None, + forced_stdin_key: bool = False, +) -> tuple[list[str], dict[str, str], bool, str]: + """Build ``(argv, environment, key_on_stdin, note)`` for the agent. + + Separated from the two ways of running it - a background ``Popen`` and a + blocking ``run`` - so that the privilege drop and the key transport cannot + differ between them. They did once, and the difference showed up only in the + steps that used the blocking form. + """ + + python, agent_script, _ = resolve_agent_command(run_as) + prefix, problem = _spawn_prefix(run_as) + if problem: + raise RuntimeError(problem) + + note = "" + on_stdin = forced_stdin_key + if not on_stdin: + survives, why = _env_survives_sudo(run_as, "UMBRA_NODE_KEY") + if survives: + pass + elif prefix and shutil.which("sudo"): + # Recoverable: fall back to the pipe rather than refusing to run. + on_stdin, note = True, why + else: + raise RuntimeError(why) + + environment = _agent_environment(extra_env) + if on_stdin: + environment.pop("UMBRA_NODE_KEY", None) + else: + environment["UMBRA_NODE_KEY"] = key + + command = prefix + [python, agent_script, "--config", str(config_path), "--log-level", "INFO"] + if extra_args: + command += extra_args + if on_stdin: + command += ["--node-key-stdin"] + return command, environment, on_stdin, note + + +def launch_agent_process( + *, + config_path: Path, + key: str, + run_as: str, + log_path: Path, + extra_env: dict[str, str] | None = None, + extra_args: list[str] | None = None, + forced_stdin_key: bool = False, +) -> tuple[subprocess.Popen, str]: + """Start the agent as ``run_as`` and return ``(process, how_the_key_arrived)``. + + One place decides how the node key reaches the agent, because there are four + call sites and the failure mode of getting it wrong is quiet: the agent starts, + logs nothing about credentials, and a hundred lines later the proof of concept + reports that the node never came online. + + Two transports, neither of which puts the key in a file: + + ``environment`` the key sits in the child's environment. Right when the + privilege drop preserves the environment, and wrong when it + does not - sudo resets it by policy, so the variable is + simply gone by the time the agent looks. + ``stdin`` the key is written to a pipe the agent reads once. Works + whatever sudo does to the environment, and keeps the key out + of argv, where every user on the host can read it from + /proc. Used when the environment cannot be trusted, which is + detected by asking rather than by finding out. + """ + + command, environment, on_stdin, note = agent_command( + config_path=config_path, + key=key, + run_as=run_as, + extra_env=extra_env, + extra_args=extra_args, + forced_stdin_key=forced_stdin_key, + ) + process = subprocess.Popen( + command, + stdin=subprocess.PIPE if on_stdin else None, + stdout=log_path.open("wb"), + stderr=subprocess.STDOUT, + env=environment, + text=True, + # Its own session, which makes it a process-group leader. Two reasons, + # both learned the hard way: + # + # - sudo forks. The Popen handle is sudo, and sudo forks the agent + # before exec'ing, so signalling the handle kills the *supervisor* and + # orphans the worker. The step that is supposed to demonstrate a power + # cut was killing a wrapper and leaving the agent happily renewing its + # lease, which is the opposite of a power cut and the reason a killed + # agent's job appeared not to come back. + # - the agent no longer shares the controller's group, so a Ctrl-C aimed + # at the controller does not take the workers with it, or the reverse. + start_new_session=True, + ) + if on_stdin: + # Written and closed at once, so a slow agent never sees a writer still + # attached to its stdin and cannot mistake the pipe for a terminal. + try: + process.stdin.write(key + "\n") + process.stdin.flush() + finally: + process.stdin.close() + return process, note + + +def signal_agent_group(process: subprocess.Popen, number: int) -> None: + """Signal the whole process group, not just the process we spawned. + + Falls back to the single process if the group is already gone, because a + process that exited between the check and the signal raises, and "the agent is + already dead" is not an error worth raising for. + """ + + try: + os.killpg(os.getpgid(process.pid), number) + except (ProcessLookupError, PermissionError): + try: + process.send_signal(number) + except ProcessLookupError: + pass + + +def agent_pids_for_config(config_path: Path) -> list[int]: + """Every live process running the agent against ``config_path``, by pid. + + Found by reading /proc rather than by holding on to a pid, which is the whole + point. The process this driver spawned is the ``sudo`` wrapper; the worker is + the wrapper's child, owned by a different account, and the two are not + interchangeable. Checking the wrapper told us nothing about the worker, and + "the wrapper died" was once reported as "the agent died" - which is how a + proof of concept claimed to demonstrate a power cut while leaving the agent + running, renewing its lease, and holding a job for ever. + + Each config file is unique to one agent in one run's working directory, so + matching on it identifies the worker without trusting the driver, the wrapper, + or anything else this process happens to have spawned. + """ + + needle = str(config_path) + alive: set[int] = set() + for entry in Path("/proc").iterdir(): + if not entry.name.isdigit(): + continue + try: + cmdline = (entry / "cmdline").read_bytes().split(b"\0") + except OSError: + continue # the process exited between listing and reading + if any(arg.decode(errors="replace") == needle for arg in cmdline if arg): + alive.add(int(entry.name)) + return sorted(alive) + + +def watch_agent_pids(config_path: Path, *, window: float = 1.5) -> list[int]: + """Every pid that ever ran the agent against ``config_path`` while we watched. + + Sampling repeatedly and keeping the union, rather than stopping at the first + hit. Two processes match - the ``sudo`` wrapper this driver spawned and the + worker underneath it - and they are not simultaneous: the wrapper appears + first, and the worker may be gone within a hundred milliseconds of appearing, + because the fault under test exits as soon as it claims its job. A single + early sample therefore finds the wrapper and never the worker, which is how a + check meant to prove the agent vanished came to prove only that its launcher + did. + + ``/proc//cmdline`` is world-readable, so the worker of another account is + visible here even though ``/proc//environ`` is not. That asymmetry is + also why the node key travels in the environment rather than on a command + line: the environment is private, argv is not. + """ + + seen: set[int] = set() + deadline = time.time() + window + while time.time() < deadline: + seen.update(agent_pids_for_config(config_path)) + time.sleep(0.02) + return sorted(seen) + + +def check_agent_survived( + process: subprocess.Popen, log_path: Path, *, who: str +) -> None: + """Fail loudly and usefully if an agent died at startup. + + The alternative is an empty log and a "node did not come online" much further + on, which points at the controller rather than at the thing that failed - and + which, in this project's history, has been the shape of three separate bugs. + """ + + time.sleep(0.4) + if process.poll() is None: + return + try: + detail = log_path.read_text(errors="replace").strip().splitlines()[-3:] + except OSError: + detail = [] + raise RuntimeError( + f"the agent for {who} exited immediately (status {process.returncode}). " + + (f"Its log says: {' | '.join(detail)}" if detail else "and wrote nothing.") + ) + + def start_agent( *, agent_name: str, @@ -221,6 +566,7 @@ def start_agent( kinds: list[str] | None = None, log_path: Path, extra_env: dict[str, str] | None = None, + key_on_stdin: bool = False, ) -> AgentHandle: """Write a config the agent user can read, and start the agent as them. @@ -232,7 +578,11 @@ def start_agent( config = { "controller_url": controller_url, - "node_key": "", # supplied via UMBRA_NODE_KEY + # Deliberately empty: the key is supplied by the environment, or on stdin + # when the environment cannot be trusted to survive the privilege drop. + # A config file holding a secret is a secret that eventually lands in a + # backup, a tar, or a support bundle. + "node_key": "", "node_name": agent_name, "kinds": kinds or [], "lease_seconds": 30, @@ -240,39 +590,71 @@ def start_agent( } config_path = workdir / f"node-{agent_name}.json" config_path.write_text(json.dumps(config, indent=2)) - # World-readable on purpose: the config holds no secret. The key travels in - # the environment, so a config file that had to be protected would be a sign - # the key was in the wrong place. config_path.chmod(0o644) - agent_script = Path(__file__).resolve().parents[1] / "umbratrans_web" / "node_agent.py" - python = sys.executable - command = [python, str(agent_script), "--config", str(config_path), "--log-level", "INFO"] - if run_as and run_as != pwd.getpwuid(os.getuid()).pw_name: - command = ["setpriv", "--reuid", run_as, "--regid", run_as, "--init-groups", "--"] + command - environment = _agent_environment(extra_env) - environment["UMBRA_NODE_KEY"] = key - log = log_path.open("wb") - process = subprocess.Popen(command, stdout=log, stderr=subprocess.STDOUT, env=environment) - return AgentHandle( + process, note = launch_agent_process( + config_path=config_path, + key=key, + run_as=run_as, + log_path=log_path, + extra_env=extra_env, + forced_stdin_key=key_on_stdin, + ) + handle = AgentHandle( name=agent_name, node_id=node_id, key=key, user=run_as or pwd.getpwuid(os.getuid()).pw_name, config_path=config_path, process=process, + key_transport=note or "environment", ) + check_agent_survived(process, log_path, who=agent_name) + return handle -def stop_agent(agent: AgentHandle, *, signal_number: int = signal.SIGTERM) -> None: +def stop_agent(agent: AgentHandle, *, signal_number: int = signal.SIGTERM) -> list[int]: + """Stop an agent, and return the pids that refused to stop. + + A file the agent watches, not a signal. Signalling is not available: this + process does not own the agent, and kill(2) against another account's process + fails. Signalling the ``sudo`` wrapper that *is* ours succeeds, kills the + wrapper, and orphans the worker - so the driver would report success and leave + an agent renewing leases. That happened on every run until the pids were + counted. + """ + if agent.process is None: - return - agent.process.send_signal(signal_number) + return [] + stop_file = agent.config_path.with_suffix(".stop") + if agent.stop_file: + stop_file = agent.stop_file + try: + stop_file.write_text("stop\n") + except OSError: + pass + + deadline = time.time() + 6 + while time.time() < deadline: + if not agent_pids_for_config(agent.config_path): + return [] + time.sleep(0.2) + + # Out of patience. Kill what we can - the wrapper - and then check again, + # reporting anything left rather than assuming it went. + signal_agent_group(agent.process, signal_number) try: agent.process.wait(timeout=5) except subprocess.TimeoutExpired: - agent.process.kill() + signal_agent_group(agent.process, signal.SIGKILL) agent.process.wait(timeout=5) + leftovers = agent_pids_for_config(agent.config_path) + if leftovers: + print( + f"\033[31mWARNING\033[0m {len(leftovers)} agent process(es) for {agent.name} " + f"survived a stop: {leftovers}. They will keep renewing their leases." + ) + return leftovers def run_agent_once( @@ -286,20 +668,27 @@ def run_agent_once( agent whose lease behaviour is untested. """ - agent_script = Path(__file__).resolve().parents[1] / "umbratrans_web" / "node_agent.py" - python = sys.executable - uid = pwd.getpwnam(agent.user) - command = [ - python, str(agent_script), "--config", str(agent.config_path), - "--log-level", "INFO", "--max-iterations", str(iterations), - ] - if agent.user != pwd.getpwuid(os.getuid()).pw_name: - command = ["setpriv", "--reuid", agent.user, "--regid", str(uid.pw_gid), "--init-groups", "--"] + command - environment = _agent_environment() - environment["UMBRA_NODE_KEY"] = agent.key - completed = subprocess.run( - command, capture_output=True, text=True, timeout=timeout, env=environment + command, environment, on_stdin, _ = agent_command( + config_path=agent.config_path, + key=agent.key, + run_as=agent.user, + extra_args=["--max-iterations", str(iterations)], ) + if on_stdin: + # A pipe is the transport here too. Feeding it from a string rather than + # PIPE is simpler, and the amount is one line, so it cannot block. + completed = subprocess.run( + command, + input=agent.key + "\n", + capture_output=True, + text=True, + timeout=timeout, + env=environment, + ) + else: + completed = subprocess.run( + command, capture_output=True, text=True, timeout=timeout, env=environment + ) log_path.write_text(completed.stdout + completed.stderr) return completed.stdout + completed.stderr @@ -387,6 +776,25 @@ def demonstrate( # -- 3. the agent authenticates and the node comes online ------------------ _head("3. The agent heartbeats, and only then is the node online") + _, _, which = resolve_agent_command(agent_user) + _note(f"running the {which}") + if which.startswith("repository") and agent_user != pwd.getpwuid(os.getuid()).pw_name: + _note("WARNING: the installed copy is missing, so this is testing the repository") + _note("copy under another user - which that user cannot execute. Run: sudo") + _note("poc/poc_node_user.sh setup") + # Report the key's route up front rather than letting a missing key surface + # later as "the node never came online". + _command, _env, on_stdin, why = agent_command( + config_path=workdir / "node-poc-agent.json", + key="probe", + run_as=agent_user, + ) + if on_stdin: + _note("node key transport: a pipe, because sudo will not carry the environment") + for line in why.splitlines(): + _note(f" {line}") + else: + _note("node key transport: the agent's environment, never a file and never argv") devices, measurements = measure() _note(f"measured on this host as {local_name}:") for device in devices: @@ -507,12 +915,19 @@ def demonstrate( {"kind": "sleep", "payload": {"seconds": 0.6}, "idempotency_key": f"poc-concurrent-{index}"}, ) # Both agents run at once, so the contention is real rather than staged. - runners = [subprocess.Popen( - [sys.executable, str(Path(__file__).resolve().parents[1] / "umbratrans_web" / "node_agent.py"), - "--config", str(handle.config_path), "--log-level", "INFO", "--max-iterations", "40"], - stdout=(workdir / f"{handle.name}.log").open("wb"), stderr=subprocess.STDOUT, - env={**_agent_environment(), "UMBRA_NODE_KEY": handle.key}) - for handle in (agent, controller_agent)] + runners = [] + for handle, extra in ((agent, ["--max-iterations", "40"]), + (controller_agent, ["--max-iterations", "40"])): + log_file = workdir / f"{handle.name}.log" + process, _ = launch_agent_process( + config_path=handle.config_path, + key=handle.key, + run_as=agent_user, + log_path=log_file, + extra_args=extra, + ) + check_agent_survived(process, log_file, who=f"{handle.name} (second)") + runners.append(process) try: deadline = time.monotonic() + 60 concurrent = [] @@ -560,55 +975,125 @@ def demonstrate( {"kind": "sleep", "payload": {"seconds": 600}, "idempotency_key": "poc-long"}, ) _note(f"submitted a 600-second job {long_job['id'][:8]}") - # Run the agent for real, let it claim, then SIGKILL it. Not a clean stop: a - # clean stop would release the lease, and a clean stop is not the failure - # being tested. SIGKILL is the closest thing to a power cut available. - victim = subprocess.Popen( - [sys.executable, str(Path(__file__).resolve().parents[1] / "umbratrans_web" / "node_agent.py"), - "--config", str(agent.config_path), "--log-level", "INFO"], - stdout=(workdir / "agent-victim.log").open("wb"), stderr=subprocess.STDOUT, - env={**_agent_environment(), "UMBRA_NODE_KEY": agent.key}, + + # The agent exits abruptly the moment it claims this job: no cleanup, no lease + # release, no goodbye. It does that from the inside, because the alternative + # does not work. + # + # Signalling it from here was the previous approach, and it never was a power + # cut. This driver runs as `umbra`; the agent runs as `umbra-poc`. It cannot + # signal a process owned by another account. Signalling the *sudo wrapper* it + # had spawned does work - that wrapper is its own - and killing it leaves the + # worker running, renewing its lease, holding the job for ever. kill(2) on a + # process group reports success as soon as one member has been signalled, so + # the wrapper's death was read as the agent's. Every run leaked an agent that + # way, and the "power cut" that was printed above it described something that + # had not happened. + # + # What makes this honest rather than merely different: the worker is + # identified by its own pid, found in /proc by its config path, and the run + # fails if that pid is still there afterwards. + victim_log = workdir / "agent-victim.log" + victim_config = workdir / "node-poc-victim.json" + victim_config.write_text(agent.config_path.read_text()) + victim_config.chmod(0o644) + _note("the agent will exit abruptly on claiming it, from the inside") + victim, _ = launch_agent_process( + config_path=victim_config, + key=agent.key, + run_as=agent_user, + log_path=victim_log, + extra_args=["--fault-exit-after-claim", "1"], ) - claimed_state = wait_for_status(admin, organization_id, long_job["id"], "running", timeout=25) - _note(f"claimed: status={claimed_state['status'] if claimed_state else '?'} " - f"attempt={claimed_state['attempt'] if claimed_state else '?'}") - if not claimed_state or claimed_state["status"] != "running": + + # Catch it while it is still there, so "it is gone" means something: a pid + # that was never observed is not evidence of anything. + worker_pids = watch_agent_pids(victim_config, window=3.0) + _note(f"the driver holds pid {victim.pid} (the sudo wrapper); every process running " + f"the agent was {worker_pids}") + if len(worker_pids) < 2: + _note("note: only one process was seen - the wrapper. The worker is short-lived " + "here, so this check is weaker than it looks.") + + claimed = wait_for_status(admin, organization_id, long_job["id"], "running", timeout=25) + if not claimed: failures += 1 - _fail("the agent never claimed the long job, so there is no lease to expire") + _fail("no agent ever claimed the long job, so there is no lease to expire") else: _pass("the agent holds a lease on a job it will take ten minutes to finish") - events_before = admin.get(f"/api/compute/jobs/{long_job['id']}/events") - _note("events before the kill: " + " -> ".join(e["event"] for e in events_before)) - _note("SIGKILL: no cleanup, no lease release, no goodbye - exactly like a power cut") - victim.send_signal(signal.SIGKILL) - victim.wait(timeout=5) - time.sleep(2) - still_running = admin.get(f"/api/compute/jobs?organization_id={organization_id}") - held = next(j for j in still_running if j["id"] == long_job["id"]) - _note(f"immediately after the kill: status={held['status']} " - f"(the lease is still held - nothing released it)") - if held["status"] == "running": - _pass("the job is still held by a node that no longer exists") + deadline = time.time() + 20 + while time.time() < deadline and agent_pids_for_config(victim_config): + time.sleep(0.2) + gone = agent_pids_for_config(victim_config) + if not worker_pids: + failures += 1 + _fail("the agent worker was never observed, so its absence proves nothing") + elif gone: + failures += 1 + _fail(f"the agent worker {gone} is still running after it claimed the job, so " + "nothing below this means anything") else: - _note(f"note: status was already {held['status']}") + _pass("the worker is gone: its own pid is no longer in /proc") - _note("waiting for the lease to expire (30s), then sweeping") - time.sleep(31) - swept = admin.post("/api/compute/leases/sweep", {}) - recovered = next( + held = next( j for j in admin.get(f"/api/compute/jobs?organization_id={organization_id}") if j["id"] == long_job["id"] ) - _note(f"after the sweep: status={recovered['status']} attempt={recovered['attempt']}") - _note(f"error: {recovered['error']!r}") - events_after = admin.get(f"/api/compute/jobs/{long_job['id']}/events") - _note("event trail: " + " -> ".join(e["event"] for e in events_after)) - if long_job["id"] in swept and recovered["status"] in ("queued", "failed"): - _pass("the expired lease returned the job to the queue rather than losing it (I26)") + _note(f"after the agent vanished: status={held['status']} " + f"(the lease is still held - nothing released it)") + if held["status"] == "running": + _pass("the job is still held by an agent that no longer exists") else: failures += 1 - _fail("the expired lease did not return the job") + _fail(f"the job was already {held['status']}, so it was never left holding") + + # Wait for the *controller's own* sweeper, without asking anyone to sweep. + # The previous version of this step posted to the administrative sweep route + # itself, which made recovery look automatic while depending entirely on a + # human noticing a stalled job - and there was no timer, so that endpoint's + # docstring described something that did not exist. + _note("waiting for the controller's own sweeper - nobody is asked to sweep") + recovered = None + deadline = time.time() + 45 + while time.time() < deadline: + current = next( + j for j in admin.get(f"/api/compute/jobs?organization_id={organization_id}") + if j["id"] == long_job["id"] + ) + if current["status"] != "running": + recovered = current + break + time.sleep(1) + if recovered is None: + # Recovery did not happen on its own. Ask once, by hand, to separate "the + # mechanism is broken" from "the timer is missing", which are different + # problems and were previously reported as one "failed". + swept = admin.post("/api/compute/leases/sweep", {}) + recovered = next( + j for j in admin.get(f"/api/compute/jobs?organization_id={organization_id}") + if j["id"] == long_job["id"] + ) + _note(f"after a manual sweep: status={recovered['status']} swept={swept}") + if recovered["status"] != "running": + failures += 1 + _fail("the timer did not recover it, but a manual sweep did. The lease " + "mechanism works; the sweeper does not run.") + else: + failures += 1 + _fail("the lease did not expire even under a manual sweep, so the mechanism " + "itself is broken") + else: + _note(f"recovered without being asked: status={recovered['status']} " + f"attempt={recovered['attempt']}") + _note(f"error: {recovered['error']!r}") + _note("event trail: " + " -> ".join( + e["event"] for e in admin.get(f"/api/compute/jobs/{long_job['id']}/events"))) + if recovered["status"] in ("queued", "failed"): + _pass("a node that vanished had its job returned by the controller alone (I26)") + else: + failures += 1 + _fail(f"the job left 'running' as {recovered['status']!r} rather than being returned") # -- 9. slow is eligible, impossible is blocked --------------------------- _head("9. A model that does not fit in VRAM runs on RAM; one nothing can run is blocked") @@ -718,12 +1203,20 @@ def main(argv: list[str] | None = None) -> int: workdir = args.workdir or Path(tempfile.mkdtemp(prefix="umbra-poc-")) workdir.mkdir(parents=True, exist_ok=True) - # 0750, not the 0700 mkdtemp gives and not the 0755 a default umask would. - # The agent runs as another user and has to be able to read its config out of - # here, and the directory has to be traversable for that to work - while the - # node key, which is the thing worth protecting, is in the environment rather - # than in the file. - workdir.chmod(0o750) + # 0755, not the 0700 mkdtemp gives and not the 0750 that seemed tidier. + # The agent runs as a different user and has to traverse this directory to + # read its own config, and a different user is not the owner and not in the + # group, so anything less than o+x denies it outright. 0750 produced + # "Permission denied" on a file whose own mode was already 0644. + # + # What that costs: anyone on the box can read the agent's logs here. They hold + # job identifiers, node names and timings - fleet information, not secrets - + # and the node key is in the environment rather than in these files, which is + # the reason this is acceptable. For a real deployment put the workdir in the + # agent's own state directory and let the agent own it; here the controller + # has to write into it too, and two owners would mean a group that undoes the + # separation this is demonstrating. + workdir.chmod(0o755) print("\033[1mUmbraTrans compute proof of concept\033[0m") _note(f"controller : {args.controller}") _note(f"workdir : {workdir}") diff --git a/umbratrans_web/app.py b/umbratrans_web/app.py index 9d859e0..7bb9a18 100644 --- a/umbratrans_web/app.py +++ b/umbratrans_web/app.py @@ -20,7 +20,9 @@ Design decisions worth stating explicitly: from __future__ import annotations +import asyncio import logging +from contextlib import asynccontextmanager, suppress from datetime import datetime, timezone from http.cookies import SimpleCookie from pathlib import Path @@ -382,6 +384,90 @@ def _begin_setup_if_needed(gate: SetupGate, service: AccountService) -> None: LOGGER.warning("could not determine whether first-run setup is required", exc_info=True) +#: How often expired leases are swept, in seconds. Shorter than the shortest +#: lease any node may be configured with, so a job is never recovered later than +#: one lease period after its holder stopped renewing. +LEASE_SWEEP_SECONDS = 10.0 + + +async def _sweep_expired_leases(app: FastAPI) -> None: + """Return jobs whose holder stopped renewing, without being asked. + + The lease is the whole recovery mechanism: a worker that is ``kill -9``'d + never releases anything, so the expiry *is* the signal. And a signal nobody + listens to is not a signal. + + This used to exist only as an administrator endpoint, on the understanding + that a controller timer would call it. There was no such timer, so the + recovery path depended on a human noticing a stalled job and posting to an + administrative route by hand. That is not a mechanism; it is a diagnosis. The + proof of concept masked it, because the demonstration swept by hand itself. + + Errors are logged and the loop continues. A sweeper that dies on one bad lease + is a queue that silently stops recovering, which is the exact failure it + exists to prevent, and the failure would be invisible. + """ + + import asyncio + + log = logging.getLogger("umbratrans_web.compute") + while True: + try: + await asyncio.sleep(LEASE_SWEEP_SECONDS) + repository = getattr(app.state, "compute_repository", None) + recovered = repository.sweep_expired_leases() if repository is not None else [] + if recovered: + log.warning( + "returned %d job(s) whose lease expired: %s", + len(recovered), + ", ".join(str(job_id) for job_id in recovered), + ) + except asyncio.CancelledError: + raise + except Exception: + log.exception("lease sweep failed; will try again in %.0fs", LEASE_SWEEP_SECONDS) + + +@asynccontextmanager +async def _lifespan(app: FastAPI): + """Run the lease sweeper for as long as the application serves. + + The compute repository is built here rather than on first use, because the + sweeper needs one before any request arrives. A controller restarted while a + job was running must recover that job: if the repository were built lazily, + nothing would touch the compute surface, nothing would build it, and the + sweeper would have nothing to sweep. + + A deployment with no database configured gets no repository and no sweeper, + which is not a degraded compute tier but the absence of one. + """ + + if getattr(app.state.settings, "database_dsn", None): + from .compute import build_compute_repository + + try: + app.state.compute_repository = build_compute_repository( + app.state.settings, getattr(app.state, "password_service", None) + ) + except Exception: + # Refusing to serve would turn a compute problem into an outage of + # the account and project surfaces, which do not need it. The sweeper + # skips its round and the compute routes report the real error. + logging.getLogger("umbratrans_web.compute").exception( + "compute repository unavailable at startup; lease recovery is off until " + "a compute request builds one" + ) + + task = asyncio.create_task(_sweep_expired_leases(app)) + app.state.lease_sweeper = task + try: + yield + finally: + task.cancel() + with suppress(asyncio.CancelledError): + await task + + def create_app( *, settings: Settings | None = None, @@ -436,6 +522,7 @@ def create_app( docs_url=None, redoc_url=None, openapi_url=None, + lifespan=_lifespan, ) app.state.settings = resolved_settings app.state.repository = resolved_repository diff --git a/umbratrans_web/compute.py b/umbratrans_web/compute.py index 83dcb94..d110a3a 100644 --- a/umbratrans_web/compute.py +++ b/umbratrans_web/compute.py @@ -1454,3 +1454,20 @@ __all__ = [ "RUNNING", "TERMINAL_STATUSES", ] + + +def build_compute_repository(settings, passwords=None) -> "ComputeRepository": + """Build the compute repository from settings. + + Shared by the request dependency and by the application's startup, because + the lease sweeper needs a repository before any request arrives. Building it + lazily in the sweeper instead would mean a controller restarted mid-job never + recovers that job: nothing would ever touch the compute surface, so the + repository would never be built, so the sweeper would have nothing to sweep, + so the job would sit ``running`` for ever - which is the failure the sweeper + exists to prevent, reached by the route the sweeper takes to prevent it. + """ + + if not getattr(settings, "database_dsn", None): + raise RepositoryError("compute_unavailable", "No database is configured.") + return ComputeRepository(settings.database_dsn, passwords=passwords) diff --git a/umbratrans_web/node_agent.py b/umbratrans_web/node_agent.py index f6802c5..bffa533 100644 --- a/umbratrans_web/node_agent.py +++ b/umbratrans_web/node_agent.py @@ -34,7 +34,7 @@ import sys import time import urllib.error import urllib.request -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace from pathlib import Path from typing import Any, Mapping @@ -73,9 +73,37 @@ class NodeConfig: #: does not produce a tight reconnect loop. retry_backoff_seconds: float = 2.0 max_retry_backoff_seconds: float = 30.0 + #: A file whose existence means "stop". Optional, and the mechanism by which + #: something that is not allowed to signal this process can still ask it to + #: stop cleanly. + stop_file: Path | None = None + #: Fault injection: die abruptly after claiming this many jobs, with no + #: cleanup and no lease release. See :meth:`NodeAgent.run`. + exit_abruptly_after_claim: int = 0 @classmethod - def load(cls, path: Path) -> NodeConfig: + def key_from_stdin(cls) -> str: + """Read the node key from stdin, one line, and let go of it. + + A third place for the key, after the environment and the config file, for + the case where the environment cannot be trusted to survive the trip to + this process. The ordinary way to run an agent as another unix user is + ``sudo -u``, and sudo resets the environment by policy - so the key in + the environment silently does not arrive, and the agent reports only that + it found no key. A pipe sidesteps the policy: the value never appears in + argv, where every user on the host can read it from /proc, and never in a + file, which outlives the process. + """ + line = sys.stdin.readline().strip() + if not line: + raise AgentError( + "--node-key-stdin was given but stdin was empty. The writer closed the " + "pipe without sending a key." + ) + return line + + @classmethod + def load(cls, path: Path, *, key_from_stdin: bool = False) -> NodeConfig: raw = json.loads(path.read_text()) # The key is deliberately *not* in this list. It may come from the # environment, which is the preferred arrangement: a config file holding @@ -85,11 +113,21 @@ class NodeConfig: missing = [key for key in ("controller_url", "node_name") if not raw.get(key)] if missing: raise AgentError(f"{path} is missing: {', '.join(missing)}") - key = os.environ.get("UMBRA_NODE_KEY") or raw.get("node_key") + key = os.environ.get("UMBRA_NODE_KEY") + if not key and key_from_stdin: + key = cls.key_from_stdin() + key = key or raw.get("node_key") if not key: + hint = ( + "Pass --node-key-stdin if the key is coming through a pipe, which is what " + "happens when sudo resets the environment on the way here." + if key_from_stdin + else "If this agent is started through sudo, check whether sudo is " + "resetting the environment, and pass --node-key-stdin instead." + ) raise AgentError( f"{path} has no node_key and UMBRA_NODE_KEY is not set. Put the key in one " - "of them - the environment is preferable, so the file stays safe to copy." + f"of them - the environment is preferable, so the file stays safe to copy. {hint}" ) return cls( controller_url=str(raw["controller_url"]).rstrip("/"), @@ -98,6 +136,8 @@ class NodeConfig: kinds=list(raw.get("kinds", [])), lease_seconds=int(raw.get("lease_seconds", 60)), poll_interval_seconds=float(raw.get("poll_interval_seconds", 1.0)), + stop_file=Path(raw["stop_file"]) if raw.get("stop_file") else None, + exit_abruptly_after_claim=int(raw.get("exit_abruptly_after_claim", 0)), ) @@ -400,6 +440,7 @@ class NodeAgent: self._handlers = handlers if handlers is not None else default_handlers() self._stop = False self._active = 0 + self._claims_made = 0 #: What this node has told the controller is resident. Reported on every #: heartbeat so the console can distinguish a node that is idle from one #: that is idle and out of memory. §7.3 @@ -464,6 +505,20 @@ class NodeAgent: job["kind"], job_id[:8], placement.get("node_name", "?"), placement.get("device_kind", "?"), placement.get("reason", "?"), ) + self._claims_made += 1 + if ( + self._config.exit_abruptly_after_claim + and self._claims_made >= self._config.exit_abruptly_after_claim + ): + # Here, between taking the job and doing it. Not after the work: a + # fault that fires once the job has finished tests a finished job, + # and the lease it is meant to strand is already released. + LOGGER.warning( + "fault injection: exiting abruptly having claimed %d job(s), without " + "running it and without releasing the lease", + self._claims_made, + ) + os._exit(137) # A placement that names something to evict says so before the work starts, # because the cost of the eviction is somebody else's warm model. if placement.get("would_evict"): @@ -545,11 +600,21 @@ class NodeAgent: self.beat(force=True) iterations = 0 while not self._stop: + # Asked to stop by file. A supervisor that cannot signal this process - + # because it does not own it, which is the normal arrangement when an + # agent runs as a service account - still has to be able to stop it + # cleanly. Without this, stopping an agent meant killing the *sudo + # wrapper*, which orphaned the worker: every proof-of-concept run left + # an agent behind, quietly renewing leases for jobs nobody was watching. + if self._config.stop_file is not None and self._config.stop_file.exists(): + LOGGER.info("stop file %s appeared; stopping", self._config.stop_file) + break worked = False try: worked = self.claim_and_run() except Exception: # noqa: BLE001 - the loop outlives any single failure LOGGER.exception("unexpected error in the claim/run cycle") + if not self.beat(): # Backing off rather than spinning. A controller that is down # should not be hammered by every node on the network at once. @@ -568,16 +633,32 @@ class NodeAgent: def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="UmbraTrans compute node agent") parser.add_argument("--config", required=True, type=Path, help="path to the node JSON config") + parser.add_argument( + "--node-key-stdin", + action="store_true", + help="read the node key from stdin instead of the environment; for use under " + "sudo, which resets the environment by policy", + ) parser.add_argument("--log-level", default="INFO") parser.add_argument("--once", action="store_true", help="one claim/heartbeat cycle and exit") parser.add_argument("--max-iterations", type=int, default=None) + parser.add_argument( + "--fault-exit-after-claim", + type=int, + default=None, + metavar="N", + help="fault injection: exit abruptly with no cleanup after claiming N jobs, " + "so lease recovery can be exercised against a process that really vanished", + ) args = parser.parse_args(argv) logging.basicConfig( level=getattr(logging, str(args.log_level).upper(), logging.INFO), format="%(asctime)s %(levelname)-7s %(name)s: %(message)s", ) - config = NodeConfig.load(args.config) + config = NodeConfig.load(args.config, key_from_stdin=args.node_key_stdin) + if args.fault_exit_after_claim is not None: + config = replace(config, exit_abruptly_after_claim=args.fault_exit_after_claim) client = ControllerClient(config.controller_url, config.node_key) agent = NodeAgent(config, client) LOGGER.info( diff --git a/umbratrans_web/routes_compute.py b/umbratrans_web/routes_compute.py index eb79847..044f478 100644 --- a/umbratrans_web/routes_compute.py +++ b/umbratrans_web/routes_compute.py @@ -41,7 +41,6 @@ from .errors import ( AuthenticationRequiredError, ForbiddenError, InputError, - RepositoryError, ) from .models import Principal from .schemas import RequestModel @@ -339,16 +338,11 @@ def compute_repository(request: Request): cached = getattr(request.app.state, "compute_repository", None) if cached is not None: return cached - from .compute import ComputeRepository + from .compute import build_compute_repository - settings = request.app.state.settings - if not settings.database_dsn: - raise RepositoryError("compute_unavailable", "No database is configured.") - repository = ComputeRepository( - settings.database_dsn, - passwords=request.app.state.password_service - if hasattr(request.app.state, "password_service") - else None, + repository = build_compute_repository( + request.app.state.settings, + getattr(request.app.state, "password_service", None), ) request.app.state.compute_repository = repository return repository diff --git a/umbratrans_web/tests/test_compute.py b/umbratrans_web/tests/test_compute.py index 96637da..800212a 100644 --- a/umbratrans_web/tests/test_compute.py +++ b/umbratrans_web/tests/test_compute.py @@ -25,6 +25,7 @@ from __future__ import annotations import concurrent.futures import os +import time from pathlib import Path from uuid import UUID, uuid4 @@ -834,3 +835,144 @@ def test_a_node_heartbeat_works_without_a_csrf_token_but_an_admin_call_does_not( assert client.post(f"/api/compute/nodes?organization_id={new_uuid()}", json={"name": "x"}).json()[ "error" ]["code"] == "csrf_token_missing" + + +# -------------------------------------------------------------------------- +# The sweeper, which is the difference between a recovery mechanism and a +# diagnosis. It was documented as "called by the controller's own timer" and there +# was no timer, so recovery depended on a human noticing a stalled job and +# posting to an administrative route by hand. +# -------------------------------------------------------------------------- + + +def test_an_application_sweeps_expired_leases_with_nobody_asking(compute, organization): + """A job whose agent vanished comes back on its own, not on request. + + The mechanism under test is the loop in ``app._sweep_expired_leases``. Calling + ``sweep_expired_leases`` directly would prove nothing: it always worked when + called. What was missing was anything that called it, and a test of the + function alone would have walked straight past that - as the proof of concept + did, by sweeping by hand and recording the result as automatic. + """ + + import asyncio + + from umbratrans_web.app import LEASE_SWEEP_SECONDS + + class _App: + def __init__(self) -> None: + self.state = self + self.compute_repository = compute + + app = _App() + + async def _one_round() -> list: + """The loop's body, with its sleep removed so the test does not wait.""" + + await asyncio.sleep(0) + return compute.sweep_expired_leases() + + node = _online(compute, _node(compute, organization, "swept"), + [_device(_node(compute, organization, "s"))]) + compute.submit_job(organization, kind="sleep", requested_by="test", max_attempts=3) + claimed = compute.claim_job(node.id, organization, lease_seconds=1) + assert claimed is not None + + # Not yet expired: the loop must not rescue a lease that is still live, or it + # would duplicate every job that a healthy agent was working on. + assert asyncio.run(_one_round()) == [] + assert compute.get_job(claimed.job.id).status == RUNNING + + time.sleep(1.05) + swept = asyncio.run(_one_round()) + assert claimed.job.id in swept, "a lease nobody renewed did not come back" + recovered = compute.get_job(claimed.job.id) + assert recovered.status == QUEUED, "returned to the queue, not lost" + assert "stopped renewing" in (recovered.error or "") + + # An interval that is not shorter than a lease would let a job sit past its + # own expiry before anyone looked at it. + assert LEASE_SWEEP_SECONDS > 0 + + +def test_the_sweeper_survives_a_repository_that_raises_and_stays_cancellable(monkeypatch): + """One bad round must not stop recovery for ever. + + A sweeper that dies is worse than no sweeper, because its absence is silent: + the queue simply stops returning work and nothing in the logs says why. + """ + + import asyncio + + from umbratrans_web import app as app_module + from umbratrans_web.app import _sweep_expired_leases + + # The loop sleeps before its first round, so a test that only waits a moment + # never reaches the work at all. Shortening the interval is what makes this a + # test of the loop rather than of asyncio. + monkeypatch.setattr(app_module, "LEASE_SWEEP_SECONDS", 0.01) + + calls = {"count": 0} + + class _Flaky: + def sweep_expired_leases(self, **_): + calls["count"] += 1 + if calls["count"] == 1: + raise RuntimeError("database is away") + return [] + + class _State: + compute_repository = _Flaky() + + class _App: + def __init__(self) -> None: + self.state = _State() + + app = _App() + + async def _survive_and_stop() -> None: + task = asyncio.ensure_future(_sweep_expired_leases(app)) + # Long enough for the loop to sleep, raise, log, and start again. + await asyncio.sleep(0.05) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + # Cancel has to work, or shutting the application down hangs. + assert task.cancelled() or task.done() + + asyncio.run(_survive_and_stop()) + assert calls["count"] >= 2, ( + f"the loop made {calls['count']} attempt(s); a sweeper that stops on the first " + "failure is worse than none, because its absence is silent" + ) + + +def test_the_sweeper_is_a_no_op_before_the_compute_surface_is_used(): + """No database, no repository, and the loop must not invent one. + + Opening a connection pool here would mean an account-only deployment + acquired a compute dependency at boot. It should skip the round instead. + """ + + import asyncio + + from umbratrans_web.app import _sweep_expired_leases + + class _Bare: + pass + + class _App: + def __init__(self) -> None: + self.state = _Bare() + + app = _App() + + async def _stop() -> None: + task = asyncio.ensure_future(_sweep_expired_leases(app)) + await asyncio.sleep(0) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + asyncio.run(_stop()) + assert not hasattr(app.state, "compute_repository"), "built one nobody asked for" diff --git a/umbratrans_web/tests/test_node_agent.py b/umbratrans_web/tests/test_node_agent.py new file mode 100644 index 0000000..f386ce1 --- /dev/null +++ b/umbratrans_web/tests/test_node_agent.py @@ -0,0 +1,303 @@ +"""The node agent, driven as the process it actually is. + +Every test here starts a real ``node_agent.py`` subprocess and points it at a stub +controller over real HTTP. That is heavier than calling the agent's methods, and +the weight is the point: the two behaviours worth testing here - stopping without +a signal, and dying without releasing anything - are about *processes*. ``os._exit`` +cannot be observed in-process at all, and a stop file has no meaning without +something polling for it in a loop. A test that called ``NodeAgent.run`` in-process +would have had to mock both away and would have proved only that a mock was called. +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import Any + +import pytest + +AGENT = Path(__file__).resolve().parents[1] / "node_agent.py" +NODE_KEY = "test-key-never-sent-anywhere-real" + + +class _StubController: + """Just enough of ``/api/node`` to let an agent run. + + Counts what it was asked for, because that is how the tests tell a stopped + agent from a running one and a dead one from a working one. A test that + asserted on a log line would pass on a log line written by an agent that had + already stopped. + """ + + def __init__(self) -> None: + self.lock = threading.Lock() + self.heartbeats = 0 + self.claims = 0 + self.renews = 0 + self.completes = 0 + self.keys: set[str] = set() + self.offers_job = True + self.server: ThreadingHTTPServer | None = None + + @property + def url(self) -> str: + assert self.server is not None + return f"http://127.0.0.1:{self.server.server_address[1]}" + + def __enter__(self) -> "_StubController": + controller = self + + class Handler(BaseHTTPRequestHandler): + def log_message(self, *args: Any) -> None: # noqa: A003 - stdlib name + pass + + def _read(self) -> dict[str, Any]: + length = int(self.headers.get("Content-Length") or 0) + if not length: + return {} + return json.loads(self.rfile.read(length) or b"{}") + + def _send(self, payload: Any) -> None: + body = b"" if payload is None else json.dumps(payload).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_POST(self) -> None: # noqa: N802 - stdlib name + self._read() + authorization = self.headers.get("Authorization") or "" + with controller.lock: + controller.keys.add(authorization) + if self.path.startswith("/api/node/heartbeat"): + controller.heartbeats += 1 + self._send({"status": "online", "schedulable": True}) + elif self.path.startswith("/api/node/jobs/claim"): + controller.claims += 1 + if not controller.offers_job: + self._send(None) + return + self._send( + { + "job": { + "id": "job-1", + "kind": "sleep", + "payload": {"seconds": 0}, + }, + "placement": { + "node_name": "stub", + "device_kind": "cpu", + "reason": "test", + }, + "lease_expires_at": "2026-01-01T00:00:00+00:00", + } + ) + elif "/renew" in self.path: + controller.renews += 1 + self._send({"ok": True}) + elif self.path.endswith("/complete"): + controller.completes += 1 + self._send({"ok": True}) + else: + self._send({}) + + self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + threading.Thread(target=self.server.serve_forever, daemon=True).start() + return self + + def __exit__(self, *_: Any) -> None: + assert self.server is not None + self.server.shutdown() + self.server.server_close() + + def snapshot(self) -> tuple[int, int, int]: + with self.lock: + return self.heartbeats, self.claims, self.completes + + +def _write_config(path: Path, controller: _StubController, **extra: Any) -> Path: + path.write_text( + json.dumps( + { + "controller_url": controller.url, + "node_name": "stub-node", + "kinds": [], + "lease_seconds": 30, + "poll_interval_seconds": 0.05, + "node_key": "", + **extra, + } + ) + ) + return path + + +def _agent_environment(*, with_key: bool = True) -> dict[str, str]: + """A database-free environment, because the agent never had one. + + The key is here unless a test is specifically about how it arrives, since that + is the arrangement a node actually runs under. + """ + + environment = {k: v for k, v in os.environ.items() if not k.startswith("UMBRATRANS")} + if with_key: + environment["UMBRA_NODE_KEY"] = NODE_KEY + else: + environment.pop("UMBRA_NODE_KEY", None) + return environment + + +def _run_agent(config: Path, *args: str, timeout: float = 20.0, with_key: bool = True): + environment = _agent_environment(with_key=with_key) + return subprocess.run( + [sys.executable, str(AGENT), "--config", str(config), "--log-level", "INFO", *args], + capture_output=True, + text=True, + env=environment, + timeout=timeout, + ) + + +def test_a_stop_file_stops_an_agent_that_cannot_be_signalled(tmp_path: Path) -> None: + """The agent stops when a file appears, with no signal involved. + + A node agent normally runs as a service account, so its supervisor is not + allowed to signal it - kill(2) across accounts fails. Stopping it therefore has + to work without a signal, or every stop leaks a running agent. That is not + hypothetical: the proof of concept leaked one on every run, because the only + thing it could signal was the ``sudo`` wrapper it had itself spawned. + """ + + with _StubController() as controller: + stop_file = tmp_path / "node.stop" + config = _write_config(tmp_path / "node.json", controller, stop_file=str(stop_file)) + + started = subprocess.Popen( + [sys.executable, str(AGENT), "--config", str(config), "--log-level", "INFO"], + stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, + env=_agent_environment(), + ) + try: + deadline = time.time() + 10 + while time.time() < deadline and controller.snapshot()[0] == 0: + time.sleep(0.05) + before = controller.snapshot() + assert before[0] > 0, "the agent never reached the controller, so the stop " \ + "would not prove anything" + + stop_file.write_text("stop\n") + assert started.wait(timeout=15) == 0, "the agent did not stop cleanly" + + # And it really did stop: no further traffic at all after the file. + settled = controller.snapshot() + time.sleep(0.6) + assert controller.snapshot() == settled, "traffic continued after the stop file" + finally: + if started.poll() is None: + started.kill() + started.wait(timeout=5) + + +def test_fault_injection_leaves_a_job_unclaimed_by_anything_that_finished_it( + tmp_path: Path, +) -> None: + """The agent vanishes holding a lease, having done nothing. + + Exit code 137 and no completion call. The absence of the completion is the + assertion that matters: an agent that ran the job and then died would also + exit abruptly, and the lease would already be released by the completion, so + the recovery path would never see a stranded job at all. + """ + + with _StubController() as controller: + config = _write_config(tmp_path / "node.json", controller) + finished = _run_agent( + config, "--fault-exit-after-claim", "1", "--max-iterations", "400" + ) + + assert finished.returncode == 137, ( + f"expected the fault exit (137), got {finished.returncode}: " + f"{finished.stderr[-400:]}" + ) + assert "without running it" in finished.stderr, finished.stderr[-400:] + assert controller.claims >= 1, "it never claimed anything, so nothing was stranded" + assert controller.completes == 0, "the job was completed, so no lease was stranded" + + +def test_the_node_key_is_never_in_the_config_file_it_is_written_beside(tmp_path: Path) -> None: + """A config file holding a secret is a secret that ends up in a backup. + + The key arrives by environment or by stdin, never from the file, even when the + file is the only thing that could plausibly supply it. + """ + + with _StubController() as controller: + config = _write_config(tmp_path / "node.json", controller) + assert NODE_KEY not in config.read_text() + + by_env = _run_agent(config, "--max-iterations", "1") + assert by_env.returncode == 0, by_env.stderr[-400:] + + by_stdin = _run_agent_stdin(config) + assert by_stdin.returncode == 0, by_stdin.stderr[-400:] + + # The stub recorded the header the agent actually sent. + assert f"Bearer {NODE_KEY}" in controller.keys + + +def _run_agent_stdin(config: Path): + return subprocess.run( + [sys.executable, str(AGENT), "--config", str(config), "--log-level", "INFO", + "--node-key-stdin", "--max-iterations", "1"], + input=NODE_KEY + "\n", + capture_output=True, + text=True, + env=_agent_environment(), + timeout=20, + ) + + +def test_the_key_must_arrive_by_one_of_the_two_ways(tmp_path: Path) -> None: + """Neither route supplied means a clear message naming the sudo cause. + + The bare message this replaces said only "no node_key". Reaching it in + practice means sudo reset the environment, which is a fact about the host's + sudoers policy and not about the agent, so the agent says so. + """ + + with _StubController() as controller: + config = _write_config(tmp_path / "node.json", controller) + finished = _run_agent(config, "--max-iterations", "1", with_key=False) + assert finished.returncode != 0 + assert "UMBRA_NODE_KEY is not set" in finished.stderr, finished.stderr[-300:] + assert "sudo" in finished.stderr, ( + "the failure under sudo is the environment being reset; the message should " + f"say so, got: {finished.stderr[-300:]}" + ) + + +def test_an_empty_pipe_is_reported_as_an_empty_pipe(tmp_path: Path) -> None: + """A closed stdin with nothing written is not a key that happens to be blank.""" + + with _StubController() as controller: + config = _write_config(tmp_path / "node.json", controller) + finished = subprocess.run( + [sys.executable, str(AGENT), "--config", str(config), "--node-key-stdin", + "--max-iterations", "1"], + input="", capture_output=True, text=True, + env=_agent_environment(with_key=False), timeout=20, + ) + assert finished.returncode != 0 + assert "stdin was empty" in finished.stderr, finished.stderr[-300:] + + +if __name__ == "__main__": # pragma: no cover + sys.exit(pytest.main([__file__, "-v"])) \ No newline at end of file