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
 16
 17import httpx
 18from pydantic import ValidationError
 19
 20from ctfy.challenge.deception import DeceptionSummary
 21from ctfy.core.awd_network import MatchNetwork
 22from ctfy.core.constants import DEFAULT_CLIENT_TIMEOUT
 23from ctfy.core.exceptions import NodeRequestError
 24from ctfy.sdk.base import BaseHttpClient
 25
 26
 27def _raise_for_node_status(resp: httpx.Response) -> None:
 28    """Like ``resp.raise_for_status()`` but preserves the node's error body.
 29
 30    On a 4xx/5xx, pull the FastAPI ``{"detail": ...}`` field (or the raw
 31    body as a fallback) and raise :class:`NodeRequestError` so the real
 32    failure reaches the caller instead of httpx's generic status string.
 33    """
 34    if not resp.is_error:
 35        return
 36    detail = ""
 37    try:
 38        body = resp.json()
 39    except (json.JSONDecodeError, ValueError):
 40        body = None
 41    if isinstance(body, dict):
 42        d = body.get("detail")
 43        if isinstance(d, str):
 44            detail = d
 45        elif d is not None:
 46            detail = json.dumps(d)
 47    if not detail:
 48        detail = (resp.text or "").strip()
 49    raise NodeRequestError(resp.status_code, detail)
 50
 51
 52class NodeClient(BaseHttpClient):
 53    """Thin sync HTTP client; one instance per node URL."""
 54
 55    def __init__(
 56        self,
 57        node_url: str,
 58        token: str,
 59        *,
 60        timeout: int = DEFAULT_CLIENT_TIMEOUT,
 61    ) -> None:
 62        super().__init__(f"{node_url.rstrip('/')}/api/v1", token, timeout=timeout)
 63
 64    # -- lifecycle ----------------------------------------------------------
 65
 66    def start_instance(
 67        self,
 68        *,
 69        challenge_id: str,
 70        instance_id: str,
 71        ttl: int,
 72        answers: dict[str, str],
 73        proxy_output_dir: str | None = None,
 74        env: dict[str, str] | None = None,
 75        publish_gamebox: bool = False,
 76        match_network: MatchNetwork | None = None,
 77    ) -> dict[str, Any]:
 78        resp = self.request(
 79            "POST",
 80            "/instances",
 81            json={
 82                "challenge_id": challenge_id,
 83                "instance_id": instance_id,
 84                "ttl": ttl,
 85                "answers": answers,
 86                "proxy_output_dir": proxy_output_dir,
 87                "env": env or {},
 88                "publish_gamebox": publish_gamebox,
 89                # ``None`` rather than an omitted key: the node's model
 90                # defaults to None either way, and sending it explicitly
 91                # keeps the body's shape stable across both branches.
 92                "match_network": (dataclasses.asdict(match_network) if match_network else None),
 93            },
 94        )
 95        _raise_for_node_status(resp)
 96        body: dict[str, Any] = resp.json()
 97        return body
 98
 99    def stop_instance(self, instance_id: str) -> None:
