"""
test_webauthn
~~~~~~~~~~~~~~~~~~~

WebAuthn tests

:copyright: (c) 2021-2025 by J. Christopher Wagner (jwag).
:license: MIT, see LICENSE for more details.

"""

from base64 import urlsafe_b64encode
import copy
import datetime
import json
import re

import pytest

from tests.test_two_factor import tf_in_session
from tests.test_utils import (
    FakeSerializer,
    authenticate,
    capture_flashes,
    check_location,
    get_existing_session,
    get_form_action,
    get_form_input_value,
    is_authenticated,
    json_authenticate,
    logout,
    reset_fresh,
    setup_tf_sms,
    verify_token,
)

from flask_security import (
    WebauthnUtil,
    user_authenticated,
    wan_registered,
    wan_deleted,
)

# We can't/don't test the actual client-side javascript and browser APIs - so
# to create reproducible tests, use view_scaffold, set breakpoints in the views and
# cut-and-paste the responses. That requires that 'challenge' and 'rp_origin' be
# identical between view_scaffold and tests here.
CHALLENGE = "smCCiy_k2CqQydSQ_kPEjV5a2d0ApfatcpQ1aXDmQPo"
REG_DATA1 = {
    "id": "wUUqNOjY35dcT-vpikZpZx-T91NjIe4PqrV8j7jYPOc",
    "rawId": "wUUqNOjY35dcT-vpikZpZx-T91NjIe4PqrV8j7jYPOc",
    "type": "public-key",
    "response": {
        "attestationObject": "o2NmbXRkbm9uZWdhdHRTdG10oGhhdXRoRGF0YVikSZYN5YgOjGh0NB"
        "cPZHZgW4_krrmihjLHmVzzuoMdl2NFAAAAAQAAAAAAAAAAAAAAAAAAA"
        "AAAIMFFKjTo2N-XXE_r6YpGaWcfk_dTYyHuD6q1fI-42DznpQECAy"
        "YgASFYIFRipoWMEiDuCtLUvSlqCFZBqxvUuNqZKavlWgvN2BK8Il"
        "ggLOV4eez9k0det5oIZGyKanGkmWa0hygnjjFmf8Rep6c",
        "clientDataJSON": "eyJ0eXBlIjoid2ViYXV0aG4uY3JlYXRlIiwiY2hhbGxlbmdlIjoiYzIxR"
        "FEybDVYMnN5UTNGUmVXUlRVVjlyVUVWcVZqVmhNbVF3UVhCbVlYUmpjRk"
        "V4WVZoRWJWRlFidyIsIm9yaWdpbiI6Imh0dHA6Ly9sb2NhbGhvc3Q6NT"
        "AwMSIsImNyb3NzT3JpZ2luIjpmYWxzZX0",
        "transports": ["usb"],
    },
    "extensions": '{"credProps": {}}',
}
SIGNIN_DATA1 = {
    "id": "wUUqNOjY35dcT-vpikZpZx-T91NjIe4PqrV8j7jYPOc",
    "rawId": "wUUqNOjY35dcT-vpikZpZx-T91NjIe4PqrV8j7jYPOc",
    "type": "public-key",
    "response": {
        "authenticatorData": "SZYN5YgOjGh0NBcPZHZgW4_krrmihjLHmVzzuoMdl2MBAAAABQ==",
        "clientDataJSON": "eyJ0eXBlIjoid2ViYXV0aG4uZ2V0IiwiY2hhbGxlbmdlIjoiYzIxRFEy"
        "bDVYMnN5UTNGUmVXUlRVVjlyVUVWcVZqVmhNbVF3UVhCbVlYUmpjRkV4"
        "WVZoRWJWRlFidyIsIm9yaWdpbiI6Imh0dHA6Ly9sb2NhbGhvc3Q6NTAw"
        "MSIsImNyb3NzT3JpZ2luIjpmYWxzZX0=",
        "signature": "MEUCIH5VdRXxfnoxfrVk72gvWAn91QH-l2UrIohk5YOWi9XpAiEAn6f9oHtFS"
        "68HVf6K_Ku0L33C0sID2HzpJWSiTNgJlbU=",
    },
    "assertionClientExtensions": "{}",
}
REG_DATA2 = {
    "id": "lpMv8FTVHVSxteQJ3N4azlSxXiBJADA7IK-NleETceZYODy51_Cqt7Rx6pfVP1BI",
    "rawId": "lpMv8FTVHVSxteQJ3N4azlSxXiBJADA7IK-NleETceZYODy51_Cqt7Rx6pfVP1BI",
    "type": "public-key",
    "response": {
        "attestationObject": "o2NmbXRkbm9uZWdhdHRTdG10oGhhdXRoRGF0YVjCSZYN5YgOj"
        "Gh0NBcPZHZgW4_krrmihjLHmVzzuoMdl2PFAAAAAgAAAAAAAAAAA"
        "AAAAAAAAAAAMJaTL_BU1R1UsbXkCdzeGs5UsV4gSQAwOyCvjZXhE3"
        "HmWDg8udfwqre0ceqX1T9QSKUBAgMmIAEhWCCWky_wVNUdVL"
        "G15AncLU8mBQCtY10BjnSDoOUlRjkU1CJYIIA1U9vNDpZ"
        "TihC2x0CxRZ-trF_zazYosuEqYdHSOIjZoWtjcmVkUHJvdGVjdAI",
        "clientDataJSON": "eyJ0eXBlIjoid2ViYXV0aG4uY3JlYXRlIiwiY2hhbGxlbmdlIjoi"
        "YzIxRFEybDVYMnN5UTNGUmVXUlRVVjlyVUVWcVZqVmhNbVF3UV"
        "hCbVlYUmpjRkV4WVZoRWJWRlFidyIsIm9yaWdpbiI6Imh0dHA6L"
        "y9sb2NhbGhvc3Q6NTAwMSIsImNyb3NzT3JpZ2luIjpmYWxzZX0",
        "transports": ["nfc", "usb"],
    },
    "extensions": '{"credProps": {"rk": True}}',
}

# This has user_verification=True - i.e. a multi-factor capable key
REG_DATA_UV = {
    "id": "s3xZpfGy0ZH-sSkfxIsgChwbkw_O0jOFtZeJ1LXUMEa8atG1oEskNqmFJCfgKZGy",
    "rawId": "s3xZpfGy0ZH-sSkfxIsgChwbkw_O0jOFtZeJ1LXUMEa8atG1oEskNqmFJCfgKZGy",
    "type": "public-key",
    "response": {
        "attestationObject": "o2NmbXRkbm9uZWdhdHRTdG10oGhhdXRoRGF0YVjC"
        "SZYN5YgOjGh0NBcPZHZgW4_krrmihjLHmVzzuoMdl2PFAAAABAAAAA"
        "AAAAAAAAAAAAAAAAAAMLN8WaXxstGR_rEpH8SLIAocG5MPztIzhbWXi"
        "dS11DBGvGrRtaBLJDaphSQn4CmRsqUBAgMmIAEhWCCzfFml8bLRkf"
        "6xKR_EUnaoI333MuxRlv5-LwojDibdTyJYIFMifFwn-RfkDDgsTHF"
        "jWgE6bld-Jc4nhFMTkQja9P8IoWtjcmVkUHJvdGVjdAI",
        "clientDataJSON": "eyJ0eXBlIjoid2ViYXV0aG4uY3JlYXRlIiwiY2hhbGxlbmdlIjoiYzI"
        "xRFEybDVYMnN5UTNGUmVXUlRVVjlyVUVWcVZqVmhNbVF3UVhCbVlY"
        "UmpjRkV4WVZoRWJWRlFidyIsIm9yaWdpbiI6Imh0dHA6Ly9sb2Nhb"
        "Ghvc3Q6NTAwMSIsImNyb3NzT3JpZ2luIjpmYWxzZX0",
        "transports": ["nfc", "usb"],
    },
    "extensions": '{"credProps":{"rk":true}}',
}

