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 _extract_items, _raise_for_status
  8from ctfy.sdk.base import BaseHttpClient
  9from ctfy.server.models import (
 10    AdminChallengeLatencyBucket,
 11    AdminChallengeStatsRow,
 12    AdminChallengeTimeseries,
 13    AdminFeedbackRow,
 14    AdminTaskInfo,
 15    ChallengeBuildStateResponse,
 16    ChallengePullStateResponse,
 17    QuestionAttemptResetInfo,
 18)
 19
 20
 21class AdminChallengesResource:
 22    """Operator challenge tooling: reload the catalog, pre-build on nodes,
 23    inspect attempt/solve stats, audit feedback, reset attempt caps."""
 24
 25    def __init__(self, http: BaseHttpClient) -> None:
 26        self._http = http
 27
 28    def rescan(self) -> AdminTaskInfo:
 29        """Submit a catalog-rescan background task (reload the platform's
 30        spec cache + fan the rescan out to every online node). Returns the
 31        task record; poll ``admin.tasks.get(id)`` (or watch the Tasks page)
 32        for the per-node outcome in ``result``. Wraps
 33        ``POST /admin/challenges/rescan``."""
 34        resp = self._http.request("POST", "/admin/challenges/rescan")
 35        _raise_for_status(resp)
 36        return AdminTaskInfo.model_validate(resp.json())
 37
 38    def build(self, challenge_id: str) -> AdminTaskInfo:
 39        """Submit a single-challenge pre-build background task.
 40
 41        Fans the build out to every online node and polls it to
 42        completion; poll :meth:`build_state` (or the Tasks surface) to
 43        watch ``building`` → ``built`` / ``failed``. Returns the task
 44        record. Wraps ``POST /admin/challenges/{id}/build``."""
 45        resp = self._http.request("POST", f"/admin/challenges/{challenge_id}/build")
 46        _raise_for_status(resp)
 47        return AdminTaskInfo.model_validate(resp.json())
 48
 49    def build_all(self, *, competition_id: str = "") -> AdminTaskInfo:
 50        """Submit a bulk pre-build background task. With ``competition_id``
 51        the build is scoped to that competition's effective challenge set;
 52        otherwise the whole catalog. The task kicks each online node off
 53        and polls them to completion; the per-node outcome + final
 54        per-challenge status land in ``result``. Returns the task record.
 55        Wraps ``POST /admin/challenges/build-all``."""
 56        params: dict[str, Any] = {}
 57        if competition_id:
 58            params["competition_id"] = competition_id
 59        resp = self._http.request("POST", "/admin/challenges/build-all", params=params)
 60        _raise_for_status(resp)
 61        return AdminTaskInfo.model_validate(resp.json())
 62
 63    def build_state(self) -> ChallengeBuildStateResponse:
 64        """Aggregate per-challenge build state across every online node.
 65
 66        Returns one row per known challenge with an ``aggregated``
 67        worst-case status plus the per-node breakdown for the modal
 68        drill-down. Wraps ``GET /admin/challenges/build-state``."""
 69        resp = self._http.request("GET", "/admin/challenges/build-state")
 70        _raise_for_status(resp)
 71        return ChallengeBuildStateResponse.model_validate(resp.json())
 72
 73    def pull(self, challenge_id: str) -> AdminTaskInfo:
 74        """Submit a single-challenge pre-pull background task (pull-side
 75        twin of :meth:`build`: warms registry-only ``image:`` services).
 76        Returns the task record. Wraps ``POST /admin/challenges/{id}/pull``."""
 77        resp = self._http.request("POST", f"/admin/challenges/{challenge_id}/pull")
 78        _raise_for_status(resp)
 79        return AdminTaskInfo.model_validate(resp.json())
 80
 81    def pull_all(self, *, competition_id: str = "") -> AdminTaskInfo:
 82        """Submit a bulk pre-pull background task (pull-side twin of
 83        :meth:`build_all`). Returns the task record. Wraps
 84        ``POST /admin/challenges/pull-all``."""
 85        params: dict[str, Any] = {}
 86        if competition_id:
 87            params["competition_id"] = competition_id
 88        resp = self._http.request("POST", "/admin/challenges/pull-all", params=params)
 89        _raise_for_status(resp)
 90        return AdminTaskInfo.model_validate(resp.json())
 91
 92    def pull_state(self) -> ChallengePullStateResponse:
 93        """Aggregate per-challenge pull state across every online node.
 94
 95        Pull-side twin of :meth:`build_state`. Wraps
 96        ``GET /admin/challenges/pull-state``."""
 97        resp = self._http.request("GET", "/admin/challenges/pull-state")
 98        _raise_for_status(resp)
 99        return ChallengePullStateResponse.model_validate(resp.json())
100
101    def stats(self, offset: int = 0, limit: int = 50) -> list[AdminChallengeStatsRow]:
102        """Per-challenge attempt / solve / first-blood counts."""
103        resp = self._http.request(
104            "GET",
105            "/admin/challenges/stats",
106            params={"offset": offset, "limit": limit},
107        )
108        _raise_for_status(resp)
109        return [AdminChallengeStatsRow.model_validate(r) for r in resp.json()["items"]]
110
111    def timeseries(
112        self,
113        challenge_id: str,
114        *,
115        window: int = 86400,
116        bucket: int = 3600,
117    ) -> AdminChallengeTimeseries:
118        """Per-challenge timeseries for the row-level drawer on
119        ``/admin/challenges``."""
120        resp = self._http.request(
121            "GET",
122            f"/admin/challenges/{challenge_id}/timeseries",
123            params={"window": window, "bucket": bucket},
124        )
125        _raise_for_status(resp)
126        return AdminChallengeTimeseries.model_validate(resp.json())
127
128    def latency_histogram(self) -> list[AdminChallengeLatencyBucket]:
129        """Instance-start latency distribution across challenges (the
130        admin challenges latency chart)."""
131        resp = self._http.request("GET", "/admin/challenges/latency-histogram")
132        _raise_for_status(resp)
133        return [AdminChallengeLatencyBucket.model_validate(b) for b in resp.json()]
134
135    def feedback(self, challenge_id: str, *, competition_id: str = "") -> list[AdminFeedbackRow]:
136        """Admin audit list: every reaction row on a challenge with
137        denormalised user identity. Requires admin role."""
138        params: dict[str, Any] = {}
139        if competition_id:
140            params["competition_id"] = competition_id
141        resp = self._http.request(
142            "GET", f"/admin/challenges/{challenge_id}/feedback", params=params
143        )
144        _raise_for_status(resp)
145        return _extract_items(resp.json(), AdminFeedbackRow)
146
147    def reset_question_attempts(
148        self,
149        team_id: str,
150        challenge_id: str,
151        question_id: str,
152        *,
153        reason: str = "",
154    ) -> QuestionAttemptResetInfo:
155        """Reset the per-question wrong-attempt counter for one team.
156
157        Writes a fresh baseline timestamp; the cap counter ignores all
158        wrong submissions before the new baseline, so the team gets a
159        fresh batch of attempts on this question without any audit
160        rows being deleted. Returns the new baseline row.
161        Wraps ``POST /admin/teams/{team_id}/questions/{challenge_id}/{question_id}/reset-attempts``."""
162        resp = self._http.request(
163            "POST",
164            f"/admin/teams/{team_id}/questions/{challenge_id}/{question_id}/reset-attempts",
165            json={"reason": reason},
166        )
167        _raise_for_status(resp)
168        return QuestionAttemptResetInfo.model_validate(resp.json())