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 json
 13from typing import Any
 14
 15import httpx
 16
 17from ctfy.core.constants import DEFAULT_CLIENT_TIMEOUT
 18from ctfy.core.exceptions import NodeRequestError
 19from ctfy.sdk.base import BaseHttpClient
 20
 21
 22def _raise_for_node_status(resp: httpx.Response) -> None:
 23    """Like ``resp.raise_for_status()`` but preserves the node's error body.
 24
 25    On a 4xx/5xx, pull the FastAPI ``{"detail": ...}`` field (or the raw
 26    body as a fallback) and raise :class:`NodeRequestError` so the real
 27    failure reaches the caller instead of httpx's generic status string.
 28    """
 29    if not resp.is_error:
 30        return
 31    detail = ""
 32    try:
 33        body = resp.json()
 34    except (json.JSONDecodeError, ValueError):
 35        body = None
 36    if isinstance(body, dict):
 37        d = body.get("detail")
 38        if isinstance(d, str):
 39            detail = d
 40        elif d is not None:
 41            detail = json.dumps(d)
 42    if not detail:
 43        detail = (resp.text or "").strip()
 44    raise NodeRequestError(resp.status_code, detail)
 45
 46
 47class NodeClient(BaseHttpClient):
 48    """Thin sync HTTP client; one instance per node URL."""
 49
 50    def __init__(
 51        self,
 52        node_url: str,
 53        token: str,
 54        *,
 55        timeout: int = DEFAULT_CLIENT_TIMEOUT,
 56    ) -> None:
 57        super().__init__(f"{node_url.rstrip('/')}/api/v1", token, timeout=timeout)
 58
 59    # -- lifecycle ----------------------------------------------------------
 60
 61    def start_instance(
 62        self,
 63        *,
 64        challenge_id: str,
 65        instance_id: str,
 66        ttl: int,
 67        answers: dict[str, str],
 68        proxy_output_dir: str | None = None,
 69    ) -> dict[str, Any]:
 70        resp = self.request(
 71            "POST",
 72            "/instances",
 73            json={
 74                "challenge_id": challenge_id,
 75                "instance_id": instance_id,
 76                "ttl": ttl,
 77                "answers": answers,
 78                "proxy_output_dir": proxy_output_dir,
 79            },
 80        )
 81        _raise_for_node_status(resp)
 82        body: dict[str, Any] = resp.json()
 83        return body
 84
 85    def stop_instance(self, instance_id: str) -> None:
 86        resp = self.request("DELETE", f"/instances/{instance_id}")
 87        resp.raise_for_status()
 88
 89    def stop_all(self) -> None:
 90        resp = self.request("POST", "/admin/stop-all")
 91        resp.raise_for_status()
 92
 93    def rescan_challenges(self) -> dict[str, Any]:
 94        """Tell the node to drop its spec cache and re-scan challenges_dir.
 95
 96        Returns ``{total, added, removed}`` so the platform can report
 97        the per-node outcome of a cluster-wide rescan."""
 98        resp = self.request("POST", "/admin/rescan-challenges")
 99        resp.raise_for_status()