SIGNIN_DATA_UV = {
    "id": "s3xZpfGy0ZH-sSkfxIsgChwbkw_O0jOFtZeJ1LXUMEa8atG1oEskNqmFJCfgKZGy",
    "rawId": "s3xZpfGy0ZH-sSkfxIsgChwbkw_O0jOFtZeJ1LXUMEa8atG1oEskNqmFJCfgKZGy",
    "type": "public-key",
    "response": {
        "authenticatorData": "SZYN5YgOjGh0NBcPZHZgW4_krrmihjLHmVzzuoMdl2MFAAAABQ==",
        "clientDataJSON": "eyJ0eXBlIjoid2ViYXV0aG4uZ2V0IiwiY2hhbGxlbmdlIjoiYzIxRFEy"
        "bDVYMnN5UTNGUmVXUlRVVjlyVUVWcVZqVmhNbVF3UVhCbVlYUmpjRkV4W"
        "VZoRWJWRlFidyIsIm9yaWdpbiI6Imh0dHA6Ly9sb2NhbGhvc3Q6NTAwMSI"
        "sImNyb3NzT3JpZ2luIjpmYWxzZX0=",
        "signature": "MEUCIQDR0m9Ob4nqVGiAPUf1Tu5XohDh2frl1LJ6G41GURlUIgIgKUPfkw"
        "AjP2863L2nDhcR2EKqoGEQLqlQ5xymZstyO6o=",
    },
    "assertionClientExtensions": "{}",
}

REG_DATA_UH = {
    "id": "rHb1OXVM--dgGcWg0u3cfomyc-Tu4l4kK8GjVkS8bms-foXmBAlWHyTzuhgGgCnx",
    "rawId": "rHb1OXVM--dgGcWg0u3cfomyc-Tu4l4kK8GjVkS8bms-foXmBAlWHyTzuhgGgCnx",
    "type": "public-key",
    "response": {
        "attestationObject": "o2NmbXRkbm9uZWdhdHRTdG10oGhhdXRoRGF0YVjCSZYN5YgOjGh0NBc"
        "PZHZgW4_krrmihjLHmVzzuoMd"
        "l2PFAAAAAgAAAAAAAAAAAAAAAAAAAAAAMKx29Tl1TPvnYBnFoNLt3H6J"
        "snPk7uJeJCvBo1ZEvG5rPn6F"
        "5gQJVh8k87oYBoAp8aUBAgMmIAEhWCCsdvU5dUz752AZxaDSyN-ocBL"
        "Bo99GevEWTnUxSkMRICJYIILE"
        "DLF8cQNM5l6ZgDxIYpvU88xgbq44lmR6oCBbNaHhoWtjcmVkUHJvdG"
        "VjdAI",
        "clientDataJSON": "eyJ0eXBlIjoid2ViYXV0aG4uY3JlYXRlIiwiY2hhbGxlbmdlIjoiYzIx"
        "RFEybDVYMnN5UTNGUmVXUlRVVjl"
        "yVUVWcVZqVmhNbVF3UVhCbVlYUmpjRkV4WVZoRWJWRlFidyIsIm9yaWdpbi"
        "I6Imh0dHA6Ly9sb2NhbGhvc3"
        "Q6NTAwMSIsImNyb3NzT3JpZ2luIjpmYWxzZX0",
        "transports": ["nfc", "usb"],
    },
    "extensions": '{"credProps":{"rk": true}}"',
}
SIGNIN_DATA_UH = {
    "id": "rHb1OXVM--dgGcWg0u3cfomyc-Tu4l4kK8GjVkS8bms-foXmBAlWHyTzuhgGgCnx",
    "rawId": "rHb1OXVM--dgGcWg0u3cfomyc-Tu4l4kK8GjVkS8bms-foXmBAlWHyTzuhgGgCnx",
    "type": "public-key",
    "response": {
        "authenticatorData": "SZYN5YgOjGh0NBcPZHZgW4_krrmihjLHmVzzuoMdl2MFAAAABQ==",
        "clientDataJSON": "eyJ0eXBlIjoid2ViYXV0aG4uZ2V0IiwiY2hhbGxlbmdlIjoiYzIxRFEy"
        "bDVYMnN5UTNGUmVXUlRVVjlyVUVWcVZqVmhNbVF3UVhCbVlYUmpjR"
        "kV4WVZoRWJWRlFidyIsIm9yaWdpbiI6Imh0dHA6Ly9sb2NhbGhvc3Q"
        "6NTAwMSIsImNyb3NzT3JpZ2luIjpmYWxzZX0=",
        "signature": "MEUCIQCbKwaQv_GzrWfc0nVXqhe6WZs_5Sb2b7xHC9iDW9aHeQIgF2PlfM7FdyV"
        "xcPhofekJjLBgDMTbK4mwIWHgiExZ54s=",
        "userHandle": "NTgxMTU3YmM2MGU3NGM1OTg0OTBjYTI1ZTgwNjc4MDY=",
    },
    "assertionClientExtensions": "{}",
}


class HackWebauthnUtil(WebauthnUtil):
    def generate_challenge(self, nbytes=None):
        return CHALLENGE

    def origin(self):
        # This is from view_scaffold
        return "http://localhost:5001"


pytestmark = pytest.mark.webauthn(webauthn_util_cls=HackWebauthnUtil)


def _register_start(
    client, name="testr1", usage="secondary", endpoint="wan-register", csrf_token=None
):
    response = client.post(
        endpoint, data=dict(name=name, usage=usage, csrf_token=csrf_token)
    )
    matcher = re.match(
        r".*handleRegister\(\'(.*)\'\).*",
        response.data.decode("utf-8"),
        re.IGNORECASE | re.DOTALL,
    )
    register_options = json.loads(matcher.group(1))
    response_url = get_form_action(response)
    return register_options, response_url


def _register_start_json(client, name="testr1", usage="secondary", csrf_token=None):
    response = client.post(
        "wan-register", json=dict(name=name, usage=usage, csrf_token=csrf_token)
    )
    register_options = response.json["response"]["credential_options"]
    response_url = f'wan-register/{response.json["response"]["wan_state"]}'
    return register_options, response_url


def reg_2_keys(client):
    # Register 2 keys - one first, one secondary
    # This can be used by other tests outside this module.
    authenticate(client)
    register_options, response_url = _register_start_json(
        client, name="first", usage="first"
    )
    response = client.post(response_url, json=dict(credential=json.dumps(REG_DATA_UV)))
    assert response.status_code == 200

    register_options, response_url = _register_start_json(
        client, name="second", usage="secondary"
    )
    response = client.post(response_url, json=dict(credential=json.dumps(REG_DATA1)))
    assert response.status_code == 200

    return {
        "first": {"id": REG_DATA_UV["id"], "signin": SIGNIN_DATA_UV},
        "secondary": {"id": REG_DATA1["id"], "signin": SIGNIN_DATA1},
    }


def reg_first_key(client, csrf_token=None):
    # Register a primary key - assumes already authenticated
    # This can be used by other tests outside this module.
    register_options, response_url = _register_start_json(
        client, name="first", usage="first", csrf_token=csrf_token
    )
    response = client.post(
        response_url,
        json=dict(credential=json.dumps(REG_DATA_UV), csrf_token=csrf_token),
    )
    assert response.status_code == 200
    return {"id": REG_DATA_UV["id"], "signin": SIGNIN_DATA_UV}


def _signin_start(
    client,
    identity=None,
    endpoint="wan-signin",
    csrf_token=None,
):
    response = client.post(
        endpoint, data=dict(identity=identity, csrf_token=csrf_token)
    )
    matcher = re.match(
        r".*handleSignin\(\'(.*)\'\).*",
        response.data.decode("utf-8"),
        re.IGNORECASE | re.DOTALL,
    )
    signin_options = json.loads(matcher.group(1))
    response_url = get_form_action(response)
    return signin_options, response_url


def _signin_start_json(client, identity=None, remember=False, endpoint="wan-signin"):
    headers = {"Accept": "application/json", "Content-Type": "application/json"}

    response = client.post(
        endpoint, headers=headers, json=dict(identity=identity, remember=remember)
    )
    signin_options = response.json["response"]["credential_options"]
    response_url = f'wan-signin/{response.json["response"]["wan_state"]}'
    return signin_options, response_url, response.json


def wan_signin(client, identity, signin_data, wan_signin_url):
    # perform complete sign in - useful for tests outside this module.
    signin_options, response_url = _signin_start(
        client, identity, endpoint=wan_signin_url
    )
    response = client.post(
        response_url,
        data=dict(credential=json.dumps(signin_data)),
        follow_redirects=True,
    )
    assert response.status_code == 200
    return response


def reset_signcount(app, email, keyname):
    # Due to replay attack prevention, we can only use a key once since the server
    # increments the sign_count and we can't do that on the client side!
    with app.app_context():
        user = app.security.datastore.find_user(email=email)
        cred = [c for c in user.webauthn if c.name == keyname][0]
        cred.sign_count = cred.sign_count - 1
        app.security.datastore.put(cred)
        app.security.datastore.commit()


