views.py

"""
flask_security.views
~~~~~~~~~~~~~~~~~~~~

Flask-Security views module

:copyright: (c) 2012 by Matt Wright.
:copyright: (c) 2019-2026 by J. Christopher Wagner (jwag).
:license: MIT, see LICENSE for more details.

CSRF is tricky. By default all our forms have CSRF protection built in via
Flask-WTF. This is regardless of authentication method or whether the request
is Form or JSON based. Form-based 'just works' since when rendering the form
(on GET), the CSRF token is automatically populated.
We want to handle:
    - JSON requests where CSRF token is in a header (e.g. X-CSRF-Token)
    - Option to skip CSRF when using a token to authenticate (rather than session)
      (CSRF_PROTECT_MECHANISMS)
    - Option to skip CSRF for 'login'/unauthenticated requests
      (CSRF_IGNORE_UNAUTH_ENDPOINTS)
This is complicated by the fact that the only way to disable form CSRF is to
pass in meta={csrf: false} at form instantiation time.

Be aware that for CSRF to work, caller MUST pass in session cookie. So
for pure API, and no session cookie - there is no way to support CSRF-Login
so app must set CSRF_IGNORE_UNAUTH_ENDPOINTS (or use CSRF/session cookie for logging
in then once they have a token, no need for cookie).

"""

from __future__ import annotations
from functools import partial
import time
import typing as t

from flask import (
    Blueprint,
    after_this_request,
    current_app,
    jsonify,
    request,
    session,
)
from flask_login import current_user

from .changeable import change_user_password
from .change_email import change_email, change_email_confirm
from .change_username import change_username
from .confirmable import (
    confirm_email_token_status,
    confirm_user,
    send_confirmation_instructions,
)
from .decorators import anonymous_user_required, auth_required, unauth_csrf
from .forms import (
    _setup_methods_xlate,
    ChangePasswordForm,
    DummyForm,
    ForgotPasswordForm,
    LoginForm,
    LogoutForm,
    build_form_from_request,
    build_form,
    form_errors_munge,
    ResetPasswordForm,
    SendConfirmationForm,
    TwoFactorVerifyCodeForm,
    TwoFactorSetupForm,
    TwoFactorRescueForm,
    UsernameRecoveryForm,
    VerifyForm,
)
from .passwordless import login_token_status, send_login_instructions
from .proxies import _security, _datastore
from .quart_compat import get_quart_status
from .signals import tf_profile_changed
from .tokens import refresh
from .unified_signin import (
    us_signin,
    us_signin_send_code,
    us_setup,
    us_setup_validate,
    us_verify,
    us_verify_link,
    us_verify_send_code,
)
from .recoverable import (
    reset_password_token_status,
    send_reset_password_instructions,
    update_password,
    send_username_recovery_email,
)
from .registerable import register_user, register_existing
from .recovery_codes import mf_recovery, mf_recovery_codes
from .tf_plugin import (
    tf_check_state,
    tf_illegal_state,
    tf_set_validity_token_cookie,
)
from .twofactor import (
    complete_two_factor_process,
    set_rescue_options,
    tf_clean_session,
    tf_disable,
)
from .utils import (
    base_render_json,
    check_and_update_authn_fresh,
    check_and_get_token_status,
    config_value as cv,
    confirm_redirect,
    do_flash,
    get_identity_attributes,
    get_message,
    get_post_login_redirect,
    get_post_logout_redirect,
    get_post_register_redirect,
    get_post_verify_redirect,
    get_request_attr,
    get_within_delta,
    get_url,
    handle_already_auth,
    hash_password,
    is_user_authenticated,
    localize_callback,
    login_user,
    logout_user,
    propagate_next,
    send_mail,
    slash_url_suffix,
    url_for_security,
    view_commit,
    allowed_auth_token,
    set_request_attr,
)
from .webauthn import (
    has_webauthn,
    webauthn_delete,
    webauthn_register,
    webauthn_register_response,
    webauthn_signin,
    webauthn_signin_response,
    webauthn_verify,
    webauthn_verify_response,
)

if get_quart_status():  # pragma: no cover
    from quart import make_response, redirect
else:
    from flask import make_response, redirect

if t.TYPE_CHECKING:  # pragma: no cover
    from flask.typing import ResponseValue


def default_render_json(payload, code, headers, user):
    """Default JSON response handler."""
    # Force Content-Type header to json.
    if headers is None:
        headers = dict()
    headers["Content-Type"] = "application/json"
    payload = dict(meta=dict(code=code), response=payload)
    return make_response(jsonify(payload), code, headers)


def _ctx(endpoint):
    return _security._run_ctx_processor(endpoint)


@unauth_csrf()
def login() -> ResponseValue:
    """View function for login view"""
    form = t.cast(LoginForm, build_form_from_request("login_form"))

    if is_user_authenticated(current_user):
        return handle_already_auth(
            form, payload={"identity_attributes": get_identity_attributes()}
        )

    # Clean out any potential old session info - in case of previous
    # aborted 2FA attempt.
    tf_clean_session()

    if form.validate_on_submit():
        assert form.user is not None
        remember_me = form.remember.data if "remember" in form else None
        response = _security.two_factor_plugins.tf_enter(
            form.user,
            remember_me,
            "password",
            next_loc=propagate_next(request.url, form),
        )
        if response:
            return response
        # two factor not required - login user
        after_this_request(view_commit)
        login_user(form.user, remember=remember_me, authn_via=["password"])

        if _security._want_json(request):
            return base_render_json(
                form,
                include_auth_token=allowed_auth_token(form.user),
                additional=dict(tf_required=False),
            )
        return redirect(get_post_login_redirect())

    if request.method == "POST" and cv("RETURN_GENERIC_RESPONSES"):
        # Validation failed - make sure PII error messages are generic
        fields_to_squash = dict(
            email=dict(replace_msg="GENERIC_AUTHN_FAILED"),
            password=dict(replace_msg="GENERIC_AUTHN_FAILED"),
        )
        if hasattr(form, "username"):
            fields_to_squash["username"] = dict(replace_msg="GENERIC_AUTHN_FAILED")
        form_errors_munge(form, fields_to_squash)

    if request.method == "GET":
        # set CSRF COOKIE if configured. This is the equivalent of forms and
        # base_render_json always sending the csrf_token
        session["fs_cc"] = "set"

    if _security._want_json(request):
        payload = {
            "identity_attributes": get_identity_attributes(),
        }
        return base_render_json(form, additional=payload)

    if rurl := confirm_redirect(form, "email"):
        return rurl

    return _security.render_template(
        cv("LOGIN_USER_TEMPLATE"),
        login_user_form=form,
        identity_attributes=get_identity_attributes(),
        **_ctx("login"),
    )


