diff --git a/.gitignore b/.gitignore index 2dea2f9..5eb9221 100644 --- a/.gitignore +++ b/.gitignore @@ -17,3 +17,7 @@ id_rsa* # visual brainstorming scratch .superpowers/ testing/ + +# The worker VM's secrets and the contract files copied onto it by hand +deploy/worker-vm/worker.env +deploy/worker-vm/deployments/ diff --git a/control-plane/mergit_cp/app.py b/control-plane/mergit_cp/app.py index 9232c12..e983a43 100644 --- a/control-plane/mergit_cp/app.py +++ b/control-plane/mergit_cp/app.py @@ -14,6 +14,7 @@ from . import ( auth, chain_events, + health, oauth, otp, redeem, @@ -110,6 +111,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]: allow_headers=["Content-Type", "Idempotency-Key"], expose_headers=["Retry-After"], # the app says how long to wait after a 429 (redeem codes) ) + app.include_router(health.router) app.include_router(auth.router) app.include_router(auth.account_router) app.include_router(routes_account.router) diff --git a/control-plane/mergit_cp/health.py b/control-plane/mergit_cp/health.py new file mode 100644 index 0000000..001802c --- /dev/null +++ b/control-plane/mergit_cp/health.py @@ -0,0 +1,37 @@ +"""``GET /healthz``: is this API able to serve, i.e. is its database reachable? + +Used by the compose healthcheck and by deploy/server/update.sh. It is deliberately public and says nothing +beyond ok or not ok: no version, no settings, no error text. +""" + +import asyncio +import logging +from typing import Annotated + +from fastapi import APIRouter, Depends +from fastapi.responses import JSONResponse + +from .db import Pool +from .deps import get_pool + +log = logging.getLogger(__name__) + +router = APIRouter(tags=["health"]) + +# Under a proxy's own health timeout, so a stuck database answers 503 instead of hanging the check. +DB_CHECK_TIMEOUT_S = 3.0 + + +async def _database_answers(pool: Pool) -> None: + async with pool.connection() as conn: + await conn.execute("SELECT 1") + + +@router.get("/healthz", include_in_schema=False) +async def healthz(pool: Annotated[Pool, Depends(get_pool)]) -> JSONResponse: + try: + await asyncio.wait_for(_database_answers(pool), DB_CHECK_TIMEOUT_S) + except Exception: # any failure, including a timeout or an exhausted pool, means "not healthy" + log.warning("healthz: the database did not answer", exc_info=True) + return JSONResponse({"ok": False}, status_code=503) + return JSONResponse({"ok": True}) diff --git a/control-plane/mergit_cp/pr_store.py b/control-plane/mergit_cp/pr_store.py index 96d0fdf..78bbe45 100644 --- a/control-plane/mergit_cp/pr_store.py +++ b/control-plane/mergit_cp/pr_store.py @@ -125,9 +125,15 @@ def connect(cls, database_url: str, worker_id: str = "api") -> PgPrStore: @staticmethod def _open(database_url: str) -> psycopg.Connection[dict[str, Any]]: + # connect_timeout: a database that is down or unreachable must not hang this reconnect forever. return cast( "psycopg.Connection[dict[str, Any]]", - psycopg.connect(database_url, autocommit=True, row_factory=dict_row), # pyright: ignore[reportArgumentType] + psycopg.connect( + database_url, + autocommit=True, + row_factory=dict_row, # pyright: ignore[reportArgumentType] + connect_timeout=10, + ), ) def _retrying(self, fn: Callable[[], _T]) -> _T: diff --git a/control-plane/mergit_worker/__main__.py b/control-plane/mergit_worker/__main__.py index c471ead..e83121f 100644 --- a/control-plane/mergit_worker/__main__.py +++ b/control-plane/mergit_worker/__main__.py @@ -23,6 +23,7 @@ from .providers import llm_provider from .publish import ReceiptPublisher from .sandbox import Sandbox, SandboxConfig, reap_stale +from .startup import WorkerIdLock, agents_without_key, sweep_workspaces from .tools import ToolLimits from .worker import Worker, WorkerDeps from .workspace import GitLimits @@ -141,6 +142,34 @@ def main() -> int: return 2 worker_id = cfg.worker_id or socket.gethostname() + if not cfg.worker_id: + log.warning( + "WORKER_ID is not set, so this worker is named after its hostname (%s). A container's hostname " + "changes whenever it is recreated, and a restarted worker would then no longer find the jobs and " + "sandbox containers its previous run left. Set WORKER_ID to a fixed name, one per process.", + worker_id, + ) + try: + no_key = agents_without_key(cfg.agents_file) + except ConfigError as exc: + log.error("configuration error: %s", exc) + return 2 + if no_key: + names = ", ".join(f"MERGIT_AGENT_KEY_{i}" for i in no_key) + if cfg.require_all_agent_keys: + log.error("REQUIRE_ALL_AGENT_KEYS is set and these are missing or malformed: %s", names) + return 2 + log.warning( + "no usable key for agent(s) %s: a job for them fails, and the requester loses an attempt", names + ) + lock = WorkerIdLock.acquire(cfg.database_url.get_secret_value(), worker_id) + if lock is None: + log.error( + "another worker process is already running as %r. Starting here would fail its running jobs and " + "remove its sandbox containers. Give each process its own WORKER_ID.", + worker_id, + ) + return 2 log.info("model provider: %s", llm_provider()) built = build_worker(cfg, worker_id) log.info("failed %d jobs this worker left running before a restart", built.jobs.fail_own_running()) @@ -160,6 +189,10 @@ def main() -> int: else: log.info("no GITHUB_TOKEN: pull-request requests will not be handled by this worker") log.info("removed %d stale sandbox containers", reap_stale(worker_id)) + log.info( + "removed %d abandoned workspace directories", + sweep_workspaces(cfg.sandbox_workspace_root, cfg.job_lease_s), + ) stop = threading.Event() @@ -172,6 +205,7 @@ def request_stop(_signum: int, _frame: object) -> None: built.worker.run_forever(stop) finally: built.close() + lock.close() return 0 diff --git a/control-plane/mergit_worker/config.py b/control-plane/mergit_worker/config.py index 8380e11..487e9d6 100644 --- a/control-plane/mergit_worker/config.py +++ b/control-plane/mergit_worker/config.py @@ -63,6 +63,9 @@ class WorkerConfig(BaseSettings): # Under the 1800 s dispatch deadline, leaving time to run Φ, seal and publish. job_deadline_s: float = Field(default=1500.0, gt=0) worker_id: str | None = None # defaults to the hostname; one process per id + # Refuse to start unless every agent in the agents file has a MERGIT_AGENT_KEY_. Off by default: a + # worker may deliberately serve only some of the agents (startup.agents_without_key logs the rest). + require_all_agent_keys: bool = False # Opus 5.5's thinking tokens count against this; 8192 made a truncated turn too likely. model_max_tokens: int = Field(default=16_000, ge=256) poll_interval_s: float = Field(default=2.0, gt=0) diff --git a/control-plane/mergit_worker/jobs.py b/control-plane/mergit_worker/jobs.py index 3720cc2..01961be 100644 --- a/control-plane/mergit_worker/jobs.py +++ b/control-plane/mergit_worker/jobs.py @@ -97,9 +97,15 @@ def connect(cls, database_url: str, worker_id: str) -> PgJobStore: @staticmethod def _open(database_url: str) -> psycopg.Connection[dict[str, Any]]: + # connect_timeout: a database that is down or unreachable must not hang this reconnect forever. return cast( "psycopg.Connection[dict[str, Any]]", - psycopg.connect(database_url, autocommit=True, row_factory=dict_row), # pyright: ignore[reportArgumentType] + psycopg.connect( + database_url, + autocommit=True, + row_factory=dict_row, # pyright: ignore[reportArgumentType] + connect_timeout=10, + ), ) def _retrying(self, fn: Callable[[], _T]) -> _T: diff --git a/control-plane/mergit_worker/startup.py b/control-plane/mergit_worker/startup.py new file mode 100644 index 0000000..f2f04d5 --- /dev/null +++ b/control-plane/mergit_worker/startup.py @@ -0,0 +1,111 @@ +"""Start-up checks and housekeeping for one worker process (plan task O10). + +The worker runs on its own VM, restarted by Docker and redeployed by hand, so what it leaves behind when it +stops has to be cleaned up by the next start: jobs, sandbox containers (sandbox.reap_stale) and workspace +directories (here). Everything below is a plain function over a path, an environment or a database URL, so it +is tested without a Docker daemon. +""" + +from __future__ import annotations + +import json +import logging +import os +import re +import shutil +import time +from collections.abc import Mapping +from pathlib import Path + +import psycopg + +from .config import ConfigError + +log = logging.getLogger(__name__) + +_AGENT_KEY = re.compile(r"^0x[0-9a-fA-F]{64}$") +WORKSPACE_PREFIX = "ws-" # workspace.Workspace.clone names its directories ws- + + +def agents_without_key(agents_file: Path, environ: Mapping[str, str] | None = None) -> list[str]: + """Agent ids in the agents file that have no usable ``MERGIT_AGENT_KEY_`` here. + + The key is read when a job for that agent starts, so a missing one is otherwise found only after a + requester has picked the agent, funded the plan and burned an attempt on it. + """ + env = os.environ if environ is None else environ + try: + registry: list[dict[str, object]] = json.loads(agents_file.read_text())["agents"] + except (OSError, KeyError, ValueError) as exc: + raise ConfigError(f"cannot read agent configuration: {exc}") from exc + ids = [str(a["agent_id"]) for a in registry if "agent_id" in a] + return [i for i in ids if not _AGENT_KEY.match(env.get(f"MERGIT_AGENT_KEY_{i}", ""))] + + +def _remove_workspace(path: Path) -> None: + """Delete one workspace tree, including directories a sandbox left unreadable. + + Never follows a symlink: a sandboxed agent can plant one, and chmod or rmtree through it would reach + outside the workspace root. + """ + for dirpath, dirnames, _files in os.walk(path, followlinks=False): + os.chmod(dirpath, 0o700) + for name in dirnames: + child = os.path.join(dirpath, name) + if not os.path.islink(child): + os.chmod(child, 0o700) + shutil.rmtree(path) + + +def sweep_workspaces(root: Path, older_than_s: float, *, now: float | None = None) -> int: + """Remove ``ws-*`` directories under ``root`` that nothing can still be using. Returns how many. + + A crash or a SIGKILL skips the normal clean-up, and ``reap_stale`` removes containers only, so without + this the VM's disk fills one abandoned checkout at a time. ``older_than_s`` must exceed the longest a + job can run (the lease, which is also when its row is reclaimed), so a job still in flight in another + process on this host is never touched. + """ + if not root.is_dir(): + return 0 + cutoff = (time.time() if now is None else now) - older_than_s + removed = 0 + for entry in root.iterdir(): + if not entry.name.startswith(WORKSPACE_PREFIX) or entry.is_symlink() or not entry.is_dir(): + continue + try: + if entry.stat().st_mtime >= cutoff: + continue + _remove_workspace(entry) + removed += 1 + except OSError as exc: + log.warning("could not remove stale workspace %s: %s", entry, exc) + return removed + + +class WorkerIdLock: + """A session-level Postgres advisory lock named for the worker id, held while the process lives. + + A worker id must belong to one process. At start-up a worker fails every job still ``running`` under its + id and removes every sandbox container labelled with it, so a second process sharing the id would + destroy the first one's live work. Postgres releases the lock when the connection ends, whether the + process exits cleanly or is killed, so a crashed worker never blocks its own restart. + """ + + def __init__(self, conn: psycopg.Connection[tuple[object, ...]]) -> None: + self._conn = conn + + @classmethod + def acquire(cls, database_url: str, worker_id: str) -> WorkerIdLock | None: + # Without a timeout, a database that is down or unreachable (a dead host, a network partition) hangs + # this call forever instead of letting start-up fail and Docker's restart policy retry it. + conn = psycopg.connect(database_url, autocommit=True, connect_timeout=10) # pyright: ignore[reportArgumentType] + row = conn.execute( + "SELECT pg_try_advisory_lock(hashtextextended(%s, 0))", (f"mergit_worker:{worker_id}",) + ).fetchone() + if row is not None and row[0] is True: + return cls(conn) + conn.close() + return None + + def close(self) -> None: + self._conn.close() diff --git a/control-plane/tests/test_caddy_routes.py b/control-plane/tests/test_caddy_routes.py index 087cd3a..5d48061 100644 --- a/control-plane/tests/test_caddy_routes.py +++ b/control-plane/tests/test_caddy_routes.py @@ -32,11 +32,22 @@ WITHHELD = ("/v0/account",) +def _block(host: str) -> str: + text = CADDYFILE.read_text() + match = re.search(rf"^{re.escape(host)} \{{\n(.*?)\n\}}$", text, re.S | re.M) + assert match, f"no {host} block in the Caddyfile" + return match.group(1) + + def _proxied_patterns() -> list[str]: + """The browser routes: one shared snippet, imported by the site (mergit.site and app.mergit.site) and by + api.mergit.site.""" text = CADDYFILE.read_text() - block = re.search(r"(?m)^mergit\.site, app\.mergit\.site \{.*?@api path ([^\n]+)", text, re.S) - assert block, "the mergit.site block has no `@api path` matcher" - return block.group(1).split() + snippet = re.search(r"^\(api_routes\) \{\n\s*@api path ([^\n]+)\n\}", text, re.M) + assert snippet, "the (api_routes) snippet has no `@api path` matcher" + for host in ("mergit.site, app.mergit.site", "api.mergit.site"): + assert "import api_routes" in _block(host), f"{host} must import the shared route list" + return snippet.group(1).split() def _browser_paths() -> list[str]: @@ -89,3 +100,55 @@ def test_a_withheld_route_is_a_real_route() -> None: def test_the_check_would_notice_a_route_it_does_not_know() -> None: assert not any(fnmatch("/v0/brand-new/thing", pat) for pat in _proxied_patterns()) + + +def test_the_api_host_serves_only_what_it_means_to() -> None: + """api.mergit.site used to proxy everything, so the worker's internal route and the API's docs pages + were reachable from anywhere. Now: the browser routes, /healthz, and the internal route from the + worker VM only.""" + block = _block("api.mergit.site") + assert "handle @api {" in block and "handle /healthz {" in block + assert re.search(r"handle \{\s*respond \"not found\" 404\s*\}", block), "the catch-all must be a 404" + assert not re.search(r"handle \{\s*reverse_proxy", block), "a catch-all reverse_proxy exposes every route" + + +def test_the_internal_route_is_proxied_only_from_the_worker_vm() -> None: + block = _block("api.mergit.site") + matcher = re.search(r"@worker \{(.*?)\}", block, re.S) + assert matcher, "no @worker matcher" + assert "path /v0/internal/*" in matcher.group(1) + assert "remote_ip {$WORKER_VM_IP:" in matcher.group(1), "the internal route must be limited by IP" + assert "handle @worker" in block + + +def test_the_worker_vm_default_address_is_one_nothing_can_use() -> None: + """If WORKER_VM_IP is unset the rule must admit nobody. 192.0.2.0/24 is reserved for documentation.""" + default = re.search(r"remote_ip \{\$WORKER_VM_IP:([^}]+)\}", CADDYFILE.read_text()) + assert default and default.group(1).startswith("192.0.2."), "the fallback must be a documentation address" + compose = (CADDYFILE.parent / "compose.server.yaml").read_text() + assert "WORKER_VM_IP: ${WORKER_VM_IP:-192.0.2.1}" in compose, ( + "compose must pass the same unusable default" + ) + + +def test_chain_events_is_never_reachable_through_caddy() -> None: + """Only the gateway calls it, over the compose network. Comments may mention it; configuration may not.""" + config = "\n".join( + line for line in CADDYFILE.read_text().splitlines() if not line.lstrip().startswith("#") + ) + assert "chain-events" not in config + + +def test_every_internal_route_can_reach_the_api_from_the_worker_vm() -> None: + """A new worker-facing route must not end up proxied nowhere. Today: /v0/internal/progress.""" + block = _block("api.mergit.site") + pattern = re.search(r"path (/v0/internal/\*)", block) + assert pattern + internal_paths = { + re.sub(r"\{[^}]+\}", "x", r.path) + for r in routes_runs.internal_router.routes + if hasattr(r, "path") # pyright: ignore[reportUnknownArgumentType] + } # pyright: ignore[reportAttributeAccessIssue,reportUnknownMemberType,reportUnknownVariableType] + assert internal_paths, "the internal router has no routes?" + uncovered = [p for p in internal_paths if not fnmatch(p, pattern.group(1))] + assert uncovered == [], f"worker-facing routes the Caddy rule does not cover: {uncovered}" diff --git a/control-plane/tests/test_cp_postgres_hba.py b/control-plane/tests/test_cp_postgres_hba.py new file mode 100644 index 0000000..7015952 --- /dev/null +++ b/control-plane/tests/test_cp_postgres_hba.py @@ -0,0 +1,69 @@ +"""cp_postgres must not let the superuser in from outside the compose network (deploy/postgres/pg_hba.conf). + +CP_DB_BIND publishes the port on the private network for the worker VM. With the image's default rules the +superuser would log in from there too, with the password in compose.controlplane.yaml. +""" + +import re +from pathlib import Path + +DEPLOY = Path(__file__).resolve().parents[2] / "deploy" + + +def _rules() -> list[list[str]]: + text = (DEPLOY / "postgres" / "pg_hba.conf").read_text() + return [line.split() for line in text.splitlines() if line.strip() and not line.lstrip().startswith("#")] + + +def _compose_subnet() -> str: + """The CIDR pinned in compose.controlplane.yaml's `networks.default.ipam`, which pg_hba.conf must match + exactly: a looser or different rule there would silently stop matching this file's `all all` line.""" + compose = (DEPLOY / "compose.controlplane.yaml").read_text() + match = re.search(r"subnet:\s*(\S+)\s*}", compose) + assert match, "no pinned subnet in compose.controlplane.yaml's networks.default.ipam" + return match.group(1) + + +def test_cp_postgres_uses_the_repository_hba_file() -> None: + compose = (DEPLOY / "compose.controlplane.yaml").read_text() + block = re.search(r"^ cp_postgres:\n(.*?)\n\n", compose, re.S | re.M) + assert block, "no cp_postgres service" + assert "hba_file=/etc/postgresql/mergit/pg_hba.conf" in block.group(1) + assert '"./postgres:/etc/postgresql/mergit:ro"' in block.group(1) + + +def test_the_compose_network_is_pinned_not_left_to_docker_to_pick() -> None: + # `samenet` in pg_hba.conf would match whatever subnet Docker assigns, including a bridge gateway address + # a published port can see as its source under some Docker network settings. A fixed CIDR cannot. + subnet = _compose_subnet() + assert re.fullmatch(r"\d+\.\d+\.\d+\.\d+/\d+", subnet), subnet + + +def test_only_the_worker_role_may_log_in_from_outside_the_compose_network() -> None: + compose_subnet = _compose_subnet() + remote = [ + r + for r in _rules() + if r[0].startswith("host") and r[3] not in ("127.0.0.1/32", "::1/128", "samenet", compose_subnet) + ] + assert remote == [["host", "mergit_cp", "mergit_worker", "all", "scram-sha-256"]] + + +def test_the_compose_network_rule_uses_the_pinned_subnet_not_samenet() -> None: + compose_subnet = _compose_subnet() + broad = [r for r in _rules() if r[0:3] == ["host", "all", "all"] and r[-1] == "scram-sha-256"] + assert broad == [["host", "all", "all", compose_subnet, "scram-sha-256"]] + + +def test_no_rule_trusts_a_network_address() -> None: + trusted = [r for r in _rules() if r[0].startswith("host") and r[-1] == "trust"] + assert all(r[3] in ("127.0.0.1/32", "::1/128") for r in trusted) + + +def test_the_api_port_is_not_published_to_every_interface() -> None: + # On the server, compose.server.yaml resets this to nothing and Caddy reaches `api` by name. Published to + # 0.0.0.0 here, a run without compose.server.yaml would expose the unproxied API to the internet. + compose = (DEPLOY / "compose.controlplane.yaml").read_text() + block = re.search(r"^ api:\n(.*?)\n\n", compose, re.S | re.M) + assert block, "no api service" + assert re.search(r'ports:\s*\["127\.0\.0\.1:8000:8000"\]', block.group(1)) diff --git a/control-plane/tests/test_health.py b/control-plane/tests/test_health.py new file mode 100644 index 0000000..c1535e8 --- /dev/null +++ b/control-plane/tests/test_health.py @@ -0,0 +1,85 @@ +"""/healthz tells a proxy or a deploy script whether this API can serve (its database answers).""" + +from __future__ import annotations + +import asyncio +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from typing import Any, cast + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from mergit_cp import health + +pytestmark = pytest.mark.postgres + + +def test_healthz_is_ok_when_the_database_answers(client: TestClient) -> None: + response = client.get("/healthz") + assert response.status_code == 200 + assert response.json() == {"ok": True} + + +def test_healthz_needs_no_session_and_leaks_nothing(client: TestClient) -> None: + body = client.get("/healthz").text + for word in ("version", "postgres", "password", "secret", "token"): + assert word not in body.lower() + + +class _Conn: + def __init__(self, hang: bool) -> None: + self._hang = hang + + async def execute(self, _sql: str) -> None: + if self._hang: + await asyncio.sleep(60) + raise OSError("connection refused") + + +class _BrokenPool: + def __init__(self, hang: bool = False) -> None: + self._hang = hang + + @asynccontextmanager + async def connection(self) -> AsyncIterator[_Conn]: + yield _Conn(self._hang) + + +def _app_with(pool: object) -> FastAPI: + app = FastAPI() + app.state.pool = pool + app.include_router(health.router) + return app + + +def test_healthz_is_503_when_the_database_errors() -> None: + with TestClient(_app_with(_BrokenPool())) as c: + response = c.get("/healthz") + assert response.status_code == 503 + assert response.json() == {"ok": False} + assert "refused" not in response.text # the error text stays in the log + + +def test_healthz_is_503_not_a_hang_when_the_database_stalls(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(health, "DB_CHECK_TIMEOUT_S", 0.2) + with TestClient(_app_with(_BrokenPool(hang=True))) as c: + response = c.get("/healthz") + assert response.status_code == 503 + + +def test_the_database_probe_is_one_select(monkeypatch: pytest.MonkeyPatch) -> None: + seen: list[str] = [] + + class Conn: + async def execute(self, sql: str) -> None: + seen.append(sql) + + class Pool: + @asynccontextmanager + async def connection(self) -> AsyncIterator[Conn]: + yield Conn() + + asyncio.run(health._database_answers(cast(Any, Pool()))) + assert seen == ["SELECT 1"] diff --git a/control-plane/tests/test_startup.py b/control-plane/tests/test_startup.py new file mode 100644 index 0000000..23482e8 --- /dev/null +++ b/control-plane/tests/test_startup.py @@ -0,0 +1,182 @@ +"""Start-up checks and housekeeping for a worker that lives on its own VM (plan task O10).""" + +from __future__ import annotations + +import json +import os +from pathlib import Path + +import pytest + +from mergit_worker.config import ConfigError +from mergit_worker.startup import agents_without_key, sweep_workspaces + +KEY_A = "0x" + "a" * 64 +KEY_B = "0x" + "b" * 64 + + +def agents_file(tmp_path: Path, *ids: str) -> Path: + path = tmp_path / "agents.json" + path.write_text(json.dumps({"agents": [{"agent_id": i, "name": f"agent-{i}"} for i in ids]})) + return path + + +# -- agents_without_key -------------------------------------------------------------------------------- + + +def test_every_agent_with_a_key_means_nothing_is_missing(tmp_path: Path) -> None: + env = {"MERGIT_AGENT_KEY_1": KEY_A, "MERGIT_AGENT_KEY_2": KEY_B} + assert agents_without_key(agents_file(tmp_path, "1", "2"), env) == [] + + +def test_an_agent_with_no_key_is_reported(tmp_path: Path) -> None: + assert agents_without_key(agents_file(tmp_path, "1", "2"), {"MERGIT_AGENT_KEY_1": KEY_A}) == ["2"] + + +@pytest.mark.parametrize( + "bad", + ["", "0x1234", "0x" + "g" * 64, KEY_A[2:], KEY_A + "00", " " + KEY_A], + ids=["empty", "short", "not-hex", "no-0x", "too-long", "leading-space"], +) +def test_a_malformed_key_counts_as_missing(tmp_path: Path, bad: str) -> None: + assert agents_without_key(agents_file(tmp_path, "1"), {"MERGIT_AGENT_KEY_1": bad}) == ["1"] + + +def test_a_key_for_another_agent_does_not_cover_this_one(tmp_path: Path) -> None: + assert agents_without_key(agents_file(tmp_path, "10"), {"MERGIT_AGENT_KEY_1": KEY_A}) == ["10"] + + +def test_an_unreadable_agents_file_is_a_config_error_not_a_crash(tmp_path: Path) -> None: + with pytest.raises(ConfigError): + agents_without_key(tmp_path / "missing.json", {}) + broken = tmp_path / "broken.json" + broken.write_text("{not json") + with pytest.raises(ConfigError): + agents_without_key(broken, {}) + wrong_shape = tmp_path / "shape.json" + wrong_shape.write_text('{"not_agents": []}') + with pytest.raises(ConfigError): + agents_without_key(wrong_shape, {}) + + +def test_reads_the_process_environment_by_default(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("MERGIT_AGENT_KEY_7", KEY_A) + assert agents_without_key(agents_file(tmp_path, "7")) == [] + + +# -- sweep_workspaces ---------------------------------------------------------------------------------- + +HOUR = 3600.0 + + +def workspace(root: Path, name: str, age_s: float, now: float) -> Path: + ws = root / name + (ws / "work").mkdir(parents=True) + (ws / "git").mkdir() + (ws / "work" / "file.txt").write_text("checkout") + os.utime(ws, (now - age_s, now - age_s)) + return ws + + +def test_an_old_workspace_is_removed_and_a_recent_one_is_kept(tmp_path: Path) -> None: + now = 1_000_000.0 + old = workspace(tmp_path, "ws-old", 2 * HOUR, now) + fresh = workspace(tmp_path, "ws-fresh", 60, now) + assert sweep_workspaces(tmp_path, HOUR, now=now) == 1 + assert not old.exists() + assert fresh.exists() + + +def test_a_workspace_exactly_at_the_limit_is_kept(tmp_path: Path) -> None: + # A job still inside its lease must never lose its checkout. + now = 1_000_000.0 + edge = workspace(tmp_path, "ws-edge", HOUR, now) + assert sweep_workspaces(tmp_path, HOUR, now=now) == 0 + assert edge.exists() + + +def test_only_workspace_directories_are_touched(tmp_path: Path) -> None: + now = 1_000_000.0 + other_dir = tmp_path / "keep-me" + other_dir.mkdir() + loose_file = tmp_path / "ws-not-a-directory" + loose_file.write_text("x") + for p in (other_dir, loose_file): + os.utime(p, (now - 5 * HOUR, now - 5 * HOUR)) + assert sweep_workspaces(tmp_path, HOUR, now=now) == 0 + assert other_dir.exists() and loose_file.exists() + + +def test_a_missing_root_is_nothing_to_do(tmp_path: Path) -> None: + assert sweep_workspaces(tmp_path / "does-not-exist", HOUR) == 0 + + +def test_a_workspace_a_sandbox_left_unreadable_is_still_removed(tmp_path: Path) -> None: + now = 1_000_000.0 + ws = workspace(tmp_path, "ws-locked", 2 * HOUR, now) + locked = ws / "work" / "locked" + locked.mkdir() + (locked / "inner.txt").write_text("x") + locked.chmod(0o000) + try: + assert sweep_workspaces(tmp_path, HOUR, now=now) == 1 + finally: + if locked.exists(): + locked.chmod(0o700) + assert not ws.exists() + + +def test_a_planted_symlink_is_removed_but_never_followed(tmp_path: Path) -> None: + """A sandboxed agent controls the contents of its workspace. A link to a directory elsewhere must not + make the sweep chmod or delete what it points at.""" + now = 1_000_000.0 + outside = tmp_path / "outside" + outside.mkdir() + (outside / "precious.txt").write_text("keep") + outside.chmod(0o755) + root = tmp_path / "root" + root.mkdir() + ws = workspace(root, "ws-evil", 2 * HOUR, now) + (ws / "work" / "escape").symlink_to(outside, target_is_directory=True) + (ws / "work" / "dangling").symlink_to(tmp_path / "nowhere") + + assert sweep_workspaces(root, HOUR, now=now) == 1 + + assert not ws.exists() + assert (outside / "precious.txt").read_text() == "keep" + assert outside.stat().st_mode & 0o777 == 0o755 + + +def test_a_workspace_root_that_is_itself_a_symlink_entry_is_skipped(tmp_path: Path) -> None: + now = 1_000_000.0 + real = tmp_path / "elsewhere" + real.mkdir() + os.utime(real, (now - 5 * HOUR, now - 5 * HOUR)) + root = tmp_path / "root" + root.mkdir() + (root / "ws-link").symlink_to(real, target_is_directory=True) + assert sweep_workspaces(root, HOUR, now=now) == 0 + assert real.exists() + + +def test_a_removal_failure_is_logged_and_the_rest_still_go( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + now = 1_000_000.0 + stuck = workspace(tmp_path, "ws-a-stuck", 2 * HOUR, now) + other = workspace(tmp_path, "ws-b-other", 2 * HOUR, now) + + import mergit_worker.startup as startup + + real = startup._remove_workspace + + def flaky(path: Path) -> None: + if path.name == "ws-a-stuck": + raise PermissionError("busy") + real(path) + + monkeypatch.setattr(startup, "_remove_workspace", flaky) + with caplog.at_level("WARNING"): + assert sweep_workspaces(tmp_path, HOUR, now=now) == 1 + assert stuck.exists() and not other.exists() + assert "ws-a-stuck" in caplog.text diff --git a/control-plane/tests/test_worker_role.py b/control-plane/tests/test_worker_role.py new file mode 100644 index 0000000..7140d3f --- /dev/null +++ b/control-plane/tests/test_worker_role.py @@ -0,0 +1,213 @@ +"""The worker VM connects to the control-plane database as its own least-privilege role (plan task O10). + +``deploy/worker-vm/db-role.sql`` is the grant list. These tests run it as shipped against a database built by +the real migrations, then act as that role: every statement the worker issues must work, and nothing else may. +""" + +from __future__ import annotations + +import asyncio +from collections.abc import Iterator +from pathlib import Path + +import psycopg +import pytest +from psycopg import errors +from psycopg.conninfo import make_conninfo + +from mergit_cp.db import migrate +from mergit_cp.pr_store import _CHANGES, PgPrStore +from mergit_worker import jobs as worker_jobs +from mergit_worker.jobs import PgJobStore +from mergit_worker.startup import WorkerIdLock + +pytestmark = pytest.mark.postgres + +GRANTS = Path(__file__).resolve().parents[2] / "deploy" / "worker-vm" / "db-role.sql" +ROLE = "mergit_worker" +PASSWORD = "role-test-password" +DB = "worker_role_test" +TASK = "0x" + "0" * 64 +PLAN = "00000000-0000-0000-0000-000000000000" +WORKER_TABLES = {"jobs", "pr_requests", "plans", "tasks"} +SENSITIVE = {"sessions", "users", "identities", "wallets", "email_otps", "oauth_states", "wallet_interest"} + + +async def _migrate(url: str) -> None: + async with await psycopg.AsyncConnection.connect(url) as conn: + await migrate(conn) + + +@pytest.fixture(scope="module") +def admin_url(database_url: str) -> Iterator[str]: + """A throwaway database with every migration applied and the grants file run, as the real script does.""" + with psycopg.connect(database_url, autocommit=True) as root: + root.execute(f"DROP DATABASE IF EXISTS {DB}") + root.execute(f"CREATE DATABASE {DB}") + url = make_conninfo(database_url, dbname=DB) + asyncio.run(_migrate(url)) + with psycopg.connect(database_url, autocommit=True) as root: + root.execute(f"DROP ROLE IF EXISTS {ROLE}") # a previous run's role, owned by nothing now + root.execute(f"CREATE ROLE {ROLE} LOGIN PASSWORD '{PASSWORD}'") + with psycopg.connect(url, autocommit=True) as conn: + conn.execute(GRANTS.read_text()) # pyright: ignore[reportArgumentType] + yield url + with psycopg.connect(database_url, autocommit=True) as root: + root.execute(f"DROP DATABASE {DB} WITH (FORCE)") + root.execute(f"DROP ROLE {ROLE}") + + +@pytest.fixture +def role_url(admin_url: str) -> str: + return make_conninfo(admin_url, user=ROLE, password=PASSWORD) + + +def _tables(admin_url: str) -> set[str]: + with psycopg.connect(admin_url) as conn: + rows = conn.execute("SELECT tablename FROM pg_tables WHERE schemaname = 'public'").fetchall() + return {r[0] for r in rows} + + +# -- everything the worker does works ------------------------------------------------------------------ + + +def test_the_job_queue_methods_work_as_the_role(role_url: str) -> None: + store = PgJobStore.connect(role_url, "role-test") + try: + assert store.fail_own_running() == 0 + assert store.reclaim_stale(3600.0) == 0 + assert store.claim() is None # SELECT ... FOR UPDATE SKIP LOCKED, then UPDATE + store.fail(TASK, "nothing to fail") # UPDATE ... WHERE task_id + finally: + store.close() + + +def test_the_job_loading_queries_work_as_the_role(role_url: str) -> None: + """_LOAD joins plans and tasks, and a dependent task reads its dependencies' patches: both need SELECT.""" + with psycopg.connect(role_url, autocommit=True) as conn: + assert conn.execute(worker_jobs._LOAD, {"task_id": TASK}).fetchall() == [] + assert ( + conn.execute(worker_jobs._DEPENDENCY_PATCHES, {"plan_uuid": PLAN, "indexes": [0]}).fetchall() + == [] + ) + params = {"status": "done", "holds": True, "root": None, "notes": None, "patch": None} + assert conn.execute(worker_jobs._REPORT, {**params, "task_id": TASK, "worker": "w"}).rowcount == 0 + + +def test_the_pull_request_queue_methods_work_as_the_role(role_url: str) -> None: + store = PgPrStore.connect(role_url, "role-test") + try: + assert store.claim() is None + assert store.requeue_own_running() == 0 + assert store.reclaim_stale(3600.0) == 0 + assert store.plan(PLAN) is None # SELECT FROM plans + assert store.request(PLAN) is None + store.finish(PLAN, "branch", "a" * 40, 1, "https://example.test/pr/1") + store.fail(PLAN, "nothing to fail") + finally: + store.close() + with psycopg.connect(role_url, autocommit=True) as conn: + assert conn.execute(_CHANGES, {"plan": PLAN}).fetchall() == [] # tasks LEFT JOIN jobs + + +# -- and nothing else does ------------------------------------------------------------------------------ + + +def test_the_sensitive_tables_exist_so_the_next_test_is_not_vacuous(admin_url: str) -> None: + assert _tables(admin_url) >= SENSITIVE + + +def test_the_role_cannot_read_any_table_it_was_not_given(admin_url: str, role_url: str) -> None: + denied = sorted(_tables(admin_url) - WORKER_TABLES) + assert set(denied) >= SENSITIVE + with psycopg.connect(role_url, autocommit=True) as conn: + for table in denied: + with pytest.raises(errors.InsufficientPrivilege): + conn.execute(f'SELECT * FROM "{table}" LIMIT 1') # pyright: ignore[reportArgumentType] + + +@pytest.mark.parametrize( + "sql", + [ + "DELETE FROM jobs", + "DELETE FROM pr_requests", + "INSERT INTO pr_requests (plan_uuid, requested_by) VALUES ('{plan}', 'x')", + "UPDATE plans SET state = 'done'", + "DELETE FROM plans", + "UPDATE tasks SET title = 'x'", + "DELETE FROM tasks", + "TRUNCATE jobs", + "CREATE TABLE stolen (x int)", + "ALTER TABLE jobs ADD COLUMN x int", + "DROP TABLE jobs", + ], + ids=lambda s: s.split(" (")[0], +) +def test_the_role_cannot_change_more_than_the_worker_needs(role_url: str, sql: str) -> None: + with psycopg.connect(role_url, autocommit=True) as conn, pytest.raises(errors.InsufficientPrivilege): + conn.execute(sql.format(plan=PLAN)) # pyright: ignore[reportArgumentType] + + +def test_running_the_grants_again_takes_extra_rights_away(admin_url: str, role_url: str) -> None: + with psycopg.connect(admin_url, autocommit=True) as conn: + conn.execute(f"GRANT SELECT ON sessions TO {ROLE}") + with psycopg.connect(role_url, autocommit=True) as conn: + conn.execute("SELECT 1 FROM sessions LIMIT 1") # the leak the next run must close + with psycopg.connect(admin_url, autocommit=True) as conn: + conn.execute(GRANTS.read_text()) # pyright: ignore[reportArgumentType] + with psycopg.connect(role_url, autocommit=True) as conn, pytest.raises(errors.InsufficientPrivilege): + conn.execute("SELECT 1 FROM sessions LIMIT 1") + + +# -- one process per worker id -------------------------------------------------------------------------- + + +def test_a_second_process_with_the_same_worker_id_is_refused(database_url: str) -> None: + first = WorkerIdLock.acquire(database_url, "dup-id") + assert first is not None + try: + assert WorkerIdLock.acquire(database_url, "dup-id") is None + finally: + first.close() + + +def test_different_worker_ids_do_not_block_each_other(database_url: str) -> None: + a = WorkerIdLock.acquire(database_url, "id-a") + b = WorkerIdLock.acquire(database_url, "id-b") + assert a is not None and b is not None + a.close() + b.close() + + +def test_the_id_is_free_again_once_the_holder_exits(database_url: str) -> None: + first = WorkerIdLock.acquire(database_url, "restart-id") + assert first is not None + first.close() + again = WorkerIdLock.acquire(database_url, "restart-id") + assert again is not None + again.close() + + +def test_a_killed_worker_does_not_block_its_own_restart(database_url: str) -> None: + """A SIGKILLed container never calls close(); Postgres drops the session and with it the lock.""" + dead = WorkerIdLock.acquire(database_url, "killed-id") + assert dead is not None + with psycopg.connect(database_url, autocommit=True) as admin: + admin.execute( + "SELECT pg_terminate_backend(pid) FROM pg_stat_activity" + " WHERE pid <> pg_backend_pid() AND query LIKE '%pg_try_advisory_lock%'" + ) + restarted = None + for _ in range(50): # the server notices the dropped connection almost at once + restarted = WorkerIdLock.acquire(database_url, "killed-id") + if restarted is not None: + break + asyncio.run(asyncio.sleep(0.1)) + assert restarted is not None + restarted.close() + + +def test_the_role_can_take_the_lock_too(role_url: str) -> None: + lock = WorkerIdLock.acquire(role_url, "role-lock-id") + assert lock is not None + lock.close() diff --git a/deploy/Caddyfile b/deploy/Caddyfile index a387ba3..1213af7 100644 --- a/deploy/Caddyfile +++ b/deploy/Caddyfile @@ -25,6 +25,15 @@ rewrite @spa /index.html } +# The API routes a browser calls. One list for the site (mergit.site and app.mergit.site) and api.mergit.site, so the +# two cannot drift; control-plane/tests/test_caddy_routes.py fails when a route in the API is missing from it. +# The bare /v0/account is not listed on purpose: it is DELETE /v0/account, deleting the account. It stays off the public +# origin until the gaps in what a deletion removes and in the once-per-person free credits are decided (see +# accounts.delete_account). /v0/account/* is every other account route. +(api_routes) { + @api path /v0/auth/* /v0/account/* /v0/wallet-interest /v0/plans /v0/plans/* /v0/agents /v0/agents/* +} + # The site: the frontend on live Base Sepolia data, and the orchestration API on the same origin. It is served on both # mergit.site and app.mergit.site, identically. Each page calls /v0/auth, /v0/account, /v0/wallet-interest, /v0/plans # and /v0/agents (with its /reputation) on its own hostname, so there is no CORS. The gateway and the worker reach the API directly over the @@ -33,10 +42,7 @@ # send people back to, since each allows one callback address), EXTRA_DOMAINS=mergit.site (wallet sign-in is accepted # from both) and COOKIE_DOMAIN=mergit.site (the session cookie is shared, so signing in on one signs in on the other). mergit.site, app.mergit.site { - # The bare /v0/account is not listed on purpose: it is DELETE /v0/account, deleting the account. It stays off the public - # origin until the gaps in what a deletion removes and in the once-per-person free credits are decided (see - # accounts.delete_account). /v0/account/* below is every other account route. - @api path /v0/auth/* /v0/account/* /v0/wallet-interest /v0/plans /v0/plans/* /v0/agents /v0/agents/* + import api_routes # On every response, the page's and the API's. HSTS has no includeSubDomains or preload: other names under this # domain are not all known to be HTTPS-only, and a preload entry cannot be taken back quickly. header { @@ -69,8 +75,28 @@ store.mergit.site { reverse_proxy store:8081 } -# The same API on its own name, for anything that is not the web app. Optional: the site uses its own hostnames above. +# The API on its own name. It serves the browser routes (api_routes) and /healthz, and the one route the agent worker +# calls from its own VM: /v0/internal/progress, only from the IP in WORKER_VM_IP (deploy/.env, passed in by +# compose.server.yaml; the default is a documentation address nothing can use). Everything else is 404: the +# gateway reaches /v0/chain-events over the compose network, and the API's own docs pages are not for the public. +# The worker's bearer token is still checked by the API; the IP rule only keeps the route off the open internet. # Needs an A record for `api`. It sets its own CORS headers (CORS_ORIGINS) and a Secure, HttpOnly session cookie. api.mergit.site { - reverse_proxy api:8000 + import api_routes + @worker { + path /v0/internal/* + remote_ip {$WORKER_VM_IP:192.0.2.1} + } + handle @api { + reverse_proxy api:8000 + } + handle @worker { + reverse_proxy api:8000 + } + handle /healthz { + reverse_proxy api:8000 + } + handle { + respond "not found" 404 + } } diff --git a/deploy/compose.controlplane.yaml b/deploy/compose.controlplane.yaml index da4e76d..9c04623 100644 --- a/deploy/compose.controlplane.yaml +++ b/deploy/compose.controlplane.yaml @@ -8,14 +8,28 @@ # The API and the worker get separate secret files so each holds only what it needs: neither has the gateway or # treasury key, and only the worker has an agent runtime key and the Docker socket. +# Pinned so deploy/postgres/pg_hba.conf can match it by an exact CIDR instead of `samenet`, which can be widened by +# how a given host's Docker handles published ports (see the comment there). Change both together. +networks: + default: + ipam: + config: [{ subnet: 172.30.9.0/24 }] + services: cp_postgres: image: postgres:16 restart: unless-stopped # The control plane does not share the gateway's database: both define tables named plans, tasks and outbox. environment: { POSTGRES_PASSWORD: mergit, POSTGRES_DB: mergit_cp } - volumes: ["cp_pgdata:/var/lib/postgresql/data"] + # deploy/postgres/pg_hba.conf: from outside this compose network, only the mergit_worker role may log in. The + # folder is mounted, not the file, so a `git pull` that replaces the file is seen (apply it with a restart). + command: ["postgres", "-c", "hba_file=/etc/postgresql/mergit/pg_hba.conf"] + volumes: ["cp_pgdata:/var/lib/postgresql/data", "./postgres:/etc/postgresql/mergit:ro"] healthcheck: { test: ["CMD", "pg_isready", "-U", "postgres"], interval: 2s, retries: 30 } + # Only the worker VM connects from outside (deploy/worker-vm/README.md), as the restricted mergit_worker role. + # Loopback until CP_DB_BIND (deploy/.env) names this server's private-network address. Never 0.0.0.0: Docker's + # published ports skip the host firewall, so the port would be open to the internet. + ports: ["${CP_DB_BIND:-127.0.0.1}:5433:5432"] # The gateway posts chain events to the control plane instead of the host. gateway: @@ -42,11 +56,22 @@ services: image: mergit-api:local restart: unless-stopped env_file: secrets/api.env - ports: ["8000:8000"] + # Loopback only: on the server, compose.server.yaml resets this to nothing and Caddy reaches `api` by name + # instead (deploy/Caddyfile). Published to all interfaces this would be the unproxied API, open to the + # internet and skipping the Caddy lockdown, for a run without compose.server.yaml. + ports: ["127.0.0.1:8000:8000"] depends_on: cp_postgres: { condition: service_healthy } volumes: - ../../mergit-contracts/deployments/agents-base-sepolia.json:/agents.json:ro + # /healthz answers 200 only while the database does. The slim image has no curl, so python makes the request; + # urlopen raises on a 503, which is a non-zero exit. + healthcheck: + test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/healthz', timeout=3)"] + interval: 15s + timeout: 5s + retries: 3 + start_period: 40s environment: DATABASE_URL: postgres://postgres:mergit@cp_postgres:5432/mergit_cp GATEWAY_URL: http://gateway:8082 diff --git a/deploy/compose.server.yaml b/deploy/compose.server.yaml index d92d014..9ee88c4 100644 --- a/deploy/compose.server.yaml +++ b/deploy/compose.server.yaml @@ -20,6 +20,10 @@ services: - /srv/mergit/sites:/srv:ro - caddy_data:/data - caddy_config:/config + environment: + # The agent worker's VM (deploy/worker-vm/README.md): the only address allowed to call the API's internal + # routes. The default is a documentation address nothing can use, i.e. nobody. + WORKER_VM_IP: ${WORKER_VM_IP:-192.0.2.1} depends_on: [store, gateway] volumes: { caddy_data: {}, caddy_config: {} } diff --git a/deploy/postgres/pg_hba.conf b/deploy/postgres/pg_hba.conf new file mode 100644 index 0000000..32d4628 --- /dev/null +++ b/deploy/postgres/pg_hba.conf @@ -0,0 +1,16 @@ +# Who may log in to cp_postgres (deploy/compose.controlplane.yaml). It replaces the image's default, whose last line, +# `host all all all scram-sha-256`, lets the superuser in from any address that reaches the port, with the password +# in compose.controlplane.yaml. Once CP_DB_BIND publishes the port on the private network (deploy/worker-vm), that +# address would include the worker VM, the machine most likely to be compromised, and the restricted role would be +# pointless. So: the containers on this compose network may use any role, and anything else only mergit_worker. +# +# The compose network is pinned to the subnet below (deploy/compose.controlplane.yaml, `networks.default.ipam`) +# and matched here by that exact CIDR, not `samenet`: `samenet` matches any address sharing a subnet with any of +# the server's interfaces, which under some Docker network settings (docker-proxy's own bridge gateway, +# `iptables: false`) is the address a published port sees a connection arrive from, superuser password and all. +# A fixed, narrow CIDR cannot widen that way. Keep the two in step if the subnet is ever changed. +local all all trust +host all all 127.0.0.1/32 trust +host all all ::1/128 trust +host all all 172.30.9.0/24 scram-sha-256 +host mergit_cp mergit_worker all scram-sha-256 diff --git a/deploy/server/README.md b/deploy/server/README.md index 1df3fa9..bede8a4 100644 --- a/deploy/server/README.md +++ b/deploy/server/README.md @@ -10,7 +10,8 @@ One server (Hetzner CPX32, Nuremberg) runs everything that used to run on a lapt | `store.mergit.site` | evidence store (receipt bundles) | Caddy, then `store:8081` | | _(not public)_ | validator, Postgres | compose network only | | `app.mergit.site/v0/{auth,account,plans,agents}` | orchestration API (control plane), same origin as the app | Caddy, then `api:8000`; needs `deploy/compose.controlplane.yaml` | -| `api.mergit.site` | the same API on its own name (optional) | Caddy, then `api:8000`; needs an A record | +| `api.mergit.site` | the same API on its own name: browser routes, `/healthz`, and `/v0/internal/*` from the worker VM only (`WORKER_VM_IP`); everything else 404 | Caddy, then `api:8000`; needs an A record | +| _(its own VM)_ | the agent worker | `deploy/worker-vm/README.md` | Everything is testnet money. The Hetzner firewall (`mergit-fw`) allows only 22, 80 and 443. diff --git a/deploy/server/update.sh b/deploy/server/update.sh index a01fc49..ffd6a36 100755 --- a/deploy/server/update.sh +++ b/deploy/server/update.sh @@ -32,5 +32,18 @@ case "$what" in esac case "$what" in all) deploy/server/build-sites.sh all ;; web|app) deploy/server/build-sites.sh "$what" ;; esac +# `up -d` returns when the containers have started, not when the API has run its start-up migrations, so give it a +# moment instead of failing a good deploy on the first request. +wait_healthy() { # name url [tries] + local tries=${3:-30} + for _ in $(seq "$tries"); do + if curl -fsS -m 5 "$2/healthz" >/dev/null 2>&1; then echo "$1: ok"; return 0; fi + sleep 2 + done + echo "$1 did not become healthy at $2/healthz" >&2 + return 1 +} + compose ps for h in gateway store; do printf '%s: ' "$h"; curl -fsS "https://$h.mergit.site/healthz"; echo; done +wait_healthy api https://api.mergit.site diff --git a/deploy/worker-vm/README.md b/deploy/worker-vm/README.md new file mode 100644 index 0000000..ee2f2bb --- /dev/null +++ b/deploy/worker-vm/README.md @@ -0,0 +1,65 @@ +# The agent worker on its own VM (plan task O10, issue #26) + +The worker starts sandbox containers through the host's Docker socket, and the sandboxes run code a model wrote. Whoever +controls the worker process controls its host, so it gets a VM with nothing else on it. This VM holds its agent keys, +model keys and one restricted database role. It has no gateway key, no treasury key, no web app and no database. + +``` + worker VM ──(private network, :5433, role mergit_worker)──▶ server: cp_postgres + ──https──▶ gateway.mergit.site (receipt.sealed) + ──https──▶ store.mergit.site (receipt bundles) + ──https──▶ api.mergit.site/v0/internal/progress (live steps; allowed only from this VM's IP) +``` + +## 1. On the server (once) + +1. A second Hetzner server in the same project, plus a private network joining both. Note the server's private address. +2. `deploy/.env` (git-ignored; compose reads it): `CP_DB_BIND=` and `WORKER_VM_IP=`. + Leave `CP_DB_BIND` unset and Postgres stays on loopback. Never set it to `0.0.0.0`: Docker's published ports skip the host firewall. + Through that port only `mergit_worker` can log in (`deploy/postgres/pg_hba.conf`); the `postgres` superuser cannot. +3. `docker compose -f deploy/compose.testnet.yaml -f deploy/compose.controlplane.yaml -f deploy/compose.server.yaml up -d`. +4. Make the role (prints its URL once): `deploy/worker-vm/create-db-role.sh`. It needs `cp_postgres` up and the + control plane's tables already created, which is why it comes after step 3, not before. It grants only + `jobs`, `pr_requests` (read and update) and `plans`, `tasks` (read). No sessions, users, wallets or email + addresses. +5. DNS: an A record for `api` to the server (it exists today). The Hetzner firewall stays 22, 80, 443; the private network is not filtered by it. + +## 2. On the worker VM + +```bash +git clone https://github.com/mergit-io/mergit.git && cd mergit/deploy/worker-vm +sudo install -d -o "$USER" /var/lib/mergit/workspaces +docker build -t mergit-sandbox:dev ../../control-plane/sandbox +mkdir deployments && scp server:mergit/mergit-contracts/deployments/{base-sepolia,agents-base-sepolia}.json deployments/ +cp worker.env.example worker.env && chmod 600 worker.env # fill it in: names only in the example, values never committed +docker compose build && ./preflight.sh # every line must say ok +docker compose up -d && docker compose logs -f +``` + +Expected in the log: `failed N jobs`, `reclaimed N jobs`, `removed N stale sandbox containers`, `removed N abandoned workspace directories`, then it polls. + +## 3. Cutting over from the worker on the main server + +The main server still runs a worker in `compose.controlplane.yaml`. Move in this order, so there is never a gap: + +1. Start the VM worker (step 2) and confirm it claims a job: `docker compose logs worker | grep claimed`. +2. Then, on the server: `docker compose -f deploy/compose.testnet.yaml -f deploy/compose.controlplane.yaml rm -sf worker`. +3. A follow-up change deletes the `worker` service from `compose.controlplane.yaml` so `update.sh` stops recreating it. Until then a push to + `main` brings the old worker back: run step 2 again, or remove it first. + +Two workers on one queue are safe (`FOR UPDATE SKIP LOCKED`); they must have different `WORKER_ID`s. + +## Things that go wrong + +| Symptom | Cause | +| --- | --- | +| `another worker process is already running as 'x'` | Two processes share a `WORKER_ID`. Each id is one process. A killed worker never blocks its own restart. | +| `REQUIRE_ALL_AGENT_KEYS is set and these are missing` | `agents-base-sepolia.json` lists an agent with no `MERGIT_AGENT_KEY_`. Add it, or the requester loses an attempt on that agent. | +| `WORKER_ID is not set` | The hostname is the container id and changes on every recreate, so the previous run's jobs and sandboxes are not found. Set it. | +| Live steps missing in the app, jobs still finish | The API rejected `/v0/internal/progress`: `WORKER_VM_IP` is wrong or unset. Progress is best-effort. | +| `database refused` in preflight | `CP_DB_BIND` unset, wrong private address, or the password: rerun `create-db-role.sh`. | +| A deploy killed a running job | The job is `error` and the next agent takes over. `stop_grace_period` is 40 minutes so `docker compose stop` and `down` wait for the job. A host reboot does not: drain first (`docker compose stop`), then reboot. | +| Disk filling | `docker system df` and `du -sh /var/lib/mergit/workspaces`. Old workspaces are swept at every start; images are not: `docker image prune`. | + +A dead worker's jobs are reclaimed only when a worker next starts (lease: one hour). Watch that the VM is up: +`docker compose ps` should say `Up`, and `docker compose logs --since 10m` should show polling or claims. diff --git a/deploy/worker-vm/compose.yaml b/deploy/worker-vm/compose.yaml new file mode 100644 index 0000000..7560764 --- /dev/null +++ b/deploy/worker-vm/compose.yaml @@ -0,0 +1,45 @@ +# The agent worker on its own VM (plan task O10). Runbook: deploy/worker-vm/README.md. +# cd deploy/worker-vm && ./preflight.sh && docker compose up -d --build +# The build context is the repository root (mergit_receipt is a path dependency on ../runtime/py), so run this from +# a full checkout of this repository, not a copy of this folder. +# +# Why its own machine: the worker starts sandbox containers through the host's Docker socket, and the sandboxes run +# code a model wrote. Whoever controls this process controls this host. Here that is a VM with nothing else on it: +# no database, no gateway key, no treasury key. What it holds is its agent keys, model keys and one restricted +# database role (db-role.sql), all in worker.env, which is copied by hand and never committed. + +services: + worker: + build: { context: ../.., dockerfile: control-plane/worker.Dockerfile } + image: mergit-worker:local + restart: unless-stopped + init: true + env_file: worker.env + environment: + REQUIRE_ALL_AGENT_KEYS: "true" + GATEWAY_URL: https://gateway.mergit.site + STORE_URL: https://store.mergit.site + CONTROL_PLANE_URL: https://api.mergit.site + DEPLOYMENT_FILE: /deployments/base-sepolia.json + AGENTS_FILE: /agents.json + SANDBOX_IMAGE: mergit-sandbox:dev + SANDBOX_WORKSPACE_ROOT: /var/lib/mergit/workspaces + # A job runs for up to job_deadline_s (25 minutes) plus checking and publishing. On SIGTERM the worker finishes + # the job it is on and then exits, so give it longer than any job. Docker's default of 10 seconds would SIGKILL + # it mid-job: the job stays `running` until its lease expires and the requester loses an attempt. + stop_grace_period: 40m + stop_signal: SIGTERM + security_opt: ["no-new-privileges:true"] + # A worker that logs a lot must not fill the VM's disk. + logging: + driver: json-file + options: { max-size: "20m", max-file: "5" } + volumes: + - /var/run/docker.sock:/var/run/docker.sock + # The identical path on the host and in the container: sandboxes are started through the host daemon, which + # resolves this bind source itself (control-plane/README.md). + - /var/lib/mergit/workspaces:/var/lib/mergit/workspaces + # Copied from mergit-contracts/deployments by hand (README step 4). Missing files make Docker create empty + # DIRECTORIES here, so preflight.sh checks they are real files first. + - ./deployments/base-sepolia.json:/deployments/base-sepolia.json:ro + - ./deployments/agents-base-sepolia.json:/agents.json:ro diff --git a/deploy/worker-vm/create-db-role.sh b/deploy/worker-vm/create-db-role.sh new file mode 100755 index 0000000..389ba25 --- /dev/null +++ b/deploy/worker-vm/create-db-role.sh @@ -0,0 +1,33 @@ +#!/usr/bin/env bash +# Creates the worker VM's database role (mergit_worker) with only the rights in db-role.sql, or resets its password. +# deploy/worker-vm/create-db-role.sh # makes a password and prints it once +# WORKER_DB_PASSWORD=... deploy/worker-vm/create-db-role.sh # uses one you chose (A-Z a-z 0-9 . _ ~ -, 24+ long) +# Run it on the server that runs cp_postgres, after the control plane has started once (it creates the tables). +# Safe to run again: it re-sets the password and takes back any right that is not in db-role.sql. +# PSQL replaces how psql is reached, for testing this script against a throwaway database. +set -euo pipefail +cd "$(dirname "$0")/../.." + +pw=${WORKER_DB_PASSWORD:-$(head -c 24 /dev/urandom | od -An -tx1 | tr -d ' \n')} +[[ $pw =~ ^[A-Za-z0-9._~-]{24,}$ ]] || { echo "WORKER_DB_PASSWORD must be 24 or more of A-Z a-z 0-9 . _ ~ -" >&2; exit 1; } + +if [ -n "${PSQL:-}" ]; then + read -ra psql <<<"$PSQL" +else + psql=(docker compose -f deploy/compose.testnet.yaml -f deploy/compose.controlplane.yaml + exec -T cp_postgres psql -U postgres -d mergit_cp -v ON_ERROR_STOP=1 -q) +fi + +# The password travels on stdin, never on a command line, where `ps` would show it. +{ + printf "\\\\set pw '%s'\n" "$pw" + cat <<'SQL' +SELECT format('CREATE ROLE mergit_worker LOGIN PASSWORD %L', :'pw') + WHERE NOT EXISTS (SELECT FROM pg_roles WHERE rolname = 'mergit_worker') \gexec +SELECT format('ALTER ROLE mergit_worker LOGIN PASSWORD %L', :'pw') \gexec +SQL + cat deploy/worker-vm/db-role.sql +} | "${psql[@]}" + +echo "role mergit_worker is ready. Put this in the worker VM's env file (use the private address, not a public one):" +echo "DATABASE_URL=postgres://mergit_worker:${pw}@:5433/mergit_cp?sslmode=prefer" diff --git a/deploy/worker-vm/db-role.sql b/deploy/worker-vm/db-role.sql new file mode 100644 index 0000000..2765688 --- /dev/null +++ b/deploy/worker-vm/db-role.sql @@ -0,0 +1,25 @@ +-- What the worker VM's database role may do. Idempotent; run as a superuser against the control-plane database +-- AFTER the role exists (create-db-role.sh creates it and then runs this file). +-- +-- The worker reads and writes the job queue directly (mergit_worker/jobs.py) and the pull-request queue +-- (mergit_cp/pr_store.py), and reads the plan and task it is working on. Nothing else. In particular it gets no +-- access to sessions, users, identities, wallets, email_otps, oauth_states or wallet_interest: the worker VM +-- runs model-written code and holds the Docker socket, so it is the machine most likely to be compromised, and +-- a stolen worker credential must not expose sign-in sessions or email addresses. +-- +-- If a new worker query needs another table, add it here AND in tests/test_worker_role.py, which runs every +-- statement the worker issues as this role. + +DO $$ +BEGIN + EXECUTE format('GRANT CONNECT ON DATABASE %I TO mergit_worker', current_database()); +END +$$; + +GRANT USAGE ON SCHEMA public TO mergit_worker; + +-- Start from nothing, so re-running after a tightening takes the old rights away. +REVOKE ALL ON ALL TABLES IN SCHEMA public FROM mergit_worker; + +GRANT SELECT, UPDATE ON jobs, pr_requests TO mergit_worker; +GRANT SELECT ON plans, tasks TO mergit_worker; diff --git a/deploy/worker-vm/preflight.sh b/deploy/worker-vm/preflight.sh new file mode 100755 index 0000000..60a2b77 --- /dev/null +++ b/deploy/worker-vm/preflight.sh @@ -0,0 +1,71 @@ +#!/usr/bin/env bash +# Checks the worker VM is ready, before `docker compose up`. One line per check; exits non-zero if any failed. +# cd deploy/worker-vm && ./preflight.sh +# It reads worker.env as text and never runs it, and never prints a secret. +# API_URL / GATEWAY_URL / STORE_URL override the hosted addresses (for a staging stack). +set -uo pipefail +cd "$(dirname "$0")" +api=${API_URL:-https://api.mergit.site} +gateway=${GATEWAY_URL:-https://gateway.mergit.site} +store=${STORE_URL:-https://store.mergit.site} +workspaces=${WORKSPACES:-/var/lib/mergit/workspaces} +bad=0 +ok() { echo "ok: $*"; } +fail() { echo "FAIL: $*" >&2; bad=1; } +get() { grep -E "^$1=" worker.env 2>/dev/null | tail -1 | cut -d= -f2-; } + +command -v docker >/dev/null || fail "docker is not installed" +docker info >/dev/null 2>&1 && ok "the Docker daemon answers" || fail "cannot reach the Docker daemon (is this user in the docker group?)" + +if [ -f worker.env ]; then + ok "worker.env exists" + [ "$(stat -c %a worker.env 2>/dev/null)" = 600 ] && ok "worker.env is mode 600" || fail "worker.env holds secrets: chmod 600 worker.env" + for v in WORKER_ID DATABASE_URL EVENTS_TOKEN STORE_WRITE_TOKEN WORKER_TOKEN; do + [ -n "$(get $v)" ] && ok "$v is set" || fail "$v is empty or missing in worker.env" + done + case "$(get DATABASE_URL)" in *CHANGE-ME*) fail "DATABASE_URL still has the CHANGE-ME placeholder" ;; esac + case "$(get DATABASE_URL)" in *@127.0.0.1*|*@localhost*) fail "DATABASE_URL points at this machine: use the server's private address" ;; esac + [ -n "$(get ANTHROPIC_API_KEY)$(get GROQ_API_KEY)" ] && ok "a model key is set" || fail "set ANTHROPIC_API_KEY or GROQ_API_KEY" +else + fail "worker.env is missing: copy worker.env.example and fill it in" +fi + +for f in deployments/base-sepolia.json deployments/agents-base-sepolia.json; do + if [ -f "$f" ]; then ok "$f is a file"; else + fail "$f is not a file. Copy it from mergit-contracts/deployments. If a directory of that name exists, Docker made it: remove it" + fi +done + +if [ -f deployments/agents-base-sepolia.json ] && [ -f worker.env ]; then + ids=$(python3 -c 'import json,sys; print(*[str(a["agent_id"]) for a in json.load(open(sys.argv[1]))["agents"]])' \ + deployments/agents-base-sepolia.json 2>/dev/null) || ids="" + if [ -z "$ids" ]; then fail "cannot read the agent ids from deployments/agents-base-sepolia.json"; fi + for id in $ids; do + [[ "$(get "MERGIT_AGENT_KEY_$id")" =~ ^0x[0-9a-fA-F]{64}$ ]] \ + && ok "MERGIT_AGENT_KEY_$id is a 32-byte hex key" \ + || fail "MERGIT_AGENT_KEY_$id is missing or malformed (the worker refuses to start without it)" + done +fi + +docker image inspect mergit-sandbox:dev >/dev/null 2>&1 && ok "the sandbox image exists" \ + || fail "no sandbox image: docker build -t mergit-sandbox:dev ../../control-plane/sandbox" +[ -d "$workspaces" ] && [ -w "$workspaces" ] && ok "$workspaces exists and is writable" \ + || fail "$workspaces is missing or not writable: sudo install -d -o \$USER $workspaces" +free_gb=$(df --output=avail -BG "$workspaces" 2>/dev/null | tail -1 | tr -dc 0-9) +[ "${free_gb:-0}" -ge 10 ] && ok "${free_gb} GB free for checkouts" || fail "under 10 GB free where $workspaces lives (${free_gb:-?} GB)" + +for pair in "api:$api" "gateway:$gateway" "store:$store"; do + name=${pair%%:*}; url=${pair#*:} + curl -fsS -m 8 -o /dev/null "$url/healthz" 2>/dev/null && ok "$name answers at $url/healthz" || fail "$name did not answer at $url/healthz" +done + +if docker image inspect mergit-worker:local >/dev/null 2>&1 && [ -f worker.env ]; then + docker run --rm --env-file worker.env --entrypoint python mergit-worker:local -c \ + "import os, psycopg; psycopg.connect(os.environ['DATABASE_URL'], connect_timeout=8).close()" >/dev/null 2>&1 \ + && ok "the database accepts this worker's credentials" \ + || fail "the database refused or did not answer (private address, firewall, password, CP_DB_BIND on the server?)" +else + echo "skip: the database check needs the worker image (docker compose build), run this again after it" +fi + +[ "$bad" = 0 ] && echo "ready." || { echo "not ready: fix the FAIL lines above." >&2; exit 1; } diff --git a/deploy/worker-vm/worker.env.example b/deploy/worker-vm/worker.env.example new file mode 100644 index 0000000..6066885 --- /dev/null +++ b/deploy/worker-vm/worker.env.example @@ -0,0 +1,30 @@ +# Copy to worker.env on the worker VM (chmod 600) and fill in. Never commit worker.env. +# Names only here; the values come from the server's deploy/secrets/*.env and from your model provider. + +# A fixed name, ONE PER PROCESS. The worker finds, fails and cleans up what its previous run left by this name, and +# refuses to start if another process already holds it. Never reuse a name for a second worker on the same database. +WORKER_ID=worker-vm-1 + +# Printed once by deploy/worker-vm/create-db-role.sh on the server. Use the server's PRIVATE address. +DATABASE_URL=postgres://mergit_worker:CHANGE-ME@10.0.0.2:5433/mergit_cp?sslmode=prefer + +# Same values as the server's deploy/secrets/worker.env (run `grep -E '^(EVENTS|STORE_WRITE|WORKER)_TOKEN' there`). +EVENTS_TOKEN= +STORE_WRITE_TOKEN= +WORKER_TOKEN= + +# One key per agent in agents-base-sepolia.json, 0x plus 64 hex characters. The worker refuses to start without +# every one (REQUIRE_ALL_AGENT_KEYS in compose.yaml). From mergit-contracts/.env.services. +MERGIT_AGENT_KEY_1= + +# A model key: Anthropic's, Groq's, or both. Anthropic is used when its key is present. +ANTHROPIC_API_KEY= +GROQ_API_KEY= +# GROQ_MODEL=openai/gpt-oss-120b +# LLM_PROVIDER=anthropic + +# Optional. Only the pull-request gate needs it; without it the worker still runs tasks. +# GITHUB_TOKEN= + +# Optional. gVisor, if runsc is installed on this VM. Untested here: leave unset unless you have tried it. +# SANDBOX_RUNTIME=runsc