import uctypes
import ustruct
import utime
from micropython import const

from trezor import config, io, log, loop, ui, utils, workflow
from trezor.crypto import aes, der, hashlib, hmac, random
from trezor.crypto.curve import nist256p1
from trezor.ui.confirm import CONFIRMED, Confirm, ConfirmPageable, Pageable
from trezor.ui.text import Text

from apps.common import cbor, storage
from apps.webauthn.confirm import ConfirmContent, ConfirmInfo

if __debug__:
    from apps.debug import confirm_signal

if False:
    from typing import Any, Coroutine, List, Optional
    from apps.webauthn.credential import Credential, Fido2Credential, U2fCredential


_CID_BROADCAST = const(0xFFFFFFFF)  # broadcast channel id

# types of frame
_TYPE_MASK = const(0x80)  # frame type mask
_TYPE_INIT = const(0x80)  # initial frame identifier
_TYPE_CONT = const(0x00)  # continuation frame identifier

# U2F HID sizes
_HID_RPT_SIZE = const(64)
_FRAME_INIT_SIZE = const(57)
_FRAME_CONT_SIZE = const(59)
_MAX_U2FHID_MSG_PAYLOAD_LEN = const(_FRAME_INIT_SIZE + 128 * _FRAME_CONT_SIZE)

# types of cmd
_CMD_PING = const(0x81)  # echo data through local processor only
_CMD_MSG = const(0x83)  # send U2F message frame
_CMD_LOCK = const(0x84)  # send lock channel command
_CMD_INIT = const(0x86)  # channel initialization
_CMD_WINK = const(0x88)  # send device identification wink
_CMD_CBOR = const(0x90)  # send encapsulated CTAP CBOR encoded message
_CMD_CANCEL = const(0x91)  # cancel any outstanding requests on this CID
_CMD_KEEPALIVE = const(0xBB)  # processing a message
_CMD_ERROR = const(0xBF)  # error response

# types for the msg cmd
_MSG_REGISTER = const(0x01)  # registration command
_MSG_AUTHENTICATE = const(0x02)  # authenticate/sign command
_MSG_VERSION = const(0x03)  # read version string command

# types for the cbor cmd
_CBOR_MAKE_CREDENTIAL = const(0x01)  # generate new credential command
_CBOR_GET_ASSERTION = const(0x02)  # authenticate command
_CBOR_GET_INFO = const(0x04)  # report AAGUID and device capabilities
_CBOR_CLIENT_PIN = const(0x06)  # PIN and pinToken management
_CBOR_RESET = const(0x07)  # factory reset, invalidating all generated credentials
_CBOR_GET_NEXT_ASSERTION = const(0x08)  # obtain the next per-credential signature

# CBOR MakeCredential command parameter keys
_MAKECRED_CMD_CLIENT_DATA_HASH = const(0x01)  # bytes, required
_MAKECRED_CMD_RP = const(0x02)  # map, required
_MAKECRED_CMD_USER = const(0x03)  # map, required
_MAKECRED_CMD_PUB_KEY_CRED_PARAMS = const(0x04)  # array of maps, required
_MAKECRED_CMD_EXCLUDE_LIST = const(0x05)  # array of maps, optional
_MAKECRED_CMD_EXTENSIONS = const(0x06)  # map, optional
_MAKECRED_CMD_OPTIONS = const(0x07)  # map, optional
_MAKECRED_CMD_PIN_AUTH = const(0x08)  # bytes, optional

# CBOR MakeCredential response member keys
_MAKECRED_RESP_FMT = const(0x01)  # str, required
_MAKECRED_RESP_AUTH_DATA = const(0x02)  # bytes, required
_MAKECRED_RESP_ATT_STMT = const(0x03)  # map, required

# CBOR GetAssertion command parameter keys
_GETASSERT_CMD_RP_ID = const(0x01)  # str, required
_GETASSERT_CMD_CLIENT_DATA_HASH = const(0x02)  # bytes, required
_GETASSERT_CMD_ALLOW_LIST = const(0x03)  # array of maps, optional
_GETASSERT_CMD_EXTENSIONS = const(0x04)  # map, optional
_GETASSERT_CMD_OPTIONS = const(0x05)  # map, optional
_GETASSERT_CMD_PIN_AUTH = const(0x06)  # bytes, optional

# CBOR GetAssertion response member keys
_GETASSERT_RESP_CREDENTIAL = const(0x01)  # map, optional
_GETASSERT_RESP_AUTH_DATA = const(0x02)  # bytes, required
_GETASSERT_RESP_SIGNATURE = const(0x03)  # bytes, required
_GETASSERT_RESP_USER = const(0x04)  # map, optional
_GETASSERT_RESP_PUB_KEY_CREDENTIAL_USER_ENTITY = const(0x04)  # map, optional
_GETASSERT_RESP_NUM_OF_CREDENTIALS = const(0x05)  # int, optional

# CBOR GetInfo response member keys
_GETINFO_RESP_VERSIONS = const(0x01)  # array of str, required
_GETINFO_RESP_EXTENSIONS = const(0x02)  # array of str, optional
_GETINFO_RESP_AAGUID = const(0x03)  # bytes(16), required
_GETINFO_RESP_OPTIONS = const(0x04)  # map, optional
_GETINFO_RESP_PIN_PROTOCOLS = const(0x06)  # list of unsigned integers, optional

# CBOR ClientPin command parameter keys
_CLIENTPIN_CMD_PIN_PROTOCOL = const(0x01)  # unsigned int, required
_CLIENTPIN_CMD_SUBCOMMAND = const(0x02)  # unsigned int, required
_CLIENTPIN_SUBCMD_GET_KEY_AGREEMENT = const(0x02)

# CBOR ClientPin response member keys
_CLIENTPIN_RESP_KEY_AGREEMENT = const(0x01)  # COSE_Key, optional

# status codes for the keepalive cmd
_KEEPALIVE_STATUS_PROCESSING = const(0x01)  # still processing the current request
_KEEPALIVE_STATUS_UP_NEEDED = const(0x02)  # waiting for user presence

# time intervals and timeouts
_KEEPALIVE_INTERVAL_MS = const(80)  # interval between keepalive commands
_CTAP_HID_TIMEOUT_MS = const(500)
_U2F_CONFIRM_TIMEOUT_MS = const(10 * 1000)
_FIDO2_CONFIRM_TIMEOUT_MS = const(60 * 1000)

# CBOR object signing and encryption algorithms and keys
_COSE_ALG_KEY = const(3)
_COSE_ALG_ES256 = const(-7)  # ECDSA P-256 with SHA-256
_COSE_ALG_ECDH_ES_HKDF_256 = const(-25)  # Ephemeral-static ECDH with HKDF SHA-256
_COSE_KEY_TYPE_KEY = const(1)
_COSE_KEY_TYPE_EC2 = const(2)  # elliptic curve keys with x- and y-coordinate pair
_COSE_CURVE_KEY = const(-1)  # elliptic curve identifier
_COSE_CURVE_P256 = const(1)  # P-256 curve
_COSE_X_COORD_KEY = const(-2)  # x coordinate of the public key
_COSE_Y_COORD_KEY = const(-3)  # y coordinate of the public key

# hid error codes
_ERR_NONE = const(0x00)  # no error
_ERR_INVALID_CMD = const(0x01)  # invalid command
_ERR_INVALID_PAR = const(0x02)  # invalid parameter
_ERR_INVALID_LEN = const(0x03)  # invalid message length
_ERR_INVALID_SEQ = const(0x04)  # invalid message sequencing
_ERR_MSG_TIMEOUT = const(0x05)  # message has timed out
_ERR_CHANNEL_BUSY = const(0x06)  # channel busy
_ERR_LOCK_REQUIRED = const(0x0A)  # command requires channel lock
_ERR_INVALID_CID = const(0x0B)  # command not allowed on this cid
_ERR_CBOR_UNEXPECTED_TYPE = const(0x11)  # invalid/unexpected CBOR
_ERR_INVALID_CBOR = const(0x12)  # error when parsing CBOR
_ERR_MISSING_PARAMETER = const(0x14)  # missing non-optional parameter
_ERR_CREDENTIAL_EXCLUDED = const(0x19)  # valid credential found in the exclude list
_ERR_UNSUPPORTED_ALGORITHM = const(0x26)  # requested COSE algorithm not supported
_ERR_OPERATION_DENIED = const(0x27)  # user declined or timed out
_ERR_KEY_STORE_FULL = const(0x28)  # internal key storage is full
_ERR_UNSUPPORTED_OPTION = const(0x2B)  # unsupported option
_ERR_INVALID_OPTION = const(0x2C)  # not a valid option for current operation
_ERR_KEEPALIVE_CANCEL = const(0x2D)  # pending keep alive was cancelled
_ERR_NO_CREDENTIALS = const(0x2E)  # no valid credentials provided
_ERR_NOT_ALLOWED = const(0x30)  # continuation command not allowed
_ERR_PIN_AUTH_INVALID = const(0x33)  # pinAuth verification failed
_ERR_OTHER = const(0x7F)  # other unspecified error
_ERR_EXTENSION_FIRST = const(0xE0)  # extension specific error

