def idp_initiated()

in redash/authentication/saml_auth.py [0:0]

• 38 lines of code
• 10 McCabe index (conditional complexity)

def idp_initiated(org_slug=None):
    if not current_org.get_setting("auth_saml_enabled"):
        logger.error("SAML Login is not enabled")
        return redirect(url_for("redash.index", org_slug=org_slug))

    index_url = url_for("redash.index", org_slug=org_slug)
    unsafe_next_path = request.args.get("next", index_url)
    next_path = get_next_path(unsafe_next_path)

    saml_client = get_saml_client(current_org, next_url=next_path)
    saml_client_urls_upgrade(saml_client)

    try:
        authn_response = saml_client.parse_authn_request_response(
            request.form["SAMLResponse"], entity.BINDING_HTTP_POST
        )
    except Exception:
        logger.error("Failed to parse SAML response", exc_info=True)
        flash("SAML login failed. Please try again later.")
        return redirect(url_for("redash.login", org_slug=org_slug))

    authn_response.get_identity()
    user_info = authn_response.get_subject()
    email = user_info.text

    try:
        name = "%s %s" % (authn_response.ava['firstName'][0], authn_response.ava['lastName'][0])
    except Exception:
        name = email.split('@')[0]

    attributes = {}
    if authn_response.ava:
        for k, v in authn_response.ava.items():
            if len(v) == 1:
                attributes[k] = v[0]
            else:
                attributes[k] = v

        # name = "%s %s" % (
    #     authn_response.ava["FirstName"][0],
    #     authn_response.ava["LastName"][0],
    # )

    # This is what as known as "Just In Time (JIT) provisioning".
    # What that means is that, if a user in a SAML assertion
    # isn't in the user store, we create that user first, then log them in
    user = create_and_login_user(current_org, name, email, attributes=attributes)
    if user is None:
        return logout_and_redirect_to_index()

    if "RedashGroups" in authn_response.ava:
        group_names = authn_response.ava.get("RedashGroups")
        user.update_group_assignments(group_names)

    # url = url_for("redash.index", org_slug=org_slug)

    return redirect(next_path)