@auth_required(lambda: cv("API_ENABLED_METHODS"))
def verify():
    """View function which handles a reauthentication request."""
    form = t.cast(VerifyForm, build_form_from_request("verify_form", user=current_user))

    if form.validate_on_submit():
        # form may have called verify_and_update_password()
        after_this_request(view_commit)

        # verified - so set freshness time.
        session["fs_paa"] = time.time()

        if _security._want_json(request):
            return base_render_json(
                form, include_auth_token=allowed_auth_token(form.user)
            )
        do_flash(*get_message("REAUTHENTICATION_SUCCESSFUL"))
        return redirect(get_post_verify_redirect())

    webauthn_available = has_webauthn(current_user, cv("WAN_ALLOW_AS_VERIFY"))
    if _security._want_json(request):
        payload = {
            "has_webauthn_verify_credential": webauthn_available,
            "oauth_enabled": cv("OAUTH_ENABLE"),
            "oauth_providers": (
                _security.oauthglue.provider_names if cv("OAUTH_ENABLE") else []
            ),
        }
        return base_render_json(form, additional=payload)

    return _security.render_template(
        cv("VERIFY_TEMPLATE"),
        verify_form=form,
        has_webauthn_verify_credential=webauthn_available,
        wan_verify_form=build_form("wan_verify_form"),
        **_ctx("verify"),
    )


def logout():
    """View function which handles a logout request.
    logout has never been CSRF protected.
    As part of the refresh_token feature, logout now has a form defined
    which a client can pass a refresh token that will be revoked as part of logout
    (if the refresh token is managed with a cookie, the value will be set into the
    form).
    The cookie AND refresh tracker/token will be revoked.
    """
    tf_clean_session()

    if is_user_authenticated(current_user):
        # Until we implement logout CSRF - shouldn't logout but NOT revoke refresh token
        set_request_attr("csrf_valid", True)
        form = t.cast(
            LogoutForm, build_form_from_request("logout_form", user=current_user)
        )

        if form.validate_on_submit():
            # if they passed a refresh token - revoke it
            from .tokens import _revoke_refresh_tracker

            if form.refresh_tracker and not form.refresh_tracker.revoked_at:
                _revoke_refresh_tracker(
                    form.refresh_tracker, form.refresh_errors, current_user
                )
        logout_user()

    # No body is required - so if a POST and json - return OK
    if request.method == "POST" and _security._want_json(request):
        return _security._render_json({}, 200, None, None)

    return redirect(get_post_logout_redirect())


@anonymous_user_required
@unauth_csrf()
def register() -> ResponseValue:
    """View function which handles a registration request."""

    if (_security.confirmable or request.is_json) and _security.forms[
        "confirm_register_form"
    ].cls:
        form_name = "confirm_register_form"
    else:
        form_name = "register_form"
    form = build_form_from_request(form_name)

    if form.validate_on_submit():
        after_this_request(view_commit)
        user = register_user(form)
        form.user = user

        # The 'auto-login' feature probably should be removed - I can't imagine
        # an application that would want random email accounts. It has been like this
        # since the beginning. Note that we still enforce 2FA - however for unified
        # signin - we adhere to historic behavior.
        if not _security.confirmable or cv("LOGIN_WITHOUT_CONFIRMATION"):
            response = _security.two_factor_plugins.tf_enter(
                form.user, False, "register", next_loc=propagate_next(request.url, form)
            )
            if response:
                return response
            # two factor not required - login user.
            login_user(user, authn_via=["register"])
            if _security._want_json(request):
                return base_render_json(
                    form,
                    include_auth_token=allowed_auth_token(form.user),
                    additional=dict(tf_required=False),
                )

        if not _security._want_json(request):
            return redirect(get_post_register_redirect())

        return base_render_json(form)

    # Here on GET or failed validate
    if request.method == "POST" and cv("RETURN_GENERIC_RESPONSES"):
        gr = register_existing(form)
        if gr:
            if _security._want_json(request):
                return base_render_json(form)

            return redirect(get_post_register_redirect())

    if _security._want_json(request):
        return base_render_json(form)

    return _security.render_template(
        cv("REGISTER_USER_TEMPLATE"),
        register_user_form=form,
        **_ctx("register"),
    )


@unauth_csrf()
def send_login():
    """View function that sends login instructions for passwordless login"""
    form = build_form_from_request("passwordless_login_form")

    if form.validate_on_submit():
        send_login_instructions(form.user)
        if not _security._want_json(request):
            do_flash(*get_message("LOGIN_EMAIL_SENT", email=form.user.email))

    if _security._want_json(request):
        return base_render_json(form)

    return _security.render_template(
        cv("SEND_LOGIN_TEMPLATE"), send_login_form=form, **_ctx("send_login")
    )


