mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-10-25 00:12:08 +08:00
Refactor authenticate controller into a services layer.
Features better typing which did find some issues with Exception handling.
This commit is contained in:
John Chilton
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user