def test_basic(app, clients, get_message):
    auths = []

    @user_authenticated.connect_via(app)
    def authned(myapp, user, **extra_args):
        auths.append((user.email, extra_args["authn_via"]))

    @wan_registered.connect_via(app)
    def pc(sender, user, name, **extra_args):
        assert name == "testr1"
        assert len(user.webauthn) == 1

    authenticate(clients)

    response = clients.get("/wan-register")
    # default config allows for both primary and secondary usage
    # so form should have selector
    assert get_form_input_value(response, "usage-0")
    assert get_form_input_value(response, "usage-1")

    # post with no name
    response = clients.post("/wan-register", data=dict())
    assert get_message("WEBAUTHN_NAME_REQUIRED") in response.data

    register_options, response_url = _register_start(clients, usage="first")
    assert register_options["rp"]["name"] == "My Flask App"
    assert register_options["user"]["name"] == "matt@lp.com"
    assert not register_options["excludeCredentials"]
    assert register_options["authenticatorSelection"]["residentKey"] == "preferred"
    assert register_options["extensions"]["credProps"]

    # Register using the static data above
    response = clients.post(
        response_url, data=dict(credential=json.dumps(REG_DATA1)), follow_redirects=True
    )
    assert response.status_code == 200
    assert get_message("WEBAUTHN_REGISTER_SUCCESSFUL", name="testr1") in response.data
    assert b"testr1" in response.data

    # sign in - simple case use identity so we get back allowCredentials
    logout(clients)
    signin_options, response_url = _signin_start(clients, "matt@lp.com")
    assert signin_options["timeout"] == app.config["SECURITY_WAN_SIGNIN_TIMEOUT"]
    assert signin_options["userVerification"] == "preferred"
    allow_credentials = signin_options["allowCredentials"]
    assert len(allow_credentials) == 1
    assert allow_credentials[0]["id"] == REG_DATA1["id"]
    assert allow_credentials[0]["transports"] == ["usb"]

    response = clients.post(
        response_url,
        data=dict(credential=json.dumps(SIGNIN_DATA1)),
        follow_redirects=True,
    )
    assert response.status_code == 200
    assert b"Welcome matt@lp.com" in response.data
    assert len(auths) == 2
    assert auths[1][1] == ["webauthn"]

    # verify actually logged in
    response = clients.get("/profile", follow_redirects=False)
    assert response.status_code == 200


def test_basic_json(app, clients, get_message):
    headers = {"Accept": "application/json", "Content-Type": "application/json"}

    auths = []

    @user_authenticated.connect_via(app)
    def authned(myapp, user, **extra_args):
        auths.append((user.email, extra_args["authn_via"]))

    authenticate(clients)

    # post with no name
    response = clients.post("/wan-register", json=dict())
    assert response.status_code == 400
    assert response.json["response"]["errors"][0].encode("utf-8") == get_message(
        "WEBAUTHN_NAME_REQUIRED"
    )

    register_options, response_url = _register_start_json(clients, usage="first")
    assert register_options["rp"]["name"] == "My Flask App"
    assert register_options["user"]["name"] == "matt@lp.com"

    # Register using the static data above
    response = clients.post(response_url, json=dict(credential=json.dumps(REG_DATA1)))
    assert response.status_code == 200

    # reset lastuse_datetime so we can verify signing in correctly alters it
    fake_dt = datetime.datetime(2020, 4, 7, 9, 27)
    with app.app_context():
        user = app.security.datastore.find_user(email="matt@lp.com")
        for cred in user.webauthn:
            cred.lastuse_datetime = fake_dt
            app.security.datastore.put(cred)
        app.security.datastore.commit()
        if hasattr(app.security.datastore.db, "close_db") and callable(
            app.security.datastore.db.close_db
        ):
            app.security.datastore.db.close_db(None)

    response = clients.get("/wan-register", headers=headers)
    active_creds = response.json["response"]["registered_credentials"]
    assert active_creds[0]["name"] == "testr1"
    assert datetime.datetime.fromisoformat(active_creds[0]["lastuse"]) == fake_dt

    # sign in - simple case use identity so we get back allowCredentials
    logout(clients)
    signin_options, response_url, rjson = _signin_start_json(clients, "matt@lp.com")
    assert signin_options["userVerification"] == "preferred"
    allow_credentials = signin_options["allowCredentials"]
    assert len(allow_credentials) == 1
    assert allow_credentials[0]["id"] == REG_DATA1["id"]
    assert "user" not in rjson["response"]

    response = clients.post(
        f"{response_url}?include_auth_token",
        json=dict(credential=json.dumps(SIGNIN_DATA1)),
    )
    assert response.status_code == 200
    assert response.json["response"]["user"]["email"] == "matt@lp.com"
    assert auths[1][1] == ["webauthn"]

    # verify actually logged in
    assert is_authenticated(clients, get_message)
    verify_token(clients, response.json["response"]["user"]["authentication_token"])

    # fetch credentials and verify lastuse was updated
    response = clients.get("/wan-register", headers=headers)
    active_creds = response.json["response"]["registered_credentials"]
    assert datetime.datetime.fromisoformat(active_creds[0]["lastuse"]) != fake_dt
    assert active_creds[0]["transports"] == ["usb"]
    assert active_creds[0]["usage"] == "first"

    logout(clients)
    # verify that unknown identities are just ignored when USER_HINTS is True
    response = clients.post("/wan-signin", json=dict(identity="whoami@lp.com"))
    assert response.status_code == 200


@pytest.mark.settings(wan_allow_user_hints=False)
def test_basic_json_nohints(app, client, get_message):
    # Test that with no hints allowed, we don't get any credentials and we can still
    # sign in.
    authenticate(client)
    register_options, response_url = _register_start_json(client, usage="first")
    response = client.post(response_url, json=dict(credential=json.dumps(REG_DATA1)))
    assert response.status_code == 200
    # With no hints we default to requiring a resident key
    # With allow as primary we default to requiring a cross-platform key
    assert (
        register_options["authenticatorSelection"]["authenticatorAttachment"]
        == "cross-platform"
    )
    assert register_options["authenticatorSelection"]["residentKey"] == "required"
    logout(client)

    signin_options, response_url, rjson = _signin_start_json(client, "matt@lp.com")
    allow_credentials = signin_options["allowCredentials"]
    assert len(allow_credentials) == 0
    assert "user" not in rjson["response"]

    response = client.post(
        response_url,
        json=dict(credential=json.dumps(SIGNIN_DATA1)),
    )
    assert response.status_code == 200
    assert response.json["response"]["user"]["email"] == "matt@lp.com"


def test_usage(app, client, get_message):
    authenticate(client)
    register_options, response_url = _register_start_json(client, usage="secondary")
    response = client.post(response_url, json=dict(credential=json.dumps(REG_DATA1)))
    assert response.status_code == 200
    logout(client)

    signin_options, response_url, _ = _signin_start_json(client, "matt@lp.com")
    response = client.post(
        response_url,
        json=dict(credential=json.dumps(SIGNIN_DATA1)),
    )
    assert response.status_code == 400
    assert response.json["response"]["errors"][0].encode("utf-8") == get_message(
        "WEBAUTHN_CREDENTIAL_WRONG_USAGE"
    )


def test_constraints(app, clients, get_message):
    """Test that nickname is unique for a given user but different users
    can have the same nickname.
    Also that credential_id is unique across the app.
    """
    authenticate(clients)
    register_options, response_url = _register_start_json(clients, name="testr3")
    response = clients.post(response_url, json=dict(credential=json.dumps(REG_DATA1)))
    assert response.status_code == 200

    # register same name again
    response = clients.post("wan-register", json=dict(name="testr3"))
    assert response.status_code == 400
    assert response.json["response"]["errors"][0].encode("utf-8") == get_message(
        "WEBAUTHN_NAME_INUSE", name="testr3"
    )

    logout(clients)

    # Different user - should get credential id in use error
    authenticate(clients, email="joe@lp.com")
    register_options, response_url = _register_start_json(clients, name="testr3")
    response = clients.post(response_url, json=dict(credential=json.dumps(REG_DATA2)))
    assert response.status_code == 200

    # Try to register with identical credential ID as other user
    register_options, response_url = _register_start_json(clients, name="testr4")
    response = clients.post(response_url, json=dict(credential=json.dumps(REG_DATA1)))
    assert response.status_code == 400
    assert response.json["response"]["errors"][0].encode("utf-8") == get_message(
        "WEBAUTHN_CREDENTIAL_ID_INUSE"
    )


