import flask import jwt from werkzeug.exceptions import BadRequest from atlas_um import consts from atlas_um.analytics import user_events from atlas_um.audit import user_audit from atlas_um.consts import SystemEvents from atlas_um.extensions import usm from atlas_um.helpers.domains import get_allowed_root_domain from atlas_um.helpers.domains import get_root_domain from atlas_um.helpers.tokens import get_safe_repr from atlas_um.logs import logger from atlas_um.helpers.auth_proxy import get_auth_proxy from atlas_um.tokens.services import InvalidateRefreshToken logout = flask.Blueprint("usm_logout", __name__) @logout.route("", methods=["GET"]) def get_logout(): # assigning here, to reuse later for both callback and cleanup cases flask.session[consts.REFERRER_SESSION_NAME] = flask.request.referrer dna_bearer_token = flask.request.cookies.get( flask.current_app.config.get("DNA_BEARER_TOKEN_COOKIE_NAME"), None ) if dna_bearer_token is None: logger.warning("Logout without bearer token") return process_logout() usm_id_token = flask.request.cookies.get( consts.USM_ID_TOKEN_COOKIE_NAME, None ) if usm_id_token is None: logger.warning("Logout without USM ID token") return process_logout(dna_bearer_token) state = usm.generate_state() usm_url = usm.logout_url(usm_id_token, state) flask.session[consts.STATE_SESSION_NAME] = state flask.session[consts.AUTH_PROXY_SESSION_NAME] = get_auth_proxy() logger.info("Redirecting to USM OIDC logout flow") return flask.redirect(usm_url) @logout.route("/callback", methods=["GET"]) def get_logout_callback(): # standard OIDC checks state_param = flask.request.args.get("state", None) if state_param is None: logger.warning("No state in query on logout callback") raise BadRequest("No state in query") session_state = flask.session.get(consts.STATE_SESSION_NAME, None) if session_state is None: logger.warning( "State param is missing from session on logout callback" ) raise BadRequest(f"{consts.STATE_SESSION_NAME} missing from session") if session_state != state_param: logger.warning( "No match for state in session/query on logout callback" ) raise BadRequest( f"{consts.STATE_SESSION_NAME} " f"in session does not match state in query" ) dna_refresh_token = flask.request.cookies.get( flask.current_app.config.get("DNA_REFRESH_TOKEN_COOKIE_NAME"), None ) if dna_refresh_token: logger.bind(refresh_token=get_safe_repr(dna_refresh_token)).info( "Invalidating refresh token" ) InvalidateRefreshToken.execute(dna_refresh_token) dna_bearer_token = flask.request.cookies.get( flask.current_app.config.get("DNA_BEARER_TOKEN_COOKIE_NAME"), None ) if dna_bearer_token is None: logger.warning("Logout callback without bearer token") return process_logout() return process_logout(dna_bearer_token) def process_logout(dna_bearer_token=""): # nosec user_id = "" if dna_bearer_token: try: user_id = jwt.decode( dna_bearer_token, options={"verify_signature": False} ).get("sub") except Exception: logger.warning("Logout with corrupted bearer token") auth_proxy = flask.session.get(consts.AUTH_PROXY_SESSION_NAME) if auth_proxy: back_to = auth_proxy resp = flask.make_response( flask.render_template( "auth_proxy.html", auth_proxy=auth_proxy, ) ) else: back_to = flask.url_for("common.userinfo") resp = flask.make_response(flask.redirect(back_to)) domain = get_allowed_root_domain() product_domain = domain if auth_proxy: product_domain = get_root_domain(auth_proxy) or domain domain_cookie = f".{domain}" if domain else None resp.set_cookie( flask.current_app.config.get("DNA_BEARER_TOKEN_COOKIE_NAME"), "", expires=0, httponly=True, domain=domain_cookie, secure=True, ) resp.set_cookie( flask.current_app.config.get("DNA_REFRESH_TOKEN_COOKIE_NAME"), "", expires=0, httponly=True, domain=domain_cookie, secure=True, ) resp.set_cookie( consts.USM_ID_TOKEN_COOKIE_NAME, "", expires=0, httponly=True, domain=domain_cookie, secure=True, ) if user_id: user_audit.log( SystemEvents.successful_logout, domain=product_domain, user_id=user_id, token=get_safe_repr(dna_bearer_token), ) user_events.log( SystemEvents.successful_logout, domain=product_domain, user_id=user_id, ) logger.bind( user_id=user_id, back_to=back_to, auth_proxy=auth_proxy, domain=domain, product_domain=product_domain, domain_cookie=domain_cookie, ).info("Processing user logout") return resp