"""Utils for interacting with JWT Auth.""" import os import re from typing import Callable, Iterable, Mapping, Protocol from urllib.parse import urljoin from .base import JWTAuth from .constants import ( DEFAULT_LEEWAY, PROD_API_AUDIENCE, PROD_AUDIENCE, PROD_ENVIRONMENT, PROD_ISSUERS, QA_API_AUDIENCE, QA_AUDIENCE, QA_ENVIRONMENT, QA_ISSUERS, TOKEN_TYPE, ) from .exceptions import JWTAuthError def get_default_jwks_url(environment: str) -> str: """Get default JWKS url.""" issuers_env = os.environ.get("AUTH_ISSUERS") if issuers_env: issuers = issuers_env.replace(" ", "").split(",") else: issuers = PROD_ISSUERS if environment == "prod" else QA_ISSUERS return urljoin(issuers[0], ".well-known/jwks.json") def get_default_audience(environment: str) -> Iterable[str]: """Get default audience, given environment.""" audience_env = os.environ.get("THEORCHARD_API_AUDIENCE") m2m_api_audience_env = os.environ.get("M2M_API_AUDIENCE") if m2m_api_audience_env: m2m_api_audience = m2m_api_audience_env else: m2m_api_audience = ( PROD_AUDIENCE if environment == PROD_ENVIRONMENT else QA_AUDIENCE ) if audience_env: orchard_api_audience = audience_env else: orchard_api_audience = ( PROD_API_AUDIENCE if environment == PROD_ENVIRONMENT else QA_API_AUDIENCE ) return [m2m_api_audience, orchard_api_audience] def get_default_issuer(environment: str) -> Iterable[str] | None: """Get default issuer, given environment.""" if environment == QA_ENVIRONMENT: return QA_ISSUERS elif environment == PROD_ENVIRONMENT: return PROD_ISSUERS return None def get_default_leeway(environment: str) -> int: """Get default leeway, given environment.""" env_leeway = os.environ.get("AUTH_LEEWAY") if env_leeway and env_leeway.isnumeric(): leeway = int(env_leeway) else: leeway = DEFAULT_LEEWAY if environment == PROD_ENVIRONMENT and leeway != DEFAULT_LEEWAY: raise ValueError(f"leeway must be 0 in PROD. Found: {leeway}") return leeway def jwt_auth_enabled_for_env(environment: str) -> bool: """Determine whether JWTAuth is enabled, given environment.""" if os.environ.get("JWT_AUTH_ENABLED"): return True else: return environment in [QA_ENVIRONMENT, PROD_ENVIRONMENT] def jwt_auth_from_environment(environment: str) -> JWTAuth: """Get JWT Auth instance.""" return JWTAuth( jwks_url=get_default_jwks_url(environment), audience=get_default_audience(environment), issuer=get_default_issuer(environment), leeway=get_default_leeway(environment), ) class RequestContext(Protocol): """RequestContext keeps track of request-level data during a request.""" authorization: str | None def get_token_string_from_headers( headers: Mapping[str, str], request_context_func: Callable[[], RequestContext | None] | None = None, ) -> str: """Get access token string.""" authorization: str | None = None if request_context_func: request_context = request_context_func() if request_context: authorization = request_context.authorization if not authorization: authorization = headers.get("authorization") if not authorization: raise JWTAuthError( 'Missing "Authorization" in headers.', code="missing_authorization" ) token_parts = authorization.split(maxsplit=1) if len(token_parts) != 2: raise JWTAuthError( 'Invalid "Authorization" header.', code="invalid_authorization" ) token_type, token_string = token_parts if token_type.lower() != TOKEN_TYPE.lower(): raise JWTAuthError( 'Invalid "Authorization" token type.', code="invalid_authorization_token_type", ) return token_string def is_path_match_re(path: str, *, match: list[str] | None) -> bool: """Return True if path matches any of the regexes in match, False otherwise.""" if match is None: return False for p in match: if re.compile(p).match(path): return True return False