def test_bad_data_register(app, client, get_message):
    authenticate(client)
    register_options, response_url = _register_start_json(client, name="testr3")

    # first try mangling json - should get API_ERROR
    response = client.post(response_url, json=dict(credential='"hi there"'))
    assert response.status_code == 400
    assert response.json["response"]["errors"][0].encode("utf-8") == get_message(
        "API_ERROR"
    )
    response = client.post(response_url, json=dict(credential=""))
    assert response.status_code == 400
    assert response.json["response"]["errors"][0].encode("utf-8") == get_message(
        "API_ERROR"
    )

    # Now pass incorrect keys
    bad_register = copy.deepcopy(REG_DATA1)
    del bad_register["rawId"]
    response = client.post(response_url, json=dict(credential=json.dumps(bad_register)))
    assert response.status_code == 400
    assert response.json["response"]["errors"][0].encode("utf-8") == get_message(
        "API_ERROR"
    )

    # now muck with attestation - should get VERIFY ERROR
    bad_register = copy.deepcopy(REG_DATA1)
    bad_register["rawId"] = "unknown"
    response = client.post(response_url, json=dict(credential=json.dumps(bad_register)))
    assert response.status_code == 400
    assert response.json["response"]["errors"][0].encode("utf-8") == get_message(
        "WEBAUTHN_NO_VERIFY", cause="id and raw_id were not equivalent"
    )

    # same with forms
    with capture_flashes() as flashes:
        response = client.post(
            response_url,
            data=dict(credential=json.dumps(bad_register)),
            follow_redirects=False,
        )
        assert response.status_code == 302
        assert "/wan-register" in response.location
    assert flashes[0]["category"] == "error"
    assert flashes[0]["message"].encode("utf-8") == get_message(
        "WEBAUTHN_NO_VERIFY", cause="id and raw_id were not equivalent"
    )


def test_bad_data_signin(app, client, get_message):
    authenticate(client)
    register_options, response_url = _register_start_json(client, usage="first")
    response = client.post(response_url, json=dict(credential=json.dumps(REG_DATA1)))
    assert response.status_code == 200

    logout(client)
    signin_options, response_url, _ = _signin_start_json(client, "matt@lp.com")
    response = client.post(response_url, json=dict(credential='"hi there"'))
    assert response.status_code == 400
    assert response.json["response"]["errors"][0].encode("utf-8") == get_message(
        "API_ERROR"
    )
    response = client.post(response_url, json=dict(credential=""))
    assert response.status_code == 400
    assert response.json["response"]["errors"][0].encode("utf-8") == get_message(
        "API_ERROR"
    )

    # Now pass incorrect keys
    bad_signin = copy.deepcopy(SIGNIN_DATA1)
    del bad_signin["rawId"]
    response = client.post(response_url, json=dict(credential=json.dumps(bad_signin)))
    assert response.status_code == 400
    assert response.json["response"]["errors"][0].encode("utf-8") == get_message(
        "API_ERROR"
    )

    # now muck with attestation - should get VERIFY ERROR
    bad_signin = copy.deepcopy(SIGNIN_DATA1)
    bad_signin["response"]["signature"] = bad_signin["response"]["signature"].replace(
        "M", "N"
    )
    response = client.post(response_url, json=dict(credential=json.dumps(bad_signin)))
    assert response.status_code == 400
    assert response.json["response"]["errors"][0].encode("utf-8") == get_message(
        "WEBAUTHN_NO_VERIFY", cause="Could not verify authentication signature"
    )


def test_delete(app, clients, get_message):
    @wan_deleted.connect_via(app)
    def pc(sender, user, name, **extra_args):
        assert name == "testr3"
        assert len(user.webauthn) == 1

    authenticate(clients)
    register_options, response_url = _register_start(clients, name="testr3")
    response = clients.post(
        response_url, data=dict(credential=json.dumps(REG_DATA1)), follow_redirects=True
    )
    assert response.status_code == 200
    assert get_message("WEBAUTHN_REGISTER_SUCCESSFUL", name="testr3") in response.data

    response = clients.get("/wan-register")
    assert b"testr3" in response.data

    # Make sure GET works - this is important if we get a freshness redirect when
    # attempting to delete - the verify endpoint will redirect back to here.
    response = clients.get("/wan-delete", follow_redirects=False)
    assert response.status_code == 302

    """
    response = clients.post("/wan-delete")
    assert get_message("WEBAUTHN_NAME_REQUIRED") in response.data
    """

    response = clients.post(
        "/wan-delete", data=dict(name="testr1"), follow_redirects=True
    )
    assert get_message("WEBAUTHN_NAME_NOT_FOUND", name="testr1") in response.data

    with capture_flashes() as flashes:
        response = clients.post(
            "/wan-delete", data=dict(name="testr3"), follow_redirects=True
        )
    assert flashes[0]["category"] == "info"
    assert flashes[0]["message"].encode("utf-8") == get_message(
        "WEBAUTHN_CREDENTIAL_DELETED", name="testr3"
    )
    response = clients.get("/wan-register")
    assert b"testr3" not in response.data


def test_delete_json(app, clients, get_message):
    headers = {"Accept": "application/json", "Content-Type": "application/json"}

    authenticate(clients)
    register_options, response_url = _register_start_json(clients, name="testr3")
    response = clients.post(response_url, json=dict(credential=json.dumps(REG_DATA1)))
    assert response.status_code == 200

    response = clients.get("/wan-register", headers=headers)
    active_creds = response.json["response"]["registered_credentials"]
    assert active_creds[0]["name"] == "testr3"

    response = clients.post("/wan-delete", json=dict())
    assert response.status_code == 400
    assert response.json["response"]["errors"][0].encode("utf=8") == get_message(
        "WEBAUTHN_NAME_REQUIRED"
    )

    response = clients.post("/wan-delete", json=dict(name="testr1"))
    assert response.status_code == 400
    assert response.json["response"]["errors"][0].encode("utf=8") == get_message(
        "WEBAUTHN_NAME_NOT_FOUND", name="testr1"
    )

    response = clients.post("/wan-delete", json=dict(name="testr3"))
    assert response.status_code == 200


def test_disabled_account(app, client, get_message):
    # With USER_HINTS enabled, should get 200 on initial signin POST, but
    # not receive a list of registered credentials.
    authenticate(client)

    register_options, response_url = _register_start_json(
        client, name="testr3", usage="first"
    )
    response = client.post(response_url, json=dict(credential=json.dumps(REG_DATA1)))
    assert response.status_code == 200
    logout(client)

    with app.test_request_context("/"):
        user = app.security.datastore.find_user(email="matt@lp.com")
        app.security.datastore.deactivate_user(user)
        app.security.datastore.commit()

    signin_options, response_url = _signin_start(client, "matt@lp.com")
    assert response.status_code == 200
    allow_credentials = signin_options["allowCredentials"]
    assert len(allow_credentials) == 0

    # Now set USER_HINTS false and should get 400 on second POST
    app.config["SECURITY_WAN_ALLOW_USER_HINTS"] = False

    # Identity should be ignored
    signin_options, response_url = _signin_start(client, "matt@lp.com")
    allow_credentials = signin_options["allowCredentials"]
    assert len(allow_credentials) == 0

    response = client.post(
        response_url,
        json=dict(credential=json.dumps(SIGNIN_DATA1)),
    )
    assert response.status_code == 400
    assert response.json["response"]["errors"][0].encode("utf-8") == get_message(
        "DISABLED_ACCOUNT"
    )


def test_unk_credid(app, client, get_message):
    authenticate(client)

    register_options, response_url = _register_start_json(
        client, name="testr3", usage="first"
    )
    response = client.post(response_url, json=dict(credential=json.dumps(REG_DATA1)))
    assert response.status_code == 200
    logout(client)

    signin_options, response_url, _ = _signin_start_json(client, "matt@lp.com")
    assert len(signin_options["allowCredentials"]) == 1

    bad_signin = copy.deepcopy(SIGNIN_DATA1)
    bad_signin["rawId"] = bad_signin["rawId"].replace("w", "d")

    response = client.post(
        response_url,
        json=dict(credential=json.dumps(bad_signin)),
    )
    assert response.status_code == 400
    assert response.json["response"]["field_errors"]["credential"][0].encode(
        "utf-8"
    ) == get_message("WEBAUTHN_UNKNOWN_CREDENTIAL_ID")

    # same with forms
    with capture_flashes() as flashes:
        response = client.post(
            response_url,
            data=dict(credential=json.dumps(bad_signin)),
            follow_redirects=False,
        )
        assert response.status_code == 302
        assert "/wan-signin" in response.location
    assert flashes[0]["category"] == "error"
    assert flashes[0]["message"].encode("utf-8") == get_message(
        "WEBAUTHN_UNKNOWN_CREDENTIAL_ID"
    )


