From b0bf99a6eee381a208c3de4015fb511026602922 Mon Sep 17 00:00:00 2001 From: kstyagi Date: Fri, 25 Sep 2026 17:39:06 +0530 Subject: [PATCH] feat: Add guest execution agent and vsock support - Introduced a standalone guest execution agent to facilitate command execution within the guest VM over Firecracker's vsock. - Enhanced FirecrackerClient to support vsock configuration, validating guest CID and UDS path. - Updated configuration to include vsock settings by default. - Refactored FirecrackerProcessManager to manage runtime sockets more effectively. - Implemented guest agent installation and provisioning in the image store. - Added API endpoints for executing commands in the VM and installing the guest agent. - Improved error handling for guest execution failures and agent availability. --- DEVELOPER.md | 84 ++++++- README.md | 37 ++- USAGE.md | 158 ++++++++++++- tests/integration/test_guest_exec.py | 89 +++++++ tests/test_firecracker_config.py | 28 +++ tests/test_guest_agent_images.py | 199 ++++++++++++++++ tests/test_guest_exec.py | 289 ++++++++++++++++++++++ tests/test_guest_exec_lifecycle.py | 97 ++++++++ tests/test_install.py | 6 +- tests/test_lifecycle.py | 1 + tests/test_process.py | 26 ++ tests/test_server.py | 94 +++++++- uvm/api_models.py | 39 ++- uvm/config.py | 3 + uvm/errors.py | 12 + uvm/firecracker/api.py | 6 + uvm/firecracker/config.py | 3 + uvm/firecracker/process.py | 28 ++- uvm/guest_agent.py | 342 +++++++++++++++++++++++++++ uvm/guest_exec.py | 114 +++++++++ uvm/images.py | 165 +++++++++++++ uvm/install.py | 18 +- uvm/lifecycle.py | 42 ++++ uvm/routers/vms.py | 29 +++ uvm/server.py | 14 +- 25 files changed, 1893 insertions(+), 30 deletions(-) create mode 100644 tests/integration/test_guest_exec.py create mode 100644 tests/test_guest_agent_images.py create mode 100644 tests/test_guest_exec.py create mode 100644 tests/test_guest_exec_lifecycle.py create mode 100644 uvm/guest_agent.py create mode 100644 uvm/guest_exec.py diff --git a/DEVELOPER.md b/DEVELOPER.md index d40b05e..1b4dfa0 100644 --- a/DEVELOPER.md +++ b/DEVELOPER.md @@ -17,7 +17,7 @@ start/stop/restart, CPU/RAM/storage resizing, deletion, fixed-target bulk operat logs, doctor, and explicit port edits/sync. The UVM API and Nest dashboard expose health, installation, VM lifecycle controls, and managed port-policy controls, host diagnostics, bounded VM-log tails, and fixed-target bulk -operations. Interactive SSH remains CLI-only. +operations, and bounded guest command execution over vsock. Interactive SSH remains CLI-only. There is no autostart or background reconciler, multi-host control plane, tenancy/quotas, or UVM-provided Jailer/cgroup/namespace hardening. Private port @@ -89,6 +89,8 @@ Packet integration tests must remain inside disposable isolated namespaces. │ ├── install.py # Apt, Firecracker, kernel, rootfs installation │ ├── images.py # Template lookup, private disks, and guest credentials │ ├── known_hosts.py # Deletion-only cleanup of local SSH host-key caches +│ ├── guest_agent.py # Standalone guest daemon, job scopes, and bounded wire protocol +│ ├── guest_exec.py # Host-side guest sessions and execution results │ ├── network.py # IP/MAC/TAP/bridge/NAT/forward rules │ ├── firewall.py # Owned nft bridge tables and offline conntrack cleanup │ ├── lifecycle.py # Lifecycle, bulk operations, recovery, port intent @@ -238,7 +240,8 @@ Default runtime storage is `/var/lib/uvm`: │ ├── rootfs.ext4 │ ├── config.json │ ├── firecracker.log -│ └── firecracker.sock +│ ├── firecracker.sock +│ └── vsock.sock ├── state.json ├── state.lock ├── operations.lock @@ -518,8 +521,9 @@ The `firecracker/` package has three separate responsibilities. 2. `PUT /boot-source` 3. `PUT /drives/rootfs` 4. `PUT /network-interfaces/eth0` -5. `PUT /entropy` with `{}` (a host-backed virtio RNG device) -6. `PUT /actions` with `InstanceStart` +5. `PUT /vsock` with a guest CID and a per-VM Unix socket path +6. `PUT /entropy` with `{}` (a host-backed virtio RNG device) +7. `PUT /actions` with `InstanceStart` Every generated config includes `entropy`, including retained-disk boots. An empty object enables the device with default settings; do not skip it with @@ -607,7 +611,7 @@ fail locally with `Network is unreachable`. `ImageStore.configure_guest_network()` fills in the assigned address/prefix and gateway during creation and stopped-disk resume. It only specializes the exact -shipped profile from recognized UVM Jammy v1/v2/v3 images. Missing, customized, +shipped profile from recognized UVM Jammy v1/v2/v3/v4 images. Missing, customized, symlinked, hard-linked, or unexpectedly owned/mode-set profiles are left intact. An already-specialized profile is a no-op. Custom images retain responsibility for their own guest networking. @@ -617,7 +621,8 @@ For migration, the image layer sparse-copies the private disk and runs committed guest edits cannot be overwritten based on stale offline reads. The updated profile is read back and the filesystem checked before fsync and atomic disk replacement. Failed staging keeps the original disk available for -retry. The built template's recipe marker is v3; cached older templates are +retry. The built template's recipe marker is v4 (v3 introduced the route fix); +cached older templates are also handled by per-VM provisioning without requiring an installation refresh. ### Integrity Model @@ -894,6 +899,8 @@ points expose the remaining management routes below. | `DELETE` | `/vms/{identifier}` | `vms.py` | Remove runtime files/state; optional `timeout` query parameter | | `POST` | `/vms/{vm_id}/ports` | `vms.py` | Exact ID only; body has `mode` and `ports` | | `POST` | `/vms/{vm_id}/ports/sync` | `vms.py` | Exact ID only; reapply managed policy | +| `POST` | `/vms/{vm_id}/exec` | `vms.py` | Exact ID; `command`, strict `sudo`, bounded `timeout_seconds` | +| `POST` | `/vms/{vm_id}/agent/install` | `vms.py` | Exact ID; install/upgrade the agent on a stopped disk | | `GET` | `/vms/{identifier}/logs` | `vms.py` | Bounded non-following log tail | All management routes use `get_authorized_application()` and require: @@ -940,10 +947,62 @@ marks a response truncated when an oversized line or that bound prevents a full tail. Bulk bodies use an action plus either explicit identifiers or `all_vms`; the router resolves them before `run_many()` executes the fixed IDs under one operation lock. The API does not expose interactive SSH, raw Firecracker calls, -arbitrary command execution, or host path access. Public `/`, `/dashboard`, +arbitrary host command execution, or host path access. Public `/`, `/dashboard`, `/docs`, `/redoc`, and `/openapi.json` do not perform privileged management requests. +### Guest Execution + +`LifecycleService.execute()` resolves an exact registry ID, validates canonical +paths and an observed running VMM with no pending operation, and connects under +`operations.lock`. `GuestSession` checks the Unix peer PID against the selected +VMM, sends Firecracker's `CONNECT 1070\n`, consumes its `OK` acknowledgement, and +checks the agent's protocol version, VM ID, and configured username. The lock is +released before submission/waiting; the established socket pins the original +VMM and is never reconnected across a concurrent restart or address reuse. + +`guest_agent.py` is both a host-importable protocol module and a standalone +stdlib-only Python 3.10+ guest daemon. Keep it free of imports from the rest of +UVM. Version 1 uses length-prefixed JSON with bounds before allocation, absolute +socket deadlines, a 64 KiB UTF-8 command limit, and 1 MiB combined raw output. +The synchronous HTTP response exposes only exit code, stdout/stderr, timeout, +and truncation. Captured text uses UTF-8 replacement. Nonzero process exits are +HTTP 200 results; agent failures are 502, and failures before submission are 503. +Lost/malformed responses after submission carry `outcome_unknown: true`. There +is no command retry, durable job registry, streaming response, or reconnection. + +The root systemd agent accepts only host-CID connections and limits concurrent +connections/jobs to eight. Each request creates a delegated cgroup v2 job scope. +A fresh child interpreter joins the scope before initializing the chosen user's +supplementary groups, GID, UID, home, and environment, then execs `/bin/sh -c`. +`sudo` selects guest root directly; otherwise the recorded username is used. +Password authentication is not involved, and a recorded `root` login remains +root for both flag values. Input is closed and there is no PTY. + +Both output pipes are drained concurrently. The guest enforces the deadline, +signals the process group, escalates via `cgroup.kill`, and confirms cleanup. +The leader is observed with `waitid(WNOWAIT)` until signaling finishes, preventing +PGID reuse before reap. Normal exits clean up remaining background descendants +too. An unconfirmed cleanup is an unknown outcome. Stopping the agent service +also kills its job scopes; dropping an HTTP connection does not cancel a job. + +The v4 Ubuntu builder installs Python and the service. Creation provisions the +VM ID/username in root-only `/etc/uvm-agent.json` before marking the disk ready. +Older templates are left for an explicit stopped-disk `agent/install` request. +Provisioning and upgrades stage a sparse copy, replay/check ext4, validate guest +path ownership and recorded agent-file hashes, enable the service, verify writes, +and publish atomically. A root-only guest manifest records managed file hashes +so later upgrades preserve guest edits. An older guest missing Python installs +it through apt inside the guest on its next boot; no guest binaries execute on +the host. This bootstrap requires working guest egress until the runtime is ready. + +The guest needs systemd, Python 3.10+, built-in virtio-vsock, and cgroup v2 with +`cgroup.kill` (Linux 5.14+). Host socket paths are derived from Settings, so the +registry schema stays at 3. Each Firecracker instance has an independent vsock +device and uses CID 3. The process adapter retains both sockets while identity +is uncertain and removes them only after confirmed VMM exit. IPv4 port policies +do not govern this host-only management channel. + ## 14. Configuration Reference The environment-variable table is in [README.md](README.md#configuration); @@ -968,6 +1027,8 @@ Expected failures inherit from `UvmError`: | `ConflictError` | Ambiguous identifier, allocation conflict, or unsafe/pending operation | | `CommandError` | Required host command failure | | `FirecrackerError` | VMM process or Unix API failure | +| `GuestAgentUnavailableError` | Agent connection/readiness failure before submission (HTTP 503) | +| `GuestExecutionError` | Execution failure or unknown outcome after submission (HTTP 502) | | `TapCreationError` | TAP setup failed after interface creation | CLI code catches `UvmError` and prints `uvm: error: ...` without a traceback. @@ -1050,6 +1111,15 @@ addresses/gateways, custom subnets, idempotent migration, journal-replay rechecks, and retention of guest data, credentials and keys on failures. Lifecycle tests require confirmed process exit before network-profile repair. +Guest execution tests use real child processes with fake cgroup scopes, local +Unix sockets emulating Firecracker's vsock handshake, API validation fixtures, +and real temporary ext4 upgrades. They cover credential selection, concurrent +output draining, timeout cleanup, connection loss, VM targeting/locking, and +upgrade rollback. `tests/integration/test_guest_exec.py` is an opt-in smoke test +against a prepared disposable VM with a non-root login and all TCP ports closed; +see [USAGE.md](USAGE.md#18-verification). Unit tests do not validate the real +guest vsock device or cgroup kernel path. + ### Isolated Packet Tests The opt-in packet harness must run **only in disposable isolated user/network diff --git a/README.md b/README.md index 6f83eca..ce6e31c 100644 --- a/README.md +++ b/README.md @@ -339,6 +339,8 @@ The server exposes these routes: | `DELETE` | `/vms/{identifier}` | Delete a VM and release its resources; optional `timeout` query parameter | | `POST` | `/vms/{vm_id}/ports` | Set, open, or close private port rules for an exact VM ID | | `POST` | `/vms/{vm_id}/ports/sync` | Reapply a managed private port policy for an exact VM ID | +| `POST` | `/vms/{vm_id}/exec` | Execute a command/script inside a running VM over vsock, with optional root privileges | +| `POST` | `/vms/{vm_id}/agent/install` | Install/upgrade the execution agent on a stopped VM's private disk | | `GET` | `/vms/{identifier}/logs` | Read a bounded, non-following tail of a VM log | API port rules are objects such as `{"port": 80, "protocol": "HTTP"}`; @@ -357,6 +359,39 @@ Bulk requests use an action (`start`, `stop`, `restart`, `delete`, or for each fixed target and do not prompt for confirmation. Interactive SSH remains CLI-only. The API and dashboard are not background reconcilers. +### Execute Commands Without SSH + +Use an exact VM ID from create/list: + +```sh +curl -X POST \ + -H "X-UVM-Token: $UVM_API_TOKEN" \ + -H "Content-Type: application/json" \ + -d '{"command":"whoami","sudo":true,"timeout_seconds":60}' \ + "http://127.0.0.1:8000/vms//exec" +``` + +The response contains `exit_code`, `stdout`, `stderr`, `timed_out`, and +`truncated`. Commands and multiline scripts run through `/bin/sh` inside the +guest. `sudo: false` (the default) uses the VM's configured username; `true` +uses guest root without a password prompt. A VM configured with username `root` +still runs as root when `sudo` is false. + +Execution uses a guest systemd agent over Firecracker vsock. It works with SSH +disabled and with an empty TCP allowlist. Rebuild the default Ubuntu template +with `uvm install --force` using your configured checksums to include the agent +and Python in new VMs. Restart the management server to load these routes. + +For an existing VM, call **stop → `POST /vms/{vm_id}/agent/install` → start**. +Installation stages and verifies a private disk copy, preserving guest data and +credentials. If Python is missing, the installed bootstrap service acquires it +inside the guest on the next boot; this one-time step needs guest internet access. +The guest needs Python 3.10+, systemd with cgroup v2, and a vsock-capable Linux +5.14+ kernel; UVM's default Linux 6.1 kernel provides the required drivers. + +See [execution examples and limits](USAGE.md#execute-commands-without-ssh) for +programmatic calls, local script submission, timeout behavior, and upgrade steps. + ### API Security The server controls privileged host operations. `UVM_API_TOKEN` is required @@ -654,7 +689,7 @@ guest connectivity on an isolated Linux host. - No quotas or cgroup CPU, memory, I/O, or process limits are applied. - Firecracker is launched directly by this local CLI. Jailer/cgroup hardening is not implemented by the current command set. -- No VM autostart, background reconciliation, guest readiness probing, +- No VM autostart, background reconciliation, guest application health probing, snapshots, DHCP, public-port publishing, or multi-host scheduling. - The API uses one shared token, not users, roles, or tenants. UVM is not a hardened multi-tenant platform. diff --git a/USAGE.md b/USAGE.md index 507c4f6..cefc9d5 100644 --- a/USAGE.md +++ b/USAGE.md @@ -873,6 +873,8 @@ curl --cacert /path/to/ca.pem \ | `DELETE` | `/vms/{identifier}` | Delete VM and private disk; accepts `timeout` query parameter | | `POST` | `/vms/{vm_id}/ports` | Set, open, or close private ports for an exact VM ID | | `POST` | `/vms/{vm_id}/ports/sync` | Reapply the policy for an exact VM ID | +| `POST` | `/vms/{vm_id}/exec` | Execute a guest command/script over vsock, with optional root privileges | +| `POST` | `/vms/{vm_id}/agent/install` | Install/upgrade the agent on a stopped VM's disk | | `GET` | `/vms/{identifier}/logs` | Read a bounded, non-following log tail | Individual lifecycle requests do not expose a CLI dry-run flag and do not prompt @@ -1046,6 +1048,137 @@ the other policy operations. `open`, `close`, and `sync` reject a legacy policy. Mutating a live policy is subject to the same containment and confirmation rules as the CLI. +### Execute Commands Without SSH + +`POST /vms/{vm_id}/exec` accepts an **exact VM ID** and a JSON body: + +| Field | Contract | +|---|---| +| `command` | Required, non-empty command or multiline script; at most 65,536 UTF-8 bytes; no NUL characters | +| `sudo` | Strict boolean, default `false`; select guest root instead of the VM's configured login user | +| `timeout_seconds` | Finite JSON number greater than 0 and at most 3600; default 60 | + +```sh +export VM_ID='' +curl --fail-with-body -X POST \ + -H "X-UVM-Token: $UVM_API_TOKEN" \ + -H "Content-Type: application/json" \ + -d '{"command":"whoami","sudo":true,"timeout_seconds":60}' \ + "$UVM_API_URL/vms/$VM_ID/exec" +``` + +Example response: + +```json +{"exit_code":0,"stdout":"root\n","stderr":"","timed_out":false,"truncated":false} +``` + +The agent runs `/bin/sh -c COMMAND` in the selected user's home directory, with +that user's UID and groups and a fresh `HOME`/`USER`/`PATH` environment. Stdin is +closed and no TTY is allocated. The entire script, including pipes and +redirections, uses the selected identity. For a Bash script already in the VM, +use a command such as `bash /opt/jobs/process.sh`. The flag selects an execution +identity directly; it does not invoke an interactive `sudo` password prompt. +For VMs whose configured username is `root`, both flag values run as root. + +Submit a local script's contents using `jq` to handle JSON quoting: + +```sh +jq -n --rawfile command ./script.sh \ + '{command:$command,sudo:false,timeout_seconds:300}' | + curl --fail-with-body -X POST \ + -H "X-UVM-Token: $UVM_API_TOKEN" \ + -H "Content-Type: application/json" \ + --data-binary @- "$UVM_API_URL/vms/$VM_ID/exec" +``` + +Or call it from Python using the standard library: + +```python +import json +import os +from urllib.request import Request, urlopen + +request = Request( + f"{os.environ['UVM_API_URL']}/vms/{os.environ['VM_ID']}/exec", + data=json.dumps({"command": "id && uname -a", "sudo": False}).encode(), + headers={"X-UVM-Token": os.environ["UVM_API_TOKEN"], + "Content-Type": "application/json"}, + method="POST", +) +with urlopen(request, timeout=75) as response: + result = json.load(response) +print(result["stdout"]) +print(result["exit_code"]) +``` + +**Behavior and limits:** + +- The VM must be running with a known process identity and no pending lifecycle + operation. Guest readiness is checked separately by connecting to its agent. +- The call waits for completion. Nonzero command exits still return HTTP 200; + inspect `exit_code`. Negative codes indicate a terminating signal. +- The deadline is enforced inside the guest. On timeout the agent terminates + the job, including descendants in its cgroup, and returns partial output with + `timed_out: true` and `exit_code: null`. Connection and cleanup time are additional + to `timeout_seconds`; give the HTTP client about 15 extra seconds. +- The agent captures up to **1 MiB combined stdout/stderr**, then drains and + discards excess data. `truncated` reports this limit; invalid UTF-8 is replaced. +- Up to eight connections/jobs per VM are admitted. Excess connections fail + before submission. Execution does not hold the global lifecycle lock while + waiting, so another request can stop or restart the VM. +- A normal script exit also cleans up background descendants in the job's + cgroup. Use a guest service manager for workloads that should outlive the call. +- A lost connection after submission returns HTTP 502 with + `error.outcome_unknown: true`. The command may already have produced effects; + it is never automatically resubmitted. An HTTP client disconnect does not + itself cancel the guest job; the guest deadline still applies. +- Results are returned in the response, not stored as durable jobs. Commands + and their captured output are not copied to the VM registry or host traces. + +#### Prepare New or Existing VMs + +The default Ubuntu template recipe v4 includes Python 3 and `uvm-agent.service`. +Rebuild it with `uvm install --force` using the checksum settings from the +installation section, then create new VMs. Cached older templates and custom +images require the explicit agent installation below. Restart a running Nest or +UVM API server to load the new code before making these requests. + +For an existing VM, run these requests in order, checking each succeeds: + +```sh +curl --fail-with-body -X POST -H "X-UVM-Token: $UVM_API_TOKEN" \ + "$UVM_API_URL/vms/$VM_ID/stop" +curl --fail-with-body -X POST -H "X-UVM-Token: $UVM_API_TOKEN" \ + "$UVM_API_URL/vms/$VM_ID/agent/install" +curl --fail-with-body -X POST -H "X-UVM-Token: $UVM_API_TOKEN" \ + "$UVM_API_URL/vms/$VM_ID/start" +``` + +Installation returns `{"id":"...","status":"installed","restart_required":true}`. +It requires a cleanly stopped, proven-offline private disk and does not start the +VM itself. Updates are checked on a sparse copy before publication. Guest files, +credentials, and SSH keys are retained; customized agent files are preserved +rather than overwritten. Repeating installation also upgrades unedited managed +agent files using their recorded checksums. + +The guest image must provide systemd and apt (Ubuntu 22.04 or newer is the +supported baseline). If Python 3.10+ is missing, a bootstrap service installs +`python3` inside the VM on its next boot. This one-time install needs guest DNS, +internet access, and working apt repositories. The service retries failed startup; +execution returns 503 until it can connect. New v4 templates already include +the runtime and do not need this download. + +The guest kernel needs built-in `CONFIG_VSOCKETS=y`, `CONFIG_VIRTIO_VSOCKETS=y`, +and `CONFIG_CGROUPS=y`, plus cgroup v2 and `cgroup.kill` (Linux 5.14+). UVM's +default Linux 6.1 kernel supports these. Refresh an older installed kernel as +described in the installation/entropy recovery sections before booting. + +All transport is via a per-VM Unix socket and Firecracker vsock port 1070. This +management channel is independent of the IPv4 port policy, including TCP/22; +the guest agent accepts connections from the host CID only. VM execution does +not need SSH credentials, SSH host-key caches, or an open guest TCP port. + ### Diagnostics and Log Tails The dashboard's System view calls the protected diagnostics endpoint. It runs @@ -1095,6 +1228,8 @@ Use `{"force": true}` to refresh all downloaded artifacts. | `409` | Name/identifier conflict, incompatible retained state, or an unsafe/pending operation. | | `422` | Invalid JSON body, extra request fields, or invalid UVM input. | | `500` | Host command, Firecracker, state, or unexpected operational failure. | +| `502` | Guest execution failure or lost/invalid execution response; inspect `error.outcome_unknown`. | +| `503` | Guest agent unavailable, not ready, or at its connection limit before command submission. | UVM operation errors use this envelope: @@ -1103,6 +1238,8 @@ UVM operation errors use this envelope: ``` FastAPI request-schema errors use FastAPI's normal `detail` response format. +Guest execution errors additionally include a boolean `outcome_unknown` in the +`error` object; `true` means the caller must not assume the command did not run. ## 13. Configuration Scenarios @@ -1347,7 +1484,8 @@ Keep these constraints in mind when deciding whether UVM fits a scenario: DNS publication, guest reverse proxy/TLS setup, DHCP, load balancing, snapshots, multi-network tenancy, or multi-host scheduling. - The API uses one shared bearer token, not individual users, roles, or tenants. -- There is no guest readiness probe beyond Firecracker process state. +- Execution probes guest-agent readiness; VM list/status still describes the + Firecracker process rather than guest application health. - UVM is not a hardened multi-tenant platform. ## 17. Safe Operating Checklist @@ -1389,6 +1527,24 @@ upstreams to check NAT, offline startup, reconnects and uplink changes. It does not boot Firecracker or validate the host's real TAP setup. Run actual VM/SSH/egress checks on an isolated Linux host before relying on the deployment. +The execution smoke tests target an already-created **disposable VM on an +isolated host**, with a non-root login, an installed/running agent, and an empty +TCP allowlist. They check both privilege modes, output and exit status, bounded +output, and cgroup cleanup of a descendant that starts its own session. They do +not create, stop, or delete VMs: + +```sh +UVM_EXEC_INTEGRATION=1 UVM_EXEC_VM_ID='' \ + python3 -B -m unittest discover -s tests/integration -p test_guest_exec.py -v +``` + +Set `UVM_API_URL` and `UVM_API_TOKEN` as in the API examples. For TLS, the Python +client uses the system trust store (or `SSL_CERT_FILE`). Without the explicit +integration flag these tests skip and make no requests. The ordinary unit tests +exercise real child processes with fake cgroup scopes, local Unix-socket protocol +fixtures, and temporary ext4 upgrades; they do not prove the real vsock/cgroup +path inside a VM. + ## 19. Get Help ```sh diff --git a/tests/integration/test_guest_exec.py b/tests/integration/test_guest_exec.py new file mode 100644 index 0000000..72ee11b --- /dev/null +++ b/tests/integration/test_guest_exec.py @@ -0,0 +1,89 @@ +"""Opt-in checks against an already-provisioned disposable VM on an isolated host.""" + +from __future__ import annotations + +import json +import os +import unittest +from urllib.parse import quote +from urllib.request import Request, urlopen + + +@unittest.skipUnless(os.environ.get("UVM_EXEC_INTEGRATION") == "1", "set UVM_EXEC_INTEGRATION=1 on an isolated VM host") +class GuestExecIntegrationTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + cls.api_url = os.environ.get("UVM_API_URL", "http://127.0.0.1:8000").rstrip("/") + cls.token = os.environ["UVM_API_TOKEN"] + cls.vm_id = os.environ["UVM_EXEC_VM_ID"] + cls.path = f"/vms/{quote(cls.vm_id, safe='')}" + vm = cls.request("GET", cls.path) + if (vm["id"] != cls.vm_id or vm["observed_status"] != "running" + or vm["username"] == "root" or vm["ports"] != []): + raise RuntimeError("provide a running disposable VM with a non-root login, installed agent, and empty TCP allowlist") + cls.username = vm["username"] + + @classmethod + def request(cls, method, path, body=None): + request = Request( + cls.api_url + path, method=method, + data=json.dumps(body).encode("utf-8") if body is not None else None, + headers={"X-UVM-Token": cls.token, "Content-Type": "application/json"}, + ) + with urlopen(request, timeout=30) as response: + return json.load(response) + + def execute(self, command, *, sudo=False, timeout=10): + return self.request("POST", self.path + "/exec", { + "command": command, "sudo": sudo, "timeout_seconds": timeout, + }) + + def test_user_and_root_execution_work_with_all_tcp_ports_closed(self) -> None: + ordinary = self.execute("id -u; whoami") + self.assertEqual(ordinary["exit_code"], 0) + uid, username = ordinary["stdout"].splitlines() + self.assertNotEqual(uid, "0") + self.assertEqual(username, self.username) + elevated = self.execute("id -u", sudo=True) + self.assertEqual(elevated["exit_code"], 0) + self.assertEqual(elevated["stdout"], "0\n") + + def test_multiline_script_returns_output_and_nonzero_status(self) -> None: + result = self.execute("printf 'hello\\n'\nprintf 'problem\\n' >&2\nexit 7") + self.assertEqual(result, { + "exit_code": 7, "stdout": "hello\n", "stderr": "problem\n", "timed_out": False, "truncated": False, + }) + + def test_timeout_kills_a_descendant_that_started_a_new_session(self) -> None: + result = self.execute("""python3 - <<'PY' +import os, time +child = os.fork() +if child == 0: + os.setsid() +else: + print(child, flush=True) +time.sleep(30) +PY""", timeout=2) + self.assertTrue(result["timed_out"]) + self.assertIsNone(result["exit_code"]) + child = int(result["stdout"].strip()) + checked = self.execute(f"""python3 - <<'PY' +from pathlib import Path +try: + state = Path('/proc/{child}/stat').read_text().rsplit(') ', 1)[1].split()[0] +except FileNotFoundError: + state = 'gone' +assert state in ('gone', 'Z'), state +PY""") + self.assertEqual(checked["exit_code"], 0, checked["stderr"]) + + def test_output_is_bounded_while_both_pipes_are_drained(self) -> None: + result = self.execute("""python3 - <<'PY' +import os +for _ in range(32): + os.write(1, b'o' * 65536) + os.write(2, b'e' * 65536) +PY""") + self.assertEqual(result["exit_code"], 0) + self.assertTrue(result["truncated"]) + self.assertEqual(len(result["stdout"]) + len(result["stderr"]), 1024 * 1024) diff --git a/tests/test_firecracker_config.py b/tests/test_firecracker_config.py index c65521f..65704f0 100644 --- a/tests/test_firecracker_config.py +++ b/tests/test_firecracker_config.py @@ -78,6 +78,7 @@ class FirecrackerConfigTests(unittest.TestCase): self.assertEqual(config["drives"][0]["path_on_host"], "/vm/rootfs.ext4") self.assertIn("ip=10.42.0.2::10.42.0.1", config["boot-source"]["boot_args"]) self.assertNotIn("password", str(config)) + self.assertEqual(config["vsock"], {"guest_cid": 3, "uds_path": "/var/lib/uvm/vms/vm-test/vsock.sock"}) def test_configures_firecracker_in_required_order_before_start(self) -> None: vm = VmRecord( @@ -107,6 +108,7 @@ class FirecrackerConfigTests(unittest.TestCase): "/boot-source", "/drives/rootfs", "/network-interfaces/eth0", + "/vsock", "/entropy", "/actions", ], @@ -124,6 +126,7 @@ class FirecrackerConfigTests(unittest.TestCase): "drives": [{}], "network-interfaces": [{}], "entropy": {}, + "vsock": {"guest_cid": 3, "uds_path": "/tmp/vsock.sock"}, } client = RecordingFirecrackerClient() @@ -147,6 +150,31 @@ class FirecrackerConfigTests(unittest.TestCase): client.configure_and_start(config) self.assertEqual(client.calls, []) + def test_missing_invalid_or_failed_vsock_configuration_prevents_guest_boot(self) -> None: + config = { + "machine-config": {"vcpu_count": 1, "mem_size_mib": 512}, "boot-source": {}, + "drives": [{}], "network-interfaces": [{}], "entropy": {}, + } + for vsock in (None, {}, {"guest_cid": True, "uds_path": "/tmp/vsock.sock"}, + {"guest_cid": 2, "uds_path": "/tmp/vsock.sock"}, {"guest_cid": 3, "uds_path": ""}): + with self.subTest(vsock=vsock): + client = RecordingFirecrackerClient() + config["vsock"] = vsock + with self.assertRaisesRegex(FirecrackerError, "vsock"): + client.configure_and_start(config) + self.assertEqual(client.calls, []) + config["vsock"] = {"guest_cid": 3, "uds_path": "/tmp/vsock.sock"} + client = RecordingFirecrackerClient() + + def request(method, path, body): + client.calls.append((method, path, body)) + if path == "/vsock": + raise FirecrackerError("vsock unavailable") + + with patch.object(client, "_request", side_effect=request), self.assertRaises(FirecrackerError): + client.configure_and_start(config) + self.assertNotIn("/actions", [path for _method, path, _body in client.calls]) + def test_uses_the_configured_network_netmask_in_guest_boot_arguments(self) -> None: settings = Settings( network=IPv4Network("10.50.0.0/16"), diff --git a/tests/test_guest_agent_images.py b/tests/test_guest_agent_images.py new file mode 100644 index 0000000..714bf8f --- /dev/null +++ b/tests/test_guest_agent_images.py @@ -0,0 +1,199 @@ +from __future__ import annotations + +import hashlib +import json +import shutil +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from uvm.config import Settings +from uvm.errors import UvmError +from uvm.images import GUEST_AGENT_PREPARE, ImageStore, guest_agent_files +from uvm.system import CommandRunner + + +class GuestRuntimeBootstrapTests(unittest.TestCase): + def test_runtime_install_is_guest_only_skipped_when_ready_and_stops_on_apt_failure(self) -> None: + with tempfile.TemporaryDirectory() as temporary: + directory = Path(temporary) + python = directory / "python3" + apt = directory / "apt-get" + ready = directory / "ready" + python.write_text( + f"#!{sys.executable}\nimport os, sys\nfrom pathlib import Path\n" + "sys.exit(0 if Path(os.environ['READY']).exists() else 1)\n", encoding="utf-8", + ) + apt.write_text( + f"#!{sys.executable}\nimport json, os, sys\nfrom pathlib import Path\n" + "print(json.dumps(sys.argv[1:]))\n" + "if sys.argv[-1] == 'update' and os.environ['FAIL_UPDATE'] == '1': sys.exit(100)\n" + "if sys.argv[-1] == 'python3': Path(os.environ['READY']).touch()\n", encoding="utf-8", + ) + python.chmod(0o755) + apt.chmod(0o755) + script = GUEST_AGENT_PREPARE.replace("/usr/bin/python3", str(python)).replace("/usr/bin/apt-get", str(apt)) + for installed, fail in ((True, False), (False, False), (False, True)): + with self.subTest(installed=installed, fail=fail): + ready.unlink(missing_ok=True) + if installed: + ready.touch() + result = CommandRunner(emit=None).run(( + "env", f"READY={ready}", f"FAIL_UPDATE={int(fail)}", "/bin/sh", "-c", script, + ), check=False, capture=True) + calls = [json.loads(line) for line in result.stdout.splitlines()] + self.assertEqual(result.returncode, 100 if fail else 0) + self.assertEqual(len(calls), 0 if installed else 1 if fail else 2) + if calls: + self.assertEqual(calls[0][-1], "update") + self.assertIn("APT::Update::Error-Mode=any", calls[0]) + if not installed and not fail: + self.assertEqual(calls[-1][-2:], ["install", "python3"]) + + +@unittest.skipUnless(all(shutil.which(tool) for tool in ("mkfs.ext4", "debugfs", "e2fsck", "cp")), + "e2fsprogs and coreutils are required for offline guest agent checks") +class GuestAgentImageTests(unittest.TestCase): + def setUp(self) -> None: + temporary = tempfile.TemporaryDirectory() + self.addCleanup(temporary.cleanup) + self.base = Path(temporary.name) + self.settings = Settings(base=self.base / "uvm") + self.settings.images_dir.mkdir(parents=True) + self.disk = self.base / "VM with spaces" / "rootfs.ext4" + self.disk.parent.mkdir() + self.runner = CommandRunner(emit=None) + self.images = ImageStore(self.settings, self.runner) + self.retained = { + "/etc/shadow": "worker:$6$retained-hash:1:0:99999:7:::\n", + "/etc/ssh/ssh_host_ed25519_key": "retained private SSH key\n", + "/home/worker/data": "retained application data\n", + } + + def make_disk(self, marker="uvm-ubuntu-jammy-v3\n") -> None: + root = self.base / "tree" + payloads = { + **self.retained, "/etc/uvm-image": marker, + "/usr/lib/systemd/systemd": "systemd fixture", "/usr/bin/apt-get": "apt fixture", + } + for path, content in payloads.items(): + target = root / path.lstrip("/") + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(content, encoding="utf-8") + target.chmod(0o644) + for directory in (root, *(path for path in root.rglob("*") if path.is_dir())): + directory.chmod(0o755) + with self.settings.rootfs_image.open("wb") as image: + image.truncate(32 * 1024 * 1024) + self.runner.run(("mkfs.ext4", "-q", "-F", "-b", "4096", "-d", root, self.settings.rootfs_image), capture=True) + # mkfs -d retains the creating user's ownership. Model the root-built + # guest while allowing this test itself to run without root. + paths = {"/"} + for path in payloads: + paths.add(path) + paths.update(str(parent) for parent in Path(path).parents) + commands = "\n".join(f"set_inode_field {path} {field} 0" for path in paths for field in ("uid", "gid")) + self.runner.run(("debugfs", "-w", self.settings.rootfs_image), input_text=commands + "\n", capture=True) + self.runner.run(("cp", "--sparse=always", self.settings.rootfs_image, self.disk), capture=True) + self.disk.chmod(0o600) + + def guest(self, command: str, *, disk: Path | None = None) -> str: + return self.runner.run(("debugfs", "-R", command, disk or self.disk), capture=True).stdout + + def snapshot(self) -> tuple[int, bytes]: + return self.disk.stat().st_ino, hashlib.sha256(self.disk.read_bytes()).digest() + + def replace_guest_file(self, path: str, contents: str, *, disk: Path | None = None) -> None: + source = self.base / "edited" + source.write_text(contents, encoding="utf-8") + source.chmod(0o644) + self.runner.run(("debugfs", "-w", disk or self.disk), input_text=( + f"rm {path}\nwrite {source} {path}\nset_inode_field {path} uid 0\nset_inode_field {path} gid 0\n" + ), capture=True) + + def test_existing_guest_is_upgraded_atomically_with_private_identity_and_retained_data(self) -> None: + self.make_disk() + template = hashlib.sha256(self.settings.rootfs_image.read_bytes()).digest() + self.assertTrue(self.images.provision_guest_agent(self.disk, "vm-test", "worker", install=True)) + self.assertEqual(json.loads(self.guest("cat /etc/uvm-agent.json")), {"vm_id": "vm-test", "username": "worker"}) + self.assertRegex(self.guest("stat /etc/uvm-agent.json"), r"Mode:\s+0?600") + self.assertRegex(self.guest("stat /etc/uvm-agent.json"), r"User:\s+0\s+Group:\s+0") + for path, content in guest_agent_files().items(): + self.assertEqual(self.guest(f"cat {path}"), content.decode("utf-8")) + self.assertIn('Fast link dest: "/etc/systemd/system/uvm-agent.service"', + self.guest("stat /etc/systemd/system/multi-user.target.wants/uvm-agent.service")) + self.assertEqual(self.guest("cat /etc/uvm-image"), "uvm-ubuntu-jammy-v3\n") + for path, content in self.retained.items(): + self.assertEqual(self.guest(f"cat {path}"), content) + self.assertEqual(hashlib.sha256(self.settings.rootfs_image.read_bytes()).digest(), template) + self.assertEqual(self.disk.stat().st_mode & 0o777, 0o600) + self.runner.run(("e2fsck", "-f", "-n", self.disk), capture=True) + self.assertEqual(list(self.disk.parent.iterdir()), [self.disk]) + self.images.provision_guest_agent(self.disk, "vm-test", "worker", install=True) + self.assertEqual(self.guest("cat /home/worker/data"), self.retained["/home/worker/data"]) + + def test_current_templates_are_configured_and_older_templates_require_explicit_install(self) -> None: + self.make_disk() + before = self.snapshot() + self.assertFalse(self.images.provision_guest_agent(self.disk, "vm-test", "worker")) + self.assertEqual(self.snapshot(), before) + self.replace_guest_file("/etc/uvm-image", "uvm-ubuntu-jammy-v4\n") + self.assertTrue(self.images.provision_guest_agent(self.disk, "vm-test", "worker")) + + def test_agent_updates_use_recorded_hashes_but_preserve_guest_customizations(self) -> None: + self.make_disk() + self.images.provision_guest_agent(self.disk, "vm-test", "worker", install=True) + updated = guest_agent_files() + updated["/usr/lib/uvm/agent.py"] += b"\n# next packaged revision\n" + with patch("uvm.images.guest_agent_files", return_value=updated): + self.images.provision_guest_agent(self.disk, "vm-test", "worker", install=True) + self.assertTrue(self.guest("cat /usr/lib/uvm/agent.py").endswith("# next packaged revision\n")) + self.replace_guest_file("/usr/lib/uvm/agent.py", "guest customization\n") + before = self.snapshot() + with self.assertRaisesRegex(UvmError, "customized guest agent file"): + self.images.provision_guest_agent(self.disk, "vm-test", "worker", install=True) + self.assertEqual(self.snapshot(), before) + self.assertEqual(list(self.disk.parent.iterdir()), [self.disk]) + + def test_failure_during_provisioning_keeps_the_original_disk(self) -> None: + self.make_disk() + before = self.snapshot() + write = self.images._write_guest_file + + def fail(candidate, path, source): + self.assertNotEqual(candidate, self.disk) + write(candidate, path, source) + raise UvmError("simulated write failure") + + with patch.object(self.images, "_write_guest_file", side_effect=fail), self.assertRaisesRegex(UvmError, "write failure"): + self.images.provision_guest_agent(self.disk, "vm-test", "worker", install=True) + self.assertEqual(self.snapshot(), before) + self.assertEqual(list(self.disk.parent.iterdir()), [self.disk]) + + def test_guest_edits_revealed_by_journal_replay_are_checked_before_replacement(self) -> None: + self.make_disk() + self.images.provision_guest_agent(self.disk, "vm-test", "worker", install=True) + before = self.snapshot() + run = self.runner.run + + def replay(command, **options): + result = run(command, **options) + if command[:3] == ("e2fsck", "-f", "-p"): + self.replace_guest_file("/usr/lib/uvm/agent.py", "journal-committed guest edit\n", disk=command[-1]) + return result + + with patch.object(self.runner, "run", side_effect=replay), self.assertRaisesRegex(UvmError, "customized"): + self.images.provision_guest_agent(self.disk, "vm-test", "worker", install=True) + self.assertEqual(self.snapshot(), before) + + def test_unsafe_guest_directory_and_shared_disk_are_rejected(self) -> None: + self.make_disk() + with self.assertRaisesRegex(UvmError, "unsafe"): + self.images.provision_guest_agent(self.settings.rootfs_image, "vm-test", "worker", install=True) + self.runner.run(("debugfs", "-w", "-R", "set_inode_field /etc mode 040777", self.disk), capture=True) + before = self.snapshot() + with self.assertRaisesRegex(UvmError, "unsafe guest agent directory"): + self.images.provision_guest_agent(self.disk, "vm-test", "worker", install=True) + self.assertEqual(self.snapshot(), before) diff --git a/tests/test_guest_exec.py b/tests/test_guest_exec.py new file mode 100644 index 0000000..0a8bc14 --- /dev/null +++ b/tests/test_guest_exec.py @@ -0,0 +1,289 @@ +from __future__ import annotations + +import os +import pwd +import shlex +import socket +import struct +import sys +import tempfile +import threading +import time +import unittest +from contextlib import contextmanager +from pathlib import Path +from unittest.mock import Mock, patch + +from uvm.errors import GuestAgentUnavailableError, GuestExecutionError, ValidationError +from uvm import guest_agent as agent +from uvm.guest_exec import GuestSession, validate_exec + + +class FakeScope: + """Use real subprocesses/process groups without touching the host's cgroups.""" + + def __init__(self) -> None: + self.temporary = tempfile.TemporaryDirectory() + self.path = Path(self.temporary.name) + (self.path / "cgroup.procs").touch() + self.kills = 0 + self.joined_pid = None + + def kill(self) -> None: + self.kills += 1 + self.joined_pid = (self.path / "cgroup.procs").read_text() + + def close(self) -> None: + self.temporary.cleanup() + + +class GuestCommandTests(unittest.TestCase): + def setUp(self) -> None: + self.account = pwd.getpwuid(os.geteuid()) + self.scope = FakeScope() + self.addCleanup(self.scope.temporary.cleanup) + + def run_command(self, command, timeout=5): + return agent.run_command(command, self.account.pw_name, timeout, scope_factory=lambda: self.scope) + + def test_multiline_script_has_separate_output_exit_status_and_no_stdin(self) -> None: + result = self.run_command("printf 'hello\\n'\nprintf 'problem\\n' >&2\nread input || exit 7") + self.assertEqual(result, { + "exit_code": 7, "stdout": "hello\n", "stderr": "problem\n", "timed_out": False, "truncated": False, + }) + self.assertTrue(self.scope.joined_pid.isdigit()) + self.assertGreater(self.scope.kills, 0) + + def test_command_enters_scope_before_running_and_uses_account_home_and_environment(self) -> None: + with patch.dict(os.environ, {"UVM_API_TOKEN": "must-not-reach-guest-command"}): + result = self.run_command('printf "%s\\n" "$$" "$HOME" "$USER" "$UVM_API_TOKEN"; pwd; id -u') + self.assertEqual(result["exit_code"], 0) + self.assertEqual(result["stdout"].splitlines(), [ + self.scope.joined_pid, self.account.pw_dir, self.account.pw_name, "", self.account.pw_dir, str(os.geteuid()), + ]) + + def test_timeout_kills_the_shell_and_children_that_ignore_sigterm(self) -> None: + start = time.monotonic() + result = self.run_command("trap '' TERM; sleep 30 & printf '%s\\n' \"$!\"; wait", timeout=0.25) + self.assertTrue(result["timed_out"]) + self.assertIsNone(result["exit_code"]) + self.assertLess(time.monotonic() - start, 4) + child = int(result["stdout"]) + try: + state = Path(f"/proc/{child}/stat").read_text().split(") ", 1)[1].split()[0] + except FileNotFoundError: + state = "gone" + self.assertIn(state, {"gone", "Z"}) + + def test_large_stdout_and_stderr_are_drained_without_unbounded_capture(self) -> None: + script = "import os; [(os.write(1, b'o'*65536), os.write(2, b'e'*65536)) for _ in range(32)]" + result = self.run_command(f"{shlex.quote(sys.executable)} -c {shlex.quote(script)}") + self.assertEqual(result["exit_code"], 0) + self.assertEqual(len(result["stdout"]) + len(result["stderr"]), agent.MAX_OUTPUT_BYTES) + self.assertTrue(result["truncated"]) + self.assertFalse(result["timed_out"]) + + def test_non_utf8_output_is_replaced_without_losing_the_exit_code(self) -> None: + result = self.run_command("printf '\\377\\000'; exit 255") + self.assertEqual(result["stdout"], "\ufffd\0") + self.assertEqual(result["exit_code"], 255) + + def test_cgroup_cleanup_failure_is_an_unknown_outcome_and_still_reaps_the_process(self) -> None: + with patch.object(self.scope, "kill", side_effect=OSError("cannot kill scope")): + with self.assertRaises(agent.ExecutionFailure) as raised: + self.run_command("exit 0") + self.assertTrue(raised.exception.outcome_unknown) + + def test_privilege_drop_initializes_groups_before_uid_and_exec(self) -> None: + account = Mock(pw_name="worker", pw_uid=1001, pw_gid=1002, pw_dir="/home/worker") + events = [] + with ( + patch.object(agent.pwd, "getpwnam", return_value=account), + patch.object(agent.os, "geteuid", return_value=0), + patch.object(agent.os, "initgroups", side_effect=lambda *args: events.append(("groups", args))), + patch.object(agent.os, "setgid", side_effect=lambda value: events.append(("gid", value))), + patch.object(agent.os, "setuid", side_effect=lambda value: events.append(("uid", value))), + patch.object(agent.os, "chdir") as chdir, + patch.object(agent.os, "umask"), + patch.object(agent.os, "execve") as execute, + ): + agent.child_main(self.scope.path, "worker", "printf x > file") + self.assertEqual(events, [("groups", ("worker", 1002)), ("gid", 1002), ("uid", 1001)]) + chdir.assert_called_once_with("/home/worker") + self.assertEqual(execute.call_args.args[:2], ("/bin/sh", ["sh", "-c", "printf x > file"])) + self.assertEqual(execute.call_args.args[2]["HOME"], "/home/worker") + + +class GuestProtocolTests(unittest.TestCase): + def test_validates_byte_limits_types_finite_deadlines_and_text_before_transport(self) -> None: + for command, sudo, timeout in ( + ("", False, 1), (" \n", False, 1), ("a\0b", False, 1), ("\ud800", False, 1), + ("x" * (agent.MAX_COMMAND_BYTES + 1), False, 1), + ("\u00e9" * agent.MAX_COMMAND_BYTES, False, 1), + ("id", "false", 1), ("id", 0, 1), ("id", False, True), + ("id", False, 0), ("id", False, -1), ("id", False, float("nan")), + ("id", False, float("inf")), ("id", False, 3601), ("id", False, "60"), + ): + with self.subTest(command=command[:20], sudo=sudo, timeout=timeout), self.assertRaises(ValidationError): + validate_exec(command, sudo, timeout) + validate_exec("id", False, 0.1) + + def test_oversized_frame_is_rejected_before_reading_a_body(self) -> None: + left, right = socket.socketpair() + with left, right: + left.sendall(struct.pack("!I", agent.MAX_REQUEST_BYTES + 1)) + with self.assertRaises(agent.ProtocolError): + agent.receive_frame(right, agent.MAX_REQUEST_BYTES, time.monotonic() + 1) + + def test_read_deadline_applies_to_the_whole_frame(self) -> None: + left, right = socket.socketpair() + with left, right: + left.sendall(struct.pack("!I", 5) + b"{") + start = time.monotonic() + with self.assertRaises(TimeoutError): + agent.receive_frame(right, 1024, time.monotonic() + 0.05) + self.assertLess(time.monotonic() - start, 0.5) + + def test_agent_maps_sudo_to_root_and_rejects_bad_requests_without_launching(self) -> None: + config = {"vm_id": "vm-test", "username": "worker"} + for sudo in (False, True, "true"): + with self.subTest(sudo=sudo): + left, right = socket.socketpair() + result = {"exit_code": 0, "stdout": "", "stderr": "", "timed_out": False, "truncated": False} + runner = Mock(return_value=result) + thread = threading.Thread(target=agent.serve_connection, args=(right, config), kwargs={"runner": runner}) + thread.start() + with left: + deadline = time.monotonic() + 2 + self.assertEqual(agent.receive_frame(left, 4096, deadline), {"version": 1, **config}) + agent.send_frame(left, {"version": 1, "command": "id", "sudo": sudo, "timeout_seconds": 2}, 4096, deadline) + reply = agent.receive_frame(left, 4096, deadline) + thread.join(2) + self.assertFalse(thread.is_alive()) + if type(sudo) is bool: + self.assertEqual(reply, result) + runner.assert_called_once_with("id", "root" if sudo else "worker", 2) + else: + self.assertFalse(reply["outcome_unknown"]) + runner.assert_not_called() + + def test_guest_local_connections_are_rejected_before_a_worker_is_started(self) -> None: + connection = Mock() + listener = Mock() + listener.__enter__ = Mock(return_value=listener) + listener.__exit__ = Mock(return_value=False) + listener.accept.side_effect = [(connection, (3, 9999)), KeyboardInterrupt()] + with ( + patch.object(agent.sys, "argv", ["agent.py"]), patch.object(agent.os, "geteuid", return_value=0), + patch.object(agent, "load_config", return_value={"vm_id": "vm-test", "username": "root"}), + patch.object(agent.socket, "socket", return_value=listener), patch.object(agent.threading, "Thread") as worker, + self.assertRaises(KeyboardInterrupt), + ): + agent.main() + connection.close.assert_called_once() + worker.assert_not_called() + + +class GuestTransportTests(unittest.TestCase): + @contextmanager + def firecracker(self, handler): + with tempfile.TemporaryDirectory() as temporary: + path = Path(temporary) / "vsock.sock" + errors = [] + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as listener: + listener.bind(str(path)) + listener.listen(1) + listener.settimeout(3) + + def serve(): + try: + connection, _peer = listener.accept() + with connection: + handler(connection) + except (BrokenPipeError, EOFError, ConnectionResetError): + pass + except BaseException as error: + errors.append(error) + + thread = threading.Thread(target=serve, daemon=True) + thread.start() + try: + yield path + finally: + thread.join(4) + self.assertFalse(thread.is_alive()) + if errors: + raise errors[0] + + def handshake(self, connection, **changes): + deadline = time.monotonic() + 2 + self.assertEqual(agent.receive_exact(connection, len(b"CONNECT 1070\n"), deadline), b"CONNECT 1070\n") + # The acknowledgement and JSON frame can be coalesced on the wire. + connection.sendall(b"OK 1234\n") + agent.send_frame(connection, {"version": 1, "vm_id": "vm-test", "username": "worker", **changes}, 4096, deadline) + + def test_round_trip_preserves_command_text_and_a_nonzero_exit(self) -> None: + command = "printf '%s\\n' 'quotes; $(literal)'\nexit 7" + expected = {"exit_code": 7, "stdout": "quotes; $(literal)\n", "stderr": "", "timed_out": False, "truncated": False} + + def handle(connection): + self.handshake(connection) + request = agent.receive_frame(connection, agent.MAX_REQUEST_BYTES, time.monotonic() + 2) + self.assertEqual(request, {"version": 1, "command": command, "sudo": True, "timeout_seconds": 3}) + agent.send_frame(connection, expected, agent.MAX_RESPONSE_BYTES, time.monotonic() + 2) + + with self.firecracker(handle) as path: + session = GuestSession.connect(path, "vm-test", "worker", os.getpid()) + try: + result = session.execute(command, sudo=True, timeout_seconds=3) + finally: + session.close() + self.assertEqual(result.exit_code, 7) + self.assertEqual(result.stdout, expected["stdout"]) + + def test_wrong_vm_or_process_identity_prevents_command_submission(self) -> None: + for changes, pid in (({"vm_id": "vm-other"}, os.getpid()), ({"version": 2}, os.getpid()), + ({"username": "root"}, os.getpid()), ({}, os.getpid() + 100000)): + with self.subTest(changes=changes, pid=pid): + def handle(connection): + self.handshake(connection, **changes) + self.assertEqual(connection.recv(1), b"") + + with self.firecracker(handle) as path, self.assertRaises(GuestAgentUnavailableError): + GuestSession.connect(path, "vm-test", "worker", pid) + + def test_connection_loss_after_submission_has_unknown_outcome_without_retry(self) -> None: + submissions = [] + + def handle(connection): + self.handshake(connection) + submissions.append(agent.receive_frame(connection, agent.MAX_REQUEST_BYTES, time.monotonic() + 2)) + + with self.firecracker(handle) as path: + session = GuestSession.connect(path, "vm-test", "worker", os.getpid()) + with self.assertRaises(GuestExecutionError) as raised: + try: + session.execute("touch example", sudo=False, timeout_seconds=1) + finally: + session.close() + self.assertTrue(raised.exception.outcome_unknown) + self.assertEqual(len(submissions), 1) + + def test_malformed_result_is_not_reported_as_success(self) -> None: + for result in ({"exit_code": True, "stdout": "", "stderr": "", "timed_out": False, "truncated": False}, + {"error": "bad", "outcome_unknown": "false"}, {"error": "\ud800", "outcome_unknown": False}, + {"exit_code": 0}): + with self.subTest(result=result): + def handle(connection): + self.handshake(connection) + agent.receive_frame(connection, agent.MAX_REQUEST_BYTES, time.monotonic() + 2) + agent.send_frame(connection, result, agent.MAX_RESPONSE_BYTES, time.monotonic() + 2) + + with self.firecracker(handle) as path: + session = GuestSession.connect(path, "vm-test", "worker", os.getpid()) + with self.assertRaises(GuestExecutionError) as raised: + try: + session.execute("id", sudo=False, timeout_seconds=1) + finally: + session.close() + self.assertTrue(raised.exception.outcome_unknown) diff --git a/tests/test_guest_exec_lifecycle.py b/tests/test_guest_exec_lifecycle.py new file mode 100644 index 0000000..4401244 --- /dev/null +++ b/tests/test_guest_exec_lifecycle.py @@ -0,0 +1,97 @@ +from __future__ import annotations + +from pathlib import Path +from unittest.mock import Mock + +from tests.test_lifecycle import LifecycleTestCase +from uvm.errors import ConflictError, GuestAgentUnavailableError, GuestExecutionError, NotFoundError +from uvm.guest_exec import ExecResult + + +class GuestExecLifecycleTests(LifecycleTestCase): + def test_create_provisions_agent_before_the_disk_is_ready_or_the_guest_is_started(self) -> None: + def provision(disk, vm_id, username): + self.host.call("images.provision_agent", vm_id) + vm = self.store.get(vm_id) + self.assertFalse(vm.disk_ready) + self.assertEqual(self.process.observe(vm), "gone") + self.assertEqual(disk, Path(vm.disk)) + self.assertEqual(username, "worker") + + self.provision_agent.side_effect = provision + vm = self.create_vm(username="worker") + self.assert_call_order(vm, "images.provision_agent", "process.start") + + def test_execution_pins_the_exact_vm_under_lock_and_waits_without_holding_it(self) -> None: + vm = self.create_vm(name="worker", ports=()) + result = ExecResult(7, "output", "problem", False, False) + session = Mock() + + def connect(path, vm_id, username, pid): + self.assertTrue(self.store.operation_active) + self.assertEqual((path, vm_id, username, pid), (self.settings.guest_socket_path(vm.id), vm.id, vm.username, vm.pid)) + return session + + def execute(command, **options): + self.assertFalse(self.store.operation_active) + self.assertFalse(self.store.transaction_active) + self.service.stop(vm.id) + return result + + self.service._guest_connector = Mock(side_effect=connect) + session.execute.side_effect = execute + self.assertEqual(self.service.execute(vm.id, "exit 7", sudo=True, timeout_seconds=2), result) + session.execute.assert_called_once_with("exit 7", sudo=True, timeout_seconds=2) + session.close.assert_called_once() + + def test_execution_rejects_aliases_stopped_unknown_and_pending_vms_before_connecting(self) -> None: + vm = self.create_vm(name="worker") + self.service._guest_connector = Mock() + for alias in (vm.name, vm.guest_ip): + with self.assertRaises(NotFoundError): + self.service.execute(alias, "id") + for changes in ({"status": "stopped"}, {"operation": "policy"}, {"operation": "delete"}): + self.persist(vm, **changes) + with self.assertRaises(ConflictError): + self.service.execute(vm.id, "id") + self.persist(vm, status="running", operation=None) + self.process.unknown.add(vm.id) + with self.assertRaises(ConflictError): + self.service.execute(vm.id, "id") + self.service._guest_connector.assert_not_called() + + def test_unavailable_or_lost_agent_does_not_retry_or_mutate_vm_state(self) -> None: + vm = self.create_vm() + before = self.settings.state_path.read_bytes() + self.service._guest_connector = Mock(side_effect=GuestAgentUnavailableError("not ready")) + with self.assertRaises(GuestAgentUnavailableError): + self.service.execute(vm.id, "id") + self.service._guest_connector.assert_called_once() + session = Mock() + session.execute.side_effect = GuestExecutionError("connection lost", outcome_unknown=True) + self.service._guest_connector = Mock(return_value=session) + with self.assertRaises(GuestExecutionError): + self.service.execute(vm.id, "id") + session.execute.assert_called_once() + session.close.assert_called_once() + self.assertEqual(self.settings.state_path.read_bytes(), before) + + def test_agent_upgrade_requires_a_stopped_proven_offline_disk_and_retains_identity(self) -> None: + vm = self.create_vm(username="worker") + self.provision_agent.reset_mock() + with self.assertRaises(ConflictError): + self.service.install_guest_agent(vm.id) + self.provision_agent.assert_not_called() + stopped = self.service.stop(vm.id) + before = stopped.to_dict() + + def provision(disk, vm_id, username, *, install): + self.assertTrue(self.store.operation_active) + self.assertEqual(self.process.observe(self.store.get(vm_id)), "gone") + self.assertEqual((disk, username, install), (Path(vm.disk), "worker", True)) + + self.provision_agent.side_effect = provision + self.assertEqual(self.service.install_guest_agent(vm.id).to_dict(), before) + self.provision_agent.assert_called_once() + self.assertEqual(self.store.get(vm.id).to_dict(), before) + self.assertEqual(len(self.process.started), 1) diff --git a/tests/test_install.py b/tests/test_install.py index 92cfe78..73b7047 100644 --- a/tests/test_install.py +++ b/tests/test_install.py @@ -216,13 +216,14 @@ class UbuntuBuildRunner: return CommandResult(args, 100, stdout="Reading package lists...\n", stderr="E: unmet guest dependencies\n") status = "\n\n".join( f"Package: {package}\nStatus: install ok installed\nVersion: 1.0" - for package in ("apt", "sudo", "openssh-server", "systemd-sysv") + for package in ("apt", "sudo", "openssh-server", "systemd-sysv", "python3") ) + "\n" files = { "etc/sudoers": "root ALL=(ALL:ALL) ALL\n", "etc/ssh/sshd_config": "UsePAM yes\n", "etc/ssh/ssh_host_rsa_key": "template host key", "usr/bin/sudo": "sudo binary", + "usr/bin/python3": "python binary", "usr/sbin/sshd": "sshd binary", "usr/lib/systemd/systemd": "init binary", "var/lib/dpkg/status": status, "var/lib/dpkg/info/sudo.list": "/usr/bin/sudo\n", @@ -319,6 +320,7 @@ class UbuntuRootfsBuildTests(unittest.TestCase): self.assertLess(runner.calls.index(bootstrap), runner.calls.index(packages)) self.assertIn("sudo", packages) self.assertIn("openssh-server", packages) + self.assertIn("python3", packages) self.assertIn("--mount-proc", packages) self.assertIn("--kill-child", packages) self.assertEqual(settings.rootfs_image.stat().st_mode & 0o777, 0o600) @@ -335,6 +337,8 @@ class UbuntuRootfsBuildTests(unittest.TestCase): self.assertEqual(guest("cat /var/lib/dpkg/info/sudo.list").stdout, "/usr/bin/sudo\n") self.assertRegex(guest("stat /usr/bin/sudo").stdout, r"Mode:\s+0?4755") self.assertEqual(guest("cat /etc/uvm-image").stdout, _ROOTFS_RECIPE) + self.assertIn("socket.AF_VSOCK", guest("cat /usr/lib/uvm/agent.py").stdout) + self.assertIn("Delegate=yes", guest("cat /etc/systemd/system/uvm-agent.service").stdout) self.assertIn("nameserver 1.1.1.1", guest("cat /etc/resolv.conf").stdout) self.assertIn("KeepConfiguration=yes", guest("cat /etc/systemd/network/10-uvm.network").stdout) self.assertIn("jammy-security main universe", guest("cat /etc/apt/sources.list").stdout) diff --git a/tests/test_lifecycle.py b/tests/test_lifecycle.py index c882ca4..45d0b8b 100644 --- a/tests/test_lifecycle.py +++ b/tests/test_lifecycle.py @@ -310,6 +310,7 @@ class LifecycleTestCase(unittest.TestCase): self.configure_network = self.enterContext(patch.object( self.images, "configure_guest_network", side_effect=self.configure_guest_network, )) + self.provision_agent = self.enterContext(patch.object(self.images, "provision_guest_agent")) self.copy_disk = self.enterContext(patch.object( self.images, "create_vm_disk", wraps=self.images.create_vm_disk, )) diff --git a/tests/test_process.py b/tests/test_process.py index 754c873..f0ed3f7 100644 --- a/tests/test_process.py +++ b/tests/test_process.py @@ -113,6 +113,28 @@ class ProcessTestCase(unittest.TestCase): class FirecrackerProcessSafetyTests(ProcessTestCase): + def test_vsock_socket_is_removed_only_after_confirmed_process_exit(self) -> None: + path = self.manager._settings.guest_socket_path(self.vm.id) + path.parent.mkdir(parents=True) + path.touch() + self.vm.process_start_time = None + with self.assertRaises(FirecrackerError): + self.manager.terminate(self.vm) + self.assertTrue(path.exists()) + self.vm.boot_id = OLD_BOOT_ID + self.manager.terminate(self.vm) + self.assertFalse(path.exists()) + + def test_unowned_vsock_socket_keeps_a_missing_pid_ambiguous(self) -> None: + path = self.manager._settings.guest_socket_path(self.vm.id) + path.parent.mkdir(parents=True) + path.touch() + self.vm.pid = self.vm.process_start_time = None + self.assertEqual(self.manager.observe(self.vm), "unknown") + with self.assertRaises(FirecrackerError): + self.manager.terminate(self.vm) + self.assertTrue(path.exists()) + def test_observes_a_matching_identity_as_alive(self) -> None: self.assertEqual(self.manager.observe(self.vm), "alive") self.assertTrue(self.manager.is_alive(self.vm)) @@ -808,6 +830,10 @@ class FirecrackerLegacyAdoptionTests(ProcessTestCase): def root_lstat(path: Path) -> os.stat_result: metadata = list(real_lstat(path)) metadata[4] = 0 + # Model protected system ancestors independently of the caller's + # TMPDIR permissions. Unsafe-path fixtures live inside self.directory. + if path in self.directory.parents: + metadata[0] &= ~0o022 return os.stat_result(metadata) self.root_lstat = root_lstat diff --git a/tests/test_server.py b/tests/test_server.py index 6431517..5c88e34 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -12,7 +12,11 @@ from unittest.mock import Mock, patch from uvm.cli import build_parser, main from uvm.config import Settings from uvm.domain import PortRule, VmRecord, VmResources, VmSpec -from uvm.errors import ConfigurationError, ConflictError, NotFoundError, UvmError, ValidationError +from uvm.errors import ( + ConfigurationError, ConflictError, GuestAgentUnavailableError, GuestExecutionError, + NotFoundError, UvmError, ValidationError, +) +from uvm.guest_exec import ExecResult from uvm.lifecycle import ListedVm, OperationResult from uvm.server import _http_status_for, create_api, run_server @@ -141,6 +145,8 @@ class ServerErrorTests(unittest.TestCase): (ConfigurationError("invalid settings"), 422), (UvmError("this operation needs root"), 403), (UvmError("host operation failed"), 500), + (GuestAgentUnavailableError("not ready"), 503), + (GuestExecutionError("lost connection", outcome_unknown=True), 502), ): with self.subTest(error=error): self.assertEqual(_http_status_for(error), expected) @@ -159,6 +165,7 @@ class FakeLifecycle: def __init__(self) -> None: self.last_spec: VmSpec | None = None self.last_resize: dict | None = None + self.last_execution: dict | None = None self.observed_status = "running" self.policy_status = "enforced" self.vm = VmRecord( @@ -256,6 +263,19 @@ class FakeLifecycle: self.inspect(identifier) return SimpleNamespace(content=f"last {tail} log lines", truncated=False) + def execute(self, vm_id: str, command: str, *, sudo: bool, timeout_seconds: float) -> ExecResult: + if vm_id != self.vm.id: + raise NotFoundError(f"VM not found: {vm_id}") + self.last_execution = dict(vm_id=vm_id, command=command, sudo=sudo, timeout_seconds=timeout_seconds) + return ExecResult(7, "hello\n", "problem\n", False, False) + + def install_guest_agent(self, vm_id: str) -> VmRecord: + if vm_id != self.vm.id: + raise NotFoundError(f"VM not found: {vm_id}") + if self.vm.status != "stopped": + raise ConflictError("stop the VM before installing the guest agent") + return self.vm + def diagnostics(self) -> list[dict]: return [{"name": "kvm", "ok": True, "message": "available"}] @@ -341,6 +361,8 @@ class ApiRouterTests(unittest.TestCase): ("DELETE", "/vms/test-vm", {}), ("POST", "/vms/test-vm/ports", {}), ("POST", "/vms/test-vm/ports/sync", {}), + ("POST", "/vms/vm-test/exec", {"command": "id"}), + ("POST", "/vms/vm-test/agent/install", {}), ("GET", "/diagnostics", {}), ("POST", "/exit", {}), ("POST", "/install", {}), @@ -655,6 +677,74 @@ class ApiRouterTests(unittest.TestCase): self.assertEqual(sync.json()["id"], "vm-test") self.assertEqual(sync.json()["policy_status"], "inactive") + def test_exec_preserves_script_text_privileges_and_nonzero_exit_results(self) -> None: + command = "printf 'hello\\n'\nprintf 'problem\\n' >&2\nexit 7" + response = self.client.post("/vms/vm-test/exec", headers=self.headers, + json={"command": command, "sudo": True, "timeout_seconds": 12.5}) + self.assertEqual(response.status_code, 200, response.text) + self.assertEqual(response.json(), { + "exit_code": 7, "stdout": "hello\n", "stderr": "problem\n", "timed_out": False, "truncated": False, + }) + self.assertEqual(self.lifecycle.last_execution, { + "vm_id": "vm-test", "command": command, "sudo": True, "timeout_seconds": 12.5, + }) + self.assertNotIn("guest-secret", response.text) + default = self.client.post("/vms/vm-test/exec", headers=self.headers, json={"command": "id"}) + self.assertEqual(default.status_code, 200) + self.assertEqual(self.lifecycle.last_execution["sudo"], False) + self.assertEqual(self.lifecycle.last_execution["timeout_seconds"], 60) + + def test_exec_rejects_malformed_requests_before_calling_the_service(self) -> None: + for payload in ( + {}, {"command": ""}, {"command": " \n"}, {"command": "a\0b"}, {"command": 42}, + {"command": "x" * 65537}, {"command": "\u00e9" * 40000}, + {"command": "id", "sudo": "false"}, {"command": "id", "sudo": 1}, + {"command": "id", "timeout_seconds": True}, {"command": "id", "timeout_seconds": "10"}, + {"command": "id", "timeout_seconds": 0}, {"command": "id", "timeout_seconds": 3601}, + {"command": "id", "username": "root"}, {"command": "id", "host_path": "/etc/passwd"}, + ): + with self.subTest(payload=str(payload)[:100]): + response = self.client.post("/vms/vm-test/exec", headers=self.headers, json=payload) + self.assertEqual(response.status_code, 422, response.text) + self.assertIsNone(self.lifecycle.last_execution) + + def test_exec_errors_distinguish_vm_state_agent_readiness_and_unknown_outcomes(self) -> None: + for error, expected in ((NotFoundError("missing"), 404), (ConflictError("stopped"), 409), + (GuestAgentUnavailableError("not ready"), 503), + (GuestExecutionError("lost response", outcome_unknown=True), 502)): + with self.subTest(error=error), patch.object(self.lifecycle, "execute", side_effect=error): + response = self.client.post("/vms/vm-test/exec", headers=self.headers, json={"command": "id"}) + self.assertEqual(response.status_code, expected) + if isinstance(error, GuestExecutionError): + self.assertTrue(response.json()["error"]["outcome_unknown"]) + with patch.object(self.lifecycle, "execute", return_value=ExecResult(None, "partial", "", True, True)): + response = self.client.post("/vms/vm-test/exec", headers=self.headers, json={"command": "sleep 10"}) + self.assertEqual(response.status_code, 200) + self.assertIsNone(response.json()["exit_code"]) + self.assertTrue(response.json()["timed_out"]) + self.assertTrue(response.json()["truncated"]) + + def test_exec_rejects_nonfinite_and_invalid_unicode_json_without_serialization_errors(self) -> None: + for body in ('{"command":"id","timeout_seconds":NaN}', + '{"command":"id","timeout_seconds":Infinity}', + '{"command":"\\ud800"}', '{"command":"id","sudo":NaN}'): + with self.subTest(body=body): + response = self.client.post("/vms/vm-test/exec", headers={**self.headers, "Content-Type": "application/json"}, + content=body) + self.assertEqual(response.status_code, 422, response.text) + self.assertIsNone(self.lifecycle.last_execution) + + def test_agent_install_requires_stopping_and_uses_the_existing_vm_disk(self) -> None: + blocked = self.client.post("/vms/vm-test/agent/install", headers=self.headers) + self.assertEqual(blocked.status_code, 409) + self.client.post("/vms/vm-test/stop", headers=self.headers) + with patch.object(self.lifecycle, "install_guest_agent", wraps=self.lifecycle.install_guest_agent) as install: + response = self.client.post("/vms/vm-test/agent/install", headers=self.headers) + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json(), {"id": "vm-test", "status": "installed", "restart_required": True}) + install.assert_called_once_with("vm-test") + self.assertEqual(self.lifecycle.vm.status, "stopped") + def test_bulk_action_resolves_targets_before_running_fixed_ids(self) -> None: with ( patch.object(self.lifecycle, "resolve_targets", wraps=self.lifecycle.resolve_targets) as resolve, @@ -786,6 +876,8 @@ class ApiRouterTests(unittest.TestCase): ("/vms/{vm_id}", "delete"), ("/vms/{vm_id}/ports", "post"), ("/vms/{vm_id}/ports/sync", "post"), + ("/vms/{vm_id}/exec", "post"), + ("/vms/{vm_id}/agent/install", "post"), ("/diagnostics", "get"), }, ) diff --git a/uvm/api_models.py b/uvm/api_models.py index 2c7317a..686801e 100644 --- a/uvm/api_models.py +++ b/uvm/api_models.py @@ -5,9 +5,12 @@ from __future__ import annotations from ipaddress import IPv4Address from typing import Literal -from pydantic import BaseModel, ConfigDict, Field, SecretStr +from pydantic import BaseModel, ConfigDict, Field, SecretStr, model_validator from .domain import VmRecord +from .errors import ValidationError +from .guest_agent import DEFAULT_TIMEOUT_SECONDS, MAX_COMMAND_BYTES, MAX_TIMEOUT_SECONDS +from .guest_exec import validate_exec class HealthResponse(BaseModel): @@ -112,6 +115,40 @@ class VmLogResponse(BaseModel): truncated: bool +class VmExecRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + + command: str = Field(strict=True, min_length=1, max_length=MAX_COMMAND_BYTES, repr=False, + description="Command or multiline script interpreted by /bin/sh inside the guest; at most 64 KiB UTF-8") + sudo: bool = Field(default=False, strict=True, description="Run as guest root instead of the VM's configured login user") + timeout_seconds: float = Field(default=DEFAULT_TIMEOUT_SECONDS, strict=True, gt=0, + le=MAX_TIMEOUT_SECONDS, allow_inf_nan=False) + + @model_validator(mode="before") + @classmethod + def validate_execution(cls, value: object) -> object: + # Use the domain error envelope: echoing NaN or unpaired surrogates in + # FastAPI's validation-error input can itself fail JSON serialization. + if not isinstance(value, dict) or set(value) - {"command", "sudo", "timeout_seconds"}: + raise ValidationError("execution requires an object with command, sudo, and timeout_seconds fields only") + validate_exec(value.get("command"), value.get("sudo", False), value.get("timeout_seconds", DEFAULT_TIMEOUT_SECONDS)) + return value + + +class VmExecResponse(BaseModel): + exit_code: int | None + stdout: str + stderr: str + timed_out: bool + truncated: bool + + +class VmAgentInstallResponse(BaseModel): + id: str + status: Literal["installed"] = "installed" + restart_required: bool = True + + class DiagnosticResponse(BaseModel): name: str ok: bool diff --git a/uvm/config.py b/uvm/config.py index b908db6..1d37543 100644 --- a/uvm/config.py +++ b/uvm/config.py @@ -122,6 +122,9 @@ class Settings: def vm_dir(self, vm_id: str) -> Path: return self.vms_dir / vm_id + def guest_socket_path(self, vm_id: str) -> Path: + return self.vm_dir(vm_id) / "vsock.sock" + @classmethod def from_environment(cls) -> "Settings": """Load optional deployment overrides while retaining script defaults.""" diff --git a/uvm/errors.py b/uvm/errors.py index 41c1569..742ca81 100644 --- a/uvm/errors.py +++ b/uvm/errors.py @@ -39,6 +39,18 @@ class FirecrackerError(UvmError): """Raised when Firecracker cannot be started or configured.""" +class GuestAgentUnavailableError(UvmError): + """The VM's execution agent could not be reached before submission.""" + + +class GuestExecutionError(UvmError): + """Guest execution failed; a lost response does not prove the command failed.""" + + def __init__(self, message: str, *, outcome_unknown: bool = False) -> None: + super().__init__(message) + self.outcome_unknown = outcome_unknown + + class TapCreationError(UvmError): """Raised when TAP setup fails after creating a host interface.""" diff --git a/uvm/firecracker/api.py b/uvm/firecracker/api.py index 326e7fc..719b4c6 100644 --- a/uvm/firecracker/api.py +++ b/uvm/firecracker/api.py @@ -36,6 +36,7 @@ class FirecrackerClient: drives = config["drives"] interfaces = config["network-interfaces"] entropy = config.get("entropy") + vsock = config.get("vsock") if not isinstance(machine, Mapping) or not isinstance(boot, Mapping): raise FirecrackerError("invalid Firecracker configuration") if not isinstance(drives, list) or not drives: @@ -44,6 +45,10 @@ class FirecrackerClient: raise FirecrackerError("Firecracker configuration has no network interface") if not isinstance(entropy, Mapping): raise FirecrackerError("Firecracker configuration has no valid entropy device") + if (not isinstance(vsock, Mapping) or type(vsock.get("guest_cid")) is not int + or not 3 <= vsock["guest_cid"] < 2**32 - 1 + or not isinstance(vsock.get("uds_path"), str) or not vsock["uds_path"]): + raise FirecrackerError("Firecracker configuration has no valid vsock device") self._request( "PUT", @@ -57,6 +62,7 @@ class FirecrackerClient: self._request("PUT", "/boot-source", boot) self._request("PUT", "/drives/rootfs", drives[0]) self._request("PUT", "/network-interfaces/eth0", interfaces[0]) + self._request("PUT", "/vsock", vsock) self._request("PUT", "/entropy", entropy) self._request("PUT", "/actions", {"action_type": "InstanceStart"}) diff --git a/uvm/firecracker/config.py b/uvm/firecracker/config.py index ff8d80c..41e9338 100644 --- a/uvm/firecracker/config.py +++ b/uvm/firecracker/config.py @@ -72,6 +72,9 @@ def build_config( ], # Supply host-backed randomness before guest services such as sshd start. "entropy": {}, + # Each Firecracker instance has its own Unix-socket-backed vsock device; + # CIDs need not be allocated across these independent devices. + "vsock": {"guest_cid": 3, "uds_path": str(settings.guest_socket_path(vm.id))}, } diff --git a/uvm/firecracker/process.py b/uvm/firecracker/process.py index 8264a2c..13f28b0 100644 --- a/uvm/firecracker/process.py +++ b/uvm/firecracker/process.py @@ -124,7 +124,7 @@ class FirecrackerProcessManager: ) from error socket_path = Path(vm.socket) - self._remove_socket(socket_path) + self._remove_runtime_sockets(vm) log_path = Path(vm.log) process: _PopenLike | None = None @@ -275,14 +275,16 @@ class FirecrackerProcessManager: if vm.pid is None: if vm.spawn_pending or vm.operation == "spawning" or vm.process_start_time is not None: return "unknown" - try: - Path(vm.socket).lstat() - except FileNotFoundError: - return "gone" - except OSError: + for socket_path in (Path(vm.socket), self._settings.guest_socket_path(vm.id)): + try: + socket_path.lstat() + except FileNotFoundError: + continue + except OSError: + return "unknown" + # An unowned socket may belong to a child whose identity was never saved. return "unknown" - # An unowned socket may belong to a child whose identity was never saved. - return "unknown" + return "gone" stat = self._process_stat(vm.pid) if stat is None: @@ -330,14 +332,14 @@ class FirecrackerProcessManager: f"refusing to stop Firecracker PID {vm.pid}: process identity is ambiguous" ) if observation == "gone": - self._remove_socket(Path(vm.socket)) + self._remove_runtime_sockets(vm) return try: pidfd = _open_pidfd(vm.pid) except ProcessLookupError: self._reap_child(vm.pid) - self._remove_socket(Path(vm.socket)) + self._remove_runtime_sockets(vm) return except (OSError, NotImplementedError) as error: raise FirecrackerError( @@ -369,7 +371,7 @@ class FirecrackerProcessManager: "process identity and socket retained" ) self._reap_child(vm.pid) - self._remove_socket(Path(vm.socket)) + self._remove_runtime_sockets(vm) except (OSError, NotImplementedError) as error: raise FirecrackerError( f"could not safely stop Firecracker process {vm.pid}: {error}" @@ -455,6 +457,10 @@ class FirecrackerProcessManager: if process is not None and process.poll() is not None: self._children.pop(pid, None) + def _remove_runtime_sockets(self, vm: VmRecord) -> None: + self._remove_socket(self._settings.guest_socket_path(vm.id)) + self._remove_socket(Path(vm.socket)) + @staticmethod def _remove_socket(socket_path: Path) -> None: try: diff --git a/uvm/guest_agent.py b/uvm/guest_agent.py new file mode 100644 index 0000000..0611e4f --- /dev/null +++ b/uvm/guest_agent.py @@ -0,0 +1,342 @@ +"""Standalone, stdlib-only guest execution agent (Python 3.10+). + +This file is copied into guest images and must not import the host application. +The host also imports its wire-format helpers; importing it never starts a server. +""" + +from __future__ import annotations + +import json +import math +import os +import pwd +import re +import selectors +import signal +import socket +import stat +import struct +import subprocess +import sys +import threading +import time +import uuid +from pathlib import Path + + +PROTOCOL_VERSION = 1 +AGENT_PORT = 1070 +DEFAULT_TIMEOUT_SECONDS = 60.0 +MAX_TIMEOUT_SECONDS = 3600.0 +MAX_COMMAND_BYTES = 65536 +MAX_OUTPUT_BYTES = 1024 * 1024 # Combined stdout and stderr, before UTF-8 decoding. +MAX_REQUEST_BYTES = 6 * MAX_COMMAND_BYTES + 4096 +MAX_RESPONSE_BYTES = 6 * MAX_OUTPUT_BYTES + 4096 +CONNECT_TIMEOUT_SECONDS = 5.0 +CLEANUP_SECONDS = 5.0 +MAX_CONCURRENT_COMMANDS = 8 +CONFIG_PATH = Path("/etc/uvm-agent.json") + + +class ProtocolError(ValueError): + """An invalid or oversized wire message; never include command text.""" + + +class ExecutionFailure(Exception): + def __init__(self, message: str, *, outcome_unknown: bool = False) -> None: + super().__init__(message) + self.outcome_unknown = outcome_unknown + + +def validate_command(command: object, sudo: object, timeout_seconds: object) -> None: + if not isinstance(command, str) or not command.strip() or "\0" in command: + raise ProtocolError("command must be non-empty text without NUL characters") + try: + size = len(command.encode("utf-8")) + except UnicodeError as error: + raise ProtocolError("command must be valid UTF-8 text") from error + if size > MAX_COMMAND_BYTES: + raise ProtocolError(f"command exceeds {MAX_COMMAND_BYTES} UTF-8 bytes") + if type(sudo) is not bool: + raise ProtocolError("sudo must be a boolean") + try: + valid_timeout = ( + type(timeout_seconds) in (int, float) + and math.isfinite(timeout_seconds) + and 0 < timeout_seconds <= MAX_TIMEOUT_SECONDS + ) + except OverflowError: + valid_timeout = False + if not valid_timeout: + raise ProtocolError(f"timeout_seconds must be finite and greater than 0, up to {MAX_TIMEOUT_SECONDS:g}") + + +def _remaining(deadline: float) -> float: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError("guest protocol deadline exceeded") + return remaining + + +def receive_exact(connection: socket.socket, size: int, deadline: float) -> bytes: + parts = bytearray() + while len(parts) < size: + connection.settimeout(_remaining(deadline)) + chunk = connection.recv(size - len(parts)) + if not chunk: + raise EOFError("guest connection closed") + parts.extend(chunk) + return bytes(parts) + + +def receive_frame(connection: socket.socket, limit: int, deadline: float) -> dict: + size, = struct.unpack("!I", receive_exact(connection, 4, deadline)) + if not 0 < size <= limit: + raise ProtocolError("invalid guest message length") + try: + message = json.loads(receive_exact(connection, size, deadline)) + except (ValueError, UnicodeError, RecursionError) as error: + raise ProtocolError("invalid guest JSON message") from error + if not isinstance(message, dict): + raise ProtocolError("guest message must be an object") + return message + + +def send_frame(connection: socket.socket, message: dict, limit: int, deadline: float) -> None: + payload = json.dumps(message, ensure_ascii=True, allow_nan=False, separators=(",", ":")).encode("ascii") + if len(payload) > limit: + raise ProtocolError("guest message exceeds its size limit") + connection.settimeout(_remaining(deadline)) + connection.sendall(struct.pack("!I", len(payload)) + payload) + + +class JobScope: + """A delegated guest cgroup, keeping detached descendants in the same job.""" + + def __init__(self) -> None: + entries = Path("/proc/self/cgroup").read_text(encoding="ascii").splitlines() + unified = next((entry[3:] for entry in entries if entry.startswith("0::")), None) + if unified is None or ".." in Path(unified).parts: + raise ExecutionFailure("guest execution requires a delegated cgroup v2 hierarchy") + parent = Path("/sys/fs/cgroup") / unified.lstrip("/") + self.path = parent / f"uvm-exec-{uuid.uuid4().hex}" + self.path.mkdir(mode=0o700) + if not (self.path / "cgroup.kill").is_file(): + self.path.rmdir() + raise ExecutionFailure("guest execution requires cgroup.kill (Linux 5.14 or newer)") + + def kill(self) -> None: + (self.path / "cgroup.kill").write_text("1", encoding="ascii") + + def close(self) -> None: + # cgroup.events reports descendants as well as processes in this cgroup. + deadline = time.monotonic() + 1.0 + while "populated 1" in (self.path / "cgroup.events").read_text(encoding="ascii"): + if time.monotonic() >= deadline: + raise ExecutionFailure("guest job cleanup could not be confirmed", outcome_unknown=True) + time.sleep(0.01) + for child in sorted(self.path.rglob("*"), key=lambda path: len(path.parts), reverse=True): + if child.is_dir(): + child.rmdir() + self.path.rmdir() + + +def _signal_group(process: subprocess.Popen, number: int) -> None: + # Do not reap the group leader until after signaling: its PID pins the PGID. + try: + os.killpg(process.pid, number) + except ProcessLookupError: + pass + + +def _exited(process: subprocess.Popen) -> bool: + return os.waitid(os.P_PID, process.pid, os.WEXITED | os.WNOHANG | os.WNOWAIT) is not None + + +def _cleanup_job(process: subprocess.Popen | None, scope: JobScope) -> None: + failed = False + try: + scope.kill() + except (OSError, ExecutionFailure): + failed = True + if process is not None: + try: + _signal_group(process, signal.SIGKILL) + process.wait(timeout=1.0) + except (OSError, subprocess.TimeoutExpired): + failed = True + finally: + for name in ("stdout", "stderr"): + getattr(process, name).close() + try: + scope.close() + except (OSError, ExecutionFailure): + failed = True + if failed: + raise ExecutionFailure("guest job cleanup could not be confirmed", outcome_unknown=process is not None) + + +def run_command(command: str, username: str, timeout_seconds: float, *, scope_factory=JobScope) -> dict: + """Run one non-interactive guest job, drain both pipes, and bound its lifetime.""" + scope = scope_factory() + process = None + buffers = {"stdout": bytearray(), "stderr": bytearray()} + captured = 0 + truncated = False + timed_out = False + try: + # The fresh child joins its scope BEFORE dropping credentials or executing + # user text. Avoid preexec_fn in the threaded agent. + process = subprocess.Popen( + [sys.executable, "-I", str(Path(__file__).resolve()), "--child", str(scope.path), username, command], + stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + start_new_session=True, env={"PATH": "/usr/sbin:/usr/bin:/sbin:/bin", "LANG": "C.UTF-8"}, + ) + deadline = time.monotonic() + timeout_seconds + terminating = None + killed = False + with selectors.DefaultSelector() as selector: + for name in buffers: + stream = getattr(process, name) + os.set_blocking(stream.fileno(), False) + selector.register(stream, selectors.EVENT_READ, name) + while True: + now = time.monotonic() + exited = _exited(process) + if terminating is None and (exited or now >= deadline): + timed_out = not exited + terminating = now + _signal_group(process, signal.SIGTERM if timed_out else signal.SIGKILL) + if terminating is not None: + if not killed and (exited or now >= terminating + 0.5): + scope.kill() + _signal_group(process, signal.SIGKILL) + killed = True + if killed and exited and not selector.get_map(): + break + if now >= terminating + 3.0: + raise ExecutionFailure("guest job cleanup could not be confirmed", outcome_unknown=True) + for key, _events in selector.select(0.05): + chunk = os.read(key.fileobj.fileno(), 65536) + if not chunk: + selector.unregister(key.fileobj) + continue + keep = min(len(chunk), MAX_OUTPUT_BYTES - captured) + buffers[key.data].extend(chunk[:keep]) + captured += keep + truncated |= keep < len(chunk) + # Finally below kills the scope and reaps the leader before returning. + except OSError as error: + raise ExecutionFailure("guest command execution failed", outcome_unknown=process is not None) from error + finally: + _cleanup_job(process, scope) + return { + "exit_code": None if timed_out else process.returncode, + "stdout": buffers["stdout"].decode("utf-8", errors="replace"), + "stderr": buffers["stderr"].decode("utf-8", errors="replace"), + "timed_out": timed_out, + "truncated": truncated, + } + + +def child_main(scope: Path, username: str, command: str) -> None: + """Enter the job cgroup and account in a fresh interpreter, then replace it.""" + (scope / "cgroup.procs").write_text(str(os.getpid()), encoding="ascii") + account = pwd.getpwnam(username) + if os.geteuid() == 0: + os.initgroups(account.pw_name, account.pw_gid) + os.setgid(account.pw_gid) + os.setuid(account.pw_uid) + elif (account.pw_uid, account.pw_gid) != (os.geteuid(), os.getegid()): + raise PermissionError("guest agent cannot assume the requested account") + os.chdir(account.pw_dir) + os.umask(0o022) + environment = { + "HOME": account.pw_dir, "USER": account.pw_name, "LOGNAME": account.pw_name, + "SHELL": "/bin/sh", "PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin", + "LANG": "C.UTF-8", + } + os.execve("/bin/sh", ["sh", "-c", command], environment) + + +def serve_connection(connection: socket.socket, config: dict, *, runner=run_command) -> None: + with connection: + try: + deadline = time.monotonic() + CONNECT_TIMEOUT_SECONDS + send_frame(connection, {"version": PROTOCOL_VERSION, **config}, 4096, deadline) + request = receive_frame(connection, MAX_REQUEST_BYTES, deadline) + if set(request) != {"version", "command", "sudo", "timeout_seconds"}: + raise ProtocolError("invalid execution request fields") + if type(request["version"]) is not int or request["version"] != PROTOCOL_VERSION: + raise ProtocolError("unsupported guest protocol version") + validate_command(request["command"], request["sudo"], request["timeout_seconds"]) + username = "root" if request["sudo"] else config["username"] + result = runner(request["command"], username, request["timeout_seconds"]) + send_frame(connection, result, MAX_RESPONSE_BYTES, time.monotonic() + CONNECT_TIMEOUT_SECONDS) + except (ProtocolError, ExecutionFailure, OSError, KeyError) as error: + # Never log request text, guest output, or credentials to the console. + try: + send_frame(connection, { + "error": "guest execution failed: " + str(error), + "outcome_unknown": isinstance(error, ExecutionFailure) and error.outcome_unknown, + }, 4096, time.monotonic() + CONNECT_TIMEOUT_SECONDS) + except (OSError, ProtocolError): + pass + except EOFError: + pass + + +def load_config(path: Path = CONFIG_PATH) -> dict: + descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW) + with os.fdopen(descriptor, encoding="utf-8") as source: + metadata = os.fstat(source.fileno()) + if (not stat.S_ISREG(metadata.st_mode) or metadata.st_uid != 0 + or metadata.st_mode & 0o077 or metadata.st_size > 4096): + raise ValueError("guest agent configuration must be a root-owned private regular file") + config = json.load(source) + if (not isinstance(config, dict) or set(config) != {"vm_id", "username"} + or not isinstance(config["vm_id"], str) + or re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,80}", config["vm_id"]) is None + or not isinstance(config["username"], str) + or re.fullmatch(r"[a-z_][a-z0-9_-]{0,31}\$?", config["username"]) is None): + raise ValueError("invalid guest agent configuration") + pwd.getpwnam(config["username"]) + return config + + +def main() -> None: + if len(sys.argv) == 5 and sys.argv[1] == "--child": + try: + child_main(Path(sys.argv[2]), sys.argv[3], sys.argv[4]) + except (OSError, KeyError) as error: + print(f"uvm-agent: could not launch guest command: {error}", file=sys.stderr) + sys.exit(126) + return + if len(sys.argv) != 1 or os.geteuid() != 0: + raise SystemExit("uvm-agent must run as a root systemd service") + config = load_config() + slots = threading.BoundedSemaphore(MAX_CONCURRENT_COMMANDS) + + def handle(connection: socket.socket) -> None: + try: + serve_connection(connection, config) + finally: + slots.release() + + with socket.socket(socket.AF_VSOCK, socket.SOCK_STREAM) as listener: + listener.bind((socket.VMADDR_CID_ANY, AGENT_PORT)) + listener.listen(MAX_CONCURRENT_COMMANDS) + while True: + connection, peer = listener.accept() + # Guest-local callers must not use the root agent to elevate themselves. + if peer[0] != socket.VMADDR_CID_HOST: + connection.close() + continue + if not slots.acquire(blocking=False): + connection.close() + continue + threading.Thread(target=handle, args=(connection,), daemon=True).start() + + +if __name__ == "__main__": + main() diff --git a/uvm/guest_exec.py b/uvm/guest_exec.py new file mode 100644 index 0000000..d31141e --- /dev/null +++ b/uvm/guest_exec.py @@ -0,0 +1,114 @@ +"""Host-side, one-shot guest execution over Firecracker's vsock Unix socket.""" + +from __future__ import annotations + +import socket +import stat +import struct +import time +from dataclasses import dataclass +from pathlib import Path + +from .errors import GuestAgentUnavailableError, GuestExecutionError, ValidationError +from .guest_agent import ( + AGENT_PORT, CLEANUP_SECONDS, CONNECT_TIMEOUT_SECONDS, MAX_OUTPUT_BYTES, + MAX_REQUEST_BYTES, MAX_RESPONSE_BYTES, PROTOCOL_VERSION, ProtocolError, + receive_exact, receive_frame, send_frame, validate_command, +) + + +@dataclass(frozen=True, slots=True) +class ExecResult: + exit_code: int | None + stdout: str + stderr: str + timed_out: bool + truncated: bool + + +def validate_exec(command: object, sudo: object, timeout_seconds: object) -> None: + try: + validate_command(command, sudo, timeout_seconds) + except ProtocolError as error: + raise ValidationError(str(error)) from error + + +class GuestSession: + """A connected session pins the selected VMM; never reconnect after submission.""" + + def __init__(self, connection: socket.socket) -> None: + self._connection = connection + + @classmethod + def connect(cls, path: Path, vm_id: str, username: str, pid: int) -> GuestSession: + connection = None + try: + connection = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + metadata = path.lstat() + if not stat.S_ISSOCK(metadata.st_mode): + raise ProtocolError("guest socket is not a Unix socket") + deadline = time.monotonic() + CONNECT_TIMEOUT_SECONDS + connection.settimeout(CONNECT_TIMEOUT_SECONDS) + connection.connect(str(path)) + peer_pid, _uid, _gid = struct.unpack("3i", connection.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12)) + if peer_pid != pid: + raise ProtocolError("guest socket does not belong to the selected VM process") + connection.sendall(f"CONNECT {AGENT_PORT}\n".encode("ascii")) + acknowledgement = bytearray() + while not acknowledgement.endswith(b"\n") and len(acknowledgement) < 32: + acknowledgement.extend(receive_exact(connection, 1, deadline)) + parts = bytes(acknowledgement).split() + if (not acknowledgement.endswith(b"\n") or len(parts) != 2 + or parts[0] != b"OK" or not parts[1].isdigit()): + raise ProtocolError("invalid Firecracker vsock acknowledgement") + hello = receive_frame(connection, 4096, deadline) + if (set(hello) != {"version", "vm_id", "username"} + or type(hello["version"]) is not int or hello["version"] != PROTOCOL_VERSION + or hello["vm_id"] != vm_id or hello["username"] != username): + raise ProtocolError("guest agent identity or protocol version does not match this VM") + return cls(connection) + except (OSError, EOFError, ProtocolError) as error: + if connection is not None: + connection.close() + raise GuestAgentUnavailableError( + "guest agent is unavailable or not ready; install it with POST /vms/{vm_id}/agent/install " + "while stopped, then start the VM (vsock-capable kernel required)" + ) from error + + def execute(self, command: str, *, sudo: bool, timeout_seconds: float) -> ExecResult: + validate_exec(command, sudo, timeout_seconds) + try: + send_frame(self._connection, { + "version": PROTOCOL_VERSION, "command": command, + "sudo": sudo, "timeout_seconds": timeout_seconds, + }, MAX_REQUEST_BYTES, time.monotonic() + CONNECT_TIMEOUT_SECONDS) + result = receive_frame( + self._connection, MAX_RESPONSE_BYTES, + time.monotonic() + timeout_seconds + CLEANUP_SECONDS + CONNECT_TIMEOUT_SECONDS, + ) + if "error" in result: + if (set(result) != {"error", "outcome_unknown"} or not isinstance(result["error"], str) + or len(result["error"]) > 4096 or type(result["outcome_unknown"]) is not bool): + raise ProtocolError("invalid guest failure response") + result["error"].encode("utf-8") # Reject invalid Unicode before rendering an HTTP error. + raise GuestExecutionError(result["error"], outcome_unknown=result["outcome_unknown"]) + if set(result) != {"exit_code", "stdout", "stderr", "timed_out", "truncated"}: + raise ProtocolError("invalid guest result fields") + if (type(result["timed_out"]) is not bool or type(result["truncated"]) is not bool + or not isinstance(result["stdout"], str) or not isinstance(result["stderr"], str) + or (result["timed_out"] and result["exit_code"] is not None) + or (not result["timed_out"] and ( + type(result["exit_code"]) is not int or not -64 <= result["exit_code"] <= 255))): + raise ProtocolError("invalid guest result types") + if sum(len(result[name].encode("utf-8")) for name in ("stdout", "stderr")) > 3 * MAX_OUTPUT_BYTES: + raise ProtocolError("guest output exceeds its limit") + return ExecResult(**result) + except (OSError, EOFError, ProtocolError, UnicodeError) as error: + raise GuestExecutionError( + "guest execution connection was lost or returned an invalid result; " + "the command may have run and will not be retried", + outcome_unknown=True, + ) from error + + def close(self) -> None: + self._connection.close() diff --git a/uvm/images.py b/uvm/images.py index 259b4e5..6abe778 100644 --- a/uvm/images.py +++ b/uvm/images.py @@ -2,6 +2,8 @@ from __future__ import annotations +import hashlib +import json import os import re import stat @@ -11,6 +13,7 @@ from dataclasses import dataclass from ipaddress import IPv4Address from pathlib import Path +from . import guest_agent from .config import MAX_STORAGE_MIB, MIN_STORAGE_MIB, Settings from .errors import UvmError, ValidationError from .integrity import load_manifest, verify_file @@ -46,6 +49,54 @@ _LEGACY_GUEST_NETWORK_TEMPLATE = GUEST_NETWORK_TEMPLATE.replace( "KeepConfiguration=yes", "KeepConfiguration=static", ) +GUEST_AGENT_SERVICE = """\ +[Unit] +Description=UVM guest command execution agent +After=local-fs.target network.target +ConditionPathExists=/etc/uvm-agent.json +StartLimitIntervalSec=0 + +[Service] +Type=simple +ExecStartPre=/bin/sh /usr/lib/uvm/prepare-agent.sh +ExecStart=/usr/bin/python3 -I /usr/lib/uvm/agent.py +Restart=on-failure +RestartSec=5 +TimeoutStartSec=600 +TimeoutStopSec=5 +KillMode=control-group +Delegate=yes +UMask=0077 + +[Install] +WantedBy=multi-user.target +""" + +# Runs only INSIDE the VM. Old Jammy guests can acquire the runtime without SSH +# or executing a retained guest's binaries in a privileged host chroot. +GUEST_AGENT_PREPARE = """\ +set -eu +check_python() { + /usr/bin/python3 -I -c 'import sys, json, math, pwd, selectors, socket, struct, subprocess, threading, uuid; assert sys.version_info >= (3, 10)' 2>/dev/null +} +if check_python; then + exit 0 +fi +export DEBIAN_FRONTEND=noninteractive LC_ALL=C +/usr/bin/apt-get -o DPkg::Lock::Timeout=300 -o APT::Update::Error-Mode=any update +/usr/bin/apt-get -o DPkg::Lock::Timeout=300 -y --no-install-recommends install python3 +check_python +""" + + +def guest_agent_files() -> dict[str, bytes]: + """The same standalone payload is used by the builder and offline upgrades.""" + return { + "/usr/lib/uvm/agent.py": Path(guest_agent.__file__).read_bytes(), + "/usr/lib/uvm/prepare-agent.sh": GUEST_AGENT_PREPARE.encode("utf-8"), + "/etc/systemd/system/uvm-agent.service": GUEST_AGENT_SERVICE.encode("utf-8"), + } + @dataclass(frozen=True, slots=True) class GuestAssets: @@ -232,6 +283,7 @@ class ImageStore: ) if marker is None or marker.path.read_text(encoding="utf-8") not in { "uvm-ubuntu-jammy-v1\n", "uvm-ubuntu-jammy-v2\n", "uvm-ubuntu-jammy-v3\n", + "uvm-ubuntu-jammy-v4\n", }: return None profile = self._read_guest_file( @@ -246,6 +298,119 @@ class ImageStore: return None return profile + def provision_guest_agent(self, disk: Path, vm_id: str, username: str, *, install: bool = False) -> bool: + """Configure new templates, or explicitly install on a proven-offline disk. + + The lifecycle lock and confirmed VMM exit are required. A checked sparse + copy preserves the original disk on failure and makes retries atomic. + """ + parse_username(username) + if re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,80}", vm_id) is None: + raise ValidationError("invalid VM ID for guest agent configuration") + try: + metadata = disk.lstat() + if (not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1 + or disk.resolve() == self._settings.rootfs_image.resolve()): + raise UvmError("private VM disk is unsafe for guest agent installation") + with tempfile.TemporaryDirectory(prefix="uvm-guest-agent-") as name: + files = Path(name) + if not install: + marker = self._read_guest_file(disk, "/etc/uvm-image", files / "marker", + required=False, regular_only=True) + if marker is None or marker.path.read_text(encoding="utf-8") != "uvm-ubuntu-jammy-v4\n": + return False + descriptor, candidate = tempfile.mkstemp(prefix=".rootfs-agent-", suffix=".ext4", dir=disk.parent) + os.close(descriptor) + temporary = Path(candidate) + try: + self._copy_sparse(disk, temporary) + temporary.chmod(0o600) + checked = self._runner.run(("e2fsck", "-f", "-p", temporary), check=False, capture=True) + if checked.returncode not in (0, 1): + raise UvmError("filesystem check failed before guest agent installation") + # All ownership/content checks happen AFTER journal replay. + self._install_agent_files(temporary, files, vm_id, username) + self._runner.run(("e2fsck", "-f", "-n", temporary), capture=True) + with temporary.open("rb") as image: + os.fsync(image.fileno()) + os.replace(temporary, disk) + directory = os.open(disk.parent, os.O_RDONLY) + try: + os.fsync(directory) + finally: + os.close(directory) + finally: + temporary.unlink(missing_ok=True) + return True + except (OSError, UnicodeError) as error: + raise UvmError(f"could not provision guest agent: {error}") from error + + def _install_agent_files(self, disk: Path, files: Path, vm_id: str, username: str) -> None: + for binary in ("/usr/lib/systemd/systemd", "/usr/bin/apt-get"): + result = self._runner.run(("debugfs", "-R", f"stat {binary}", disk), capture=True) + if re.search(r"Type:\s+regular\b", result.stdout) is None: + raise UvmError("guest agent installation requires a systemd/apt image (Ubuntu 22.04 or newer)") + for directory in ("/etc", "/usr", "/usr/lib", "/usr/lib/uvm", "/etc/systemd", + "/etc/systemd/system", "/etc/systemd/system/multi-user.target.wants"): + self._agent_directory(disk, directory) + payloads = guest_agent_files() + payloads["/etc/uvm-agent.json"] = (json.dumps({"vm_id": vm_id, "username": username}) + "\n").encode("utf-8") + manifest_path = "/etc/uvm-agent-files.json" + previous = self._agent_file(disk, manifest_path, files / "manifest") + old_hashes = {} + if previous is not None: + try: + old_hashes = json.loads(previous.path.read_bytes()) + except (ValueError, UnicodeError) as error: + raise UvmError("invalid guest agent ownership manifest") from error + if (not isinstance(old_hashes, dict) or set(old_hashes) - set(payloads) + or any(not isinstance(value, str) or re.fullmatch(r"[0-9a-f]{64}", value) is None + for value in old_hashes.values())): + raise UvmError("invalid guest agent ownership manifest") + hashes = {path: hashlib.sha256(content).hexdigest() for path, content in payloads.items()} + for index, (path, content) in enumerate(payloads.items()): + existing = self._agent_file(disk, path, files / f"existing-{index}") + if existing is not None: + digest = hashlib.sha256(existing.path.read_bytes()).hexdigest() + if digest not in {hashes[path], old_hashes.get(path)}: + raise UvmError(f"refusing to overwrite a customized guest agent file: {path}") + source = files / f"payload-{index}" + source.write_bytes(content) + self._write_guest_file(disk, path, _GuestFile(source, 0o600 if path.endswith(".json") else 0o644, 0, 0)) + source = files / "new-manifest" + source.write_text(json.dumps(hashes, sort_keys=True) + "\n", encoding="utf-8") + self._write_guest_file(disk, manifest_path, _GuestFile(source, 0o600, 0, 0)) + link = "/etc/systemd/system/multi-user.target.wants/uvm-agent.service" + target = "/etc/systemd/system/uvm-agent.service" + result = self._runner.run(("debugfs", "-R", f"stat {link}", disk), capture=True) + if "file not found" in (result.stdout + result.stderr).lower(): + self._runner.run(("debugfs", "-w", "-R", f"symlink {link} {target}", disk), capture=True) + result = self._runner.run(("debugfs", "-R", f"stat {link}", disk), capture=True) + if (re.search(r"Type:\s+symlink\b", result.stdout) is None + or f'Fast link dest: "{target}"' not in result.stdout): + raise UvmError("could not enable guest agent without replacing a customized service link") + + def _agent_directory(self, disk: Path, path: str) -> None: + result = self._runner.run(("debugfs", "-R", f"stat {path}", disk), capture=True) + if "file not found" in (result.stdout + result.stderr).lower(): + self._create_guest_directory(disk, path, 0o755, 0, 0) + return + mode, owner = _MODE_PATTERN.search(result.stdout), _OWNER_PATTERN.search(result.stdout) + if (re.search(r"Type:\s+directory\b", result.stdout) is None or mode is None or owner is None + or int(owner.group(1)) != 0 or int(mode.group(1), 8) & 0o022): + raise UvmError(f"unsafe guest agent directory: {path}") + + def _agent_file(self, disk: Path, path: str, destination: Path) -> _GuestFile | None: + result = self._runner.run(("debugfs", "-R", f"stat {path}", disk), capture=True) + if "file not found" in (result.stdout + result.stderr).lower(): + return None + if re.search(r"Type:\s+regular\b", result.stdout) is None: + raise UvmError(f"unsafe guest agent file: {path}") + source = self._read_guest_file(disk, path, destination) + if (source.uid != 0 or source.mode & 0o022 or source.links != 1 or source.has_extended_attributes): + raise UvmError(f"unsafe guest agent file: {path}") + return source + @staticmethod def _validate_storage(storage_mib: int) -> None: if type(storage_mib) is not int or not MIN_STORAGE_MIB <= storage_mib <= MAX_STORAGE_MIB: diff --git a/uvm/install.py b/uvm/install.py index 3237b3b..b662893 100644 --- a/uvm/install.py +++ b/uvm/install.py @@ -14,17 +14,17 @@ from pathlib import Path from .config import Settings from .errors import UvmError -from .images import GUEST_NETWORK_TEMPLATE, GuestAssets +from .images import GUEST_NETWORK_TEMPLATE, GuestAssets, guest_agent_files from .integrity import load_manifest, require_checksum, sha256_file, verify_file, write_manifest from .state import StateStore from .system import CommandRunner, check_kvm, ensure_data_directories, require_root _UBUNTU_RELEASE = "jammy" -_ROOTFS_RECIPE = "uvm-ubuntu-jammy-v3\n" +_ROOTFS_RECIPE = "uvm-ubuntu-jammy-v4\n" _GUEST_PACKAGES = ( "ca-certificates", "openssh-server", "sudo", "systemd-sysv", "udev", - "dbus", "libpam-systemd", "iproute2", "iputils-ping", "curl", + "dbus", "libpam-systemd", "iproute2", "iputils-ping", "curl", "python3", ) # Keep the image mountable by the older Firecracker quickstart guest kernels. _EXT4_FEATURES = ( @@ -397,7 +397,14 @@ class Installer: machine_id.symlink_to("../../../etc/machine-id") for key in (root / "etc/ssh").glob("ssh_host_*_key*"): key.unlink() - self._runner.run(("systemctl", "--root", root, "enable", "ssh.service", "systemd-networkd.service"), capture=True) + for guest_path, contents in guest_agent_files().items(): + path = root / guest_path.lstrip("/") + path.parent.mkdir(parents=True, exist_ok=True) + path.parent.chmod(0o755) + path.write_bytes(contents) + path.chmod(0o644) + self._runner.run(("systemctl", "--root", root, "enable", "ssh.service", "systemd-networkd.service", + "uvm-agent.service"), capture=True) self._runner.run(("systemctl", "--root", root, "set-default", "multi-user.target"), capture=True) @staticmethod @@ -406,6 +413,7 @@ class Installer: "etc/passwd", "etc/shadow", "etc/group", "etc/sudoers", "etc/ssh/sshd_config", "usr/bin/sudo", "usr/bin/apt-get", "usr/sbin/sshd", "usr/lib/systemd/systemd", "var/lib/dpkg/status", "var/lib/dpkg/info/apt.list", "var/lib/dpkg/info/sudo.list", + "usr/bin/python3", "usr/lib/uvm/agent.py", "etc/systemd/system/uvm-agent.service", ): if not (root / relative).is_file(): raise UvmError(f"built Ubuntu rootfs is missing /{relative}") @@ -414,7 +422,7 @@ class Installer: fields = dict(line.split(": ", 1) for line in paragraph.splitlines() if ": " in line) if fields.get("Status") == "install ok installed": installed.add(fields.get("Package")) - if not {"apt", "sudo", "openssh-server", "systemd-sysv"}.issubset(installed): + if not {"apt", "sudo", "openssh-server", "systemd-sysv", "python3"}.issubset(installed): raise UvmError("built Ubuntu rootfs has incomplete apt/dpkg package metadata") if not (root / "usr/bin/sudo").stat().st_mode & 0o4000: raise UvmError("built Ubuntu rootfs sudo is missing its setuid permission") diff --git a/uvm/lifecycle.py b/uvm/lifecycle.py index bb18eb8..236521b 100644 --- a/uvm/lifecycle.py +++ b/uvm/lifecycle.py @@ -17,6 +17,8 @@ from .errors import ConflictError, NotFoundError, UvmError, ValidationError from .firecracker.api import FirecrackerClient from .firecracker.config import build_config, write_config from .firecracker.process import FirecrackerProcessManager, ProcessInfo +from .guest_agent import DEFAULT_TIMEOUT_SECONDS +from .guest_exec import ExecResult, GuestSession, validate_exec from .images import ImageStore from .integrity import load_manifest, verify_file from .known_hosts import KnownHostsManager @@ -58,6 +60,7 @@ class LifecycleService: network: NetworkManager, process: FirecrackerProcessManager, *, known_hosts: KnownHostsManager, client_factory: Callable[[Path, float], FirecrackerClient] = FirecrackerClient, + guest_connector: Callable[[Path, str, str, int], GuestSession] = GuestSession.connect, clock: Callable[[], float] = time.time, ) -> None: self._settings = settings @@ -67,6 +70,7 @@ class LifecycleService: self._process = process self._known_hosts = known_hosts self._client_factory = client_factory + self._guest_connector = guest_connector self._clock = clock def create(self, spec: VmSpec) -> VmRecord: @@ -90,6 +94,7 @@ class LifecycleService: disk = self._images.create_vm_disk(assets.rootfs, Path(vm.disk), storage_mib=spec.storage_mib) self._images.provision_credentials(disk, vm.username, vm.password) self._images.configure_guest_network(disk, vm.guest_ip) + self._images.provision_guest_agent(disk, vm.id, vm.username) vm.disk_ready = True self._save(vm) self._boot(vm, assets.kernel) @@ -494,6 +499,41 @@ class LifecycleService: def find_for_ssh(self, identifier: str) -> VmRecord: return self._state_store.find_by_id_or_ip(identifier) + def execute( + self, vm_id: str, command: str, *, sudo: bool = False, + timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS, + ) -> ExecResult: + """Pin a running VM's guest connection under the lock, then wait unlocked.""" + require_root() + validate_exec(command, sudo, timeout_seconds) + with self._state_store.operation_lock(): + vm = self._load_for_operation(vm_id, exact_id=True) + self._not_deleting(vm) + if (vm.status != "running" or vm.operation is not None or vm.pending_resize is not None + or self._process.observe(vm) != "alive"): + raise ConflictError("guest execution requires a running VM with no pending lifecycle operation") + session = self._guest_connector(self._settings.guest_socket_path(vm.id), vm.id, vm.username, vm.pid) + try: + # This session stays attached to that VMM even if the VM is stopped, + # deleted, or restarted concurrently. Never reconnect/retry a command. + return session.execute(command, sudo=sudo, timeout_seconds=timeout_seconds) + finally: + session.close() + + def install_guest_agent(self, vm_id: str) -> VmRecord: + """Explicitly install/upgrade the agent on a cleanly stopped private disk.""" + require_root() + with self._state_store.operation_lock(): + vm = self._load_for_operation(vm_id, exact_id=True) + self._not_deleting(vm) + if (vm.status != "stopped" or vm.operation is not None or vm.pending_resize is not None + or self._process.observe(vm) != "gone"): + raise ConflictError("stop the VM and finish pending operations before installing the guest agent") + if not vm.disk_ready or not vm.disk: + raise ConflictError("guest agent installation requires a completely provisioned private disk") + self._images.provision_guest_agent(Path(vm.disk), vm.id, vm.username, install=True) + return vm + def log_path(self, identifier: str) -> Path: vm = self._state_store.find_by_id_or_ip(identifier) self._validate_paths(vm) @@ -566,6 +606,8 @@ class LifecycleService: directory = self._settings.vm_dir(vm.id).absolute() if directory.is_symlink(): raise ConflictError("VM directory must not be a symbolic link") + if self._settings.guest_socket_path(vm.id).is_symlink(): + raise ConflictError("unsafe guest vsock path") for field, filename in (("socket", "firecracker.sock"), ("config", "config.json"), ("log", "firecracker.log"), ("disk", "rootfs.ext4")): path = getattr(vm, field) diff --git a/uvm/routers/vms.py b/uvm/routers/vms.py index f06d666..7475ad4 100644 --- a/uvm/routers/vms.py +++ b/uvm/routers/vms.py @@ -9,8 +9,11 @@ from ..api_models import ( PortSyncResponse, PortUpdateRequest, VmActionResultResponse, + VmAgentInstallResponse, VmBulkActionRequest, VmCreateRequest, + VmExecRequest, + VmExecResponse, VmLogResponse, VmResponse, VmResizeRequest, @@ -202,6 +205,32 @@ def sync_ports( ) +@router.post("/{vm_id}/exec", response_model=VmExecResponse) +def execute_in_vm( + vm_id: str, + request: VmExecRequest, + application: Application = Depends(get_authorized_application), +) -> VmExecResponse: + """Execute a bounded, non-interactive guest command over vsock (exact VM ID).""" + result = application.lifecycle.execute( + vm_id, request.command, sudo=request.sudo, timeout_seconds=request.timeout_seconds, + ) + return VmExecResponse( + exit_code=result.exit_code, stdout=result.stdout, stderr=result.stderr, + timed_out=result.timed_out, truncated=result.truncated, + ) + + +@router.post("/{vm_id}/agent/install", response_model=VmAgentInstallResponse) +def install_guest_agent( + vm_id: str, + application: Application = Depends(get_authorized_application), +) -> VmAgentInstallResponse: + """Install/upgrade the execution agent on a stopped VM's retained private disk.""" + vm = application.lifecycle.install_guest_agent(vm_id) + return VmAgentInstallResponse(id=vm.id) + + @router.get("/{identifier}/logs", response_model=VmLogResponse) def get_vm_logs( identifier: str, diff --git a/uvm/server.py b/uvm/server.py index 7bfcdc6..b87c218 100644 --- a/uvm/server.py +++ b/uvm/server.py @@ -9,7 +9,10 @@ import ssl from typing import Any from .app import Application, build_application -from .errors import ConfigurationError, ConflictError, NotFoundError, UvmError, ValidationError +from .errors import ( + ConfigurationError, ConflictError, GuestAgentUnavailableError, GuestExecutionError, + NotFoundError, UvmError, ValidationError, +) def create_api(application: Application | None = None, *, host: str | None = None) -> Any: @@ -49,9 +52,12 @@ def create_api(application: Application | None = None, *, host: str | None = Non @api.exception_handler(UvmError) async def handle_uvm_error(_request: Any, error: UvmError) -> Any: + detail: dict[str, Any] = {"message": str(error)} + if isinstance(error, GuestExecutionError): + detail["outcome_unknown"] = error.outcome_unknown return JSONResponse( status_code=_http_status_for(error), - content={"error": {"message": str(error)}}, + content={"error": detail}, ) api.include_router(health_router) @@ -130,6 +136,10 @@ def _validate_server_settings( def _http_status_for(error: UvmError) -> int: + if isinstance(error, GuestAgentUnavailableError): + return 503 + if isinstance(error, GuestExecutionError): + return 502 if isinstance(error, NotFoundError): return 404 if isinstance(error, ConflictError):