ctfy.sdk.admin_resources.challenges

  1"""``client.admin.challenges`` — catalog rescan, pre-build, stats, feedback audit."""
  2
  3from __future__ import annotations
  4
  5from typing import Any
  6
  7from ctfy.sdk._helpers import PagedList, _extract_items, _raise_for_status
  8from ctfy.sdk.base import BaseHttpClient
  9from ctfy.server.models import (
 10    AdminChallengeLatencyBucket,
 11    AdminChallengeQuestions,
 12    AdminChallengeStatsRow,
 13    AdminChallengeTimeseries,
 14    AdminFeedbackRow,
 15    AdminTaskInfo,
 16    ChallengeBuildStateResponse,
 17    ChallengePullStateResponse,
 18    DeceptionStats,
 19    LlmBudgetPage,
 20    LlmBudgetResetInfo,
 21    QuestionAttemptResetInfo,
 22)
 23
 24
 25class AdminChallengesResource:
 26    """Operator challenge tooling: reload the catalog, pre-build on nodes,
 27    inspect attempt/solve stats, audit feedback, reset attempt caps."""
 28
 29    def __init__(self, http: BaseHttpClient) -> None:
 30        self._http = http
 31
 32    def rescan(self) -> AdminTaskInfo:
 33        """Submit a catalog-rescan background task (reload the platform's
 34        spec cache + fan the rescan out to every online node). Returns the
 35        task record; poll ``admin.tasks.get(id)`` (or watch the Tasks page)
 36        for the per-node outcome in ``result``. Wraps
 37        ``POST /admin/challenges/rescan``."""
 38        resp = self._http.request("POST", "/admin/challenges/rescan")
 39        _raise_for_status(resp)
 40        return AdminTaskInfo.model_validate(resp.json())
 41
 42    def build(self, challenge_id: str) -> AdminTaskInfo:
 43        """Submit a single-challenge pre-build background task.
 44
 45        Fans the build out to every online node and polls it to
 46        completion; poll :meth:`build_state` (or the Tasks surface) to
 47        watch ``building`` → ``built`` / ``failed``. Returns the task
 48        record. Wraps ``POST /admin/challenges/{id}/build``."""
 49        resp = self._http.request("POST", f"/admin/challenges/{challenge_id}/build")
 50        _raise_for_status(resp)
 51        return AdminTaskInfo.model_validate(resp.json())
 52
 53    def build_all(self, *, competition_id: str = "") -> AdminTaskInfo:
 54        """Submit a bulk pre-build background task. With ``competition_id``
 55        the build is scoped to that competition's effective challenge set;
 56        otherwise the whole catalog. The task kicks each online node off
 57        and polls them to completion; the per-node outcome + final
 58        per-challenge status land in ``result``. Returns the task record.
 59        Wraps ``POST /admin/challenges/build-all``."""
 60        params: dict[str, Any] = {}
 61        if competition_id:
 62            params["competition_id"] = competition_id
 63        resp = self._http.request("POST", "/admin/challenges/build-all", params=params)
 64        _raise_for_status(resp)
 65        return AdminTaskInfo.model_validate(resp.json())
 66
 67    def questions(self, challenge_id: str) -> AdminChallengeQuestions:
 68        """Every milestone a challenge declares, gated ones included.
 69
 70        The catalog projection omits questions behind ``requires`` so a
 71        player cannot read a challenge's attack chain off its public page;
 72        this is the organiser's way back in — "which milestones exist, and
 73        why did this team not get credit for one". Wraps
 74        ``GET /admin/challenges/{id}/questions``."""
 75        resp = self._http.request("GET", f"/admin/challenges/{challenge_id}/questions")
 76        _raise_for_status(resp)
 77        return AdminChallengeQuestions.model_validate(resp.json())
 78
 79    def build_state(self) -> ChallengeBuildStateResponse:
 80        """Aggregate per-challenge build state across every online node.
 81
 82        Returns one row per known challenge with an ``aggregated``
 83        worst-case status plus the per-node breakdown for the modal
 84        drill-down. Wraps ``GET /admin/challenges/build-state``."""
 85        resp = self._http.request("GET", "/admin/challenges/build-state")
 86        _raise_for_status(resp)
 87        return ChallengeBuildStateResponse.model_validate(resp.json())
 88
 89    def pull(self, challenge_id: str) -> AdminTaskInfo:
 90        """Submit a single-challenge pre-pull background task (pull-side
 91        twin of :meth:`build`: warms registry-only ``image:`` services).
 92        Returns the task record. Wraps ``POST /admin/challenges/{id}/pull``."""
 93        resp = self._http.request("POST", f"/admin/challenges/{challenge_id}/pull")
 94        _raise_for_status(resp)
 95        return AdminTaskInfo.model_validate(resp.json())
 96
 97    def pull_all(self, *, competition_id: str = "") -> AdminTaskInfo:
 98        """Submit a bulk pre-pull background task (pull-side twin of
 99        :meth:`build_all`). Returns the task record. Wraps
100        ``POST /admin/challenges/pull-all``."""
101        params: dict[str, Any] = {}
102        if competition_id:
103            params["competition_id"] = competition_id
104        resp = self._http.request("POST", "/admin/challenges/pull-all", params=params)
105        _raise_for_status(resp)
106        return AdminTaskInfo.model_validate(resp.json())
107
108    def pull_state(self) -> ChallengePullStateResponse:
109        """Aggregate per-challenge pull state across every online node.
110
111        Pull-side twin of :meth:`build_state`. Wraps
112        ``GET /admin/challenges/pull-state``."""
113        resp = self._http.request("GET", "/admin/challenges/pull-state")
114        _raise_for_status(resp)
115        return ChallengePullStateResponse.model_validate(resp.json())
116
117    def stats(self, offset: int = 0, limit: int = 50) -> list[AdminChallengeStatsRow]:
118        """Per-challenge attempt / solve / first-blood counts."""
119        resp = self._http.request(
120            "GET",
121            "/admin/challenges/stats",
122            params={"offset": offset, "limit": limit},
123        )
124        _raise_for_status(resp)
125        return [AdminChallengeStatsRow.model_validate(r) for r in resp.json()["items"]]
126
127    def timeseries(
128        self,
129        challenge_id: str,
130        *,
131        window: int = 86400,
132        bucket: int = 3600,
133    ) -> AdminChallengeTimeseries:
134        """Per-challenge timeseries for the row-level drawer on
135        ``/admin/challenges``."""
136        resp = self._http.request(
137            "GET",
138            f"/admin/challenges/{challenge_id}/timeseries",
139            params={"window": window, "bucket": bucket},
140        )
141        _raise_for_status(resp)
142        return AdminChallengeTimeseries.model_validate(resp.json())
143
144    def latency_histogram(self) -> list[AdminChallengeLatencyBucket]:
145        """Instance-start latency distribution across challenges (the
146        admin challenges latency chart)."""
147        resp = self._http.request("GET", "/admin/challenges/latency-histogram")
148        _raise_for_status(resp)
149        return [AdminChallengeLatencyBucket.model_validate(b) for b in resp.json()]
150
151    def feedback(
152        self, challenge_id: str, *, competition_id: str = ""
153    ) -> PagedList[AdminFeedbackRow]:
154        """Admin audit list: every reaction row on a challenge with
155        denormalised user identity. Requires admin role."""
156        params: dict[str, Any] = {}
157        if competition_id:
158            params["competition_id"] = competition_id
159        resp = self._http.request(
160            "GET", f"/admin/challenges/{challenge_id}/feedback", params=params
161        )
162        _raise_for_status(resp)
163        return _extract_items(resp.json(), AdminFeedbackRow)
164
165    def reset_question_attempts(
166        self,
167        team_id: str,
168        challenge_id: str,
169        question_id: str,
170        *,
171        reason: str = "",
172    ) -> QuestionAttemptResetInfo:
173        """Reset the per-question wrong-attempt counter for one team.
174
175        Writes a fresh baseline timestamp; the cap counter ignores all
176        wrong submissions before the new baseline, so the team gets a
177        fresh batch of attempts on this question without any audit
178        rows being deleted. Returns the new baseline row.
179        Wraps ``POST /admin/teams/{team_id}/questions/{challenge_id}/{question_id}/reset-attempts``."""
180        resp = self._http.request(
181            "POST",
182            f"/admin/teams/{team_id}/questions/{challenge_id}/{question_id}/reset-attempts",
183            json={"reason": reason},
184        )
185        _raise_for_status(resp)
186        return QuestionAttemptResetInfo.model_validate(resp.json())
187
188    def list_llm_budgets(self, *, competition_id: str = "") -> LlmBudgetPage:
189        """Who has spent what against the challenge-LLM gateway.
190
191        The reader for :meth:`reset_llm_budget` — without it an organiser
192        can only hand back an allowance to a team somebody already named.
193        ⚠️ ``gateway_configured`` is on the envelope because an empty
194        list otherwise reads as "nobody spent anything" when it means
195        "this deployment enforces no budgets", and those call for
196        opposite actions.
197        Wraps ``GET /admin/llm-budgets``."""
198        params = {"competition_id": competition_id} if competition_id else None
199        resp = self._http.request("GET", "/admin/llm-budgets", params=params)
200        _raise_for_status(resp)
201        return LlmBudgetPage.model_validate(resp.json())
202
203    def deception_stats(
204        self,
205        *,
206        competition_id: str = "",
207        challenge_id: str = "",
208    ) -> DeceptionStats:
209        """How often an assessing agent obeyed a directive planted in a
210        challenge's own output.
211
212        ⚠️ ``resisted`` is an **upper bound**, and the response carries
213        that caveat on the wire rather than leaving it to documentation:
214        the probe runs inside a container the player is invited to
215        compromise, so a missing compliance record is not evidence of
216        restraint. ``comply_rate`` is ``None`` rather than ``0.0`` when
217        nothing has been contacted — 0/0 rendered as 0% would assert
218        that no agent was fooled.
219
220        Nothing here is scored; it reports, it does not rule.
221        Wraps ``GET /admin/deception/stats``."""
222        params = {
223            k: v
224            for k, v in (
225                ("competition_id", competition_id),
226                ("challenge_id", challenge_id),
227            )
228            if v
229        }
230        resp = self._http.request("GET", "/admin/deception/stats", params=params or None)
231        _raise_for_status(resp)
232        return DeceptionStats.model_validate(resp.json())
233
234    def reset_llm_budget(
235        self,
236        team_id: str,
237        competition_id: str,
238        challenge_id: str,
239        *,
240        reason: str = "",
241    ) -> LlmBudgetResetInfo:
242        """Give one team its LLM allowance back on one challenge.
243
244        Clears the gateway's live tally **and** the platform's durable
245        total — either alone un-does itself, so the route does both or
246        refuses. ⚠️ Always answers 200; read ``.outcome`` rather than the
247        status, because ``unreachable`` means nothing changed and is
248        worth retrying.
249        Wraps ``POST /admin/teams/{team_id}/llm-budget/{competition_id}/{challenge_id}/reset``."""
250        resp = self._http.request(
251            "POST",
252            f"/admin/teams/{team_id}/llm-budget/{competition_id}/{challenge_id}/reset",
253            json={"reason": reason},
254        )
255        _raise_for_status(resp)
256        return LlmBudgetResetInfo.model_validate(resp.json())