@pytest.mark.settings(wan_allow_as_first_factor=False)
def test_no_first_factor(app, client, get_message):
    # make sure that is app not configured to allow a webauthn key as a 'first'
    # authenticator, that the endpoint 'disappears'.
    authenticate(client)

    register_options, response_url = _register_start_json(client, name="testr3")
    response = client.post(response_url, json=dict(credential=json.dumps(REG_DATA1)))
    assert response.status_code == 200
    logout(client)

    response = client.post("wan-signin", json=dict(identity="matt@lp.com"))
    assert response.status_code == 404


@pytest.mark.two_factor()
@pytest.mark.unified_signin()
def test_tf(app, client, get_message):
    # Test using webauthn key as a second factor
    # Register 2 keys - one "first" one "secondary"
    keys = reg_2_keys(client)
    logout(client)

    # log back in - should require MFA.
    response = client.post(
        "/us-signin",
        data=dict(identity="matt@lp.com", passcode="password", remember=True),
        follow_redirects=True,
    )
    assert response.status_code == 200
    assert b"Use Your WebAuthn Security Key as a Second Factor" in response.data
    # we should have a wan key available
    assert b'action="/wan-signin' in response.data

    # verify NOT logged in
    response = client.get("/profile", follow_redirects=False)
    assert "/login" in response.location

    signin_options, response_url = _signin_start(client, "matt@lp.com")
    assert len(signin_options["allowCredentials"]) == 1
    assert signin_options["allowCredentials"][0]["id"] == keys["secondary"]["id"]
    response = client.post(
        response_url,
        json=dict(credential=json.dumps(keys["secondary"]["signin"])),
    )
    assert response.status_code == 200

    # verify actually logged in
    response = client.get("/profile", follow_redirects=False)
    assert response.status_code == 200


@pytest.mark.two_factor()
@pytest.mark.unified_signin()
def test_tf_json(app, client, get_message):
    # Test using webauthn key as a second factor
    # Register 2 keys - one "first" one "secondary"
    keys = reg_2_keys(client)
    logout(client)

    # log back in - should require MFA.
    response = client.post(
        "/us-signin",
        json=dict(identity="matt@lp.com", passcode="password", remember=True),
    )
    assert response.status_code == 200
    assert response.json["response"]["tf_method"] == "webauthn"
    assert response.json["response"]["tf_required"]
    assert response.json["response"]["tf_state"] == "ready"
    assert response.json["response"]["tf_signin_url"] == "/wan-signin"

    # verify NOT logged in
    response = client.get("/profile", headers={"accept": "application/json"})
    assert response.status_code == 401

    # For secondary, identity is stored in session
    signin_options, response_url, _ = _signin_start_json(client, "")
    assert len(signin_options["allowCredentials"]) == 1
    assert signin_options["allowCredentials"][0]["id"] == keys["secondary"]["id"]
    response = client.post(
        response_url,
        json=dict(credential=json.dumps(keys["secondary"]["signin"])),
    )
    assert response.status_code == 200

    # verify actually logged in
    response = client.get("/profile", headers={"accept": "application/json"})
    assert response.status_code == 200


@pytest.mark.two_factor()
@pytest.mark.settings(two_factor_always_validate=False)
def test_tf_validity_window(app, client, get_message):
    # Test with a two-factor validity setting - we don't get re-prompted.
    authenticate(client)
    assert not client.get_cookie("tf_validity")
    register_options, response_url = _register_start_json(client)
    client.post(response_url, json=dict(credential=json.dumps(REG_DATA1)))
    logout(client)

    # login - should require second factor
    response = client.post(
        "/login",
        data=dict(email="matt@lp.com", password="password"),
        follow_redirects=True,
    )
    assert b"Use Your WebAuthn Security Key as a Second Factor" in response.data
    with client.session_transaction() as session:
        assert "tf_user_id" in session

    signin_options, response_url = _signin_start(client, "matt@lp.com")
    response = client.post(response_url, json=dict(credential=json.dumps(SIGNIN_DATA1)))
    assert response.status_code == 200

    # verify actually logged in
    response = client.get("/profile", follow_redirects=False)
    assert response.status_code == 200
    reset_signcount(app, "matt@lp.com", "testr1")
    logout(client)

    # since we didn't specify 'remember' previously - should still require 2FA
    response = client.post(
        "/login",
        data=dict(email="matt@lp.com", password="password", remember=True),
        follow_redirects=True,
    )
    assert b"Use Your WebAuthn Security Key as a Second Factor" in response.data

    signin_options, response_url = _signin_start(client, "matt@lp.com")
    response = client.post(response_url, json=dict(credential=json.dumps(SIGNIN_DATA1)))
    assert response.status_code == 200
    assert client.get_cookie("tf_validity")
    logout(client)

    # since we did specify 'remember' previously - should not require 2FA
    response = client.post(
        "/login",
        data=dict(email="matt@lp.com", password="password"),
        follow_redirects=True,
    )
    # verify actually logged in
    response = client.get("/profile", follow_redirects=False)
    assert response.status_code == 200

    # Since logged in all tf related attributes in session should be gone.
    assert not tf_in_session(get_existing_session(client))


@pytest.mark.two_factor()
@pytest.mark.settings(two_factor_always_validate=False)
def test_tf_validity_window_json(app, client, get_message):
    # Test with a two-factor validity setting - we don't get re-prompted.
    # This also relies on the tf_validity_cookie
    json_authenticate(client)
    register_options, response_url = _register_start_json(client)
    client.post(response_url, json=dict(credential=json.dumps(REG_DATA1)))
    logout(client)

    response = client.post(
        "/login", json=dict(email="matt@lp.com", password="password", remember=True)
    )
    assert response.status_code == 200
    assert response.json["response"]["tf_required"]

    signin_options, response_url = _signin_start(client, "matt@lp.com")
    response = client.post(response_url, json=dict(credential=json.dumps(SIGNIN_DATA1)))
    assert response.status_code == 200
    logout(client)

    # Sign in again - shouldn't require 2FA
    response = client.post(
        "/login",
        json=dict(
            email="matt@lp.com",
            password="password",
            remember=True,
        ),
    )
    assert response.status_code == 200
    response = client.get("/profile", follow_redirects=False)
    assert response.status_code == 200


@pytest.mark.settings(wan_register_within="1 seconds")
def test_register_timeout(app, client, get_message):
    authenticate(client)

    app.security.wan_serializer = FakeSerializer(1.0)
    register_options, response_url = _register_start_json(client, name="testr3")

    response = client.post(response_url, json=dict(credential=json.dumps(REG_DATA1)))
    assert response.status_code == 400
    assert response.json["response"]["errors"][0].encode("utf-8") == get_message(
        "WEBAUTHN_EXPIRED", within=app.config["SECURITY_WAN_REGISTER_WITHIN"]
    )


@pytest.mark.settings(wan_signin_within="2 seconds")
def test_signin_timeout(app, client, get_message):
    authenticate(client)

    register_options, response_url = _register_start_json(client, name="testr3")
    response = client.post(response_url, json=dict(credential=json.dumps(REG_DATA1)))
    assert response.status_code == 200
    logout(client)

    app.security.wan_serializer = FakeSerializer(2.0)
    signin_options, response_url, _ = _signin_start_json(client, "matt@lp.com")
    response = client.post(
        response_url,
        json=dict(credential=json.dumps(SIGNIN_DATA1)),
    )
    assert response.status_code == 400
    assert response.json["response"]["errors"][0].encode("utf-8") == get_message(
        "WEBAUTHN_EXPIRED", within=app.config["SECURITY_WAN_SIGNIN_WITHIN"]
    )


def test_bad_token(app, client, get_message):
    authenticate(client)

    response = client.post("/wan-register/not a token", json=dict())
    assert response.status_code == 400
    assert response.json["response"]["errors"][0].encode("utf-8") == get_message(
        "API_ERROR"
    )
    # same w/o json
    response = client.post(
        "/wan-register/not a token", data=dict(), follow_redirects=True
    )
    assert get_message("API_ERROR") in response.data
    response = client.post(
        "/wan-register/not a token", data=dict(), follow_redirects=False
    )
    assert "/wan-register" in response.location

    # Test wan-verify
    response = client.post("/wan-verify/not a token", json=dict())
    assert response.status_code == 400
    assert response.json["response"]["errors"][0].encode("utf-8") == get_message(
        "API_ERROR"
    )
    # same w/o json
    response = client.post(
        "/wan-verify/not a token", data=dict(), follow_redirects=True
    )
    assert get_message("API_ERROR") in response.data
    response = client.post(
        "/wan-verify/not a token", data=dict(), follow_redirects=False
    )
    assert "/wan-verify" in response.location

    # Test signin
    logout(client)

    response = client.post("/wan-signin/not a token", json=dict())
    assert response.status_code == 400
    assert response.json["response"]["errors"][0].encode("utf-8") == get_message(
        "API_ERROR"
    )
    # same w/o json
    response = client.post(
        "/wan-signin/not a token", data=dict(), follow_redirects=True
    )
    assert get_message("API_ERROR") in response.data
    response = client.post(
        "/wan-signin/not a token", data=dict(), follow_redirects=False
    )
    assert "/wan-signin" in response.location