# command status responses
_SW_NO_ERROR = const(0x9000)
_SW_WRONG_LENGTH = const(0x6700)
_SW_DATA_INVALID = const(0x6984)
_SW_CONDITIONS_NOT_SATISFIED = const(0x6985)
_SW_WRONG_DATA = const(0x6A80)
_SW_INS_NOT_SUPPORTED = const(0x6D00)
_SW_CLA_NOT_SUPPORTED = const(0x6E00)

# init response
_CAPFLAG_WINK = const(0x01)  # device supports _CMD_WINK
_CAPFLAG_CBOR = const(0x04)  # device supports _CMD_CBOR
_U2FHID_IF_VERSION = const(2)  # interface version

# register response
_U2F_REGISTER_ID = const(0x05)  # version 2 registration identifier
_U2F_ATT_PRIV_KEY = b"q&\xac+\xf6D\xdca\x86\xad\x83\xef\x1f\xcd\xf1*W\xb5\xcf\xa2\x00\x0b\x8a\xd0'\xe9V\xe8T\xc5\n\x8b"
_U2F_ATT_CERT = b"0\x82\x01\x180\x81\xc0\x02\t\x00\xb1\xd9\x8fBdr\xd3,0\n\x06\x08*\x86H\xce=\x04\x03\x020\x151\x130\x11\x06\x03U\x04\x03\x0c\nTrezor U2F0\x1e\x17\r160429133153Z\x17\r260427133153Z0\x151\x130\x11\x06\x03U\x04\x03\x0c\nTrezor U2F0Y0\x13\x06\x07*\x86H\xce=\x02\x01\x06\x08*\x86H\xce=\x03\x01\x07\x03B\x00\x04\xd9\x18\xbd\xfa\x8aT\xac\x92\xe9\r\xa9\x1f\xcaz\xa2dT\xc0\xd1s61M\xde\x83\xa5K\x86\xb5\xdfN\xf0Re\x9a\x1do\xfc\xb7F\x7f\x1a\xcd\xdb\x8a3\x08\x0b^\xed\x91\x89\x13\xf4C\xa5&\x1b\xc7{h`o\xc10\n\x06\x08*\x86H\xce=\x04\x03\x02\x03G\x000D\x02 $\x1e\x81\xff\xd2\xe5\xe6\x156\x94\xc3U.\x8f\xeb\xd7\x1e\x895\x92\x1c\xb4\x83ACq\x1cv\xea\xee\xf3\x95\x02 _\x80\xeb\x10\xf2\\\xcc9\x8b<\xa8\xa9\xad\xa4\x02\x7f\x93\x13 w\xb7\xab\xcewFZ'\xf5=3\xa1\x1d"
_BOGUS_APPID = b"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
_AAGUID = (
    b"\xd6\xd0\xbd\xc3b\xee\xc4\xdb\xde\x8dzenJD\x87"
)  # First 16 bytes of SHA-256("TREZOR 2")
_BOGUS_PRIV_KEY = b"\xAA" * 32

# authentication control byte
_AUTH_ENFORCE = const(0x03)  # enforce user presence and sign
_AUTH_CHECK_ONLY = const(0x07)  # check only
_AUTH_FLAG_UP = const(1 << 0)  # user present
_AUTH_FLAG_UV = const(1 << 2)  # user verified
_AUTH_FLAG_AT = const(1 << 6)  # attested credential data included
_AUTH_FLAG_ED = const(1 << 7)  # extension data included

# common raw message format (ISO7816-4:2005 mapping)
_APDU_CLA = const(0)  # uint8_t cla;        // Class - reserved
_APDU_INS = const(1)  # uint8_t ins;        // U2F instruction
_APDU_P1 = const(2)  # uint8_t p1;         // U2F parameter 1
_APDU_P2 = const(3)  # uint8_t p2;         // U2F parameter 2
_APDU_LC1 = const(4)  # uint8_t lc1;        // Length field, set to zero
_APDU_LC2 = const(5)  # uint8_t lc2;        // Length field, MSB
_APDU_LC3 = const(6)  # uint8_t lc3;        // Length field, LSB
_APDU_DATA = const(7)  # uint8_t data[1];    // Data field

# Dialog results
_RESULT_NONE = const(0)
_RESULT_CONFIRM = const(1)  # User confirmed.
_RESULT_DECLINE = const(2)  # User declined.
_RESULT_CANCEL = const(3)  # Request was cancelled by _CMD_CANCEL.
_RESULT_TIMEOUT = const(4)  # Request exceeded _FIDO2_CONFIRM_TIMEOUT_MS.

# Generate the authenticatorKeyAgreementKey used for ECDH in authenticatorClientPIN getKeyAgreement.
_KEY_AGREEMENT_PRIVKEY = nist256p1.generate_secret()
_KEY_AGREEMENT_PUBKEY = nist256p1.publickey(_KEY_AGREEMENT_PRIVKEY, False)

# FIDO2 configuration.
_ALLOW_FIDO2 = True
_ALLOW_RESIDENT_CREDENTIALS = True

# The attestation type to use in MakeCredential responses. If false, then self attestation will be used.
_USE_BASIC_ATTESTATION = False

_AUTOCONFIRM = False


class CborError(Exception):
    def __init__(self, code: int):
        self.code = code


def frame_init() -> dict:
    # uint32_t cid;     // Channel identifier
    # uint8_t cmd;      // Command - b7 set
    # uint8_t bcnth;    // Message byte count - high part
    # uint8_t bcntl;    // Message byte count - low part
    # uint8_t data[HID_RPT_SIZE - 7];   // Data payload
    return {
        "cid": 0 | uctypes.UINT32,
        "cmd": 4 | uctypes.UINT8,
        "bcnt": 5 | uctypes.UINT16,
        "data": (7 | uctypes.ARRAY, (_HID_RPT_SIZE - 7) | uctypes.UINT8),
    }


def frame_cont() -> dict:
    # uint32_t cid;                     // Channel identifier
    # uint8_t seq;                      // Sequence number - b7 cleared
    # uint8_t data[HID_RPT_SIZE - 5];   // Data payload
    return {
        "cid": 0 | uctypes.UINT32,
        "seq": 4 | uctypes.UINT8,
        "data": (5 | uctypes.ARRAY, (_HID_RPT_SIZE - 5) | uctypes.UINT8),
    }


def resp_cmd_init() -> dict:
    # uint8_t nonce[8];         // Client application nonce
    # uint32_t cid;             // Channel identifier
    # uint8_t versionInterface; // Interface version
    # uint8_t versionMajor;     // Major version number
    # uint8_t versionMinor;     // Minor version number
    # uint8_t versionBuild;     // Build version number
    # uint8_t capFlags;         // Capabilities flags
    return {
        "nonce": (0 | uctypes.ARRAY, 8 | uctypes.UINT8),
        "cid": 8 | uctypes.UINT32,
        "versionInterface": 12 | uctypes.UINT8,
        "versionMajor": 13 | uctypes.UINT8,
        "versionMinor": 14 | uctypes.UINT8,
        "versionBuild": 15 | uctypes.UINT8,
        "capFlags": 16 | uctypes.UINT8,
    }


def resp_cmd_register(khlen: int, certlen: int, siglen: int) -> dict:
    cert_ofs = 67 + khlen
    sig_ofs = cert_ofs + certlen
    status_ofs = sig_ofs + siglen
    # uint8_t registerId;       // Registration identifier (U2F_REGISTER_ID)
    # uint8_t pubKey[65];       // Generated public key
    # uint8_t keyHandleLen;     // Length of key handle
    # uint8_t keyHandle[khlen]; // Key handle
    # uint8_t cert[certlen];    // Attestation certificate
    # uint8_t sig[siglen];      // Registration signature
    # uint16_t status;
    return {
        "registerId": 0 | uctypes.UINT8,
        "pubKey": (1 | uctypes.ARRAY, 65 | uctypes.UINT8),
        "keyHandleLen": 66 | uctypes.UINT8,
        "keyHandle": (67 | uctypes.ARRAY, khlen | uctypes.UINT8),
        "cert": (cert_ofs | uctypes.ARRAY, certlen | uctypes.UINT8),
        "sig": (sig_ofs | uctypes.ARRAY, siglen | uctypes.UINT8),
        "status": status_ofs | uctypes.UINT16,
    }


# index of keyHandleLen in req_cmd_authenticate struct
_REQ_CMD_AUTHENTICATE_KHLEN = const(64)


