ctfy.sdk.admin_resources.users

  1"""``client.admin.users`` — user roster + role mutation (admin)."""
  2
  3from __future__ import annotations
  4
  5import builtins
  6
  7from ctfy.sdk._helpers import _raise_for_status
  8from ctfy.sdk.base import BaseHttpClient
  9from ctfy.server.models import (
 10    AdminTokenRevokeResponse,
 11    AdminUserInfo,
 12    TokenTraceResponse,
 13)
 14
 15
 16class AdminUsersResource:
 17    """Admin-or-better user listing, lookup and promotion."""
 18
 19    def __init__(self, http: BaseHttpClient) -> None:
 20        self._http = http
 21
 22    def list(
 23        self,
 24        offset: int = 0,
 25        limit: int = 50,
 26        *,
 27        q: str = "",
 28        role: str = "",
 29        kind: str = "",
 30        verified: str = "",
 31        sort: str = "",
 32        dir: str = "",
 33    ) -> builtins.list[AdminUserInfo]:
 34        """Users known to the platform, with role + promotion audit.
 35
 36        ``q`` (matches display name / email / id), ``role``, ``kind``
 37        (``human`` / ``agent`` / ``service``), ``verified``
 38        (``verified`` / ``unverified``), ``sort`` (``name`` / ``email`` /
 39        ``created`` / ``last_active`` / ``solves`` / ``attempts``) and
 40        ``dir`` (``asc`` / ``desc``) are applied **server-side**, before
 41        the page is sliced.
 42
 43        ⚠️ ``kind`` and ``verified`` are separate axes from ``role``, not
 44        refinements of it: every synthetic agent entrant is role
 45        ``user``, and whether an address was proved says nothing about
 46        privilege. A caller narrowing with ``role`` alone cannot reach
 47        either question.
 48        """
 49        params: dict[str, str | int] = {"offset": offset, "limit": limit}
 50        params.update(
 51            {
 52                k: v
 53                for k, v in (
 54                    ("q", q),
 55                    ("role", role),
 56                    ("kind", kind),
 57                    ("verified", verified),
 58                    ("sort", sort),
 59                    ("dir", dir),
 60                )
 61                if v
 62            }
 63        )
 64        resp = self._http.request("GET", "/admin/users", params=params)
 65        _raise_for_status(resp)
 66        return [AdminUserInfo.model_validate(u) for u in resp.json()["items"]]
 67
 68    def resend_verification(self, user_id: str) -> None:
 69        """Mail this account's own address a fresh verification code.
 70
 71        ⚠️ **Writes nothing.** The account still has to receive the code
 72        and exchange it, so a bounce leaves it exactly as unproved as it
 73        was — which is why this is a separate call from
 74        :meth:`verify_email` rather than a flag on it.
 75        """
 76        resp = self._http.request("POST", f"/admin/users/{user_id}/resend-verification")
 77        _raise_for_status(resp)
 78
 79    def verify_email(self, user_id: str) -> AdminUserInfo:
 80        """Vouch for an address the deployment cannot reach at all.
 81
 82        ⚠️ **This writes a permanent claim in somebody else's name**, and
 83        there is no call that undoes it: the stamp fills and never
 84        overwrites. The row records which operator vouched, so a vouch
 85        stays legible as a vouch rather than passing for evidence.
 86        """
 87        resp = self._http.request("POST", f"/admin/users/{user_id}/verify-email")
 88        _raise_for_status(resp)
 89        return AdminUserInfo.model_validate(resp.json())
 90
 91    def get(self, user_id: str) -> AdminUserInfo:
 92        """Admin-or-better: single-user lookup with the audit fields
 93        (role, promoted_by, promoted_at, current team summary). Skips
 94        the list scan :meth:`list` would otherwise force."""
 95        resp = self._http.request("GET", f"/admin/users/{user_id}")
 96        _raise_for_status(resp)
 97        return AdminUserInfo.model_validate(resp.json())
 98
 99    def set_standings_excluded(self, user_id: str, excluded: bool) -> AdminUserInfo:
