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