def req_cmd_authenticate(khlen: int) -> dict:
    # uint8_t chal[32];         // Challenge
    # uint8_t appId[32];        // Application id
    # uint8_t keyHandleLen;     // Length of key handle
    # uint8_t keyHandle[khlen]; // Key handle
    return {
        "chal": (0 | uctypes.ARRAY, 32 | uctypes.UINT8),
        "appId": (32 | uctypes.ARRAY, 32 | uctypes.UINT8),
        "keyHandleLen": 64 | uctypes.UINT8,
        "keyHandle": (65 | uctypes.ARRAY, khlen | uctypes.UINT8),
    }


def resp_cmd_authenticate(siglen: int) -> dict:
    status_ofs = 5 + siglen
    # uint8_t flags;        // U2F_AUTH_FLAG_ values
    # uint32_t ctr;         // Counter field (big-endian)
    # uint8_t sig[siglen];  // Signature
    # uint16_t status;
    return {
        "flags": 0 | uctypes.UINT8,
        "ctr": 1 | uctypes.UINT32,
        "sig": (5 | uctypes.ARRAY, siglen | uctypes.UINT8),
        "status": status_ofs | uctypes.UINT16,
    }


def overlay_struct(buf, desc):
    desc_size = uctypes.sizeof(desc, uctypes.BIG_ENDIAN)
    if desc_size > len(buf):
        raise ValueError("desc is too big (%d > %d)" % (desc_size, len(buf)))
    return uctypes.struct(uctypes.addressof(buf), desc, uctypes.BIG_ENDIAN)


def make_struct(desc):
    desc_size = uctypes.sizeof(desc, uctypes.BIG_ENDIAN)
    buf = bytearray(desc_size)
    return buf, uctypes.struct(uctypes.addressof(buf), desc, uctypes.BIG_ENDIAN)


class Msg:
    def __init__(
        self, cid: int, cla: int, ins: int, p1: int, p2: int, lc: int, data: bytes
    ) -> None:
        self.cid = cid
        self.cla = cla
        self.ins = ins
        self.p1 = p1
        self.p2 = p2
        self.lc = lc
        self.data = data


class Cmd:
    def __init__(self, cid: int, cmd: int, data: bytes) -> None:
        self.cid = cid
        self.cmd = cmd
        self.data = data

    def to_msg(self) -> Msg:
        cla = self.data[_APDU_CLA]
        ins = self.data[_APDU_INS]
        p1 = self.data[_APDU_P1]
        p2 = self.data[_APDU_P2]
        lc = (
            (self.data[_APDU_LC1] << 16)
            + (self.data[_APDU_LC2] << 8)
            + (self.data[_APDU_LC3])
        )
        data = self.data[_APDU_DATA : _APDU_DATA + lc]
        return Msg(self.cid, cla, ins, p1, p2, lc, data)


async def read_cmd(iface: io.HID) -> Optional[Cmd]:
    desc_init = frame_init()
    desc_cont = frame_cont()
    read = loop.wait(iface.iface_num() | io.POLL_READ)

    buf = await read
    while True:
        ifrm = overlay_struct(buf, desc_init)
        bcnt = ifrm.bcnt
        data = ifrm.data
        datalen = len(data)
        seq = 0

        if ifrm.cmd & _TYPE_MASK == _TYPE_CONT:
            # unexpected cont packet, abort current msg
            if __debug__:
                log.warning(__name__, "_TYPE_CONT")
            return None

        if ifrm.cid == 0 or ((ifrm.cid == _CID_BROADCAST) and (ifrm.cmd != _CMD_INIT)):
            # CID 0 is reserved for future use and _CID_BROADCAST is reserved for channel allocation
            await send_cmd(cmd_error(ifrm.cid, _ERR_INVALID_CID), iface)
            return None

        if bcnt > _MAX_U2FHID_MSG_PAYLOAD_LEN:
            # invalid payload length, abort current msg
            if __debug__:
                log.warning(__name__, "_MAX_U2FHID_MSG_PAYLOAD_LEN")
            await send_cmd(cmd_error(ifrm.cid, _ERR_INVALID_LEN), iface)
            return None

        if datalen < bcnt:
            databuf = bytearray(bcnt)
            utils.memcpy(databuf, 0, data, 0, bcnt)
            data = databuf
        else:
            data = data[:bcnt]

        while datalen < bcnt:
            buf = await loop.race(read, loop.sleep(_CTAP_HID_TIMEOUT_MS * 1000))
            if not isinstance(buf, (bytes, bytearray)):
                await send_cmd(cmd_error(ifrm.cid, _ERR_MSG_TIMEOUT), iface)
                return None

            cfrm = overlay_struct(buf, desc_cont)

            if cfrm.seq == _CMD_INIT:
                # _CMD_INIT frame, cancels current channel
                break

            if cfrm.cid != ifrm.cid:
                # cont frame for a different channel, reply with BUSY and abort
                if __debug__:
                    log.warning(__name__, "_ERR_CHANNEL_BUSY")
                await send_cmd(cmd_error(cfrm.cid, _ERR_CHANNEL_BUSY), iface)
                continue

            if cfrm.seq != seq:
                # cont frame for this channel, but incorrect seq number, abort
                # current msg
                if __debug__:
                    log.warning(__name__, "_ERR_INVALID_SEQ")
                await send_cmd(cmd_error(cfrm.cid, _ERR_INVALID_SEQ), iface)
                return None

            datalen += utils.memcpy(data, datalen, cfrm.data, 0, bcnt - datalen)
            seq += 1
        else:
            return Cmd(ifrm.cid, ifrm.cmd, data)


async def send_cmd(cmd: Cmd, iface: io.HID) -> None:
    init_desc = frame_init()
    cont_desc = frame_cont()
    offset = 0
    seq = 0
    datalen = len(cmd.data)

    buf, frm = make_struct(init_desc)
    frm.cid = cmd.cid
    frm.cmd = cmd.cmd
    frm.bcnt = datalen

    offset += utils.memcpy(frm.data, 0, cmd.data, offset, datalen)
    iface.write(buf)

    if offset < datalen:
        frm = overlay_struct(buf, cont_desc)

    write = loop.wait(iface.iface_num() | io.POLL_WRITE)
    while offset < datalen:
        frm.seq = seq
        copied = utils.memcpy(frm.data, 0, cmd.data, offset, datalen)
        offset += copied
        if copied < _FRAME_CONT_SIZE:
            frm.data[copied:] = bytearray(_FRAME_CONT_SIZE - copied)
        while True:
            await write
            if iface.write(buf) > 0:
                break
        seq += 1


def send_cmd_sync(cmd: Cmd, iface: io.HID) -> None:
    init_desc = frame_init()
    cont_desc = frame_cont()
    offset = 0
    seq = 0
    datalen = len(cmd.data)

    buf, frm = make_struct(init_desc)
    frm.cid = cmd.cid
    frm.cmd = cmd.cmd
    frm.bcnt = datalen

    offset += utils.memcpy(frm.data, 0, cmd.data, offset, datalen)
    iface.write(buf)

    if offset < datalen:
        frm = overlay_struct(buf, cont_desc)

    while offset < datalen:
        frm.seq = seq
        copied = utils.memcpy(frm.data, 0, cmd.data, offset, datalen)
        offset += copied
        if copied < _FRAME_CONT_SIZE:
            frm.data[copied:] = bytearray(_FRAME_CONT_SIZE - copied)
        iface.write_blocking(buf, 1000)
        seq += 1


async def handle_reports(iface: io.HID) -> None:
    dialog_mgr = DialogManager(iface)

    while True:
        try:
            req = await read_cmd(iface)
            if req is None:
                continue
            if dialog_mgr.is_busy() and req.cid not in (
                dialog_mgr.get_cid(),
                _CID_BROADCAST,
            ):
                resp = cmd_error(req.cid, _ERR_CHANNEL_BUSY)  # type: Optional[Cmd]
            else:
                resp = dispatch_cmd(req, dialog_mgr)
            if resp is not None:
                await send_cmd(resp, iface)
        except Exception as e:
            log.exception(__name__, e)


class KeepaliveCallback:
    def __init__(self, cid: int, iface: io.HID) -> None:
        self.cid = cid
        self.iface = iface

    def __call__(self) -> None:
        send_cmd_sync(cmd_keepalive(self.cid, _KEEPALIVE_STATUS_PROCESSING), self.iface)


async def verify_user(keepalive_callback: KeepaliveCallback) -> bool:
    from apps.common.request_pin import verify_user_pin, PinCancelled, PinInvalid
    import trezor.pin

    try:
        trezor.pin.keepalive_callback = keepalive_callback
        await verify_user_pin()
        ret = True
    except (PinCancelled, PinInvalid):
        ret = False
    finally:
        trezor.pin.keepalive_callback = None

    return ret


