fief_client.integrations.cli

  1"""CLI integration."""
  2
  3import functools
  4import http
  5import http.server
  6import json
  7import pathlib
  8import queue
  9import typing
 10import urllib.parse
 11import webbrowser
 12
 13from yaspin import yaspin
 14from yaspin.spinners import Spinners
 15
 16from fief_client import (
 17    Fief,
 18    FiefAccessTokenExpired,
 19    FiefAccessTokenInfo,
 20    FiefTokenResponse,
 21    FiefUserInfo,
 22)
 23from fief_client.pkce import get_code_challenge, get_code_verifier
 24
 25
 26class FiefAuthError(Exception):
 27    """
 28    Base error for FiefAuth integration.
 29    """
 30
 31
 32class FiefAuthNotAuthenticatedError(FiefAuthError):
 33    """
 34    The user is not authenticated.
 35    """
 36
 37    pass
 38
 39
 40class FiefAuthAuthorizationCodeMissingError(FiefAuthError):
 41    """
 42    The authorization code was not found in the redirection URL.
 43    """
 44
 45    pass
 46
 47
 48class FiefAuthRefreshTokenMissingError(FiefAuthError):
 49    """
 50    The refresh token is missing in the saved credentials.
 51    """
 52
 53    pass
 54
 55
 56class CallbackHTTPServer(http.server.ThreadingHTTPServer):
 57    pass
 58
 59
 60class CallbackHTTPRequestHandler(http.server.BaseHTTPRequestHandler):
 61    def __init__(
 62        self,
 63        *args,
 64        queue: "queue.Queue[str]",
 65        render_success_page,
 66        render_error_page,
 67        **kwargs,
 68    ) -> None:
 69        self.queue = queue
 70        self.render_success_page = render_success_page
 71        self.render_error_page = render_error_page
 72        super().__init__(*args, **kwargs)
 73
 74    def log_message(self, format: str, *args: typing.Any) -> None:
 75        pass
 76
 77    def do_GET(self):
 78        parsed_url = urllib.parse.urlparse(self.path)
 79        query_params = urllib.parse.parse_qs(parsed_url.query)
 80
 81        try:
 82            code = query_params["code"][0]
 83        except (KeyError, IndexError):
 84            output = self.render_error_page(query_params).encode("utf-8")
 85            self.send_response(http.HTTPStatus.BAD_REQUEST)
 86            self.send_header("Content-type", "text/html; charset=utf-8")
 87            self.send_header("Content-Length", str(len(output)))
 88            self.end_headers()
 89            self.wfile.write(output)
 90        else:
 91            self.queue.put(code)
 92
 93            output = self.render_success_page().encode("utf-8")
 94            self.send_response(http.HTTPStatus.OK)
 95            self.send_header("Content-type", "text/html; charset=utf-8")
 96            self.send_header("Content-Length", str(len(output)))
 97            self.end_headers()
 98            self.wfile.write(output)
 99