100        body: dict[str, Any] = resp.json()
101        return body
102
103    # -- admin pre-build (image cache warming) ------------------------------
104
105    def build_challenge(self, challenge_id: str) -> dict[str, Any]:
106        """Ask the node to pre-build images for *challenge_id*.
107
108        Fires-and-returns: the node persists ``status="building"`` and
109        spawns a daemon thread; the body returned here is that initial
110        state row. Polling :meth:`get_build_state` is how the platform
111        learns when it lands on ``built`` / ``failed``.
112        """
113        resp = self.request("POST", f"/admin/challenges/{challenge_id}/build")
114        _raise_for_node_status(resp)
115        body: dict[str, Any] = resp.json()
116        return body
117
118    def build_all_challenges(self) -> dict[str, Any]:
119        """Queue background pre-build for every spec the node knows about.
120
121        Returns ``{queued, skipped_built, skipped_in_progress}`` so the
122        platform can report per-node what got picked up. Sequential
123        on the node side — no fan-out across the corpus.
124        """
125        resp = self.request("POST", "/admin/challenges/build-all")
126        _raise_for_node_status(resp)
127        body: dict[str, Any] = resp.json()
128        return body
129
130    def get_build_state(self) -> dict[str, Any]:
131        """Fetch every per-challenge build-state row on this node.
132
133        Returns ``{"rows": [{challenge_id, status, built_at, error}, …]}``.
134        ``status`` is one of ``unbuilt`` / ``building`` / ``built`` /
135        ``failed``; ``unbuilt`` placeholders are synthesised for specs
136        the node has seen but never been asked to build.
137        """
138        resp = self.request("GET", "/admin/challenges/build-state")
139        _raise_for_node_status(resp)
140        body: dict[str, Any] = resp.json()
141        return body
142
143    # -- status / health ----------------------------------------------------
144
145    def get_status(self, instance_id: str) -> dict[str, Any]:
146        resp = self.request("GET", f"/instances/{instance_id}/status")
147        resp.raise_for_status()
148        body: dict[str, Any] = resp.json()
149        return body
150
151    def check_health(self, instance_id: str) -> bool:
152        resp = self.request("GET", f"/instances/{instance_id}/health")
153        resp.raise_for_status()
154        return bool(resp.json().get("is_healthy"))
155
156    def node_health(self) -> dict[str, Any]:
157        """Liveness + ``{running, capacity}`` for heartbeat."""
158        resp = self.request("GET", "/health")
159        resp.raise_for_status()
160        body: dict[str, Any] = resp.json()
161        return body
162
163    # -- traffic / runtime --------------------------------------------------
164
165    def get_traffic(self, instance_id: str) -> dict[str, Any]:
166        """Fetch mitmproxy flow data; file lives on the node's FS."""
167        resp = self.request("GET", f"/instances/{instance_id}/traffic")
168        resp.raise_for_status()
169        body: dict[str, Any] = resp.json()
170        return body
171
172    def list_containers(self, instance_id: str) -> list[dict[str, Any]]:
173        """Enumerate every container in an instance's compose project.
174
175        Backs the admin-shell feature: the platform forwards a
176        ``GET /admin/instances/{id}/containers`` to the assigned node,
177        which returns one row per container (challenge services and
178        platform-injected sidecars alike). The WebSocket reverse-proxy
179        is opened separately and does not go through ``NodeClient``.
180        """
181        resp = self.request("GET", f"/instances/{instance_id}/containers")
182        _raise_for_node_status(resp)
183        body: list[dict[str, Any]] = resp.json()
184        return body
185
186    def get_container_logs(self, instance_id: str) -> str:
187        """Fetch combined container stdout/stderr for archival."""
188        resp = self.request("GET", f"/instances/{instance_id}/container-logs")
189        resp.raise_for_status()
190        return str(resp.json().get("logs") or "")
191
192    def get_pcap(self, instance_id: str) -> bytes:
193        """Fetch the raw tcpdump capture for *instance_id*.
194
195        Returns an empty ``bytes`` when no capture exists (sidecar
196        disabled, instance never reached the running state, etc.) so
197        callers can persist conditionally without try/except.
198        """
199        resp = self.request("GET", f"/instances/{instance_id}/pcap")
200        if resp.status_code == 404:
201            return b""
202        resp.raise_for_status()
203        return resp.content
204
205    def get_agent_runtime(self, instance_id: str) -> dict[str, Any]:
206        """Provider-agnostic runtime hints for attaching an agent (proxy URL,
207        CA PEM, optional Docker-specific names). Replaces the old
208        sandbox-network shape."""
209        resp = self.request("GET", f"/instances/{instance_id}/agent-runtime")
210        resp.raise_for_status()
211        body: dict[str, Any] = resp.json()
212        return body
213
214    # -- per-instance rendered attachments ---------------------------------
215
216    def list_instance_attachments(self, instance_id: str) -> dict[str, Any]:
217        """List per-instance attachments rendered into the node's workdir.
218
219        Returns the raw JSON dict — caller projects through
220        :class:`AttachmentList` for typing. Used by the platform's
221        per-instance attachment endpoint to surface team-specific
222        rendered file listings."""
223        resp = self.request("GET", f"/instances/{instance_id}/attachments")
224        resp.raise_for_status()
225        body: dict[str, Any] = resp.json()
226        return body
227
228    def get_openvpn_config(self, instance_id: str) -> bytes:
229        """Fetch the per-instance OpenVPN client config from the node.
230
231        The node reads ``/etc/openvpn/client.ovpn`` out of the instance's
232        ``openvpn`` service container (generated on first boot by the
233        ``ctfy/openvpn-base`` image's bootstrap). Returns the body
234        verbatim — the platform-side route adds the
235        ``Content-Disposition`` header before re-emitting to the player.
236
237        Raises ``httpx.HTTPStatusError`` on non-2xx — the platform's
238        proxy route catches 404 and re-emits to the player, anything
239        else is treated as a 502.
240        """
241        resp = self.request("GET", f"/instances/{instance_id}/openvpn-config")
242        resp.raise_for_status()
243        return resp.content
244
245    def download_instance_attachment(self, instance_id: str, filename: str) -> tuple[bytes, str]:
246        """Download one per-instance attachment as ``(bytes, content_type)``.
247
248        Buffers the whole body — attachments are bounded (typical case
249        is a 10 KB binary or a small text file). For huge captures the
250        caller should hand the player a CDN URL instead.
251
252        Raises ``httpx.HTTPStatusError`` on non-2xx — the platform's
253        proxy route catches 404 and re-emits to the player, anything
254        else is a 502."""
255        resp = self.request(
256            "GET",
257            f"/instances/{instance_id}/attachments/{filename}",
258        )
259        resp.raise_for_status()
260        return resp.content, resp.headers.get("content-type", "application/octet-stream")