async def confirm(*args: Any, **kwargs: Any) -> bool:
    if _AUTOCONFIRM:
        return True
    dialog = Confirm(*args, **kwargs)
    if __debug__:
        return await loop.race(dialog, confirm_signal()) is CONFIRMED
    else:
        return await dialog is CONFIRMED


async def confirm_pageable(*args: Any, **kwargs: Any) -> bool:
    if _AUTOCONFIRM:
        return True
    dialog = ConfirmPageable(*args, **kwargs)
    if __debug__:
        return await loop.race(dialog, confirm_signal()) is CONFIRMED
    else:
        return await dialog is CONFIRMED


class State:
    def __init__(self, cid: int, iface: io.HID) -> None:
        self.cid = cid
        self.iface = iface

    def keepalive_status(self) -> Optional[int]:
        return None

    def timeout_ms(self) -> int:
        raise NotImplementedError

    async def confirm_dialog(self) -> bool:
        pass

    async def on_confirm(self) -> None:
        pass

    async def on_decline(self) -> None:
        pass

    async def on_timeout(self) -> None:
        pass

    async def on_cancel(self) -> None:
        pass


class U2fState(State, ConfirmInfo):
    def __init__(
        self, cid: int, iface: io.HID, req_data: bytes, cred: "Credential"
    ) -> None:
        State.__init__(self, cid, iface)
        ConfirmInfo.__init__(self)
        self._cred = cred
        self._req_data = req_data
        self.load_icon(self._cred.rp_id_hash)

    def timeout_ms(self) -> int:
        return _U2F_CONFIRM_TIMEOUT_MS

    def app_name(self) -> str:
        return self._cred.app_name()

    def account_name(self) -> Optional[str]:
        return self._cred.account_name()


class U2fConfirmRegister(U2fState):
    def __init__(
        self, cid: int, iface: io.HID, req_data: bytes, cred: "U2fCredential"
    ) -> None:
        super().__init__(cid, iface, req_data, cred)

    async def confirm_dialog(self) -> bool:
        if self._cred.rp_id_hash == _BOGUS_APPID:
            text = Text("U2F", ui.ICON_WRONG, ui.RED)
            text.normal(
                "Another U2F device", "was used to register", "in this application."
            )
            return await confirm(text, confirm=None, cancel="Close")
        else:
            content = ConfirmContent(self)
            return await confirm(content)

    def get_header(self) -> str:
        return "U2F Register"

    def __eq__(self, other: object) -> bool:
        return (
            isinstance(other, U2fConfirmRegister)
            and self.cid == other.cid
            and self._req_data == other._req_data
        )


class U2fConfirmAuthenticate(U2fState):
    def __init__(
        self, cid: int, iface: io.HID, req_data: bytes, cred: "Credential"
    ) -> None:
        super().__init__(cid, iface, req_data, cred)

    def get_header(self) -> str:
        return "U2F Authenticate"

    async def confirm_dialog(self) -> bool:
        content = ConfirmContent(self)
        return await confirm(content)

    def __eq__(self, other: object) -> bool:
        return (
            isinstance(other, U2fConfirmAuthenticate)
            and self.cid == other.cid
            and self._req_data == other._req_data
        )


class Fido2State(State):
    def __init__(self, cid: int, iface: io.HID) -> None:
        super().__init__(cid, iface)

    def keepalive_status(self) -> int:
        return _KEEPALIVE_STATUS_UP_NEEDED

    def timeout_ms(self) -> int:
        return _FIDO2_CONFIRM_TIMEOUT_MS

    async def on_confirm(self) -> None:
        cmd = cbor_error(self.cid, _ERR_OPERATION_DENIED)
        await send_cmd(cmd, self.iface)

    async def on_decline(self) -> None:
        cmd = cbor_error(self.cid, _ERR_OPERATION_DENIED)
        await send_cmd(cmd, self.iface)

    async def on_timeout(self) -> None:
        await self.on_decline()

    async def on_cancel(self) -> None:
        cmd = cbor_error(self.cid, _ERR_KEEPALIVE_CANCEL)
        await send_cmd(cmd, self.iface)


class Fido2ConfirmMakeCredential(Fido2State, ConfirmInfo):
    def __init__(
        self,
        cid: int,
        iface: io.HID,
        client_data_hash: bytes,
        cred: "Fido2Credential",
        resident: bool,
        user_verification: bool,
    ) -> None:
        Fido2State.__init__(self, cid, iface)
        ConfirmInfo.__init__(self)
        self._client_data_hash = client_data_hash
        self._cred = cred
        self._resident = resident
        self._user_verification = user_verification
        self.load_icon(cred.rp_id_hash)

    def get_header(self) -> str:
        return "FIDO2 Register"

    def app_name(self) -> str:
        return self._cred.app_name()

    def account_name(self) -> Optional[str]:
        return self._cred.account_name()

    async def confirm_dialog(self) -> bool:
        content = ConfirmContent(self)
        if not await confirm(content):
            return False
        if self._user_verification:
            return await verify_user(KeepaliveCallback(self.cid, self.iface))
        return True

    async def on_confirm(self) -> None:
        self._cred.generate_id()
        send_cmd_sync(cmd_keepalive(self.cid, _KEEPALIVE_STATUS_PROCESSING), self.iface)
        response_data = cbor_make_credential_sign(
            self._client_data_hash, self._cred, self._user_verification
        )

        cmd = Cmd(self.cid, _CMD_CBOR, bytes([_ERR_NONE]) + response_data)
        if self._resident:
            send_cmd_sync(
                cmd_keepalive(self.cid, _KEEPALIVE_STATUS_PROCESSING), self.iface
            )
            from apps.common.storage.webauthn import store_resident_credential

            if not store_resident_credential(self._cred):
                cmd = cbor_error(self.cid, _ERR_KEY_STORE_FULL)
        await send_cmd(cmd, self.iface)


class Fido2ConfirmExcluded(Fido2ConfirmMakeCredential):
    def __init__(self, cid: int, iface: io.HID, cred: "Fido2Credential") -> None:
        super().__init__(cid, iface, b"", cred, resident=False, user_verification=False)

    async def on_confirm(self) -> None:
        cmd = cbor_error(self.cid, _ERR_CREDENTIAL_EXCLUDED)
        await send_cmd(cmd, self.iface)

        text = Text("FIDO2 Register", ui.ICON_WRONG, ui.RED)
        text.normal("This device is already", "registered with", self._cred.rp_id + ".")
        await confirm(text, confirm=None, cancel="Close")


class Fido2ConfirmGetAssertion(Fido2State, ConfirmInfo, Pageable):
    def __init__(
        self,
        cid: int,
        iface: io.HID,
        client_data_hash: bytes,
        creds: List["Credential"],
        hmac_secret: Optional[dict],
        resident: bool,
        user_verification: bool,
    ) -> None:
        Fido2State.__init__(self, cid, iface)
        ConfirmInfo.__init__(self)
        Pageable.__init__(self)
        self._client_data_hash = client_data_hash
        self._creds = creds
        self._hmac_secret = hmac_secret
        self._resident = resident
        self._user_verification = user_verification
        self.load_icon(self._creds[0].rp_id_hash)

    def get_header(self) -> str:
        return "FIDO2 Authenticate"

    def app_name(self) -> str:
        return self._creds[self.page()].app_name()

    def account_name(self) -> Optional[str]:
        return self._creds[self.page()].account_name()

    def page_count(self) -> int:
        return len(self._creds)

    async def confirm_dialog(self) -> bool:
        content = ConfirmContent(self)
        if not await confirm_pageable(self, content):
            return False
        if self._user_verification:
            return await verify_user(KeepaliveCallback(self.cid, self.iface))
        return True

    async def on_confirm(self) -> None:
        cred = self._creds[self.page()]
        try:
            send_cmd_sync(
                cmd_keepalive(self.cid, _KEEPALIVE_STATUS_PROCESSING), self.iface
            )
            response_data = cbor_get_assertion_sign(
                self._client_data_hash,
                cred.rp_id_hash,
                cred,
                self._hmac_secret,
                self._resident,
                True,
                self._user_verification,
            )
            cmd = Cmd(self.cid, _CMD_CBOR, bytes([_ERR_NONE]) + response_data)
        except CborError as e:
            cmd = cbor_error(self.cid, e.code)
        except KeyError:
            cmd = cbor_error(self.cid, _ERR_MISSING_PARAMETER)
        except Exception:
            cmd = cbor_error(self.cid, _ERR_OPERATION_DENIED)

        await send_cmd(cmd, self.iface)


class Fido2ConfirmNoPin(State):
    def __init__(self, cid: int, iface: io.HID) -> None:
        super().__init__(cid, iface)

    def timeout_ms(self) -> int:
        return _FIDO2_CONFIRM_TIMEOUT_MS

    async def confirm_dialog(self) -> bool:
        text = Text("FIDO2 Verify User", ui.ICON_WRONG, ui.RED)
        text.normal("Unable to verify user.", "Please enable PIN", "protection.")
        return await confirm(text, confirm=None, cancel="Close")


