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
from datetime import datetime, timedelta
from urllib.parse import quote
import jwt
import requests
from oauthlib.common import generate_nonce
from requests_oauthlib import OAuth2Session
from galaxy import exceptions
from galaxy import util
from galaxy.model import CustosAuthnzToken, User
from ..authnz import IdentityProvider
log = logging.getLogger(__name__)
STATE_COOKIE_NAME = 'galaxy-oidc-state'
NONCE_COOKIE_NAME = 'galaxy-oidc-nonce'
KEYCLOAK_BACKENDS = {'custos', 'cilogon', 'keycloak'}
[docs]class CustosAuthnz(IdentityProvider):
[docs] def __init__(self, provider, oidc_config, oidc_backend_config, idphint=None):
provider = provider.lower()
self.config = {'provider': provider}
self.config['verify_ssl'] = oidc_config['VERIFY_SSL']
self.config['url'] = oidc_backend_config['url']
self.config['client_id'] = oidc_backend_config['client_id']
self.config['client_secret'] = oidc_backend_config['client_secret']
self.config['redirect_uri'] = oidc_backend_config['redirect_uri']
self.config['ca_bundle'] = oidc_backend_config.get('ca_bundle', None)
self.config['extra_params'] = {
'kc_idp_hint': oidc_backend_config.get('idphint', 'oidc' if self.config['provider'] in ['custos', 'keycloak'] else 'cilogon')
}
if provider == 'cilogon':
self._load_config_for_cilogon()
elif provider == 'custos':
self._load_config_for_custos()
elif provider == 'keycloak':
self._load_config_for_keycloak()
def _decode_token_no_signature(self, token):
return jwt.decode(token, audience=self.config['client_id'], options={"verify_signature": False})
[docs] def authenticate(self, trans, idphint=None):
base_authorize_url = self.config['authorization_endpoint']
scopes = ['openid', 'email', 'profile']
if self.config['provider'] in ['custos', 'cilogon']:
scopes.append('org.cilogon.userinfo')
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 "extra_params" in self.config:
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
[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)
access_token = token['access_token']
id_token = token['id_token']
refresh_token = token['refresh_token'] if 'refresh_token' in token else None
expiration_time = datetime.now() + timedelta(seconds=token.get('expires_in', 3600))
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(id_token)
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)
email = userinfo['email']
user_id = userinfo['sub']
# 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
if not user:
existing_user = trans.sa_session.query(User).filter_by(email=email).first()
if existing_user:
# If there is only a single external authentication
# provider in use, trust the user provided and
# automatically associate.
# TODO: Future work will expand on this and provide an
# interface for when there are multiple auth providers
# allowing explicit authenticated association.
if (trans.app.config.enable_oidc
and len(trans.app.config.oidc) == 1
and len(trans.app.auth_manager.authenticators) == 0):
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.exception(message)
raise exceptions.AuthenticationFailed(message)
elif self.config['provider'] == 'custos':
login_redirect_url = f"{login_redirect_url}root/login?confirm=true&custos_token={json.dumps(token)}"
return login_redirect_url, None
else:
username = self._username_from_userinfo(trans, userinfo)
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(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)
else:
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
trans.sa_session.add(custos_authnz_token)
trans.sa_session.flush()
return "/", 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(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)
trans.sa_session.add(user)
trans.sa_session.add(custos_authnz_token)
trans.sa_session.flush()
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])
trans.sa_session.flush()
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_logout_redirect_url=None):
try:
redirect_url = self.config['end_session_endpoint']
if post_logout_redirect_url is not None:
redirect_url += f"?redirect_uri={quote(post_logout_redirect_url)}"
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.get('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
return oauth2_session.fetch_token(
token_endpoint,
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())
def _get_userinfo(self, oauth2_session):
userinfo_endpoint = self.config['userinfo_endpoint']
return oauth2_session.get(userinfo_endpoint,
verify=self._get_verify_param()).json()
def _get_custos_authnz_token(self, sa_session, user_id, provider):
return sa_session.query(CustosAuthnzToken).filter_by(
external_user_id=user_id, provider=provider).one_or_none()
def _hash_nonce(self, 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!")
def _load_config_for_cilogon(self):
# Set cilogon endpoints
self.config['authorization_endpoint'] = "https://cilogon.org/authorize"
self.config['token_endpoint'] = "https://cilogon.org/oauth2/token"
self.config['userinfo_endpoint'] = "https://cilogon.org/oauth2/userinfo"
def _load_config_for_custos(self):
self.config['well_known_oidc_config_uri'] = self._get_well_known_uri_from_url(self.config['provider'])
self.config['credential_url'] = '/'.join([self.config['url'].rstrip('/'), 'credentials'])
self._get_custos_credentials()
# Set custos endpoints
clientIdAndSec = f"{self.config['client_id']}:{self.config['client_secret']}"
eps = requests.get(self.config['well_known_oidc_config_uri'],
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)
well_known_oidc_config = eps.json()
self._load_well_known_oidc_config(well_known_oidc_config)
def _load_config_for_keycloak(self):
self.config['well_known_oidc_config_uri'] = self._get_well_known_uri_from_url(self.config['provider'])
well_known_oidc_config = self._fetch_well_known_oidc_config(self.config['well_known_oidc_config_uri'])
self._load_well_known_oidc_config(well_known_oidc_config)
def _get_custos_credentials(self):
clientIdAndSec = f"{self.config['client_id']}:{self.config['client_secret']}"
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 _get_well_known_uri_from_url(self, provider):
# TODO: Look up this URL from a Python library
if provider in ['custos', 'keycloak']:
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"
else:
raise Exception(f"Unknown Custos provider name: {provider}")
def _fetch_well_known_oidc_config(self, well_known_uri):
try:
return requests.get(well_known_uri,
verify=self._get_verify_param(),
timeout=util.DEFAULT_SOCKET_TIMEOUT).json()
except Exception:
log.error(f"Failed to load well-known OIDC config URI: {well_known_uri}")
raise
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']
def _username_from_userinfo(self, trans, userinfo):
username = userinfo.get('preferred_username', userinfo['email'])
if "@" in username:
username = username.split('@')[0] # username created from username portion of email
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