127 lines
3.7 KiB
Python
127 lines
3.7 KiB
Python
import connexion
|
|
from flask import redirect, g
|
|
|
|
from crc import sso, app, db, auth
|
|
from crc.api.common import ApiError
|
|
from crc.models.user import UserModel, UserModelSchema
|
|
|
|
"""
|
|
.. module:: crc.api.user
|
|
:synopsis: Single Sign On (SSO) user login and session handlers
|
|
"""
|
|
|
|
|
|
@auth.verify_token
|
|
def verify_token(token):
|
|
try:
|
|
resp = UserModel.decode_auth_token(token)
|
|
g.user = UserModel.query.filter_by(uid=resp).first()
|
|
except:
|
|
return False
|
|
|
|
if g.user is not None:
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
|
|
@auth.login_required
|
|
def get_current_user():
|
|
return UserModelSchema().dump(g.user)
|
|
|
|
|
|
@sso.login_handler
|
|
def sso_login(user_info):
|
|
# TODO: Get redirect URL from Shibboleth request header
|
|
_handle_login(user_info)
|
|
|
|
|
|
def _handle_login(user_info, redirect_url=app.config['FRONTEND_AUTH_CALLBACK']):
|
|
"""On successful login, adds user to database if the user is not already in the system,
|
|
then returns the frontend auth callback URL, with auth token appended.
|
|
|
|
Args:
|
|
user_info (dict of {
|
|
uid: str,
|
|
affiliation: Optional[str],
|
|
display_name: Optional[str],
|
|
email_address: Optional[str],
|
|
eppn: Optional[str],
|
|
first_name: Optional[str],
|
|
last_name: Optional[str],
|
|
title: Optional[str],
|
|
}): Dictionary of user attributes
|
|
redirect_url: Optional[str]
|
|
|
|
Returns:
|
|
Response. 302 - Redirects to the frontend auth callback URL, with auth token appended.
|
|
"""
|
|
uid = user_info['uid']
|
|
user = db.session.query(UserModel).filter(UserModel.uid == uid).first()
|
|
|
|
if user is None:
|
|
# Add new user
|
|
user = UserModelSchema().load(user_info, session=db.session)
|
|
else:
|
|
# Update existing user data
|
|
user = UserModelSchema().load(user_info, session=db.session, instance=user, partial=True)
|
|
|
|
# Build display_name if not set
|
|
if 'display_name' not in user_info or len(user_info['display_name']) == 0:
|
|
display_name_list = []
|
|
|
|
for prop in ['first_name', 'last_name']:
|
|
if prop in user_info and len(user_info[prop]) > 0:
|
|
display_name_list.append(user_info[prop])
|
|
|
|
user.display_name = ' '.join(display_name_list)
|
|
|
|
db.session.add(user)
|
|
db.session.commit()
|
|
|
|
# Return the frontend auth callback URL, with auth token appended.
|
|
auth_token = user.encode_auth_token().decode()
|
|
return redirect('%s/%s' % (redirect_url, auth_token))
|
|
|
|
|
|
def backdoor(
|
|
uid=None,
|
|
affiliation=None,
|
|
display_name=None,
|
|
email_address=None,
|
|
eppn=None,
|
|
first_name=None,
|
|
last_name=None,
|
|
title=None,
|
|
redirect_url=None,
|
|
):
|
|
"""A backdoor for end-to-end system testing that allows the system to simulate logging in as a specific user.
|
|
Only works if the application is running in a non-production environment.
|
|
|
|
Args:
|
|
uid: str
|
|
affiliation: Optional[str]
|
|
display_name: Optional[str]
|
|
email_address: Optional[str]
|
|
eppn: Optional[str]
|
|
first_name: Optional[str]
|
|
last_name: Optional[str]
|
|
title: Optional[str]
|
|
redirect_url: Optional[str]
|
|
|
|
Returns:
|
|
str. If not on production, returns the frontend auth callback URL, with auth token appended.
|
|
|
|
Raises:
|
|
ApiError. If on production, returns a 404 error.
|
|
"""
|
|
if not 'PRODUCTION' in app.config or not app.config['PRODUCTION']:
|
|
user_info = {}
|
|
for key in UserModel.__dict__.keys():
|
|
if key in connexion.request.args:
|
|
user_info[key] = connexion.request.args[key]
|
|
|
|
return _handle_login(user_info, redirect_url)
|
|
else:
|
|
raise ApiError('404', 'unknown')
|