class Fido2ConfirmNoCredentials(Fido2ConfirmGetAssertion):
    def __init__(self, cid: int, iface: io.HID, rp_id: str) -> None:

        from apps.webauthn.credential import Fido2Credential

        cred = Fido2Credential()
        cred.rp_id = rp_id
        super().__init__(
            cid, iface, b"", [cred], {}, resident=False, user_verification=False
        )

    async def on_confirm(self) -> None:
        cmd = cbor_error(self.cid, _ERR_NO_CREDENTIALS)
        await send_cmd(cmd, self.iface)

        text = Text("FIDO2 Authenticate", ui.ICON_WRONG, ui.RED)
        text.normal(
            "This device is not", "registered with", self._creds[0].app_name() + "."
        )
        await confirm(text, confirm=None, cancel="Close")


class Fido2ConfirmReset(Fido2State):
    def __init__(self, cid: int, iface: io.HID) -> None:
        super().__init__(cid, iface)

    async def confirm_dialog(self) -> bool:
        text = Text("FIDO2 Reset", ui.ICON_CONFIG)
        text.normal("Do you really want to")
        text.bold("erase all credentials?")
        return await confirm(text)

    async def on_confirm(self) -> None:
        from apps.common.storage.webauthn import erase_resident_credentials

        erase_resident_credentials()
        cmd = Cmd(self.cid, _CMD_CBOR, bytes([_ERR_NONE]))
        await send_cmd(cmd, self.iface)


class DialogManager:
    def __init__(self, iface: io.HID) -> None:
        self.iface = iface
        self._clear()

    def _clear(self) -> None:
        self.state = None  # type: Optional[State]
        self.deadline = 0
        self.result = _RESULT_NONE
        self.workflow = None  # type: Optional[Coroutine]
        self.keepalive = None  # type: Optional[Coroutine]

    def reset_timeout(self) -> None:
        if self.state is not None:
            self.deadline = utime.ticks_ms() + self.state.timeout_ms()

    def reset(self) -> None:
        if self.workflow is not None:
            loop.close(self.workflow)
        if self.keepalive is not None:
            loop.close(self.keepalive)
        self._clear()

    def get_cid(self) -> int:
        if self.state is None:
            return 0
        return self.state.cid

    def is_busy(self) -> bool:
        if utime.ticks_ms() >= self.deadline:
            self.reset()
        return bool(workflow.tasks or self.workflow)

    def compare(self, state: State) -> bool:
        if self.state != state:
            return False
        if utime.ticks_ms() >= self.deadline:
            self.reset()
            return False
        return True

    def set_state(self, state: State) -> bool:
        if self.is_busy():
            return False

        self.state = state
        self.reset_timeout()
        self.result = _RESULT_NONE
        if state.keepalive_status() is not None:
            self.keepalive = self.keepalive_loop()
            loop.schedule(self.keepalive)
        else:
            self.keepalive = None
        self.workflow = self.dialog_workflow()
        loop.schedule(self.workflow)
        return True

    async def keepalive_loop(self) -> None:
        if not isinstance(self.state, Fido2State):
            return
        while utime.ticks_ms() < self.deadline:
            cmd = cmd_keepalive(self.state.cid, self.state.keepalive_status())
            await send_cmd(cmd, self.iface)
            await loop.sleep(_KEEPALIVE_INTERVAL_MS * 1000)

        self.result = _RESULT_TIMEOUT
        self.reset()

    async def dialog_workflow(self) -> None:
        if self.workflow is None or self.state is None:
            return

        try:
            workflow.on_start(self.workflow)
            if await self.state.confirm_dialog():
                self.result = _RESULT_CONFIRM
            else:
                self.result = _RESULT_DECLINE
        finally:
            if self.keepalive is not None:
                loop.close(self.keepalive)
            self.keepalive = None

            if self.result == _RESULT_CONFIRM:
                await self.state.on_confirm()
            elif self.result == _RESULT_CANCEL:
                await self.state.on_cancel()
            elif self.result == _RESULT_TIMEOUT:
                await self.state.on_timeout()
            else:
                await self.state.on_decline()

            workflow.on_close(self.workflow)
            self.workflow = None


def dispatch_cmd(req: Cmd, dialog_mgr: DialogManager) -> Optional[Cmd]:
    if req.cmd == _CMD_MSG:
        try:
            m = req.to_msg()
        except IndexError:
            return cmd_error(req.cid, _ERR_INVALID_LEN)

        if m.cla != 0:
            if __debug__:
                log.warning(__name__, "_SW_CLA_NOT_SUPPORTED")
            return msg_error(req.cid, _SW_CLA_NOT_SUPPORTED)

        if m.lc + _APDU_DATA > len(req.data):
            if __debug__:
                log.warning(__name__, "_SW_WRONG_LENGTH")
            return msg_error(req.cid, _SW_WRONG_LENGTH)

        if m.ins == _MSG_REGISTER:
            if __debug__:
                log.debug(__name__, "_MSG_REGISTER")
            return msg_register(m, dialog_mgr)
        elif m.ins == _MSG_AUTHENTICATE:
            if __debug__:
                log.debug(__name__, "_MSG_AUTHENTICATE")
            return msg_authenticate(m, dialog_mgr)
        elif m.ins == _MSG_VERSION:
            if __debug__:
                log.debug(__name__, "_MSG_VERSION")
            return msg_version(m)
        else:
            if __debug__:
                log.warning(__name__, "_SW_INS_NOT_SUPPORTED: %d", m.ins)
            return msg_error(req.cid, _SW_INS_NOT_SUPPORTED)

    elif req.cmd == _CMD_INIT:
        if __debug__:
            log.debug(__name__, "_CMD_INIT")
        return cmd_init(req)
    elif req.cmd == _CMD_PING:
        if __debug__:
            log.debug(__name__, "_CMD_PING")
        return req
    elif req.cmd == _CMD_WINK:
        if __debug__:
            log.debug(__name__, "_CMD_WINK")
        loop.schedule(ui.alert())
        return req
    elif req.cmd == _CMD_CBOR and _ALLOW_FIDO2:
        if not req.data:
            return cmd_error(req.cid, _ERR_INVALID_LEN)
        if req.data[0] == _CBOR_MAKE_CREDENTIAL:
            if __debug__:
                log.debug(__name__, "_CBOR_MAKE_CREDENTIAL")
            return cbor_make_credential(req, dialog_mgr)
        elif req.data[0] == _CBOR_GET_ASSERTION:
            if __debug__:
                log.debug(__name__, "_CBOR_GET_ASSERTION")
            return cbor_get_assertion(req, dialog_mgr)
        elif req.data[0] == _CBOR_GET_INFO:
            if __debug__:
                log.debug(__name__, "_CBOR_GET_INFO")
            return cbor_get_info(req)
        elif req.data[0] == _CBOR_CLIENT_PIN:
            if __debug__:
                log.debug(__name__, "_CBOR_CLIENT_PIN")
            return cbor_client_pin(req)
        elif req.data[0] == _CBOR_RESET:
            if __debug__:
                log.debug(__name__, "_CBOR_RESET")
            return cbor_reset(req, dialog_mgr)
        elif req.data[0] == _CBOR_GET_NEXT_ASSERTION:
            if __debug__:
                log.debug(__name__, "_CBOR_GET_NEXT_ASSERTION")
            return cbor_error(req.cid, _ERR_NOT_ALLOWED)
        else:
            if __debug__:
                log.warning(__name__, "_ERR_INVALID_CMD _CMD_CBOR %d", req.data[0])
            return cbor_error(req.cid, _ERR_INVALID_CMD)

    elif req.cmd == _CMD_CANCEL:
        if __debug__:
            log.debug(__name__, "_CMD_CANCEL")
        dialog_mgr.result = _RESULT_CANCEL
        dialog_mgr.reset()
        return None
    else:
        if __debug__:
            log.warning(__name__, "_ERR_INVALID_CMD: %d", req.cmd)
        return cmd_error(req.cid, _ERR_INVALID_CMD)


def cmd_init(req: Cmd) -> Cmd:
    if req.cid == _CID_BROADCAST:
        # uint32_t except 0 and 0xffffffff
        resp_cid = random.uniform(0xFFFFFFFE) + 1
    else:
        resp_cid = req.cid

    buf, resp = make_struct(resp_cmd_init())
    utils.memcpy(resp.nonce, 0, req.data, 0, len(req.data))
    resp.cid = resp_cid
    resp.versionInterface = _U2FHID_IF_VERSION
    resp.versionMajor = 2
    resp.versionMinor = 0
    resp.versionBuild = 0
    resp.capFlags = _CAPFLAG_WINK | _CAPFLAG_CBOR

    return Cmd(req.cid, req.cmd, buf)


