28570052741

Committed 02 Jul 2026 06:25AM UTC coverage: 75.387% (+0.3%) from 75.073%

Build # 28570052741

Build Type

Pull #1356

github

Committed by

nss10

Commit Message

Prevent audeince mismatch error.

Pull Request Pull Request #1356: MIDRC-1285 MIDRC-1291 Task tokens + Access token denylist + Validate aud

Coverage Stats

8613 of 11425 relevant lines covered (75.39%)

0.75 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

86.1

fence/auth.py

import re
import urllib.request, urllib.parse, urllib.error
from datetime import datetime
from functools import wraps
from json import JSONDecodeError

import backoff
import flask
from flask import current_app
from authutils.errors import JWTError, JWTExpiredError
from authutils.token.validate import (
    current_token,
    require_auth_header as authutils_require_auth_header,
    set_current_token,
    validate_request,
)
from cdislogging import get_logger
import requests

from fence.authz.auth import check_arborist_auth
from fence.config import config
from fence.errors import Unauthorized, InternalError
from fence.jwt.validate import validate_jwt
from fence.models import User, IdentityProvider, query_for_user
from fence.user import get_current_user
from fence.utils import (
    clear_cookies,
    DEFAULT_BACKOFF_SETTINGS,
    allowed_login_redirects,
    domain,
)

logger = get_logger(__name__)


def require_auth_header(*args, **kwargs):
    """
    Injects the default token audience before calling authutils's `require_auth_header`
    """
    if "audience" in kwargs:
        if type(kwargs["audience"]) != list:
            kwargs["audience"] = [kwargs["audience"]]
        kwargs["audience"].append("gen3")
    else:
        kwargs["audience"] = "gen3"

    return authutils_require_auth_header(*args, **kwargs)


def get_jwt():
    """
    Return the user's JWT from authorization header. Requires flask application context.
    Raises:
        - Unauthorized, if header is missing or not in the correct format
    """
    header = flask.request.headers.get("Authorization")
    if not header:
        raise Unauthorized("missing authorization header")
    try:
        bearer, token = header.split(" ")
    except ValueError:
        msg = "authorization header not in expected format"
        logger.debug(f"{msg}. Received header: {header}")
        logger.error(f"{msg}.")
        raise Unauthorized(msg)
    if bearer.lower() != "bearer":
        raise Unauthorized("expected bearer token in auth header")
    return token


def build_redirect_url(hostname, path):
    """
    Compute a redirect given a hostname and next path where
    Args:
        hostname (str): may be empty string or a bare hostname or
               a hostname with a protocal attached (https?://...)
        path (int): is a path to attach to hostname
    Return:
        string url suitable for flask.redirect
    """
    redirect_base = hostname
    # BASE_URL may be empty or a bare hostname or a hostname with a protocol
    if bool(redirect_base) and not redirect_base.startswith("http"):
        redirect_base = "https://" + redirect_base
    return redirect_base + path


def get_ip_information_string():
    """
    Returns a string containing the client's IP address and any X-Forwarded headers.

    Returns:
        str: A formatted string containing the client's IP address and X-Forwarded headers.
    """
    x_forwarded_headers = [
        f"{header}: {value}"
        for header, value in flask.request.headers
        if "X-Forwarded" in header
    ]
    return f"flask.request.remote_addr={flask.request.remote_addr} x_forwarded_headers={x_forwarded_headers}"