@pytest.mark.settings(
    wan_register_template="custom_security/wan_register.html",
    wan_signin_template="custom_security/wan_signin.html",
    wan_verify_template="custom_security/wan_verify.html",
)
def test_wan_context_processors(client, app):
    @app.security.context_processor
    def default_ctx_processor():
        return {"global": "global"}

    @app.security.wan_register_context_processor
    def register_ctx():
        return {"foo": "register"}

    authenticate(client)

    response = client.get("wan-register")
    assert b"CUSTOM WAN REGISTER" in response.data
    assert b"global" in response.data
    assert b"register" in response.data

    response = client.post("wan-register", data=dict(name="matt@lp.com"))
    assert b"CUSTOM WAN REGISTER" in response.data
    assert b"global" in response.data
    assert b"register" in response.data
    logout(client)

    @app.security.wan_signin_context_processor
    def signin_ctx():
        return {"foo": "signin"}

    response = client.get("wan-signin")
    assert b"CUSTOM WAN SIGNIN" in response.data
    assert b"global" in response.data
    assert b"signin" in response.data

    response = client.post("wan-signin", data=dict(name="matt@lp.com"))
    assert b"CUSTOM WAN SIGNIN" in response.data
    assert b"global" in response.data
    assert b"signin" in response.data

    @app.security.wan_verify_context_processor
    def verify_ctx():
        return {"foo": "verify"}

    authenticate(client)
    response = client.get("wan-verify")
    assert b"CUSTOM WAN VERIFY" in response.data
    assert b"global" in response.data
    assert b"verify" in response.data

    response = client.post("wan-verify", data=dict(name="matt@lp.com"))
    assert b"CUSTOM WAN VERIFY" in response.data
    assert b"global" in response.data
    assert b"verify" in response.data


@pytest.mark.two_factor()
def test_alt_tf(app, client, get_message):
    # Use webauthn as primary and set up SMS as second factor
    authenticate(client)

    register_options, response_url = _register_start_json(client, usage="first")
    response = client.post(response_url, json=dict(credential=json.dumps(REG_DATA1)))
    assert response.status_code == 200

    sms_sender = setup_tf_sms(client)
    logout(client)

    # sign in using webauthn key
    signin_options, response_url, _ = _signin_start_json(client, "matt@lp.com")
    response = client.post(
        response_url,
        json=dict(credential=json.dumps(SIGNIN_DATA1)),
    )
    assert response.status_code == 200
    assert response.json["response"]["tf_required"]

    code = sms_sender.messages[0].split()[-1]
    response = client.post("/tf-validate", json=dict(code=code))
    assert response.status_code == 200
    # verify logged in
    response = client.get("/profile", follow_redirects=False)
    assert response.status_code == 200


@pytest.mark.two_factor()
def test_all_in_one(app, client, get_message):
    # Use a key that supports user_verification - we should be able to
    # use that alone.
    authenticate(client)
    register_options, response_url = _register_start_json(client, usage="first")
    response = client.post(response_url, json=dict(credential=json.dumps(REG_DATA_UV)))
    assert response.status_code == 200
    setup_tf_sms(client)

    logout(client)
    signin_options, response_url, rjson = _signin_start_json(client, "matt@lp.com")
    assert "user" not in rjson["response"]
    response = client.post(
        response_url, json=dict(credential=json.dumps(SIGNIN_DATA_UV))
    )
    assert response.json["response"]["user"]["email"] == "matt@lp.com"

    # verify actually logged in
    response = client.get("/profile", headers={"accept": "application/json"})
    assert response.status_code == 200


@pytest.mark.two_factor()
def test_all_in_one_not_allowed(app, client, get_message):
    # now test when we don't allow a key to satisfy both factors
    authenticate(client)
    register_options, response_url = _register_start_json(client, usage="first")
    response = client.post(response_url, json=dict(credential=json.dumps(REG_DATA_UV)))
    assert response.status_code == 200
    setup_tf_sms(client)
    logout(client)

    app.config["SECURITY_WAN_ALLOW_AS_MULTI_FACTOR"] = False
    signin_options, response_url, rjson = _signin_start_json(client, "matt@lp.com")
    assert "user" not in rjson["response"]

    response = client.post(
        response_url, json=dict(credential=json.dumps(SIGNIN_DATA_UV))
    )
    assert response.json["response"]["tf_required"]


def test_reset(app, client):
    headers = {"Accept": "application/json", "Content-Type": "application/json"}

    authenticate(client)
    register_options, response_url = _register_start_json(client)
    client.post(response_url, json=dict(credential=json.dumps(REG_DATA_UV)))

    response = client.get("/wan-register", headers=headers)
    active_creds = response.json["response"]["registered_credentials"]
    assert active_creds[0]["name"] == "testr1"

    with app.test_request_context("/"):
        user = app.security.datastore.find_user(email="matt@lp.com")
        app.security.datastore.webauthn_reset(user)
        app.security.datastore.commit()

    response = client.get("/wan-register", headers=headers)
    active_creds = response.json["response"]["registered_credentials"]
    assert len(active_creds) == 0


def test_user_handle(app, clients, get_message):
    """Test that we fail signin if user_handle doesn't match.
    Since we generated the SIGNIN_DATA_OH from view_scaffold - the user_handle
    has no way of matching.
    """
    authenticate(clients)
    register_options, response_url = _register_start_json(clients, usage="first")
    response = clients.post(response_url, json=dict(credential=json.dumps(REG_DATA_UH)))
    assert response.status_code == 200

    # verify can't sign in
    logout(clients)
    signin_options, response_url, _ = _signin_start_json(clients, "matt@lp.com")
    response = clients.post(
        response_url, json=dict(credential=json.dumps(SIGNIN_DATA_UH))
    )
    assert response.json["response"]["field_errors"]["credential"][0].encode(
        "utf-8"
    ) == get_message("WEBAUTHN_MISMATCH_USER_HANDLE")

    # Now change the user_handle both for the user and SIGNIN_DATA_UH
    with app.test_request_context("/"):
        user = app.security.datastore.find_user(email="matt@lp.com")
        app.security.datastore.set_webauthn_user_handle(user)
        app.security.datastore.commit()
        b64_user_handle = urlsafe_b64encode(
            user.fs_webauthn_user_handle.encode()
        ).decode("utf-8")
    upd_signin_data = copy.deepcopy(SIGNIN_DATA_UH)
    upd_signin_data["response"]["userHandle"] = b64_user_handle
    signin_options, response_url, _ = _signin_start_json(clients, "matt@lp.com")
    response = clients.post(
        response_url, json=dict(credential=json.dumps(upd_signin_data))
    )
    # verify actually logged in
    response = clients.get("/profile", headers={"accept": "application/json"})
    assert response.status_code == 200


def test_autogen_user_handle(app, client, get_message):
    # Test that is an existing user doesn't have a fs_webauthn_user_handle - it will
    # be generated.
    with app.test_request_context("/"):
        user = app.security.datastore.find_user(email="matt@lp.com")
        user.fs_webauthn_user_handle = None
        app.security.datastore.put(user)
        app.security.datastore.commit()

    authenticate(client)
    register_options, response_url = _register_start_json(client, usage="first")
    with app.test_request_context("/"):
        user = app.security.datastore.find_user(email="matt@lp.com")
        assert user.fs_webauthn_user_handle
        b64_user_handle = (
            urlsafe_b64encode(user.fs_webauthn_user_handle.encode())
            .decode("utf-8")
            .replace("=", "")
        )
        assert b64_user_handle == register_options["user"]["id"]


