import time import typing from http import HTTPStatus import requests from flask import current_app, request from jose import jwt from auth_api.errors import AuthError, Error from auth_api.logs import logger from auth_api.providers.base import BaseAuthProvider from auth_api.users import User class AtlasAuthProvider(BaseAuthProvider): def __init__(self): self._key = "" self._atlas_api_token = "" def authorize(self) -> User: """ We may have 2 kinds of tokens: - generic - without 'aud' claim, suitable for usage with all projects in a scope of end user authorization - m2m - with 'aud' claim, suitable for projects specified in 'aud' in a scope of server to server authorization """ token = self.get_token() rsa_key = self.get_key() if not rsa_key: logger.error("Unable to get Atlas key") raise AuthError(Error("no_atlas_key", "Unable to get Atlas key")) algorithms = current_app.config.get("ALGORITHMS") issuer = current_app.config.get("ATLAS_TOKEN_ISS") audience = current_app.config.get("ATLAS_API_AUDIENCE") options = self._get_validation_options(token) try: self._validate_required_claims(token) payload = jwt.decode( token, rsa_key, algorithms=algorithms, issuer=issuer, audience=audience, options=options, ) except jwt.ExpiredSignatureError: raise AuthError(Error("token_expired", "Token is expired")) except jwt.JWTClaimsError as e: logger.bind(algorithms=algorithms, issuer=issuer, audience=audience, options=options, error=e).warning( "Incorrect claims in token" ) raise AuthError(Error("invalid_claims", "Incorrect claims, please check the issuer and audience")) except AuthError as e: raise e except Exception as e: logger.bind(algorithms=algorithms, issuer=issuer, audience=audience, options=options, error=e).warning( "Unable to parse authentication token" ) raise AuthError(Error("invalid_header", "Unable to parse authentication token.")) return User(payload, current_app.config.get("RESOURCE_GROUP")) def get_key(self): if self._key: return self._key self._key = requests.get(current_app.config.get("ATLAS_PUBLIC_KEY_URL")).text return self._key def get_token(self): """Obtains the Access token from cookie or auth header.""" # case with token in header, e.g. Authorization: Bearer auth = request.headers.get("Authorization") # case with token in cookie token = request.cookies.get(current_app.config.get("ATLAS_BEARER_TOKEN_COOKIE_NAME")) # auth header is in priority if not auth and token: return token if not auth: raise AuthError( Error( "authorization_header_missing", "Authorization header or cookie 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] def get_atlas_api_token(self): try: if jwt.get_unverified_claims(self._atlas_api_token).get("exp") > int(time.time()): return self._atlas_api_token except Exception: # nosec pass self._atlas_api_token = "" res = requests.post( current_app.config.get("ATLAS_OAUTH_TOKEN_URL"), data={ "grant_type": "client_credentials", "audience": current_app.config.get("ATLAS_OAUTH_TOKEN_AUDIENCE"), "client_id": current_app.config.get("ATLAS_API_CLIENT_ID"), "client_secret": current_app.config.get("ATLAS_API_CLIENT_SECRET"), }, ) try: self._atlas_api_token = res.json().get("access_token") except Exception as e: logger.bind(error=e).error("Can`t get atlas api token") return self._atlas_api_token def _get_validation_options(self, token): """ We may have 2 kinds of tokens: - generic - without 'aud' claim, suitable for usage with all projects in a scope of end user authorization - m2m - with 'aud' claim, suitable for projects specified in 'aud' in a scope of server to server authorization This function returns params to pass the validation for both cases. We need this, as pyjwt lib has hard validation of `aud` claim on token decoding. Optionally we cat also set the strict aud validations, where the only m2m tokens are needed. """ options = {"verify_aud": False, "verify_iss": False} try: claimset = jwt.get_unverified_claims(token) except jwt.JWTError as e: logger.bind(error=e).warning("Token decoding error") return options if "aud" in claimset or current_app.config.get("ATLAS_API_AUDIENCE_IS_STRICT"): options["verify_aud"] = bool(current_app.config.get("ATLAS_API_AUDIENCE")) or current_app.config.get( "ATLAS_API_AUDIENCE_IS_STRICT" ) if "iss" in claimset: options["verify_iss"] = bool(current_app.config.get("ATLAS_TOKEN_ISS")) return options def _validate_required_claims(self, token): """ Applies 2 validation rules: 1. Validation for required aud claim for strict validation cases with m2m tokens, as it is broken it the python-jose lib itself for some reason: https://github.com/mpdavis/python-jose/blob/96474ecfb6ad3ce16f41b0814ab5126d58725e2a/jose/jwt.py#L338 2. Validation for required resource group claims, that are relevant to product. """ claimset = jwt.get_unverified_claims(token) if "aud" not in claimset and current_app.config.get("ATLAS_API_AUDIENCE_IS_STRICT"): raise jwt.JWTClaimsError("No aud in token, but it is required by settings") resource_group = current_app.config.get("RESOURCE_GROUP") if ( "aud" not in claimset and resource_group and not [claim for claim in claimset.keys() if claim.startswith(resource_group)] ): raise AuthError( Error("invalid permissions", "No product relevant claims found in token"), status_code=HTTPStatus.FORBIDDEN, )