ctfy.sdk.resources.auth

  1"""``client.auth`` — password auth, OAuth identities, API tokens, discovery."""
  2
  3from __future__ import annotations
  4
  5from ctfy.sdk._helpers import _raise_for_status
  6from ctfy.sdk.base import BaseHttpClient
  7from ctfy.server.models import (
  8    AuthTokenResponse,
  9    CreateFineGrainedTokenResponse,
 10    DeviceCodeResponse,
 11    DeviceTokenResponse,
 12    LinkedIdentity,
 13    LinkStartResponse,
 14    ProvidersResponse,
 15    TokenInfo,
 16    TokenScopesResponse,
 17)
 18
 19
 20class AuthResource:
 21    """Register / login, identity binding, token CRUD, sign-in discovery."""
 22
 23    def __init__(self, http: BaseHttpClient) -> None:
 24        self._http = http
 25
 26    def register(
 27        self,
 28        email: str,
 29        password: str,
 30        display_name: str = "",
 31    ) -> AuthTokenResponse:
 32        """Register via email + password. Returns a fresh user-kind token.
 33
 34        First-user bootstrap: when the backend has zero users, the new
 35        user is granted ``role="super_admin"``.
 36        """
 37        resp = self._http.request(
 38            "POST",
 39            "/auth/register",
 40            json={"email": email, "password": password, "display_name": display_name},
 41        )
 42        _raise_for_status(resp)
 43        return AuthTokenResponse.model_validate(resp.json())
 44
 45    def login(self, email: str, password: str) -> AuthTokenResponse:
 46        """Sign in with email + password. Returns a fresh user-kind token."""
 47        resp = self._http.request(
 48            "POST", "/auth/login", json={"email": email, "password": password}
 49        )
 50        _raise_for_status(resp)
 51        return AuthTokenResponse.model_validate(resp.json())
 52
 53    def device_code(self) -> DeviceCodeResponse:
 54        """Start a device-authorization flow (unauthenticated, RFC 8628).
 55
 56        Returns a short ``user_code`` + a ``verification_uri`` the user
 57        opens in a browser, plus the opaque ``device_code`` to poll with.
 58        """
 59        resp = self._http.request("POST", "/auth/device/code")
 60        _raise_for_status(resp)
 61        return DeviceCodeResponse.model_validate(resp.json())
 62
 63    def device_token(self, device_code: str) -> DeviceTokenResponse:
 64        """Poll for the token after the user approves (unauthenticated).
 65
 66        ``status`` is ``pending`` / ``approved`` / ``denied`` /
 67        ``expired``; ``token`` is set only when ``approved``.
 68        """
 69        resp = self._http.request(
 70            "POST", "/auth/device/token", json={"device_code": device_code}
 71        )
 72        _raise_for_status(resp)
 73        return DeviceTokenResponse.model_validate(resp.json())
 74
 75    def set_password(self, new_password: str, current_password: str = "") -> None:
 76        """Set or change the caller's password.
 77
 78        ``current_password`` is required iff the caller already has a
 79        password identity; absent on the first-time "attach password to
 80        OAuth-only account" flow.
 81        """
 82        resp = self._http.request(
 83            "POST",
 84            "/auth/password",
 85            json={"new_password": new_password, "current_password": current_password},
 86        )
 87        _raise_for_status(resp)
 88
 89    def delete_password(self) -> None:
 90        """Drop the caller's password identity. Requires ≥1 OAuth identity."""
 91        resp = self._http.request("DELETE", "/auth/password")
 92        _raise_for_status(resp)
 93
 94    def list_identities(self) -> list[LinkedIdentity]:
 95        """Bound OAuth + password identities for the calling user."""
 96        resp = self._http.request("GET", "/auth/identities")
 97        _raise_for_status(resp)
 98        return [LinkedIdentity.model_validate(i) for i in resp.json()["items"]]
 99
100    def unbind_identity(self, identity_id: str) -> None:
101        """Unbind one OAuth identity. Server blocks unlinking the last one."""
102        resp = self._http.request("DELETE", f"/auth/identities/{identity_id}")
103        _raise_for_status(resp)
104
105    def start_oauth_link(self, provider: str) -> LinkStartResponse:
106        """Begin a "link a second OAuth identity to my account" flow.
107
108        Returns the platform's signed-state authorize URL — the actual
109        flow finishes in a browser (the user opens the URL and approves
110        linking the provider to their existing account).
111        """
112        resp = self._http.request("POST", f"/auth/identities/link/{provider}")
113        _raise_for_status(resp)
114        return LinkStartResponse.model_validate(resp.json())
115
116    def list_tokens(self, offset: int = 0, limit: int = 50) -> list[TokenInfo]:
117        """List caller's tokens (user + agent). Plaintext never returned."""
118        resp = self._http.request("GET", "/auth/tokens", params={"offset": offset, "limit": limit})
119        _raise_for_status(resp)
120        return [TokenInfo.model_validate(t) for t in resp.json()["items"]]
121
122    def mint_token(
123        self,
124        label: str,
125        expires_in_days: int | None = None,
126        *,
127        competition_access: str | None = None,
128        competition_ids: list[str] | None = None,
129        permissions: dict[str, str] | None = None,
130    ) -> CreateFineGrainedTokenResponse:
131        """Mint a fresh fine-grained token (``pf_*``). Plaintext ONCE.
132
133        With no scope args the server mints the broad legacy profile
134        (every competition, full participate access) so existing
135        callers keep working unchanged. To scope it GitHub-style pass
136        ``competition_access`` (``all`` / ``selected`` / ``none``),
137        ``competition_ids`` (when ``selected``) and ``permissions``
138        (category → ``none`` / ``read`` / ``write``).
139
140        ``expires_in_days`` defaults server-side when ``None``.
141        """
142        body: dict[str, object] = {"label": label}
143        if expires_in_days is not None:
144            body["expires_in_days"] = expires_in_days
145        if competition_access is not None:
146            body["competition_access"] = competition_access
147        if competition_ids is not None:
148            body["competition_ids"] = competition_ids
149        if permissions is not None:
150            body["permissions"] = permissions
151        resp = self._http.request("POST", "/auth/tokens", json=body)
152        _raise_for_status(resp)
153        return CreateFineGrainedTokenResponse.model_validate(resp.json())
154
155    def revoke_token(self, token_id: str) -> None:
156        """Revoke one token by id."""
157        resp = self._http.request("DELETE", f"/auth/tokens/{token_id}")
158        _raise_for_status(resp)
159
160    def revoke_other_tokens(self) -> None:
161        """Revoke every token belonging to this user except the caller's own."""
162        resp = self._http.request("DELETE", "/auth/tokens/others")
163        _raise_for_status(resp)
164
165    def providers(self) -> ProvidersResponse:
166        """Which sign-in methods this deployment offers (OAuth providers
167        configured + whether password auth is enabled)."""
168        resp = self._http.request("GET", "/auth/providers")
169        _raise_for_status(resp)
170        return ProvidersResponse.model_validate(resp.json())
171
172    def token_scopes(self) -> TokenScopesResponse:
173        """The catalog of grantable fine-grained token scopes — drives
174        the token-create wizard."""
175        resp = self._http.request("GET", "/auth/tokens/scopes")
176        _raise_for_status(resp)
177        return TokenScopesResponse.model_validate(resp.json())