Source code for latch_cli.services.login

"""Service to login."""


from typing import Optional

import click
from latch_sdk_config.latch import config
from latch_sdk_config.user import user_config


[docs]def login(connection: Optional[str] = None) -> str: """Authenticates a user with Latch and persists an access token. Kicks off the three-legged OAuth2.0 flow outlined in `this RFC`_. The logic scaffolding this flow and detailed documentation about it can be found in the `latch.auth` package. The user will be redirected to a browser and prompted to login. This function meanwhile spins up a callback server on a separate thread that will be hit when the browser login is successful with an access token. .. _this RFC: https://datatracker.ietf.org/doc/html/rfc6749 """ if _browser_available() is False: token: str = click.prompt( f"Go to `{config.console_routes.developer}` and copy your API Key here", type=str, ) token = token.strip() user_config.update_token(token) return token from latch_cli.auth import PKCE, CSRFState, OAuth2 from latch_cli.constants import oauth2_constants with PKCE() as pkce: with CSRFState() as csrf_state: oauth2_flow = OAuth2(pkce, csrf_state, oauth2_constants) auth_code = oauth2_flow.authorization_request(connection) jwt = oauth2_flow.access_token_request(auth_code) # Exchange JWT from Auth0 for a persistent token issued by # LatchBio. access_jwt = _auth0_jwt_for_access_jwt(jwt) user_config.update_token(access_jwt) return access_jwt
def _browser_available() -> bool: """Returns true if browser available for login flow. Takes advantage of browser searching logic for many platforms written `here`_. .. _here: https://github.com/python/cpython/blob/3a2b89580ded72262fbea0f7ad24096a90c42b9c/Lib/webbrowser.py#L38 """ import webbrowser try: browser = webbrowser.get() if browser is not None: return True except Exception: pass return False def _auth0_jwt_for_access_jwt(token) -> str: """Requests a LatchBio minted (long-lived) acccess JWT. Uses an Auth0 token to authenticate the user. """ import latch_cli.tinyrequests as tinyrequests headers = { "Authorization": f"Bearer {token}", } url = config.api.user.jwt response = tinyrequests.post(url, headers=headers) resp = response.json() try: jwt = resp["jwt"] except KeyError as e: raise ValueError( f"Malformed response from request for access token {resp}" ) from e return jwt