ctfy.sdk.node_client

   1"""HTTP client for platform → node machine calls.
   2
   3Every call is authenticated with the target node's bearer token —
   4minted per node at registration and used in both directions. End users
   5never touch this client; they talk to the platform, which proxies to
   6the node. Node endpoints are defined in
   7``server/routes/node_instances.py``.
   8"""
   9
  10from __future__ import annotations
  11
  12import base64
  13import dataclasses
  14import json
  15from typing import Any, TypeVar
  16
  17import httpx
  18from pydantic import BaseModel, Field, ValidationError
  19
  20from ctfy.challenge.deception import DeceptionSummary
  21from ctfy.core.awd_network import MatchNetwork
  22from ctfy.core.challenge import ChallengeReloadResult
  23from ctfy.core.constants import DEFAULT_CLIENT_TIMEOUT
  24from ctfy.core.exceptions import NodeRequestError
  25from ctfy.core.failure_text import tail
  26from ctfy.core.instance_disk_retention import (
  27    CAPTURE_DROPPED_BYTES_HEADER,
  28    CAPTURE_DROPPED_SEGMENTS_HEADER,
  29)
  30from ctfy.core.log import log
  31from ctfy.core.observed_contact import ContactReading
  32from ctfy.core.prebuild_tag import ChallengeFreezeResult
  33from ctfy.sdk.base import BaseHttpClient
  34from ctfy.server.models.challenge import AttachmentList
  35from ctfy.server.models.instance import InstanceTrafficPage, InstanceTrafficWitnesses
  36from ctfy.server.models.node import (
  37    AgentRuntime,
  38    AwdAttackResult,
  39    BuildAllQueued,
  40    BuildStateRows,
  41    CheckResult,
  42    ComposeContainer,
  43    FabricProbeReport,
  44    FabricStatus,
  45    HarnessLimits,
  46    HarnessRun,
  47    InstanceStarted,
  48    InstanceStatusReport,
  49    NodeAuthoredPush,
  50    NodeAuthoredResult,
  51    NodeAuthoredSmokePush,
  52    NodeAuthoredSmokeResult,
  53    NodeDiskFree,
  54    NodeInstanceList,
  55    NodeInstanceSummary,
  56    NodeRangePush,
  57    NodeRangeResult,
  58    NodeRangeVerifyPush,
  59    NodeRangeVerifyResult,
  60    PatchVerifyResult,
  61    PullAllQueued,
  62    PullStateRows,
  63)
  64from ctfy.server.node_state import ChallengeBuildState, ChallengePullState
  65from ctfy.server.traffic_tail import coerce_page
  66
  67
  68@dataclasses.dataclass(frozen=True)
  69class PcapCapture:
  70    """A node's merged capture, and what the cap dropped from it (#1714)."""
  71
  72    data: bytes
  73    #: ``None`` = the node did not say (older build); ``0`` = nothing.
  74    dropped_bytes: int | None = None
  75    dropped_segments: int | None = None
  76
  77
  78def _header_int(resp: httpx.Response, name: str) -> int | None:
  79    raw = resp.headers.get(name)
  80    if raw is None:
  81        return None
  82    try:
  83        return max(0, int(raw))
  84    except ValueError:
  85        return None
  86
  87
  88def _raise_for_node_status(resp: httpx.Response) -> None:
  89    """Like ``resp.raise_for_status()`` but preserves the node's error body.
  90
  91    On a 4xx/5xx, pull the FastAPI ``{"detail": ...}`` field (or the raw
  92    body as a fallback) and raise :class:`NodeRequestError` so the real
  93    failure reaches the caller instead of httpx's generic status string.
  94    """
  95    if not resp.is_error:
  96        return
  97    detail = ""
  98    code = ""
  99    try:
 100        body = resp.json()
 101    except (json.JSONDecodeError, ValueError):
 102        body = None
 103    if isinstance(body, dict):
 104        d = body.get("detail")
 105        if isinstance(d, str):
 106            detail = d
 107        elif isinstance(d, dict) and isinstance(d.get("code"), str):
 108            # A node naming its failure (#1835): keep the name as a field
 109            # and the sentence as the detail, so neither a reader of the
 110            # code nor a human reading the detail meets JSON.
 111            code = str(d["code"])
 112            message = d.get("message")
 113            detail = message if isinstance(message, str) else json.dumps(d)
 114        elif d is not None:
 115            detail = json.dumps(d)
 116    if not detail:
 117        detail = (resp.text or "").strip()
 118    raise NodeRequestError(resp.status_code, detail, code=code)
 119
 120
 121_M = TypeVar("_M", bound=BaseModel)
 122
 123
 124def _parsed(model: type[_M], resp: httpx.Response) -> _M:
 125    """The node's answer, as the model the node itself declares.
 126
 127    ⚠️ **A parse failure raises rather than defaulting.** Its neighbour
 128    ``instance_health`` falls back to an empty ``DeceptionSummary``, and
 129    that is right there: the summary is decoration on a liveness answer
 130    the caller already has. These carry the *verdict* — whether a checker
 131    passed, what an attack read — and a default is a verdict nobody
 132    reached. The checker's own fail direction is closed, and an answer
 133    this client cannot parse is not an answer.
 134
 135    ⚠️ Unknown fields are ignored (pydantic's default), which is the
 136    forward-compatible half: the node and the platform are deployed
 137    separately, ``ctfy/core/version.py`` exists because they skew, and a
 138    newer node sending a field this platform has never heard of must not
 139    fail the call.
 140    """
 141    try:
 142        return model.model_validate(resp.json())
 143    except (ValidationError, ValueError) as exc:
 144        # ⚠️ ``resp.status_code``, not a synthetic one: the node
 145        # *answered*, and the failure is that its body does not fit the
 146        # contract. Reporting a 502 here would say the node was
 147        # unreachable, which sends whoever reads it to the wrong host.
 148        raise NodeRequestError(
 149            resp.status_code,
 150            f"answered something this platform cannot read as {model.__name__}: {exc}",
 151        ) from exc
 152
 153
 154def _parsed_list(model: type[_M], resp: httpx.Response) -> list[_M]:
 155    """A JSON array of *model*, with the same failure direction as
 156    :func:`_parsed` — one unreadable element fails the call rather than
 157    being dropped, because a partial list read as a whole one is how a
 158    re-adoption pass decides an instance is gone."""
 159    body = resp.json()
 160    if not isinstance(body, list):
 161        raise NodeRequestError(
 162            resp.status_code, f"expected a list of {model.__name__}, got {type(body).__name__}"
 163        )
 164    try:
 165        return [model.model_validate(item) for item in body]
 166    except (ValidationError, ValueError) as exc:
 167        raise NodeRequestError(
 168            resp.status_code,
 169            f"answered something this platform cannot read as a list of {model.__name__}: {exc}",
 170        ) from exc
 171
 172
 173class HealthReading(BaseModel):
 174    """One health poll's three answers.
 175
 176    ⚠️ **A record rather than a tuple, and the reason is recorded one
 177    subsystem over.** ``competition_results`` lost ``attempts`` and
 178    ``last_solve_at`` to a hand-made ``(rank, flags, solved)`` tuple
 179    sitting between a board that produced them and a template that
 180    consumed them; this poll had two answers and now has three, so the
 181    tuple was one field away from the same fate. Growing a record is a
 182    field; growing a tuple is every call site, silently re-bound.
 183    """
 184
 185    healthy: bool = False
 186    deception: DeceptionSummary = Field(default_factory=DeceptionSummary)
 187    contact: ContactReading = Field(default_factory=ContactReading)
 188    new_processes: dict[str, list[str]] = Field(default_factory=dict)
 189
 190
 191class NodeClient(BaseHttpClient):
 192    """Thin sync HTTP client; one instance per node URL."""
 193
 194    def __init__(
 195        self,
 196        node_url: str,
 197        token: str,
 198        *,
 199        timeout: int = DEFAULT_CLIENT_TIMEOUT,
 200        transport: httpx.BaseTransport | None = None,
 201    ) -> None:
 202        """*transport* is `BaseHttpClient`'s own seam, passed through so a
 203        test can drive a real node app over the real client — the hop the
 204        fakes on either side of it cannot see."""
 205        super().__init__(
 206            f"{node_url.rstrip('/')}/api/v1", token, timeout=timeout, transport=transport
 207        )
 208
 209    # -- lifecycle ----------------------------------------------------------
 210
 211    def start_instance(
 212        self,
 213        *,
 214        challenge_id: str,
 215        instance_id: str,
 216        ttl: int,
 217        answers: dict[str, str],
 218        proxy_output_dir: str | None = None,
 219        env: dict[str, str] | None = None,
 220        publish_gamebox: bool = False,
 221        match_network: MatchNetwork | None = None,
 222        open_shell_password: str = "",
 223        authored_revision: str = "",
 224    ) -> InstanceStarted:
 225        resp = self.request(
 226            "POST",
 227            "/instances",
 228            json={
 229                # #1835: the authored tree's revision this start must run;
 230                # a node holding another (or none) fetches it first.
 231                "authored_revision": authored_revision,
 232                "challenge_id": challenge_id,
 233                "instance_id": instance_id,
 234                "ttl": ttl,
 235                "answers": answers,
 236                "proxy_output_dir": proxy_output_dir,
 237                "env": env or {},
 238                "publish_gamebox": publish_gamebox,
 239                # ``None`` rather than an omitted key: the node's model
 240                # defaults to None either way, and sending it explicitly
 241                # keeps the body's shape stable across both branches.
 242                "match_network": (dataclasses.asdict(match_network) if match_network else None),
 243                # Classic-AWD open shell (#1725): the box's starting root
 244                # password, empty for every managed / non-AWD start.
 245                "open_shell_password": open_shell_password,
 246            },
 247        )
 248        _raise_for_node_status(resp)
 249        return _parsed(InstanceStarted, resp)
 250
 251    def stop_instance(self, instance_id: str) -> None:
 252        resp = self.request("DELETE", f"/instances/{instance_id}")
 253        resp.raise_for_status()
 254
 255    def apply_fabric(
 256        self,
 257        *,
 258        index: int,
 259        interface: dict[str, Any],
 260        match_network: str = "",
 261    ) -> FabricStatus:
 262        """Put this node on one match's WireGuard fabric.
 263
 264        ⚠️ **Returns the model, never ``resp.json()``.** The platform
 265        reads the verdict field by field, and on a dict a renamed field
 266        is ``None`` — falsy — which reads as "the arena did not open",
 267        so a wire change would quietly report every node as broken.
 268
 269        Returns the node's verdict rather than raising on a refusal: the
 270        route answers **200 with a body** precisely so that "this node is
 271        unreachable" and "this node rejected the plan" stay different
 272        facts, and a caller fanning out over an arena needs to tell them
 273        apart — one is waited on, the other is looked at.
 274
 275        *match_network* lets the node check the tunnel landed in the
 276        namespace its boxes are in, and open the bridge's one hole. It is
 277        the platform's own fact (`awd_network.network_name`), so the node
 278        is told rather than guessing.
 279        """
 280        resp = self.request(
 281            "POST",
 282            "/fabric",
 283            json={
 284                "index": index,
 285                "interface": interface,
 286                "match_network": match_network,
 287            },
 288        )
 289        resp.raise_for_status()
 290        return _parsed(FabricStatus, resp)
 291
 292    def probe_fabric(
 293        self,
 294        *,
 295        index: int,
 296        peers: list[dict[str, str]],
 297        mtu: int = 0,
 298    ) -> FabricProbeReport:
 299        """Ask named peers the three link questions. Applies nothing.
 300
 301        ⚠️ **Returns the model, never ``resp.json()``** — same reason
 302        `apply_fabric` gives: on a dict a renamed field reads as falsy,
 303        which here means "your tunnel is down" about a wire change.
 304        """
 305        resp = self.request(
 306            "POST",
 307            "/fabric/probe",
 308            json={"index": index, "peers": peers, "mtu": mtu},
 309        )
 310        resp.raise_for_status()
 311        return _parsed(FabricProbeReport, resp)
 312
 313    def remove_fabric(self, *, index: int, match_network: str = "") -> FabricStatus:
 314        """Take this node off one match's fabric and close its hole.
 315
 316        ⚠️ *match_network* is what lets the node resolve the bridge and
 317        delete the ``DOCKER-USER`` rules by name. Without it the tunnel
 318        goes and the hole stays — open on whatever docker gives that
 319        bridge name next.
 320        """
 321        resp = self.request(
 322            "DELETE",
 323            f"/fabric/{index}",
 324            params={"match_network": match_network},
 325        )
 326        resp.raise_for_status()
 327        return _parsed(FabricStatus, resp)
 328
 329    def stop_instance_service(self, instance_id: str, service: str) -> int:
 330        """Stop one compose service, leaving the instance running.
 331
 332        The exercise format's session close (§4b.4). Returns how many
 333        containers the node stopped — 0 means the challenge declares no
 334        such service, which is a real answer rather than a failure, so
 335        the caller can distinguish it from a cut tunnel.
 336        """
 337        resp = self.request(
 338            "POST",
 339            f"/instances/{instance_id}/stop-service",
 340            json={"service": service},
 341        )
 342        resp.raise_for_status()
 343        body = resp.json()
 344        return int(body.get("stopped", 0)) if isinstance(body, dict) else 0
 345
 346    def stop_all(self) -> None:
 347        resp = self.request("POST", "/admin/stop-all")
 348        resp.raise_for_status()
 349
 350    def put_authored_challenge(self, push: NodeAuthoredPush) -> NodeAuthoredResult:
 351        """Ask the node to write an approved player upload into its catalog.
 352
 353        ⚠️ **The bytes travel here, unlike `put_generated_range`.** That
 354        one sends a scenario because the node can compile it; a
 355        player-authored tree has no compiler and no corpus entry, so the
 356        archive is what there is to send.
 357
 358        The node answers **200 with ``ok: false``** for a refusal, so a
 359        node that declined the archive and a node that could not be
 360        reached stay different facts; only the second raises here.
 361        """
 362        resp = self.request("POST", "/admin/authored", json=push.model_dump(mode="json"))
 363        resp.raise_for_status()
 364        return _parsed(NodeAuthoredResult, resp)
 365
 366    def smoke_authored_challenge(
 367        self, push: NodeAuthoredSmokePush, *, timeout: int | None = None
 368    ) -> NodeAuthoredSmokeResult:
 369        """Ask the node to run an authored challenge's own reference exploit.
 370
 371        ⚠️ **Takes its own timeout.** A smoke run brings a whole estate
 372        up; the client default is sized for a status call and would give
 373        up while the node is still doing what it was asked.
 374        """
 375        kwargs: dict[str, Any] = {"json": push.model_dump(mode="json")}
 376        if timeout is not None:
 377            kwargs["timeout"] = timeout
 378        resp = self.request("POST", "/admin/authored/smoke", **kwargs)
 379        resp.raise_for_status()
 380        return _parsed(NodeAuthoredSmokeResult, resp)
 381
 382    def put_generated_range(self, push: NodeRangePush) -> NodeRangeResult:
 383        """Ask the node to build a generated range from its scenario.
 384
 385        ⚠️ **The scenario travels, not the estate.** The corpus is
 386        mounted on the node and the compiler ships inside it, so 70-odd
 387        rendered files never cross this API — and the corpus commit that
 388        rides along is what stops one scenario becoming two different
 389        estates under one id across a fleet.
 390
 391        The node answers **200 with ``ok: false``** for a refusal, so a
 392        node that declined and a node that could not be reached stay
 393        different facts; only the second raises here.
 394        """
 395        resp = self.request("POST", "/admin/ranges", json=push.model_dump(mode="json"))
 396        resp.raise_for_status()
 397        return _parsed(NodeRangeResult, resp)
 398
 399    def verify_generated_range(
 400        self, push: NodeRangeVerifyPush, *, timeout: int | None = None
 401    ) -> NodeRangeVerifyResult:
 402        """Ask the node to build a scenario, solve it, and take it down.
 403
 404        ⚠️ **The longest call in this API**, so the caller passes its own
 405        timeout: the node's budget is ``compose_timeout_s +
 406        solve_timeout_s`` and a client that used the default would give
 407        up while the node was still building — then retry, and start a
 408        second estate beside the first.
 409
 410        ⚠️ A client timeout does **not** leave a range running: the node
 411        tears down in its own ``finally``. What is lost is the answer,
 412        not the host.
 413
 414        The node answers **200 with an ``error``** for a refusal, the
 415        same split ``put_generated_range`` uses; only an unreachable node
 416        raises.
 417        """
 418        kwargs: dict[str, Any] = {"json": push.model_dump(mode="json")}
 419        if timeout is not None:
 420            # httpx takes a per-request override; the client-wide value
 421            # stays whatever the caller constructed this with.
 422            kwargs["timeout"] = timeout
 423        resp = self.request("POST", "/admin/ranges/verify", **kwargs)
 424        resp.raise_for_status()
 425        return _parsed(NodeRangeVerifyResult, resp)
 426
 427    def cancel_generated_range_verify(self, run_id: str) -> bool:
 428        """Tell the node to stop a verification it is running.
 429
 430        ⚠️ **A miss is success, not a 404 to report.** The cancel races
 431        the run's own finish, so "no such run" means the run already
 432        ended — which is what the operator asked for. The node answers
 433        ``{"cancelled": false}`` for that case rather than raising, and
 434        this returns it so a caller that wants to say "it had already
 435        finished" can.
 436        """
 437        resp = self.request("POST", f"/admin/ranges/verify/{run_id}/cancel")
 438        resp.raise_for_status()
 439        body = resp.json()
 440        return bool(body.get("cancelled"))
 441
 442    def rescan_challenges(self) -> ChallengeReloadResult:
 443        """Tell the node to drop its spec cache and re-scan challenges_dir.
 444
 445        Returns ``{total, added, removed}`` so the platform can report
 446        the per-node outcome of a cluster-wide rescan."""
 447        resp = self.request("POST", "/admin/rescan-challenges")
 448        resp.raise_for_status()
 449        return _parsed(ChallengeReloadResult, resp)
 450
 451    # -- admin pre-build (image cache warming) ------------------------------
 452
 453    def build_challenge(
 454        self, challenge_id: str, *, authored_revision: str = ""
 455    ) -> ChallengeBuildState:
 456        """Ask the node to pre-build images for *challenge_id*.
 457
 458        Fires-and-returns: the node persists ``status="building"`` and
 459        spawns a daemon thread; the body returned here is that initial
 460        state row. Polling :meth:`get_build_state` is how the platform
 461        learns when it lands on ``built`` / ``failed``. *authored_revision*
 462        is #1835's: the tree the build must be of, fetched if not held.
 463        """
 464        params = {"authored_revision": authored_revision} if authored_revision else None
 465        resp = self.request("POST", f"/admin/challenges/{challenge_id}/build", params=params)
 466        _raise_for_node_status(resp)
 467        return _parsed(ChallengeBuildState, resp)
 468
 469    def build_all_challenges(self, challenge_ids: list[str] | None = None) -> BuildAllQueued:
 470        """Queue background pre-build on the node.
 471
 472        With ``challenge_ids=None`` the node builds every spec it knows.
 473        With an explicit list (the platform scoping a bulk build to a
 474        competition's challenge set) only those are built. Returns
 475        ``{queued, skipped_built, skipped_in_progress}`` so the platform
 476        can report per-node what got picked up. Sequential on the node
 477        side — no fan-out across the corpus.
 478        """
 479        kwargs: dict[str, Any] = {}
 480        if challenge_ids is not None:
 481            kwargs["json"] = {"challenge_ids": challenge_ids}
 482        resp = self.request("POST", "/admin/challenges/build-all", **kwargs)
 483        _raise_for_node_status(resp)
 484        return _parsed(BuildAllQueued, resp)
 485
 486    def build_push_challenge(
 487        self,
 488        challenge_id: str,
 489        *,
 490        push_ticket: str = "",
 491        timeout_s: int = 1800,
 492        cleanup: bool = False,
 493        cache_max_bytes: int = 0,
 494        confine_network: bool = False,
 495        authored_revision: str = "",
 496    ) -> ChallengeFreezeResult:
 497        """Build one challenge's ``build:`` services and freeze them to the registry.
 498
 499        Layer 2 producer (P0). **Blocks** until the challenge is built +
 500        pushed (or skipped as already frozen), unlike the fire-and-return
 501        :meth:`build_challenge`, so the per-call HTTP timeout sits above
 502        the node's build budget. Returns the per-service freeze result;
 503        the platform's ``challenge_build_push`` task calls this one
 504        challenge at a time. *push_ticket* is the platform's short-lived
 505        registry credential the node's ``docker push`` authenticates with
 506        — the node runs it in a container with no ambient login.
 507        *cleanup* asks the node to remove what the build left and prune its
 508        build cache to *cache_max_bytes* afterwards, and *confine_network*
 509        keeps every ``RUN`` off the host's network (both the freeze pool, #1827).
 510        """
 511        resp = self.request(
 512            "POST",
 513            f"/admin/challenges/{challenge_id}/build-push",
 514            json={
 515                "push_ticket": push_ticket,
 516                "cleanup": cleanup,
 517                "cache_max_bytes": cache_max_bytes,
 518                "confine_network": confine_network,
 519                "authored_revision": authored_revision,
 520            },
 521            timeout=timeout_s + 60,
 522        )
 523        _raise_for_node_status(resp)
 524        return _parsed(ChallengeFreezeResult, resp)
 525
 526    def get_disk_free(self) -> NodeDiskFree:
 527        """Free space on the node's data filesystem, read now (#1827).
 528
 529        Raises on any failure — including a node predating the route, which
 530        answers 404 — and the freeze reads every failure as *cannot tell*,
 531        which stops it.
 532        """
 533        resp = self.request("GET", "/admin/disk")
 534        _raise_for_node_status(resp)
 535        return _parsed(NodeDiskFree, resp)
 536
 537    def get_build_state(self) -> BuildStateRows:
 538        """Fetch every per-challenge build-state row on this node.
 539
 540        Returns ``{"rows": [{challenge_id, status, built_at, error}, …]}``.
 541        ``status`` is one of ``unbuilt`` / ``building`` / ``built`` /
 542        ``failed``; ``unbuilt`` placeholders are synthesised for specs
 543        the node has seen but never been asked to build.
 544        """
 545        resp = self.request("GET", "/admin/challenges/build-state")
 546        _raise_for_node_status(resp)
 547        return _parsed(BuildStateRows, resp)
 548
 549    def pull_challenge(self, challenge_id: str) -> ChallengePullState:
 550        """Ask the node to pre-pull registry images for *challenge_id*.
 551
 552        Pull-side twin of :meth:`build_challenge`: the node persists
 553        ``status="pulling"`` and spawns a daemon thread; poll
 554        :meth:`get_pull_state` for the ``pulled`` / ``failed`` landing.
 555        """
 556        resp = self.request("POST", f"/admin/challenges/{challenge_id}/pull")
 557        _raise_for_node_status(resp)
 558        return _parsed(ChallengePullState, resp)
 559
 560    def pull_all_challenges(self, challenge_ids: list[str] | None = None) -> PullAllQueued:
 561        """Queue background pre-pull on the node.
 562
 563        With ``challenge_ids=None`` the node pulls every spec it knows;
 564        with an explicit list only those (the platform scoping to a
 565        competition). Returns ``{queued, skipped_pulled,
 566        skipped_in_progress}``.
 567        """
 568        kwargs: dict[str, Any] = {}
 569        if challenge_ids is not None:
 570            kwargs["json"] = {"challenge_ids": challenge_ids}
 571        resp = self.request("POST", "/admin/challenges/pull-all", **kwargs)
 572        _raise_for_node_status(resp)
 573        return _parsed(PullAllQueued, resp)
 574
 575    def get_pull_state(self) -> PullStateRows:
 576        """Fetch every per-challenge pull-state row on this node.
 577
 578        Returns ``{"rows": [{challenge_id, status, pulled_at, error}, …]}``.
 579        ``status`` is one of ``unpulled`` / ``pulling`` / ``pulled`` /
 580        ``failed``; ``unpulled`` placeholders are synthesised for specs
 581        the node has seen but never been asked to pull.
 582        """
 583        resp = self.request("GET", "/admin/challenges/pull-state")
 584        _raise_for_node_status(resp)
 585        return _parsed(PullStateRows, resp)
 586
 587    # -- status / health ----------------------------------------------------
 588
 589    def get_status(self, instance_id: str) -> InstanceStatusReport:
 590        resp = self.request("GET", f"/instances/{instance_id}/status")
 591        resp.raise_for_status()
 592        return _parsed(InstanceStatusReport, resp)
 593
 594    def list_instances(self) -> list[NodeInstanceSummary]:
 595        """Every instance the node still has bookkeeping for.
 596
 597        Used by the platform's boot re-adoption pass to decide which
 598        durable claims still have containers behind them. ``get_status``
 599        answers the same question one id at a time; asking it per claim
 600        would cost one round trip per instance to a node that may be
 601        slow or down, on the path that gates startup.
 602        """
 603        resp = self.request("GET", "/instances")
 604        resp.raise_for_status()
 605        return _parsed(NodeInstanceList, resp).items
 606
 607    def check_health(self, instance_id: str) -> bool:
 608        return self.check_health_verbose(instance_id).healthy
 609
 610    def check_health_verbose(self, instance_id: str) -> HealthReading:
 611        """Health plus whatever agent-deception telemetry the node read.
 612
 613        One call, because the platform already polls this endpoint for
 614        every live instance every tick and a second round trip per
 615        instance would be the whole cost of the feature. ``check_health``
 616        stays a bool for the callers that only ever wanted one.
 617
 618        Never an error, whatever arrives: this rides the health poll,
 619        and telemetry must not be able to mark a healthy instance
 620        unhealthy.
 621
 622        ⚠️ **The two ways this can come back empty are different facts
 623        and used to be one.** An older node sends no ``deception`` key
 624        at all — it *did not say*, which is ``channel=None`` and
 625        authorises nothing. A body this platform cannot parse is
 626        something else: the node did say, and we could not read it. That
 627        is ``unreadable``, the same value the reader gives for a missing
 628        file, because it is the same fact — there is no usable channel
 629        here — arriving through a different door.
 630
 631        The parse failure that matters is version skew in the other
 632        direction: a **newer** node reporting an ``ObservationKind`` this
 633        platform does not have. ``Observation.kind`` is typed, so one
 634        unknown member fails the whole model rather than dropping one
 635        row — which is the right refusal (the node's word is not ours to
 636        edit) and exactly why it must not land as "reported nothing".
 637        """
 638        resp = self.request("GET", f"/instances/{instance_id}/health")
 639        resp.raise_for_status()
 640        body = resp.json()
 641        raw = body.get("deception")
 642        if not raw:
 643            summary = DeceptionSummary()
 644        else:
 645            try:
 646                summary = DeceptionSummary.model_validate(raw)
 647            except ValidationError as exc:
 648                log.info(
 649                    "unreadable deception telemetry from node",
 650                    instance_id=instance_id,
 651                    error=tail(str(exc), limit=200),
 652                )
 653                summary = DeceptionSummary(channel="unreadable")
 654        raw_contact = body.get("contact")
 655        if not raw_contact:
 656            contact = ContactReading()
 657        else:
 658            try:
 659                contact = ContactReading.model_validate(raw_contact)
 660            except ValidationError as exc:
 661                log.info(
 662                    "unreadable contact telemetry from node",
 663                    instance_id=instance_id,
 664                    error=tail(str(exc), limit=200),
 665                )
 666                contact = ContactReading(channel="unreadable")
 667        raw_procs = body.get("new_processes")
 668        # ⚠️ Shape-checked rather than trusted: an older node sends
 669        # nothing and a newer one could send a shape this platform does
 670        # not model, and neither may reach a reader as *a container full
 671        # of intruders*. Anything unreadable is silence, which for a
 672        # signal built to under-report is the honest reading.
 673        procs: dict[str, list[str]] = {}
 674        if isinstance(raw_procs, dict):
 675            for container, names in raw_procs.items():
 676                if isinstance(container, str) and isinstance(names, list):
 677                    procs[container] = [str(n) for n in names]
 678        return HealthReading(
 679            healthy=bool(body.get("is_healthy")),
 680            deception=summary,
 681            contact=contact,
 682            new_processes=procs,
 683        )
 684
 685    def node_health(self) -> dict[str, Any]:
 686        """Liveness + ``{running, capacity}`` for heartbeat."""
 687        resp = self.request("GET", "/health")
 688        resp.raise_for_status()
 689        body: dict[str, Any] = resp.json()
 690        return body
 691
 692    # -- traffic / runtime --------------------------------------------------
 693
 694    def get_traffic_page(
 695        self, instance_id: str, *, cursor: str = "", limit: int = 0
 696    ) -> InstanceTrafficPage:
 697        """One page of mitmproxy flows; the capture lives on the node's FS.
 698
 699        ⚠️ **The one method that must not use ``_parsed``.** A node from
 700        before paging answers with a bare *list* — it sent the whole
 701        capture and understands no cursor — and a strict
 702        ``model_validate`` turns that supported skew into a raised
 703        request. So the body goes through ``coerce_page``, which already
 704        owns every decision about an unexpected shape, and the caller
 705        gets the model.
 706
 707        This is the same move ``coerce_page`` was written for, one layer
 708        down. Its docstring used to say the parsing belonged
 709        "platform-side … so a route and this client cannot come to
 710        disagree" — and there were *three* platform-side calls doing it.
 711        There is one call now, at the seam that knows the request
 712        cursor, which is what the fallback needs; the routes hold the
 713        type instead of re-deciding its shape.
 714
 715        ⚠️ **An empty ``cursor`` means "from the beginning".** Passing
 716        one on every poll re-reads the whole capture, which is the cost
 717        the paged read exists to remove — arriving through the caller
 718        rather than through the node.
 719        """
 720        params: dict[str, Any] = {}
 721        if cursor:
 722            params["cursor"] = cursor
 723        if limit > 0:
 724            params["limit"] = limit
 725        resp = self.request("GET", f"/instances/{instance_id}/traffic", params=params)
 726        resp.raise_for_status()
 727        page = coerce_page(resp.json(), cursor)
 728        return InstanceTrafficPage(
 729            flows=page.flows,
 730            cursor=page.cursor,
 731            truncated=page.truncated,
 732            capture=page.capture,
 733        )
 734
 735    def get_traffic_witnesses(
 736        self, instance_id: str, *, cursor: str = "", limit: int = 0
 737    ) -> InstanceTrafficWitnesses:
 738        """Who spoke to this box since *cursor*, with no bodies moved.
 739
 740        §7.1's correlation reads one gamebox per team per round. Through
 741        :meth:`get_traffic_page` that is every flow, both bodies
 742        included (~100 kB apiece by ``CLAUDE.md``'s measurement),
 743        pulled across the wire to extract a handful of addresses; the
 744        node reduces instead and this carries the result.
 745
 746        ⚠️ **Strict parsing here, unlike its neighbour, and that is the
 747        difference between an old field and an old route.**
 748        ``get_traffic_page`` is lenient because a node from before
 749        paging answers the *same* path with a bare list, which is
 750        supported skew. This route did not exist on such a node at all,
 751        so the honest answer from one is a 404 — and a lenient parse of
 752        an error body would hand the round loop an empty witness set,
 753        which reads as *nobody was there*. The caller catches the
 754        request error and records the round as unknown.
 755        """
 756        params: dict[str, Any] = {}
 757        if cursor:
 758            params["cursor"] = cursor
 759        if limit > 0:
 760            params["limit"] = limit
 761        resp = self.request("GET", f"/instances/{instance_id}/traffic-witnesses", params=params)
 762        resp.raise_for_status()
 763        return _parsed(InstanceTrafficWitnesses, resp)
 764
 765    def list_containers(self, instance_id: str) -> list[ComposeContainer]:
 766        """Enumerate every container in an instance's compose project.
 767
 768        Backs the admin-shell feature: the platform forwards a
 769        ``GET /admin/instances/{id}/containers`` to the assigned node,
 770        which returns one row per container (challenge services and
 771        platform-injected sidecars alike). The WebSocket reverse-proxy
 772        is opened separately and does not go through ``NodeClient``.
 773        """
 774        resp = self.request("GET", f"/instances/{instance_id}/containers")
 775        _raise_for_node_status(resp)
 776        return _parsed_list(ComposeContainer, resp)
 777
 778    def run_checker(
 779        self, instance_id: str, *, service: str, cmd: list[str], timeout_s: int = 30
 780    ) -> CheckResult:
 781        """Exec an author-supplied checker in a trusted judge sidecar.
 782
 783        Backs the checker / exec-judge feature: the platform forwards a
 784        verify request to the assigned node, which runs ``cmd`` inside the
 785        named sidecar and returns
 786        ``{exit_code, stdout, stderr, timed_out, error}``. The per-call HTTP
 787        timeout sits above the node's exec budget so the client doesn't
 788        abandon the request before the node returns.
 789        """
 790        resp = self.request(
 791            "POST",
 792            f"/instances/{instance_id}/check",
 793            json={"service": service, "cmd": cmd, "timeout_s": timeout_s},
 794            timeout=timeout_s + 10,
 795        )
 796        _raise_for_node_status(resp)
 797        return _parsed(CheckResult, resp)
 798
 799    def run_harness(
 800        self,
 801        *,
 802        image: str,
 803        run_id: str = "",
 804        env: dict[str, str],
 805        timeout_s: int = 1800,
 806        untrusted: bool = False,
 807        network: str = "",
 808        own_network: bool = False,
 809        match_network: str = "",
 810        match_address: str = "",
 811        brief: dict[str, Any] | None = None,
 812        target_instance: str = "",
 813        target_aliases: dict[str, str] | None = None,
 814        pull_ticket: str = "",
 815        limits: HarnessLimits | None = None,
 816    ) -> HarnessRun:
 817        """Run a one-shot evaluation-harness container on the node.
 818
 819        The node ``docker run``s ``image`` with ``env`` injected, waits for it
 820        to exit (bounded by ``timeout_s``), and returns
 821        ``{exit_code, stdout, stderr, timed_out, error}`` — the same shape as
 822        :meth:`run_checker`. The harness writes its JSON rollup to stdout. The
 823        per-call HTTP timeout sits above the node's run budget so the client
 824        doesn't abandon the request before the (long-running) harness returns.
 825
 826        ``untrusted`` runs the image under the node's isolation profile —
 827        off host networking, bounded resources, no capabilities — for an
 828        image the platform did not build. It requires a network: the
 829        node answers 400 without one, because "isolate this" and "I will
 830        not say where the target lives" do not compose.
 831
 832        Any of ``network`` (the target instance's own, when the caller
 833        knows it), ``own_network`` or ``target_instance`` satisfies that.
 834        ⚠️ Most eval dispatches need the second: there the harness
 835        *starts* the target partway through its own run, so at ``docker
 836        run`` time no target network exists and there is no name to
 837        send. The node creates a fresh bridge and removes it afterwards.
 838
 839        ``target_instance`` reverses that ordering, and is what an
 840        `engagement` uses: the platform starts the estate first and names
 841        it here, and the node wires its published containers to the
 842        agent's bridge. ⚠️ An **id**, never a network name — the name is
 843        a fact only the node's daemon can read, and ``docker run``
 844        *creates* whatever string it is handed, so a platform guessing
 845        one starts a run that reaches nothing.
 846        """
 847        resp = self.request(
 848            "POST",
 849            "/harness/run",
 850            json={
 851                "image": image,
 852                "env": env,
 853                "timeout_s": timeout_s,
 854                "untrusted": untrusted,
 855                "network": network,
 856                "own_network": own_network,
 857                "match_network": match_network,
 858                # ⚠️ **The platform-assigned arena address.** The node
 859                # connects the agent to the match network with this
 860                # ``--ip``, so the address is known before the round loop
 861                # asks (§7.1). Empty means "let docker's IPAM pick" — an
 862                # older platform's behaviour, and what keeps a node reading
 863                # this field unchanged for every non-arena dispatch.
 864                "match_address": match_address,
 865                # ⚠️ Sent even when empty, so a node reading it gets the
 866                # same shape either way; the node's default is what makes
 867                # an older platform's dispatch unchanged.
 868                "pull_ticket": pull_ticket,
 869                # ⚠️ ``None`` leaves the node's own defaults in force, which is
 870                # what a dispatch said before the ceilings were settings.
 871                "limits": limits.model_dump() if limits is not None else None,
 872                # ⚠️ Same rule, and empty is a real answer here too: a
 873                # dispatch with no run row (the conformance probe, a dry
 874                # run) has nothing to label, and a label with an empty
 875                # value would make every one of them answer one query.
 876                "run_id": run_id,
 877                # ⚠️ Sent as the already-built manifest. The platform is
 878                # the side that knows the challenge, the competition and
 879                # the run mode; a node deriving its own would be a
 880                # second implementation of a published contract.
 881                "brief": brief,
 882                "target_instance": target_instance,
 883                # ⚠️ **The names the brief published, so they resolve on
 884                # the agent's bridge too.** An estate's hosts answer to a
 885                # per-instance FQDN minted by its own resolver; joined to
 886                # a new network a container answers only to its compose
 887                # *service* name. Sending the mapping is what keeps the
 888                # address the agent was told about true under either
 889                # placement — without it the agent is handed a hostname
 890                # that does not resolve, reaches nothing, and reports an
 891                # empty result, which reads as a model that gave up.
 892                #
 893                # ⚠️ The platform's to send: it computed the surface. A
 894                # node re-deriving it from the rendered compose would be
 895                # a second implementation of the address the brief has
 896                # already promised.
 897                "target_aliases": target_aliases or {},
 898            },
 899            timeout=timeout_s + 30,
 900        )
 901        _raise_for_node_status(resp)
 902        return _parsed(HarnessRun, resp)
 903
 904    def cancel_harness(self, *, run_id: str) -> int:
 905        """Stop whatever this node is running for ``run_id``; returns how many.
 906
 907        Until this existed a dispatched container was reclaimed by
 908        **nothing but its own timeout** — ``_harness_timeout``'s docstring
 909        says exactly that — so an operator pressing cancel on a
 910        match-long run was asking for something the platform had no way
 911        to deliver.
 912
 913        ⚠️ **Zero is an ordinary answer, not a failure.** The run may have
 914        finished on its own, been dispatched to a different node, or been
 915        lost to a node restart. A caller that read zero as an error would
 916        report a broken platform every time a cancel arrived a second
 917        late.
 918        """
 919        resp = self.request("POST", "/harness/cancel", json={"run_id": run_id})
 920        body = resp.json()
 921        return int(body.get("stopped", 0)) if isinstance(body, dict) else 0
 922
 923    def verify_patch(
 924        self, challenge_id: str, files: dict[str, bytes], *, timeout_s: int = 900
 925    ) -> PatchVerifyResult:
 926        """Build a submitted patch on the node and return its verdict.
 927
 928        ``files`` maps a challenge-relative path to its new bytes; base64
 929        is applied here because a patch target may legitimately be
 930        binary. Returns ``{verdict, detail, applied, steps, error}``.
 931
 932        **An empty ``verdict`` with a populated ``error`` is not a
 933        judgement** — it means the node refused the submission or the
 934        verifier crashed. The caller must retry or retire the lease
 935        rather than write a score, since failing to judge a patch is not
 936        the same as judging it unfixed.
 937
 938        The per-call HTTP timeout sits above the node's own budget so the
 939        client doesn't abandon the request while the node is still
 940        building.
 941        """
 942        resp = self.request(
 943            "POST",
 944            "/patches/verify",
 945            json={
 946                "challenge_id": challenge_id,
 947                "files": {
 948                    path: base64.b64encode(content).decode() for path, content in files.items()
 949                },
 950                "timeout_s": timeout_s,
 951            },
 952            timeout=timeout_s + 60,
 953        )
 954        _raise_for_node_status(resp)
 955        return _parsed(PatchVerifyResult, resp)
 956
 957    def collect_patch_files(
 958        self, instance_id: str, service: str, paths: list[str]
 959    ) -> dict[str, bytes]:
 960        """Read a player's SSH edits back out of their running box.
 961
 962        ``paths`` are absolute in-container paths the *platform* derived
 963        from the pristine source tree. Returns only what could be read:
 964        a path missing from the reply means the player did not change it,
 965        and the caller falls back to the shipped file rather than
 966        recording a deletion they never made.
 967        """
 968        resp = self.request(
 969            "POST",
 970            f"/instances/{instance_id}/patch-collect",
 971            json={"service": service, "paths": paths},
 972        )
 973        _raise_for_node_status(resp)
 974        body: dict[str, str] = resp.json()
 975        return {path: base64.b64decode(blob) for path, blob in body.items()}
 976
 977    def write_awd_answers(self, boxes: dict[str, dict[str, str]]) -> dict[str, str]:
 978        """Rotate a round's flags into every box of this node, in one call.
 979
 980        ⚠️ **One call for the whole node, never one per box.** §5.2's
 981        capacity red line is 1000 teams x 3 services injected inside
 982        120 s; per-box that is 3000 round trips, batched it is about
 983        fourteen. The signature is the enforcement — a caller cannot
 984        accidentally loop.
 985
 986        Returns ``{instance_id: ""}`` for the boxes written and a reason
 987        for the ones that were not, because one team's unreachable box
 988        must not cost every other team on the node its rotation. An
 989        unrotated box keeps serving last round's flag, which anyone
 990        holding the old value can replay.
 991        """
 992        resp = self.request("POST", "/awd/answers", json={"boxes": boxes})
 993        _raise_for_node_status(resp)
 994        result: dict[str, str] = resp.json()
 995        return result
 996
 997    def probe_awd_service(
 998        self,
 999        *,
1000        instance_id: str,
1001        service_id: str,
1002        tick: int,
1003        flag: str,
1004        previous_flag: str = "",
1005        previous_flag_id: str = "",
1006    ) -> dict[str, str]:
1007        """Run one round's SLA probes against one box, from outside it.
1008
1009        The flags travel *to* the node: a checker that sourced its
1010        expected value from the box it is checking would pass on any box
1011        that returns whatever it was handed.
1012        """
1013        resp = self.request(
1014            "POST",
1015            "/awd/probe",
1016            json={
1017                "instance_id": instance_id,
1018                "service_id": service_id,
1019                "tick": tick,
1020                "flag": flag,
1021                "previous_flag": previous_flag,
1022                "previous_flag_id": previous_flag_id,
1023            },
1024        )
1025        _raise_for_node_status(resp)
1026        result: dict[str, str] = resp.json()
1027        return result
1028
1029    def attack_awd_box(
1030        self,
1031        *,
1032        instance_id: str,
1033        tick: int,
1034        timeout_s: int = 60,
1035        agent_image: str = "",
1036        counter_flag: str = "",
1037        forensic_marker: str = "",
1038        llm_api_key: str = "",
1039    ) -> AwdAttackResult:
1040        """Run this box's own reference exploit against it, once.
1041
1042        Per **box**, not per service: one `run_exploit` discovers every
1043        answer the challenge declares and prints them all.
1044
1045        ``timeout_s`` is sent because the platform is what knows how long
1046        a round is; the node clamps it, because a container held open for
1047        an hour is the node's problem whoever asked for it.
1048
1049        ``forensic_marker`` is §4.3.1's, carried only by the agent
1050        branch: `build_attack_argv` passes no environment at all, so a
1051        marker on the reference exploit would be the one value that
1052        voids `EXPLOIT007`'s guarantee.
1053
1054        ``agent_image`` + ``counter_flag`` switch the node to §4.2b's
1055        agent attacker. ⚠️ **Both or neither** — the node refuses a half
1056        request rather than falling back to the reference exploit, which
1057        would run a different format than the one configured while the
1058        board looked entirely normal.
1059        """
1060        resp = self.request(
1061            "POST",
1062            "/awd/attack",
1063            json={
1064                "instance_id": instance_id,
1065                "tick": tick,
1066                "timeout_s": timeout_s,
1067                "agent_image": agent_image,
1068                "counter_flag": counter_flag,
1069                "forensic_marker": forensic_marker,
1070                "llm_api_key": llm_api_key,
1071            },
1072        )
1073        _raise_for_node_status(resp)
1074        return _parsed(AwdAttackResult, resp)
1075
1076    def get_container_logs(self, instance_id: str) -> str:
1077        """Fetch combined container stdout/stderr for archival."""
1078        resp = self.request("GET", f"/instances/{instance_id}/container-logs")
1079        resp.raise_for_status()
1080        return str(resp.json().get("logs") or "")
1081
1082    def get_pcap(self, instance_id: str) -> PcapCapture:
1083        """Fetch the merged tcpdump capture for *instance_id*.
1084
1085        ``data`` is empty when no capture exists (sidecar disabled,
1086        instance never reached the running state, etc.) so callers can
1087        persist conditionally without try/except.
1088
1089        ⚠️ **It also says what the per-instance cap already dropped**
1090        (#1714), off the node's response headers, and ``None`` when the
1091        node did not say — an older node, which is *not reported*, never
1092        a clean capture.
1093        """
1094        resp = self.request("GET", f"/instances/{instance_id}/pcap")
1095        if resp.status_code == 404:
1096            return PcapCapture(data=b"")
1097        resp.raise_for_status()
1098        return PcapCapture(
1099            data=resp.content,
1100            dropped_bytes=_header_int(resp, CAPTURE_DROPPED_BYTES_HEADER),
1101            dropped_segments=_header_int(resp, CAPTURE_DROPPED_SEGMENTS_HEADER),
1102        )
1103
1104    def get_harness_artifact(self, artifact_id: str) -> bytes:
1105        """One finished run's collected archive, or ``b""``.
1106
1107        ⚠️ **Empty rather than an exception for a 404**, following
1108        :meth:`get_pcap`. The node reaps uncollected archives, and a
1109        platform that comes back after a restart is exactly the caller
1110        that meets one that is gone — a state its own record already
1111        describes, not a transport failure.
1112        """
1113        resp = self.request("GET", f"/harness/artifacts/{artifact_id}")
1114        if resp.status_code == 404:
1115            return b""
1116        resp.raise_for_status()
1117        return resp.content
1118
1119    def delete_harness_artifact(self, artifact_id: str) -> bool:
1120        """Tell the node the archive has been stored and can go.
1121
1122        ⚠️ **Never raises on a transport failure**, because it is called
1123        after the bytes are safely in object storage: turning a cleanup
1124        error into the run's error would replace a success with a
1125        janitorial complaint. The node's reaper collects whatever this
1126        misses.
1127        """
1128        try:
1129            resp = self.request("DELETE", f"/harness/artifacts/{artifact_id}")
1130        except Exception:
1131            return False
1132        return resp.status_code == 200
1133
1134    def get_agent_runtime(self, instance_id: str) -> AgentRuntime:
1135        """Provider-agnostic runtime hints for attaching an agent (proxy URL,
1136        CA PEM, optional Docker-specific names). Replaces the old
1137        sandbox-network shape."""
1138        resp = self.request("GET", f"/instances/{instance_id}/agent-runtime")
1139        resp.raise_for_status()
1140        return _parsed(AgentRuntime, resp)
1141
1142    # -- per-instance rendered attachments ---------------------------------
1143
1144    def list_instance_attachments(self, instance_id: str) -> AttachmentList:
1145        """List per-instance attachments rendered into the node's workdir.
1146
1147        Backs the platform's per-instance attachment endpoint, which
1148        surfaces the team-specific rendered file listing.
1149
1150        ⚠️ This used to hand back the raw dict, under a docstring saying
1151        "caller projects through :class:`AttachmentList` for typing" —
1152        the node's route already declared that model, so the type existed
1153        and this method's signature was the one place it was dropped.
1154        Projecting *here* means there is one place a malformed answer is
1155        caught rather than one per caller."""
1156        resp = self.request("GET", f"/instances/{instance_id}/attachments")
1157        resp.raise_for_status()
1158        return _parsed(AttachmentList, resp)
1159
1160    def get_vpn_config(self, instance_id: str) -> bytes:
1161        """Fetch the per-instance VPN client config from the node.
1162
1163        The node reads the backend's client config out of the
1164        instance's VPN service container (generated on first boot by
1165        the base image's bootstrap) and rewrites its endpoint to the
1166        node's public host:port. Returns the body verbatim — the
1167        platform-side route adds the ``Content-Disposition`` header
1168        before re-emitting to the player, because only it knows the
1169        backend and therefore the filename the player needs.
1170
1171        Raises ``httpx.HTTPStatusError`` on non-2xx — the platform's
1172        proxy route catches 404 and re-emits to the player, anything
1173        else is treated as a 502.
1174        """
1175        resp = self.request("GET", f"/instances/{instance_id}/vpn-config")
1176        resp.raise_for_status()
1177        return resp.content
1178
1179    def download_instance_attachment(self, instance_id: str, filename: str) -> tuple[bytes, str]:
1180        """Download one per-instance attachment as ``(bytes, content_type)``.
1181
1182        Buffers the whole body — attachments are bounded (typical case
1183        is a 10 KB binary or a small text file). For huge captures the
1184        caller should hand the player a CDN URL instead.
1185
1186        Raises ``httpx.HTTPStatusError`` on non-2xx — the platform's
1187        proxy route catches 404 and re-emits to the player, anything
1188        else is a 502."""
1189        resp = self.request(
1190            "GET",
1191            f"/instances/{instance_id}/attachments/{filename}",
1192        )
1193        resp.raise_for_status()
1194        return resp.content, resp.headers.get("content-type", "application/octet-stream")