def test_verify_json(app, client, get_message):
    # Test can re-authenticate using existing webauthn key.
    headers = {"Accept": "application/json", "Content-Type": "application/json"}

    authenticate(client)
    register_options, response_url = _register_start_json(client, usage="first")
    response = client.post(response_url, json=dict(credential=json.dumps(REG_DATA1)))
    assert response.status_code == 200

    reset_fresh(client, app.config["SECURITY_FRESHNESS"])

    response = client.get("fresh", headers=headers)
    assert response.status_code == 401
    assert response.json["response"]["reauth_required"]
    assert response.json["response"]["has_webauthn_verify_credential"]

    response = client.get("wan-verify", headers=headers)
    assert response.status_code == 200

    response = client.post("wan-verify", json=dict())
    # default webauthn_utils will set userVerification to discouraged in the
    # case of verify.
    signin_options = response.json["response"]["credential_options"]
    assert signin_options["userVerification"] == "discouraged"

    response_url = (
        f'wan-verify/{response.json["response"]["wan_state"]}?include_auth_token'
    )
    response = client.post(response_url, json=dict(credential=json.dumps(SIGNIN_DATA1)))
    assert response.status_code == 200
    verify_token(client, response.json["response"]["user"]["authentication_token"])

    response = client.get("fresh", headers=headers)
    assert response.status_code == 200


def test_verify(app, client, get_message):
    # Test can re-authenticate using existing webauthn key.
    # Forms version - verify that the 'next' qparam is properly maintained during the
    # 2 part authentication.
    authenticate(client)
    register_options, response_url = _register_start(client, usage="first")
    response = client.post(
        response_url, data=dict(credential=json.dumps(REG_DATA1)), follow_redirects=True
    )
    assert response.status_code == 200
    assert b"testr1" in response.data

    old_paa = reset_fresh(client, app.config["SECURITY_FRESHNESS"])
    response = client.get("fresh")
    assert response.location == "/verify?next=/fresh"
    signin_options, response_url = _signin_start(
        client, endpoint="wan-verify?next=/fresh"
    )

    response = client.post(
        response_url,
        data=dict(credential=json.dumps(SIGNIN_DATA1)),
        follow_redirects=False,
    )
    assert "/fresh" in response.location
    with client.session_transaction() as sess:
        assert sess["fs_paa"] > old_paa


@pytest.mark.settings(wan_signin_within="2 seconds")
def test_verify_timeout(app, client, get_message):
    authenticate(client)
    register_options, response_url = _register_start_json(client, name="testr3")
    response = client.post(response_url, json=dict(credential=json.dumps(REG_DATA1)))
    assert response.status_code == 200

    app.security.wan_serializer = FakeSerializer(2.0)
    response = client.post("wan-verify", json=dict())
    response_url = f'wan-verify/{response.json["response"]["wan_state"]}'
    response = client.post(response_url, json=dict(credential=json.dumps(SIGNIN_DATA1)))
    assert response.status_code == 400
    assert response.json["response"]["errors"][0].encode("utf-8") == get_message(
        "WEBAUTHN_EXPIRED", within=app.config["SECURITY_WAN_SIGNIN_WITHIN"]
    )


def test_verify_validate_error(app, client, get_message):
    authenticate(client)
    register_options, response_url = _register_start_json(client, name="testr3")
    response = client.post(response_url, json=dict(credential=json.dumps(REG_DATA1)))
    assert response.status_code == 200

    response = client.post("wan-verify", json=dict())
    response_url = f'wan-verify/{response.json["response"]["wan_state"]}'
    # send wrong signin data
    response = client.post(
        response_url, json=dict(credential=json.dumps(SIGNIN_DATA_UH))
    )
    assert response.status_code == 400
    assert response.json["response"]["field_errors"]["credential"][0].encode(
        "utf-8"
    ) == get_message("WEBAUTHN_UNKNOWN_CREDENTIAL_ID")

    # same thing - with forms - this should redirect to wan-verify and flash a message
    with capture_flashes() as flashes:
        response = client.post(
            response_url,
            data=dict(credential=json.dumps(SIGNIN_DATA_UH)),
            follow_redirects=False,
        )
        assert response.status_code == 302
        assert "/wan-verify" in response.location
    assert flashes[0]["category"] == "error"
    assert flashes[0]["message"].encode("utf-8") == get_message(
        "WEBAUTHN_UNKNOWN_CREDENTIAL_ID"
    )


@pytest.mark.settings(wan_allow_as_verify=None)
def test_no_verify(app, client):
    authenticate(client)
    response = client.get("/wan-verify")
    assert response.status_code == 404


def test_verify_usage_any_json(app, client, get_message):
    # Test the WAN_ALLOW_AS_VERIFY config.
    # Make sure only allowed credentials show up as options
    # Make sure if we use a disallowed credential, we get an error.
    keys = reg_2_keys(client)

    # Default WAN_ALLOW_AS_VERIFY is ["first", "secondary"]
    response = client.post("wan-verify", json=dict())
    response_url = f'wan-verify/{response.json["response"]["wan_state"]}'
    allow_credentials = response.json["response"]["credential_options"][
        "allowCredentials"
    ]
    assert len(allow_credentials) == 2

    # make sure can sign in with either
    response = client.post(
        response_url, json=dict(credential=json.dumps(keys["first"]["signin"]))
    )
    assert response.status_code == 200
    response = client.post(
        response_url, json=dict(credential=json.dumps(keys["secondary"]["signin"]))
    )
    assert response.status_code == 200


@pytest.mark.settings(wan_allow_as_verify="first")
def test_verify_usage_first_json(app, client, get_message):
    # Test the WAN_ALLOW_AS_VERIFY config.
    # Make sure only allowed credentials show up as options
    # Make sure if we use a disallowed credential, we get an error.
    keys = reg_2_keys(client)

    response = client.post("wan-verify", json=dict())
    response_url = f'wan-verify/{response.json["response"]["wan_state"]}'
    allow_credentials = response.json["response"]["credential_options"][
        "allowCredentials"
    ]
    assert len(allow_credentials) == 1
    assert allow_credentials[0]["id"] == keys["first"]["id"]

    # make sure can sign in with just "first"
    response = client.post(
        response_url, json=dict(credential=json.dumps(keys["first"]["signin"]))
    )
    assert response.status_code == 200
    # but not "secondary"
    response = client.post(
        response_url, json=dict(credential=json.dumps(keys["secondary"]["signin"]))
    )
    assert response.status_code == 400
    assert response.json["response"]["errors"][0].encode("utf-8") == get_message(
        "WEBAUTHN_CREDENTIAL_WRONG_USAGE"
    )


@pytest.mark.settings(wan_allow_as_verify="secondary")
def test_verify_usage_secondary_json(app, client, get_message):
    # Test the WAN_ALLOW_AS_VERIFY config.
    # Make sure only allowed credentials show up as options
    # Make sure if we use a disallowed credential, we get an error.
    keys = reg_2_keys(client)

    response = client.post("wan-verify", json=dict())
    response_url = f'wan-verify/{response.json["response"]["wan_state"]}'
    allow_credentials = response.json["response"]["credential_options"][
        "allowCredentials"
    ]
    assert len(allow_credentials) == 1
    assert allow_credentials[0]["id"] == keys["secondary"]["id"]

    # make sure can sign in with just "secondary"
    response = client.post(
        response_url, json=dict(credential=json.dumps(keys["first"]["signin"]))
    )
    assert response.status_code == 400
    assert response.json["response"]["errors"][0].encode("utf-8") == get_message(
        "WEBAUTHN_CREDENTIAL_WRONG_USAGE"
    )
    response = client.post(
        response_url, json=dict(credential=json.dumps(keys["secondary"]["signin"]))
    )
    assert response.status_code == 200


def test_remember_token(client):
    # test that remember token properly set on primary authn with webauthn
    authenticate(client)
    register_options, response_url = _register_start_json(
        client, name="testr3", usage="first"
    )
    response = client.post(response_url, json=dict(credential=json.dumps(REG_DATA1)))
    assert response.status_code == 200
    logout(client)

    assert not client.get_cookie("remember_token")

    headers = {"Accept": "application/json", "Content-Type": "application/json"}

    response = client.post(
        "wan-signin", headers=headers, json=dict(identity="matt@lp.com", remember=True)
    )
    response_url = f'wan-signin/{response.json["response"]["wan_state"]}'
    assert response.json["response"]["remember"]

    response = client.post(
        response_url,
        json=dict(credential=json.dumps(SIGNIN_DATA1), remember=True),
    )
    assert client.get_cookie("remember_token")
    client.delete_cookie("session")
    response = client.get("/profile")
    assert b"profile" in response.data


