diff --git a/lib/galaxy/webapps/galaxy/api/authenticate.py b/lib/galaxy/webapps/galaxy/api/authenticate.py index b35060acbde..b01dd61020f 100644 --- a/lib/galaxy/webapps/galaxy/api/authenticate.py +++ b/lib/galaxy/webapps/galaxy/api/authenticate.py @@ -15,23 +15,21 @@ Returns } """ -import logging -from base64 import b64decode -from urllib.parse import unquote - -from galaxy import exceptions -from galaxy.util import ( - smart_str, - unicodify, -) from galaxy.web import expose_api_anonymous_and_sessionless from galaxy.webapps.base.webapp import GalaxyWebTransaction -from . import BaseGalaxyAPIController - -log = logging.getLogger(__name__) +from galaxy.webapps.galaxy.services.authenticate import ( + APIKeyResponse, + AuthenticationService, +) +from . import ( + BaseGalaxyAPIController, + depends, +) class AuthenticationController(BaseGalaxyAPIController): + authentication_service = depends(AuthenticationService) + @expose_api_anonymous_and_sessionless def options(self, trans: GalaxyWebTransaction, **kwd): """ @@ -47,7 +45,7 @@ class AuthenticationController(BaseGalaxyAPIController): # trans.response.headers['Access-Control-Allow-Methods'] = 'POST, PUT, GET, OPTIONS, DELETE' @expose_api_anonymous_and_sessionless - def get_api_key(self, trans: GalaxyWebTransaction, **kwd): + def get_api_key(self, trans: GalaxyWebTransaction, **kwd) -> APIKeyResponse: """ GET /api/authenticate/baseauth returns an API key for authenticated user based on BaseAuth headers @@ -57,62 +55,4 @@ class AuthenticationController(BaseGalaxyAPIController): :raises: ObjectNotFound, HTTPBadRequest """ - identity, password = self._decode_baseauth(trans.environ.get("HTTP_AUTHORIZATION")) - # check if this is an email address or username - user = self.app.user_manager.get_user_by_identity(identity) - if not user: - raise exceptions.ObjectNotFound("The user does not exist.") - is_valid_user = self.app.auth_manager.check_password(user, password) - if is_valid_user: - key = self.app.api_keys_manager.get_or_create_api_key(user) - return dict(api_key=key) - else: - raise exceptions.AuthenticationFailed("Invalid password.") - - def _decode_baseauth(self, encoded_str): - """ - Decode an encrypted HTTP basic authentication string. Returns a tuple of - the form (email, password), and raises a HTTPBadRequest exception if - nothing could be decoded. - - :param encoded_str: BaseAuth string encoded base64 - :type encoded_str: string - - :returns: email of the user - :rtype: string - :returns: password of the user - :rtype: string - - :raises: HTTPBadRequest - """ - try: - split = encoded_str.strip().split(" ") - except AttributeError: - raise exceptions.RequestParameterInvalidException("Authentication is missing") - - # If split is only one element, try to decode the email and password - # directly. - if len(split) == 1: - try: - email, password = unicodify(b64decode(smart_str(split[0]))).split(":") - except Exception as e: - raise exceptions.ActionInputError(e) - - # If there are only two elements, check the first and ensure it says - # 'basic' so that we know we're about to decode the right thing. If not, - # bail out. - elif len(split) == 2: - if split[0].strip().lower() == "basic": - try: - email, password = unicodify(b64decode(smart_str(split[1]))).split(":") - except Exception: - raise exceptions.ActionInputError() - else: - raise exceptions.ActionInputError() - - # If there are more than 2 elements, something crazy must be happening. - # Bail. - else: - raise exceptions.ActionInputError() - - return unquote(email), unquote(password) + return self.authentication_service.get_api_key(trans.environ) diff --git a/lib/galaxy/webapps/galaxy/services/authenticate.py b/lib/galaxy/webapps/galaxy/services/authenticate.py new file mode 100644 index 00000000000..2497da4c51b --- /dev/null +++ b/lib/galaxy/webapps/galaxy/services/authenticate.py @@ -0,0 +1,92 @@ +from base64 import b64decode +from typing import ( + Any, + Dict, + Optional, + Tuple, +) +from urllib.parse import unquote + +from pydantic import BaseModel + +from galaxy import exceptions +from galaxy.auth import AuthManager +from galaxy.managers.api_keys import ApiKeyManager +from galaxy.managers.users import UserManager +from galaxy.util import ( + smart_str, + unicodify, +) + + +class APIKeyResponse(BaseModel): + api_key: str + + +class AuthenticationService: + def __init__(self, user_manager: UserManager, auth_manager: AuthManager, api_keys_manager: ApiKeyManager): + self._user_manager = user_manager + self._auth_manager = auth_manager + self._api_keys_manager = api_keys_manager + + def get_api_key(self, environ: Dict[str, Any]) -> APIKeyResponse: + auth_header = environ.get("HTTP_AUTHORIZATION") + identity, password = self._decode_baseauth(auth_header) + # check if this is an email address or username + user = self._user_manager.get_user_by_identity(identity) + if not user: + raise exceptions.ObjectNotFound("The user does not exist.") + is_valid_user = self._auth_manager.check_password(user, password) + if is_valid_user: + key = self._api_keys_manager.get_or_create_api_key(user) + return APIKeyResponse(api_key=key) + else: + raise exceptions.AuthenticationFailed("Invalid password.") + + def _decode_baseauth(self, encoded_str: Optional[Any]) -> Tuple[str, str]: + """ + Decode an encrypted HTTP basic authentication string. Returns a tuple of + the form (email, password), and raises a HTTPBadRequest exception if + nothing could be decoded. + + :param encoded_str: BaseAuth string encoded base64 + :type encoded_str: string + + :returns: email of the user + :rtype: string + :returns: password of the user + :rtype: string + + :raises: HTTPBadRequest + """ + if encoded_str is None: + raise exceptions.RequestParameterInvalidException("Authentication is missing") + + split = str(encoded_str).strip().split(" ") + + # If split is only one element, try to decode the email and password + # directly. + if len(split) == 1: + try: + email, password = unicodify(b64decode(smart_str(split[0]))).split(":") + except Exception as e: + raise exceptions.ActionInputError(e) + + # If there are only two elements, check the first and ensure it says + # 'basic' so that we know we're about to decode the right thing. If not, + # bail out. + elif len(split) == 2: + if split[0].strip().lower() == "basic": + try: + email, password = unicodify(b64decode(smart_str(split[1]))).split(":") + except Exception: + raise exceptions.ActionInputError("Invalid authorization line") + else: + raise exceptions.ActionInputError("Invalid authorization line") + + # If there are more than 2 elements, something crazy must be happening. + # Bail. + else: + raise exceptions.ActionInputError("Invalid authorization line - more than two entries") + + return unquote(email), unquote(password)