ctfy.sdk

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

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