@anonymous_user_required
def token_login(token):
    """View function that handles passwordless login via a token
    Like reset-password and confirm - this is usually a GET via an email
    so from the request we can't differentiate form-based apps from non.
    """

    expired, invalid, user = login_token_status(token)

    if not user or invalid:
        m, c = get_message("INVALID_LOGIN_TOKEN")
        if cv("REDIRECT_BEHAVIOR") == "spa":
            return redirect(get_url(cv("LOGIN_ERROR_VIEW"), qparams={c: m}))
        do_flash(m, c)
        return redirect(url_for_security("login"))
    if expired:
        send_login_instructions(user)
        m, c = get_message("LOGIN_EXPIRED", email=user.email, within=cv("LOGIN_WITHIN"))
        if cv("REDIRECT_BEHAVIOR") == "spa":
            return redirect(
                get_url(
                    cv("LOGIN_ERROR_VIEW"),
                    qparams=user.get_redirect_qparams({c: m}),
                )
            )
        do_flash(m, c)
        return redirect(url_for_security("login"))

    login_user(user, authn_via=["token"])
    after_this_request(view_commit)
    if cv("REDIRECT_BEHAVIOR") == "spa":
        return redirect(
            get_url(cv("POST_LOGIN_VIEW"), qparams=user.get_redirect_qparams())
        )

    do_flash(*get_message("PASSWORDLESS_LOGIN_SUCCESSFUL"))

    return redirect(get_post_login_redirect())


@unauth_csrf()
def send_confirmation():
    """View function which sends confirmation instructions (/confirm)."""
    form = t.cast(
        SendConfirmationForm, build_form_from_request("send_confirmation_form")
    )

    if form.validate_on_submit():
        send_confirmation_instructions(form.user)
        if not _security._want_json(request):
            do_flash(*get_message("CONFIRMATION_REQUEST", email=form.email.data))

    elif request.method == "POST" and cv("RETURN_GENERIC_RESPONSES"):
        # Here on GET or failed validate
        rinfo = dict(email=dict())
        form_errors_munge(form, rinfo)  # by suppressing errors JSON should return 200
        # Check for other errors - for default form - there aren't additional fields
        # but applications might add some (e.g. recaptcha)
        if not form.errors:
            # Make look exactly like successful (e.g. real user) request
            if not _security._want_json(request):
                do_flash(*get_message("CONFIRMATION_REQUEST", email=form.email.data))

    if _security._want_json(request):
        # Never include user info since this is an anonymous endpoint.
        return base_render_json(form, include_user=False)

    return _security.render_template(
        cv("SEND_CONFIRMATION_TEMPLATE"),
        send_confirmation_form=form,
        **_ctx("send_confirmation"),
    )


def confirm_email(token):
    """
    View function which handles an email confirmation request.
    This is always a GET from an email - so for 'spa' must always redirect.
    """

    expired, invalid, user = confirm_email_token_status(token)

    if not user or invalid or expired:
        if expired:
            m, c = get_message(
                "CONFIRMATION_EXPIRED",
                within=cv("CONFIRM_EMAIL_WITHIN"),
            )
        else:
            m, c = get_message("INVALID_CONFIRMATION_TOKEN")
        if cv("REDIRECT_BEHAVIOR") == "spa":
            return redirect(get_url(cv("CONFIRM_ERROR_VIEW"), qparams={c: m}))
        do_flash(m, c)
        return redirect(
            get_url(cv("CONFIRM_ERROR_VIEW")) or url_for_security("send_confirmation")
        )

    already_confirmed = user.confirmed_at is not None
    if already_confirmed:
        m, c = get_message("ALREADY_CONFIRMED")

        if cv("REDIRECT_BEHAVIOR") == "spa":
            # No reason to expose identity info to anyone who has the link
            return redirect(
                get_url(
                    cv("CONFIRM_ERROR_VIEW"),
                    qparams={c: m},
                )
            )

        do_flash(m, c)
        return redirect(
            get_url(cv("CONFIRM_ERROR_VIEW")) or url_for_security("send_confirmation")
        )

    confirm_user(user)
    after_this_request(view_commit)
    m, c = get_message("EMAIL_CONFIRMED")

    # ? The only case where user is logged in already would be if
    # LOGIN_WITHOUT_CONFIRMATION
    if user != current_user:
        logout_user()
        if cv("AUTO_LOGIN_AFTER_CONFIRM"):
            # N.B. this is a (small) security risk if email went to wrong place.
            # and you have the LOGIN_WITHOUT_CONFIRMATION flag since in that case
            # you can be logged in and doing stuff - but another person could
            # get the email.
            # Note also this goes against OWASP recommendations.
            response = _security.two_factor_plugins.tf_enter(
                user, False, "confirm", next_loc=propagate_next(request.url, None)
            )
            if response:
                do_flash(m, c)
                return response
            login_user(user, authn_via=["confirm"])

    if cv("REDIRECT_BEHAVIOR") == "spa":
        return redirect(
            get_url(
                cv("POST_CONFIRM_VIEW"),
                qparams=user.get_redirect_qparams({c: m}),
            )
        )
    do_flash(m, c)
    return redirect(
        get_url(cv("POST_CONFIRM_VIEW"))
        or get_url(
            cv("POST_LOGIN_VIEW") if cv("AUTO_LOGIN_AFTER_CONFIRM") else ".login"
        )
    )


@unauth_csrf()
def forgot_password():
    """View function that handles a forgotten password request (/reset).
    This is allowed for either anonymous or authenticated users. The rationale is that
    often users stay logged in for a long time and might have forgotten their password
    and might be prompted for it for sensitive operations (/verify).
    """
    form = t.cast(ForgotPasswordForm, build_form_from_request("forgot_password_form"))

    if form.validate_on_submit():
        send_reset_password_instructions(form.user)
        if not _security._want_json(request):
            do_flash(*get_message("PASSWORD_RESET_REQUEST", email=form.email.data))

    elif request.method == "POST" and cv("RETURN_GENERIC_RESPONSES"):
        # Here on failed validate (POST) and want generic responses
        rinfo = dict(email=dict())
        form_errors_munge(form, rinfo)  # by suppressing errors JSON should return 200
        # Check for other errors - for default form - there aren't additional fields
        # but applications might add some (e.g. recaptcha)
        if not form.errors:
            # No OTHER errors on form.
            # Make look exactly like successful (e.g. real user) request
            hash_password("not-a-password")  # reduce timing between successful and not.
            if not _security._want_json(request):
                do_flash(*get_message("PASSWORD_RESET_REQUEST", email=form.email.data))

    if _security._want_json(request):
        # Never include user info since this is an anonymous endpoint.
        return base_render_json(form, include_user=False)

    if rurl := confirm_redirect(form, "email"):
        return rurl

    if is_user_authenticated(current_user):
        form.email.data = current_user.email
    return _security.render_template(
        cv("FORGOT_PASSWORD_TEMPLATE"),
        forgot_password_form=form,
        **_ctx("forgot_password"),
    )


