#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Copyright 2019 SoloKeys Developers
#
# Licensed under the Apache License, Version 2.0, <LICENSE-APACHE or
# http://apache.org/licenses/LICENSE-2.0> or the MIT license <LICENSE-MIT or
# http://opensource.org/licenses/MIT>, at your option. This file may not be
# copied, modified, or distributed except according to those terms.
#

# Script for testing correctness of CTAP2/CTAP1 security token

from __future__ import print_function, absolute_import, unicode_literals
import sys, os, time, math
from random import randint
from binascii import hexlify
import array, struct, socket

from fido2.hid import CtapHidDevice, CTAPHID
from fido2.client import Fido2Client, ClientError
from fido2.ctap import CtapError
from fido2.ctap1 import CTAP1, ApduError, APDU
from fido2.ctap2 import ES256, PinProtocolV1
from fido2.utils import Timeout, sha256, hmac_sha256
from fido2.attestation import Attestation

from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes

from solo.fido2 import force_udp_backend
from solo.client import SoloClient


# Set up a FIDO 2 client using the origin https://example.com


def ForceU2F(client, device):
    client.ctap = CTAP1(device)
    client.pin_protocol = None
    client._do_make_credential = client._ctap1_make_credential
    client._do_get_assertion = client._ctap1_get_assertion


def VerifyAttestation(attest, data):
    verifier = Attestation.for_type(attest.fmt)
    verifier().verify(attest.att_statement, attest.auth_data, data.hash)


def shannon_entropy(data):
    sum = 0.0
    total = len(data)
    for x in range(0, 256):
        freq = data.count(x)
        p = freq / total
        if p > 0:
            sum -= p * math.log2(p)
    return sum


class Packet(object):
    def __init__(self, data):
        l = len(data)
        self.data = data

    def ToWireFormat(self,):
        return self.data

    @staticmethod
    def FromWireFormat(pkt_size, data):
        return Packet(data)


class Test:
    def __init__(self, msg):
        self.msg = msg

    def __enter__(self,):
        print(self.msg)

    def __exit__(self, a, b, c):
        print("Pass")