100        resp = self.request("DELETE", f"/instances/{instance_id}")
101        resp.raise_for_status()
102
103    def stop_instance_service(self, instance_id: str, service: str) -> int:
104        """Stop one compose service, leaving the instance running.
105
106        The exercise format's session close (§4b.4). Returns how many
107        containers the node stopped — 0 means the challenge declares no
108        such service, which is a real answer rather than a failure, so
109        the caller can distinguish it from a cut tunnel.
110        """
111        resp = self.request(
112            "POST",
113            f"/instances/{instance_id}/stop-service",
114            json={"service": service},
115        )
116        resp.raise_for_status()
117        body = resp.json()
118        return int(body.get("stopped", 0)) if isinstance(body, dict) else 0
119
120    def stop_all(self) -> None:
121        resp = self.request("POST", "/admin/stop-all")
122        resp.raise_for_status()
123
124    def rescan_challenges(self) -> dict[str, Any]:
125        """Tell the node to drop its spec cache and re-scan challenges_dir.
126
127        Returns ``{total, added, removed}`` so the platform can report
128        the per-node outcome of a cluster-wide rescan."""
129        resp = self.request("POST", "/admin/rescan-challenges")
130        resp.raise_for_status()
131        body: dict[str, Any] = resp.json()
132        return body
133
134    # -- admin pre-build (image cache warming) ------------------------------
135
136    def build_challenge(self, challenge_id: str) -> dict[str, Any]:
137        """Ask the node to pre-build images for *challenge_id*.
138
139        Fires-and-returns: the node persists ``status="building"`` and
140        spawns a daemon thread; the body returned here is that initial
141        state row. Polling :meth:`get_build_state` is how the platform
142        learns when it lands on ``built`` / ``failed``.
143        """
144        resp = self.request("POST", f"/admin/challenges/{challenge_id}/build")
145        _raise_for_node_status(resp)
146        body: dict[str, Any] = resp.json()
147        return body
148
149    def build_all_challenges(self, challenge_ids: list[str] | None = None) -> dict[str, Any]:
150        """Queue background pre-build on the node.
151
152        With ``challenge_ids=None`` the node builds every spec it knows.
153        With an explicit list (the platform scoping a bulk build to a
154        competition's challenge set) only those are built. Returns
155        ``{queued, skipped_built, skipped_in_progress}`` so the platform
156        can report per-node what got picked up. Sequential on the node
157        side — no fan-out across the corpus.
158        """
159        kwargs: dict[str, Any] = {}
160        if challenge_ids is not None:
161            kwargs["json"] = {"challenge_ids": challenge_ids}
162        resp = self.request("POST", "/admin/challenges/build-all", **kwargs)
163        _raise_for_node_status(resp)
164        body: dict[str, Any] = resp.json()
165        return body
166
167    def get_build_state(self) -> dict[str, Any]:
168        """Fetch every per-challenge build-state row on this node.
169
170        Returns ``{"rows": [{challenge_id, status, built_at, error}, …]}``.
171        ``status`` is one of ``unbuilt`` / ``building`` / ``built`` /
172        ``failed``; ``unbuilt`` placeholders are synthesised for specs
173        the node has seen but never been asked to build.
174        """
175        resp = self.request("GET", "/admin/challenges/build-state")
176        _raise_for_node_status(resp)
177        body: dict[str, Any] = resp.json()
178        return body
179
180    def pull_challenge(self, challenge_id: str) -> dict[str, Any]:
181        """Ask the node to pre-pull registry images for *challenge_id*.
182
183        Pull-side twin of :meth:`build_challenge`: the node persists
184        ``status="pulling"`` and spawns a daemon thread; poll
185        :meth:`get_pull_state` for the ``pulled`` / ``failed`` landing.
186        """
187        resp = self.request("POST", f"/admin/challenges/{challenge_id}/pull")
188        _raise_for_node_status(resp)
189        body: dict[str, Any] = resp.json()
190        return body
191
192    def pull_all_challenges(self, challenge_ids: list[str] | None = None) -> dict[str, Any]:
193        """Queue background pre-pull on the node.
194
195        With ``challenge_ids=None`` the node pulls every spec it knows;
196        with an explicit list only those (the platform scoping to a
197        competition). Returns ``{queued, skipped_pulled,
198        skipped_in_progress}``.
199        """
200        kwargs: dict[str, Any] = {}
201        if challenge_ids is not None:
202            kwargs["json"] = {"challenge_ids": challenge_ids}
203        resp = self.request("POST", "/admin/challenges/pull-all", **kwargs)
204        _raise_for_node_status(resp)
205        body: dict[str, Any] = resp.json()
206        return body
207
208    def get_pull_state(self) -> dict[str, Any]:
209        """Fetch every per-challenge pull-state row on this node.
210
211        Returns ``{"rows": [{challenge_id, status, pulled_at, error}, …]}``.
212        ``status`` is one of ``unpulled`` / ``pulling`` / ``pulled`` /
213        ``failed``; ``unpulled`` placeholders are synthesised for specs
214        the node has seen but never been asked to pull.
215        """
216        resp = self.request("GET", "/admin/challenges/pull-state")
217        _raise_for_node_status(resp)
218        body: dict[str, Any] = resp.json()
219        return body
220
221    # -- status / health ----------------------------------------------------
222
223    def get_status(self, instance_id: str) -> dict[str, Any]:
224        resp = self.request("GET", f"/instances/{instance_id}/status")
225        resp.raise_for_status()
226        body: dict[str, Any] = resp.json()
227        return body
228
229    def list_instances(self) -> list[dict[str, Any]]:
230        """Every instance the node still has bookkeeping for.
231
232        Used by the platform's boot re-adoption pass to decide which
233        durable claims still have containers behind them. ``get_status``
234        answers the same question one id at a time; asking it per claim
235        would cost one round trip per instance to a node that may be
236        slow or down, on the path that gates startup.
237        """
238        resp = self.request("GET", "/instances")
239        resp.raise_for_status()
240        items: list[dict[str, Any]] = resp.json().get("items", [])
241        return items
242
243    def check_health(self, instance_id: str) -> bool:
244        return self.check_health_verbose(instance_id)[0]
245
246    def check_health_verbose(self, instance_id: str) -> tuple[bool, DeceptionSummary]:
247        """Health plus whatever agent-deception telemetry the node read.
248
249        One call, because the platform already polls this endpoint for
250        every live instance every tick and a second round trip per
251        instance would be the whole cost of the feature. ``check_health``
252        stays a bool for the callers that only ever wanted one.
253
254        An older node sends no ``deception`` key, and a node whose read
255        failed sends null — both are an empty summary, never an error:
256        this rides the health poll, and telemetry must not be able to
257        mark a healthy instance unhealthy.
258        """
259        resp = self.request("GET", f"/instances/{instance_id}/health")
260        resp.raise_for_status()
261        body = resp.json()
262        raw = body.get("deception")
263        try:
264            summary = DeceptionSummary.model_validate(raw) if raw else DeceptionSummary()
265        except ValidationError:
266            summary = DeceptionSummary()
267        return bool(body.get("is_healthy")), summary
268
269    def node_health(self) -> dict[str, Any]:
270        """Liveness + ``{running, capacity}`` for heartbeat."""
271        resp = self.request("GET", "/health")
272        resp.raise_for_status()
273        body: dict[str, Any] = resp.json()
274        return body
275
276    # -- traffic / runtime --------------------------------------------------
277
278    def get_traffic_page(
279        self, instance_id: str, *, cursor: str = "", limit: int = 0
280    ) -> dict[str, Any]:
281        """One page of mitmproxy flows; the capture lives on the node's FS.
282
283        Returns the node's envelope verbatim — ``{flows, cursor,
284        truncated}``. It is deliberately *not* parsed here: this is the
285        transport, and every decision about an unexpected shape belongs
286        in one place platform-side (``traffic_tail.coerce_page``), so a
287        route and this client cannot come to disagree about what a
288        missing cursor means.
289
290        ⚠️ **An empty ``cursor`` means "from the beginning".** Passing
291        one on every poll re-reads the whole capture, which is the cost
292        the paged read exists to remove — arriving through the caller
293        rather than through the node.
294        """
295        params: dict[str, Any] = {}
296        if cursor:
297            params["cursor"] = cursor
298        if limit > 0:
299            params["limit"] = limit
300        resp = self.request("GET", f"/instances/{instance_id}/traffic", params=params)
301        resp.raise_for_status()
302        body: dict[str, Any] = resp.json()
303        return body
304
305    def list_containers(self, instance_id: str) -> list[dict[str, Any]]:
306        """Enumerate every container in an instance's compose project.
307
308        Backs the admin-shell feature: the platform forwards a
309        ``GET /admin/instances/{id}/containers`` to the assigned node,
310        which returns one row per container (challenge services and
311        platform-injected sidecars alike). The WebSocket reverse-proxy
312        is opened separately and does not go through ``NodeClient``.
313        """
314        resp = self.request("GET", f"/instances/{instance_id}/containers")
315        _raise_for_node_status(resp)
316        body: list[dict[str, Any]] = resp.json()
317        return body
318
319    def run_checker(
320        self, instance_id: str, *, service: str, cmd: list[str], timeout_s: int = 30
321    ) -> dict[str, Any]:
322        """Exec an author-supplied checker in a trusted judge sidecar.
323
324        Backs the checker / exec-judge feature: the platform forwards a
325        verify request to the assigned node, which runs ``cmd`` inside the
326        named sidecar and returns
327        ``{exit_code, stdout, stderr, timed_out, error}``. The per-call HTTP
328        timeout sits above the node's exec budget so the client doesn't
329        abandon the request before the node returns.
330        """
331        resp = self.request(
332            "POST",
333            f"/instances/{instance_id}/check",
334            json={"service": service, "cmd": cmd, "timeout_s": timeout_s},
335            timeout=timeout_s + 10,
336        )
337        _raise_for_node_status(resp)
338        body: dict[str, Any] = resp.json()
339        return body
340
341    def run_harness(
342        self, *, image: str, env: dict[str, str], timeout_s: int = 1800
343    ) -> dict[str, Any]:
344        """Run a one-shot evaluation-harness container on the node.
345
346        The node ``docker run``s ``image`` with ``env`` injected, waits for it
347        to exit (bounded by ``timeout_s``), and returns
348        ``{exit_code, stdout, stderr, timed_out, error}`` — the same shape as
349        :meth:`run_checker`. The harness writes its JSON rollup to stdout. The
350        per-call HTTP timeout sits above the node's run budget so the client
351        doesn't abandon the request before the (long-running) harness returns.
352        """
353        resp = self.request(
354            "POST",
355            "/harness/run",
356            json={"image": image, "env": env, "timeout_s": timeout_s},
357            timeout=timeout_s + 30,
358        )
359        _raise_for_node_status(resp)
360        body: dict[str, Any] = resp.json()
361        return body
362
363    def verify_patch(
364        self, challenge_id: str, files: dict[str, bytes], *, timeout_s: int = 900
365    ) -> dict[str, Any]:
366        """Build a submitted patch on the node and return its verdict.
367
368        ``files`` maps a challenge-relative path to its new bytes; base64
369        is applied here because a patch target may legitimately be
370        binary. Returns ``{verdict, detail, applied, steps, error}``.
371
372        **An empty ``verdict`` with a populated ``error`` is not a
373        judgement** — it means the node refused the submission or the
374        verifier crashed. The caller must retry or retire the lease
375        rather than write a score, since failing to judge a patch is not
376        the same as judging it unfixed.
377
378        The per-call HTTP timeout sits above the node's own budget so the
379        client doesn't abandon the request while the node is still
380        building.
381        """
382        resp = self.request(
383            "POST",
384            "/patches/verify",
385            json={
386                "challenge_id": challenge_id,
387                "files": {
388                    path: base64.b64encode(content).decode() for path, content in files.items()
389                },
390                "timeout_s": timeout_s,
391            },
392            timeout=timeout_s + 60,
393        )
394        _raise_for_node_status(resp)
395        body: dict[str, Any] = resp.json()
396        return body
397
398    def collect_patch_files(
399        self, instance_id: str, service: str, paths: list[str]
400    ) -> dict[str, bytes]:
401        """Read a player's SSH edits back out of their running box.
402
403        ``paths`` are absolute in-container paths the *platform* derived
404        from the pristine source tree. Returns only what could be read:
405        a path missing from the reply means the player did not change it,
406        and the caller falls back to the shipped file rather than
407        recording a deletion they never made.
408        """
409        resp = self.request(
410            "POST",
411            f"/instances/{instance_id}/patch-collect",
412            json={"service": service, "paths": paths},
413        )
414        _raise_for_node_status(resp)
415        body: dict[str, str] = resp.json()
416        return {path: base64.b64decode(blob) for path, blob in body.items()}
417
418    def write_awd_answers(self, boxes: dict[str, dict[str, str]]) -> dict[str, str]:
419        """Rotate a round's flags into every box of this node, in one call.
420
421        ⚠️ **One call for the whole node, never one per box.** §5.2's
422        capacity red line is 1000 teams x 3 services injected inside
423        120 s; per-box that is 3000 round trips, batched it is about
424        fourteen. The signature is the enforcement — a caller cannot
425        accidentally loop.
426
427        Returns ``{instance_id: ""}`` for the boxes written and a reason
428        for the ones that were not, because one team's unreachable box
429        must not cost every other team on the node its rotation. An
430        unrotated box keeps serving last round's flag, which anyone
431        holding the old value can replay.
432        """
433        resp = self.request("POST", "/awd/answers", json={"boxes": boxes})
434        _raise_for_node_status(resp)
435        result: dict[str, str] = resp.json()
436        return result
437
438    def probe_awd_service(
439        self,
440        *,
441        instance_id: str,
442        service_id: str,
443        tick: int,
444        flag: str,
445        previous_flag: str = "",
446        previous_flag_id: str = "",
447    ) -> dict[str, str]:
448        """Run one round's SLA probes against one box, from outside it.
449
450        The flags travel *to* the node: a checker that sourced its
451        expected value from the box it is checking would pass on any box
452        that returns whatever it was handed.
453        """
454        resp = self.request(
455            "POST",
456            "/awd/probe",
457            json={
458                "instance_id": instance_id,
459                "service_id": service_id,
460                "tick": tick,
461                "flag": flag,
462                "previous_flag": previous_flag,
463                "previous_flag_id": previous_flag_id,
464            },
465        )
466        _raise_for_node_status(resp)
467        result: dict[str, str] = resp.json()
468        return result
469
470    def get_container_logs(self, instance_id: str) -> str:
471        """Fetch combined container stdout/stderr for archival."""
472        resp = self.request("GET", f"/instances/{instance_id}/container-logs")
473        resp.raise_for_status()
474        return str(resp.json().get("logs") or "")
475
476    def get_pcap(self, instance_id: str) -> bytes:
477        """Fetch the raw tcpdump capture for *instance_id*.
478
479        Returns an empty ``bytes`` when no capture exists (sidecar
480        disabled, instance never reached the running state, etc.) so
481        callers can persist conditionally without try/except.
482        """
483        resp = self.request("GET", f"/instances/{instance_id}/pcap")
484        if resp.status_code == 404:
485            return b""
486        resp.raise_for_status()
487        return resp.content
488
489    def get_agent_runtime(self, instance_id: str) -> dict[str, Any]:
490        """Provider-agnostic runtime hints for attaching an agent (proxy URL,
491        CA PEM, optional Docker-specific names). Replaces the old
492        sandbox-network shape."""
493        resp = self.request("GET", f"/instances/{instance_id}/agent-runtime")
494        resp.raise_for_status()
495        body: dict[str, Any] = resp.json()
496        return body
497
498    # -- per-instance rendered attachments ---------------------------------
499
500    def list_instance_attachments(self, instance_id: str) -> dict[str, Any]:
501        """List per-instance attachments rendered into the node's workdir.
502
503        Returns the raw JSON dict — caller projects through
504        :class:`AttachmentList` for typing. Used by the platform's
505        per-instance attachment endpoint to surface team-specific
506        rendered file listings."""
507        resp = self.request("GET", f"/instances/{instance_id}/attachments")
508        resp.raise_for_status()
509        body: dict[str, Any] = resp.json()
510        return body
511
512    def get_vpn_config(self, instance_id: str) -> bytes:
513        """Fetch the per-instance VPN client config from the node.
514
515        The node reads the backend's client config out of the
516        instance's VPN service container (generated on first boot by
517        the base image's bootstrap) and rewrites its endpoint to the
518        node's public host:port. Returns the body verbatim — the
519        platform-side route adds the ``Content-Disposition`` header
520        before re-emitting to the player, because only it knows the
521        backend and therefore the filename the player needs.
522
523        Raises ``httpx.HTTPStatusError`` on non-2xx — the platform's
524        proxy route catches 404 and re-emits to the player, anything
525        else is treated as a 502.
526        """
527        resp = self.request("GET", f"/instances/{instance_id}/vpn-config")
528        resp.raise_for_status()
529        return resp.content
530
531    def download_instance_attachment(self, instance_id: str, filename: str) -> tuple[bytes, str]:
532        """Download one per-instance attachment as ``(bytes, content_type)``.
533
534        Buffers the whole body — attachments are bounded (typical case
535        is a 10 KB binary or a small text file). For huge captures the
536        caller should hand the player a CDN URL instead.
537
538        Raises ``httpx.HTTPStatusError`` on non-2xx — the platform's
539        proxy route catches 404 and re-emits to the player, anything
540        else is a 502."""
541        resp = self.request(
542            "GET",
543            f"/instances/{instance_id}/attachments/{filename}",
544        )
545        resp.raise_for_status()
546        return resp.content, resp.headers.get("content-type", "application/octet-stream")