@unauth_csrf()
def reset_password(token):
    """View function that handles a reset password request (/reset/<token>).

    This endpoint can be called either when authenticated or anonymous

    This is usually called via GET as part of an email link and redirects to
    a reset-password form.
    It is called via POST to actually update the password (and then redirects to
    a post reset/login view)
    If in either case the token is either invalid or expired it redirects to
    the 'forgot-password' form.

    In the case of non-form based configuration:
    For GET normal case - redirect to RESET_VIEW?token={token}
    For GET invalid case - redirect to RESET_ERROR_VIEW?error={error}
    For POST normal/successful case - return 200 with new authentication token
    For POST error case return 400
    """

    expired, invalid, user = reset_password_token_status(token)
    form = t.cast(ResetPasswordForm, build_form_from_request("reset_password_form"))
    form.user = user

    if request.method == "GET":
        if not user or invalid or expired:
            if expired:
                m, c = get_message(
                    "PASSWORD_RESET_EXPIRED",
                    within=cv("RESET_PASSWORD_WITHIN"),
                )
            else:
                m, c = get_message("INVALID_RESET_PASSWORD_TOKEN")
            if cv("REDIRECT_BEHAVIOR") == "spa":
                return redirect(get_url(cv("RESET_ERROR_VIEW"), qparams={c: m}))
            do_flash(m, c)
            return redirect(url_for_security("forgot_password"))

        # All good - for SPA - redirect to the ``reset_view``
        # Still - don't include PII such as identity and email if someone
        # intercepts link they still won't necessarily know the login identity
        # (even though they can change the password!).
        if cv("REDIRECT_BEHAVIOR") == "spa":
            return redirect(
                get_url(
                    cv("RESET_VIEW"),
                    qparams={"token": token},
                )
            )
        # for forms - render the reset password form
        return _security.render_template(
            cv("RESET_PASSWORD_TEMPLATE"),
            reset_password_form=form,
            reset_password_token=token,
            **_ctx("reset_password"),
        )

    # This is the POST case.
    if not user or invalid or expired:
        if expired:
            m, c = get_message(
                "PASSWORD_RESET_EXPIRED", within=cv("RESET_PASSWORD_WITHIN")
            )
        else:
            m, c = get_message("INVALID_RESET_PASSWORD_TOKEN")

        if _security._want_json(request):
            form.form_errors.append(m)
            return base_render_json(form, include_user=False)
        else:
            do_flash(m, c)
            return redirect(url_for_security("forgot_password"))

    if form.validate_on_submit():
        after_this_request(view_commit)
        update_password(user, form.password.data)
        if cv("AUTO_LOGIN_AFTER_RESET"):
            # backwards compat - really shouldn't do this according to OWASP
            response = _security.two_factor_plugins.tf_enter(
                form.user, False, "reset", next_loc=propagate_next(request.url, None)
            )
            if response:
                return response
            # two factor not required - just login
            login_user(user, authn_via=["reset"])
            if _security._want_json(request):
                dummy_form = DummyForm(formdata=None)
                dummy_form.user = user
                return base_render_json(
                    dummy_form,
                    include_auth_token=allowed_auth_token(form.user),
                    additional=dict(tf_required=False),
                )
            else:
                do_flash(*get_message("PASSWORD_RESET"))
                return redirect(
                    get_url(cv("POST_RESET_VIEW")) or get_url(cv("POST_LOGIN_VIEW"))
                )
        else:
            if _security._want_json(request):
                return _security._render_json({}, 200, None, None)
            else:
                do_flash(*get_message("PASSWORD_RESET_NO_LOGIN"))
                return redirect(get_url(cv("POST_RESET_VIEW")) or get_url(".login"))

    # validation failure case - for forms - we try again including the token
    # for non-forms -  we just return errors and assume caller remembers token.
    if _security._want_json(request):
        return base_render_json(form)
    return _security.render_template(
        cv("RESET_PASSWORD_TEMPLATE"),
        reset_password_form=form,
        reset_password_token=token,
        **_ctx("reset_password"),
    )


@auth_required(lambda: cv("API_ENABLED_METHODS"))
def change_password():
    """View function which handles a change password request."""
    form = t.cast(ChangePasswordForm, build_form_from_request("change_password_form"))

    if not current_user.password:
        # This is case where user registered w/o a password - since we can't
        # confirm with existing password - make sure fresh using whatever authentication
        # method they have set up.
        if not check_and_update_authn_fresh(
            cv("FRESHNESS"),
            cv("FRESHNESS_GRACE_PERIOD"),
            get_request_attr("fs_authn_via"),
        ):
            return _security._reauthn_handler(
                cv("FRESHNESS"), cv("FRESHNESS_GRACE_PERIOD")
            )

    if form.validate_on_submit():
        after_this_request(view_commit)
        change_user_password(current_user._get_current_object(), form.new_password.data)
        if _security._want_json(request):
            form.user = current_user
            return base_render_json(
                form, include_auth_token=allowed_auth_token(form.user)
            )

        do_flash(*get_message("PASSWORD_CHANGE"))
        return redirect(
            get_url(cv("POST_CHANGE_VIEW")) or get_url(cv("POST_LOGIN_VIEW"))
        )

    active_password = True if current_user.password else False
    if _security._want_json(request):
        form.user = current_user
        payload = dict(active_password=active_password)
        return base_render_json(form, additional=payload)

    return _security.render_template(
        cv("CHANGE_PASSWORD_TEMPLATE"),
        change_password_form=form,
        active_password=active_password,
        **_ctx("change_password"),
    )