class Tester:
    def __init__(self,):
        self.origin = "https://examplo.org"
        self.host = "examplo.org"
        self.user_count = 10
        self.is_sim = False

    def find_device(self,):
        print(list(CtapHidDevice.list_devices()))
        dev = next(CtapHidDevice.list_devices(), None)
        if not dev:
            raise RuntimeError("No FIDO device found")
        self.dev = dev
        self.client = Fido2Client(dev, self.origin)
        self.ctap = self.client.ctap2
        self.ctap1 = CTAP1(dev)

        # consume timeout error
        # cmd,resp = self.recv_raw()

    def set_user_count(self, count):
        self.user_count = count

    def set_sim(self, b):
        self.is_sim = b

    def send_data(self, cmd, data):
        if type(data) != type(b""):
            data = struct.pack("%dB" % len(data), *[ord(x) for x in data])
        with Timeout(1.0) as event:
            return self.dev.call(cmd, data, event)

    def send_raw(self, data, cid=None):
        if cid is None:
            cid = self.dev._dev.cid
        elif type(cid) != type(b""):
            cid = struct.pack("%dB" % len(cid), *[ord(x) for x in cid])
        if type(data) != type(b""):
            data = struct.pack("%dB" % len(data), *[ord(x) for x in data])
        data = cid + data
        l = len(data)
        if l != 64:
            pad = "\x00" * (64 - l)
            pad = struct.pack("%dB" % len(pad), *[ord(x) for x in pad])
            data = data + pad
        data = list(data)
        assert len(data) == 64
        self.dev._dev.InternalSendPacket(Packet(data))

    def send_magic_reboot(self,):
        """
        For use in simulation and testing.  Random bytes that authentictor should detect
        and then restart itself.
        """
        magic_cmd = (
            b"\xac\x10\x52\xca\x95\xe5\x69\xde\x69\xe0\x2e\xbf"
            + b"\xf3\x33\x48\x5f\x13\xf9\xb2\xda\x34\xc5\xa8\xa3"
            + b"\x40\x52\x66\x97\xa9\xab\x2e\x0b\x39\x4d\x8d\x04"
            + b"\x97\x3c\x13\x40\x05\xbe\x1a\x01\x40\xbf\xf6\x04"
            + b"\x5b\xb2\x6e\xb7\x7a\x73\xea\xa4\x78\x13\xf6\xb4"
            + b"\x9a\x72\x50\xdc"
        )
        self.dev._dev.InternalSendPacket(Packet(magic_cmd))

    def cid(self,):
        return self.dev._dev.cid

    def set_cid(self, cid):
        if type(cid) not in [type(b""), type(bytearray())]:
            cid = struct.pack("%dB" % len(cid), *[ord(x) for x in cid])
        self.dev._dev.cid = cid

    def recv_raw(self,):
        with Timeout(1.0) as t:
            cmd, payload = self.dev._dev.InternalRecv()
        return cmd, payload

    def check_error(self, data, err=None):
        assert len(data) == 1
        if err is None:
            if data[0] != 0:
                raise CtapError(data[0])
        elif data[0] != err:
            raise ValueError("Unexpected error: %02x" % data[0])

    def testFunc(self, func, test, *args, **kwargs):
        with Test(test):
            res = None
            expectedError = kwargs.get("expectedError", None)
            otherArgs = kwargs.get("other", {})
            try:
                res = func(*args, **otherArgs)
                if expectedError != CtapError.ERR.SUCCESS:
                    raise RuntimeError("Expected error to occur for test: %s" % test)
            except CtapError as e:
                if expectedError is not None:
                    if e.code != expectedError:
                        raise RuntimeError(
                            "Got error code 0x%x, expected %x" % (e.code, expectedError)
                        )
                else:
                    print(e)
        return res

    def testReset(self,):
        print("Resetting Authenticator...")
        self.ctap.reset()

    def testMC(self, test, *args, **kwargs):
        return self.testFunc(self.ctap.make_credential, test, *args, **kwargs)

    def testGA(self, test, *args, **kwargs):
        return self.testFunc(self.ctap.get_assertion, test, *args, **kwargs)

    def testCP(self, test, *args, **kwargs):
        return self.testFunc(self.ctap.client_pin, test, *args, **kwargs)

    def testPP(self, test, *args, **kwargs):
        return self.testFunc(
            self.client.pin_protocol.get_pin_token, test, *args, **kwargs
        )

    def test_long_ping(self):
        amt = 1000
        pingdata = os.urandom(amt)
        with Test("Send %d byte ping" % amt):
            try:
                t1 = time.time() * 1000
                r = self.send_data(CTAPHID.PING, pingdata)
                t2 = time.time() * 1000
                delt = t2 - t1
                # if (delt < 140 ):
                # raise RuntimeError('Fob is too fast (%d ms)' % delt)
                if delt > 555 * (amt / 1000):
                    raise RuntimeError("Fob is too slow (%d ms)" % delt)
                if r != pingdata:
                    raise ValueError("Ping data not echo'd")
            except CtapError as e:
                raise RuntimeError("ping failed")

        sys.stdout.flush()

    def test_hid(self, check_timeouts=False):
        if check_timeouts:
            with Test("idle"):
                try:
                    cmd, resp = self.recv_raw()
                except socket.timeout:
                    pass

        with Test("init"):
            r = self.send_data(CTAPHID.INIT, "\x11\x11\x11\x11\x11\x11\x11\x11")

        with Test("100 byte ping"):
            pingdata = os.urandom(100)
            try:
                r = self.send_data(CTAPHID.PING, pingdata)
                if r != pingdata:
                    raise ValueError("Ping data not echo'd")
            except CtapError as e:
                print("100 byte Ping failed:", e)
                raise RuntimeError("ping failed")

        self.test_long_ping()

        with Test("Wink"):
            r = self.send_data(CTAPHID.WINK, "")

        with Test("CBOR msg with no data"):
            try:
                r = self.send_data(CTAPHID.CBOR, "")
                if len(r) > 1 or r[0] == 0:
                    raise RuntimeError("Cbor is supposed to have payload")
            except CtapError as e:
                assert e.code == CtapError.ERR.INVALID_LENGTH

        with Test("No data in U2F msg"):
            try:
                r = self.send_data(CTAPHID.MSG, "")
                print(hexlify(r))
                if len(r) > 2:
                    raise RuntimeError("MSG is supposed to have payload")
            except CtapError as e:
                assert e.code == CtapError.ERR.INVALID_LENGTH

        with Test("Use init command to resync"):
            r = self.send_data(CTAPHID.INIT, "\x11\x22\x33\x44\x55\x66\x77\x88")

        with Test("Invalid HID command"):
            try:
                r = self.send_data(0x66, "")
                raise RuntimeError("Invalid command did not return error")
            except CtapError as e:
                assert e.code == CtapError.ERR.INVALID_COMMAND

        with Test("Sending packet with too large of a length."):
            self.send_raw("\x81\x1d\xba\x00")
            cmd, resp = self.recv_raw()
            self.check_error(resp, CtapError.ERR.INVALID_LENGTH)

        r = self.send_data(CTAPHID.PING, "\x44" * 200)
        with Test("Sending packets that skip a sequence number."):
            self.send_raw("\x81\x04\x90")
            self.send_raw("\x00")
            self.send_raw("\x01")
            # skip 2
            self.send_raw("\x03")
            cmd, resp = self.recv_raw()
            self.check_error(resp, CtapError.ERR.INVALID_SEQ)

        with Test("Resync and send ping"):
            try:
                r = self.send_data(CTAPHID.INIT, "\x11\x22\x33\x44\x55\x66\x77\x88")
                pingdata = os.urandom(100)
                r = self.send_data(CTAPHID.PING, pingdata)
                if r != pingdata:
                    raise ValueError("Ping data not echo'd")
            except CtapError as e:
                raise RuntimeError("resync fail: ", e)

        with Test("Send ping and abort it"):
            self.send_raw("\x81\x04\x00")
            self.send_raw("\x00")
            self.send_raw("\x01")
            try:
                r = self.send_data(CTAPHID.INIT, "\x11\x22\x33\x44\x55\x66\x77\x88")
            except CtapError as e:
                raise RuntimeError("resync fail: ", e)

        with Test("Send ping and abort it with different cid, expect timeout"):
            oldcid = self.cid()
            newcid = "\x11\x22\x33\x44"
            self.send_raw("\x81\x10\x00")
            self.send_raw("\x00")
            self.send_raw("\x01")
            self.set_cid(newcid)
            self.send_raw(
                "\x86\x00\x08\x11\x22\x33\x44\x55\x66\x77\x88"
            )  # init from different cid
            print("wait for init response")
            cmd, r = self.recv_raw()  # init response
            assert cmd == 0x86
            self.set_cid(oldcid)
            if check_timeouts:
                # print('wait for timeout')
                cmd, r = self.recv_raw()  # timeout response
                assert cmd == 0xBF

        with Test("Test timeout"):
            self.send_data(CTAPHID.INIT, "\x11\x22\x33\x44\x55\x66\x77\x88")
            t1 = time.time() * 1000
            self.send_raw("\x81\x04\x00")
            self.send_raw("\x00")
            self.send_raw("\x01")
            cmd, r = self.recv_raw()  # timeout response
            t2 = time.time() * 1000
            delt = t2 - t1
            assert cmd == 0xBF
            assert r[0] == CtapError.ERR.TIMEOUT
            assert delt < 1000 and delt > 400

        with Test("Test not cont"):
            self.send_data(CTAPHID.INIT, "\x11\x22\x33\x44\x55\x66\x77\x88")
            self.send_raw("\x81\x04\x00")
            self.send_raw("\x00")
            self.send_raw("\x01")
            self.send_raw("\x81\x10\x00")  # init packet
            cmd, r = self.recv_raw()  # timeout response
            assert cmd == 0xBF
            assert r[0] == CtapError.ERR.INVALID_SEQ

        if check_timeouts:
            with Test("Check random cont ignored"):
                self.send_data(CTAPHID.INIT, "\x11\x22\x33\x44\x55\x66\x77\x88")
                self.send_raw("\x01\x10\x00")
                try:
                    cmd, r = self.recv_raw()  # timeout response
                except socket.timeout:
                    pass

        with Test("Check busy"):
            t1 = time.time() * 1000
            self.send_data(CTAPHID.INIT, "\x11\x22\x33\x44\x55\x66\x77\x88")
            oldcid = self.cid()
            newcid = "\x11\x22\x33\x44"
            self.send_raw("\x81\x04\x00")
            self.set_cid(newcid)
            self.send_raw("\x81\x04\x00")
            cmd, r = self.recv_raw()  # busy response
            t2 = time.time() * 1000
            assert t2 - t1 < 100
            assert cmd == 0xBF
            assert r[0] == CtapError.ERR.CHANNEL_BUSY

            self.set_cid(oldcid)
            cmd, r = self.recv_raw()  # timeout response
            assert cmd == 0xBF
            assert r[0] == CtapError.ERR.TIMEOUT

        with Test("Check busy interleaved"):
            cid1 = "\x11\x22\x33\x44"
            cid2 = "\x01\x22\x33\x44"
            self.set_cid(cid2)
            self.send_data(CTAPHID.INIT, "\x11\x22\x33\x44\x55\x66\x77\x88")
            self.set_cid(cid1)
            self.send_data(CTAPHID.INIT, "\x11\x22\x33\x44\x55\x66\x77\x88")
            self.send_raw("\x81\x00\x63")  # echo 99 bytes first channel

            self.set_cid(cid2)  # send ping on 2nd channel
            self.send_raw("\x81\x00\x63")
            time.sleep(0.1)
            self.send_raw("\x00")

            cmd, r = self.recv_raw()  # busy response

            self.set_cid(cid1)  # finish 1st channel ping
            self.send_raw("\x00")

            self.set_cid(cid2)

            assert cmd == 0xBF
            assert r[0] == CtapError.ERR.CHANNEL_BUSY

            self.set_cid(cid1)
            cmd, r = self.recv_raw()  # ping response
            assert cmd == 0x81
            assert len(r) == 0x63

        if check_timeouts:
            with Test("Test idle, wait for timeout"):
                sys.stdout.flush()
                try:
                    cmd, resp = self.recv_raw()
                except socket.timeout:
                    pass

        with Test("Test cid 0 is invalid"):
            self.set_cid("\x00\x00\x00\x00")
            self.send_raw(
                "\x86\x00\x08\x11\x22\x33\x44\x55\x66\x77\x88", cid="\x00\x00\x00\x00"
            )
            cmd, r = self.recv_raw()  # timeout
            assert cmd == 0xBF
            assert r[0] == CtapError.ERR.INVALID_CHANNEL

        with Test("Test invalid broadcast cid use"):
            self.set_cid("\xff\xff\xff\xff")
            self.send_raw(
                "\x81\x00\x08\x11\x22\x33\x44\x55\x66\x77\x88", cid="\xff\xff\xff\xff"
            )
            cmd, r = self.recv_raw()  # timeout
            assert cmd == 0xBF
            assert r[0] == CtapError.ERR.INVALID_CHANNEL

    def test_u2f(self,):
        chal = sha256(b"AAA")
        appid = sha256(b"BBB")
        lastc = 0

        regs = []

        with Test("Check version"):
            assert self.ctap1.get_version() == "U2F_V2"

        with Test("Check bad INS"):
            try:
                res = self.ctap1.send_apdu(0, 0, 0, 0, b"")
            except ApduError as e:
                assert e.code == 0x6D00

        with Test("Check bad CLA"):
            try:
                res = self.ctap1.send_apdu(1, CTAP1.INS.VERSION, 0, 0, b"abc")
            except ApduError as e:
                assert e.code == 0x6E00

        for i in range(0, self.user_count):
            with Test(
                "U2F reg + auth %d/%d (count: %02x)" % (i + 1, self.user_count, lastc)
            ):
                reg = self.ctap1.register(chal, appid)
                reg.verify(appid, chal)
                auth = self.ctap1.authenticate(chal, appid, reg.key_handle)
                auth.verify(appid, chal, reg.public_key)

                regs.append(reg)
                # check endianness
                if lastc:
                    assert (auth.counter - lastc) < 10
                lastc = auth.counter
                if lastc > 0x80000000:
                    print("WARNING: counter is unusually high: %04x" % lastc)
                    assert 0

        for i in range(0, self.user_count):
            with Test(
                "Checking previous registration %d/%d" % (i + 1, self.user_count)
            ):
                auth = self.ctap1.authenticate(chal, appid, regs[i].key_handle)
                auth.verify(appid, chal, regs[i].public_key)

        print("Check that all previous credentials are registered...")
        for i in range(0, self.user_count):
            with Test("Check that previous credential %d is registered" % i):
                try:
                    auth = self.ctap1.authenticate(
                        chal, appid, regs[i].key_handle, check_only=True
                    )
                except ApduError as e:
                    # Indicates that key handle is registered
                    assert e.code == APDU.USE_NOT_SATISFIED

        with Test("Check an incorrect key handle is not registered"):
            kh = bytearray(regs[0].key_handle)
            kh[0] = kh[0] ^ (0x40)
            try:
                self.ctap1.authenticate(chal, appid, kh, check_only=True)
                assert 0
            except ApduError as e:
                assert e.code == APDU.WRONG_DATA

        with Test("Try to sign with incorrect key handle"):
            try:
                self.ctap1.authenticate(chal, appid, kh)
                assert 0
            except ApduError as e:
                assert e.code == APDU.WRONG_DATA

        with Test("Try to sign using an incorrect keyhandle length"):
            try:
                kh = regs[0].key_handle
                self.ctap1.authenticate(chal, appid, kh[: len(kh) // 2])
                assert 0
            except ApduError as e:
                assert e.code == APDU.WRONG_DATA

        with Test("Try to sign using an incorrect appid"):
            badid = bytearray(appid)
            badid[0] = badid[0] ^ (0x40)
            try:
                auth = self.ctap1.authenticate(chal, badid, regs[0].key_handle)
                assert 0
            except ApduError as e:
                assert e.code == APDU.WRONG_DATA

    def test_fido2_simple(self, pin_token=None):
        creds = []
        exclude_list = []
        rp = {"id": self.host, "name": "ExaRP"}
        user = {"id": b"usee_od", "name": "AB User"}
        challenge = "Y2hhbGxlbmdl"
        PIN = pin_token

        fake_id1 = array.array("B", [randint(0, 255) for i in range(0, 150)]).tobytes()
        fake_id2 = array.array("B", [randint(0, 255) for i in range(0, 73)]).tobytes()

        exclude_list.append({"id": fake_id1, "type": "public-key"})
        exclude_list.append({"id": fake_id2, "type": "public-key"})

        t1 = time.time() * 1000
        attest, data = self.client.make_credential(
            rp, user, challenge, pin=PIN, exclude_list=[]
        )
        t2 = time.time() * 1000
        VerifyAttestation(attest, data)
        print("Register time: %d ms" % (t2 - t1))

        cred = attest.auth_data.credential_data
        creds.append(cred)

        allow_list = [{"id": creds[0].credential_id, "type": "public-key"}]
        t1 = time.time() * 1000
        assertions, client_data = self.client.get_assertion(
            rp["id"], challenge, allow_list, pin=PIN
        )
        t2 = time.time() * 1000
        assertions[0].verify(client_data.hash, creds[0].public_key)

        print("Assertion time: %d ms" % (t2 - t1))

    def test_fido2_brute_force(self):
        creds = []
        exclude_list = []
        rp = {"id": self.host, "name": "ExaRP"}
        user = {"id": b"usee_od", "name": "AB User"}
        PIN = None
        abc = "abcdefghijklnmopqrstuvwxyz"
        abc += abc.upper()

        self.ctap.reset()

        for i in range(0, 2048 ** 2):
            creds = []

            challenge = "".join([abc[randint(0, len(abc) - 1)] for x in range(0, 32)])

            fake_id1 = array.array(
                "B", [randint(0, 255) for i in range(0, 150)]
            ).tobytes()
            fake_id2 = array.array(
                "B", [randint(0, 255) for i in range(0, 73)]
            ).tobytes()

            exclude_list.append({"id": fake_id1, "type": "public-key"})
            exclude_list.append({"id": fake_id2, "type": "public-key"})

            # for i in range(0,2048**2):
            for i in range(0, 1):
                t1 = time.time() * 1000
                attest, data = self.client.make_credential(
                    rp, user, challenge, pin=PIN, exclude_list=[]
                )
                print(attest.auth_data.counter)
                t2 = time.time() * 1000
                VerifyAttestation(attest, data)
                print("Register valid (%d ms)" % (t2 - t1))
            sys.stdout.flush()

            cred = attest.auth_data.credential_data
            creds.append(cred)

            # for i in range(0,2048**2):
            for i in range(0, 1):
                allow_list = [{"id": creds[0].credential_id, "type": "public-key"}]
                t1 = time.time() * 1000
                assertions, client_data = self.client.get_assertion(
                    rp["id"], challenge, allow_list, pin=PIN
                )
                t2 = time.time() * 1000
                assertions[0].verify(client_data.hash, creds[0].public_key)
                print(assertions[0].auth_data.counter)

                print("Assertion valid (%d ms)" % (t2 - t1))
                sys.stdout.flush()

    def test_fido2(self):
        def test(self, pincode=None):
            creds = []
            exclude_list = []
            rp = {"id": self.host, "name": "ExaRP"}
            user = {"id": b"usee_od", "name": "AB User"}
            challenge = "Y2hhbGxlbmdl"
            PIN = pincode

            fake_id1 = array.array(
                "B", [randint(0, 255) for i in range(0, 150)]
            ).tobytes()
            fake_id2 = array.array(
                "B", [randint(0, 255) for i in range(0, 73)]
            ).tobytes()

            exclude_list.append({"id": fake_id1, "type": "public-key"})
            exclude_list.append({"id": fake_id2, "type": "public-key"})

            # test make credential
            with Test("make %d credentials" % self.user_count):
                lastc = 0
                for i in range(0, self.user_count):
                    attest, data = self.client.make_credential(
                        rp, user, challenge, pin=PIN, exclude_list=[]
                    )
                    VerifyAttestation(attest, data)

                    # verify counter is correct
                    if lastc > 0:
                        assert attest.auth_data.counter - lastc < 10
                        assert attest.auth_data.counter - lastc > 0
                    assert attest.auth_data.counter < 0x10000
                    lastc = attest.auth_data.counter

                    cred = attest.auth_data.credential_data
                    creds.append(cred)
                    print(cred)

            if PIN is not None:
                with Test("make credential with wrong pin code"):
                    try:
                        attest, data = self.client.make_credential(
                            rp, user, challenge, pin=PIN + " ", exclude_list=[]
                        )
                    except CtapError as e:
                        assert e.code == CtapError.ERR.PIN_INVALID
                    except ClientError as e:
                        assert e.cause.code == CtapError.ERR.PIN_INVALID

            with Test("make credential with exclude list"):
                attest, data = self.client.make_credential(
                    rp, user, challenge, pin=PIN, exclude_list=exclude_list
                )
                VerifyAttestation(attest, data)
                cred = attest.auth_data.credential_data
                creds.append(cred)

            with Test("make credential with exclude list including real credential"):
                real_excl = [{"id": cred.credential_id, "type": "public-key"}]
                try:
                    attest, data = self.client.make_credential(
                        rp,
                        user,
                        challenge,
                        pin=PIN,
                        exclude_list=exclude_list + real_excl,
                    )
                    raise RuntimeError("Exclude list did not return expected error")
                except CtapError as e:
                    assert e.code == CtapError.ERR.CREDENTIAL_EXCLUDED
                except ClientError as e:
                    assert e.cause.code == CtapError.ERR.CREDENTIAL_EXCLUDED

            for i, x in enumerate(creds):
                with Test("get assertion %d" % i):
                    allow_list = [{"id": x.credential_id, "type": "public-key"}]
                    assertions, client_data = self.client.get_assertion(
                        rp["id"], challenge, allow_list, pin=PIN
                    )
                    assertions[0].verify(client_data.hash, x.public_key)

            if PIN is not None:
                with Test("get assertion with wrong pin code"):
                    try:
                        assertions, client_data = self.client.get_assertion(
                            rp["id"], challenge, allow_list, pin=PIN + " "
                        )
                    except CtapError as e:
                        assert e.code == CtapError.ERR.PIN_INVALID
                    except ClientError as e:
                        assert e.cause.code == CtapError.ERR.PIN_INVALID

            with Test("get multiple assertions"):
                allow_list = [
                    {"id": x.credential_id, "type": "public-key"} for x in creds
                ]
                assertions, client_data = self.client.get_assertion(
                    rp["id"], challenge, allow_list, pin=PIN
                )

                for ass, cred in zip(assertions, creds):
                    i += 1

                    ass.verify(client_data.hash, cred.public_key)
                    print("%d verified" % i)

        with Test("Reset device"):
            try:
                self.ctap.reset()
            except CtapError as e:
                print("Warning, reset failed: ", e)
                pass

        test(self, None)

        with Test("Set a pin code"):
            PIN = "1122aabbwfg0h9g !@#=="
            self.client.pin_protocol.set_pin(PIN)

        with Test("Illegally set pin code again"):
            try:
                self.client.pin_protocol.set_pin(PIN)
            except CtapError as e:
                assert e.code == CtapError.ERR.NOT_ALLOWED

        with Test("Change pin code"):
            PIN2 = PIN + "_pin2"
            self.client.pin_protocol.change_pin(PIN, PIN2)
            PIN = PIN2

        with Test("Change pin code using wrong pin"):
            try:
                self.client.pin_protocol.change_pin(PIN.replace("a", "b"), "1234")
            except CtapError as e:
                assert e.code == CtapError.ERR.PIN_INVALID

        with Test("MC using wrong pin"):
            try:
                self.test_fido2_simple("abcd3")
            except ClientError as e:
                assert e.cause.code == CtapError.ERR.PIN_INVALID

        with Test("get info"):
            inf = self.ctap.get_info()

        self.test_fido2_simple(PIN)

        with Test("Re-run make_credential and get_assertion tests with pin code"):
            test(self, PIN)

        with Test("Reset device"):
            try:
                self.ctap.reset()
            except CtapError as e:
                print("Warning, reset failed: ", e)

    def test_extensions(self,):
        creds = []
        exclude_list = []
        rp = {"id": self.host, "name": "ExaRP"}
        user = {"id": b"usee_od", "name": "AB User"}
        challenge = "Y2hhbGxlbmdl"
        pin_protocol = 1
        key_params = [{"type": "public-key", "alg": ES256.ALGORITHM}]
        cdh = b"123456789abcdef0123456789abcdef0"

        salt1 = b"\x5a" * 32
        salt2 = b"\x96" * 32

        self.testReset()

        with Test("Get info has hmac-secret"):
            info = self.ctap.get_info()
            assert "hmac-secret" in info.extensions

        reg = self.testMC(
            "Send MC with hmac-secret ext set to true, expect SUCCESS",
            cdh,
            rp,
            user,
            key_params,
            expectedError=CtapError.ERR.SUCCESS,
            other={"extensions": {"hmac-secret": True}, "options": {"rk": True}},
        )

        with Test("Get shared secret"):
            key_agreement, shared_secret = (
                self.client.pin_protocol._init_shared_secret()
            )
            cipher = Cipher(
                algorithms.AES(shared_secret),
                modes.CBC(b"\x00" * 16),
                default_backend(),
            )

        for salt_list in ((salt1,), (salt1, salt2)):
            enc = cipher.encryptor()
            salt_enc = b""
            for salt in salt_list:
                salt_enc += enc.update(salt)
            salt_enc += enc.finalize()

            salt_auth = hmac_sha256(shared_secret, salt_enc)[:16]

            auth = self.testGA(
                "Send GA request with %d salts hmac-secret, expect success"
                % len(salt_list),
                rp["id"],
                cdh,
                other={
                    "extensions": {
                        "hmac-secret": {1: key_agreement, 2: salt_enc, 3: salt_auth}
                    }
                },
                expectedError=CtapError.ERR.SUCCESS,
            )

            with Test(
                "Check that hmac-secret is in auth_data extensions and has %d bytes"
                % (len(salt_list) * 32)
            ):
                ext = auth.auth_data.extensions
                assert ext
                assert "hmac-secret" in ext[4]
                assert type(ext[4]["hmac-secret"]) == type(b"")
                assert len(ext[4]["hmac-secret"]) == len(salt_list) * 32

            with Test("Check that shannon_entropy of hmac-secret is good"):
                ext = auth.auth_data.extensions
                dec = cipher.decryptor()
                key = dec.update(ext[4]["hmac-secret"]) + dec.finalize()

                if len(salt_list) == 1:
                    assert shannon_entropy(ext[4]["hmac-secret"]) > 4.6
                    assert shannon_entropy(key) > 4.6
                if len(salt_list) == 2:
                    assert shannon_entropy(ext[4]["hmac-secret"]) > 5.6
                    assert shannon_entropy(key) > 5.6

    def test_fido2_other(self,):

        creds = []
        exclude_list = []
        rp = {"id": self.host, "name": "ExaRP"}
        rp2 = {"id": "solokeys.com", "name": "ExaRP"}
        user = {"id": b"usee_od", "name": "AB User"}
        user1 = {"id": b"1234567890", "name": "Conor Patrick"}
        user2 = {"id": b"oiewhfoi", "name": "Han Solo"}
        user3 = {"id": b"23ohfpjwo@@", "name": "John Smith"}
        challenge = "Y2hhbGxlbmdl"
        pin_protocol = 1
        key_params = [{"type": "public-key", "alg": ES256.ALGORITHM}]
        cdh = b"123456789abcdef0123456789abcdef0"

        def reboot():
            if self.is_sim:
                print("Sending restart command...")
                self.send_magic_reboot()
                time.sleep(0.25)
            else:
                print("Please reboot authentictor and hit enter")
                input()
                self.find_device()

        self.testReset()

        with Test("Get info"):
            info = self.ctap.get_info()

        with Test("Check FIDO2 string is in VERSIONS field"):
            assert "FIDO_2_0" in info.versions

        with Test("Check pin protocols field"):
            if len(info.pin_protocols):
                assert sum(info.pin_protocols) > 0

        with Test("Check options field"):
            for x in info.options:
                assert info.options[x] in [True, False]

        prev_reg = self.testMC(
            "Send MC request, expect success",
            cdh,
            rp,
            user,
            key_params,
            expectedError=CtapError.ERR.SUCCESS,
        )

        with Test("Check attestation format is correct"):
            assert prev_reg.fmt in ["packed", "tpm", "android-key", "adroid-safetynet"]

        with Test("Check auth_data is at least 77 bytes"):
            assert len(prev_reg.auth_data) >= 77

        allow_list = [
            {
                "id": prev_reg.auth_data.credential_data.credential_id,
                "type": "public-key",
            }
        ]

        prev_auth = self.testGA(
            "Send GA request, expect success",
            rp["id"],
            cdh,
            allow_list,
            expectedError=CtapError.ERR.SUCCESS,
        )

        with Test("Test auth_data is 37 bytes"):
            assert len(prev_auth.auth_data) == 37

        with Test("Test that auth_data.rpIdHash is correct"):
            assert sha256(rp["id"].encode()) == prev_auth.auth_data.rp_id_hash

        with Test("Check that AT flag is not set"):
            assert (prev_auth.auth_data.flags & 0xF8) == 0

        with Test("Test that user, credential and numberOfCredentials are not present"):
            assert prev_auth.user == None
            assert prev_auth.number_of_credentials == None

        self.testGA(
            "Send GA request with empty allow_list, expect NO_CREDENTIALS",
            rp["id"],
            cdh,
            [],
            expectedError=CtapError.ERR.NO_CREDENTIALS,
        )

        # apply bit flip
        badid = list(prev_reg.auth_data.credential_data.credential_id[:])
        badid[len(badid) // 2] = badid[len(badid) // 2] ^ 1
        badid = bytes(badid)

        self.testGA(
            "Send GA request with corrupt credId in allow_list, expect NO_CREDENTIALS",
            rp["id"],
            cdh,
            [{"id": badid, "type": "public-key"}],
            expectedError=CtapError.ERR.NO_CREDENTIALS,
        )

        self.testMC(
            "Send MC request with missing clientDataHash, expect error",
            None,
            rp,
            user,
            key_params,
            expectedError=CtapError.ERR.MISSING_PARAMETER,
        )

        self.testMC(
            "Send MC request with integer for clientDataHash, expect error",
            5,
            rp,
            user,
            key_params,
        )

        self.testMC(
            "Send MC request with missing user, expect error",
            cdh,
            rp,
            None,
            key_params,
            expectedError=CtapError.ERR.MISSING_PARAMETER,
        )

        self.testMC(
            "Send MC request with bytearray user, expect error",
            cdh,
            rp,
            b"1234abcd",
            key_params,
        )

        self.testMC(
            "Send MC request with missing RP, expect error",
            cdh,
            None,
            user,
            key_params,
            expectedError=CtapError.ERR.MISSING_PARAMETER,
        )

        self.testMC(
            "Send MC request with bytearray RP, expect error",
            cdh,
            b"1234abcd",
            user,
            key_params,
        )

        self.testMC(
            "Send MC request with missing pubKeyCredParams, expect error",
            cdh,
            rp,
            user,
            None,
            expectedError=CtapError.ERR.MISSING_PARAMETER,
        )

        self.testMC(
            "Send MC request with incorrect pubKeyCredParams, expect error",
            cdh,
            rp,
            user,
            b"2356",
        )

        self.testMC(
            "Send MC request with incorrect excludeList, expect error",
            cdh,
            rp,
            user,
            key_params,
            other={"exclude_list": 8},
        )

        self.testMC(
            "Send MC request with incorrect extensions, expect error",
            cdh,
            rp,
            user,
            key_params,
            other={"extensions": 8},
        )

        self.testMC(
            "Send MC request with incorrect options, expect error",
            cdh,
            rp,
            user,
            key_params,
            other={"options": 8},
        )

        self.testMC(
            "Send MC request with bad RP.name",
            cdh,
            {"id": self.host, "name": 8, "icon": "icon"},
            user,
            key_params,
        )

        self.testMC(
            "Send MC request with bad RP.id",
            cdh,
            {"id": 8, "name": "name", "icon": "icon"},
            user,
            key_params,
        )

        self.testMC(
            "Send MC request with bad RP.icon",
            cdh,
            {"id": self.host, "name": "name", "icon": 8},
            user,
            key_params,
        )

        self.testMC(
            "Send MC request with bad user.name",
            cdh,
            rp,
            {"id": b"usee_od", "name": 8},
            key_params,
        )

        self.testMC(
            "Send MC request with bad user.id",
            cdh,
            rp,
            {"id": "usee_od", "name": "name"},
            key_params,
        )

        self.testMC(
            "Send MC request with bad user.displayName",
            cdh,
            rp,
            {"id": "usee_od", "name": "name", "displayName": 8},
            key_params,
        )

        self.testMC(
            "Send MC request with bad user.icon",
            cdh,
            rp,
            {"id": "usee_od", "name": "name", "icon": 8},
            key_params,
        )

        self.testMC(
            "Send MC request with non-map pubKeyCredParams item",
            cdh,
            rp,
            user,
            ["wrong"],
        )

        self.testMC(
            "Send MC request with pubKeyCredParams item missing type field",
            cdh,
            rp,
            user,
            [{"alg": ES256.ALGORITHM}],
            expectedError=CtapError.ERR.MISSING_PARAMETER,
        )

        self.testMC(
            "Send MC request with pubKeyCredParams item with bad type field",
            cdh,
            rp,
            user,
            [{"alg": ES256.ALGORITHM, "type": b"public-key"}],
        )

        self.testMC(
            "Send MC request with pubKeyCredParams item missing alg",
            cdh,
            rp,
            user,
            [{"type": "public-key"}],
            expectedError=CtapError.ERR.MISSING_PARAMETER,
        )

        self.testMC(
            "Send MC request with pubKeyCredParams item with bad alg",
            cdh,
            rp,
            user,
            [{"alg": "7", "type": "public-key"}],
        )

        self.testMC(
            "Send MC request with pubKeyCredParams item with bogus alg, expect UNSUPPORTED_ALGORITHM",
            cdh,
            rp,
            user,
            [{"alg": 1234, "type": "public-key"}],
            expectedError=CtapError.ERR.UNSUPPORTED_ALGORITHM,
        )

        self.testMC(
            "Send MC request with pubKeyCredParams item with bogus type, expect UNSUPPORTED_ALGORITHM",
            cdh,
            rp,
            user,
            [{"alg": ES256.ALGORITHM, "type": "rot13"}],
            expectedError=CtapError.ERR.UNSUPPORTED_ALGORITHM,
        )

        self.testMC(
            "Send MC request with excludeList item with bogus type, expect SUCCESS",
            cdh,
            rp,
            user,
            key_params,
            expectedError=CtapError.ERR.SUCCESS,
            other={"exclude_list": [{"id": b"1234", "type": "rot13"}]},
        )

        self.testMC(
            "Send MC request with excludeList with bad item, expect error",
            cdh,
            rp,
            user,
            key_params,
            other={"exclude_list": ["1234"]},
        )

        self.testMC(
            "Send MC request with excludeList with item missing type field, expect error",
            cdh,
            rp,
            user,
            key_params,
            other={"exclude_list": [{"id": b"1234"}]},
        )

        self.testMC(
            "Send MC request with excludeList with item missing id field, expect error",
            cdh,
            rp,
            user,
            key_params,
            other={"exclude_list": [{"type": "public-key"}]},
        )

        self.testMC(
            "Send MC request with excludeList with item containing bad id field, expect error",
            cdh,
            rp,
            user,
            key_params,
            other={"exclude_list": [{"type": "public-key", "id": "1234"}]},
        )

        self.testMC(
            "Send MC request with excludeList with item containing bad type field, expect error",
            cdh,
            rp,
            user,
            key_params,
            other={"exclude_list": [{"type": b"public-key", "id": b"1234"}]},
        )

        self.testMC(
            "Send MC request with excludeList containing previous registration, expect CREDENTIAL_EXCLUDED",
            cdh,
            rp,
            user,
            key_params,
            other={
                "exclude_list": [
                    {
                        "type": "public-key",
                        "id": prev_reg.auth_data.credential_data.credential_id,
                    }
                ]
            },
            expectedError=CtapError.ERR.CREDENTIAL_EXCLUDED,
        )

        self.testMC(
            "Send MC request with unknown option, expect SUCCESS",
            cdh,
            rp,
            user,
            key_params,
            other={"options": {"unknown": False}},
            expectedError=CtapError.ERR.SUCCESS,
        )

        if "uv" in info.options:
            if info.options["uv"]:
                self.testMC(
                    "Send MC request with uv set to true, expect SUCCESS",
                    cdh,
                    rp,
                    user,
                    key_params,
                    other={"options": {"uv": True}},
                    expectedError=CtapError.ERR.SUCCESS,
                )
        if "up" in info.options:
            if info.options["up"]:
                self.testMC(
                    "Send MC request with up set to true, expect INVALID_OPTION",
                    cdh,
                    rp,
                    user,
                    key_params,
                    other={"options": {"up": True}},
                    expectedError=CtapError.ERR.INVALID_OPTION,
                )

        self.testGA(
            "Send GA request with missing RPID, expect MISSING_PARAMETER",
            None,
            cdh,
            allow_list,
            expectedError=CtapError.ERR.MISSING_PARAMETER,
        )

        self.testGA(
            "Send GA request with bad RPID, expect error",
            {"type": "wrong"},
            cdh,
            allow_list,
        )

        self.testGA(
            "Send GA request with missing clientDataHash, expect MISSING_PARAMETER",
            rp["id"],
            None,
            allow_list,
            expectedError=CtapError.ERR.MISSING_PARAMETER,
        )

        self.testGA(
            "Send GA request with bad clientDataHash, expect error",
            rp["id"],
            {"type": "wrong"},
            allow_list,
        )

        self.testGA(
            "Send GA request with bad allow_list, expect error",
            rp["id"],
            cdh,
            {"type": "wrong"},
        )

        self.testGA(
            "Send GA request with bad item in allow_list, expect error",
            rp["id"],
            cdh,
            allow_list + ["wrong"],
        )

        self.testGA(
            "Send GA request with unknown option, expect SUCCESS",
            rp["id"],
            cdh,
            allow_list,
            other={"options": {"unknown": True}},
            expectedError=CtapError.ERR.SUCCESS,
        )

        if "uv" in info.options:
            if info.options["uv"]:
                res = self.testGA(
                    "Send GA request with uv set to true, expect SUCCESS",
                    rp["id"],
                    cdh,
                    allow_list,
                    other={"options": {"uv": True}},
                    expectedError=CtapError.ERR.SUCCESS,
                )
                with Test("Check that UV flag is set in response"):
                    assert res.auth_data.flags & (1 << 2)
        if "up" in info.options:
            if info.options["up"]:
                res = self.testGA(
                    "Send GA request with up set to true, expect SUCCESS",
                    rp["id"],
                    cdh,
                    allow_list,
                    other={"options": {"up": True}},
                    expectedError=CtapError.ERR.SUCCESS,
                )
            with Test("Check that UP flag is set in response"):
                assert res.auth_data.flags & 1

        self.testGA(
            "Send GA request with bogus type item in allow_list, expect SUCCESS",
            rp["id"],
            cdh,
            allow_list + [{"type": "rot13", "id": b"1234"}],
            expectedError=CtapError.ERR.SUCCESS,
        )

        self.testGA(
            "Send GA request with item missing type field in allow_list, expect error",
            rp["id"],
            cdh,
            allow_list + [{"id": b"1234"}],
        )

        self.testGA(
            "Send GA request with item containing bad type field in allow_list, expect error",
            rp["id"],
            cdh,
            allow_list + [{"type": b"public-key", "id": b"1234"}],
        )

        self.testGA(
            "Send GA request with item containing bad id in allow_list, expect error",
            rp["id"],
            cdh,
            allow_list + [{"type": b"public-key", "id": 42}],
        )

        self.testGA(
            "Send GA request with item missing id in allow_list, expect error",
            rp["id"],
            cdh,
            allow_list + [{"type": b"public-key"}],
        )

        self.testReset()

        def testRk(pin_code=None):
            self.testGA(
                "Send GA request with reset auth, expect NO_CREDENTIALS",
                rp["id"],
                cdh,
                allow_list,
                expectedError=CtapError.ERR.NO_CREDENTIALS,
            )

            pin_auth = None
            if pin_code:
                with Test("Set pin code"):
                    self.client.pin_protocol.set_pin(pin_code)
                    pin_token = self.client.pin_protocol.get_pin_token(pin_code)
                    pin_auth = hmac_sha256(pin_token, cdh)[:16]

            self.testMC(
                "Send MC request with rk option set to true, expect SUCCESS",
                cdh,
                rp,
                user,
                key_params,
                other={"options": {"rk": True}, "pin_auth": pin_auth},
                expectedError=CtapError.ERR.SUCCESS,
            )

            options = {"rk": True}
            if "uv" in info.options and info.options["uv"]:
                options["uv"] = False

            for i, x in enumerate([user1, user2, user3]):
                self.testMC(
                    "Send MC request with rk option set to true, expect SUCCESS %d/3"
                    % (i + 1),
                    cdh,
                    rp2,
                    x,
                    key_params,
                    other={"options": options, "pin_auth": pin_auth},
                    expectedError=CtapError.ERR.SUCCESS,
                )

            auth1 = self.testGA(
                "Send GA request with no allow_list, expect SUCCESS",
                rp2["id"],
                cdh,
                other={"options": options, "pin_auth": pin_auth},
                expectedError=CtapError.ERR.SUCCESS,
            )

            with Test("Check that there are 3 credentials returned"):
                assert auth1.number_of_credentials == 3

            with Test("Get the next 2 assertions"):
                auth2 = self.ctap.get_next_assertion()
                auth3 = self.ctap.get_next_assertion()

            if not pin_code:
                with Test("Check only the user ID was returned"):
                    assert "id" in auth1.user.keys() and len(auth1.user.keys()) == 1
                    assert "id" in auth2.user.keys() and len(auth2.user.keys()) == 1
                    assert "id" in auth3.user.keys() and len(auth3.user.keys()) == 1
            else:
                with Test("Check that all user info was returned"):
                    for x in (auth1, auth2, auth3):
                        for y in ("name", "icon", "displayName", "id"):
                            assert y in x.user.keys()
                        assert len(x.user.keys()) == 4

            with Test("Send an extra getNextAssertion request, expect error"):
                try:
                    auth4 = self.ctap.get_next_assertion()
                    assert 0
                except CtapError as e:
                    print(e)

        testRk(None)
        #
        # print("Assuming authenticator does NOT have a display.")
        pin1 = "1234567890"
        testRk("1234567890")

        # PinProtocolV1
        res = self.testCP(
            "Test getKeyAgreement, expect SUCCESS",
            pin_protocol,
            PinProtocolV1.CMD.GET_KEY_AGREEMENT,
            expectedError=CtapError.ERR.SUCCESS,
        )

        with Test("Test getKeyAgreement has appropriate fields"):
            key = res[1]
            assert "Is public key" and key[1] == 2
            assert "Is P256" and key[-1] == 1
            assert "Is right alg" and key[3] == -7
            assert "Right key" and len(key[-3]) == 32 and type(key[-3]) == type(bytes())

        with Test("Test setting a new pin"):
            pin2 = "qwertyuiop\x11\x22\x33\x00123"
            self.client.pin_protocol.change_pin(pin1, pin2)

        with Test("Test getting new pin_auth"):
            pin_token = self.client.pin_protocol.get_pin_token(pin2)
            pin_auth = hmac_sha256(pin_token, cdh)[:16]

        res_mc = self.testMC(
            "Send MC request with new pin auth",
            cdh,
            rp,
            user,
            key_params,
            other={"pin_auth": pin_auth},
            expectedError=CtapError.ERR.SUCCESS,
        )

        with Test("Check UV flag is set"):
            assert res_mc.auth_data.flags & (1 << 2)

        res_ga = self.testGA(
            "Send GA request with no allow_list, expect SUCCESS",
            rp["id"],
            cdh,
            [
                {
                    "type": "public-key",
                    "id": res_mc.auth_data.credential_data.credential_id,
                }
            ],
            other={"pin_auth": pin_auth},
            expectedError=CtapError.ERR.SUCCESS,
        )

        with Test("Check UV flag is set"):
            assert res_ga.auth_data.flags & (1 << 2)

        self.testReset()

        with Test("Setting pin code, expect SUCCESS"):
            self.client.pin_protocol.set_pin(pin1)

        self.testReset()

        # print("Setting pin code <4 bytes, expect POLICY_VIOLATION ")
        # try:
        #     self.client.pin_protocol.set_pin("123")
        # except CtapError as e:
        #     assert e.code == CtapError.ERR.POLICY_VIOLATION
        # print("Pass")

        with Test("Setting pin code >63 bytes, expect POLICY_VIOLATION "):
            try:
                self.client.pin_protocol.set_pin("A" * 64)
                assert 0
            except CtapError as e:
                assert e.code == CtapError.ERR.PIN_POLICY_VIOLATION

        with Test("Get pin token when no pin is set, expect PIN_NOT_SET"):
            try:
                self.client.pin_protocol.get_pin_token(pin1)
                assert 0
            except CtapError as e:
                assert e.code == CtapError.ERR.PIN_NOT_SET

        with Test("Get change pin when no pin is set, expect PIN_NOT_SET"):
            try:
                self.client.pin_protocol.change_pin(pin1, "1234")
                assert 0
            except CtapError as e:
                assert e.code == CtapError.ERR.PIN_NOT_SET

        with Test("Setting pin code and get pin_token, expect SUCCESS"):
            self.client.pin_protocol.set_pin(pin1)
            pin_token = self.client.pin_protocol.get_pin_token(pin1)
            pin_auth = hmac_sha256(pin_token, cdh)[:16]

        with Test("Get info and assert that clientPin is set to true"):
            info = self.ctap.get_info()
            assert info.options["clientPin"]

        with Test("Test setting pin again fails"):
            try:
                self.client.pin_protocol.set_pin(pin1)
                assert 0
            except CtapError as e:
                print(e)

        res_mc = self.testMC(
            "Send MC request with no pin_auth, expect PIN_REQUIRED",
            cdh,
            rp,
            user,
            key_params,
            expectedError=CtapError.ERR.PIN_REQUIRED,
        )

        res_mc = self.testGA(
            "Send GA request with no pin_auth, expect PIN_REQUIRED",
            rp["id"],
            cdh,
            expectedError=CtapError.ERR.PIN_REQUIRED,
        )

        res = self.testCP(
            "Test getRetries, expect SUCCESS",
            pin_protocol,
            PinProtocolV1.CMD.GET_RETRIES,
            expectedError=CtapError.ERR.SUCCESS,
        )

        with Test("Check there is 8 pin attempts left"):
            assert res[3] == 8

        # Flip 1 bit
        pin_wrong = list(pin1)
        c = pin1[len(pin1) // 2]

        pin_wrong[len(pin1) // 2] = chr(ord(c) ^ 1)
        pin_wrong = "".join(pin_wrong)

        for i in range(1, 3):
            self.testPP(
                "Get pin_token with wrong pin code, expect PIN_INVALID (%d/2)" % i,
                pin_wrong,
                expectedError=CtapError.ERR.PIN_INVALID,
            )
            print("Check there is %d pin attempts left" % (8 - i))
            res = self.ctap.client_pin(pin_protocol, PinProtocolV1.CMD.GET_RETRIES)
            assert res[3] == (8 - i)
            print("Pass")

        for i in range(1, 3):
            self.testPP(
                "Get pin_token with wrong pin code, expect PIN_AUTH_BLOCKED %d/2" % i,
                pin_wrong,
                expectedError=CtapError.ERR.PIN_AUTH_BLOCKED,
            )

        reboot()

        with Test("Get pin_token, expect SUCCESS"):
            pin_token = self.client.pin_protocol.get_pin_token(pin1)
            pin_auth = hmac_sha256(pin_token, cdh)[:16]

        res_mc = self.testMC(
            "Send MC request with correct pin_auth",
            cdh,
            rp,
            user,
            key_params,
            other={"pin_auth": pin_auth},
            expectedError=CtapError.ERR.SUCCESS,
        )

        with Test("Test getRetries resets to 8"):
            res = self.ctap.client_pin(pin_protocol, PinProtocolV1.CMD.GET_RETRIES)
            assert res[3] == (8)

        for i in range(1, 10):
            err = CtapError.ERR.PIN_INVALID
            if i in (3, 6):
                err = CtapError.ERR.PIN_AUTH_BLOCKED
            elif i >= 9:
                err = CtapError.ERR.PIN_BLOCKED
            self.testPP(
                "Lock out authentictor and check correct error codes %d/9" % i,
                pin_wrong,
                expectedError=err,
            )

            attempts = 8 - i
            if i > 8:
                attempts = 0

            with Test("Check there is %d pin attempts left" % attempts):
                res = self.ctap.client_pin(pin_protocol, PinProtocolV1.CMD.GET_RETRIES)
                assert res[3] == attempts

            if err == CtapError.ERR.PIN_AUTH_BLOCKED:
                reboot()

        res_mc = self.testMC(
            "Send MC request with correct pin_auth, expect PIN_BLOCKED",
            cdh,
            rp,
            user,
            key_params,
            other={"pin_auth": pin_auth},
            expectedError=CtapError.ERR.PIN_BLOCKED,
        )

        reboot()

        self.testPP(
            "Get pin_token with correct pin code, expect PIN_BLOCKED",
            pin1,
            expectedError=CtapError.ERR.PIN_BLOCKED,
        )

        self.testReset()

        print("Done")

    def test_rk(self,):
        creds = []
        rp = {"id": self.host, "name": "ExaRP"}

        users = [
            {"id": b"user" + os.urandom(16), "name": "Username%d" % i}
            for i in range(0, self.user_count)
        ]
        challenge = "Y2hhbGxlbmdl"
        PIN = None
        self.ctap.reset()
        # if PIN: self.client.pin_protocol.set_pin(PIN)

        with Test("registering 1 user with RK"):
            t1 = time.time() * 1000
            attest, data = self.client.make_credential(
                rp, users[-1], challenge, pin=PIN, exclude_list=[], rk=True
            )
            t2 = time.time() * 1000
            VerifyAttestation(attest, data)
            creds.append(attest.auth_data.credential_data)

        with Test("1 assertion"):
            t1 = time.time() * 1000
            assertions, client_data = self.client.get_assertion(
                rp["id"], challenge, pin=PIN
            )
            t2 = time.time() * 1000
            assertions[0].verify(client_data.hash, creds[0].public_key)

        with Test("registering %d users with RK" % len(users)):
            for i in range(0, len(users) - 1):
                t1 = time.time() * 1000
                attest, data = self.client.make_credential(
                    rp, users[i], challenge, pin=PIN, exclude_list=[], rk=True
                )
                t2 = time.time() * 1000
                VerifyAttestation(attest, data)
                creds.append(attest.auth_data.credential_data)

        t1 = time.time() * 1000
        assertions, client_data = self.client.get_assertion(
            rp["id"], challenge, pin=PIN
        )
        t2 = time.time() * 1000

        print("Got %d assertions for %d users" % (len(assertions), len(users)))
        assert len(assertions) == len(users)

        for x, y in zip(assertions, creds):
            x.verify(client_data.hash, y.public_key)

        print("Assertion(s) valid (%d ms)" % (t2 - t1))

        with Test("register a duplicate user "):
            t1 = time.time() * 1000
            attest, data = self.client.make_credential(
                rp, users[1], challenge, pin=PIN, exclude_list=[], rk=True
            )
            t2 = time.time() * 1000
            VerifyAttestation(attest, data)
            creds = creds[:2] + creds[3:] + [attest.auth_data.credential_data]

        t1 = time.time() * 1000
        assertions, client_data = self.client.get_assertion(
            rp["id"], challenge, pin=PIN
        )
        t2 = time.time() * 1000
        with Test("check %d assertions, %d users" % (len(assertions), len(users))):
            assert len(assertions) == len(users)
            for x, y in zip(assertions, creds):
                x.verify(client_data.hash, y.public_key)

    def test_solo(self,):
        """
        Solo specific tests
        """
        # RNG command
        sc = SoloClient()
        sc.find_device(self.dev)
        sc.use_u2f()
        memmap = (0x08005000, 0x08005000 + 198 * 1024 - 8)

        total = 1024 * 16
        with Test("Gathering %d random bytes..." % total):
            entropy = b""
            while len(entropy) < total:
                entropy += sc.get_rng()

        with Test("Test entropy is close to perfect"):
            sum = shannon_entropy(entropy)
            assert sum > 7.98
        print("Entropy is %.5f bits per byte." % sum)

        with Test("Test Solo version command"):
            assert len(sc.solo_version()) == 3

        with Test("Test bootloader is not active"):
            try:
                sc.write_flash(memmap[0], b"1234")
            except ApduError:
                pass

        sc.exchange = sc.exchange_fido2
        with Test("Test Solo version and random commands with fido2 layer"):
            assert len(sc.solo_version()) == 3
            sc.get_rng()

    def test_bootloader(self,):
        sc = SoloClient()
        sc.find_device(self.dev)
        sc.use_u2f()

        memmap = (0x08005000, 0x08005000 + 198 * 1024 - 8)
        data = b"A" * 64

        with Test("Test version command"):
            assert len(sc.bootloader_version()) == 3

        with Test("Test write command"):
            sc.write_flash(memmap[0], data)

        for addr in (memmap[0] - 8, memmap[0] - 4, memmap[1], memmap[1] - 8):
            with Test("Test out of bounds write command at 0x%04x" % addr):
                try:
                    sc.write_flash(addr, data)
                except CtapError as e:
                    assert e.code == CtapError.ERR.NOT_ALLOWED

    def test_responses(self,):
        PIN = "1234"
        RPID = self.host
        for dev in CtapHidDevice.list_devices():
            print("dev", dev)
            client = Fido2Client(dev, RPID)
            ctap = client.ctap2
            # ctap.reset()
            try:
                if PIN:
                    client.pin_protocol.set_pin(PIN)
            except:
                pass

            inf = ctap.get_info()
            # print (inf)
            print("versions: ", inf.versions)
            print("aaguid: ", inf.aaguid)
            print("rk: ", inf.options["rk"])
            print("clientPin: ", inf.options["clientPin"])
            print("max_message_size: ", inf.max_msg_size)

            # rp = {'id': 'SelectDevice', 'name': 'SelectDevice'}
            rp = {"id": RPID, "name": "ExaRP"}
            user = {"id": os.urandom(10), "name": "SelectDevice"}
            user = {"id": b"21first one", "name": "single User"}
            challenge = "Y2hhbGxlbmdl"

            if 1:
                attest, data = client.make_credential(
                    rp, user, challenge, exclude_list=[], pin=PIN, rk=True
                )

                cred = attest.auth_data.credential_data
                creds = [cred]

                allow_list = [{"id": creds[0].credential_id, "type": "public-key"}]
                allow_list = []
                assertions, client_data = client.get_assertion(
                    rp["id"], challenge, pin=PIN
                )
                assertions[0].verify(client_data.hash, creds[0].public_key)

            if 0:
                print("registering 1 user with RK")
                t1 = time.time() * 1000
                attest, data = client.make_credential(
                    rp, user, challenge, pin=PIN, exclude_list=[], rk=True
                )
                t2 = time.time() * 1000
                VerifyAttestation(attest, data)
                creds = [attest.auth_data.credential_data]
                print("Register valid (%d ms)" % (t2 - t1))

                print("1 assertion")
                t1 = time.time() * 1000
                assertions, client_data = client.get_assertion(
                    rp["id"], challenge, pin=PIN
                )
                t2 = time.time() * 1000
                assertions[0].verify(client_data.hash, creds[0].public_key)
                print("Assertion valid (%d ms)" % (t2 - t1))

            # print('fmt:',attest.fmt)
            # print('rp_id_hash',attest.auth_data.rp_id_hash)
            # print('flags:', hex(attest.auth_data.flags))
            # print('count:', hex(attest.auth_data.counter))
            print("flags MC:", attest.auth_data)
            print("flags GA:", assertions[0].auth_data)
            # print('cred_id:',attest.auth_data.credential_data.credential_id)
            # print('pubkey:',attest.auth_data.credential_data.public_key)
            # print('aaguid:',attest.auth_data.credential_data.aaguid)
            # print('cred data:',attest.auth_data.credential_data)
            # print('auth_data:',attest.auth_data)
            # print('auth_data:',attest.auth_data)
            # print('alg:',attest.att_statement['alg'])
            # print('sig:',attest.att_statement['sig'])
            # print('x5c:',attest.att_statement['x5c'])
            # print('data:',data)

            print("assertion:", assertions[0])
            print("clientData:", client_data)

            print()
            # break


def test_find_brute_force():
    i = 0
    while 1:
        t1 = time.time() * 1000
        t = Tester()
        t.find_device()
        t2 = time.time() * 1000
        print("connected %d (%d ms)" % (i, t2 - t1))
        i += 1
        time.sleep(0.01)


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: %s [sim] <[u2f]|[fido2]|[rk]|[hid]|[ping]>")
        sys.exit(0)

    t = Tester()
    t.set_user_count(3)

    if "sim" in sys.argv:
        print("Using UDP backend.")
        force_udp_backend()
        t.set_sim(True)
        t.set_user_count(10)

    t.find_device()

    if "solo" in sys.argv:
        t.test_solo()

    if "u2f" in sys.argv:
        t.test_u2f()

    if "fido2" in sys.argv:
        t.test_fido2()
        t.test_fido2_other()

    if "fido2-ext" in sys.argv:
        t.test_extensions()

    if "rk" in sys.argv:
        t.test_rk()

    if "ping" in sys.argv:
        t.test_long_ping()

    # hid tests are a bit invasive and should be done last
    if "hid" in sys.argv:
        t.test_hid(check_timeouts=t.is_sim)

    if "bootloader" in sys.argv:
        if t.is_sim:
            raise RuntimeError("Cannot test bootloader in simulation yet.")
        print("Put device in bootloader mode and then hit enter")
        input()
        t.test_bootloader()

    # t.test_responses()
    # test_find_brute_force()
    # t.test_fido2_brute_force()