def _identify_user_and_update_database(
    user,
    username,
    provider,
    email=None,
    id_from_idp=None,
    username_deny_regex=None,
) -> bool:
    """
    Create a new user if one doesn't already exist in the database. Commit the user
    and associated idp information to the database.

    Args:
        user (User): user to be logged in, if it already exists, None otherwise
        username (str): specific username of user to be logged in
        provider (str): specfic idp of user to be logged in
        email (str, optional): email of user (may or may not match username depending
            on the IdP)
        id_from_idp (str, optional): id from the IDP (which may be different than
            the username)

    Return:
        User: the created or updated user
    """
    username_deny_regex = username_deny_regex or config["GLOBAL_USERNAME_DENY_REGEX"]
    if username_deny_regex:
        if re.search(pattern=username_deny_regex, string=username):
            logger.info(
                f"Blocked login of user with username {username} due to deny regex: {username_deny_regex}"
            )

            # intentionally empty message to prevent information leakage
            raise Unauthorized(message="")

    if user:
        if user.active == False:
            # Abort login if user.active == False:
            raise Unauthorized(
                "User is known but not authorized/activated in the system"
            )
        _update_users_email(user, email)
        _update_users_id_from_idp(user, id_from_idp)
        _update_users_last_auth(user)
    else:
        if not config["ALLOW_NEW_USER_ON_LOGIN"]:
            # do not create new active users automatically
            raise Unauthorized("New user is not yet authorized/activated in the system")

        # add the new user
        user = User(username=username)

        if email:
            user.email = email

        if id_from_idp:
            user.id_from_idp = id_from_idp
            # TODO: update iss_sub mapping table?

    # This expression is relevant to those users who already have user and
    # idp info persisted to the database. We avoid unnecessarily re-saving
    # that user and idp info.
    if not user.identity_provider or not user.identity_provider.name == provider:
        # setup idp connection for new user (or existing user w/o it setup)
        idp = (
            current_app.scoped_session()
            .query(IdentityProvider)
            .filter(IdentityProvider.name == provider)
            .first()
        )
        if not idp:
            idp = IdentityProvider(name=provider)

        user.identity_provider = idp
        current_app.scoped_session().add(user)
        current_app.scoped_session().commit()

    # `login_in_progress_username` stored for use by the user registration code.
    # not using `flask.session["username"]` because other code relies on it to know
    # whether a user is logged in; in this case the user isn't logged in yet.
    flask.session["login_in_progress_username"] = user.username

    flask.g.user = user
    return user


def _is_user_registration_required_before_login(user, provider) -> bool:
    auto_registration_enabled = (
        config["OPENID_CONNECT"]
        .get(provider, {})
        .get("enable_idp_users_registration", False)
    )
    # Registration is required if:
    # - Registration is enabled in the config, AND
    # - Automatic registration is NOT enabled, AND
    # - The user's registration info is empty
    return (
        config["REGISTER_USERS_ON"]
        and not auto_registration_enabled
        and user.additional_info.get("registration_info", {}) == {}
    )


def login_user_or_require_registration(
    username, provider, upstream_idp=None, shib_idp=None, email=None, id_from_idp=None
) -> bool:
    """
    Check if a user needs to go through the registration flow before being logged in. If not,
    login the user with the given username and provider. Set values in Flask session to indicate
    the user being logged in.

    Args:
        username (str): specific username of user to be logged in
        provider (str): specfic idp of user to be logged in
        upstream_idp (str, optional): upstream fence IdP
        shib_idp (str, optional): upstream shibboleth IdP
        email (str, optional): email of user (may or may not match username depending
            on the IdP)
        id_from_idp (str, optional): id from the IDP (which may be different than
            the username)

    Return:
        bool: whether the user has been logged in (if registration is enabled and the user is not
            registered, this would be False)
    """

    def log_ip(user):
        ip_info = get_ip_information_string()
        logger.info(
            f"User logged in. user.id={user.id} user.username={user.username} {ip_info}"
        )

    def set_flask_session_values(user):
        """
        Helper fuction to set user values in the session.

        Args:
            user (User): User object
        """
        flask.session["username"] = user.username
        flask.session["user_id"] = str(user.id)
        flask.session["provider"] = user.identity_provider.name
        if upstream_idp:
            flask.session["upstream_idp"] = upstream_idp
        if shib_idp:
            flask.session["shib_idp"] = shib_idp
        flask.g.user = user
        flask.g.scopes = ["_all"]
        flask.g.token = None

    user = query_for_user(session=current_app.scoped_session(), username=username)
    user = _identify_user_and_update_database(
        user, username, provider, email, id_from_idp
    )
    log_user_in = not _is_user_registration_required_before_login(user, provider)
    if log_user_in:
        set_flask_session_values(user)
        log_ip(user)
    return log_user_in


@backoff.on_exception(backoff.expo, Exception, **DEFAULT_BACKOFF_SETTINGS)
def get_openid_config_for_idp(open_id_connect):
    """
    Return openid configuration for a given provider.
    Args:
        open_id_connect (dict): fence config for idp
    Returns:
        response: response of openid configuration
    """
    well_known_url = open_id_connect["discovery_url"]
    well_known_resp = requests.get(well_known_url)
    well_known_resp.raise_for_status()
    return well_known_resp