@unauth_csrf()
def two_factor_setup():
    """View function for two-factor setup.

    This is used both for GET to fetch forms and POST to actually set configuration
    (and send token).

    There are 3 cases for setting up:
    1) initial login and application requires 2FA
    2) changing existing 2FA information
    3) user wanting to enable or disable 2FA (assuming application doesn't require it)

    In order to CHANGE/ENABLE/DISABLE a 2FA information, user must be properly logged in
    AND have a 'fresh' authentication.

    For initial login when 2FA required of course user can't be logged in - in this
    case we need to have been sent some
    state via the session as part of login to show a) who and b) that they successfully
    authenticated.
    """
    form = t.cast(TwoFactorSetupForm, build_form_from_request("two_factor_setup_form"))

    changing = is_user_authenticated(current_user)
    if not changing:
        # This is the initial login case
        if not all(k in session for k in ["tf_user_id", "tf_state"]) or session[
            "tf_state"
        ] not in ["setup_from_login", "validating_profile"]:
            # illegal call on this endpoint
            tf_clean_session()
            return tf_illegal_state(form, cv("TWO_FACTOR_ERROR_VIEW"))

        user = _datastore.find_user(fs_uniquifier=session["tf_user_id"])
        if not user:
            tf_clean_session()
            return tf_illegal_state(form, cv("TWO_FACTOR_ERROR_VIEW"))

    else:
        # Caller is changing their TFA profile. This requires a 'fresh' authentication
        # N.B unauth_csrf has done the CSRF check already.
        if not check_and_update_authn_fresh(
            cv("FRESHNESS"),
            cv("FRESHNESS_GRACE_PERIOD"),
            get_request_attr("fs_authn_via"),
        ):
            return _security._reauthn_handler(
                cv("FRESHNESS"), cv("FRESHNESS_GRACE_PERIOD")
            )
        user = current_user
    form.user = user

    if form.validate_on_submit():
        # Before storing in DB and therefore requiring 2FA we need to
        # make sure it actually works.
        # Requiring 2FA is triggered by having:
        # - BOTH tf_totp_secret and tf_primary_method in the user record
        # - OR having the application global config TWO_FACTOR_REQUIRED
        # - OR User.check_tf_required() returns True (overridden by app).
        # Until we correctly validate the 2FA - we don't set primary_method in
        # user model but use the session to store it.
        pm = form.setup.data
        if pm == "disable":
            tf_disable(user)
            after_this_request(view_commit)
            if not _security._want_json(request):
                do_flash(*get_message("TWO_FACTOR_DISABLED"))
                return redirect(get_url(cv("TWO_FACTOR_POST_SETUP_VIEW")))
            else:
                return base_render_json(form)

        # Regenerate the TOTP secret on every call of 2FA setup
        totp = _security.totp_factory.generate_totp_secret()
        phone = form.phone.data if pm == "sms" else None
        session["tf_totp_secret"] = totp
        session["tf_primary_method"] = pm
        session["tf_state"] = "validating_profile"
        # currently - state_token only works for changing TFA - not initial login
        state_token = None
        if changing:
            state = {
                "totp_secret": totp,
                "method": pm,
                "phone": phone,
            }
            state_token = _security.tf_setup_serializer.dumps(state)
        json_response = {
            "tf_state": "validating_profile",  # deprecated in 5.5.0
            "tf_primary_method": pm,  # old
            "tf_method": pm,
            "tf_state_token": state_token,
        }
        if phone:
            #  TODO dont save here - wait until complete
            user.tf_phone_number = phone
            _datastore.put(user)
            after_this_request(view_commit)

        if (
            pm == "email" or pm == "sms"
        ):  # TODO not sure this is needed - send checks this
            msg = user.tf_send_security_token(
                method=pm,
                totp_secret=totp,
                phone_number=phone,
            )
            if msg:
                # send code didn't work
                form.setup.errors = list()
                form.setup.errors.append(msg)
                if _security._want_json(request):
                    return base_render_json(
                        form, include_user=False, error_status_code=500
                    )

        qrcode_values = dict()
        if pm == "authenticator":
            authr_setup_values = _security.totp_factory.fetch_setup_values(totp, user)
            # Add all the values used in qrcode to json response
            json_response["tf_authr_key"] = authr_setup_values["key"]
            json_response["tf_authr_b32key"] = authr_setup_values["b32key"]
            json_response["tf_authr_username"] = authr_setup_values["username"]
            json_response["tf_authr_issuer"] = authr_setup_values["issuer"]
            json_response["tf_authr_uri"] = authr_setup_values["uri"]

            qrcode_values = dict(
                authr_qrcode=authr_setup_values["image"],
                authr_key=authr_setup_values["key"],
                authr_username=authr_setup_values["username"],
                authr_issuer=authr_setup_values["issuer"],
            )
        if _security._want_json(request):
            return base_render_json(form, include_user=False, additional=json_response)
        code_form = build_form("two_factor_verify_code_form")
        return _security.render_template(
            cv("TWO_FACTOR_SETUP_TEMPLATE"),
            two_factor_setup_form=form,
            two_factor_verify_code_form=code_form,
            choices=cv("TWO_FACTOR_ENABLED_METHODS"),
            chosen_method=pm,  # do not translate
            primary_method=localize_callback(
                _setup_methods_xlate[getattr(user, "tf_primary_method", None)]
            ),
            changing=changing,
            state_token=state_token,
            **qrcode_values,
            **_ctx("tf_setup"),
        )

    # We get here on GET and POST with failed validation.
    choices = cv("TWO_FACTOR_ENABLED_METHODS")[:]
    tf_required = user.check_tf_required_setup()
    if (not tf_required) and user.tf_primary_method is not None:
        choices.insert(0, "disable")

    if _security._want_json(request):
        # Provide information application/UI might need to render their own form/input
        json_response = {
            "tf_required": tf_required,
            "tf_primary_method": getattr(user, "tf_primary_method", None),  # old
            "tf_method": getattr(user, "tf_primary_method", None),
            "tf_phone_number": getattr(user, "tf_phone_number", None),
            "tf_available_methods": choices,
        }
        return base_render_json(form, include_user=False, additional=json_response)

    code_form = build_form("two_factor_verify_code_form")
    return _security.render_template(
        cv("TWO_FACTOR_SETUP_TEMPLATE"),
        two_factor_setup_form=form,
        two_factor_verify_code_form=code_form,
        choices=choices,
        chosen_method=None,
        primary_method=localize_callback(
            _setup_methods_xlate[getattr(user, "tf_primary_method", None)]
        ),
        changing=changing,
        state_token=None,
        two_factor_required=tf_required,
        **_ctx("tf_setup"),
    )


