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