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.custos_authnz
import base64
import hashlib
import json
import logging
import os
import time
from dataclasses import dataclass
from datetime import (
datetime,
timedelta,
)
from typing import (
List,
Optional,
)
from urllib.parse import quote
import jwt
import requests
from oauthlib.common import generate_nonce
from requests_oauthlib import OAuth2Session
from galaxy import (
exceptions,
util,
)
from galaxy.model import (
CustosAuthnzToken,
User,
)
from galaxy.model.base import transaction
from galaxy.model.orm.util import add_object_to_object_session
from . import IdentityProvider
try:
import pkce
except ImportError:
pkce = None # type: ignore[assignment]
log = logging.getLogger(__name__)
STATE_COOKIE_NAME = "galaxy-oidc-state"
NONCE_COOKIE_NAME = "galaxy-oidc-nonce"
VERIFIER_COOKIE_NAME = "galaxy-oidc-verifier"
KEYCLOAK_BACKENDS = {"custos", "cilogon", "keycloak"}
[docs]@dataclass
class CustosAuthnzConfiguration:
provider: str
verify_ssl: Optional[bool]
url: str
label: str
client_id: str
client_secret: str
require_create_confirmation: bool
redirect_uri: str
ca_bundle: Optional[str]
pkce_support: bool
extra_params: Optional[dict]
authorization_endpoint: Optional[str]
token_endpoint: Optional[str]
end_session_endpoint: Optional[str]
well_known_oidc_config_uri: Optional[str]
iam_client_secret: Optional[str]
userinfo_endpoint: Optional[str]
credential_url: Optional[str]
[docs]class OIDCAuthnzBase(IdentityProvider):
[docs] def __init__(self, provider, oidc_config, oidc_backend_config, idphint=None):
provider = provider.lower()
self.config = CustosAuthnzConfiguration(
provider=provider,
verify_ssl=oidc_config["VERIFY_SSL"],
url=oidc_backend_config["url"],
label=oidc_backend_config.get("label", provider.capitalize()),
client_id=oidc_backend_config["client_id"],
client_secret=oidc_backend_config["client_secret"],
require_create_confirmation=oidc_backend_config.get("require_create_confirmation", provider == "custos"),
redirect_uri=oidc_backend_config["redirect_uri"],
ca_bundle=oidc_backend_config.get("ca_bundle", None),
pkce_support=oidc_backend_config.get("pkce_support", False),
extra_params={},
authorization_endpoint=None,
token_endpoint=None,
end_session_endpoint=None,
well_known_oidc_config_uri=None,
iam_client_secret=None,
userinfo_endpoint=None,
credential_url=None,
)
def _decode_token_no_signature(self, token):
return jwt.decode(token, audience=self.config.client_id, options={"verify_signature": False})
[docs] def refresh(self, trans, custos_authnz_token):
if custos_authnz_token is None:
raise exceptions.AuthenticationFailed("cannot find authorized user while refreshing token")
id_token_decoded = self._decode_token_no_signature(custos_authnz_token.id_token)
# do not refresh tokens if they didn't reach their half lifetime
if int(id_token_decoded["iat"]) + int(id_token_decoded["exp"]) > 2 * int(time.time()):
return False
log.info(custos_authnz_token.access_token)
oauth2_session = self._create_oauth2_session()
token_endpoint = self.config.token_endpoint
if self.config.iam_client_secret:
client_secret = self.config.iam_client_secret
else:
client_secret = self.config.client_secret
clientIdAndSec = f"{self.config.client_id}:{self.config.client_secret}" # for custos
params = {
"client_id": self.config.client_id,
"client_secret": client_secret,
"refresh_token": custos_authnz_token.refresh_token,
"headers": {
"Authorization": f"Basic {util.unicodify(base64.b64encode(util.smart_str(clientIdAndSec)))}"
}, # for custos
}
token = oauth2_session.refresh_token(token_endpoint, **params)
processed_token = self._process_token(trans, oauth2_session, token, False)
custos_authnz_token.access_token = processed_token["access_token"]
custos_authnz_token.id_token = processed_token["id_token"]
custos_authnz_token.refresh_token = processed_token["refresh_token"]
custos_authnz_token.expiration_time = processed_token["expiration_time"]
custos_authnz_token.refresh_expiration_time = processed_token["refresh_expiration_time"]
trans.sa_session.add(custos_authnz_token)
trans.sa_session.flush()
return True
def _get_provider_specific_scopes(self):
return []
[docs] def authenticate(self, trans, idphint=None):
base_authorize_url = self.config.authorization_endpoint
scopes = ["openid", "email", "profile"]
scopes.extend(self._get_provider_specific_scopes())
oauth2_session = self._create_oauth2_session(scope=scopes)
nonce = generate_nonce()
nonce_hash = self._hash_nonce(nonce)
extra_params = {"nonce": nonce_hash}
if idphint is not None:
extra_params["idphint"] = idphint
if self.config.pkce_support:
if not pkce:
raise InvalidAuthnzConfigException(
"The python 'pkce' library is not installed but Galaxy is configured to use it "
"(see oidc_backends_config). Make sure pkce is installed correctly to proceed."
)
code_verifier, code_challenge = pkce.generate_pkce_pair(96)
extra_params["code_challenge"] = code_challenge
extra_params["code_challenge_method"] = "S256"
trans.set_cookie(value=code_verifier, name=VERIFIER_COOKIE_NAME)
if self.config.extra_params:
extra_params.update(self.config.extra_params)
authorization_url, state = oauth2_session.authorization_url(base_authorize_url, **extra_params)
trans.set_cookie(value=state, name=STATE_COOKIE_NAME)
trans.set_cookie(value=nonce, name=NONCE_COOKIE_NAME)
return authorization_url
def _process_token(self, trans, oauth2_session, token, validate_nonce=True):
processed_token = {}
processed_token["access_token"] = token["access_token"]
processed_token["id_token"] = token["id_token"]
processed_token["refresh_token"] = token["refresh_token"] if "refresh_token" in token else None
processed_token["expiration_time"] = datetime.now() + timedelta(seconds=token.get("expires_in", 3600))
processed_token["refresh_expiration_time"] = (
(datetime.now() + timedelta(seconds=token["refresh_expires_in"])) if "refresh_expires_in" in token else None
)
# Get nonce from token['id_token'] and validate. 'nonce' in the
# id_token is a hash of the nonce stored in the NONCE_COOKIE_NAME
# cookie.
id_token_decoded = self._decode_token_no_signature(processed_token["id_token"])
if validate_nonce:
nonce_hash = id_token_decoded["nonce"]
self._validate_nonce(trans, nonce_hash)
# Get userinfo and lookup/create Galaxy user record
if id_token_decoded.get("email", None):
userinfo = id_token_decoded
else:
userinfo = self._get_userinfo(oauth2_session)
processed_token["email"] = userinfo["email"]
processed_token["user_id"] = userinfo["sub"]
processed_token["username"] = self._username_from_userinfo(trans, userinfo)
return processed_token
[docs] def callback(self, state_token, authz_code, trans, login_redirect_url):
# Take state value to validate from token. OAuth2Session.fetch_token
# will validate that the state query parameter value on the URL matches
# this value.
state_cookie = trans.get_cookie(name=STATE_COOKIE_NAME)
oauth2_session = self._create_oauth2_session(state=state_cookie)
token = self._fetch_token(oauth2_session, trans)
processed_token = self._process_token(trans, oauth2_session, token)
user_id = processed_token["user_id"]
email = processed_token["email"]
username = processed_token["username"]
access_token = processed_token["access_token"]
id_token = processed_token["id_token"]
refresh_token = processed_token["refresh_token"]
expiration_time = processed_token["expiration_time"]
refresh_expiration_time = processed_token["refresh_expiration_time"]
# Create or update custos_authnz_token record
custos_authnz_token = self._get_custos_authnz_token(trans.sa_session, user_id, self.config.provider)
if custos_authnz_token is None:
user = trans.user
existing_user = trans.sa_session.query(User).filter_by(email=email).first()
if not user:
if existing_user:
if trans.app.config.fixed_delegated_auth:
user = existing_user
else:
message = f"There already exists a user with email {email}. To associate this external login, you must first be logged in as that existing account."
log.info(message)
login_redirect_url = (
f"{login_redirect_url}login/start"
f"?connect_external_provider={self.config.provider}"
f"&connect_external_email={email}"
f"&connect_external_label={self.config.label}"
)
return login_redirect_url, None
elif self.config.require_create_confirmation:
login_redirect_url = f"{login_redirect_url}login/start?confirm=true&provider_token={json.dumps(token)}&provider={self.config.provider}"
return login_redirect_url, None
else:
user = trans.app.user_manager.create(email=email, username=username)
if trans.app.config.user_activation_on:
trans.app.user_manager.send_activation_email(trans, email, username)
# Create a token to link this identity with an existing account
custos_authnz_token = CustosAuthnzToken(
user=user,
external_user_id=user_id,
provider=self.config.provider,
access_token=access_token,
id_token=id_token,
refresh_token=refresh_token,
expiration_time=expiration_time,
refresh_expiration_time=refresh_expiration_time,
)
label = self.config.label
if trans.app.config.fixed_delegated_auth:
redirect_url = login_redirect_url
elif existing_user and existing_user != user:
redirect_url = (
f"{login_redirect_url}user/external_ids"
f"?email_exists={email}"
f"¬ification=Your%20{label}%20identity%20has%20been%20linked"
"%20to%20your%20Galaxy%20account."
)
else:
redirect_url = (
f"{login_redirect_url}user/external_ids"
f"?notification=Your%20{label}%20identity%20has%20been%20linked"
"%20to%20your%20Galaxy%20account."
)
else:
# Identity is already linked to account - login as usual
custos_authnz_token.access_token = access_token
custos_authnz_token.id_token = id_token
custos_authnz_token.refresh_token = refresh_token
custos_authnz_token.expiration_time = expiration_time
custos_authnz_token.refresh_expiration_time = refresh_expiration_time
redirect_url = "/"
trans.sa_session.add(custos_authnz_token)
with transaction(trans.sa_session):
trans.sa_session.commit()
return redirect_url, custos_authnz_token.user
[docs] def create_user(self, token, trans, login_redirect_url):
token_dict = json.loads(token)
access_token = token_dict["access_token"]
id_token = token_dict["id_token"]
refresh_token = token_dict["refresh_token"] if "refresh_token" in token_dict else None
expiration_time = datetime.now() + timedelta(
seconds=token_dict.get("expires_in", 3600)
) # might be a problem cause times no long valid
refresh_expiration_time = (
(datetime.now() + timedelta(seconds=token_dict["refresh_expires_in"]))
if "refresh_expires_in" in token_dict
else None
)
# Get nonce from token['id_token'] and validate. 'nonce' in the
# id_token is a hash of the nonce stored in the NONCE_COOKIE_NAME
# cookie.
userinfo = self._decode_token_no_signature(id_token)
# Get userinfo and create Galaxy user record
email = userinfo["email"]
# Check if username if already taken
username = self._username_from_userinfo(trans, userinfo)
user_id = userinfo["sub"]
user = trans.app.user_manager.create(email=email, username=username)
if trans.app.config.user_activation_on:
trans.app.user_manager.send_activation_email(trans, email, username)
custos_authnz_token = CustosAuthnzToken(
external_user_id=user_id,
provider=self.config.provider,
access_token=access_token,
id_token=id_token,
refresh_token=refresh_token,
expiration_time=expiration_time,
refresh_expiration_time=refresh_expiration_time,
)
add_object_to_object_session(custos_authnz_token, user)
custos_authnz_token.user = user
trans.sa_session.add(user)
trans.sa_session.add(custos_authnz_token)
with transaction(trans.sa_session):
trans.sa_session.commit()
return login_redirect_url, user
[docs] def disconnect(self, provider, trans, email=None, disconnect_redirect_url=None):
try:
user = trans.user
index = 0
# Find CustosAuthnzToken record for this provider (should only be one)
provider_tokens = [token for token in user.custos_auth if token.provider == self.config.provider]
if len(provider_tokens) == 0:
raise Exception(f"User is not associated with provider {self.config.provider}")
if len(provider_tokens) > 1:
for idx, token in enumerate(provider_tokens):
id_token_decoded = self._decode_token_no_signature(token.id_token)
if id_token_decoded["email"] == email:
index = idx
trans.sa_session.delete(provider_tokens[index])
with transaction(trans.sa_session):
trans.sa_session.commit()
return True, "", disconnect_redirect_url
except Exception as e:
return False, f"Failed to disconnect provider {provider}: {util.unicodify(e)}", None
[docs] def logout(self, trans, post_user_logout_href=None):
if not self.config.redirect_uri:
log.error("Failed to generate logout redirect_url")
return None
try:
if self.config.end_session_endpoint:
redirect_url = self.config.end_session_endpoint
if post_user_logout_href is not None:
redirect_url += f"?redirect_uri={quote(post_user_logout_href)}"
return redirect_url
except Exception as e:
log.error("Failed to generate logout redirect_url", exc_info=e)
return None
def _create_oauth2_session(self, state=None, scope=None):
client_id = self.config.client_id
redirect_uri = self.config.redirect_uri
if redirect_uri.startswith("http://localhost") and os.environ.get("OAUTHLIB_INSECURE_TRANSPORT", None) != "1":
log.warning("Setting OAUTHLIB_INSECURE_TRANSPORT to '1' to allow plain HTTP (non-SSL) callback")
os.environ["OAUTHLIB_INSECURE_TRANSPORT"] = "1"
session = OAuth2Session(client_id, scope=scope, redirect_uri=redirect_uri, state=state)
session.verify = self._get_verify_param()
return session
def _fetch_token(self, oauth2_session, trans):
if self.config.iam_client_secret:
# Custos uses the Keycloak client secret to get the token
client_secret = self.config.iam_client_secret
else:
client_secret = self.config.client_secret
token_endpoint = self.config.token_endpoint
clientIdAndSec = f"{self.config.client_id}:{self.config.client_secret}" # for custos
params = {
"client_secret": client_secret,
"authorization_response": trans.request.url,
"headers": {
"Authorization": f"Basic {util.unicodify(base64.b64encode(util.smart_str(clientIdAndSec)))}"
}, # for custos
"verify": self._get_verify_param(),
}
if self.config.pkce_support:
code_verifier = trans.get_cookie(name=VERIFIER_COOKIE_NAME)
trans.set_cookie("", name=VERIFIER_COOKIE_NAME, age=-1)
params["code_verifier"] = code_verifier
return oauth2_session.fetch_token(token_endpoint, **params)
def _get_userinfo(self, oauth2_session):
userinfo_endpoint = self.config.userinfo_endpoint
return oauth2_session.get(userinfo_endpoint, verify=self._get_verify_param()).json()
@staticmethod
def _get_custos_authnz_token(sa_session, user_id, provider):
return sa_session.query(CustosAuthnzToken).filter_by(external_user_id=user_id, provider=provider).one_or_none()
@staticmethod
def _hash_nonce(nonce):
return hashlib.sha256(util.smart_str(nonce)).hexdigest()
def _validate_nonce(self, trans, nonce_hash):
nonce_cookie = trans.get_cookie(name=NONCE_COOKIE_NAME)
# Delete the nonce cookie
trans.set_cookie("", name=NONCE_COOKIE_NAME, age=-1)
nonce_cookie_hash = self._hash_nonce(nonce_cookie)
if nonce_hash != nonce_cookie_hash:
raise Exception("Nonce mismatch. Check that configured redirect_uri matches the URL you are using.")
def _load_config(self, headers: Optional[dict] = None, params: Optional[dict] = None):
if not headers:
headers = {}
if not params:
params = {}
self.config.well_known_oidc_config_uri = self._get_well_known_uri_from_url(self.config.provider)
if not self.config.well_known_oidc_config_uri:
log.error(f"Failed to load well-known OIDC config URI: {self.config.well_known_oidc_config_uri}")
raise Exception(f"Failed to load well-known OIDC config URI: {self.config.well_known_oidc_config_uri}")
try:
well_known_oidc_config = requests.get(
self.config.well_known_oidc_config_uri,
headers=headers,
verify=self._get_verify_param(),
timeout=util.DEFAULT_SOCKET_TIMEOUT,
params=params,
).json()
self._load_well_known_oidc_config(well_known_oidc_config)
except Exception:
log.error(f"Failed to load well-known OIDC config URI: {self.config.well_known_oidc_config_uri}")
raise
def _get_well_known_uri_from_url(self, provider):
# TODO: Look up this URL from a Python library
base_url = self.config.url
# Remove potential trailing slash to avoid "//realms"
base_url = base_url if base_url[-1] != "/" else base_url[:-1]
return f"{base_url}/.well-known/openid-configuration"
def _load_well_known_oidc_config(self, well_known_oidc_config):
self.config.authorization_endpoint = well_known_oidc_config["authorization_endpoint"]
self.config.token_endpoint = well_known_oidc_config["token_endpoint"]
self.config.userinfo_endpoint = well_known_oidc_config["userinfo_endpoint"]
self.config.end_session_endpoint = well_known_oidc_config.get("end_session_endpoint")
def _get_verify_param(self):
"""Return 'ca_bundle' if 'verify_ssl' is true and 'ca_bundle' is configured."""
# in requests_oauthlib, the verify param can either be a boolean or a CA bundle path
if self.config.ca_bundle is not None and self.config.verify_ssl:
return self.config.ca_bundle
else:
return self.config.verify_ssl
@staticmethod
def _username_from_userinfo(trans, userinfo):
username = userinfo.get("preferred_username", userinfo["email"])
if "@" in username:
username = username.split("@")[0] # username created from username portion of email
username = util.ready_name_for_url(username)
if trans.sa_session.query(trans.app.model.User).filter_by(username=username).first():
# if username already exists in database, append integer and iterate until unique username found
count = 0
while trans.sa_session.query(trans.app.model.User).filter_by(username=(f"{username}{count}")).first():
count += 1
return f"{username}{count}"
else:
return username
[docs]class OIDCAuthnzBaseKeycloak(OIDCAuthnzBase):
[docs] def __init__(self, provider, oidc_config, oidc_backend_config, idphint=None):
super().__init__(provider, oidc_config, oidc_backend_config, idphint)
self.config.extra_params = {"kc_idp_hint": oidc_backend_config.get("idphint", "oidc")}
self._load_config()
[docs]class OIDCAuthnzBaseCiLogon(OIDCAuthnzBase):
[docs] def __init__(self, provider, oidc_config, oidc_backend_config, idphint=None):
super().__init__(provider, oidc_config, oidc_backend_config, idphint)
self.config.extra_params = {"kc_idp_hint": oidc_backend_config.get("idphint", "cilogon")}
self._load_config()
def _get_provider_specific_scopes(self):
return ["org.cilogon.userinfo"]
def _get_well_known_uri_from_url(self, provider):
base_url = self.config.url
# backwards compatibility. CILogon URL is given with /authorize in the examples. not sure if
# this applies to the wild, but let's be safe here and remove the /authorize if it exists
# which will lead to the correct openid configuration
base_url = base_url if base_url.split("/")[-1] != "authorize" else "/".join(base_url.split("/")[:-1])
return f"{base_url}/.well-known/openid-configuration"
[docs]class CustosAuthFactory:
@dataclass
class _CustosAuthBasedProviderCacheItem:
created_at: datetime
item: OIDCAuthnzBase
provider: str
oidc_config: dict
oidc_backend_config: dict
idphint: str
_CustosAuthBasedProvidersCache: List[_CustosAuthBasedProviderCacheItem] = []
[docs] @staticmethod
def GetCustosBasedAuthProvider(provider, oidc_config, oidc_backend_config, idphint=None):
# see if we have a config loaded up already
for item in CustosAuthFactory._CustosAuthBasedProvidersCache:
if (
item.provider == provider
and item.oidc_config == oidc_config
and item.oidc_backend_config == oidc_backend_config
and item.idphint == idphint
):
return item.item
auth_adapter: OIDCAuthnzBase
if provider.lower() == "custos":
auth_adapter = OIDCAuthnzBaseCustos(provider, oidc_config, oidc_backend_config, idphint)
elif provider.lower() == "keycloak":
auth_adapter = OIDCAuthnzBaseKeycloak(provider, oidc_config, oidc_backend_config, idphint)
elif provider.lower() == "cilogon":
auth_adapter = OIDCAuthnzBaseCiLogon(provider, oidc_config, oidc_backend_config, idphint)
else:
raise Exception(f"Unknown Custos provider name: {provider}")
if auth_adapter:
CustosAuthFactory._CustosAuthBasedProvidersCache.append(
CustosAuthFactory._CustosAuthBasedProviderCacheItem(
created_at=datetime.now(),
item=auth_adapter,
provider=provider,
oidc_config=oidc_config,
oidc_backend_config=oidc_backend_config,
idphint=idphint,
)
)
return auth_adapter
[docs]class OIDCAuthnzBaseCustos(OIDCAuthnzBase):
[docs] def __init__(self, provider, oidc_config, oidc_backend_config, idphint=None):
super().__init__(provider, oidc_config, oidc_backend_config, idphint)
self.config.extra_params = {"kc_idp_hint": oidc_backend_config.get("idphint", "oidc")}
self._load_config_for_custos()
def _get_custos_credentials(self):
clientIdAndSec = f"{self.config.client_id}:{self.config.client_secret}"
if not self.config.credential_url:
raise Exception(
f"Error OIDC provider {self.config.provider} is of type Custos, but does not have the credential url set"
)
creds = requests.get(
self.config.credential_url,
headers={"Authorization": f"Basic {util.unicodify(base64.b64encode(util.smart_str(clientIdAndSec)))}"},
verify=False,
params={"client_id": self.config.client_id},
timeout=util.DEFAULT_SOCKET_TIMEOUT,
)
credentials = creds.json()
self.config.iam_client_secret = credentials["iam_client_secret"]
def _load_config_for_custos(self):
self.config.credential_url = f"{self.config.url.rstrip('/')}/credentials"
self._get_custos_credentials()
# Set custos endpoints
clientIdAndSec = f"{self.config.client_id}:{self.config.client_secret}"
headers = {"Authorization": f"Basic {util.unicodify(base64.b64encode(util.smart_str(clientIdAndSec)))}"}
params = {"client_id": self.config.client_id}
self._load_config(headers, params)