def msg_register(req: Msg, dialog_mgr: DialogManager) -> Cmd:
    if not storage.is_initialized():
        if __debug__:
            log.warning(__name__, "not initialized")
        return msg_error(req.cid, _SW_CONDITIONS_NOT_SATISFIED)

    # check length of input data
    if len(req.data) != 64:
        if __debug__:
            log.warning(__name__, "_SW_WRONG_LENGTH req.data")
        return msg_error(req.cid, _SW_WRONG_LENGTH)

    from apps.webauthn.credential import U2fCredential

    # parse challenge and rp_id_hash
    chal = req.data[:32]

    cred = U2fCredential()
    cred.rp_id_hash = bytes(req.data[32:])
    cred.generate_key_handle()

    # check equality with last request
    new_state = U2fConfirmRegister(req.cid, dialog_mgr.iface, req.data, cred)
    if not dialog_mgr.compare(new_state):
        if not dialog_mgr.set_state(new_state):
            return msg_error(req.cid, _SW_CONDITIONS_NOT_SATISFIED)
    dialog_mgr.reset_timeout()

    # wait for a button or continue
    if dialog_mgr.result != _RESULT_CONFIRM:
        if __debug__:
            log.info(__name__, "waiting for button")
        return msg_error(req.cid, _SW_CONDITIONS_NOT_SATISFIED)

    # sign the registration challenge and return
    if __debug__:
        log.info(__name__, "signing register")
    buf = msg_register_sign(chal, cred)

    dialog_mgr.reset()

    return Cmd(req.cid, _CMD_MSG, buf)


def msg_register_sign(challenge: bytes, cred: "U2fCredential") -> bytes:
    pubkey = nist256p1.publickey(cred.private_key(), False)

    # hash the request data together with keyhandle and pubkey
    dig = hashlib.sha256()
    dig.update(b"\x00")  # uint8_t reserved;
    dig.update(cred.rp_id_hash)  # uint8_t appId[32];
    dig.update(challenge)  # uint8_t chal[32];
    dig.update(cred.id)  # uint8_t keyHandle[64];
    dig.update(pubkey)  # uint8_t pubKey[65];

    # sign the digest and convert to der
    sig = nist256p1.sign(_U2F_ATT_PRIV_KEY, dig.digest(), False)
    sig = der.encode_seq((sig[1:33], sig[33:]))

    # pack to a response
    buf, resp = make_struct(
        resp_cmd_register(len(cred.id), len(_U2F_ATT_CERT), len(sig))
    )
    resp.registerId = _U2F_REGISTER_ID
    utils.memcpy(resp.pubKey, 0, pubkey, 0, len(pubkey))
    resp.keyHandleLen = len(cred.id)
    utils.memcpy(resp.keyHandle, 0, cred.id, 0, len(cred.id))
    utils.memcpy(resp.cert, 0, _U2F_ATT_CERT, 0, len(_U2F_ATT_CERT))
    utils.memcpy(resp.sig, 0, sig, 0, len(sig))
    resp.status = _SW_NO_ERROR

    return buf


def msg_authenticate(req: Msg, dialog_mgr: DialogManager) -> Cmd:
    if not storage.is_initialized():
        if __debug__:
            log.warning(__name__, "not initialized")
        return msg_error(req.cid, _SW_CONDITIONS_NOT_SATISFIED)

    # we need at least keyHandleLen
    if len(req.data) <= _REQ_CMD_AUTHENTICATE_KHLEN:
        if __debug__:
            log.warning(__name__, "_SW_WRONG_LENGTH req.data")
        return msg_error(req.cid, _SW_WRONG_LENGTH)

    # check keyHandleLen
    khlen = req.data[_REQ_CMD_AUTHENTICATE_KHLEN]
    auth = overlay_struct(req.data, req_cmd_authenticate(khlen))

    from apps.webauthn.credential import Credential

    cred = Credential.from_bytes(auth.keyHandle, bytes(auth.appId))
    if cred is None:
        # specific error logged in msg_authenticate_genkey
        return msg_error(req.cid, _SW_WRONG_DATA)

    # if _AUTH_CHECK_ONLY is requested, return, because keyhandle has been checked already
    if req.p1 == _AUTH_CHECK_ONLY:
        if __debug__:
            log.info(__name__, "_AUTH_CHECK_ONLY")
        return msg_error(req.cid, _SW_CONDITIONS_NOT_SATISFIED)

    # from now on, only _AUTH_ENFORCE is supported
    if req.p1 != _AUTH_ENFORCE:
        if __debug__:
            log.info(__name__, "_AUTH_ENFORCE")
        return msg_error(req.cid, _SW_WRONG_DATA)

    # check equality with last request
    new_state = U2fConfirmAuthenticate(req.cid, dialog_mgr.iface, req.data, cred)
    if not dialog_mgr.compare(new_state):
        if not dialog_mgr.set_state(new_state):
            return msg_error(req.cid, _SW_CONDITIONS_NOT_SATISFIED)
    dialog_mgr.reset_timeout()

    # wait for a button or continue
    if dialog_mgr.result != _RESULT_CONFIRM:
        if __debug__:
            log.info(__name__, "waiting for button")
        return msg_error(req.cid, _SW_CONDITIONS_NOT_SATISFIED)

    # sign the authentication challenge and return
    if __debug__:
        log.info(__name__, "signing authentication")
    buf = msg_authenticate_sign(auth.chal, auth.appId, cred)

    dialog_mgr.reset()

    return Cmd(req.cid, _CMD_MSG, buf)


def msg_authenticate_sign(
    challenge: bytes, rp_id_hash: bytes, cred: "Credential"
) -> bytes:
    flags = bytes([_AUTH_FLAG_UP])

    # get next counter
    ctr = cred.next_signature_counter()
    ctrbuf = ustruct.pack(">L", ctr)

    # hash input data together with counter
    dig = hashlib.sha256()
    dig.update(rp_id_hash)  # uint8_t appId[32];
    dig.update(flags)  # uint8_t flags;
    dig.update(ctrbuf)  # uint8_t ctr[4];
    dig.update(challenge)  # uint8_t chal[32];

    # sign the digest and convert to der
    sig = nist256p1.sign(cred.private_key(), dig.digest(), False)
    sig = der.encode_seq((sig[1:33], sig[33:]))

    # pack to a response
    buf, resp = make_struct(resp_cmd_authenticate(len(sig)))
    resp.flags = flags[0]
    resp.ctr = ctr
    utils.memcpy(resp.sig, 0, sig, 0, len(sig))
    resp.status = _SW_NO_ERROR

    return buf


def msg_version(req: Msg) -> Cmd:
    if req.data:
        return msg_error(req.cid, _SW_WRONG_LENGTH)
    return Cmd(req.cid, _CMD_MSG, b"U2F_V2\x90\x00")  # includes _SW_NO_ERROR


def msg_error(cid: int, code: int) -> Cmd:
    return Cmd(cid, _CMD_MSG, ustruct.pack(">H", code))


def cmd_error(cid: int, code: int) -> Cmd:
    return Cmd(cid, _CMD_ERROR, ustruct.pack(">B", code))


def cbor_error(cid: int, code: int) -> Cmd:
    return Cmd(cid, _CMD_CBOR, ustruct.pack(">B", code))


def credentials_from_descriptor_list(
    descriptor_list: List[dict], rp_id_hash: bytes
) -> List["Credential"]:

    from apps.webauthn.credential import Credential

    cred_list = []
    for credential_descriptor in descriptor_list:
        credential_type = credential_descriptor["type"]
        if not isinstance(credential_type, str):
            raise TypeError
        if credential_type != "public-key":
            continue

        credential_id = credential_descriptor["id"]
        if not isinstance(credential_id, (bytes, bytearray)):
            raise TypeError

        cred = Credential.from_bytes(credential_id, rp_id_hash)
        if cred is not None:
            cred_list.append(cred)

    return cred_list


def algorithms_from_pub_key_cred_params(pub_key_cred_params: List[dict]) -> List[int]:
    alg_list = []
    for pkcp in pub_key_cred_params:
        pub_key_cred_type = pkcp["type"]
        if not isinstance(pub_key_cred_type, str):
            raise TypeError
        if pub_key_cred_type != "public-key":
            continue

        pub_key_cred_alg = pkcp["alg"]
        if not isinstance(pub_key_cred_alg, int):
            raise TypeError
        alg_list.append(pub_key_cred_alg)
    return alg_list


