def get_fxa_uid_from_oauth_token()

in api/authentication.py [0:0]

• 34 lines of code
• 15 McCabe index (conditional complexity)

def get_fxa_uid_from_oauth_token(token: str, use_cache: bool = True) -> str:
    # set a default cache_timeout, but this will be overriden to match
    # the 'exp' time in the JWT returned by FxA
    cache_timeout = 60
    cache_key = get_cache_key(token)

    if not use_cache:
        fxa_resp_data = introspect_token(token)
    else:
        # set a default fxa_resp_data, so any error during introspection
        # will still cache for at least cache_timeout to prevent an outage
        # from causing useless run-away repetitive introspection requests
        fxa_resp_data = {"status_code": None, "json": {}}
        try:
            cached_fxa_resp_data = cache.get(cache_key)

            if cached_fxa_resp_data:
                fxa_resp_data = cached_fxa_resp_data
            else:
                # no cached data, get new
                fxa_resp_data = introspect_token(token)
        except AuthenticationFailed:
            raise
        finally:
            # Store potential valid response, errors, inactive users, etc. from FxA
            # for at least 60 seconds. Valid access_token cache extended after checking.
            cache.set(cache_key, fxa_resp_data, cache_timeout)

    if fxa_resp_data["status_code"] is None:
        raise APIException("Previous FXA call failed, wait to retry.")

    if not fxa_resp_data["status_code"] == 200:
        raise APIException("Did not receive a 200 response from FXA.")

    if not fxa_resp_data["json"].get("active"):
        raise AuthenticationFailed("FXA returned active: False for token.")

    # FxA user is active, check for the associated Relay account
    if (raw_fxa_uid := fxa_resp_data.get("json", {}).get("sub")) is None:
        raise NotFound("FXA did not return an FXA UID.")
    fxa_uid = str(raw_fxa_uid)

    # cache valid access_token and fxa_resp_data until access_token expiration
    # TODO: revisit this since the token can expire before its time
    if isinstance(fxa_resp_data.get("json", {}).get("exp"), int):
        # Note: FXA iat and exp are timestamps in *milliseconds*
        fxa_token_exp_time = int(fxa_resp_data["json"]["exp"] / 1000)
        now_time = int(datetime.now(UTC).timestamp())
        fxa_token_exp_cache_timeout = fxa_token_exp_time - now_time
        if fxa_token_exp_cache_timeout > cache_timeout:
            # cache until access_token expires (matched Relay user)
            # this handles cases where the token already expired
            cache_timeout = fxa_token_exp_cache_timeout
    cache.set(cache_key, fxa_resp_data, cache_timeout)

    return fxa_uid