def logout(next_url, force_era_global_logout=False):
    """
    Return a redirect which another logout from IDP or the provided redirect.
    Depending on the IDP, this logout will propogate. For example, if using
    another fence as an IDP, this will hit that fence's logout endpoint.
    Args:
        next_url (str): Final redirect desired after logout
    """
    logger.debug("IN AUTH LOGOUT, next_url = {0}".format(next_url))

    # propogate logout to IDP
    provider_logout = None
    provider = flask.session.get("provider")

    if force_era_global_logout or provider == IdentityProvider.itrust:
        safe_url = urllib.parse.quote_plus(next_url)
        provider_logout = config["ITRUST_GLOBAL_LOGOUT"] + safe_url
    elif provider == IdentityProvider.fence:
        base = config["OPENID_CONNECT"]["fence"]["api_base_url"]
        provider_logout = base + "/logout?" + urllib.parse.urlencode({"next": next_url})
    elif provider == "cognito":
        idp_openid_connect = config["OPENID_CONNECT"]["cognito"]
        well_known = None
        try:
            well_known_resp = get_openid_config_for_idp(idp_openid_connect)
            well_known = well_known_resp.json()
        except requests.exceptions.HTTPError as e:
            logger.error(
                f"Well-known endpoint returned an error status after multiple retries, Cognito Session not invalidated, Logging out of Gen3. Error: {e}"
            )
        except requests.exceptions.ConnectionError as e:
            logger.error(
                f"Could not connect to well-known endpoint, Cognito Session not invalidated, Logging out of Gen3. Error: {e} "
            )
        except JSONDecodeError as e:
            logger.error(
                f"Invalid JSON resonse from well-known, Cognito Session not invalidated, Logging out of Gen3. Error: {e}"
            )
        except Exception as e:
            logger.error(
                f"Error occured trying to get well-known, Cognito Session not invalidated, Logging out from Gen3. Error: {e}"
            )
        if well_known:
            end_session_endpoint = well_known.get("end_session_endpoint")
            # NOTE: discovery url for cognito is different than the cognito api domain url. Check the domain for the APIs like end_session_endpoint or authorization_endpoint found in the well-know openid config
            if domain(end_session_endpoint) not in allowed_login_redirects():
                logger.error(
                    f"Logout url {end_session_endpoint} not in LOGIN_REDIRECT_WHITELIST config. Cognito Session not invalidated, Logging out from Gen3."
                )
            else:
                if end_session_endpoint:
                    provider_logout = (
                        end_session_endpoint
                        + "?"
                        + urllib.parse.urlencode(
                            {
                                "client_id": idp_openid_connect["client_id"],
                                "logout_uri": next_url,  # NOTE: This needs to be set up in the cognito console for an allowed sign-out url
                            }
                        )
                    )
                else:
                    logger.error(
                        "end_session_endpoint not found in well-known config. Cognito Session not invalidated. Logging out from Gen3"
                    )

    flask.session.clear()
    try:
        redirect_response = flask.make_response(
            flask.redirect(provider_logout or urllib.parse.unquote(next_url))
        )
    except Exception as e:
        logger.error(f"Error logging out: {e}")
    clear_cookies(redirect_response)
    return redirect_response


def check_scope(scope):
    def wrapper(f):
        @wraps(f)
        def check_scope_and_call(*args, **kwargs):
            if "_all" in flask.g.scopes or scope in flask.g.scopes:
                return f(*args, **kwargs)
            else:
                raise Unauthorized(
                    "Requested scope {} can't access this endpoint".format(scope)
                )

        return check_scope_and_call

    return wrapper


