ctfy.sdk.resources.me

  1"""``client.me`` — the calling user's profile, inbox, account + progress."""
  2
  3from __future__ import annotations
  4
  5from typing import Any
  6
  7from ctfy.sdk._helpers import _raise_for_status
  8from ctfy.sdk.base import BaseHttpClient
  9from ctfy.server.models import (
 10    AccountDeletionAccepted,
 11    EmailVerificationSent,
 12    InboxResponse,
 13    MeResponse,
 14    MilestoneProgress,
 15    MyAchievementsResponse,
 16    MyFavourites,
 17    MyResults,
 18    MySolveSummary,
 19    NotificationPreferencesResponse,
 20    ReauthenticationOpened,
 21    StarGazerVerifyResponse,
 22)
 23
 24
 25class MeResource:
 26    """Self-service surface for the authenticated caller."""
 27
 28    def __init__(self, http: BaseHttpClient) -> None:
 29        self._http = http
 30
 31    def get(self) -> MeResponse:
 32        """Get the calling user's profile + auth context.
 33
 34        Per-competition team memberships ride along on
 35        ``competition_teams``; there is no global "current team" any
 36        more — pick the row whose ``competition_id`` matches the
 37        scope you care about.
 38        """
 39        resp = self._http.request("GET", "/me")
 40        _raise_for_status(resp)
 41        return MeResponse.model_validate(resp.json())
 42
 43    def inbox(self) -> InboxResponse:
 44        """Pending team invites, join requests, direct invites, announcements."""
 45        resp = self._http.request("GET", "/me/inbox")
 46        _raise_for_status(resp)
 47        return InboxResponse.model_validate(resp.json())
 48
 49    def notification_preferences(self) -> NotificationPreferencesResponse:
 50        """Which categories reach the caller by email.
 51
 52        Each cell reports ``effective`` (what happens now), ``default``
 53        (what reverting would restore) and ``explicit`` (whether the user
 54        actually chose it, or is merely inheriting). Storage is sparse, so
 55        an untouched cell has no row — ``explicit`` is how you tell.
 56        """
 57        resp = self._http.request("GET", "/me/notification-preferences")
 58        _raise_for_status(resp)
 59        return NotificationPreferencesResponse.model_validate(resp.json())
 60
 61    def set_notification_preference(
 62        self, category: str, enabled: bool, *, channel: str = "email"
 63    ) -> NotificationPreferencesResponse:
 64        """Turn one category on or off for one channel.
 65
 66        ``security`` is refused with a 400 — sign-in codes and account
 67        alerts exist to protect the account and are not switchable.
 68        """
 69        resp = self._http.request(
 70            "PATCH",
 71            "/me/notification-preferences",
 72            json={"category": category, "channel": channel, "enabled": enabled},
 73        )
 74        _raise_for_status(resp)
 75        return NotificationPreferencesResponse.model_validate(resp.json())
 76
 77    def clear_notification_preference(
 78        self, category: str, *, channel: str = "email"
 79    ) -> NotificationPreferencesResponse:
 80        """Drop the explicit cell so the default applies again.
 81
 82        This is *revert*, not *disable*: for most categories the default
 83        is on, so clearing an explicit ``False`` turns mail back on.
 84        """
 85        resp = self._http.request("DELETE", f"/me/notification-preferences/{category}/{channel}")
 86        _raise_for_status(resp)
 87        return NotificationPreferencesResponse.model_validate(resp.json())
 88
 89    def read_announcement(self, announcement_id: str) -> None:
 90        """Mark one announcement as read for the calling user."""
 91        resp = self._http.request("POST", f"/me/announcements/{announcement_id}/read")
 92        _raise_for_status(resp)
 93
 94    def read_all_announcements(self) -> None:
 95        """Mark every relevant announcement as read."""
 96        resp = self._http.request("POST", "/me/announcements/read-all")
 97        _raise_for_status(resp)
 98
 99    def update_profile(self, **fields: Any) -> MeResponse:
100        """PATCH the caller's profile.
101
102        Pass any subset of ``bio`` / ``country`` / ``website_url`` /
103        ``timezone`` / ``social_links``. Omit a key to leave it unchanged;
104        pass ``None`` to clear it. Returns the updated :class:`MeResponse`.
105
106        Example::
107
108            client.me.update_profile(bio="hacking goblins", country="JP")
109            client.me.update_profile(website_url=None)  # clear it
110        """
111        resp = self._http.request("PATCH", "/me/profile", json=fields)
112        _raise_for_status(resp)
113        return MeResponse.model_validate(resp.json())
114
115    def update_visibility(self, overrides: dict[str, bool]) -> MeResponse:
116        """PATCH per-field privacy flags on the caller's profile.
117
118        Keys outside :data:`PROFILE_VISIBILITY_KEYS` are silently dropped
119        server-side. Returns the updated :class:`MeResponse`.
120        """
121        resp = self._http.request("PATCH", "/me/profile/visibility", json=overrides)
122        _raise_for_status(resp)
123        return MeResponse.model_validate(resp.json())
124
125    def export_account(self) -> bytes:
126        """Download a ZIP archive of every piece of data the platform
127        holds about the caller (user, identities, tokens, memberships,
128        submissions, solves, activity). Returns raw bytes — write to a
129        file with ``Path(...).write_bytes(client.me.export_account())``."""
130        resp = self._http.request("GET", "/me/export")
131        _raise_for_status(resp)
132        return resp.content
133
134    def reauthenticate(
135        self,
136        *,
137        current_password: str = "",
138        code: str = "",
139    ) -> ReauthenticationOpened:
140        """Prove a factor again, opening the sudo window on **this** token.
141
142        A handful of account changes — repointing the primary address,
143        setting or removing a password, unlinking an identity, minting a
144        token, widening one, adding an address, linking an identity,
145        approving a device login — refuse with ``409
146        reauthentication_required`` unless the credential making the
147        request proved a factor inside the last
148        ``reauth_window_s`` seconds. This is how it proves one.
149
150        Signing in already counts, so a freshly-minted session needs
151        nothing; this is for a bearer that has been held past the
152        window. Read ``deletion_proof`` off :meth:`get` to learn which
153        factor this account has — it is the same answer for both — and
154        pass that one. An account with neither passes with neither.
155
156        The window lands on the token in the ``Authorization`` header
157        and no other: proving on a laptop does not open it for a bearer
158        somebody else is holding.
159        """
160        body: dict[str, str] = {}
161        if current_password:
162            body["current_password"] = current_password
163        if code:
164            body["code"] = code
165        resp = self._http.request("POST", "/me/reauthenticate", json=body)
166        _raise_for_status(resp)
167        return ReauthenticationOpened.model_validate(resp.json())
168
169    def request_reauth_code(self) -> EmailVerificationSent:
170        """Mail this account's own address a code that opens the window.
171
172        Refused (409 ``reauth_proof_required``) unless a mailed code is
173        what this account re-proves with — one holding a password
174        re-enters that instead. ⚠️ A *different* code from the deletion
175        one: digits handed over under "confirm it is you" must not also
176        erase the account.
177        """
178        resp = self._http.request("POST", "/me/reauthenticate/code")
179        _raise_for_status(resp)
180        return EmailVerificationSent.model_validate(resp.json())
181
182    def request_deletion_code(self) -> EmailVerificationSent:
183        """Mail this account's own address a code that deletes it.
184
185        Refused (409 ``deletion_proof_required``) unless a mailed code is
186        what this account's proof actually is — read ``deletion_proof``
187        off :meth:`get` first. An account with a password re-enters that
188        instead, and one with neither needs no code at all.
189        """
190        resp = self._http.request("POST", "/me/deletion-code")
191        _raise_for_status(resp)
192        return EmailVerificationSent.model_validate(resp.json())
193
194    def delete_account(
195        self,
196        confirm_display_name: str,
197        *,
198        current_password: str = "",
199        code: str = "",
200    ) -> AccountDeletionAccepted:
201        """Self-delete the calling user.
202
203        ``confirm_display_name`` must match the caller's display name
204        (falls back to email when empty) exactly — the typo gate against
205        fat-fingered destruction, and **not** the authorization half: it
206        is a public string.
207
208        The second factor is whichever ``deletion_proof`` on
209        :meth:`get` names — ``current_password`` for an account holding
210        a password, ``code`` from :meth:`request_deletion_code` for one
211        proving a mailed address, neither when it answers ``none``. The
212        server composes that answer and enforces it, so read it rather
213        than guessing; sending the wrong one is a 401.
214
215        ⚠️ Both are keyword-only and default blank, so a caller written
216        before the gate existed still compiles — and still works against
217        an account whose answer is ``none``, which is every account that
218        predates it.
219
220        ⚠️ **Read the answer: this does not always delete anything yet.**
221        With ``account_deletion_grace_s`` above zero — the default — the
222        account is *scheduled*: its tokens are revoked and a job runs the
223        cascade later, and signing in before then cancels the whole
224        thing. ``deleted`` says which happened and ``effective_at`` says
225        when it becomes final. A caller that discards this cannot tell
226        *gone* from *recoverable*, which are opposite facts with opposite
227        next steps.
228        """
229        body: dict[str, str] = {"confirm_display_name": confirm_display_name}
230        if current_password:
231            body["current_password"] = current_password
232        if code:
233            body["code"] = code
234        resp = self._http.request("DELETE", "/me", json=body)
235        _raise_for_status(resp)
236        return AccountDeletionAccepted.model_validate(resp.json())
237
238    def achievements(self, competition_id: str = "") -> MyAchievementsResponse:
239        """Caller's unlocked + locked badges (with progress hints)."""
240        params: dict[str, str] = {}
241        if competition_id:
242            params["competition_id"] = competition_id
243        resp = self._http.request("GET", "/me/achievements", params=params)
244        _raise_for_status(resp)
245        return MyAchievementsResponse.model_validate(resp.json())
246
247    def results(self) -> MyResults:
248        """我的战绩: every started event the caller entered, newest first, each
249        at that event's own official placement (never a total). Includes the
250        sittings only the caller may see -- in progress, frozen (ranked as of
251        the freeze), anonymous, disqualified -- each with ``only_you`` set."""
252        resp = self._http.request("GET", "/me/results")
253        _raise_for_status(resp)
254        return MyResults.model_validate(resp.json())
255
256    def favourites(self) -> MyFavourites:
257        """Every competition and challenge the caller has starred (#2070) —
258        whole sets, and only ever the caller's own: a favourite is a private
259        preference, never shown to another person or shared with a team."""
260        resp = self._http.request("GET", "/me/favourites")
261        _raise_for_status(resp)
262        return MyFavourites.model_validate(resp.json())
263
264    def favourite(self, kind: str, target_id: str) -> MyFavourites:
265        """Star a ``competition`` or a ``challenge``; returns the caller's full
266        set. Idempotent. A competition the caller cannot see is a 404."""
267        resp = self._http.request("PUT", f"/me/favourites/{kind}/{target_id}")
268        _raise_for_status(resp)
269        return MyFavourites.model_validate(resp.json())
270
271    def unfavourite(self, kind: str, target_id: str) -> MyFavourites:
272        """Remove a star; removing one that is not there is the same answer."""
273        resp = self._http.request("DELETE", f"/me/favourites/{kind}/{target_id}")
274        _raise_for_status(resp)
275        return MyFavourites.model_validate(resp.json())
276
277    def solves(self, *, competition_id: str = "") -> list[MySolveSummary]:
278        """Per-challenge solve summary for the caller.
279
280        Unscoped this spans every team the caller has ever played for.
281        Pass ``competition_id`` to ask the narrower question a
282        competition-scoped surface means: have I solved this *here*.
283        """
284        params: dict[str, str] = {}
285        if competition_id:
286            params["competition_id"] = competition_id
287        resp = self._http.request("GET", "/me/solves", params=params)
288        _raise_for_status(resp)
289        return [MySolveSummary.model_validate(s) for s in resp.json()]
290
291    def milestone_progress(self, *, competition_id: str = "") -> list[MilestoneProgress]:
292        """Partial-solve progress per challenge for the caller.
293
294        One row per challenge with at least one captured question;
295        ``solved_question_ids`` enumerates which milestones the user
296        has cleared. The row deliberately carries no question total:
297        the number of objectives is player-facing secret (#2435).
298        With ``competition_id`` set the result narrows to solves
299        stamped against the caller's team in that comp.
300        """
301        params: dict[str, str] = {}
302        if competition_id:
303            params["competition_id"] = competition_id
304        resp = self._http.request("GET", "/me/milestone-progress", params=params)
305        _raise_for_status(resp)
306        return [MilestoneProgress.model_validate(s) for s in resp.json()]
307
308    def verify_star_gazer(self, competition_id: str = "") -> StarGazerVerifyResponse:
309        """Verify the caller starred the ctfy GitHub repo and grant the
310        ``star_gazer`` badge if so.
311
312        Friendly failures (no GitHub identity, repo not configured, not
313        yet starred, GitHub rate-limited) come back as 200 with
314        ``verified=False`` + an actionable ``reason``."""
315        params = {"competition_id": competition_id} if competition_id else None
316        resp = self._http.request("POST", "/me/star-gazer/verify", params=params)
317        _raise_for_status(resp)
318        return StarGazerVerifyResponse.model_validate(resp.json())