@auth_required(lambda: cv("API_ENABLED_METHODS"))
def two_factor_setup_validate(token: str) -> ResponseValue:
    """
    Validate new setup.
    The token is the state variable that is signed and timed
    and contains all the state that once confirmed will be stored in the user record.
    Unlike the code in two_factor_token_validation - this works w/o a session.
    It also is JUST for setting up/changing two factor for an authenticated user.
    """
    form = t.cast(
        TwoFactorVerifyCodeForm, build_form_from_request("two_factor_verify_code_form")
    )

    expired, invalid, state = check_and_get_token_status(
        token, "tf_setup", get_within_delta("TWO_FACTOR_SETUP_WITHIN")
    )
    if invalid:
        m, c = get_message("API_ERROR")
    if expired:
        m, c = get_message(
            "TWO_FACTOR_SETUP_EXPIRED", within=cv("TWO_FACTOR_SETUP_WITHIN")
        )
    if invalid or expired:
        tf_clean_session()  # until we completely remove session based setup/state
        if _security._want_json(request):
            form.form_errors.append(m)
            return base_render_json(form, include_user=False)
        do_flash(m, c)
        return redirect(url_for_security("two_factor_setup"))

    totp_secret = state["totp_secret"]
    method = state["method"]
    phone = state["phone"]
    form.tf_totp_secret = totp_secret
    form.primary_method = method
    form.user = current_user
    form.is_setup = True

    if form.validate_on_submit():
        tf_clean_session()  # until we completely remove session based setup/state
        after_this_request(view_commit)
        _datastore.tf_set(current_user, method, totp_secret, phone)
        # TODO: should validity cookie be removed? extended? left alone?
        # Currently - leave it alone - meaning cookie still set.

        tf_profile_changed.send(
            current_app._get_current_object(),  # type: ignore[attr-defined]
            _async_wrapper=current_app.ensure_sync,
            user=current_user,
            method=method,
        )

        if _security._want_json(request):
            return base_render_json(
                form,
                include_user=False,
                additional=dict(
                    tf_method=method,
                    tf_primary_method=method,
                    tf_phone=current_user.tf_phone_number,
                ),
            )
        else:
            do_flash(*get_message("TWO_FACTOR_CHANGE_METHOD_SUCCESSFUL"))
            return redirect(get_url(cv("TWO_FACTOR_POST_SETUP_VIEW")))

    # Code not correct/outdated.
    if _security._want_json(request):
        return base_render_json(form, include_user=False)
    m, c = get_message("TWO_FACTOR_INVALID_TOKEN")
    do_flash(m, c)
    return redirect(url_for_security("two_factor_setup"))


@unauth_csrf()
def two_factor_token_validation():
    """View function for two-factor token validation

    Two cases:
    1) normal login case - everything setup correctly; normal 2FA validation
       In this case - user not logged in -
       but 'tf_state' == 'ready' or 'validating_profile'
    2) validating after CHANGE/ENABLE 2FA. In this case user logged in/authenticated
       In this case we allow a GET to get the specific enter-code form.

    """
    form = t.cast(
        TwoFactorVerifyCodeForm, build_form_from_request("two_factor_verify_code_form")
    )

    # state info in session
    pm = session.get("tf_primary_method", None)
    totp_secret = session.get("tf_totp_secret", None)
    tf_state = session.get("tf_state", None)
    tf_user_id = session.get("tf_user_id", None)

    changing = is_user_authenticated(current_user)
    if not changing:
        # This is the normal login case OR initial setup (two factor required)
        if (
            tf_state not in ["ready", "validating_profile"]
            or (tf_state == "validating_profile" and not all([pm, totp_secret]))
            or not tf_user_id
        ):
            # illegal call on this endpoint
            tf_clean_session()
            return tf_illegal_state(form, cv("TWO_FACTOR_ERROR_VIEW"))

        user = _datastore.find_user(fs_uniquifier=tf_user_id)
        form.user = user
        if not user:
            tf_clean_session()
            return tf_illegal_state(form, cv("TWO_FACTOR_ERROR_VIEW"))

        if tf_state == "ready":
            # normal login case - use saved values
            pm = user.tf_primary_method
            totp_secret = user.tf_totp_secret
    else:
        # Changing TFA profile - user is already authenticated.
        if tf_state != "validating_profile" or not all([pm, totp_secret]):
            tf_clean_session()
            # logout since this seems like attack-ish/logic error
            logout_user()
            return tf_illegal_state(form, cv("TWO_FACTOR_ERROR_VIEW"))
        form.user = current_user
        form.is_setup = True

    form.primary_method = pm
    form.tf_totp_secret = totp_secret
    if form.validate_on_submit():
        # Success - finish process based on 'changing' and clear all session variables
        completion_message, token = complete_two_factor_process(
            form.user, pm, totp_secret, changing
        )

        after_this_request(view_commit)
        if token:
            after_this_request(partial(tf_set_validity_token_cookie, token=token))

        if not _security._want_json(request):
            do_flash(*get_message(completion_message))
            if changing:
                return redirect(get_url(cv("TWO_FACTOR_POST_SETUP_VIEW")))
            else:
                return redirect(get_post_login_redirect())

        else:
            return base_render_json(
                form, include_auth_token=allowed_auth_token(form.user)
            )

    # GET or not successful POST

    # if we were trying to validate a new method
    if changing:
        if _security._want_json(request):
            return base_render_json(form)
        # allow app to fetch just this form (independent of /tf_setup)
        return _security.render_template(
            cv("TWO_FACTOR_VERIFY_CODE_TEMPLATE"),
            two_factor_verify_code_form=form,
            chosen_method=localize_callback(_setup_methods_xlate[pm]),
            **_ctx("tf_token_validation"),
        )

    # if we were trying to validate an existing method
    else:
        rescue_form = build_form("two_factor_rescue_form")
        recovery_options = set_rescue_options(rescue_form, form.user)
        if _security._want_json(request):
            return base_render_json(
                form, additional=dict(recovery_options=recovery_options)
            )
        return _security.render_template(
            cv("TWO_FACTOR_VERIFY_CODE_TEMPLATE"),
            two_factor_rescue_form=rescue_form,
            two_factor_verify_code_form=form,
            chosen_method=localize_callback(_setup_methods_xlate[pm]),
            problem=None,
            **_ctx("tf_token_validation"),
        )


