ctfy.sdk.resources.auth

  1"""``client.auth`` — password auth, OAuth identities, API tokens, discovery."""
  2
  3from __future__ import annotations
  4
  5from urllib.parse import quote
  6
  7from ctfy.sdk._helpers import _raise_for_status
  8from ctfy.sdk.base import BaseHttpClient
  9from ctfy.server.models import (
 10    AffiliationSelection,
 11    AuthTokenResponse,
 12    CreateFineGrainedTokenResponse,
 13    DeviceCodeResponse,
 14    DeviceTokenResponse,
 15    EmailVerificationSent,
 16    EmailVerificationStatus,
 17    LinkedIdentity,
 18    LinkStartResponse,
 19    ProvidersResponse,
 20    TokenInfo,
 21    TokenScopesResponse,
 22    TokenTemplatesResponse,
 23    UserEmailInfo,
 24)
 25
 26
 27class AuthResource:
 28    """Register / login, identity binding, token CRUD, sign-in discovery."""
 29
 30    def __init__(self, http: BaseHttpClient) -> None:
 31        self._http = http
 32
 33    def register(
 34        self,
 35        email: str,
 36        password: str,
 37        display_name: str = "",
 38    ) -> AuthTokenResponse:
 39        """Register via email + password. Returns a fresh user-kind token.
 40
 41        First-user bootstrap: when the backend has zero users, the new
 42        user is granted ``role="super_admin"``.
 43        """
 44        resp = self._http.request(
 45            "POST",
 46            "/auth/register",
 47            json={"email": email, "password": password, "display_name": display_name},
 48        )
 49        _raise_for_status(resp)
 50        return AuthTokenResponse.model_validate(resp.json())
 51
 52    def login(self, email: str, password: str) -> AuthTokenResponse:
 53        """Sign in with email + password. Returns a fresh user-kind token."""
 54        resp = self._http.request(
 55            "POST", "/auth/login", json={"email": email, "password": password}
 56        )
 57        _raise_for_status(resp)
 58        return AuthTokenResponse.model_validate(resp.json())
 59
 60    def device_code(self) -> DeviceCodeResponse:
 61        """Start a device-authorization flow (unauthenticated, RFC 8628).
 62
 63        Returns a short ``user_code`` + a ``verification_uri`` the user
 64        opens in a browser, plus the opaque ``device_code`` to poll with.
 65        """
 66        resp = self._http.request("POST", "/auth/device/code")
 67        _raise_for_status(resp)
 68        return DeviceCodeResponse.model_validate(resp.json())
 69
 70    def device_token(self, device_code: str) -> DeviceTokenResponse:
 71        """Poll for the token after the user approves (unauthenticated).
 72
 73        ``status`` is ``pending`` / ``approved`` / ``denied`` /
 74        ``expired``; ``token`` is set only when ``approved``.
 75        """
 76        resp = self._http.request("POST", "/auth/device/token", json={"device_code": device_code})
 77        _raise_for_status(resp)
 78        return DeviceTokenResponse.model_validate(resp.json())
 79
 80    def set_password(self, new_password: str, current_password: str = "") -> None:
 81        """Set or change the caller's password.
 82
 83        ``current_password`` is required iff the caller already has a
 84        password identity; absent on the first-time "attach password to
 85        OAuth-only account" flow.
 86        """
 87        resp = self._http.request(
 88            "POST",
 89            "/auth/password",
 90            json={"new_password": new_password, "current_password": current_password},
 91        )
 92        _raise_for_status(resp)
 93
 94    def delete_password(self) -> None:
 95        """Drop the caller's password identity. Requires ≥1 OAuth identity."""
 96        resp = self._http.request("DELETE", "/auth/password")
 97        _raise_for_status(resp)
 98
 99    def list_identities(self) -> list[LinkedIdentity]:
