import flask
from flask import g, request
from crc import app, db
from crc.api.common import ApiError
from crc.models.user import UserModel, UserModelSchema
from crc.services.ldap_service import LdapService, LdapModel
"""
.. module:: crc.api.user
:synopsis: Single Sign On (SSO) user login and session handlers
"""
def verify_token(token=None):
"""
Verifies the token for the user (if provided). If in production environment and token is not provided,
gets user from the SSO headers and returns their token.
Args:
token: Optional[str]
Returns:
token: str
Raises:
ApiError. If not on production and token is not valid, returns an 'invalid_token' 403 error.
If on production and user is not authenticated, returns a 'no_user' 403 error.
"""
failure_error = ApiError("invalid_token", "Unable to decode the token you provided. Please re-authenticate",
status_code=403)
if not _is_production() and (token is None or 'user' not in g):
g.user = UserModel.query.first()
token = g.user.encode_auth_token()
if token:
try:
token_info = UserModel.decode_auth_token(token)
g.user = UserModel.query.filter_by(uid=token_info['sub']).first()
except:
raise failure_error
if g.user is not None:
return token_info
else:
raise failure_error
# If there's no token and we're in production, get the user from the SSO headers and return their token
if not token and _is_production():
uid = _get_request_uid(request)
if uid is not None:
db_user = UserModel.query.filter_by(uid=uid).first()
if db_user is not None:
g.user = db_user
token = g.user.encode_auth_token().decode()
token_info = UserModel.decode_auth_token(token)
return token_info
else:
raise ApiError("no_user", "User not found. Please login via the frontend app before accessing this feature.",
status_code=403)
def verify_token_admin(token=None):
"""
Verifies the token for the user (if provided) in non-production environment. If in production environment,
checks that the user is in the list of authorized admins
Args:
token: Optional[str]
Returns:
token: str
"""
# If this is production, check that the user is in the list of admins
if _is_production():
uid = _get_request_uid(request)
if uid is not None and uid in app.config['ADMIN_UIDS']:
return verify_token()
# If we're not in production, just use the normal verify_token method
else:
return verify_token(token)
def get_current_user():
return UserModelSchema().dump(g.user)
def login(
uid=None,
redirect_url=None,
):
"""
In non-production environment, provides an endpoint for end-to-end system testing that allows the system
to simulate logging in as a specific user. In production environment, simply logs user in via single-sign-on
(SSO) Shibboleth authentication headers.
Args:
uid: Optional[str]
redirect_url: Optional[str]
Returns:
str. If not on production, returns the frontend auth callback URL, with auth token appended.
If on production and user is authenticated via SSO, returns the frontend auth callback URL,
with auth token appended.
Raises:
ApiError. If on production and user is not authenticated, returns a 404 error.
"""
# ----------------------------------------
# Shibboleth Authentication Headers
# ----------------------------------------
# X-Remote-Cn: Daniel Harold Funk (dhf8r)
# X-Remote-Sn: Funk
# X-Remote-Givenname: Daniel
# X-Remote-Uid: dhf8r
# Eppn: dhf8r@virginia.edu
# Cn: Daniel Harold Funk (dhf8r)
# Sn: Funk
# Givenname: Daniel
# Uid: dhf8r
# X-Remote-User: dhf8r@virginia.edu
# X-Forwarded-For: 128.143.0.10
# X-Forwarded-Host: dev.crconnect.uvadcos.io
# X-Forwarded-Server: dev.crconnect.uvadcos.io
# Connection: Keep-Alive
# If we're in production, override any uid with the uid from the SSO request headers
if _is_production():
uid = _get_request_uid(request)
if uid:
app.logger.info("SSO_LOGIN: Full URL: " + request.url)
app.logger.info("SSO_LOGIN: User Id: " + uid)
app.logger.info("SSO_LOGIN: Will try to redirect to : " + str(redirect_url))
ldap_info = LdapService().user_info(uid)
if ldap_info:
return _handle_login(ldap_info, redirect_url)
raise ApiError('404', 'unknown')
@app.route('/sso')
def sso():
response = ""
response += "Headers
"
response += ""
for k, v in request.headers:
response += "- %s %s
\n" % (k, v)
response += "Environment
"
for k, v in request.environ:
response += "- %s %s
\n" % (k, v)
return response
def _handle_login(user_info: LdapModel, redirect_url=None):
"""
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 - an ldap user_info object.
redirect_url: Optional[str]
Returns:
Response. 302 - Redirects to the frontend auth callback URL, with auth token appended.
"""
user = _upsert_user(user_info)
g.user = user
# Return the frontend auth callback URL, with auth token appended.
auth_token = user.encode_auth_token().decode()
if redirect_url is not None:
if redirect_url.find("http://") != 0 and redirect_url.find("https://") != 0:
redirect_url = "http://" + redirect_url
url = '%s?token=%s' % (redirect_url, auth_token)
app.logger.info("SSO_LOGIN: REDIRECTING TO: " + url)
return flask.redirect(url, code=302)
else:
app.logger.info("SSO_LOGIN: NO REDIRECT, JUST RETURNING AUTH TOKEN.")
return auth_token
def _upsert_user(user_info):
user = db.session.query(UserModel).filter(UserModel.uid == user_info.uid).first()
if user is None:
# Add new user
user = UserModel()
else:
user = db.session.query(UserModel).filter(UserModel.uid == user_info.uid).with_for_update().first()
user.uid = user_info.uid
user.display_name = user_info.display_name
user.email_address = user_info.email_address
user.affiliation = user_info.affiliation
user.title = user_info.title
db.session.add(user)
db.session.commit()
return user
def _get_request_uid(req):
uid = None
if _is_production():
if 'user' in g and g.user is not None:
return g.user.uid
uid = req.headers.get("Uid")
if not uid:
uid = req.headers.get("X-Remote-Uid")
if not uid:
raise ApiError("invalid_sso_credentials", "'Uid' nor 'X-Remote-Uid' were present in the headers: %s"
% str(req.headers))
return uid
def _is_production():
return 'PRODUCTION' in app.config and app.config['PRODUCTION']