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