ctfy.sdk.admin_resources.risk

  1"""``client.admin.risk`` -- the risk-control queue (#2068).
  2
  3Detection **reports**; a person decides. ``cases`` / ``case`` read the
  4organiser's queue; ``sightings`` is the one door to an account's sealed IP and
  5browser-fingerprint record and is **platform administrators only**.
  6
  7Acting on a case is the existing, separately authorised verbs:
  8``client.admin.teams`` (disqualify, ban, void solves) and
  9``client.admin.users.ban`` (platform account ban).
 10"""
 11
 12from __future__ import annotations
 13
 14from ctfy.sdk._helpers import _raise_for_status
 15from ctfy.sdk.base import BaseHttpClient
 16from ctfy.server.models import (
 17    RiskActionRow,
 18    RiskAppealList,
 19    RiskAppealView,
 20    RiskCaseDetail,
 21    RiskCaseList,
 22    RiskSettings,
 23    SightingsResponse,
 24)
 25
 26
 27class AdminRiskResource:
 28    def __init__(self, http: BaseHttpClient) -> None:
 29        self._http = http
 30
 31    def cases(
 32        self, competition_id: str, *, status: str = "", offset: int = 0, limit: int = 50
 33    ) -> RiskCaseList:
 34        """The event's risk cases, strongest first, with the count of every
 35        status. ``status`` is ``open`` / ``dismissed`` / ``actioned``."""
 36        params: dict[str, str | int] = {"offset": offset, "limit": limit}
 37        if status:
 38            params["status"] = status
 39        resp = self._http.request(
 40            "GET", f"/admin/competitions/{competition_id}/risk/cases", params=params
 41        )
 42        _raise_for_status(resp)
 43        return RiskCaseList.model_validate(resp.json())
 44
 45    def case(self, competition_id: str, case_id: str) -> RiskCaseDetail:
 46        """One case and every clue it rests on."""
 47        resp = self._http.request(
 48            "GET", f"/admin/competitions/{competition_id}/risk/cases/{case_id}"
 49        )
 50        _raise_for_status(resp)
 51        return RiskCaseDetail.model_validate(resp.json())
 52
 53    def sightings(self, user_id: str) -> SightingsResponse:
 54        """An account's IPs and browser fingerprints, opened. **Platform
 55        administrators only**; every read is an audit row. An IP is a lead,
 56        never evidence."""
 57        resp = self._http.request("GET", f"/admin/users/{user_id}/sightings")
 58        _raise_for_status(resp)
 59        return SightingsResponse.model_validate(resp.json())
 60
 61    def settings(self, competition_id: str) -> RiskSettings:
 62        """Whether the risk-control scan looks at this event (default on)."""
 63        resp = self._http.request("GET", f"/admin/competitions/{competition_id}/risk/settings")
 64        _raise_for_status(resp)
 65        return RiskSettings.model_validate(resp.json())
 66
 67    def set_enabled(self, competition_id: str, enabled: bool) -> RiskSettings:
 68        """Switch detection on or off for one event. Off stops new clues and
 69        cases; what was already found stays readable and actionable."""
 70        resp = self._http.request(
 71            "PUT",
 72            f"/admin/competitions/{competition_id}/risk/settings",
 73            json={"enabled": enabled},
 74        )
 75        _raise_for_status(resp)
 76        return RiskSettings.model_validate(resp.json())
 77
 78    def dismiss(self, competition_id: str, case_id: str, *, note: str = "") -> RiskCaseDetail:
 79        """Dismiss an open case: the organisers looked and there is nothing to
 80        do. A dismissed case re-opens by itself only on genuinely new evidence."""
 81        resp = self._http.request(
 82            "POST",
 83            f"/admin/competitions/{competition_id}/risk/cases/{case_id}/dismiss",
 84            json={"note": note},
 85        )
 86        _raise_for_status(resp)
 87        return RiskCaseDetail.model_validate(resp.json())
 88
 89    def reopen(self, competition_id: str, case_id: str) -> RiskCaseDetail:
 90        """Put a dismissed or actioned case back in the queue."""
 91        resp = self._http.request(
 92            "POST", f"/admin/competitions/{competition_id}/risk/cases/{case_id}/reopen"
 93        )
 94        _raise_for_status(resp)
 95        return RiskCaseDetail.model_validate(resp.json())
 96
 97    def actions(
 98        self, competition_id: str, *, team_id: str = "", case_id: str = ""
 99    ) -> list[RiskActionRow]:
100        """What the organisers did about teams here, newest first."""
101        params: dict[str, str] = {}
102        if team_id:
103            params["team_id"] = team_id
104        if case_id:
105            params["case_id"] = case_id
106        resp = self._http.request(
107            "GET", f"/admin/competitions/{competition_id}/risk/actions", params=params
108        )
109        _raise_for_status(resp)
110        return [RiskActionRow.model_validate(row) for row in resp.json()]
111
112    def appeals(self, competition_id: str, *, status: str = "pending") -> RiskAppealList:
113        """The appeals queue: ``pending`` (an answer is owed), ``replied`` or
114        ``all``. The words are the team's, in plain text."""
115        resp = self._http.request(
116            "GET",
117            f"/admin/competitions/{competition_id}/risk/appeals",
118            params={"status": status},
119        )
120        _raise_for_status(resp)
121        return RiskAppealList.model_validate(resp.json())
122
123    def reply(self, competition_id: str, appeal_id: str, text: str) -> RiskAppealView:
124        """Reply to an appeal; kept with the action, in plain text."""
125        resp = self._http.request(
126            "POST",
127            f"/admin/competitions/{competition_id}/risk/appeals/{appeal_id}/reply",
128            json={"text": text},
129        )
130        _raise_for_status(resp)
131        return RiskAppealView.model_validate(resp.json())