def cbor_make_credential(req: Cmd, dialog_mgr: DialogManager) -> Optional[Cmd]:
    from apps.webauthn.knownapps import knownapps

    if not storage.is_initialized():
        if __debug__:
            log.warning(__name__, "not initialized")
        return cbor_error(req.cid, _ERR_OPERATION_DENIED)

    try:
        param = cbor.decode(req.data[1:])
        rp = param[_MAKECRED_CMD_RP]
        rp_id = rp["id"]
        rp_id_hash = hashlib.sha256(rp_id).digest()

        # Prepare the new credential.
        user = param[_MAKECRED_CMD_USER]

        from apps.webauthn.credential import Fido2Credential

        cred = Fido2Credential()
        cred.rp_id = rp_id
        cred.rp_id_hash = rp_id_hash
        cred.rp_name = param[_MAKECRED_CMD_RP].get("name", None)
        cred.user_id = user["id"]
        cred.user_name = user.get("name", None)
        cred.user_display_name = user.get("displayName", None)

        # Check if any of the credential descriptors in the exclude list belong to this authenticator.
        exclude_list = param.get(_MAKECRED_CMD_EXCLUDE_LIST, [])
        if credentials_from_descriptor_list(exclude_list, rp_id_hash):
            # This authenticator is already registered.
            if not dialog_mgr.set_state(
                Fido2ConfirmExcluded(req.cid, dialog_mgr.iface, cred)
            ):
                return cmd_error(req.cid, _ERR_CHANNEL_BUSY)
            return None

        # Check that the relying party supports ECDSA P-256 with SHA-256. We don't support any other algorithms.
        pub_key_cred_params = param[_MAKECRED_CMD_PUB_KEY_CRED_PARAMS]
        if _COSE_ALG_ES256 not in algorithms_from_pub_key_cred_params(
            pub_key_cred_params
        ):
            return cbor_error(req.cid, _ERR_UNSUPPORTED_ALGORITHM)

        # Get options.
        options = param.get(_MAKECRED_CMD_OPTIONS, {})
        resident_key = options.get("rk", False)
        user_verification = options.get("uv", False)

        # Get supported extensions.
        cred.hmac_secret = param.get(_MAKECRED_CMD_EXTENSIONS, {}).get(
            "hmac-secret", False
        )

        client_data_hash = param[_MAKECRED_CMD_CLIENT_DATA_HASH]
    except TypeError:
        return cbor_error(req.cid, _ERR_CBOR_UNEXPECTED_TYPE)
    except KeyError:
        return cbor_error(req.cid, _ERR_MISSING_PARAMETER)
    except Exception:
        return cbor_error(req.cid, _ERR_INVALID_CBOR)

    cred.use_sign_count = knownapps.get(rp_id_hash, {}).get("use_sign_count", True)

    # Check data types.
    if (
        not cred.check_data_types()
        or not isinstance(user.get("icon", ""), str)
        or not isinstance(rp.get("icon", ""), str)
        or not isinstance(client_data_hash, (bytes, bytearray))
        or not isinstance(resident_key, bool)
        or not isinstance(user_verification, bool)
    ):
        return cbor_error(req.cid, _ERR_CBOR_UNEXPECTED_TYPE)

    # Check options.
    if "up" in options:
        return cbor_error(req.cid, _ERR_INVALID_OPTION)

    if resident_key and not _ALLOW_RESIDENT_CREDENTIALS:
        return cbor_error(req.cid, _ERR_UNSUPPORTED_OPTION)

    if user_verification and not config.has_pin():
        # User verification requested, but PIN is not enabled.
        state_set = dialog_mgr.set_state(Fido2ConfirmNoPin(req.cid, dialog_mgr.iface))
        if state_set:
            return cbor_error(req.cid, _ERR_UNSUPPORTED_OPTION)
        else:
            return cmd_error(req.cid, _ERR_CHANNEL_BUSY)

    # Check that the pinAuth parameter is absent. Client PIN is not supported.
    if _MAKECRED_CMD_PIN_AUTH in param:
        return cbor_error(req.cid, _ERR_PIN_AUTH_INVALID)

    # Ask user to confirm registration.
    state_set = dialog_mgr.set_state(
        Fido2ConfirmMakeCredential(
            req.cid,
            dialog_mgr.iface,
            client_data_hash,
            cred,
            resident_key,
            user_verification,
        )
    )

    if not state_set:
        return cmd_error(req.cid, _ERR_CHANNEL_BUSY)

    return None


def cbor_make_credential_sign(
    client_data_hash: bytes, cred: "Fido2Credential", user_verification: bool
) -> bytes:
    privkey = cred.private_key()
    pubkey = nist256p1.publickey(privkey, False)

    flags = _AUTH_FLAG_UP | _AUTH_FLAG_AT
    if user_verification:
        flags |= _AUTH_FLAG_UV

    # Encode the authenticator data (Credential ID, its public key and extensions).
    credential_pub_key = cbor.encode(
        {
            _COSE_ALG_KEY: _COSE_ALG_ES256,
            _COSE_KEY_TYPE_KEY: _COSE_KEY_TYPE_EC2,
            _COSE_CURVE_KEY: _COSE_CURVE_P256,
            _COSE_X_COORD_KEY: pubkey[1:33],
            _COSE_Y_COORD_KEY: pubkey[33:],
        }
    )

    att_cred_data = (
        _AAGUID + len(cred.id).to_bytes(2, "big") + cred.id + credential_pub_key
    )

    extensions = b""
    if cred.hmac_secret:
        extensions = cbor.encode({"hmac-secret": True})
        flags |= _AUTH_FLAG_ED

    ctr = cred.next_signature_counter()

    authenticator_data = (
        cred.rp_id_hash
        + bytes([flags])
        + ctr.to_bytes(4, "big")
        + att_cred_data
        + extensions
    )

    # Compute the attestation signature of the authenticator data.
    if _USE_BASIC_ATTESTATION:
        privkey = _U2F_ATT_PRIV_KEY

    dig = hashlib.sha256()
    dig.update(authenticator_data)
    dig.update(client_data_hash)
    sig = nist256p1.sign(privkey, dig.digest(), False)
    sig = der.encode_seq((sig[1:33], sig[33:]))

    # Encode the authenticatorMakeCredential response data.
    attestation_statement = {"alg": _COSE_ALG_ES256, "sig": sig}
    if _USE_BASIC_ATTESTATION:
        attestation_statement["x5c"] = [_U2F_ATT_CERT]

    return cbor.encode(
        {
            _MAKECRED_RESP_FMT: "packed",
            _MAKECRED_RESP_AUTH_DATA: authenticator_data,
            _MAKECRED_RESP_ATT_STMT: attestation_statement,
        }
    )


def cbor_get_assertion(req: Cmd, dialog_mgr: DialogManager) -> Optional[Cmd]:
    if not storage.is_initialized():
        if __debug__:
            log.warning(__name__, "not initialized")
        return cbor_error(req.cid, _ERR_OPERATION_DENIED)

    try:
        param = cbor.decode(req.data[1:])
        rp_id = param[_GETASSERT_CMD_RP_ID]
        rp_id_hash = hashlib.sha256(rp_id).digest()

        allow_list = param.get(_GETASSERT_CMD_ALLOW_LIST, [])
        if allow_list:
            # Get all credentials from the allow list that belong to this authenticator.
            cred_list = credentials_from_descriptor_list(allow_list, rp_id_hash)
            for cred in cred_list:
                if cred.rp_id is None:
                    cred.rp_id = rp_id
            resident = False
        else:
            # Allow list is empty. Get resident credentials.
            if _ALLOW_RESIDENT_CREDENTIALS:
                from apps.common.storage.webauthn import get_resident_credentials

                cred_list = get_resident_credentials(rp_id_hash)
            else:
                cred_list = []
            resident = True

        # Sort credentials by time of creation.
        cred_list.sort()

        # Check that the pinAuth parameter is absent. Client PIN is not supported.
        if _GETASSERT_CMD_PIN_AUTH in param:
            return cbor_error(req.cid, _ERR_PIN_AUTH_INVALID)

        # Get options.
        options = param.get(_GETASSERT_CMD_OPTIONS, {})
        user_presence = options.get("up", True)
        user_verification = options.get("uv", False)

        # Get supported extensions.
        hmac_secret = param.get(_GETASSERT_CMD_EXTENSIONS, {}).get("hmac-secret", None)

        client_data_hash = param[_GETASSERT_CMD_CLIENT_DATA_HASH]
    except TypeError:
        return cbor_error(req.cid, _ERR_CBOR_UNEXPECTED_TYPE)
    except KeyError:
        return cbor_error(req.cid, _ERR_MISSING_PARAMETER)
    except Exception:
        return cbor_error(req.cid, _ERR_INVALID_CBOR)

    # Check data types.
    if (
        not isinstance(hmac_secret, (dict, type(None)))
        or not isinstance(client_data_hash, (bytes, bytearray))
        or not isinstance(user_presence, bool)
        or not isinstance(user_verification, bool)
    ):
        return cbor_error(req.cid, _ERR_CBOR_UNEXPECTED_TYPE)

    # Check options.
    if "rk" in options:
        return cbor_error(req.cid, _ERR_INVALID_OPTION)

    if user_verification and not config.has_pin():
        # User verification requested, but PIN is not enabled.
        state_set = dialog_mgr.set_state(Fido2ConfirmNoPin(req.cid, dialog_mgr.iface))
        if state_set:
            return cbor_error(req.cid, _ERR_UNSUPPORTED_OPTION)
        else:
            return cmd_error(req.cid, _ERR_CHANNEL_BUSY)

    if not cred_list:
        # No credentials. This authenticator is not registered.
        if user_presence:
            state_set = dialog_mgr.set_state(
                Fido2ConfirmNoCredentials(req.cid, dialog_mgr.iface, rp_id)
            )
        else:
            return cbor_error(req.cid, _ERR_NO_CREDENTIALS)
    elif not user_presence and not user_verification:
        # Silent authentication.
        try:
            response_data = cbor_get_assertion_sign(
                client_data_hash,
                rp_id_hash,
                cred_list[0],
                hmac_secret,
                resident,
                user_presence,
                user_verification,
            )
            return Cmd(req.cid, _CMD_CBOR, bytes([_ERR_NONE]) + response_data)
        except Exception:
            return cbor_error(req.cid, _ERR_OPERATION_DENIED)
    else:
        # Ask user to confirm one of the credentials.
        state_set = dialog_mgr.set_state(
            Fido2ConfirmGetAssertion(
                req.cid,
                dialog_mgr.iface,
                client_data_hash,
                cred_list,
                hmac_secret,
                resident,
                user_verification,
            )
        )

    if not state_set:
        return cmd_error(req.cid, _ERR_CHANNEL_BUSY)

    return None


