import typing from flask import request from jose import jwt from auth_api import config from auth_api.errors import AuthError, Error from auth_api.providers.atlas import AtlasAuthProvider from auth_api.providers.auth0 import Auth0AuthProvider class AuthProviderFactory: """ Returns needed auth provider based on request context. Current implementation use `iss` claim from token to determine the provider. Actual providers instances are created on factory creation, and then it returns the actual instances to ensure that we have all needed data to validate the requests (e.g. rsa keys from external sources) """ def __init__(self): self._atlas = AtlasAuthProvider() self._auth0 = Auth0AuthProvider() def create(self) -> typing.Union[Auth0AuthProvider, AtlasAuthProvider]: if self.get_iss() == config.ATLAS_TOKEN_ISS: return self._atlas return self._auth0 def get_iss(self) -> str: token = self.get_token() try: payload = jwt.get_unverified_claims(token) except Exception: raise AuthError(Error("invalid_header", "Unable to parse authentication token.")) return payload.get("iss") def get_token(self) -> str: """Obtains the Access token from cookie or auth header.""" token = request.cookies.get(config.ATLAS_BEARER_TOKEN_COOKIE_NAME) if token: return token auth = request.headers.get("Authorization", None) if not auth: raise AuthError(Error("authorization_header_missing", "Authorization header is expected")) parts = auth.split() if parts[0].lower() != "bearer": raise AuthError(Error("invalid_header", "Authorization header must start with Bearer")) elif len(parts) == 1: raise AuthError(Error("invalid_header", "Token not found")) elif len(parts) > 2: raise AuthError(Error("invalid_header", "Authorization header must be Bearer token")) return parts[1]