@anonymous_user_required
@unauth_csrf()
def two_factor_rescue():
    """Function that handles a situation where user can't
    enter his two-factor validation code

    User must have already provided valid username/password.
    User must have already established 2FA

    """
    form = t.cast(
        TwoFactorRescueForm, build_form_from_request("two_factor_rescue_form")
    )

    form.user = tf_check_state(["ready"])
    if not form.user:
        return tf_illegal_state(form, cv("TWO_FACTOR_ERROR_VIEW"))

    recovery_options = set_rescue_options(form, form.user)
    rproblem = ""
    if form.validate_on_submit():
        raction = form.help_setup.data
        rproblem = raction
        if raction == "email":
            msg = form.user.tf_send_security_token(
                method="email",
                totp_secret=form.user.tf_totp_secret,
                phone_number=getattr(form.user, "tf_phone_number", None),
            )
            if msg:
                rproblem = ""
                form.help_setup.errors.append(msg)
                if _security._want_json(request):
                    return base_render_json(
                        form, include_user=False, error_status_code=500
                    )
            # drop through to GET path
        elif raction == "recovery_code":
            return redirect(url_for_security("mf_recovery"))

        # send app provider a mail message regarding trouble
        elif raction == "help":
            send_mail(
                cv("EMAIL_SUBJECT_TWO_FACTOR_RESCUE"),
                cv("TWO_FACTOR_RESCUE_MAIL"),
                "two_factor_rescue",
                user=form.user,
            )
            # drop through to GET path
        else:
            return "", 404

    if _security._want_json(request):
        return base_render_json(
            form, include_user=False, additional=dict(recovery_options=recovery_options)
        )

    code_form = build_form("two_factor_verify_code_form")
    return _security.render_template(
        cv("TWO_FACTOR_VERIFY_CODE_TEMPLATE"),
        two_factor_verify_code_form=code_form,
        two_factor_rescue_form=form,
        chosen_method=localize_callback(
            _setup_methods_xlate[form.user.tf_primary_method]
        ),
        rescue_mail=cv("TWO_FACTOR_RESCUE_MAIL"),
        problem=rproblem,
        **_ctx("tf_token_validation"),
    )


@anonymous_user_required
@unauth_csrf()
def recover_username():
    """View function for username recovery"""

    form = t.cast(
        UsernameRecoveryForm, build_form_from_request("username_recovery_form")
    )

    if form.validate_on_submit():
        send_username_recovery_email(form.user)

        if _security._want_json(request):
            return base_render_json(form, include_user=False)

        do_flash(*get_message("USERNAME_RECOVERY_REQUEST"))

        return redirect(url_for_security("login"))
    elif request.method == "POST" and cv("RETURN_GENERIC_RESPONSES"):
        rinfo = dict(email=dict())
        form_errors_munge(form, rinfo)
        if not form.errors:
            if not _security._want_json(request):
                do_flash(*get_message("USERNAME_RECOVERY_REQUEST"))

    if _security._want_json(request):
        return base_render_json(form, include_user=False)

    if rurl := confirm_redirect(form, "email"):
        return rurl

    return _security.render_template(
        cv("USERNAME_RECOVERY_TEMPLATE"),
        username_recovery_form=form,
        **_ctx("recover_username"),
    )