def cbor_get_assertion_hmac_secret(
    cred: "Credential", hmac_secret: dict
) -> Optional[bytes]:
    key_agreement = hmac_secret[1]  # The public key of platform key agreement key.
    # NOTE: We should check the key_agreement[_COSE_ALG_KEY] here, but to avoid compatibility issues we don't,
    # because there is currently no valid value which describes the actual key agreement algorithm.
    if (
        key_agreement[_COSE_KEY_TYPE_KEY] != _COSE_KEY_TYPE_EC2
        or key_agreement[_COSE_CURVE_KEY] != _COSE_CURVE_P256
    ):
        return None

    x = key_agreement[_COSE_X_COORD_KEY]
    y = key_agreement[_COSE_Y_COORD_KEY]
    salt_enc = hmac_secret[2]  # The encrypted salt.
    salt_auth = hmac_secret[3]  # The HMAC of the encrypted salt.
    if (
        len(x) != 32
        or len(y) != 32
        or len(salt_auth) != 16
        or len(salt_enc) not in (32, 64)
    ):
        raise CborError(_ERR_INVALID_LEN)

    # Compute the ECDH shared secret.
    ecdh_result = nist256p1.multiply(_KEY_AGREEMENT_PRIVKEY, b"\04" + x + y)
    shared_secret = hashlib.sha256(ecdh_result[1:33]).digest()

    # Check the authentication tag and decrypt the salt.
    tag = hmac.Hmac(shared_secret, salt_enc, hashlib.sha256).digest()[:16]
    if not utils.consteq(tag, salt_auth):
        raise CborError(_ERR_EXTENSION_FIRST)
    salt = aes(aes.CBC, shared_secret).decrypt(salt_enc)

    # Get cred_random - a constant symmetric key associated with the credential.
    cred_random = cred.hmac_secret_key()
    if cred_random is None:
        # The credential does not have the hmac-secret extension enabled.
        return None

    # Compute the hmac-secret output.
    output = hmac.Hmac(cred_random, salt[:32], hashlib.sha256).digest()
    if len(salt) == 64:
        output += hmac.Hmac(cred_random, salt[32:], hashlib.sha256).digest()

    # Encrypt the hmac-secret output.
    return aes(aes.CBC, shared_secret).encrypt(output)


def cbor_get_assertion_sign(
    client_data_hash: bytes,
    rp_id_hash: bytes,
    cred: "Credential",
    hmac_secret: Optional[dict],
    resident: bool,
    user_presence: bool,
    user_verification: bool,
) -> bytes:
    # Process extensions
    extensions = {}

    # Spec deviation: Do not reveal hmac-secret during silent authentication.
    if hmac_secret and user_presence:
        encrypted_output = cbor_get_assertion_hmac_secret(cred, hmac_secret)
        if encrypted_output is not None:
            extensions["hmac-secret"] = encrypted_output

    # Encode the authenticator data.
    flags = 0
    if user_presence:
        flags |= _AUTH_FLAG_UP
    if user_verification:
        flags |= _AUTH_FLAG_UV

    encoded_extensions = b""
    if extensions:
        flags |= _AUTH_FLAG_ED
        encoded_extensions = cbor.encode(extensions)

    ctr = cred.next_signature_counter()

    authenticator_data = (
        rp_id_hash + bytes([flags]) + ctr.to_bytes(4, "big") + encoded_extensions
    )

    # Sign the authenticator data and the client data hash.
    dig = hashlib.sha256()
    dig.update(authenticator_data)
    dig.update(client_data_hash)
    if user_presence:
        privkey = cred.private_key()
    else:
        # Spec deviation: Use a bogus key during silent authentication.
        privkey = _BOGUS_PRIV_KEY
    sig = nist256p1.sign(privkey, dig.digest(), False)
    sig = der.encode_seq((sig[1:33], sig[33:]))

    # Encode the authenticatorGetAssertion response data.
    response = {
        _GETASSERT_RESP_CREDENTIAL: {"type": "public-key", "id": cred.id},
        _GETASSERT_RESP_AUTH_DATA: authenticator_data,
        _GETASSERT_RESP_SIGNATURE: sig,
    }

    if resident and user_presence and cred.user_id is not None:
        response[_GETASSERT_RESP_USER] = {"id": cred.user_id}

    return cbor.encode(response)


def cbor_get_info(req: Cmd) -> Cmd:
    response_data = {
        _GETINFO_RESP_VERSIONS: ["U2F_V2", "FIDO_2_0"],
        _GETINFO_RESP_EXTENSIONS: ["hmac-secret"],
        _GETINFO_RESP_AAGUID: _AAGUID,
        _GETINFO_RESP_OPTIONS: {
            "rk": _ALLOW_RESIDENT_CREDENTIALS,
            "up": True,
            "uv": True,
        },
        _GETINFO_RESP_PIN_PROTOCOLS: [1],
    }
    return Cmd(req.cid, _CMD_CBOR, bytes([_ERR_NONE]) + cbor.encode(response_data))


def cbor_client_pin(req: Cmd) -> Cmd:
    try:
        param = cbor.decode(req.data[1:])
        pin_protocol = param[_CLIENTPIN_CMD_PIN_PROTOCOL]
        subcommand = param[_CLIENTPIN_CMD_SUBCOMMAND]
    except Exception:
        return cbor_error(req.cid, _ERR_INVALID_CBOR)

    if pin_protocol != 1:
        return cbor_error(req.cid, _ERR_PIN_AUTH_INVALID)

    # We only support the get key agreement command which is required for the hmac-secret extension.
    if subcommand != _CLIENTPIN_SUBCMD_GET_KEY_AGREEMENT:
        return cbor_error(req.cid, _ERR_UNSUPPORTED_OPTION)

    # Encode the public key of the authenticator key agreement key.
    # NOTE: There is currently no valid value for _COSE_ALG_KEY which describes the actual
    # key agreement algorithm as specified, but _COSE_ALG_ECDH_ES_HKDF_256 is allegedly
    # recommended by the latest draft of the CTAP2 spec.
    response_data = {
        _CLIENTPIN_RESP_KEY_AGREEMENT: {
            _COSE_ALG_KEY: _COSE_ALG_ECDH_ES_HKDF_256,
            _COSE_KEY_TYPE_KEY: _COSE_KEY_TYPE_EC2,
            _COSE_CURVE_KEY: _COSE_CURVE_P256,
            _COSE_X_COORD_KEY: _KEY_AGREEMENT_PUBKEY[1:33],
            _COSE_Y_COORD_KEY: _KEY_AGREEMENT_PUBKEY[33:],
        }
    }

    return Cmd(req.cid, _CMD_CBOR, bytes([_ERR_NONE]) + cbor.encode(response_data))


def cbor_reset(req: Cmd, dialog_mgr: DialogManager) -> Optional[Cmd]:
    if not storage.is_initialized():
        if __debug__:
            log.warning(__name__, "not initialized")
        return cbor_error(req.cid, _ERR_OPERATION_DENIED)

    if not dialog_mgr.set_state(Fido2ConfirmReset(req.cid, dialog_mgr.iface)):
        return cmd_error(req.cid, _ERR_CHANNEL_BUSY)
    return None


def cmd_keepalive(cid: int, status: int) -> Cmd:
    return Cmd(cid, _CMD_KEEPALIVE, bytes([status]))
