Warning
This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.
Source code for galaxy.authnz.managers
import builtins
import copy
import json
import logging
import os
import random
import string
import requests
from cloudauthz import CloudAuthz
from cloudauthz.exceptions import CloudAuthzBaseException
from galaxy import (
exceptions,
model,
)
from galaxy.util import (
asbool,
etree,
listify,
parse_xml,
string_as_bool,
unicodify,
)
from .custos_authnz import (
CustosAuthnz,
KEYCLOAK_BACKENDS,
)
from .psa_authnz import (
BACKENDS_NAME,
on_the_fly_config,
PSAAuthnz,
Storage,
Strategy,
)
log = logging.getLogger(__name__)
# Note: This if for backward compatibility. Icons can be specified in oidc_backends_config.xml.
DEFAULT_OIDC_IDP_ICONS = {
"google": "https://developers.google.com/identity/images/btn_google_signin_light_normal_web.png",
"elixir": "https://elixir-europe.org/sites/default/files/images/login-button-orange.png",
"okta": "https://www.okta.com/sites/all/themes/Okta/images/blog/Logos/Okta_Logo_BrightBlue_Medium.png",
}
[docs]class AuthnzManager:
[docs] def __init__(self, app, oidc_config_file, oidc_backends_config_file):
"""
:type app: galaxy.app.UniverseApplication
:param app:
:type config: string
:param config: sets the path for OIDC configuration
file (e.g., oidc_backends_config.xml).
"""
self.app = app
self.allowed_idps = None
self._parse_oidc_config(oidc_config_file)
self._parse_oidc_backends_config(oidc_backends_config_file)
def _parse_oidc_config(self, config_file):
self.oidc_config = {}
try:
tree = parse_xml(config_file)
root = tree.getroot()
if root.tag != "OIDC":
raise etree.ParseError(
"The root element in OIDC_Config xml file is expected to be `OIDC`, "
"found `{}` instead -- unable to continue.".format(root.tag)
)
for child in root:
if child.tag != "Setter":
log.error(
"Expect a node with `Setter` tag, found a node with `{}` tag instead; "
"skipping this node.".format(child.tag)
)
continue
if "Property" not in child.attrib or "Value" not in child.attrib or "Type" not in child.attrib:
log.error(
"Could not find the node attributes `Property` and/or `Value` and/or `Type`;"
" found these attributes: `{}`; skipping this node.".format(child.attrib)
)
continue
try:
if child.get("Type") == "bool":
func = string_as_bool
else:
func = getattr(builtins, child.get("Type"))
except AttributeError:
log.error(
"The value of attribute `Type`, `{}`, is not a valid built-in type;" " skipping this node"
).format(child.get("Type"))
continue
self.oidc_config[child.get("Property")] = func(child.get("Value"))
except ImportError:
raise
except etree.ParseError as e:
raise etree.ParseError(f"Invalid configuration at `{config_file}`: {e} -- unable to continue.")
def _get_idp_icon(self, idp):
return self.oidc_backends_config[idp].get("icon") or DEFAULT_OIDC_IDP_ICONS.get(idp)
def _parse_oidc_backends_config(self, config_file):
self.oidc_backends_config = {}
self.oidc_backends_implementation = {}
try:
tree = parse_xml(config_file)
root = tree.getroot()
if root.tag != "OIDC":
raise etree.ParseError(
"The root element in OIDC config xml file is expected to be `OIDC`, "
"found `{}` instead -- unable to continue.".format(root.tag)
)
for child in root:
if child.tag != "provider":
log.error(
"Expect a node with `provider` tag, found a node with `{}` tag instead; "
"skipping the node.".format(child.tag)
)
continue
if "name" not in child.attrib:
log.error(f"Could not find a node attribute 'name'; skipping the node '{child.tag}'.")
continue
idp = child.get("name").lower()
if idp in BACKENDS_NAME:
self.oidc_backends_config[idp] = self._parse_idp_config(child)
self.oidc_backends_implementation[idp] = "psa"
self.app.config.oidc[idp] = {"icon": self._get_idp_icon(idp)}
elif idp in KEYCLOAK_BACKENDS:
self.oidc_backends_config[idp] = self._parse_custos_config(child)
self.oidc_backends_implementation[idp] = "custos"
self.app.config.oidc[idp] = {"icon": self._get_idp_icon(idp)}
else:
raise etree.ParseError("Unknown provider specified")
if len(self.oidc_backends_config) == 0:
raise etree.ParseError("No valid provider configuration parsed.")
except ImportError:
raise
except etree.ParseError as e:
raise etree.ParseError(f"Invalid configuration at `{config_file}`: {e} -- unable to continue.")
def _parse_idp_config(self, config_xml):
rtv = {
"client_id": config_xml.find("client_id").text,
"client_secret": config_xml.find("client_secret").text,
"redirect_uri": config_xml.find("redirect_uri").text,
"enable_idp_logout": asbool(config_xml.findtext("enable_idp_logout", "false")),
}
if config_xml.find("label") is not None:
rtv["label"] = config_xml.find("label").text
if config_xml.find("require_create_confirmation") is not None:
rtv["require_create_confirmation"] = asbool(config_xml.find("require_create_confirmation").text)
if config_xml.find("prompt") is not None:
rtv["prompt"] = config_xml.find("prompt").text
if config_xml.find("api_url") is not None:
rtv["api_url"] = config_xml.find("api_url").text
if config_xml.find("url") is not None:
rtv["url"] = config_xml.find("url").text
if config_xml.find("icon") is not None:
rtv["icon"] = config_xml.find("icon").text
if config_xml.find("extra_scopes") is not None:
rtv["extra_scopes"] = listify(config_xml.find("extra_scopes").text)
if config_xml.find("tenant_id") is not None:
rtv["tenant_id"] = config_xml.find("tenant_id").text
if config_xml.find("pkce_support") is not None:
rtv["pkce_support"] = asbool(config_xml.find("pkce_support").text)
return rtv
def _parse_custos_config(self, config_xml):
rtv = {
"url": config_xml.find("url").text,
"client_id": config_xml.find("client_id").text,
"client_secret": config_xml.find("client_secret").text,
"redirect_uri": config_xml.find("redirect_uri").text,
"enable_idp_logout": asbool(config_xml.findtext("enable_idp_logout", "false")),
}
if config_xml.find("label") is not None:
rtv["label"] = config_xml.find("label").text
if config_xml.find("require_create_confirmation") is not None:
rtv["require_create_confirmation"] = asbool(config_xml.find("require_create_confirmation").text)
if config_xml.find("credential_url") is not None:
rtv["credential_url"] = config_xml.find("credential_url").text
if config_xml.find("well_known_oidc_config_uri") is not None:
rtv["well_known_oidc_config_uri"] = config_xml.find("well_known_oidc_config_uri").text
if config_xml.findall("allowed_idp") is not None:
self.allowed_idps = list(map(lambda idp: idp.text, config_xml.findall("allowed_idp")))
if config_xml.find("ca_bundle") is not None:
rtv["ca_bundle"] = config_xml.find("ca_bundle").text
if config_xml.find("icon") is not None:
rtv["icon"] = config_xml.find("icon").text
if config_xml.find("pkce_support") is not None:
rtv["pkce_support"] = asbool(config_xml.find("pkce_support").text)
return rtv
[docs] def get_allowed_idps(self):
# None, if no allowed idp list is set, and a list of EntityIDs if configured (in oidc_backend)
return self.allowed_idps
def _unify_provider_name(self, provider):
if provider.lower() in self.oidc_backends_config:
return provider.lower()
for k, v in BACKENDS_NAME.items():
if v == provider:
return k.lower()
return None
def _get_authnz_backend(self, provider, idphint=None):
unified_provider_name = self._unify_provider_name(provider)
if unified_provider_name in self.oidc_backends_config:
provider = unified_provider_name
identity_provider_class = self._get_identity_provider_class(self.oidc_backends_implementation[provider])
try:
if provider in KEYCLOAK_BACKENDS:
return (
True,
"",
identity_provider_class(
unified_provider_name,
self.oidc_config,
self.oidc_backends_config[unified_provider_name],
idphint=idphint,
),
)
else:
return (
True,
"",
identity_provider_class(
unified_provider_name, self.oidc_config, self.oidc_backends_config[unified_provider_name]
),
)
except Exception as e:
log.exception(f"An error occurred when loading {identity_provider_class.__name__}")
return False, unicodify(e), None
else:
msg = f"The requested identity provider, `{provider}`, is not a recognized/expected provider."
log.debug(msg)
return False, msg, None
@staticmethod
def _get_identity_provider_class(implementation):
if implementation == "psa":
return PSAAuthnz
elif implementation == "custos":
return CustosAuthnz
else:
return None
def _extend_cloudauthz_config(self, cloudauthz, request, sa_session, user_id):
config = copy.deepcopy(cloudauthz.config)
if cloudauthz.provider == "aws":
success, message, backend = self._get_authnz_backend(cloudauthz.authn.provider)
strategy = Strategy(request, None, Storage, backend.config)
on_the_fly_config(sa_session)
try:
config["id_token"] = cloudauthz.authn.get_id_token(strategy)
except requests.exceptions.HTTPError as e:
msg = (
"Sign-out from Galaxy and remove its access from `{}`, then log back in using `{}` "
"account.".format(self._unify_provider_name(cloudauthz.authn.provider), cloudauthz.authn.uid)
)
log.debug(
"Failed to get/refresh ID token for user with ID `{}` for assuming authz_id `{}`. "
"User may not have a refresh token. If the problem persists, set the `prompt` key to "
"`consent` in `oidc_backends_config.xml`, then restart Galaxy and ask user to: {}"
"Error Message: `{}`".format(user_id, cloudauthz.id, msg, e.response.text)
)
raise exceptions.AuthenticationFailed(
err_msg="An error occurred getting your ID token. {}. If the problem persists, please "
"contact Galaxy admin.".format(msg)
)
return config
[docs] @staticmethod
def can_user_assume_authn(trans, authn_id):
qres = trans.sa_session.query(model.UserAuthnzToken).get(authn_id)
if qres is None:
msg = "Authentication record with the given `authn_id` (`{}`) not found.".format(
trans.security.encode_id(authn_id)
)
log.debug(msg)
raise exceptions.ObjectNotFound(msg)
if qres.user_id != trans.user.id:
msg = "The request authentication with ID `{}` is not accessible to user with ID " "`{}`.".format(
trans.security.encode_id(authn_id), trans.security.encode_id(trans.user.id)
)
log.warning(msg)
raise exceptions.ItemAccessibilityException(msg)
[docs] @staticmethod
def try_get_authz_config(sa_session, user_id, authz_id):
"""
It returns a cloudauthz config (see model.CloudAuthz) with the
given ID; and raise an exception if either a config with given
ID does not exist, or the configuration is defined for a another
user than trans.user.
:type trans: galaxy.webapps.base.webapp.GalaxyWebTransaction
:param trans: Galaxy web transaction
:type authz_id: int
:param authz_id: The ID of a CloudAuthz configuration to be used for
getting temporary credentials.
:rtype : model.CloudAuthz
:return: a cloudauthz configuration.
"""
qres = sa_session.query(model.CloudAuthz).get(authz_id)
if qres is None:
raise exceptions.ObjectNotFound("An authorization configuration with given ID not found.")
if user_id != qres.user_id:
msg = (
"The request authorization configuration (with ID:`{}`) is not accessible for user with "
"ID:`{}`.".format(qres.id, user_id)
)
log.warning(msg)
raise exceptions.ItemAccessibilityException(msg)
return qres
[docs] def refresh_expiring_oidc_tokens_for_provider(self, trans, auth):
try:
success, message, backend = self._get_authnz_backend(auth.provider)
if success is False:
msg = f"An error occurred when refreshing user token on `{auth.provider}` identity provider: {message}"
log.error(msg)
return False
refreshed = backend.refresh(trans, auth)
if refreshed:
log.debug(f"Refreshed user token via `{auth.provider}` identity provider")
return True
except Exception as e:
msg = f"An error occurred when refreshing user token: {e}"
log.error(msg)
return False
[docs] def refresh_expiring_oidc_tokens(self, trans, user=None):
user = trans.user or user
if not isinstance(user, model.User):
return
for auth in user.custos_auth or []:
self.refresh_expiring_oidc_tokens_for_provider(trans, auth)
for auth in user.social_auth or []:
self.refresh_expiring_oidc_tokens_for_provider(trans, auth)
[docs] def authenticate(self, provider, trans, idphint=None):
"""
:type provider: string
:param provider: set the name of the identity provider to be
used for authentication flow.
:type trans: GalaxyWebTransaction
:param trans: Galaxy web transaction.
:return: an identity provider specific authentication redirect URI.
"""
try:
success, message, backend = self._get_authnz_backend(provider, idphint=idphint)
if success is False:
return False, message, None
elif provider in KEYCLOAK_BACKENDS:
if self.allowed_idps and (idphint not in self.allowed_idps):
msg = f"An error occurred when authenticating a user. Invalid EntityID: `{idphint}`"
log.exception(msg)
return False, msg, None
return (
True,
f"Redirecting to the `{provider}` identity provider for authentication",
backend.authenticate(trans, idphint),
)
return (
True,
f"Redirecting to the `{provider}` identity provider for authentication",
backend.authenticate(trans),
)
except Exception:
msg = f"An error occurred when authenticating a user on `{provider}` identity provider"
log.exception(msg)
return False, msg, None
[docs] def callback(self, provider, state_token, authz_code, trans, login_redirect_url, idphint=None):
try:
success, message, backend = self._get_authnz_backend(provider, idphint=idphint)
if success is False:
return False, message, (None, None)
return success, message, backend.callback(state_token, authz_code, trans, login_redirect_url)
except exceptions.AuthenticationFailed:
raise
except Exception:
msg = f"An error occurred when handling callback from `{provider}` identity provider. Please contact an administrator for assistance."
log.exception(msg)
return False, msg, (None, None)
[docs] def create_user(self, provider, token, trans, login_redirect_url):
try:
success, message, backend = self._get_authnz_backend(provider)
if success is False:
return False, message, (None, None)
return success, message, backend.create_user(token, trans, login_redirect_url)
except exceptions.AuthenticationFailed:
log.exception("Error creating user")
raise
except Exception:
msg = f"An error occurred when creating a user with `{provider}` identity provider. Please contact an administrator for assistance."
log.exception(msg)
return False, msg, (None, None)
[docs] def logout(self, provider, trans, post_user_logout_href=None):
"""
Log the user out of the identity provider.
:type provider: string
:param provider: set the name of the identity provider.
:type trans: GalaxyWebTransaction
:param trans: Galaxy web transaction.
:type post_user_logout_href: string
:param post_user_logout_href: (Optional) URL for identity provider
to redirect to after logging user out.
:return: a tuple (success boolean, message, redirect URI)
"""
try:
# check if logout is enabled for this idp and return false if not
unified_provider_name = self._unify_provider_name(provider)
if self.oidc_backends_config[unified_provider_name]["enable_idp_logout"] is False:
return False, f"IDP logout is not enabled for {provider}", None
success, message, backend = self._get_authnz_backend(provider)
if success is False:
return False, message, None
return True, message, backend.logout(trans, post_user_logout_href)
except Exception:
msg = f"An error occurred when logging out from `{provider}` identity provider. Please contact an administrator for assistance."
log.exception(msg)
return False, msg, None
[docs] def disconnect(self, provider, trans, email=None, disconnect_redirect_url=None, idphint=None):
try:
success, message, backend = self._get_authnz_backend(provider, idphint=idphint)
if success is False:
return False, message, None
elif provider in KEYCLOAK_BACKENDS:
return backend.disconnect(provider, trans, email, disconnect_redirect_url)
return backend.disconnect(provider, trans, disconnect_redirect_url)
except Exception:
msg = (
"An error occurred when disconnecting authentication with `{}` identity provider for user `{}`".format(
provider, trans.user.username
)
)
log.exception(msg)
return False, msg, None
[docs] def get_cloud_access_credentials(self, cloudauthz, sa_session, user_id, request=None):
"""
This method leverages CloudAuthz (https://github.com/galaxyproject/cloudauthz)
to request a cloud-based resource provider (e.g., Amazon AWS, Microsoft Azure)
for temporary access credentials to a given resource.
It first checks if a cloudauthz config with the given ID (`authz_id`) is
available and can be assumed by the user, and raises an exception if either
is false. Otherwise, it then extends the cloudauthz configuration as required
by the CloudAuthz library for the provider specified in the configuration.
For instance, it adds on-the-fly values such as a valid OpenID Connect
identity token, as required by CloudAuthz for AWS. Then requests temporary
credentials from the CloudAuthz library using the updated configuration.
:type cloudauthz: CloudAuthz
:param cloudauthz: an instance of CloudAuthz to be used for getting temporary
credentials.
:type sa_session: sqlalchemy.orm.scoping.scoped_session
:param sa_session: SQLAlchemy database handle.
:type user_id: int
:param user_id: Decoded Galaxy user ID.
:type request: galaxy.web.framework.base.Request
:param request: Encapsulated HTTP(S) request.
:rtype: dict
:return: a dictionary containing credentials to access a cloud-based
resource provider. See CloudAuthz (https://github.com/galaxyproject/cloudauthz)
for details on the content of this dictionary.
"""
config = self._extend_cloudauthz_config(cloudauthz, request, sa_session, user_id)
try:
ca = CloudAuthz()
log.info(
"Requesting credentials using CloudAuthz with config id `{}` on be half of user `{}`.".format(
cloudauthz.id, user_id
)
)
credentials = ca.authorize(cloudauthz.provider, config)
return credentials
except CloudAuthzBaseException as e:
log.info(e)
raise exceptions.AuthenticationFailed(e)
except NotImplementedError as e:
log.info(e)
raise exceptions.RequestParameterInvalidException(e)
[docs] def get_cloud_access_credentials_in_file(self, new_file_path, cloudauthz, sa_session, user_id, request=None):
"""
This method leverages CloudAuthz (https://github.com/galaxyproject/cloudauthz)
to request a cloud-based resource provider (e.g., Amazon AWS, Microsoft Azure)
for temporary access credentials to a given resource.
This method uses the `get_cloud_access_credentials` method to obtain temporary
credentials, and persists them to a (temporary) file, and returns the file path.
:type new_file_path: str
:param new_file_path: Where dataset files are saved on temporary storage.
See `app.config.new_file_path`.
:type cloudauthz: CloudAuthz
:param cloudauthz: an instance of CloudAuthz to be used for getting temporary
credentials.
:type sa_session: sqlalchemy.orm.scoping.scoped_session
:param sa_session: SQLAlchemy database handle.
:type user_id: int
:param user_id: Decoded Galaxy user ID.
:type request: galaxy.web.framework.base.Request
:param request: [Optional] Encapsulated HTTP(S) request.
:rtype: str
:return: The filename to which credentials are written.
"""
filename = os.path.abspath(
os.path.join(
new_file_path,
"cd_"
+ "".join(random.SystemRandom().choice(string.ascii_uppercase + string.digits) for _ in range(11)),
)
)
credentials = self.get_cloud_access_credentials(cloudauthz, sa_session, user_id, request)
log.info(
"Writing credentials generated using CloudAuthz with config id `{}` to the following file: `{}`"
"".format(cloudauthz.id, filename)
)
with open(filename, "w") as f:
f.write(json.dumps(credentials))
return filename