def create_blueprint(app, state, import_name):
    """Creates the security extension blueprint"""

    bp = Blueprint(
        cv("BLUEPRINT_NAME", app=app),
        import_name,
        url_prefix=cv("URL_PREFIX", app=app),
        subdomain=cv("SUBDOMAIN", app=app),
        template_folder="templates",
        static_folder=cv("STATIC_FOLDER", app),
        static_url_path=cv("STATIC_FOLDER_URL", app),
    )

    if cv("LOGOUT_METHODS", app=app) is not None:
        bp.route(
            cv("LOGOUT_URL", app=app),
            methods=cv("LOGOUT_METHODS", app=app),
            endpoint="logout",
        )(logout)

    login_url = cv("LOGIN_URL", app=app)
    if state.passwordless:
        bp.route(login_url, methods=["GET", "POST"], endpoint="login")(send_login)
        bp.route(
            login_url + slash_url_suffix(login_url, "<token>"),
            endpoint="token_login",
        )(token_login)
    elif cv("US_SIGNIN_REPLACES_LOGIN", app=app):
        bp.route(login_url, methods=["GET", "POST"], endpoint="login")(us_signin)

    else:
        bp.route(login_url, methods=["GET", "POST"], endpoint="login")(login)

    if cv("FRESHNESS", app=app).total_seconds() >= 0:
        bp.route(cv("VERIFY_URL", app=app), methods=["GET", "POST"], endpoint="verify")(
            verify
        )

    if state.unified_signin:
        us_signin_url = cv("US_SIGNIN_URL", app=app)
        us_signin_send_code_url = cv("US_SIGNIN_SEND_CODE_URL", app=app)
        us_setup_url = cv("US_SETUP_URL", app=app)
        us_verify_url = cv("US_VERIFY_URL", app=app)
        us_verify_send_code_url = cv("US_VERIFY_SEND_CODE_URL", app=app)
        us_verify_link_url = cv("US_VERIFY_LINK_URL", app=app)
        bp.route(us_signin_url, methods=["GET", "POST"], endpoint="us_signin")(
            us_signin
        )
        bp.route(
            us_signin_send_code_url,
            methods=["POST"],
            endpoint="us_signin_send_code",
        )(us_signin_send_code)

        bp.route(us_setup_url, methods=["GET", "POST"], endpoint="us_setup")(us_setup)
        bp.route(
            us_setup_url + slash_url_suffix(us_setup_url, "<token>"),
            methods=["POST"],
            endpoint="us_setup_validate",
        )(us_setup_validate)

        # Freshness verification
        if cv("FRESHNESS", app=app).total_seconds() >= 0:
            bp.route(us_verify_url, methods=["GET", "POST"], endpoint="us_verify")(
                us_verify
            )
            bp.route(
                us_verify_send_code_url,
                methods=["POST"],
                endpoint="us_verify_send_code",
            )(us_verify_send_code)

        bp.route(us_verify_link_url, methods=["GET"], endpoint="us_verify_link")(
            us_verify_link
        )

    if state.two_factor:
        two_factor_setup_url = cv("TWO_FACTOR_SETUP_URL", app=app)
        two_factor_token_validation_url = cv("TWO_FACTOR_TOKEN_VALIDATION_URL", app=app)
        two_factor_rescue_url = cv("TWO_FACTOR_RESCUE_URL", app=app)
        bp.route(
            two_factor_setup_url,
            methods=["GET", "POST"],
            endpoint="two_factor_setup",
        )(two_factor_setup)
        bp.route(
            two_factor_setup_url + slash_url_suffix(two_factor_setup_url, "<token>"),
            methods=["POST"],
            endpoint="two_factor_setup_validate",
        )(two_factor_setup_validate)
        bp.route(
            two_factor_token_validation_url,
            methods=["GET", "POST"],
            endpoint="two_factor_token_validation",
        )(two_factor_token_validation)
        bp.route(
            two_factor_rescue_url,
            methods=["GET", "POST"],
            endpoint="two_factor_rescue",
        )(two_factor_rescue)

    if state.registerable:
        bp.route(
            cv("REGISTER_URL", app=app), methods=["GET", "POST"], endpoint="register"
        )(register)

    if state.recoverable:
        reset_url = cv("RESET_URL", app=app)
        bp.route(reset_url, methods=["GET", "POST"], endpoint="forgot_password")(
            forgot_password
        )
        bp.route(
            reset_url + slash_url_suffix(reset_url, "<token>"),
            methods=["GET", "POST"],
            endpoint="reset_password",
        )(reset_password)

    if state.username_recovery:
        username_recovery_url = cv("USERNAME_RECOVERY_URL", app=app)
        bp.route(
            username_recovery_url,
            methods=["GET", "POST"],
            endpoint="recover_username",
        )(recover_username)

    if state.changeable:
        bp.route(
            cv("CHANGE_URL", app=app),
            methods=["GET", "POST"],
            endpoint="change_password",
        )(change_password)

    if state.change_email:
        change_email_url = cv("CHANGE_EMAIL_URL", app=app)
        bp.route(
            change_email_url,
            methods=["GET", "POST"],
            endpoint="change_email",
        )(change_email)
        bp.route(
            change_email_url + slash_url_suffix(change_email_url, "<token>"),
            methods=["GET"],
            endpoint="change_email_confirm",
        )(change_email_confirm)

    if state.change_username:
        bp.route(
            cv("CHANGE_USERNAME_URL", app=app),
            methods=["GET", "POST"],
            endpoint="change_username",
        )(change_username)

    if state.confirmable:
        confirm_url = cv("CONFIRM_URL", app=app)
        bp.route(confirm_url, methods=["GET", "POST"], endpoint="send_confirmation")(
            send_confirmation
        )
        bp.route(
            confirm_url + slash_url_suffix(confirm_url, "<token>"),
            methods=["GET", "POST"],
            endpoint="confirm_email",
        )(confirm_email)

    if cv("MULTI_FACTOR_RECOVERY_CODES", app) and state.support_mfa:
        multi_factor_recovery_codes_url = cv("MULTI_FACTOR_RECOVERY_CODES_URL", app=app)
        multi_factor_recovery_url = cv("MULTI_FACTOR_RECOVERY_URL", app=app)
        bp.route(
            multi_factor_recovery_codes_url,
            methods=["GET", "POST"],
            endpoint="mf_recovery_codes",
        )(mf_recovery_codes)
        bp.route(
            multi_factor_recovery_url,
            methods=["GET", "POST"],
            endpoint="mf_recovery",
        )(mf_recovery)

    if state.webauthn:
        wan_register_url = cv("WAN_REGISTER_URL", app=app)
        wan_signin_url = cv("WAN_SIGNIN_URL", app=app)
        wan_delete_url = cv("WAN_DELETE_URL", app=app)
        wan_verify_url = cv("WAN_VERIFY_URL", app=app)
        bp.route(
            wan_register_url,
            methods=["GET", "POST"],
            endpoint="wan_register",
        )(webauthn_register)
        bp.route(
            wan_register_url + slash_url_suffix(wan_register_url, "<token>"),
            methods=["POST"],
            endpoint="wan_register_response",
        )(webauthn_register_response)

        bp.route(wan_signin_url, methods=["GET", "POST"], endpoint="wan_signin")(
            webauthn_signin
        )
        bp.route(
            wan_signin_url + slash_url_suffix(wan_signin_url, "<token>"),
            methods=["POST"],
            endpoint="wan_signin_response",
        )(webauthn_signin_response)
        bp.route(wan_delete_url, methods=["GET", "POST"], endpoint="wan_delete")(
            webauthn_delete
        )
        if cv("FRESHNESS", app=app).total_seconds() >= 0 and cv(
            "WAN_ALLOW_AS_VERIFY", app=app
        ):
            bp.route(wan_verify_url, methods=["GET", "POST"], endpoint="wan_verify")(
                webauthn_verify
            )
            bp.route(
                wan_verify_url + slash_url_suffix(wan_verify_url, "<token>"),
                methods=["POST"],
                endpoint="wan_verify_response",
            )(webauthn_verify_response)

    if state.refresh_token:
        refresh_url = cv("REFRESH_TOKEN_URL", app=app)
        bp.route(
            refresh_url,
            methods=["POST"],
            endpoint="refresh_token",
        )(refresh)
    return bp