ctfy.sdk

 73class AdminClient:
 74    """Admin / operator surface, grouped into resource namespaces.
 75
 76    Namespaces: :attr:`users`, :attr:`competitions`, :attr:`announcements`,
 77    :attr:`achievements`, :attr:`challenges`, :attr:`instances`,
 78    :attr:`records`, :attr:`nodes`, :attr:`observability`, :attr:`settings`,
 79    :attr:`competition_admins`.
 80    """
 81
 82    def __init__(self, http: BaseHttpClient) -> None:
 83        #: Shared transport (the parent PlatformClient when reached via
 84        #: ``client.admin``, or an owned client when built via ``connect``).
 85        self._http = http
 86
 87    @classmethod
 88    def connect(
 89        cls,
 90        server_url: str,
 91        token: str = "",
 92        *,
 93        max_retries: int = 3,
 94        timeout: int = DEFAULT_CLIENT_TIMEOUT,
 95    ) -> AdminClient:
 96        """Build an ``AdminClient`` that owns its own transport.
 97
 98        For operator tooling that doesn't already hold a
 99        :class:`PlatformClient`. The returned client is a context manager
100        that closes its transport on exit.
101        """
102        base = server_url.rstrip("/")
103        http = BaseHttpClient(f"{base}/api/v1", token, timeout=timeout, max_retries=max_retries)
104        return cls(http)
105
106    @cached_property
107    def users(self) -> AdminUsersResource:
108        return AdminUsersResource(self._http)
109
110    @cached_property
111    def competitions(self) -> AdminCompetitionsResource:
112        return AdminCompetitionsResource(self._http)
113
114    @cached_property
115    def announcements(self) -> AdminAnnouncementsResource:
116        return AdminAnnouncementsResource(self._http)
117
118    @cached_property
119    def achievements(self) -> AdminAchievementsResource:
120        return AdminAchievementsResource(self._http)
121
122    @cached_property
123    def challenges(self) -> AdminChallengesResource:
124        return AdminChallengesResource(self._http)
125
126    @cached_property
127    def email(self) -> AdminEmailResource:
128        return AdminEmailResource(self._http)
129
130    @cached_property
131    def patches(self) -> AdminPatchesResource:
132        return AdminPatchesResource(self._http)
133
134    @cached_property
135    def awd(self) -> AdminAwdResource:
136        return AdminAwdResource(self._http)
137
138    @cached_property
139    def series(self) -> AdminSeriesResource:
140        return AdminSeriesResource(self._http)
141
142    @cached_property
143    def instances(self) -> AdminInstancesResource:
144        return AdminInstancesResource(self._http)
145
146    @cached_property
147    def records(self) -> AdminRecordsResource:
148        return AdminRecordsResource(self._http)
149
150    @cached_property
151    def nodes(self) -> AdminNodesResource:
152        return AdminNodesResource(self._http)
153
154    @cached_property
155    def observability(self) -> AdminObservabilityResource:
156        return AdminObservabilityResource(self._http)
157
158    @cached_property
159    def settings(self) -> AdminSettingsResource:
160        return AdminSettingsResource(self._http)
161
162    @cached_property
163    def competition_admins(self) -> AdminCompetitionAdminsResource:
164        return AdminCompetitionAdminsResource(self._http)
165
166    @cached_property
167    def competition_invites(self) -> AdminCompetitionInvitesResource:
168        return AdminCompetitionInvitesResource(self._http)
169
170    @cached_property
171    def registrations(self) -> AdminRegistrationsResource:
172        """Entrant roster, eligibility review, and CSV export."""
173        return AdminRegistrationsResource(self._http)
174
175    @cached_property
176    def teams(self) -> AdminTeamsResource:
177        """Enforcement against a squad — disqualify and reinstate.
178
179        Separate from ``registrations`` on purpose: that one rules on
180        eligibility (which decides prizes), this one decides whether the
181        team is in the event at all.
182        """
183        return AdminTeamsResource(self._http)
184
185    @cached_property
186    def tasks(self) -> AdminTasksResource:
187        return AdminTasksResource(self._http)
188
189    @cached_property
190    def scheduled_jobs(self) -> AdminScheduledJobsResource:
191        return AdminScheduledJobsResource(self._http)
192
193    @cached_property
194    def eval_models(self) -> AdminEvalModelsResource:
195        return AdminEvalModelsResource(self._http)
196
197    @cached_property
198    def eval_vendors(self) -> AdminEvalVendorsResource:
199        return AdminEvalVendorsResource(self._http)
200
201    @cached_property
202    def eval_harnesses(self) -> AdminEvalHarnessesResource:
203        return AdminEvalHarnessesResource(self._http)
204
205    @cached_property
206    def eval_runs(self) -> AdminEvalRunsResource:
207        return AdminEvalRunsResource(self._http)
208
209    @cached_property
210    def eval_campaigns(self) -> AdminEvalCampaignsResource:
211        return AdminEvalCampaignsResource(self._http)
212
213    def close(self) -> None:
214        """Close the underlying transport.
215
216        Only call this on a standalone client built via :meth:`connect`;
217        when reached via ``PlatformClient.admin`` the transport is shared
218        with — and closed by — the parent player client.
219        """
220        self._http.close()
221
222    def __enter__(self) -> Self:
223        return self
224
225    def __exit__(
226        self,
227        exc_type: type[BaseException] | None,
228        exc: BaseException | None,
229        tb: TracebackType | None,
230    ) -> None:
231        self.close()

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