100        """Take an account off the public standings, or put it back.
101
102        Requires **super-admin**. For the deployment's own staff and test
103        accounts, which otherwise sit on the leaderboard beside the
104        people the event is for.
105
106        ⚠️ **Not a disqualification.** The account keeps playing, keeps
107        its solves and still sees its own rank; what changes is whether
108        anybody else does. The route's own docstring makes the same
109        distinction — conflating the two would put a scarlet letter on
110        an organiser's test account.
111
112        ⚠️ Idempotent: setting it to the value it already holds returns
113        the row unchanged rather than re-stamping it.
114        """
115        resp = self._http.request(
116            "PATCH",
117            f"/admin/users/{user_id}/standings",
118            json={"excluded": excluded},
119        )
120        _raise_for_status(resp)
121        return AdminUserInfo.model_validate(resp.json())
122
123    def trace_token(self, token: str) -> TokenTraceResponse:
124        """Attribute a leaked bearer token to its owner and its lineage.
125
126        Requires **super-admin**. The sibling of
127        :meth:`RecordsResource.trace_answer`, on the other kind of
128        secret that ends up in a paste bin: ``team_tokens`` stores only
129        a SHA-256 hash, so an operator holding a ``pf_…`` from a log has
130        nothing to grep for and no way to learn whose it is.
131
132        ``chain`` is the whole **rotation lineage**, oldest first —
133        ``rotate`` mints a replacement row linked by ``rotated_from``
134        rather than re-secreting the old one, so a secret found in a
135        six-month-old log resolves to a revoked row whose successor is
136        the credential still authenticating requests. ``live`` marks the
137        one worth revoking; a token that was never rotated yields a
138        chain of one.
139
140        ⚠️ A string this deployment never minted comes back
141        ``found=False`` at 200 rather than 404 — "not ours" is an answer
142        somebody asked for, and a 404 reads as a wrong URL.
143
144        ⚠️ Nothing derived from the secret comes back, not even its
145        hash: that digest is unsalted, so it *is* the stored verifier.
146
147        The token travels in the request body, never a query string, so
148        a live credential does not land in an access log on the way.
149        """
150        resp = self._http.request("POST", "/admin/token-trace", json={"token": token})
151        _raise_for_status(resp)
152        return TokenTraceResponse.model_validate(resp.json())
153
154    def revoke_token(self, token_id: str) -> AdminTokenRevokeResponse:
155        """End a token this super-admin does not own — the revoke half of
156        :meth:`trace_token`.
157
158        Requires **super-admin**. Owner-revoke (``DELETE
159        /auth/tokens/{id}``) refuses a token whose owner is not the
160        caller; this is the operator door for a leaked ``pf_…`` — pass
161        the ``token_id`` that ``trace_token`` marked *live*. Especially
162        for a synthetic **agent** entrant, whose owner is an ``agent:…``
163        account nobody can log in as, this is the only sanctioned path.
164
165        ⚠️ Idempotent: re-revoking an already-revoked row returns
166        ``already_revoked=True`` at 200, not an error. A ``token_id``
167        that resolves to nothing is a 404 (unlike ``trace_token``'s
168        200 miss — here the id came from a trace, so a miss is a typo).
169        """
170        resp = self._http.request("POST", f"/admin/tokens/{token_id}/revoke")
171        _raise_for_status(resp)
172        return AdminTokenRevokeResponse.model_validate(resp.json())
173
174    def set_role(self, user_id: str, role: str) -> AdminUserInfo:
175        """Promote / demote between ``"user"`` and ``"admin"``.
176
177        Requires **super-admin**. ``super_admin`` role is env-anchored and
178        cannot be changed via API; trying returns 403.
179        """
180        resp = self._http.request(
181            "PATCH",
182            f"/admin/users/{user_id}/role",
183            json={"role": role},
184        )
185        _raise_for_status(resp)
186        return AdminUserInfo.model_validate(resp.json())