100        """Bound OAuth + password identities for the calling user."""
101        resp = self._http.request("GET", "/auth/identities")
102        _raise_for_status(resp)
103        return [LinkedIdentity.model_validate(i) for i in resp.json()["items"]]
104
105    def unbind_identity(self, identity_id: str) -> None:
106        """Unbind one OAuth identity. Server blocks unlinking the last one."""
107        resp = self._http.request("DELETE", f"/auth/identities/{identity_id}")
108        _raise_for_status(resp)
109
110    def start_oauth_link(self, provider: str) -> LinkStartResponse:
111        """Begin a "link a second OAuth identity to my account" flow.
112
113        Returns the platform's signed-state authorize URL — the actual
114        flow finishes in a browser (the user opens the URL and approves
115        linking the provider to their existing account).
116        """
117        resp = self._http.request("POST", f"/auth/identities/link/{provider}")
118        _raise_for_status(resp)
119        return LinkStartResponse.model_validate(resp.json())
120
121    def list_tokens(self, offset: int = 0, limit: int = 50) -> list[TokenInfo]:
122        """List caller's tokens (user + agent). Plaintext never returned."""
123        resp = self._http.request("GET", "/auth/tokens", params={"offset": offset, "limit": limit})
124        _raise_for_status(resp)
125        return [TokenInfo.model_validate(t) for t in resp.json()["items"]]
126
127    def token_templates(self) -> TokenTemplatesResponse:
128        """The named starting points ``mint_token(template=…)`` accepts.
129
130        Carries each template's competition access, its dense
131        permission map and its suggested lifetime — the mechanical
132        content, which is all a non-browser caller needs. Human copy
133        lives in the web client's locale bundles.
134        """
135        resp = self._http.request("GET", "/auth/tokens/templates")
136        _raise_for_status(resp)
137        return TokenTemplatesResponse.model_validate(resp.json())
138
139    def mint_token(
140        self,
141        label: str,
142        expires_in_days: int | None = None,
143        *,
144        template: str | None = None,
145        competition_access: str | None = None,
146        competition_ids: list[str] | None = None,
147        permissions: dict[str, str] | None = None,
148    ) -> CreateFineGrainedTokenResponse:
149        """Mint a fresh fine-grained token (``pf_*``). Plaintext ONCE.
150
151        With no scope args the server mints the broad legacy profile
152        (every competition, full participate access) so existing
153        callers keep working unchanged. To scope it GitHub-style pass
154        ``competition_access`` (``all`` / ``selected`` / ``none``),
155        ``competition_ids`` (when ``selected``) and ``permissions``
156        (category → ``none`` / ``read`` / ``write``).
157
158        ``template`` names a preset (see :meth:`token_templates`) and is
159        a *starting point*: anything else passed alongside it wins, so
160        taking ``agent_play`` and narrowing it to two competitions is
161        one call. An unknown name is refused rather than defaulted —
162        the fall-back would be the broad profile, i.e. a wider token
163        than was asked for.
164
165        ``expires_in_days`` defaults server-side when ``None``.
166        """
167        body: dict[str, object] = {"label": label}
168        if expires_in_days is not None:
169            body["expires_in_days"] = expires_in_days
170        if template is not None:
171            body["template"] = template
172        if competition_access is not None:
173            body["competition_access"] = competition_access
174        if competition_ids is not None:
175            body["competition_ids"] = competition_ids
176        if permissions is not None:
177            body["permissions"] = permissions
178        resp = self._http.request("POST", "/auth/tokens", json=body)
179        _raise_for_status(resp)
180        return CreateFineGrainedTokenResponse.model_validate(resp.json())
181
182    def update_token_scopes(
183        self,
184        token_id: str,
185        *,
186        label: str | None = None,
187        template: str | None = None,
188        competition_access: str | None = None,
189        competition_ids: list[str] | None = None,
190        permissions: dict[str, str] | None = None,
191    ) -> TokenInfo:
192        """Edit a live token's label, competition reach and permissions.
193
194        **Partial**: only what you pass moves. Anything left ``None`` is
195        not sent at all, so editing permissions cannot blank the
196        competition scope by omission.
197
198        ``competition_ids`` has two absences that differ: ``None``
199        leaves the list alone, ``[]`` clears it.
200
201        The secret, the expiry and the kind are unreachable from here —
202        :meth:`rotate_token` and :meth:`renew_token` are those, each
203        with its own refusals. A change that actually widens the token
204        mails its owner, in the same unswitchable ``security``
205        category as the mint.
206        """
207        body: dict[str, object] = {}
208        if label is not None:
209            body["label"] = label
210        if template is not None:
211            body["template"] = template
212        if competition_access is not None:
213            body["competition_access"] = competition_access
214        if competition_ids is not None:
215            body["competition_ids"] = competition_ids
216        if permissions is not None:
217            body["permissions"] = permissions
218        resp = self._http.request("PATCH", f"/auth/tokens/{token_id}", json=body)
219        _raise_for_status(resp)
220        return TokenInfo.model_validate(resp.json())
221
222    def renew_token(self, token_id: str, expires_in_days: int | None = None) -> TokenInfo:
223        """Push a token's expiry out. The secret does not change.
224
225        Nothing to redeploy — which is the whole difference from
226        :meth:`rotate_token`. The new lifetime is measured from now,
227        never added to what is left; ``0`` means never expires and
228        ``None`` takes the server default.
229
230        Refused on a token that has **already** lapsed: expiry is the
231        one thing that kills a leaked credential nobody noticed, so
232        reviving a dead secret is the rotation's job.
233        """
234        body: dict[str, object] = {}
235        if expires_in_days is not None:
236            body["expires_in_days"] = expires_in_days
237        resp = self._http.request("POST", f"/auth/tokens/{token_id}/renew", json=body)
238        _raise_for_status(resp)
239        return TokenInfo.model_validate(resp.json())
240
241    def rotate_token(
242        self, token_id: str, expires_in_days: int | None = None
243    ) -> CreateFineGrainedTokenResponse:
244        """Replace a token's secret, carrying its scope across.
245
246        The answer to a leak and to a token that has already expired.
247        The plaintext comes back exactly once, the predecessor is
248        revoked in the same operation, and the reply's ``id`` is a new
249        one — a rotation is two linked rows, so an audit can still
250        resolve the old secret's hash.
251        """
252        body: dict[str, object] = {}
253        if expires_in_days is not None:
254            body["expires_in_days"] = expires_in_days
255        resp = self._http.request("POST", f"/auth/tokens/{token_id}/rotate", json=body)
256        _raise_for_status(resp)
257        return CreateFineGrainedTokenResponse.model_validate(resp.json())
258
259    def revoke_token(self, token_id: str) -> None:
260        """Revoke one token by id."""
261        resp = self._http.request("DELETE", f"/auth/tokens/{token_id}")
262        _raise_for_status(resp)
263
264    def revoke_other_tokens(self) -> None:
265        """Revoke every token belonging to this user except the caller's own."""
266        resp = self._http.request("DELETE", "/auth/tokens/others")
267        _raise_for_status(resp)
268
269    def providers(self) -> ProvidersResponse:
270        """Which sign-in methods this deployment offers (OAuth providers
271        configured + whether password auth is enabled)."""
272        resp = self._http.request("GET", "/auth/providers")
273        _raise_for_status(resp)
274        return ProvidersResponse.model_validate(resp.json())
275
276    def token_scopes(self) -> TokenScopesResponse:
277        """The catalog of grantable fine-grained token scopes — drives
278        the token-create wizard."""
279        resp = self._http.request("GET", "/auth/tokens/scopes")
280        _raise_for_status(resp)
281        return TokenScopesResponse.model_validate(resp.json())
282
283    def request_email_code(self, email: str) -> dict[str, object]:
284        """Send a one-time sign-in code to ``email``.
285
286        Always reports success for a well-formed address whether or not
287        an account exists — the endpoint is deliberately not an account
288        enumeration oracle, and verification creates the account if
289        needed.
290        """
291        resp = self._http.request("POST", "/auth/email-code", json={"email": email})
292        _raise_for_status(resp)
293        return dict(resp.json())
294
295    def verify_email_code(self, email: str, code: str) -> AuthTokenResponse:
296        """Exchange a sign-in code for a session token.
297
298        Creates the account on first use, which also verifies the address
299        by construction — unlike password registration, which never did.
300        """
301        resp = self._http.request(
302            "POST", "/auth/email-code/verify", json={"email": email, "code": code}
303        )
304        _raise_for_status(resp)
305        return AuthTokenResponse.model_validate(resp.json())
306
307    def request_email_verification(self) -> EmailVerificationSent:
308        """Mail a verification code to the calling account's own address.
309
310        ⚠️ **Takes no address.** The code goes to whatever is already on
311        the account, which is what stops this being a way to have the
312        platform mail arbitrary inboxes and what stops a caller
313        attaching somebody else's address to their own account as
314        verified.
315
316        Wrapped rather than left browser-only because a deployment that
317        sets ``email_verification_required`` refuses *every* write from
318        an unproved account — including from a ``user``-kind token minted
319        by ``ctfy login``, which is exactly the credential this client
320        holds. Without these two methods a CLI user on such a deployment
321        would have no way forward that did not involve a browser.
322        """
323        resp = self._http.request("POST", "/auth/email-verification")
324        _raise_for_status(resp)
325        return EmailVerificationSent.model_validate(resp.json())
326
327    def list_emails(self) -> list[UserEmailInfo]:
328        """Every address on the calling account, oldest claim first.
329
330        ``users.email`` is the *primary* — the address the platform mails
331        and the one ``email_verification_required`` rules on — and these
332        rows are the set it is chosen from, each carrying its own proof.
333        """
334        resp = self._http.request("GET", "/me/emails")
335        _raise_for_status(resp)
336        return [UserEmailInfo.model_validate(e) for e in resp.json()["items"]]
337
338    def add_email(self, email: str) -> EmailVerificationSent:
339        """Mail a code to an address you want to attach to this account.
340
341        ⚠️ **Unlike :meth:`request_email_verification`, this one does take
342        an address, and the difference is where the code goes.** That one
343        mails whatever is already on the account, so a parameter would
344        make it a way to mail arbitrary inboxes on demand. Here the code
345        goes *to the address you name*, so the exchange proves nothing
346        unless you can read that mailbox — naming it is the act.
347
348        ⚠️ **Read ``handover`` on the answer before redeeming the code.**
349        An address another account already holds is not refused here; the
350        exchange *moves* it. ``"orphans"`` means that account has no
351        other way in and will be unreachable afterwards, which
352        :meth:`verify_added_email` refuses unless you say
353        ``acknowledge_orphan=True``.
354        """
355        resp = self._http.request("POST", "/me/emails", json={"email": email})
356        _raise_for_status(resp)
357        return EmailVerificationSent.model_validate(resp.json())
358
359    def verify_added_email(
360        self, email: str, code: str, *, acknowledge_orphan: bool = False
361    ) -> UserEmailInfo:
362        """Exchange the code for a proved address on this account.
363
364        Carries the address as well as the code because the code table is
365        keyed by ``(address, purpose)`` — asking which address this code
366        is for *is* the lookup, not a convenience.
367
368        ⚠️ **``acknowledge_orphan`` defaults to refusing**, and the empty
369        case has to: a caller that never heard of handovers must not be
370        able to leave another account unreachable by omission. It is
371        needed only when :meth:`add_email` answered ``"orphans"`` — every
372        other move is free, and the losing account is mailed either way.
373        """
374        resp = self._http.request(
375            "POST",
376            "/me/emails/verify",
377            json={"email": email, "code": code, "acknowledge_orphan": acknowledge_orphan},
378        )
379        _raise_for_status(resp)
380        return UserEmailInfo.model_validate(resp.json())
381
382    def set_primary_email(self, email: str) -> UserEmailInfo:
383        """Point the account's primary address at one of its proved ones.
384
385        The verification stamp moves with the address; only a proved
386        address may be chosen, because the primary is where the platform
387        writes and what the verification gate rules on.
388        """
389        resp = self._http.request("POST", f"/me/emails/{quote(email, safe='')}/primary")
390        _raise_for_status(resp)
391        return UserEmailInfo.model_validate(resp.json())
392
393    def remove_email(self, email: str) -> None:
394        """Drop a secondary address. The primary is refused — make another
395        one primary first."""
396        resp = self._http.request("DELETE", f"/me/emails/{quote(email, safe='')}")
397        _raise_for_status(resp)
398
399    def publish_affiliation(self, email: str) -> AffiliationSelection:
400        """Show an institution badge derived from one proved address.
401
402        Independent of :meth:`set_primary_email` — two choices over one
403        set of addresses, the mailbox the platform writes to and the one
404        a stranger's badge is derived from. Only a *proved* address is
405        accepted (409 otherwise): the badge is a published claim about a
406        person, so the platform shows only what it checked.
407
408        Nothing about the institution is stored. It is resolved on every
409        read, so the badge disappears by itself when the address is
410        removed, loses its proof, is handed to another account, or the
411        operator deletes the catalogue row.
412        """
413        resp = self._http.request("POST", f"/me/emails/{quote(email, safe='')}/affiliation")
414        _raise_for_status(resp)
415        return AffiliationSelection.model_validate(resp.json())
416
417    def clear_affiliation(self) -> AffiliationSelection:
418        """Stop showing a badge.
419
420        Takes no address, and is therefore never refused: a version
421        addressed by address would refuse to stop publishing a badge
422        derived from an address that has already been removed.
423        """
424        resp = self._http.request("DELETE", "/me/affiliation")
425        _raise_for_status(resp)
426        return AffiliationSelection.model_validate(resp.json())
427
428    def submit_email_verification(self, code: str) -> EmailVerificationStatus:
429        """Exchange the code for the verified stamp.
430
431        Every rejection is the same ``invalid_code``, deliberately:
432        separating "wrong" from "expired" from "already used" tells an
433        attacker which guesses were structurally close and tells a
434        legitimate caller nothing beyond "request a new one".
435        """
436        resp = self._http.request("POST", "/auth/email-verification/verify", json={"code": code})
437        _raise_for_status(resp)
438        return EmailVerificationStatus.model_validate(resp.json())