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 email_verification_status(self) -> EmailVerificationStatus:
328        """Where the account stands, including whether a code is outstanding.
329
330        ``code_sent_at`` / ``code_expires_at`` are ``None`` when no
331        redeemable code exists (never sent, the send failed, expired).
332        """
333        resp = self._http.request("GET", "/auth/email-verification")
334        _raise_for_status(resp)
335        return EmailVerificationStatus.model_validate(resp.json())
336
337    def list_emails(self) -> list[UserEmailInfo]:
338        """Every address on the calling account, oldest claim first.
339
340        ``users.email`` is the *primary* — the address the platform mails
341        and the one ``email_verification_required`` rules on — and these
342        rows are the set it is chosen from, each carrying its own proof.
343        """
344        resp = self._http.request("GET", "/me/emails")
345        _raise_for_status(resp)
346        return [UserEmailInfo.model_validate(e) for e in resp.json()["items"]]
347
348    def add_email(self, email: str) -> EmailVerificationSent:
349        """Mail a code to an address you want to attach to this account.
350
351        ⚠️ **Unlike :meth:`request_email_verification`, this one does take
352        an address, and the difference is where the code goes.** That one
353        mails whatever is already on the account, so a parameter would
354        make it a way to mail arbitrary inboxes on demand. Here the code
355        goes *to the address you name*, so the exchange proves nothing
356        unless you can read that mailbox — naming it is the act.
357
358        ⚠️ **Read ``handover`` on the answer before redeeming the code.**
359        An address another account already holds is not refused here; the
360        exchange *moves* it. ``"orphans"`` means that account has no
361        other way in and will be unreachable afterwards, which
362        :meth:`verify_added_email` refuses unless you say
363        ``acknowledge_orphan=True``.
364        """
365        resp = self._http.request("POST", "/me/emails", json={"email": email})
366        _raise_for_status(resp)
367        return EmailVerificationSent.model_validate(resp.json())
368
369    def verify_added_email(
370        self, email: str, code: str, *, acknowledge_orphan: bool = False
371    ) -> UserEmailInfo:
372        """Exchange the code for a proved address on this account.
373
374        Carries the address as well as the code because the code table is
375        keyed by ``(address, purpose)`` — asking which address this code
376        is for *is* the lookup, not a convenience.
377
378        ⚠️ **``acknowledge_orphan`` defaults to refusing**, and the empty
379        case has to: a caller that never heard of handovers must not be
380        able to leave another account unreachable by omission. It is
381        needed only when :meth:`add_email` answered ``"orphans"`` — every
382        other move is free, and the losing account is mailed either way.
383        """
384        resp = self._http.request(
385            "POST",
386            "/me/emails/verify",
387            json={"email": email, "code": code, "acknowledge_orphan": acknowledge_orphan},
388        )
389        _raise_for_status(resp)
390        return UserEmailInfo.model_validate(resp.json())
391
392    def set_primary_email(self, email: str) -> UserEmailInfo:
393        """Point the account's primary address at one of its proved ones.
394
395        The verification stamp moves with the address; only a proved
396        address may be chosen, because the primary is where the platform
397        writes and what the verification gate rules on.
398        """
399        resp = self._http.request("POST", f"/me/emails/{quote(email, safe='')}/primary")
400        _raise_for_status(resp)
401        return UserEmailInfo.model_validate(resp.json())
402
403    def remove_email(self, email: str) -> None:
404        """Drop a secondary address. The primary is refused — make another
405        one primary first."""
406        resp = self._http.request("DELETE", f"/me/emails/{quote(email, safe='')}")
407        _raise_for_status(resp)
408
409    def publish_affiliation(self, email: str) -> AffiliationSelection:
410        """Show an institution badge derived from one proved address.
411
412        Independent of :meth:`set_primary_email` — two choices over one
413        set of addresses, the mailbox the platform writes to and the one
414        a stranger's badge is derived from. Only a *proved* address is
415        accepted (409 otherwise): the badge is a published claim about a
416        person, so the platform shows only what it checked.
417
418        Nothing about the institution is stored. It is resolved on every
419        read, so the badge disappears by itself when the address is
420        removed, loses its proof, is handed to another account, or the
421        operator deletes the catalogue row.
422        """
423        resp = self._http.request("POST", f"/me/emails/{quote(email, safe='')}/affiliation")
424        _raise_for_status(resp)
425        return AffiliationSelection.model_validate(resp.json())
426
427    def clear_affiliation(self) -> AffiliationSelection:
428        """Stop showing a badge.
429
430        Takes no address, and is therefore never refused: a version
431        addressed by address would refuse to stop publishing a badge
432        derived from an address that has already been removed.
433        """
434        resp = self._http.request("DELETE", "/me/affiliation")
435        _raise_for_status(resp)
436        return AffiliationSelection.model_validate(resp.json())
437
438    def submit_email_verification(self, code: str) -> EmailVerificationStatus:
439        """Exchange the code for the verified stamp.
440
441        Every rejection is the same ``invalid_code``, deliberately:
442        separating "wrong" from "expired" from "already used" tells an
443        attacker which guesses were structurally close and tells a
444        legitimate caller nothing beyond "request a new one".
445        """
446        resp = self._http.request("POST", "/auth/email-verification/verify", json={"code": code})
447        _raise_for_status(resp)
448        return EmailVerificationStatus.model_validate(resp.json())