100        self.server.shutdown()
101
102
103class FiefAuth:
104    """
105    Helper class to integrate Fief authentication in a CLI tool.
106
107    **Example:**
108
109    ```py
110    from fief_client import Fief
111    from fief_client.integrations.cli import FiefAuth
112
113    fief = Fief(
114        "https://example.fief.dev",
115        "YOUR_CLIENT_ID",
116    )
117    auth = FiefAuth(fief, "./credentials.json")
118    ```
119    """
120
121    _userinfo: typing.Optional[FiefUserInfo] = None
122    _tokens: typing.Optional[FiefTokenResponse] = None
123
124    def __init__(self, client: Fief, credentials_path: str) -> None:
125        """
126        :param client: Instance of a Fief client.
127        :param credentials_path: Path where the credentials will be stored on the user machine.
128        We recommend you to use a library like [appdir](https://github.com/ActiveState/appdirs)
129        to determine a reasonable path depending on the user's operating system.
130        """
131        self.client = client
132        self.credentials_path = pathlib.Path(credentials_path)
133        self._load_stored_credentials()
134
135    def access_token_info(self, refresh: bool = True) -> FiefAccessTokenInfo:
136        """
137        Return credentials information saved on disk.
138
139        Optionally, it can automatically get a fresh `access_token` if
140        the saved one is expired.
141
142        :param refresh: Whether the client should automatically refresh the token.
143        Defaults to `True`.
144
145        :raises: `FiefAuthNotAuthenticatedError` if the user is not authenticated.
146        :raises: `fief_client.FiefAccessTokenExpired` if the access token is expired and automatic refresh is disabled.
147        """
148        if self._tokens is None:
149            raise FiefAuthNotAuthenticatedError()
150
151        access_token = self._tokens["access_token"]
152        try:
153            return self.client.validate_access_token(access_token)
154        except FiefAccessTokenExpired:
155            if refresh:
156                self._refresh_access_token()
157                return self.access_token_info()
158            raise
159
160    def current_user(self, refresh: bool = False) -> FiefUserInfo:
161        """
162        Return user information saved on disk.
163
164        Optionally, it can automatically refresh it from the server if there
165        is a valid access token.
166
167        :param refresh: Whether the client should refresh the user information.
168        Defaults to `False`.
169
170        :raises: `FiefAuthNotAuthenticatedError` if the user is not authenticated.
171        """
172        if self._tokens is None or self._userinfo is None:
173            raise FiefAuthNotAuthenticatedError()
174        if refresh:
175            access_token_info = self.access_token_info()
176            userinfo = self.client.userinfo(access_token_info["access_token"])
177            self._save_credentials(self._tokens, userinfo)
178        return self._userinfo
179
180    def authorize(
181        self,
182        server_address: tuple[str, int] = ("localhost", 51562),
183        redirect_path: str = "/callback",
184        *,
185        scope: typing.Optional[list[str]] = None,
186        lang: typing.Optional[str] = None,
187        extras_params: typing.Optional[typing.Mapping[str, str]] = None,
188    ) -> tuple[FiefTokenResponse, FiefUserInfo]:
189        """
190        Perform a user authentication with the Fief server.
191
192        It'll automatically open the user's default browser and redirect them
193        to the Fief authorization page.
194
195        Under the hood, the client opens a temporary web server.
196
197        After a successful authentication, Fief will redirect to this web server
198        so the client can catch the authorization code and generate a valid access token.
199
200        Finally, it'll automatically save the credentials on disk.
201
202        :param server_address: The address of the temporary web server the client should open.
203        It's a tuple composed of the IP and the port. Defaults to `("localhost", 51562)`.
204        :param redirect_path: Redirect URI where Fief will redirect after a successful authentication.
205        Defaults to `/callback`.
206        :param scope: Optional list of scopes to ask for.
207        The client will **always** ask at least for `openid` and `offline_access`.
208        :param lang: Optional parameter to set the user locale on the authentication pages.
209        Should be a valid [RFC 3066](https://www.rfc-editor.org/rfc/rfc3066) language identifier, like `fr` or `pt-PT`.
210        :param extras_params: Optional dictionary containing [specific parameters](https://docs.fief.dev/going-further/authorize-url/).
211
212        **Example:**
213
214        ```py
215        tokens, userinfo = auth.authorize()
216        ```
217        """
218        redirect_uri = f"http://{server_address[0]}:{server_address[1]}{redirect_path}"
219
220        scope_set: set[str] = set(scope) if scope else set()
221        scope_set.add("openid")
222        scope_set.add("offline_access")
223
224        code_verifier = get_code_verifier()
225        code_challenge = get_code_challenge(code_verifier)
226
227        authorization_url = self.client.auth_url(
228            redirect_uri,
229            scope=list(scope_set),
230            code_challenge=code_challenge,
231            code_challenge_method="S256",
232            lang=lang,
233            extras_params=extras_params,
234        )
235        webbrowser.open(authorization_url)
236
237        with yaspin(
238            text="Please complete authentication in your browser.",
239            spinner=Spinners.dots,
240        ) as spinner:
241            code_queue: queue.Queue[str] = queue.Queue()
242            server = CallbackHTTPServer(
243                server_address,
244                functools.partial(
245                    CallbackHTTPRequestHandler,
246                    queue=code_queue,
247                    render_success_page=self.render_success_page,
248                    render_error_page=self.render_error_page,
249                ),
250            )
251
252            server.serve_forever()
253
254            try:
255                code = code_queue.get(block=False)
256            except queue.Empty as e:
257                raise FiefAuthAuthorizationCodeMissingError() from e
258
259            spinner.text = "Getting a token..."
260
261            tokens, userinfo = self.client.auth_callback(
262                code, redirect_uri, code_verifier=code_verifier
263            )
264            self._save_credentials(tokens, userinfo)
265
266            spinner.ok("Successfully authenticated")
267
268        return tokens, userinfo
269
270    def render_success_page(self) -> str:
271        """
272        Generate the HTML page that'll be shown to the user after a successful redirection.
273
274        By default, it just tells the user that it can go back to the CLI.
275
276        You can override this method if you want to customize this page.
277        """
278        return f"""
279        <html>
280            <head>
281                <link href="{self.client.base_url}/static/auth.css" rel="stylesheet">
282            </head>
283            <body class="antialiased">
284                <main>
285                    <div class="relative flex">
286                        <div class="w-full">
287                            <div class="min-h-screen h-full flex flex flex-col after:flex-1">
288                                <div class="flex-1"></div>
289                                <div class="w-full max-w-sm mx-auto px-4 py-8 text-center">
290                                    <h1 class="text-3xl text-accent font-bold mb-6">Done! You can go back to your terminal!</h1>
291                                </div>
292                            </div>
293                        </div>
294                    </div>
295                </main>
296                <script>
297                    window.addEventListener("DOMContentLoaded", () => {{
298                        setTimeout(() => {{
299                            window.close();
300                        }}, 5000);
301                    }});
302                </script>
303            </body>
304        </html>
305        """
306
307    def render_error_page(self, query_params: dict[str, typing.Any]) -> str:
308        """
309        Generate the HTML page that'll be shown to the user when something goes wrong during redirection.
310
311        You can override this method if you want to customize this page.
312        """
313        return f"""
314        <html>
315            <head>
316                <link href="{self.client.base_url}/static/auth.css" rel="stylesheet">
317            </head>
318            <body class="antialiased">
319                <main>
320                    <div class="relative flex">
321                        <div class="w-full">
322                            <div class="min-h-screen h-full flex flex flex-col after:flex-1">
323                                <div class="flex-1"></div>
324                                <div class="w-full max-w-sm mx-auto px-4 py-8 text-center">
325                                    <h1 class="text-3xl text-accent font-bold mb-6">Something went wrong! You're not authenticated.</h1>
326                                    <p>Error detail: {json.dumps(query_params)}</p>
327                                </div>
328                            </div>
329                        </div>
330                    </div>
331                </main>
332            </body>
333        </html>
334        """
335
336    def _refresh_access_token(self):
337        refresh_token = self._tokens.get("refresh_token")
338        if refresh_token is None:
339            raise FiefAuthRefreshTokenMissingError()
340        tokens, userinfo = self.client.auth_refresh_token(refresh_token)
341        self._save_credentials(tokens, userinfo)
342
343    def _load_stored_credentials(self):
344        if self.credentials_path.exists():
345            with open(self.credentials_path) as file:
346                try:
347                    data = json.loads(file.read())
348                    self._userinfo = data["userinfo"]
349                    self._tokens = data["tokens"]
350                except json.decoder.JSONDecodeError:
351                    pass
352
353    def _save_credentials(self, tokens: FiefTokenResponse, userinfo: FiefUserInfo):
354        self._tokens = tokens
355        self._userinfo = userinfo
356        with open(self.credentials_path, "w") as file:
357            data = {"userinfo": userinfo, "tokens": tokens}
358            file.write(json.dumps(data))
359
360
361__all__ = [
362    "FiefAuth",
363    "FiefAuthError",
364    "FiefAuthNotAuthenticatedError",
365    "FiefAuthAuthorizationCodeMissingError",
366    "FiefAuthRefreshTokenMissingError",
367]