def login_required(scope=None):
    """
    Create decorator to require a user session
    """

    def decorator(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            if flask.session.get("username"):
                is_logged_in = login_user_or_require_registration(
                    flask.session["username"], flask.session["provider"]
                )
                if not is_logged_in:
                    raise Unauthorized("Please register to login")
                return f(*args, **kwargs)

            eppn = None
            if config["LOGIN_OPTIONS"]:
                enable_shib = "shibboleth" in [
                    option["idp"] for option in config["LOGIN_OPTIONS"]
                ]
            else:
                # fall back on "providers"
                enable_shib = "shibboleth" in (
                    config.get("ENABLED_IDENTITY_PROVIDERS") or {}
                ).get("providers", {})

            if enable_shib and "SHIBBOLETH_HEADER" in config:
                eppn = flask.request.headers.get(config["SHIBBOLETH_HEADER"])

            if config.get("MOCK_AUTH") is True:
                eppn = "test"
            # if there is authorization header for oauth
            if "Authorization" in flask.request.headers:
                has_oauth(scope=scope)
                return f(*args, **kwargs)
            # if there is shibboleth session, then create user session and
            # log user in
            elif eppn:
                username = eppn.split("!")[-1]
                flask.session["username"] = username
                flask.session["provider"] = IdentityProvider.itrust
                is_logged_in = login_user_or_require_registration(
                    username, flask.session["provider"]
                )
                if not is_logged_in:
                    raise Unauthorized("Please register to login")
                return f(*args, **kwargs)
            else:
                raise Unauthorized("Please login")

        return wrapper

    return decorator


def has_oauth(scope=None):
    scope = scope or set()
    scope.update({"openid"})
    try:
        access_token_claims = validate_jwt(
            scope=scope,
            purpose="access",
        )
    except JWTError as e:
        raise Unauthorized("failed to validate token: {}".format(e))
    if "sub" in access_token_claims:
        user_id = access_token_claims["sub"]
        user = (
            current_app.scoped_session().query(User).filter_by(id=int(user_id)).first()
        )
        if not user:
            raise Unauthorized("no user found with id: {}".format(user_id))
        # set some application context for current user
        flask.g.user = user
    # set some application context for current client id
    # client_id should be None if the field doesn't exist or is empty
    flask.g.client_id = access_token_claims.get("azp") or None
    flask.g.token = access_token_claims


def get_user_from_claims(claims):
    return (
        current_app.scoped_session()
        .query(User)
        .filter(User.id == claims["sub"])
        .first()
    )


def admin_login_required(function):
    """Use the check_arborist_auth decorator checking on admin authorization."""
    return check_arborist_auth(["/services/fence/admin"], "*")(function)


def _update_users_email(user, email):
    """
    Update email if provided and doesn't match db entry.
    """
    if email and user.email != email:
        logger.info(
            f"Updating username {user.username}'s email from {user.email} to {email}"
        )
        user.email = email

        current_app.scoped_session().add(user)
        current_app.scoped_session().commit()


def _update_users_id_from_idp(user, id_from_idp):
    """
    Update id_from_idp if provided and doesn't match db entry.
    """
    if id_from_idp and user.id_from_idp != id_from_idp:
        logger.info(
            f"Updating username {user.username}'s id_from_idp from {user.id_from_idp} to {id_from_idp}"
        )
        user.id_from_idp = id_from_idp

        current_app.scoped_session().add(user)
        current_app.scoped_session().commit()


def _update_users_last_auth(user):
    """
    Update _last_auth.
    """
    logger.info(f"Updating username {user.username}'s _last_auth.")
    user._last_auth = datetime.now()

    current_app.scoped_session().add(user)
    current_app.scoped_session().commit()

1	import re	1✔
2	import urllib.request, urllib.parse, urllib.error	1✔
3	from datetime import datetime	1✔
4	from functools import wraps	1✔
5	from json import JSONDecodeError	1✔
6
7	import backoff	1✔
8	import flask	1✔
9	from flask import current_app	1✔
10	from authutils.errors import JWTError, JWTExpiredError	1✔
11	from authutils.token.validate import (	1✔
12	current_token,
13	require_auth_header as authutils_require_auth_header,
14	set_current_token,
15	validate_request,
16	)
17	from cdislogging import get_logger	1✔
18	import requests	1✔
19
20	from fence.authz.auth import check_arborist_auth	1✔
21	from fence.config import config	1✔
22	from fence.errors import Unauthorized, InternalError	1✔
23	from fence.jwt.validate import validate_jwt	1✔
24	from fence.models import User, IdentityProvider, query_for_user	1✔
25	from fence.user import get_current_user	1✔
26	from fence.utils import (	1✔
27	clear_cookies,
28	DEFAULT_BACKOFF_SETTINGS,
29	allowed_login_redirects,
30	domain,
31	)
32
33	logger = get_logger(__name__)	1✔
34
35
36	def require_auth_header(args, *kwargs):	1✔
37	"""
38	Injects the default token audience before calling authutils's `require_auth_header`
39	"""
40	if "audience" in kwargs:	1✔
41	if type(kwargs["audience"]) != list:	×
42	kwargs["audience"] = [kwargs["audience"]]	×
43	kwargs["audience"].append("gen3")	×
44	else:
45	kwargs["audience"] = "gen3"	1✔
46
47	return authutils_require_auth_header(args, *kwargs)	1✔
48
49
50	def get_jwt():	1✔
51	"""
52	Return the user's JWT from authorization header. Requires flask application context.
53	Raises:
54	- Unauthorized, if header is missing or not in the correct format
55	"""
56	header = flask.request.headers.get("Authorization")	1✔
57	if not header:	1✔
58	raise Unauthorized("missing authorization header")	1✔
59	try:	1✔
60	bearer, token = header.split(" ")	1✔
61	except ValueError:	×
62	msg = "authorization header not in expected format"	×
63	logger.debug(f"{msg}. Received header: {header}")	×
64	logger.error(f"{msg}.")	×
65	raise Unauthorized(msg)	×
66	if bearer.lower() != "bearer":	1✔
67	raise Unauthorized("expected bearer token in auth header")	×
68	return token	1✔
69
70
71	def build_redirect_url(hostname, path):	1✔
72	"""
73	Compute a redirect given a hostname and next path where
74	Args:
75	hostname (str): may be empty string or a bare hostname or
76	a hostname with a protocal attached (https?://...)
77	path (int): is a path to attach to hostname
78	Return:
79	string url suitable for flask.redirect
80	"""
81	redirect_base = hostname	1✔
82	# BASE_URL may be empty or a bare hostname or a hostname with a protocol
83	if bool(redirect_base) and not redirect_base.startswith("http"):	1✔
84	redirect_base = "https://" + redirect_base	1✔
85	return redirect_base + path	1✔
86
87
88	def get_ip_information_string():	1✔
89	"""
90	Returns a string containing the client's IP address and any X-Forwarded headers.
91
92	Returns:
93	str: A formatted string containing the client's IP address and X-Forwarded headers.
94	"""
95	x_forwarded_headers = [	1✔
96	f"{header}: {value}"
97	for header, value in flask.request.headers
98	if "X-Forwarded" in header
99	]
100	return f"flask.request.remote_addr={flask.request.remote_addr} x_forwarded_headers={x_forwarded_headers}"	1✔
101
102
103	def _identify_user_and_update_database(	1✔
104	user,
105	username,
106	provider,
107	email=None,
108	id_from_idp=None,
109	username_deny_regex=None,
110	) -> bool:
111	"""
112	Create a new user if one doesn't already exist in the database. Commit the user
113	and associated idp information to the database.
114
115	Args:
116	user (User): user to be logged in, if it already exists, None otherwise
117	username (str): specific username of user to be logged in
118	provider (str): specfic idp of user to be logged in
119	email (str, optional): email of user (may or may not match username depending
120	on the IdP)
121	id_from_idp (str, optional): id from the IDP (which may be different than
122	the username)
123
124	Return:
125	User: the created or updated user
126	"""
127	username_deny_regex = username_deny_regex or config["GLOBAL_USERNAME_DENY_REGEX"]	1✔
128	if username_deny_regex:	1✔
129	if re.search(pattern=username_deny_regex, string=username):	1✔
130	logger.info(	1✔
131	f"Blocked login of user with username {username} due to deny regex: {username_deny_regex}"
132	)
133
134	# intentionally empty message to prevent information leakage
135	raise Unauthorized(message="")	1✔
136
137	if user:	1✔
138	if user.active == False:	1✔
139	# Abort login if user.active == False:
140	raise Unauthorized(	1✔
141	"User is known but not authorized/activated in the system"
142	)
143	_update_users_email(user, email)	1✔
144	_update_users_id_from_idp(user, id_from_idp)	1✔
145	_update_users_last_auth(user)	1✔
146	else:
147	if not config["ALLOW_NEW_USER_ON_LOGIN"]:	1✔
148	# do not create new active users automatically
149	raise Unauthorized("New user is not yet authorized/activated in the system")	1✔
150
151	# add the new user
152	user = User(username=username)	1✔
153
154	if email:	1✔
155	user.email = email	1✔
156
157	if id_from_idp:	1✔
158	user.id_from_idp = id_from_idp	1✔
159	# TODO: update iss_sub mapping table?
160
161	# This expression is relevant to those users who already have user and
162	# idp info persisted to the database. We avoid unnecessarily re-saving
163	# that user and idp info.
164	if not user.identity_provider or not user.identity_provider.name == provider:	1✔
165	# setup idp connection for new user (or existing user w/o it setup)
166	idp = (	1✔
167	current_app.scoped_session()
168	.query(IdentityProvider)
169	.filter(IdentityProvider.name == provider)
170	.first()
171	)
172	if not idp:	1✔
173	idp = IdentityProvider(name=provider)	1✔
174
175	user.identity_provider = idp	1✔
176	current_app.scoped_session().add(user)	1✔
177	current_app.scoped_session().commit()	1✔
178
179	# `login_in_progress_username` stored for use by the user registration code.
180	# not using `flask.session["username"]` because other code relies on it to know
181	# whether a user is logged in; in this case the user isn't logged in yet.
182	flask.session["login_in_progress_username"] = user.username	1✔
183
184	flask.g.user = user	1✔
185	return user	1✔
186
187
188	def _is_user_registration_required_before_login(user, provider) -> bool:	1✔
189	auto_registration_enabled = (	1✔
190	config["OPENID_CONNECT"]
191	.get(provider, {})
192	.get("enable_idp_users_registration", False)
193	)
194	# Registration is required if:
195	# - Registration is enabled in the config, AND
196	# - Automatic registration is NOT enabled, AND
197	# - The user's registration info is empty
198	return (	1✔
199	config["REGISTER_USERS_ON"]
200	and not auto_registration_enabled
201	and user.additional_info.get("registration_info", {}) == {}
202	)
203
204
205	def login_user_or_require_registration(	1✔
206	username, provider, upstream_idp=None, shib_idp=None, email=None, id_from_idp=None
207	) -> bool:
208	"""
209	Check if a user needs to go through the registration flow before being logged in. If not,
210	login the user with the given username and provider. Set values in Flask session to indicate
211	the user being logged in.
212
213	Args:
214	username (str): specific username of user to be logged in
215	provider (str): specfic idp of user to be logged in
216	upstream_idp (str, optional): upstream fence IdP
217	shib_idp (str, optional): upstream shibboleth IdP
218	email (str, optional): email of user (may or may not match username depending
219	on the IdP)
220	id_from_idp (str, optional): id from the IDP (which may be different than
221	the username)
222
223	Return:
224	bool: whether the user has been logged in (if registration is enabled and the user is not
225	registered, this would be False)
226	"""
227
228	def log_ip(user):	1✔
229	ip_info = get_ip_information_string()	1✔
230	logger.info(	1✔
231	f"User logged in. user.id={user.id} user.username={user.username} {ip_info}"
232	)
233
234	def set_flask_session_values(user):	1✔
235	"""
236	Helper fuction to set user values in the session.
237
238	Args:
239	user (User): User object
240	"""
241	flask.session["username"] = user.username	1✔
242	flask.session["user_id"] = str(user.id)	1✔
243	flask.session["provider"] = user.identity_provider.name	1✔
244	if upstream_idp:	1✔
245	flask.session["upstream_idp"] = upstream_idp	×
246	if shib_idp:	1✔
247	flask.session["shib_idp"] = shib_idp	×
248	flask.g.user = user	1✔
249	flask.g.scopes = ["_all"]	1✔
250	flask.g.token = None	1✔
251
252	user = query_for_user(session=current_app.scoped_session(), username=username)	1✔
253	user = _identify_user_and_update_database(	1✔
254	user, username, provider, email, id_from_idp
255	)
256	log_user_in = not _is_user_registration_required_before_login(user, provider)	1✔
257	if log_user_in:	1✔
258	set_flask_session_values(user)	1✔
259	log_ip(user)	1✔
260	return log_user_in	1✔
261
262
263	@backoff.on_exception(backoff.expo, Exception, **DEFAULT_BACKOFF_SETTINGS)	1✔
264	def get_openid_config_for_idp(open_id_connect):	1✔
265	"""
266	Return openid configuration for a given provider.
267	Args:
268	open_id_connect (dict): fence config for idp
269	Returns:
270	response: response of openid configuration
271	"""
272	well_known_url = open_id_connect["discovery_url"]	1✔
273	well_known_resp = requests.get(well_known_url)	1✔
274	well_known_resp.raise_for_status()	1✔
275	return well_known_resp	1✔
276
277
278	def logout(next_url, force_era_global_logout=False):	1✔
279	"""
280	Return a redirect which another logout from IDP or the provided redirect.
281	Depending on the IDP, this logout will propogate. For example, if using
282	another fence as an IDP, this will hit that fence's logout endpoint.
283	Args:
284	next_url (str): Final redirect desired after logout
285	"""
286	logger.debug("IN AUTH LOGOUT, next_url = {0}".format(next_url))	1✔
287
288	# propogate logout to IDP
289	provider_logout = None	1✔
290	provider = flask.session.get("provider")	1✔
291
292	if force_era_global_logout or provider == IdentityProvider.itrust:	1✔
293	safe_url = urllib.parse.quote_plus(next_url)	1✔
294	provider_logout = config["ITRUST_GLOBAL_LOGOUT"] + safe_url	1✔
295	elif provider == IdentityProvider.fence:	1✔
296	base = config["OPENID_CONNECT"]["fence"]["api_base_url"]	1✔
297	provider_logout = base + "/logout?" + urllib.parse.urlencode({"next": next_url})	1✔
298	elif provider == "cognito":	1✔
299	idp_openid_connect = config["OPENID_CONNECT"]["cognito"]	1✔
300	well_known = None	1✔
301	try:	1✔
302	well_known_resp = get_openid_config_for_idp(idp_openid_connect)	1✔
303	well_known = well_known_resp.json()	1✔
304	except requests.exceptions.HTTPError as e:	1✔
305	logger.error(	1✔
306	f"Well-known endpoint returned an error status after multiple retries, Cognito Session not invalidated, Logging out of Gen3. Error: {e}"
307	)
308	except requests.exceptions.ConnectionError as e:	1✔
309	logger.error(	1✔
310	f"Could not connect to well-known endpoint, Cognito Session not invalidated, Logging out of Gen3. Error: {e} "
311	)
312	except JSONDecodeError as e:	×
313	logger.error(	×
314	f"Invalid JSON resonse from well-known, Cognito Session not invalidated, Logging out of Gen3. Error: {e}"
315	)
316	except Exception as e:	×
317	logger.error(	×
318	f"Error occured trying to get well-known, Cognito Session not invalidated, Logging out from Gen3. Error: {e}"
319	)
320	if well_known:	1✔
321	end_session_endpoint = well_known.get("end_session_endpoint")	1✔
322	# NOTE: discovery url for cognito is different than the cognito api domain url. Check the domain for the APIs like end_session_endpoint or authorization_endpoint found in the well-know openid config
323	if domain(end_session_endpoint) not in allowed_login_redirects():	1✔
324	logger.error(	1✔
325	f"Logout url {end_session_endpoint} not in LOGIN_REDIRECT_WHITELIST config. Cognito Session not invalidated, Logging out from Gen3."
326	)
327	else:
328	if end_session_endpoint:	1✔
329	provider_logout = (	1✔
330	end_session_endpoint
331	+ "?"
332	+ urllib.parse.urlencode(
333	{
334	"client_id": idp_openid_connect["client_id"],
335	"logout_uri": next_url, # NOTE: This needs to be set up in the cognito console for an allowed sign-out url
336	}
337	)
338	)
339	else:
340	logger.error(	×
341	"end_session_endpoint not found in well-known config. Cognito Session not invalidated. Logging out from Gen3"
342	)
343
344	flask.session.clear()	1✔
345	try:	1✔
346	redirect_response = flask.make_response(	1✔
347	flask.redirect(provider_logout or urllib.parse.unquote(next_url))
348	)
349	except Exception as e:	×
350	logger.error(f"Error logging out: {e}")	×
351	clear_cookies(redirect_response)	1✔
352	return redirect_response	1✔
353
354
355	def check_scope(scope):	1✔
356	def wrapper(f):	×
357	@wraps(f)	×
358	def check_scope_and_call(args, *kwargs):	×
359	if "_all" in flask.g.scopes or scope in flask.g.scopes:	×
360	return f(args, *kwargs)	×
361	else:
362	raise Unauthorized(	×
363	"Requested scope {} can't access this endpoint".format(scope)
364	)
365
366	return check_scope_and_call	×
367
368	return wrapper	×
369
370
371	def login_required(scope=None):	1✔
372	"""
373	Create decorator to require a user session
374	"""
375
376	def decorator(f):	1✔
377	@wraps(f)	1✔
378	def wrapper(args, *kwargs):	1✔
379	if flask.session.get("username"):	1✔
380	is_logged_in = login_user_or_require_registration(	1✔
381	flask.session["username"], flask.session["provider"]
382	)
383	if not is_logged_in:	1✔
384	raise Unauthorized("Please register to login")	×
385	return f(args, *kwargs)	1✔
386
387	eppn = None	1✔
388	if config["LOGIN_OPTIONS"]:	1✔
389	enable_shib = "shibboleth" in [	1✔
390	option["idp"] for option in config["LOGIN_OPTIONS"]
391	]
392	else:
393	# fall back on "providers"
394	enable_shib = "shibboleth" in (	×
395	config.get("ENABLED_IDENTITY_PROVIDERS") or {}
396	).get("providers", {})
397
398	if enable_shib and "SHIBBOLETH_HEADER" in config:	1✔
399	eppn = flask.request.headers.get(config["SHIBBOLETH_HEADER"])	1✔
400
401	if config.get("MOCK_AUTH") is True:	1✔
402	eppn = "test"	1✔
403	# if there is authorization header for oauth
404	if "Authorization" in flask.request.headers:	1✔
405	has_oauth(scope=scope)	1✔
406	return f(args, *kwargs)	1✔
407	# if there is shibboleth session, then create user session and
408	# log user in
409	elif eppn:	1✔
410	username = eppn.split("!")[-1]	1✔
411	flask.session["username"] = username	1✔
412	flask.session["provider"] = IdentityProvider.itrust	1✔
413	is_logged_in = login_user_or_require_registration(	1✔
414	username, flask.session["provider"]
415	)
416	if not is_logged_in:	1✔
417	raise Unauthorized("Please register to login")	×
418	return f(args, *kwargs)	1✔
419	else:
420	raise Unauthorized("Please login")	1✔
421
422	return wrapper	1✔
423
424	return decorator	1✔
425
426
427	def has_oauth(scope=None):	1✔
428	scope = scope or set()	1✔
429	scope.update({"openid"})	1✔
430	try:	1✔
431	access_token_claims = validate_jwt(	1✔
432	scope=scope,
433	purpose="access",
434	)
435	except JWTError as e:	1✔
436	raise Unauthorized("failed to validate token: {}".format(e))	×
437	if "sub" in access_token_claims:	1✔
438	user_id = access_token_claims["sub"]	1✔
439	user = (	1✔
440	current_app.scoped_session().query(User).filter_by(id=int(user_id)).first()
441	)
442	if not user:	1✔
443	raise Unauthorized("no user found with id: {}".format(user_id))	×
444	# set some application context for current user
445	flask.g.user = user	1✔
446	# set some application context for current client id
447	# client_id should be None if the field doesn't exist or is empty
448	flask.g.client_id = access_token_claims.get("azp") or None	1✔
449	flask.g.token = access_token_claims	1✔
450
451
452	def get_user_from_claims(claims):	1✔
453	return (	1✔
454	current_app.scoped_session()
455	.query(User)
456	.filter(User.id == claims["sub"])
457	.first()
458	)
459
460
461	def admin_login_required(function):	1✔
462	"""Use the check_arborist_auth decorator checking on admin authorization."""
463	return check_arborist_auth(["/services/fence/admin"], "*")(function)	1✔
464
465
466	def _update_users_email(user, email):	1✔
467	"""
468	Update email if provided and doesn't match db entry.
469	"""
470	if email and user.email != email:	1✔
471	logger.info(	1✔
472	f"Updating username {user.username}'s email from {user.email} to {email}"
473	)
474	user.email = email	1✔
475
476	current_app.scoped_session().add(user)	1✔
477	current_app.scoped_session().commit()	1✔
478
479
480	def _update_users_id_from_idp(user, id_from_idp):	1✔
481	"""
482	Update id_from_idp if provided and doesn't match db entry.
483	"""
484	if id_from_idp and user.id_from_idp != id_from_idp:	1✔
485	logger.info(	1✔
486	f"Updating username {user.username}'s id_from_idp from {user.id_from_idp} to {id_from_idp}"
487	)
488	user.id_from_idp = id_from_idp	1✔
489
490	current_app.scoped_session().add(user)	1✔
491	current_app.scoped_session().commit()	1✔
492
493
494	def _update_users_last_auth(user):	1✔
495	"""
496	Update _last_auth.
497	"""
498	logger.info(f"Updating username {user.username}'s _last_auth.")	1✔
499	user._last_auth = datetime.now()	1✔
500
501	current_app.scoped_session().add(user)	1✔
502	current_app.scoped_session().commit()	1✔

uc-cdis / fence / 28570052741

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous