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 collections.abc import Iterable, Mapping, Sequence
  16from typing import Any, TypeVar
  17from urllib.parse import quote
  18
  19import httpx
  20from pydantic import BaseModel, Field, ValidationError
  21
  22from ctfy.challenge.deception import DeceptionSummary
  23from ctfy.core.agent_log import NodeAgentLog
  24from ctfy.core.attachment_layout import (
  25    InvalidAttachmentName,
  26    attachment_url_path,
  27    component_refusal,
  28)
  29from ctfy.core.awd_network import MatchNetwork
  30from ctfy.core.challenge import ChallengeReloadResult
  31from ctfy.core.constants import (
  32    DEFAULT_CLIENT_TIMEOUT,
  33    HARNESS_CREATE_TIMEOUT_S,
  34    HARNESS_IMAGE_INSPECT_TIMEOUT_S,
  35    IMAGE_PULL_TIMEOUT_S,
  36)
  37from ctfy.core.enums import InternetAccess, TcpAccess
  38from ctfy.core.exceptions import NodeRequestError
  39from ctfy.core.failure_text import tail
  40from ctfy.core.freeze_budget import freeze_client_wait_s, freeze_deadline_s
  41from ctfy.core.instance_disk_retention import (
  42    CAPTURE_DROPPED_BYTES_HEADER,
  43    CAPTURE_DROPPED_SEGMENTS_HEADER,
  44)
  45from ctfy.core.judge_budget import DEFAULT_JUDGE_TIMEOUT_S, judge_exec_s
  46from ctfy.core.judge_budget import platform_wait_s as judge_platform_wait_s
  47from ctfy.core.log import log
  48from ctfy.core.node_features import CAPABILITY_FEATURES, feature_enabled
  49from ctfy.core.observed_contact import ContactReading
  50from ctfy.core.patch_budget import patch_client_wait_s
  51from ctfy.core.prebuild_tag import ChallengeFreezeResult
  52from ctfy.core.resource_budget import ResourceBudget
  53from ctfy.core.smoke_budget import SOLUTION_BUILD_TIMEOUT_S
  54from ctfy.core.smoke_live import NodeSmokeProgress
  55from ctfy.sdk.base import RETRY_UNSENT, BaseHttpClient
  56from ctfy.server.models.challenge import AttachmentList
  57from ctfy.server.models.instance import InstanceTrafficPage, InstanceTrafficWitnesses
  58from ctfy.server.models.node import (
  59    AgentIngressPeer,
  60    AgentIngressPeerRemoved,
  61    AgentIngressPeerRequest,
  62    AgentRuntime,
  63    AwdAttackResult,
  64    BuildAllQueued,
  65    BuildPushCancelled,
  66    BuildPushProgress,
  67    BuildStateRows,
  68    CheckResult,
  69    ComposeContainer,
  70    FabricProbeReport,
  71    FabricStatus,
  72    HarnessImagePolicy,
  73    HarnessLimits,
  74    HarnessRegistryAuth,
  75    HarnessRun,
  76    HarnessRunResult,
  77    InstanceHeld,
  78    InstanceInternetChange,
  79    InstanceInternetChanged,
  80    InstanceStarted,
  81    InstanceStatusReport,
  82    NodeAuthoredPush,
  83    NodeAuthoredResult,
  84    NodeAuthoredSmokePush,
  85    NodeAuthoredSmokeResult,
  86    NodeCapabilities,
  87    NodeDiskFree,
  88    NodeInstanceList,
  89    NodeInstanceSummary,
  90    NodeRangePush,
  91    NodeRangeResult,
  92    NodeRangeVerifyPush,
  93    NodeRangeVerifyResult,
  94    NodeSmokeEntrance,
  95    NodeSmokeExploitOutcome,
  96    PatchVerifyResult,
  97    PullAllQueued,
  98    PullStateRows,
  99)
 100from ctfy.server.node_state import ChallengeBuildState, ChallengePullState
 101from ctfy.server.traffic_tail import coerce_page
 102
 103
 104@dataclasses.dataclass(frozen=True)
 105class PcapCapture:
 106    """A node's merged capture, and what the cap dropped from it (#1714)."""
 107
 108    data: bytes
 109    #: ``None`` = the node did not say (older build); ``0`` = nothing.
 110    dropped_bytes: int | None = None
 111    dropped_segments: int | None = None
 112
 113
 114def _header_int(resp: httpx.Response, name: str) -> int | None:
 115    raw = resp.headers.get(name)
 116    if raw is None:
 117        return None
 118    try:
 119        return max(0, int(raw))
 120    except ValueError:
 121        return None
 122
 123
 124def _raise_for_node_status(resp: httpx.Response) -> None:
 125    """Like ``resp.raise_for_status()`` but preserves the node's error body.
 126
 127    On a 4xx/5xx, pull the FastAPI ``{"detail": ...}`` field (or the raw
 128    body as a fallback) and raise :class:`NodeRequestError` so the real
 129    failure reaches the caller instead of httpx's generic status string.
 130    """
 131    if not resp.is_error:
 132        return
 133    detail = ""
 134    code = ""
 135    try:
 136        body = resp.json()
 137    except (json.JSONDecodeError, ValueError):
 138        body = None
 139    if isinstance(body, dict):
 140        d = body.get("detail")
 141        if isinstance(d, str):
 142            detail = d
 143        elif isinstance(d, dict) and isinstance(d.get("code"), str):
 144            # A node naming its failure (#1835): keep the name as a field
 145            # and the sentence as the detail, so neither a reader of the
 146            # code nor a human reading the detail meets JSON.
 147            code = str(d["code"])
 148            message = d.get("message")
 149            detail = message if isinstance(message, str) else json.dumps(d)
 150        elif d is not None:
 151            detail = json.dumps(d)
 152    if not detail:
 153        detail = (resp.text or "").strip()
 154    raise NodeRequestError(resp.status_code, detail, code=code)
 155
 156
 157_M = TypeVar("_M", bound=BaseModel)
 158
 159
 160def _parsed(model: type[_M], resp: httpx.Response) -> _M:
 161    """The node's answer, as the model the node itself declares.
 162
 163    ⚠️ **A parse failure raises rather than defaulting.** Its neighbour
 164    ``instance_health`` falls back to an empty ``DeceptionSummary``, and
 165    that is right there: the summary is decoration on a liveness answer
 166    the caller already has. These carry the *verdict* — whether a checker
 167    passed, what an attack read — and a default is a verdict nobody
 168    reached. The checker's own fail direction is closed, and an answer
 169    this client cannot parse is not an answer.
 170
 171    ⚠️ Unknown fields are ignored (pydantic's default), which is the
 172    forward-compatible half: the node and the platform are deployed
 173    separately, ``ctfy/core/version.py`` exists because they skew, and a
 174    newer node sending a field this platform has never heard of must not
 175    fail the call.
 176    """
 177    try:
 178        return model.model_validate(resp.json())
 179    except (ValidationError, ValueError) as exc:
 180        # ⚠️ ``resp.status_code``, not a synthetic one: the node
 181        # *answered*, and the failure is that its body does not fit the
 182        # contract. Reporting a 502 here would say the node was
 183        # unreachable, which sends whoever reads it to the wrong host.
 184        raise NodeRequestError(
 185            resp.status_code,
 186            f"answered something this platform cannot read as {model.__name__}: {exc}",
 187        ) from exc
 188
 189
 190def _parsed_list(model: type[_M], resp: httpx.Response) -> list[_M]:
 191    """A JSON array of *model*, with the same failure direction as
 192    :func:`_parsed` — one unreadable element fails the call rather than
 193    being dropped, because a partial list read as a whole one is how a
 194    re-adoption pass decides an instance is gone."""
 195    body = resp.json()
 196    if not isinstance(body, list):
 197        raise NodeRequestError(
 198            resp.status_code, f"expected a list of {model.__name__}, got {type(body).__name__}"
 199        )
 200    try:
 201        return [model.model_validate(item) for item in body]
 202    except (ValidationError, ValueError) as exc:
 203        raise NodeRequestError(
 204            resp.status_code,
 205            f"answered something this platform cannot read as a list of {model.__name__}: {exc}",
 206        ) from exc
 207
 208
 209class HealthReading(BaseModel):
 210    """One health poll's three answers.
 211
 212    ⚠️ **A record rather than a tuple, and the reason is recorded one
 213    subsystem over.** ``competition_results`` lost ``attempts`` and
 214    ``last_solve_at`` to a hand-made ``(rank, flags, solved)`` tuple
 215    sitting between a board that produced them and a template that
 216    consumed them; this poll had two answers and now has three, so the
 217    tuple was one field away from the same fate. Growing a record is a
 218    field; growing a tuple is every call site, silently re-bound.
 219    """
 220
 221    healthy: bool = False
 222    deception: DeceptionSummary = Field(default_factory=DeceptionSummary)
 223    contact: ContactReading = Field(default_factory=ContactReading)
 224    new_processes: dict[str, list[str]] = Field(default_factory=dict)
 225
 226
 227def masked_capabilities(caps: NodeCapabilities, disabled: Iterable[str]) -> NodeCapabilities:
 228    """*caps* with every feature in *disabled* answered ``False`` (#3037).
 229
 230    The effective value is *supported by the node* AND *enabled by the
 231    admin*. ⚠️ Only names that are features are masked
 232    (`ctfy.core.node_features.CAPABILITY_FEATURES`), so an unknown or stale name
 233    in the row turns nothing off — the switch fails open. A placement switch
 234    (``public_ports``) is not a capability and is asked by placement instead.
 235    """
 236    off = [f for f in CAPABILITY_FEATURES if not feature_enabled(disabled, f)]
 237    if not off:
 238        return caps
 239    update: dict[str, object] = dict.fromkeys(off, False)
 240    for flag in off:
 241        reason = f"{flag}_reason"
 242        if reason in NodeCapabilities.model_fields:
 243            update[reason] = "switched off by an admin on this node"
 244    return caps.model_copy(update=update)
 245
 246
 247class NodeClient(BaseHttpClient):
 248    """Thin sync HTTP client; one instance per node URL."""
 249
 250    def __init__(
 251        self,
 252        node_url: str,
 253        token: str,
 254        *,
 255        timeout: int = DEFAULT_CLIENT_TIMEOUT,
 256        max_retries: int = 2,
 257        transport: httpx.BaseTransport | None = None,
 258        disabled_features: Iterable[str] = (),
 259    ) -> None:
 260        """*disabled_features* are the features an admin switched off on this
 261        node (#3037, `ctfy.core.node_features`): :meth:`capabilities` answers
 262        ``False`` for each, whatever the node says, so a caller reads the
 263        effective value and cannot forget the switch. ``()`` is the raw answer
 264        — every platform construction passes the node row's set
 265        (`test_every_node_client_carries_the_switches`).
 266
 267        *transport* is `BaseHttpClient`'s own seam, passed through so a
 268        test can drive a real node app over the real client — the hop the
 269        fakes on either side of it cannot see.
 270
 271        *max_retries* is passed through too: a best-effort read or a stop
 272        is tried once (#2812), because a retry on a timeout multiplies the
 273        one number that has to stay small.
 274
 275        ⚠️ **A call that starts long work retries only a request that never
 276        left** (#3162, :data:`~ctfy.sdk.base.RETRY_UNSENT`): every POST here
 277        that names its own HTTP timeout says so per call, because a
 278        ``ReadTimeout`` is a request the node has and is still running.
 279        ``start_instance`` keeps the default — the node replays it from the
 280        instance id. Guarded by `ctfy/tests/test_long_node_work_is_sent_once.py`."""
 281        super().__init__(
 282            f"{node_url.rstrip('/')}/api/v1",
 283            token,
 284            timeout=timeout,
 285            max_retries=max_retries,
 286            transport=transport,
 287        )
 288        self.disabled_features = frozenset(disabled_features)
 289
 290    # -- lifecycle ----------------------------------------------------------
 291
 292    def start_instance(
 293        self,
 294        *,
 295        challenge_id: str,
 296        instance_id: str,
 297        ttl: int,
 298        answers: dict[str, str],
 299        proxy_output_dir: str | None = None,
 300        env: dict[str, str] | None = None,
 301        publish_gamebox: bool = False,
 302        match_network: MatchNetwork | None = None,
 303        open_shell_password: str = "",
 304        authored_revision: str = "",
 305        internet: InternetAccess = InternetAccess.ALLOW,
 306        tcp_access: TcpAccess = TcpAccess.TLS_SNI,
 307        egress_allow_hosts: list[str] | None = None,
 308        resource_budget: ResourceBudget | None = None,
 309        compose_up_timeout_s: int = 0,
 310    ) -> InstanceStarted:
 311        resp = self.request(
 312            "POST",
 313            "/instances",
 314            json={
 315                # #1835: the authored tree's revision this start must run;
 316                # a node holding another (or none) fetches it first.
 317                "authored_revision": authored_revision,
 318                "challenge_id": challenge_id,
 319                "instance_id": instance_id,
 320                "ttl": ttl,
 321                "answers": answers,
 322                "proxy_output_dir": proxy_output_dir,
 323                "env": env or {},
 324                "publish_gamebox": publish_gamebox,
 325                # ``None`` rather than an omitted key: the node's model
 326                # defaults to None either way, and sending it explicitly
 327                # keeps the body's shape stable across both branches.
 328                "match_network": (dataclasses.asdict(match_network) if match_network else None),
 329                # Classic-AWD open shell (#1725): the box's starting root
 330                # password, empty for every managed / non-AWD start.
 331                "open_shell_password": open_shell_password,
 332                # #1969: ``deny`` holds the box off the internet; the hosts
 333                # are the platform-internal ones it still reaches.
 334                "internet": str(internet),
 335                # #2082: how a player reaches the box's TCP services.
 336                "tcp_access": str(tcp_access),
 337                "egress_allow_hosts": list(egress_allow_hosts or []),
 338                # The CPU / memory a player-authored box is held to; ``None``
 339                # for the corpus and from an older platform.
 340                "resource_budget": (
 341                    resource_budget.model_dump() if resource_budget is not None else None
 342                ),
 343                # #3165: the platform's ``compose_up_timeout_s``, so the node
 344                # spends the budget the platform waits on; ``0`` = the node's own.
 345                "compose_up_timeout_s": compose_up_timeout_s,
 346            },
 347        )
 348        _raise_for_node_status(resp)
 349        return _parsed(InstanceStarted, resp)
 350
 351    def stop_instance(self, instance_id: str) -> None:
 352        resp = self.request("DELETE", f"/instances/{instance_id}")
 353        resp.raise_for_status()
 354
 355    def apply_fabric(
 356        self,
 357        *,
 358        index: int,
 359        interface: dict[str, Any],
 360        match_network: str = "",
 361    ) -> FabricStatus:
 362        """Put this node on one match's WireGuard fabric.
 363
 364        ⚠️ **Returns the model, never ``resp.json()``.** The platform
 365        reads the verdict field by field, and on a dict a renamed field
 366        is ``None`` — falsy — which reads as "the arena did not open",
 367        so a wire change would quietly report every node as broken.
 368
 369        Returns the node's verdict rather than raising on a refusal: the
 370        route answers **200 with a body** precisely so that "this node is
 371        unreachable" and "this node rejected the plan" stay different
 372        facts, and a caller fanning out over an arena needs to tell them
 373        apart — one is waited on, the other is looked at.
 374
 375        *match_network* lets the node check the tunnel landed in the
 376        namespace its boxes are in, and open the bridge's one hole. It is
 377        the platform's own fact (`awd_network.network_name`), so the node
 378        is told rather than guessing.
 379        """
 380        resp = self.request(
 381            "POST",
 382            "/fabric",
 383            json={
 384                "index": index,
 385                "interface": interface,
 386                "match_network": match_network,
 387            },
 388        )
 389        resp.raise_for_status()
 390        return _parsed(FabricStatus, resp)
 391
 392    def probe_fabric(
 393        self,
 394        *,
 395        index: int,
 396        peers: list[dict[str, str]],
 397        mtu: int = 0,
 398    ) -> FabricProbeReport:
 399        """Ask named peers the three link questions. Applies nothing.
 400
 401        ⚠️ **Returns the model, never ``resp.json()``** — same reason
 402        `apply_fabric` gives: on a dict a renamed field reads as falsy,
 403        which here means "your tunnel is down" about a wire change.
 404        """
 405        resp = self.request(
 406            "POST",
 407            "/fabric/probe",
 408            json={"index": index, "peers": peers, "mtu": mtu},
 409        )
 410        resp.raise_for_status()
 411        return _parsed(FabricProbeReport, resp)
 412
 413    def remove_fabric(self, *, index: int, match_network: str = "") -> FabricStatus:
 414        """Take this node off one match's fabric and close its hole.
 415
 416        ⚠️ *match_network* is what lets the node resolve the bridge and
 417        delete the ``DOCKER-USER`` rules by name. Without it the tunnel
 418        goes and the hole stays — open on whatever docker gives that
 419        bridge name next.
 420        """
 421        resp = self.request(
 422            "DELETE",
 423            f"/fabric/{index}",
 424            params={"match_network": match_network},
 425        )
 426        resp.raise_for_status()
 427        return _parsed(FabricStatus, resp)
 428
 429    def stop_instance_service(self, instance_id: str, service: str) -> int:
 430        """Stop one compose service, leaving the instance running.
 431
 432        The exercise format's session close (§4b.4). Returns how many
 433        containers the node stopped — 0 means the challenge declares no
 434        such service, which is a real answer rather than a failure, so
 435        the caller can distinguish it from a cut tunnel.
 436        """
 437        resp = self.request(
 438            "POST",
 439            f"/instances/{instance_id}/stop-service",
 440            json={"service": service},
 441        )
 442        resp.raise_for_status()
 443        body = resp.json()
 444        return int(body.get("stopped", 0)) if isinstance(body, dict) else 0
 445
 446    def capabilities(
 447        self, *, harness_admission: bool = False, agent_ingress: bool = False
 448    ) -> NodeCapabilities:
 449        """What this node can do beyond the original verbs.
 450
 451        ⚠️ **A 404 is an answer, not an error**: a node from before hold /
 452        resume has no such route, and "cannot hold" is exactly what the
 453        platform needs to hear -- the caller then falls back to deleting.
 454        Any other failure raises, because an unreachable node is not a node
 455        that cannot hold.
 456        """
 457        # ⚠️ #2356: ``harness_admission`` makes the node prove it can hold
 458        # an admitted run, which writes its host firewall — asked only by
 459        # the admission paths, never by hold / internet.
 460        # #3020: ``agent_ingress`` makes the node converge its listener,
 461        # asked only before handing a published target over WireGuard.
 462        params = {
 463            **({"harness_admission": "true"} if harness_admission else {}),
 464            **({"agent_ingress": "true"} if agent_ingress else {}),
 465        }
 466        resp = self.request("GET", "/capabilities", params=params or None)
 467        if resp.status_code == 404:
 468            return NodeCapabilities()
 469        resp.raise_for_status()
 470        return masked_capabilities(_parsed(NodeCapabilities, resp), self.disabled_features)
 471
 472    def create_agent_ingress_peer(
 473        self, *, run_id: str, public_key: str, ports: list[int], ttl_s: int
 474    ) -> AgentIngressPeer:
 475        """Add *run_id*'s peer to the node's agent-ingress listener (#3020).
 476
 477        Only the public key crosses; the caller keeps the private half.
 478        Raises on any non-2xx (the node's refusal is in the detail).
 479        """
 480        body = AgentIngressPeerRequest(
 481            run_id=run_id, public_key=public_key, ports=ports, ttl_s=ttl_s
 482        )
 483        resp = self.request("POST", "/agent-ingress/peers", json=body.model_dump(mode="json"))
 484        resp.raise_for_status()
 485        return _parsed(AgentIngressPeer, resp)
 486
 487    def delete_agent_ingress_peer(self, run_id: str) -> bool:
 488        """Revoke *run_id*'s peer; ``False`` when the node held none."""
 489        resp = self.request("DELETE", f"/agent-ingress/peers/{run_id}")
 490        resp.raise_for_status()
 491        return _parsed(AgentIngressPeerRemoved, resp).removed
 492
 493    def hold_instance(self, instance_id: str, mode: str) -> int:
 494        """Pause or stop an instance, removing nothing. Returns containers held."""
 495        resp = self.request("POST", f"/instances/{instance_id}/hold", json={"mode": mode})
 496        resp.raise_for_status()
 497        return _parsed(InstanceHeld, resp).count
 498
 499    def resume_instance(self, instance_id: str) -> int:
 500        """Bring a held instance back. Returns containers brought back."""
 501        resp = self.request("POST", f"/instances/{instance_id}/resume")
 502        resp.raise_for_status()
 503        return _parsed(InstanceHeld, resp).count
 504
 505    def change_instance_internet(
 506        self, instance_id: str, change: InstanceInternetChange
 507    ) -> InstanceInternetChanged:
 508        """Change a running box's internet verdict (#2067).
 509
 510        Raises :class:`NodeRequestError` on any refusal, carrying the node's
 511        ``code`` (:data:`~ctfy.core.exceptions.NODE_INTERNET_CHANGE_FAILED`
 512        when its firewall could not converge — the node has then put its
 513        record back, so nothing changed).
 514        """
 515        resp = self.request(
 516            "POST", f"/instances/{instance_id}/internet", json=change.model_dump(mode="json")
 517        )
 518        _raise_for_node_status(resp)
 519        return _parsed(InstanceInternetChanged, resp)
 520
 521    def stop_all(self) -> None:
 522        resp = self.request("POST", "/admin/stop-all")
 523        resp.raise_for_status()
 524
 525    def put_authored_challenge(self, push: NodeAuthoredPush) -> NodeAuthoredResult:
 526        """Ask the node to write an approved player upload into its catalog.
 527
 528        ⚠️ **The bytes travel here, unlike `put_generated_range`.** That
 529        one sends a scenario because the node can compile it; a
 530        player-authored tree has no compiler and no corpus entry, so the
 531        archive is what there is to send.
 532
 533        The node answers **200 with ``ok: false``** for a refusal, so a
 534        node that declined the archive and a node that could not be
 535        reached stay different facts; only the second raises here.
 536        """
 537        resp = self.request("POST", "/admin/authored", json=push.model_dump(mode="json"))
 538        resp.raise_for_status()
 539        return _parsed(NodeAuthoredResult, resp)
 540
 541    def smoke_authored_challenge(
 542        self, push: NodeAuthoredSmokePush, *, timeout: int | None = None
 543    ) -> NodeAuthoredSmokeResult:
 544        """Ask the node to run an authored challenge's own reference exploit.
 545
 546        ⚠️ **Takes its own timeout.** A smoke run brings a whole estate
 547        up; the client default is sized for a status call and would give
 548        up while the node is still doing what it was asked.
 549        """
 550        kwargs: dict[str, Any] = {"json": push.model_dump(mode="json")}
 551        if timeout is not None:
 552            kwargs["timeout"] = timeout
 553        resp = self.request("POST", "/admin/authored/smoke", retry_on=RETRY_UNSENT, **kwargs)
 554        resp.raise_for_status()
 555        return _parsed(NodeAuthoredSmokeResult, resp)
 556
 557    def smoke_progress(self, run_id: str, since: int = 0) -> NodeSmokeProgress | None:
 558        """The steps and lines a running smoke has produced since *since* (#2145).
 559
 560        ``None`` when the node holds no feed for *run_id* — a node older than
 561        the feed, or a run that never started — which a watcher reads as
 562        *unwatched*. Any other failure raises: the caller decides that a
 563        watcher's trouble must never fail the run it watches.
 564        """
 565        # ⚠️ A short timeout of its own: this is read once a second beside a
 566        # run that may be minutes long, and a read that hangs on a busy worker
 567        # must not hold the watcher past the run it watches.
 568        resp = self.request(
 569            "GET",
 570            f"/admin/authored/smoke/{run_id}/progress",
 571            params={"since": since},
 572            timeout=10,
 573        )
 574        if resp.status_code == 404:
 575            return None
 576        resp.raise_for_status()
 577        return _parsed(NodeSmokeProgress, resp)
 578
 579    def smoke_entrance(self, run_id: str) -> NodeSmokeEntrance:
 580        """The player's entrance to the smoke box held for *run_id* (#2819 W2).
 581
 582        ``pending`` / ``ready`` / ``gone``; a 404 (a node older than the
 583        route) raises, which the caller reads as *this node cannot hold*.
 584        """
 585        resp = self.request("GET", f"/admin/authored/smoke/{run_id}/entrance", timeout=15)
 586        resp.raise_for_status()
 587        return _parsed(NodeSmokeEntrance, resp)
 588
 589    def post_smoke_exploit(self, run_id: str, outcome: NodeSmokeExploitOutcome) -> bool:
 590        """Hand the held smoke box its exploit outcome (#2819 W2).
 591
 592        ``True`` when the node took it; ``False`` when it holds no box for
 593        *run_id* any more (it tore down on its own timeout) or had one.
 594        """
 595        resp = self.request(
 596            "POST",
 597            f"/admin/authored/smoke/{run_id}/exploit",
 598            json=outcome.model_dump(mode="json"),
 599            timeout=30,
 600            # First outcome wins on the node, so a resend is harmless; but a
 601            # read timeout is a post that may have landed (#3162).
 602            retry_on=RETRY_UNSENT,
 603        )
 604        if resp.status_code == 404:
 605            return False
 606        resp.raise_for_status()
 607        return bool(resp.json().get("taken"))
 608
 609    def harness_log(self, run_id: str, after: int = 0) -> NodeAgentLog | None:
 610        """An agent run's console lines numbered above *after* (#2401).
 611
 612        ``None`` when the node holds no console for *run_id* — a node older
 613        than the feed, or a run that never started there — which a watcher
 614        reads as *unwatched*. Any other failure raises: the caller decides
 615        that a watcher's trouble must never fail the run it watches.
 616        """
 617        # ⚠️ A short timeout of its own, as ``smoke_progress``: read once a
 618        # second beside a run that may be hours long.
 619        resp = self.request(
 620            "GET", f"/harness/runs/{run_id}/log", params={"after": after}, timeout=10
 621        )
 622        if resp.status_code == 404:
 623            return None
 624        resp.raise_for_status()
 625        return _parsed(NodeAgentLog, resp)
 626
 627    def harness_result(self, run_id: str) -> HarnessRunResult | None:
 628        """What the node knows of a run's harness call (#3021).
 629
 630        ``None`` when the node has no such route — a node older than the
 631        result store, which can never return a run's outcome after its call
 632        is gone. ``state="unknown"`` is the node answering that it holds
 633        nothing for *run_id*. Any other failure raises: an unreachable node
 634        is neither of those, and the caller bounds how long it waits on one.
 635        """
 636        resp = self.request("GET", f"/harness/runs/{run_id}/result", timeout=30)
 637        if resp.status_code == 404:
 638            return None
 639        resp.raise_for_status()
 640        return _parsed(HarnessRunResult, resp)
 641
 642    def put_generated_range(self, push: NodeRangePush) -> NodeRangeResult:
 643        """Ask the node to build a generated range from its scenario.
 644
 645        ⚠️ **The scenario travels, not the estate.** The corpus is
 646        mounted on the node and the compiler ships inside it, so 70-odd
 647        rendered files never cross this API — and the corpus commit that
 648        rides along is what stops one scenario becoming two different
 649        estates under one id across a fleet.
 650
 651        The node answers **200 with ``ok: false``** for a refusal, so a
 652        node that declined and a node that could not be reached stay
 653        different facts; only the second raises here.
 654        """
 655        resp = self.request("POST", "/admin/ranges", json=push.model_dump(mode="json"))
 656        resp.raise_for_status()
 657        return _parsed(NodeRangeResult, resp)
 658
 659    def verify_generated_range(
 660        self, push: NodeRangeVerifyPush, *, timeout: int | None = None
 661    ) -> NodeRangeVerifyResult:
 662        """Ask the node to build a scenario, solve it, and take it down.
 663
 664        ⚠️ **The longest call in this API**, so the caller passes its own
 665        timeout: the node's budget is ``compose_timeout_s +
 666        solve_timeout_s`` and a client that used the default would give
 667        up while the node was still building — then retry, and start a
 668        second estate beside the first.
 669
 670        ⚠️ A client timeout does **not** leave a range running: the node
 671        tears down in its own ``finally``. What is lost is the answer,
 672        not the host.
 673
 674        The node answers **200 with an ``error``** for a refusal, the
 675        same split ``put_generated_range`` uses; only an unreachable node
 676        raises.
 677        """
 678        kwargs: dict[str, Any] = {"json": push.model_dump(mode="json")}
 679        if timeout is not None:
 680            # httpx takes a per-request override; the client-wide value
 681            # stays whatever the caller constructed this with.
 682            kwargs["timeout"] = timeout
 683        resp = self.request("POST", "/admin/ranges/verify", retry_on=RETRY_UNSENT, **kwargs)
 684        resp.raise_for_status()
 685        return _parsed(NodeRangeVerifyResult, resp)
 686
 687    def cancel_generated_range_verify(self, run_id: str) -> bool:
 688        """Tell the node to stop a verification it is running.
 689
 690        ⚠️ **A miss is success, not a 404 to report.** The cancel races
 691        the run's own finish, so "no such run" means the run already
 692        ended — which is what the operator asked for. The node answers
 693        ``{"cancelled": false}`` for that case rather than raising, and
 694        this returns it so a caller that wants to say "it had already
 695        finished" can.
 696        """
 697        resp = self.request("POST", f"/admin/ranges/verify/{run_id}/cancel")
 698        resp.raise_for_status()
 699        body = resp.json()
 700        return bool(body.get("cancelled"))
 701
 702    def rescan_challenges(self) -> ChallengeReloadResult:
 703        """Tell the node to drop its spec cache and re-scan challenges_dir.
 704
 705        Returns ``{total, added, removed}`` so the platform can report
 706        the per-node outcome of a cluster-wide rescan."""
 707        resp = self.request("POST", "/admin/rescan-challenges")
 708        resp.raise_for_status()
 709        return _parsed(ChallengeReloadResult, resp)
 710
 711    # -- admin pre-build (image cache warming) ------------------------------
 712
 713    def build_challenge(
 714        self,
 715        challenge_id: str,
 716        *,
 717        authored_revision: str = "",
 718        resource_budget: ResourceBudget | None = None,
 719    ) -> ChallengeBuildState:
 720        """Ask the node to pre-build images for *challenge_id*.
 721
 722        Fires-and-returns: the node persists ``status="building"`` and
 723        spawns a daemon thread; the body returned here is that initial
 724        state row. Polling :meth:`get_build_state` is how the platform
 725        learns when it lands on ``built`` / ``failed``. *authored_revision*
 726        is #1835's: the tree the build must be of, fetched if not held.
 727        """
 728        params = {"authored_revision": authored_revision} if authored_revision else None
 729        # #2504: an authored tree pre-builds under this budget, or not at all.
 730        kwargs: dict[str, Any] = {}
 731        if resource_budget is not None:
 732            kwargs["json"] = {"resource_budget": resource_budget.model_dump()}
 733        resp = self.request(
 734            "POST", f"/admin/challenges/{challenge_id}/build", params=params, **kwargs
 735        )
 736        _raise_for_node_status(resp)
 737        return _parsed(ChallengeBuildState, resp)
 738
 739    def build_all_challenges(
 740        self,
 741        challenge_ids: list[str] | None = None,
 742        *,
 743        resource_budget: ResourceBudget | None = None,
 744    ) -> BuildAllQueued:
 745        """Queue background pre-build on the node.
 746
 747        With ``challenge_ids=None`` the node builds every spec it knows.
 748        With an explicit list (the platform scoping a bulk build to a
 749        competition's challenge set) only those are built. Returns
 750        ``{queued, skipped_built, skipped_in_progress}`` so the platform
 751        can report per-node what got picked up. Sequential on the node
 752        side — no fan-out across the corpus.
 753        """
 754        kwargs: dict[str, Any] = {}
 755        body: dict[str, Any] = {}
 756        if challenge_ids is not None:
 757            body["challenge_ids"] = challenge_ids
 758        if resource_budget is not None:
 759            # #2504: what an authored id in the scope pre-builds under.
 760            body["resource_budget"] = resource_budget.model_dump()
 761        if body:
 762            kwargs["json"] = body
 763        resp = self.request("POST", "/admin/challenges/build-all", **kwargs)
 764        _raise_for_node_status(resp)
 765        return _parsed(BuildAllQueued, resp)
 766
 767    def build_push_challenge(
 768        self,
 769        challenge_id: str,
 770        *,
 771        push_ticket: str = "",
 772        deadline_s: int = 0,
 773        cleanup: bool = False,
 774        cache_max_bytes: int = 0,
 775        confine_network: bool = False,
 776        authored_revision: str = "",
 777        run_id: str = "",
 778        resource_budget: ResourceBudget | None = None,
 779    ) -> ChallengeFreezeResult:
 780        """Build one challenge's ``build:`` services and freeze them to the registry.
 781
 782        Layer 2 producer (P0). **Blocks** until the challenge is built +
 783        pushed (or skipped as already frozen), unlike the fire-and-return
 784        :meth:`build_challenge`. ⚠️ *deadline_s* bounds the node's whole
 785        freeze (#3167, `ctfy.core.freeze_budget`; ``0`` sends the node's own
 786        default), and the HTTP wait is derived from it —
 787        `freeze_client_wait_s`, past the node's own answer — so the client
 788        never abandons a freeze the node is still running. Sent once: a
 789        ``ReadTimeout`` is a request the node has (#3162). Returns the per-service freeze result;
 790        the platform's ``challenge_build_push`` task calls this one
 791        challenge at a time. *push_ticket* is the platform's short-lived
 792        registry credential the node's ``docker push`` authenticates with
 793        — the node runs it in a container with no ambient login.
 794        *cleanup* asks the node to remove what the build left and prune its
 795        build cache to *cache_max_bytes* afterwards, and *confine_network*
 796        keeps every ``RUN`` off the host's network (both the freeze pool, #1827).
 797        *run_id* names this call for :meth:`cancel_build_push` and
 798        :meth:`build_push_progress` while it is in flight (#1949).
 799        """
 800        resp = self.request(
 801            "POST",
 802            f"/admin/challenges/{challenge_id}/build-push",
 803            json={
 804                "push_ticket": push_ticket,
 805                "cleanup": cleanup,
 806                "cache_max_bytes": cache_max_bytes,
 807                "confine_network": confine_network,
 808                "authored_revision": authored_revision,
 809                "run_id": run_id,
 810                "deadline_s": deadline_s,
 811                # #2504: the build limits an authored tree is frozen under;
 812                # a node refuses one that arrives without them.
 813                "resource_budget": (
 814                    resource_budget.model_dump() if resource_budget is not None else None
 815                ),
 816            },
 817            timeout=freeze_client_wait_s(deadline_s or freeze_deadline_s()),
 818            retry_on=RETRY_UNSENT,
 819        )
 820        _raise_for_node_status(resp)
 821        return _parsed(ChallengeFreezeResult, resp)
 822
 823    def cancel_build_push(self, run_id: str) -> BuildPushCancelled:
 824        """Cancel the freeze :meth:`build_push_challenge` started under *run_id* (#1949).
 825
 826        Idempotent; ``found`` false means the node holds no such run.
 827        """
 828        resp = self.request("POST", "/admin/build-push/cancel", json={"run_id": run_id})
 829        _raise_for_node_status(resp)
 830        return _parsed(BuildPushCancelled, resp)
 831
 832    def build_push_progress(self, run_id: str) -> BuildPushProgress:
 833        """What the freeze running under *run_id* is doing now (#1949)."""
 834        resp = self.request("GET", f"/admin/build-push/{run_id}/progress")
 835        _raise_for_node_status(resp)
 836        return _parsed(BuildPushProgress, resp)
 837
 838    def get_disk_free(self) -> NodeDiskFree:
 839        """Free space on the node's data filesystem, read now (#1827).
 840
 841        Raises on any failure — including a node predating the route, which
 842        answers 404 — and the freeze reads every failure as *cannot tell*,
 843        which stops it.
 844        """
 845        resp = self.request("GET", "/admin/disk")
 846        _raise_for_node_status(resp)
 847        return _parsed(NodeDiskFree, resp)
 848
 849    def get_build_state(self) -> BuildStateRows:
 850        """Fetch every per-challenge build-state row on this node.
 851
 852        Returns ``{"rows": [{challenge_id, status, built_at, error}, …]}``.
 853        ``status`` is one of ``unbuilt`` / ``building`` / ``built`` /
 854        ``failed``; ``unbuilt`` placeholders are synthesised for specs
 855        the node has seen but never been asked to build.
 856        """
 857        resp = self.request("GET", "/admin/challenges/build-state")
 858        _raise_for_node_status(resp)
 859        return _parsed(BuildStateRows, resp)
 860
 861    def pull_challenge(self, challenge_id: str) -> ChallengePullState:
 862        """Ask the node to pre-pull registry images for *challenge_id*.
 863
 864        Pull-side twin of :meth:`build_challenge`: the node persists
 865        ``status="pulling"`` and spawns a daemon thread; poll
 866        :meth:`get_pull_state` for the ``pulled`` / ``failed`` landing.
 867        """
 868        resp = self.request("POST", f"/admin/challenges/{challenge_id}/pull")
 869        _raise_for_node_status(resp)
 870        return _parsed(ChallengePullState, resp)
 871
 872    def pull_all_challenges(self, challenge_ids: list[str] | None = None) -> PullAllQueued:
 873        """Queue background pre-pull on the node.
 874
 875        With ``challenge_ids=None`` the node pulls every spec it knows;
 876        with an explicit list only those (the platform scoping to a
 877        competition). Returns ``{queued, skipped_pulled,
 878        skipped_in_progress}``.
 879        """
 880        kwargs: dict[str, Any] = {}
 881        if challenge_ids is not None:
 882            kwargs["json"] = {"challenge_ids": challenge_ids}
 883        resp = self.request("POST", "/admin/challenges/pull-all", **kwargs)
 884        _raise_for_node_status(resp)
 885        return _parsed(PullAllQueued, resp)
 886
 887    def get_pull_state(self) -> PullStateRows:
 888        """Fetch every per-challenge pull-state row on this node.
 889
 890        Returns ``{"rows": [{challenge_id, status, pulled_at, error}, …]}``.
 891        ``status`` is one of ``unpulled`` / ``pulling`` / ``pulled`` /
 892        ``failed``; ``unpulled`` placeholders are synthesised for specs
 893        the node has seen but never been asked to pull.
 894        """
 895        resp = self.request("GET", "/admin/challenges/pull-state")
 896        _raise_for_node_status(resp)
 897        return _parsed(PullStateRows, resp)
 898
 899    # -- status / health ----------------------------------------------------
 900
 901    def get_status(self, instance_id: str) -> InstanceStatusReport:
 902        resp = self.request("GET", f"/instances/{instance_id}/status")
 903        resp.raise_for_status()
 904        return _parsed(InstanceStatusReport, resp)
 905
 906    def list_instances(self) -> list[NodeInstanceSummary]:
 907        """Every instance the node still has bookkeeping for.
 908
 909        Used by the platform's boot re-adoption pass to decide which
 910        durable claims still have containers behind them. ``get_status``
 911        answers the same question one id at a time; asking it per claim
 912        would cost one round trip per instance to a node that may be
 913        slow or down, on the path that gates startup.
 914        """
 915        resp = self.request("GET", "/instances")
 916        resp.raise_for_status()
 917        return _parsed(NodeInstanceList, resp).items
 918
 919    def check_health(self, instance_id: str) -> bool:
 920        return self.check_health_verbose(instance_id).healthy
 921
 922    def check_health_verbose(self, instance_id: str) -> HealthReading:
 923        """Health plus whatever agent-deception telemetry the node read.
 924
 925        One call, because the platform already polls this endpoint for
 926        every live instance every tick and a second round trip per
 927        instance would be the whole cost of the feature. ``check_health``
 928        stays a bool for the callers that only ever wanted one.
 929
 930        Never an error, whatever arrives: this rides the health poll,
 931        and telemetry must not be able to mark a healthy instance
 932        unhealthy.
 933
 934        ⚠️ **The two ways this can come back empty are different facts
 935        and used to be one.** An older node sends no ``deception`` key
 936        at all — it *did not say*, which is ``channel=None`` and
 937        authorises nothing. A body this platform cannot parse is
 938        something else: the node did say, and we could not read it. That
 939        is ``unreadable``, the same value the reader gives for a missing
 940        file, because it is the same fact — there is no usable channel
 941        here — arriving through a different door.
 942
 943        The parse failure that matters is version skew in the other
 944        direction: a **newer** node reporting an ``ObservationKind`` this
 945        platform does not have. ``Observation.kind`` is typed, so one
 946        unknown member fails the whole model rather than dropping one
 947        row — which is the right refusal (the node's word is not ours to
 948        edit) and exactly why it must not land as "reported nothing".
 949        """
 950        resp = self.request("GET", f"/instances/{instance_id}/health")
 951        resp.raise_for_status()
 952        body = resp.json()
 953        raw = body.get("deception")
 954        if not raw:
 955            summary = DeceptionSummary()
 956        else:
 957            try:
 958                summary = DeceptionSummary.model_validate(raw)
 959            except ValidationError as exc:
 960                log.info(
 961                    "unreadable deception telemetry from node",
 962                    instance_id=instance_id,
 963                    error=tail(str(exc), limit=200),
 964                )
 965                summary = DeceptionSummary(channel="unreadable")
 966        raw_contact = body.get("contact")
 967        if not raw_contact:
 968            contact = ContactReading()
 969        else:
 970            try:
 971                contact = ContactReading.model_validate(raw_contact)
 972            except ValidationError as exc:
 973                log.info(
 974                    "unreadable contact telemetry from node",
 975                    instance_id=instance_id,
 976                    error=tail(str(exc), limit=200),
 977                )
 978                contact = ContactReading(channel="unreadable")
 979        raw_procs = body.get("new_processes")
 980        # ⚠️ Shape-checked rather than trusted: an older node sends
 981        # nothing and a newer one could send a shape this platform does
 982        # not model, and neither may reach a reader as *a container full
 983        # of intruders*. Anything unreadable is silence, which for a
 984        # signal built to under-report is the honest reading.
 985        procs: dict[str, list[str]] = {}
 986        if isinstance(raw_procs, dict):
 987            for container, names in raw_procs.items():
 988                if isinstance(container, str) and isinstance(names, list):
 989                    procs[container] = [str(n) for n in names]
 990        return HealthReading(
 991            healthy=bool(body.get("is_healthy")),
 992            deception=summary,
 993            contact=contact,
 994            new_processes=procs,
 995        )
 996
 997    def node_health(self) -> dict[str, Any]:
 998        """Liveness + ``{running, capacity}`` for heartbeat."""
 999        resp = self.request("GET", "/health")
