fief_client.integrations.cli

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