From bdbac4e64b8433aa5b90e2b6f5c5462eeac3ec94 Mon Sep 17 00:00:00 2001 From: Paulo Date: Mon, 7 Sep 2026 20:34:18 +0200 Subject: [PATCH] Release orphan boxes hourly and report a missing secrets exchange (DRU-477) A run that dies without its cleanup leaves its box until the lease ends. An hourly task releases such a box sooner, and the release revokes the identity. The records themselves need no reaping: the issuer already denies an identity whose run ended or whose lease passed. druks doctor probes the secrets exchange and names the fix when it is down. --- backend/druks/core/tasks.py | 27 ++++++++--- backend/druks/doctor.py | 27 +++++++++++ backend/druks/sandbox/models.py | 23 +++++++-- backend/tests/test_doctor.py | 61 +++++++++++++++++++++++ backend/tests/test_orphan_boxes.py | 78 ++++++++++++++++++++++++++++++ docs/concepts.md | 3 +- docs/deployment.md | 7 +++ docs/troubleshooting.md | 13 +++++ 8 files changed, 225 insertions(+), 14 deletions(-) create mode 100644 backend/tests/test_orphan_boxes.py diff --git a/backend/druks/core/tasks.py b/backend/druks/core/tasks.py index 02f899c4..cc74bc2e 100644 --- a/backend/druks/core/tasks.py +++ b/backend/druks/core/tasks.py @@ -5,6 +5,8 @@ from druks.harnesses.directory import refresh_added_catalogs from druks.harnesses.providers import get_provider, get_providers from druks.sandbox import gate +from druks.sandbox.client import sandbox_client +from druks.sandbox.models import SandboxIdentity from druks.secrets.models import VaultSecret from druks.workflows import task @@ -20,12 +22,23 @@ async def reap_deleted_files() -> None: @task(every="*/15 * * * *") async def refresh_tokens() -> None: - # Every 15 min. With an ~8h Claude TTL refreshed at <2h remaining (and - # codex ~10d at <24h), this keeps both tokens alive with a wide margin - # while doing almost nothing on most ticks. + # Fifteen minutes keeps every token inside its refresh margin and does + # almost nothing on most ticks. await _refresh() +@task(every="0 * * * *") +async def release_orphan_boxes() -> None: + await _release_orphan_boxes() + + +async def _release_orphan_boxes() -> None: + # A run that died without its cleanup leaves its box until the lease ends. + for identity in await SandboxIdentity.list_orphans(): + logger.info("releasing the orphan box %s of run %s", identity.host_id, identity.run_id) + await sandbox_client.release(host_id=identity.host_id) + + @task(every="0 6 * * *") async def refresh_catalogs() -> None: for provider in get_providers(): @@ -36,10 +49,9 @@ async def refresh_catalogs() -> None: async def _refresh() -> dict[str, object]: subscriptions = await VaultSecret.list_subscriptions() - # A rotation ends the token every box holds, so a due rotation runs only - # while its subscription is idle, or once urgent. rotate_token no-ops - # outside the margin and requests a refresh for every live box. Snapshot - # plain values: each refresh commits and expires the session's ORM objects. + # A rotation ends the token every box holds, so a due one waits for an + # idle subscription unless it is urgent. Plain values: each refresh + # commits and expires the session's rows. rows = [ ( subscription.audience_name, @@ -106,4 +118,3 @@ def _log_result(result: RotationResult) -> None: result.provider, result.subscription_id, ) - # "fresh" and "locked" (another worker owns this row's refresh) are quiet no-ops. diff --git a/backend/druks/doctor.py b/backend/druks/doctor.py index a3efe94a..64246a8f 100644 --- a/backend/druks/doctor.py +++ b/backend/druks/doctor.py @@ -303,6 +303,32 @@ async def _drukbox_doctor(settings: Settings): await api.aclose() +async def check_secrets_exchange(settings: Settings) -> CheckResult: + if not settings.sandbox.service_url: + return CheckResult( + name="secrets_exchange", ok=True, detail="not configured (sandbox execution is off)" + ) + url = f"{settings.sandbox.exchange_url.rstrip('/')}/healthz" + try: + async with httpx.AsyncClient(timeout=3.0) as http: + status = (await http.get(url)).status_code + except httpx.HTTPError as error: + return CheckResult( + name="secrets_exchange", + ok=False, + detail=f"drukbox-exchange is unreachable at {url}: {error}. " + "Start it: docker compose up -d drukbox-exchange", + ) + if status != 200: + return CheckResult( + name="secrets_exchange", + ok=False, + detail=f"drukbox-exchange answered {status} at {url}. " + "Read its log: docker compose logs drukbox-exchange", + ) + return CheckResult(name="secrets_exchange", ok=True, detail=url) + + async def check_sandbox_e2e(settings: Settings) -> CheckResult | list[CheckResult]: """Provision a real VM, exercise the acquire and reattach dial paths, and probe each registered harness CLI's presence on the image. Costs one @@ -570,6 +596,7 @@ async def _run_app_check(app_name: str, check) -> CheckResult: check_database, check_redis, check_drukbox, + check_secrets_exchange, check_capability_modules, check_apps, check_declared_sandboxes, diff --git a/backend/druks/sandbox/models.py b/backend/druks/sandbox/models.py index 76140f5a..7906de97 100644 --- a/backend/druks/sandbox/models.py +++ b/backend/druks/sandbox/models.py @@ -88,7 +88,6 @@ async def create( identity = cls( run_id=run_id, scoped_to=scoped_to, - # Own rows: the caller's values stay values. secret_refs=[ SecretRef( name=ref.name, @@ -110,9 +109,8 @@ async def create( entries = {} for ref in secret_refs: secret = await VaultSecret.get(ref.secret_id) - # A custom entry binds the host the proxy swaps at, the variable - # the box exports, and the header the value fills. A catalog entry - # leaves those to Drukbox. + # A custom entry names its host, variable, and header; a catalog + # entry leaves those to Drukbox. fields = {} if ref.host: header = secret.header or BEARER_HEADER @@ -188,9 +186,24 @@ async def bind(self, host_id: str) -> None: await db_session().commit() async def revoke(self) -> None: - self.revoked_at = Base.utc_now() + # A second revoke keeps the first stamp. + self.revoked_at = self.revoked_at or Base.utc_now() await db_session().commit() + @classmethod + async def list_orphans(cls) -> list["SandboxIdentity"]: + """The identities of boxes whose run ended: bound, inside their lease, not revoked.""" + rows = await db_session().scalars( + select(cls) + .options(selectinload(cls.run)) + .where( + cls.host_id.is_not(None), + cls.revoked_at.is_(None), + cls.expires_at > Base.utc_now(), + ) + ) + return [identity for identity in rows if not identity.run.is_active] + @classmethod async def revoke_for_host(cls, engine, host_id: str) -> None: # Own transaction: a box release can run outside a step session. diff --git a/backend/tests/test_doctor.py b/backend/tests/test_doctor.py index 62917d7e..024cd7c6 100644 --- a/backend/tests/test_doctor.py +++ b/backend/tests/test_doctor.py @@ -198,6 +198,66 @@ async def test_drukbox_passes_when_unconfigured(tmp_path: Path) -> None: assert "not configured" in result.detail +_SANDBOX = {"service_url": "https://sb.test", "service_token": "t"} + + +def _exchange(monkeypatch, *, status: int = 200, error: Exception | None = None) -> list[str]: + # The exchange as the probe sees it: a status, or a connection error. + asked: list[str] = [] + + class FakeClient: + def __init__(self, **_kwargs) -> None: + pass + + async def __aenter__(self): + return self + + async def __aexit__(self, *_exc) -> None: + pass + + async def get(self, url: str): + asked.append(url) + if error: + raise error + return SimpleNamespace(status_code=status) + + monkeypatch.setattr(doctor.httpx, "AsyncClient", FakeClient) + return asked + + +async def test_secrets_exchange_is_not_probed_when_sandbox_execution_is_off( + tmp_path: Path, monkeypatch +) -> None: + asked = _exchange(monkeypatch) + + result = await doctor.check_secrets_exchange(make_settings(tmp_path)) + + assert result.ok + assert "not configured" in result.detail + assert asked == [] + + +async def test_secrets_exchange_reports_a_healthy_exchange(tmp_path: Path, monkeypatch) -> None: + _exchange(monkeypatch, status=200) + + result = await doctor.check_secrets_exchange(make_settings(tmp_path, sandbox=_SANDBOX)) + + assert result.ok + assert result.detail == "http://127.0.0.1:8781/healthz" + + +async def test_secrets_exchange_names_the_service_and_the_fix_when_unreachable( + tmp_path: Path, monkeypatch +) -> None: + _exchange(monkeypatch, error=httpx.ConnectError("refused")) + + result = await doctor.check_secrets_exchange(make_settings(tmp_path, sandbox=_SANDBOX)) + + assert not result.ok + assert "drukbox-exchange is unreachable at http://127.0.0.1:8781/healthz" in result.detail + assert "docker compose up -d drukbox-exchange" in result.detail + + async def test_declared_sandboxes_pass_when_none_are_declared( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, @@ -300,6 +360,7 @@ async def test_run_checks_covers_all_check_names(tmp_path: Path) -> None: "database", "redis", "drukbox", + "secrets_exchange", "capability_modules", } diff --git a/backend/tests/test_orphan_boxes.py b/backend/tests/test_orphan_boxes.py new file mode 100644 index 00000000..1dbf1ab9 --- /dev/null +++ b/backend/tests/test_orphan_boxes.py @@ -0,0 +1,78 @@ +from datetime import UTC, datetime, timedelta +from types import SimpleNamespace + +from conftest import connect_provider +from druks.core import tasks +from druks.database import db_session +from druks.durable.engine import _step_engine +from druks.harnesses.providers import AnthropicProvider +from druks.sandbox.models import SandboxIdentity, SecretRef +from druks.testing import seed_run +from druks_field_notes.workflows import Summarize + + +async def _identity(run_id: str, *, state: str = "running", host_id: str = "") -> SandboxIdentity: + await seed_run(db_session(), kind=Summarize.kind, run_id=run_id, state=state) + subscription = await connect_provider( + AnthropicProvider, {"claudeAiOauth": {"accessToken": "test-token"}} + ) + identity, _ = await SandboxIdentity.create( + run_id=run_id, + scoped_to="workflow", + secret_refs=[SecretRef(name="anthropic", secret_id=subscription.id)], + ) + if host_id: + await identity.bind(host_id) + return identity + + +def _drukbox(monkeypatch) -> list[str]: + # A release ends the identity first, as the real client does. + released: list[str] = [] + + async def release(*, host_id: str): + released.append(host_id) + await SandboxIdentity.revoke_for_host(_step_engine(), host_id) + + monkeypatch.setattr(tasks, "sandbox_client", SimpleNamespace(release=release)) + return released + + +async def test_a_finished_runs_box_is_released_and_a_live_runs_box_stays(druks_db, monkeypatch): + await _identity("run-1", state="finished", host_id="host-dead") + await _identity("run-2", host_id="host-live") + released = _drukbox(monkeypatch) + + await tasks._release_orphan_boxes() + + assert released == ["host-dead"] + + +async def test_a_box_past_its_lease_is_left_to_drukbox(druks_db, monkeypatch): + identity = await _identity("run-1", state="finished", host_id="host-old") + identity.expires_at = datetime.now(UTC) - timedelta(minutes=1) + await db_session().commit() + released = _drukbox(monkeypatch) + + await tasks._release_orphan_boxes() + + assert released == [] + + +async def test_an_identity_without_a_box_has_nothing_to_release(druks_db, monkeypatch): + await _identity("run-1", state="finished") + released = _drukbox(monkeypatch) + + await tasks._release_orphan_boxes() + + assert released == [] + + +async def test_a_second_tick_finds_nothing_to_do(druks_db, monkeypatch): + await _identity("run-1", state="finished", host_id="host-dead") + released = _drukbox(monkeypatch) + + await tasks._release_orphan_boxes() + await tasks._release_orphan_boxes() + + assert released == ["host-dead"] diff --git a/docs/concepts.md b/docs/concepts.md index 64917eb4..00dd0544 100644 --- a/docs/concepts.md +++ b/docs/concepts.md @@ -200,7 +200,8 @@ subscription is idle or the token is urgent. One rotator runs at a time, and new calls wait for it. After a rotation, Druks requests a refresh from the secrets exchange for every live sandbox on that subscription. A provider can revoke the previous token at the rotation. Druks revokes the identity when it -releases the sandbox, and a terminal run denies every fetch. The identity +releases the sandbox, and a terminal run denies every fetch. An hourly task +releases the sandbox of a run that ended without its cleanup. The identity expires with the sandbox lease. ## Events, signals, webhooks, and subjects diff --git a/docs/deployment.md b/docs/deployment.md index 5749f5d2..24e47612 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -212,6 +212,13 @@ identity bearer and nothing else, and it answers `value` and the provider's `POST /refresh//` on `[sandbox].exchange_url`, one request for each live sandbox on the subscription. +An identity dies with its sandbox. Druks revokes it before it deletes the +sandbox, and the issuer denies a terminal run's identity before any cleanup. +A run that dies without its cleanup leaves its sandbox until the lease ends. +The `release_orphan_boxes` task runs every hour and releases such a sandbox +sooner. Drukbox reaps a sandbox at the end of its lease in any case. `druks doctor` probes the exchange at +`[sandbox].exchange_url` on `/healthz` and names the corrective action. + Drukbox encrypts the secret entries of each sandbox with `SECRETS_KEY`. The installer generates `[secrets].drukbox_secrets_key` and renders it as `SECRETS_KEY` for the API and the exchange. Pin the proxy image with diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 13650985..01a58694 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -160,6 +160,19 @@ Make sure that the service listens at `[sandbox].service_url` in `druks.toml`. For remote providers, a healthy Drukbox API does not prove SSH access. Then follow with `druks doctor --sandbox`. +### The secrets exchange is unreachable + +`druks doctor` reports `secrets_exchange` with the URL it probed. Run: + +```bash +docker compose up -d drukbox-exchange +docker compose logs --tail=200 drukbox-exchange +``` + +The exchange binds `127.0.0.1:8781` on the Druks host. Until it answers, a +sandbox gets no value for its placeholders, and each agent call fails at the +provider with an authentication error. + ### A sandbox process appears stuck Druks copies the dashboard transcript from files that a detached VM process writes.