@pytest.mark.two_factor()
@pytest.mark.unified_signin()
def test_remember_token_tf(client):
    # test that remember token properly set after secondary authn with webauthn
    authenticate(client)
    register_options, response_url = _register_start_json(client, name="testr3")
    response = client.post(response_url, json=dict(credential=json.dumps(REG_DATA1)))
    assert response.status_code == 200
    logout(client)

    assert not client.get_cookie("remember_token")

    # login again - should require MFA
    response = client.post(
        "/us-signin",
        json=dict(identity="matt@lp.com", passcode="password", remember=True),
    )
    assert response.status_code == 200
    assert response.json["response"]["tf_method"] == "webauthn"
    assert response.json["response"]["tf_required"]
    with client.session_transaction() as session:
        assert session["tf_remember_login"]

    signin_options, response_url, _ = _signin_start_json(client, "matt@lp.com")
    response = client.post(
        response_url,
        json=dict(credential=json.dumps(SIGNIN_DATA1), remember=True),
    )
    assert client.get_cookie("remember_token")
    client.delete_cookie("session")
    response = client.get("/profile")
    assert b"profile" in response.data


@pytest.mark.settings(
    wan_post_register_view="/post_register",
)
def test_post_register_redirect(app, client, get_message):
    authenticate(client)

    register_options, response_url = _register_start_json(client, name="testr3")
    response = client.post(
        response_url,
        data=dict(credential=json.dumps(REG_DATA1)),
        follow_redirects=False,
    )
    assert response.status_code == 302
    assert "/post_register" in response.location


class MyWebauthnUtil(HackWebauthnUtil):
    def user_verification(self, user, usage):
        from webauthn.helpers.structs import UserVerificationRequirement

        return UserVerificationRequirement.REQUIRED


@pytest.mark.two_factor()
@pytest.mark.unified_signin()
@pytest.mark.webauthn(webauthn_util_cls=MyWebauthnUtil)
def test_uv_required(client):
    # Override WebauthnUtils to require user-verification on signin.
    keys = reg_2_keys(client)
    logout(client)

    # log back in - should require MFA.
    response = client.post(
        "/us-signin",
        json=dict(identity="matt@lp.com", passcode="password", remember=True),
    )
    assert response.status_code == 200
    assert response.json["response"]["tf_required"]

    # since we always REQUIRE user_verification in our WebauthUtil this should fail
    signin_options, response_url, _ = _signin_start_json(client, "")
    response = client.post(
        response_url,
        json=dict(credential=json.dumps(keys["secondary"]["signin"])),
    )
    assert response.status_code == 400
    assert (
        "User verification is required"
        in response.json["response"]["field_errors"]["credential"][0]
    )

    logout(client)

    # Try signing in with 'first' WebAuthn key - this DOES have UV set so should work.
    signin_options, response_url, _ = _signin_start_json(client, "")
    response = client.post(
        response_url,
        json=dict(credential=json.dumps(keys["first"]["signin"])),
    )
    assert response.status_code == 200
    assert response.json["response"]["user"]["email"] == "matt@lp.com"


@pytest.mark.settings(multi_factor_recovery_codes=True)
def test_mf(client):
    # Test using recovery codes in-liu of a webauthn second factor
    # Note that we are allowed to generate recovery codes even if we don't yet have
    # an established 2nd factor
    headers = {"Accept": "application/json", "Content-Type": "application/json"}
    authenticate(client)
    response = client.post("/mf-recovery-codes", headers=headers)
    codes = response.json["response"]["recovery_codes"]
    assert len(codes) == 5

    # setup webauthn
    register_options, response_url = _register_start_json(client, name="testr3")
    response = client.post(response_url, json=dict(credential=json.dumps(REG_DATA1)))
    assert response.status_code == 200
    logout(client)

    response = client.post(
        "/login", json=dict(email="matt@lp.com", password="password")
    )
    assert response.status_code == 200
    assert response.json["response"]["tf_required"]

    # oh no - we forgot our webauthn key
    """ Right now tf-rescue is part of TWO_FACTOR - not WEBAUTHN
    response = client.get("/tf-rescue", headers=headers)
    options = response.json["response"]["recovery_options"]
    assert "recovery_code" in options.keys()
    assert "/mf-recovery" in options["recovery_code"]
    """

    response = client.post(
        "/mf-recovery", data=dict(code=codes[0]), follow_redirects=True
    )
    assert response.status_code == 200

    # verify actually logged in
    response = client.get("/profile", follow_redirects=False)
    assert response.status_code == 200


@pytest.mark.settings(url_prefix="/auth")
def test_login_next(app, client, get_message):
    # Test that ?next=/xx is propagated through login/wan-signin templates as well as
    # views.
    # Also - use a different blueprint prefix - we rarely test that....
    authenticate(client, endpoint="/auth/login")
    register_options, response_url = _register_start(
        client, name="testr3", usage="first", endpoint="/auth/wan-register"
    )
    response = client.post(
        response_url, data=dict(credential=json.dumps(REG_DATA1)), follow_redirects=True
    )
    assert response.status_code == 200
    assert get_message("WEBAUTHN_REGISTER_SUCCESSFUL", name="testr3") in response.data
    logout(client, endpoint="/auth/logout")

    response = client.get("profile", follow_redirects=True)
    assert "?next=/profile" in response.request.url
    # pull webauthn form action out of login_form - should have ?next=...
    webauthn_url = get_form_action(response, 1)

    signin_options, response_url = _signin_start(
        client, "matt@lp.com", endpoint=webauthn_url
    )
    response = client.post(
        response_url,
        data=dict(credential=json.dumps(SIGNIN_DATA1)),
        follow_redirects=True,
    )
    assert response.status_code == 200
    assert b"Profile Page" in response.data

    # Try form.next
    logout(client, endpoint="/auth/logout")
    reset_signcount(app, "matt@lp.com", "testr3")

    response = client.post(
        "/auth/wan-signin", data=dict(identity="matt@lp.com", next="/im-in")
    )
    response_url = get_form_action(response)

    next_loc = get_form_input_value(response, "next")
    response = client.post(
        response_url,
        data=dict(credential=json.dumps(SIGNIN_DATA1), next=next_loc),
        follow_redirects=False,
    )
    assert "/im-in" in response.location


@pytest.mark.flask_async()
def test_async(app, client, get_message):
    auths = []

    @user_authenticated.connect_via(app)
    async def authned(myapp, user, **extra_args):
        auths.append((user.email, extra_args["authn_via"]))

    @wan_registered.connect_via(app)
    async def pc(sender, user, name, **extra_args):
        assert name == "testr1"
        assert len(user.webauthn) == 1

    @wan_deleted.connect_via(app)
    async def wan_delete(sender, user, name, **extra_args):
        assert name == "testr1"

    authenticate(client)

    register_options, response_url = _register_start(client, usage="first")
    response = client.post(
        response_url, data=dict(credential=json.dumps(REG_DATA1)), follow_redirects=True
    )
    assert response.status_code == 200

    # sign in - simple case use identity so we get back allowCredentials
    logout(client)
    signin_options, response_url = _signin_start(client, "matt@lp.com")
    response = client.post(
        response_url,
        data=dict(credential=json.dumps(SIGNIN_DATA1)),
        follow_redirects=True,
    )
    assert response.status_code == 200
    assert b"Welcome matt@lp.com" in response.data
    assert len(auths) == 2
    assert auths[1][1] == ["webauthn"]

    # test delete signal
    response = client.post(
        "/wan-delete", data=dict(name="testr1"), follow_redirects=True
    )


@pytest.mark.csrf()
@pytest.mark.settings(
    wan_post_register_view="/done-registration",
    post_login_view="/post-login",
)
def test_csrf(app, client, get_message):
    response = client.get("/login")
    csrf_token = get_form_input_value(response, "csrf_token")
    authenticate(client, csrf=True)

    register_options, response_url = _register_start(
        client, usage="first", csrf_token=csrf_token
    )
    data = dict(credential=json.dumps(REG_DATA1))
    response = client.post(response_url, data=data, follow_redirects=True)
    assert (
        b"The CSRF token is missing" in response.data
    )  # this should have been flashed

    data["csrf_token"] = csrf_token
    response = client.post(response_url, data=data)
    assert check_location(app, response.location, "/done-registration")
    logout(client)

    # use old csrf_token - should fail and we should get the error in the template
    response = client.post(
        "wan-signin", data=dict(identity="matt@lp.com", csrf_token=csrf_token)
    )
    assert b"The CSRF tokens do not match." in response.data

    response = client.get("/wan-signin")
    csrf_token = get_form_input_value(response, "csrf_token")
    signin_options, response_url = _signin_start(
        client, "matt@lp.com", csrf_token=csrf_token
    )
    data = dict(credential=json.dumps(SIGNIN_DATA1))
    response = client.post(response_url, data=data, follow_redirects=True)
    assert (
        b"The CSRF token is missing" in response.data
    )  # this should have been flashed

    data["csrf_token"] = csrf_token
    response = client.post(response_url, data=data)
    assert check_location(app, response.location, "/post-login")