1000        resp.raise_for_status()
1001        body: dict[str, Any] = resp.json()
1002        return body
1003
1004    # -- traffic / runtime --------------------------------------------------
1005
1006    def get_traffic_page(
1007        self, instance_id: str, *, cursor: str = "", limit: int = 0
1008    ) -> InstanceTrafficPage:
1009        """One page of mitmproxy flows; the capture lives on the node's FS.
1010
1011        ⚠️ **The one method that must not use ``_parsed``.** A node from
1012        before paging answers with a bare *list* — it sent the whole
1013        capture and understands no cursor — and a strict
1014        ``model_validate`` turns that supported skew into a raised
1015        request. So the body goes through ``coerce_page``, which already
1016        owns every decision about an unexpected shape, and the caller
1017        gets the model.
1018
1019        This is the same move ``coerce_page`` was written for, one layer
1020        down. Its docstring used to say the parsing belonged
1021        "platform-side … so a route and this client cannot come to
1022        disagree" — and there were *three* platform-side calls doing it.
1023        There is one call now, at the seam that knows the request
1024        cursor, which is what the fallback needs; the routes hold the
1025        type instead of re-deciding its shape.
1026
1027        ⚠️ **An empty ``cursor`` means "from the beginning".** Passing
1028        one on every poll re-reads the whole capture, which is the cost
1029        the paged read exists to remove — arriving through the caller
1030        rather than through the node.
1031        """
1032        params: dict[str, Any] = {}
1033        if cursor:
1034            params["cursor"] = cursor
1035        if limit > 0:
1036            params["limit"] = limit
1037        resp = self.request("GET", f"/instances/{instance_id}/traffic", params=params)
1038        resp.raise_for_status()
1039        page = coerce_page(resp.json(), cursor)
1040        return InstanceTrafficPage(
1041            flows=page.flows,
1042            cursor=page.cursor,
1043            truncated=page.truncated,
1044            capture=page.capture,
1045        )
1046
1047    def get_traffic_witnesses(
1048        self, instance_id: str, *, cursor: str = "", limit: int = 0
1049    ) -> InstanceTrafficWitnesses:
1050        """Who spoke to this box since *cursor*, with no bodies moved.
1051
1052        §7.1's correlation reads one gamebox per team per round. Through
1053        :meth:`get_traffic_page` that is every flow, both bodies
1054        included (~100 kB apiece by ``CLAUDE.md``'s measurement),
1055        pulled across the wire to extract a handful of addresses; the
1056        node reduces instead and this carries the result.
1057
1058        ⚠️ **Strict parsing here, unlike its neighbour, and that is the
1059        difference between an old field and an old route.**
1060        ``get_traffic_page`` is lenient because a node from before
1061        paging answers the *same* path with a bare list, which is
1062        supported skew. This route did not exist on such a node at all,
1063        so the honest answer from one is a 404 — and a lenient parse of
1064        an error body would hand the round loop an empty witness set,
1065        which reads as *nobody was there*. The caller catches the
1066        request error and records the round as unknown.
1067        """
1068        params: dict[str, Any] = {}
1069        if cursor:
1070            params["cursor"] = cursor
1071        if limit > 0:
1072            params["limit"] = limit
1073        resp = self.request("GET", f"/instances/{instance_id}/traffic-witnesses", params=params)
1074        resp.raise_for_status()
1075        return _parsed(InstanceTrafficWitnesses, resp)
1076
1077    def list_containers(self, instance_id: str) -> list[ComposeContainer]:
1078        """Enumerate every container in an instance's compose project.
1079
1080        Backs the admin-shell feature: the platform forwards a
1081        ``GET /admin/instances/{id}/containers`` to the assigned node,
1082        which returns one row per container (challenge services and
1083        platform-injected sidecars alike). The WebSocket reverse-proxy
1084        is opened separately and does not go through ``NodeClient``.
1085        """
1086        resp = self.request("GET", f"/instances/{instance_id}/containers")
1087        _raise_for_node_status(resp)
1088        return _parsed_list(ComposeContainer, resp)
1089
1090    def run_checker(
1091        self,
1092        instance_id: str,
1093        *,
1094        service: str,
1095        cmd: list[str],
1096        timeout_s: int = DEFAULT_JUDGE_TIMEOUT_S,
1097    ) -> CheckResult:
1098        """Exec an author-supplied checker in a trusted judge sidecar.
1099
1100        Backs the checker / exec-judge feature: the platform forwards a
1101        verify request to the assigned node, which runs ``cmd`` inside the
1102        named sidecar and returns
1103        ``{exit_code, stdout, stderr, timed_out, error}``.
1104
1105        ⚠️ *timeout_s* is the author's, and the node clamps it (#3169): the
1106        clamped budget is what is sent and what the wait is derived from
1107        (`core.judge_budget.platform_wait_s`), so ``0`` no longer waits 10 s
1108        on a node running 30, and ``600`` no longer holds a threadpool slot
1109        for 610 s on a node that stops at 120. Sent once (``RETRY_UNSENT``).
1110        """
1111        exec_s = judge_exec_s(timeout_s)
1112        resp = self.request(
1113            "POST",
1114            f"/instances/{instance_id}/check",
1115            json={"service": service, "cmd": cmd, "timeout_s": exec_s},
1116            timeout=judge_platform_wait_s(exec_s),
1117            retry_on=RETRY_UNSENT,
1118        )
1119        _raise_for_node_status(resp)
1120        return _parsed(CheckResult, resp)
1121
1122    def run_harness(
1123        self,
1124        *,
1125        image: str,
1126        run_id: str = "",
1127        env: dict[str, str],
1128        timeout_s: int = 1800,
1129        untrusted: bool = False,
1130        network: str = "",
1131        own_network: bool = False,
1132        match_network: str = "",
1133        match_address: str = "",
1134        brief: dict[str, Any] | None = None,
1135        target_instance: str = "",
1136        target_aliases: dict[str, str] | None = None,
1137        target_access: str = "",
1138        vpn_config: str = "",
1139        estate_zones: Sequence[str] = (),
1140        target_endpoints: Sequence[str] = (),
1141        target_hosts: Mapping[str, str] | None = None,
1142        target_datagram_endpoints: Sequence[str] = (),
1143        pull_ticket: str = "",
1144        registry_auth: HarnessRegistryAuth | None = None,
1145        limits: HarnessLimits | None = None,
1146        admission_profile: bool = False,
1147        egress_allow: list[str] | None = None,
1148        internet: bool = True,
1149        rootfs_budget_bytes: int = 0,
1150        unpack_budget_bytes: int = 0,
1151        disk_low_watermark_bytes: int = 0,
1152        ingress: bool = True,
1153        local_ingress: bool = False,
1154        platform_image: bool = False,
1155        image_policy: HarnessImagePolicy | None,
1156        attachments: dict[str, str] | None = None,
1157        keep_work: bool = False,
1158        resume_work: bool = False,
1159        exploit: dict[str, Any] | None = None,
1160    ) -> HarnessRun:
1161        """Run a one-shot evaluation-harness container on the node.
1162
1163        ``exploit`` (#2819 W2) runs a smoke's reference exploit in the fence
1164        instead of an agent (``HarnessExploit``'s fields); sent only when set.
1165
1166        ``image_policy`` (#2608) is **required, with no default**: every
1167        caller says whether this image is a user's (the registry rules the
1168        node re-checks before pulling) or the platform's own (``None``). An
1169        untrusted run without one is refused by the node.
1170
1171        The node ``docker run``s ``image`` with ``env`` injected, waits for it
1172        to exit (bounded by ``timeout_s``), and returns
1173        ``{exit_code, stdout, stderr, timed_out, error}`` — the same shape as
1174        :meth:`run_checker`. The harness writes its JSON rollup to stdout. The
1175        per-call HTTP timeout sits above the node's run budget so the client
1176        doesn't abandon the request before the (long-running) harness returns.
1177
1178        ``untrusted`` runs the image under the node's isolation profile —
1179        off host networking, bounded resources, no capabilities — for an
1180        image the platform did not build. It requires a network: the
1181        node answers 400 without one, because "isolate this" and "I will
1182        not say where the target lives" do not compose.
1183
1184        Any of ``network`` (the target instance's own, when the caller
1185        knows it), ``own_network`` or ``target_instance`` satisfies that.
1186        ⚠️ Most eval dispatches need the second: there the harness
1187        *starts* the target partway through its own run, so at ``docker
1188        run`` time no target network exists and there is no name to
1189        send. The node creates a fresh bridge and removes it afterwards.
1190
1191        ``target_instance`` reverses that ordering, and is what an
1192        `engagement` uses: the platform starts the estate first and names
1193        it here, and the node wires its published containers to the
1194        agent's bridge. ⚠️ An **id**, never a network name — the name is
1195        a fact only the node's daemon can read, and ``docker run``
1196        *creates* whatever string it is handed, so a platform guessing
1197        one starts a run that reaches nothing.
1198        """
1199        resp = self.request(
1200            "POST",
1201            "/harness/run",
1202            json={
1203                "image": image,
1204                "env": env,
1205                "timeout_s": timeout_s,
1206                "untrusted": untrusted,
1207                "network": network,
1208                "own_network": own_network,
1209                "match_network": match_network,
1210                # ⚠️ **The platform-assigned arena address.** The node
1211                # connects the agent to the match network with this
1212                # ``--ip``, so the address is known before the round loop
1213                # asks (§7.1). Empty means "let docker's IPAM pick" — an
1214                # older platform's behaviour, and what keeps a node reading
1215                # this field unchanged for every non-arena dispatch.
1216                "match_address": match_address,
1217                # ⚠️ Sent even when empty, so a node reading it gets the
1218                # same shape either way; the node's default is what makes
1219                # an older platform's dispatch unchanged.
1220                "pull_ticket": pull_ticket,
1221                # #3475: the owner's registry login, sent **only when set**,
1222                # so a node that predates the field sees the body it always did.
1223                **(
1224                    {"registry_auth": registry_auth.model_dump()}
1225                    if registry_auth is not None
1226                    else {}
1227                ),
1228                # ⚠️ ``None`` leaves the node's own defaults in force, which is
1229                # what a dispatch said before the ceilings were settings.
1230                "limits": limits.model_dump() if limits is not None else None,
1231                # ⚠️ Same rule, and empty is a real answer here too: a
1232                # dispatch with no run row (the conformance probe, a dry
1233                # run) has nothing to label, and a label with an empty
1234                # value would make every one of them answer one query.
1235                "run_id": run_id,
1236                # ⚠️ Sent as the already-built manifest. The platform is
1237                # the side that knows the challenge, the competition and
1238                # the run mode; a node deriving its own would be a
1239                # second implementation of a published contract.
1240                "brief": brief,
1241                "target_instance": target_instance,
1242                # ⚠️ **The names the brief published, so they resolve on
1243                # the agent's bridge too.** An estate's hosts answer to a
1244                # per-instance FQDN minted by its own resolver; joined to
1245                # a new network a container answers only to its compose
1246                # *service* name. Sending the mapping is what keeps the
1247                # address the agent was told about true under either
1248                # placement — without it the agent is handed a hostname
1249                # that does not resolve, reaches nothing, and reports an
1250                # empty result, which reads as a model that gave up.
1251                #
1252                # ⚠️ The platform's to send: it computed the surface. A
1253                # node re-deriving it from the rendered compose would be
1254                # a second implementation of the address the brief has
1255                # already promised.
1256                "target_aliases": target_aliases or {},
1257                # #2550: ``{name: base64}`` for ``/ctfy/work/attachments/``.
1258                # ⚠️ Sent only when there are some, so every other dispatch
1259                # is byte-identical to before.
1260                **({"attachments": attachments} if attachments else {}),
1261                # #2550 restarts: sent only when asked for, like the files.
1262                **({"keep_work": True} if keep_work else {}),
1263                **({"resume_work": True} if resume_work else {}),
1264                # #2861 slice B: the target access the platform read off the
1265                # target's node. ⚠️ Sent only when there is one, so every
1266                # other dispatch is byte-identical to before; and only to a
1267                # node whose ``capabilities().harness_access`` said it reads
1268                # them, or to the target's own node (where an older node's
1269                # local read is still right).
1270                **(
1271                    {
1272                        "target_access": target_access,
1273                        "vpn_config": vpn_config,
1274                        "estate_zones": list(estate_zones),
1275                        "target_endpoints": list(target_endpoints),
1276                    }
1277                    if target_access
1278                    else {}
1279                ),
1280                # #3020: names → the target node's ingress address, for a
1281                # published target handed over the agent-ingress tunnel. Sent
1282                # only when there are some, and only to a node whose
1283                # ``capabilities().harness_ingress_access`` said it reads them.
1284                **({"target_hosts": dict(target_hosts)} if target_hosts else {}),
1285                # #3446: the published endpoints that are UDP, which the node
1286                # never TCP-probes. Sent only when there are some.
1287                **(
1288                    {"target_datagram_endpoints": list(target_datagram_endpoints)}
1289                    if target_datagram_endpoints
1290                    else {}
1291                ),
1292                # #2356: ⚠️ sent only when asked for, so a dispatch that does
1293                # not ask is byte-identical to an older platform's — and an
1294                # older node, which would ignore the fields, is never sent
1295                # them: the caller checks ``capabilities().harness_admission``.
1296                **(
1297                    {
1298                        "admission_profile": True,
1299                        "egress_allow": list(egress_allow or []),
1300                        "internet": internet,
1301                        "rootfs_budget_bytes": rootfs_budget_bytes,
1302                        "unpack_budget_bytes": unpack_budget_bytes,
1303                        "disk_low_watermark_bytes": disk_low_watermark_bytes,
1304                        "ingress": ingress,
1305                        # #2819 W1, F2: sent only when asked for.
1306                        **({"local_ingress": True} if local_ingress else {}),
1307                    }
1308                    if admission_profile
1309                    else {}
1310                ),
1311                "image_policy": image_policy.model_dump() if image_policy is not None else None,
1312                # #2608 N4: only for a row with no owner; see the node's request.
1313                **({"platform_image": True} if platform_image else {}),
1314                **({"exploit": exploit} if exploit is not None else {}),
1315            },
1316            timeout=(
1317                timeout_s
1318                # #2819 W2: the exploit's `solution/Dockerfile` builds first.
1319                + (SOLUTION_BUILD_TIMEOUT_S if exploit is not None else 0)
1320                + IMAGE_PULL_TIMEOUT_S
1321                + HARNESS_IMAGE_INSPECT_TIMEOUT_S
1322                + HARNESS_CREATE_TIMEOUT_S
1323                + 30
1324            ),
1325            retry_on=RETRY_UNSENT,
1326        )
1327        _raise_for_node_status(resp)
1328        return _parsed(HarnessRun, resp)
1329
1330    def cancel_harness(self, *, run_id: str) -> int:
1331        """Stop whatever this node is running for ``run_id``; returns how many.
1332
1333        Until this existed a dispatched container was reclaimed by
1334        **nothing but its own timeout** — ``_harness_timeout``'s docstring
1335        says exactly that — so an operator pressing cancel on a
1336        match-long run was asking for something the platform had no way
1337        to deliver.
1338
1339        ⚠️ **Zero is an ordinary answer, not a failure.** The run may have
1340        finished on its own, been dispatched to a different node, or been
1341        lost to a node restart. A caller that read zero as an error would
1342        report a broken platform every time a cancel arrived a second
1343        late.
1344        """
1345        resp = self.request("POST", "/harness/cancel", json={"run_id": run_id})
1346        body = resp.json()
1347        return int(body.get("stopped", 0)) if isinstance(body, dict) else 0
1348
1349    def verify_patch(
1350        self, challenge_id: str, files: dict[str, bytes], *, deadline_s: int
1351    ) -> PatchVerifyResult:
1352        """Build a submitted patch on the node and return its verdict.
1353
1354        ``files`` maps a challenge-relative path to its new bytes; base64
1355        is applied here because a patch target may legitimately be
1356        binary. Returns ``{verdict, detail, applied, steps, error}``.
1357
1358        **An empty ``verdict`` with a populated ``error`` is not a
1359        judgement** — it means the node refused the submission, the
1360        verifier crashed, or the run's deadline ran out. The caller must
1361        retry or retire the lease rather than write a score, since failing
1362        to judge a patch is not the same as judging it unfixed.
1363
1364        ⚠️ *deadline_s* bounds the node's whole run (#3164,
1365        `ctfy.core.patch_budget`), and the HTTP wait is derived from it —
1366        `patch_client_wait_s`, past the node's own answer — so the client
1367        never abandons a request the node is still working on. Sent once:
1368        a ``ReadTimeout`` is a request the node has (#3162).
1369        """
1370        resp = self.request(
1371            "POST",
1372            "/patches/verify",
1373            json={
1374                "challenge_id": challenge_id,
1375                "files": {
1376                    path: base64.b64encode(content).decode() for path, content in files.items()
1377                },
1378                "deadline_s": deadline_s,
1379            },
1380            timeout=patch_client_wait_s(deadline_s),
1381            retry_on=RETRY_UNSENT,
1382        )
1383        _raise_for_node_status(resp)
1384        return _parsed(PatchVerifyResult, resp)
1385
1386    def collect_patch_files(
1387        self, instance_id: str, service: str, paths: list[str]
1388    ) -> dict[str, bytes]:
1389        """Read a player's SSH edits back out of their running box.
1390
1391        ``paths`` are absolute in-container paths the *platform* derived
1392        from the pristine source tree. Returns only what could be read:
1393        a path missing from the reply means the player did not change it,
1394        and the caller falls back to the shipped file rather than
1395        recording a deletion they never made.
1396        """
1397        resp = self.request(
1398            "POST",
1399            f"/instances/{instance_id}/patch-collect",
1400            json={"service": service, "paths": paths},
1401        )
1402        _raise_for_node_status(resp)
1403        body: dict[str, str] = resp.json()
1404        return {path: base64.b64decode(blob) for path, blob in body.items()}
1405
1406    def write_awd_answers(self, boxes: dict[str, dict[str, str]]) -> dict[str, str]:
1407        """Rotate a round's flags into every box of this node, in one call.
1408
1409        ⚠️ **One call for the whole node, never one per box.** §5.2's
1410        capacity red line is 1000 teams x 3 services injected inside
1411        120 s; per-box that is 3000 round trips, batched it is about
1412        fourteen. The signature is the enforcement — a caller cannot
1413        accidentally loop.
1414
1415        Returns ``{instance_id: ""}`` for the boxes written and a reason
1416        for the ones that were not, because one team's unreachable box
1417        must not cost every other team on the node its rotation. An
1418        unrotated box keeps serving last round's flag, which anyone
1419        holding the old value can replay.
1420        """
1421        resp = self.request("POST", "/awd/answers", json={"boxes": boxes})
1422        _raise_for_node_status(resp)
1423        result: dict[str, str] = resp.json()
1424        return result
1425
1426    def probe_awd_service(
1427        self,
1428        *,
1429        instance_id: str,
1430        service_id: str,
1431        tick: int,
1432        flag: str,
1433        previous_flag: str = "",
1434        previous_flag_id: str = "",
1435    ) -> dict[str, str]:
1436        """Run one round's SLA probes against one box, from outside it.
1437
1438        The flags travel *to* the node: a checker that sourced its
1439        expected value from the box it is checking would pass on any box
1440        that returns whatever it was handed.
1441        """
1442        resp = self.request(
1443            "POST",
1444            "/awd/probe",
1445            json={
1446                "instance_id": instance_id,
1447                "service_id": service_id,
1448                "tick": tick,
1449                "flag": flag,
1450                "previous_flag": previous_flag,
1451                "previous_flag_id": previous_flag_id,
1452            },
1453        )
1454        _raise_for_node_status(resp)
1455        result: dict[str, str] = resp.json()
1456        return result
1457
1458    def attack_awd_box(
1459        self,
1460        *,
1461        instance_id: str,
1462        tick: int,
1463        timeout_s: int = 60,
1464        agent_image: str = "",
1465        counter_flag: str = "",
1466        forensic_marker: str = "",
1467        llm_api_key: str = "",
1468    ) -> AwdAttackResult:
1469        """Run this box's own reference exploit against it, once.
1470
1471        Per **box**, not per service: one `run_exploit` discovers every
1472        answer the challenge declares and prints them all.
1473
1474        ``timeout_s`` is sent because the platform is what knows how long
1475        a round is; the node clamps it, because a container held open for
1476        an hour is the node's problem whoever asked for it.
1477
1478        ``forensic_marker`` is §4.3.1's, carried only by the agent
1479        branch: `build_attack_argv` passes no environment at all, so a
1480        marker on the reference exploit would be the one value that
1481        voids `EXPLOIT007`'s guarantee.
1482
1483        ``agent_image`` + ``counter_flag`` switch the node to §4.2b's
1484        agent attacker. ⚠️ **Both or neither** — the node refuses a half
1485        request rather than falling back to the reference exploit, which
1486        would run a different format than the one configured while the
1487        board looked entirely normal.
1488        """
1489        resp = self.request(
1490            "POST",
1491            "/awd/attack",
1492            json={
1493                "instance_id": instance_id,
1494                "tick": tick,
1495                "timeout_s": timeout_s,
1496                "agent_image": agent_image,
1497                "counter_flag": counter_flag,
1498                "forensic_marker": forensic_marker,
1499                "llm_api_key": llm_api_key,
1500            },
1501        )
1502        _raise_for_node_status(resp)
1503        return _parsed(AwdAttackResult, resp)
1504
1505    def get_container_logs(self, instance_id: str) -> str:
1506        """Fetch combined container stdout/stderr for archival."""
1507        resp = self.request("GET", f"/instances/{instance_id}/container-logs")
1508        resp.raise_for_status()
1509        return str(resp.json().get("logs") or "")
1510
1511    def get_pcap(self, instance_id: str) -> PcapCapture:
1512        """Fetch the merged tcpdump capture for *instance_id*.
1513
1514        ``data`` is empty when no capture exists (sidecar disabled,
1515        instance never reached the running state, etc.) so callers can
1516        persist conditionally without try/except.
1517
1518        ⚠️ **It also says what the per-instance cap already dropped**
1519        (#1714), off the node's response headers, and ``None`` when the
1520        node did not say — an older node, which is *not reported*, never
1521        a clean capture.
1522        """
1523        resp = self.request("GET", f"/instances/{instance_id}/pcap")
1524        if resp.status_code == 404:
1525            return PcapCapture(data=b"")
1526        resp.raise_for_status()
1527        return PcapCapture(
1528            data=resp.content,
1529            dropped_bytes=_header_int(resp, CAPTURE_DROPPED_BYTES_HEADER),
1530            dropped_segments=_header_int(resp, CAPTURE_DROPPED_SEGMENTS_HEADER),
1531        )
1532
1533    def get_harness_artifact(self, artifact_id: str) -> bytes:
1534        """One finished run's collected archive, or ``b""``.
1535
1536        ⚠️ **Empty rather than an exception for a 404**, following
1537        :meth:`get_pcap`. The node reaps uncollected archives, and a
1538        platform that comes back after a restart is exactly the caller
1539        that meets one that is gone — a state its own record already
1540        describes, not a transport failure.
1541        """
1542        resp = self.request("GET", f"/harness/artifacts/{artifact_id}")
1543        if resp.status_code == 404:
1544            return b""
1545        resp.raise_for_status()
1546        return resp.content
1547
1548    def delete_harness_artifact(self, artifact_id: str) -> bool:
1549        """Tell the node the archive has been stored and can go.
1550
1551        ⚠️ **Never raises on a transport failure**, because it is called
1552        after the bytes are safely in object storage: turning a cleanup
1553        error into the run's error would replace a success with a
1554        janitorial complaint. The node's reaper collects whatever this
1555        misses.
1556        """
1557        try:
1558            resp = self.request("DELETE", f"/harness/artifacts/{artifact_id}")
1559        except Exception:
1560            return False
1561        return resp.status_code == 200
1562
1563    def get_agent_runtime(self, instance_id: str) -> AgentRuntime:
1564        """Provider-agnostic runtime hints for attaching an agent (proxy URL,
1565        CA PEM, optional Docker-specific names). Replaces the old
1566        sandbox-network shape."""
1567        resp = self.request("GET", f"/instances/{instance_id}/agent-runtime")
1568        resp.raise_for_status()
1569        return _parsed(AgentRuntime, resp)
1570
1571    # -- per-instance rendered attachments ---------------------------------
1572
1573    def list_instance_attachments(self, instance_id: str) -> AttachmentList:
1574        """List per-instance attachments rendered into the node's workdir.
1575
1576        Backs the platform's per-instance attachment endpoint, which
1577        surfaces the team-specific rendered file listing.
1578
1579        ⚠️ This used to hand back the raw dict, under a docstring saying
1580        "caller projects through :class:`AttachmentList` for typing" —
1581        the node's route already declared that model, so the type existed
1582        and this method's signature was the one place it was dropped.
1583        Projecting *here* means there is one place a malformed answer is
1584        caught rather than one per caller."""
1585        resp = self.request("GET", f"/instances/{instance_id}/attachments")
1586        resp.raise_for_status()
1587        return _parsed(AttachmentList, resp)
1588
1589    def get_vpn_config(self, instance_id: str) -> bytes:
1590        """Fetch the per-instance VPN client config from the node.
1591
1592        The node reads the backend's client config out of the
1593        instance's VPN service container (generated on first boot by
1594        the base image's bootstrap) and rewrites its endpoint to the
1595        node's public host:port. Returns the body verbatim — the
1596        platform-side route adds the ``Content-Disposition`` header
1597        before re-emitting to the player, because only it knows the
1598        backend and therefore the filename the player needs.
1599
1600        Raises ``httpx.HTTPStatusError`` on non-2xx — the platform's
1601        proxy route catches 404 and re-emits to the player, anything
1602        else is treated as a 502.
1603        """
1604        resp = self.request("GET", f"/instances/{instance_id}/vpn-config")
1605        resp.raise_for_status()
1606        return resp.content
1607
1608    def get_vpn_access(self, instance_id: str) -> tuple[bytes, tuple[str, ...] | None]:
1609        """The player's VPN config plus the estate's resolver zones (#2861 B).
1610
1611        The config is :meth:`get_vpn_config`'s, byte for byte; the zones ride
1612        a response header the node adds only when asked. ``None`` for the
1613        zones is a node that did not say (older, or docker unreadable there)
1614        — never "no zones". Raises like :meth:`get_vpn_config`.
1615        """
1616        resp = self.request(
1617            "GET", f"/instances/{instance_id}/vpn-config", params={"estate_zones": "true"}
1618        )
1619        resp.raise_for_status()
1620        raw = resp.headers.get("X-Ctfy-Estate-Zones")
1621        return resp.content, (tuple(raw.split()) if raw is not None else None)
1622
1623    def download_instance_attachment(self, instance_id: str, filename: str) -> tuple[bytes, str]:
1624        """Download one per-instance attachment as ``(bytes, content_type)``.
1625
1626        Buffers the whole body — attachments are bounded (typical case
1627        is a 10 KB binary or a small text file). For huge captures the
1628        caller should hand the player a CDN URL instead.
1629
1630        Raises ``httpx.HTTPStatusError`` on non-2xx — the platform's
1631        proxy route catches 404 and re-emits to the player, anything
1632        else is a 502. Invalid names or instance keys raise
1633        ``InvalidAttachmentName`` before any request is sent."""
1634        # The instance is a single route key, never part of the file name.
1635        if component_refusal(instance_id) is not None:
1636            raise InvalidAttachmentName("Invalid attachment instance")
1637        resp = self.request(
1638            "GET",
1639            f"/instances/{quote(instance_id, safe='')}/attachments/{attachment_url_path(filename)}",
1640        )
1641        resp.raise_for_status()
1642        return resp.content, resp.headers.get("content-type", "application/octet-stream")