# Copyright (c) 2020 Yubico AB
# All rights reserved.
#
#   Redistribution and use in source and binary forms, with or
#   without modification, are permitted provided that the following
#   conditions are met:
#
#    1. Redistributions of source code must retain the above copyright
#       notice, this list of conditions and the following disclaimer.
#    2. Redistributions in binary form must reproduce the above
#       copyright notice, this list of conditions and the following
#       disclaimer in the documentation and/or other materials provided
#       with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
# FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
# COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
# BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.

from __future__ import annotations

import logging
import os
import struct
import sys
from enum import IntEnum, IntFlag, unique
from threading import Event
from typing import Callable, Iterator

from ..ctap import STATUS, CtapDevice, CtapError
from ..utils import LOG_LEVEL_TRAFFIC
from .base import HidDescriptor

logger = logging.getLogger(__name__)


if sys.platform.startswith("linux"):
    from . import linux as backend
elif sys.platform.startswith("win32"):
    from . import windows as backend
elif sys.platform.startswith("darwin"):
    from . import macos as backend
elif sys.platform.startswith("freebsd"):
    from . import freebsd as backend
elif sys.platform.startswith("netbsd"):
    from . import netbsd as backend
elif sys.platform.startswith("openbsd"):
    from . import openbsd as backend
else:
    raise Exception("Unsupported platform")


list_descriptors = backend.list_descriptors
get_descriptor = backend.get_descriptor
open_connection = backend.open_connection


class ConnectionFailure(Exception):
    """The CTAP connection failed or returned an invalid response."""


@unique
class CTAPHID(IntEnum):
    PING = 0x01
    MSG = 0x03
    LOCK = 0x04
    INIT = 0x06
    WINK = 0x08
    CBOR = 0x10
    CANCEL = 0x11

    ERROR = 0x3F
    KEEPALIVE = 0x3B

    VENDOR_FIRST = 0x40


@unique
class CAPABILITY(IntFlag):
    WINK = 0x01
    LOCK = 0x02  # Not used
    CBOR = 0x04
    NMSG = 0x08

    def supported(self, flags: CAPABILITY) -> bool:
        return bool(flags & self)


TYPE_INIT = 0x80


class CtapHidDevice(CtapDevice):
    """
    CtapDevice implementation using the HID transport.

    :cvar descriptor: Device descriptor.
    """

    def __init__(self, descriptor: HidDescriptor, connection):
        self.descriptor = descriptor
        self._packet_size = descriptor.report_size_out
        self._connection = connection

        nonce = os.urandom(8)
        self._channel_id = 0xFFFFFFFF
        response = self.call(CTAPHID.INIT, nonce)
        r_nonce, response = response[:8], response[8:]
        if r_nonce != nonce:
            raise ConnectionFailure("Wrong nonce")
        (
            self._channel_id,
            self._u2fhid_version,
            v1,
            v2,
            v3,
            self._capabilities,
        ) = struct.unpack_from(">IBBBBB", response)
        self._device_version = (v1, v2, v3)

    def __repr__(self):
        return f"CtapHidDevice({self.descriptor.path!r})"

    @property
    def version(self) -> int:
        """CTAP HID protocol version."""
        return self._u2fhid_version

    @property
    def device_version(self) -> tuple[int, int, int]:
        """Device version number."""
        return self._device_version

    @property
    def capabilities(self) -> int:
        """Capabilities supported by the device."""
        return self._capabilities

    @property
    def product_name(self) -> str | None:
        """Product name of device."""
        return self.descriptor.product_name

    @property
    def serial_number(self) -> str | None:
        """Serial number of device."""
        return self.descriptor.serial_number

    def _send_cancel(self):
        packet = struct.pack(">IB", self._channel_id, TYPE_INIT | CTAPHID.CANCEL).ljust(
            self._packet_size, b"\0"
        )
        logger.log(LOG_LEVEL_TRAFFIC, "SEND: %s", packet.hex())
        self._connection.write_packet(packet)

    def call(
        self,
        cmd: int,
        data: bytes = b"",
        event: Event | None = None,
        on_keepalive: Callable[[STATUS], None] | None = None,
    ) -> bytes:
        event = event or Event()

        while True:
            try:
                return self._do_call(cmd, data, event, on_keepalive)
            except CtapError as e:
                if e.code == CtapError.ERR.CHANNEL_BUSY:
                    if not event.wait(0.1):
                        logger.warning("CTAP channel busy, trying again...")
                        continue  # Keep retrying on BUSY while not cancelled
                raise

    def _do_call(self, cmd, data, event, on_keepalive):
        remaining = data
        seq = 0

        # Send request
        header = struct.pack(">IBH", self._channel_id, TYPE_INIT | cmd, len(remaining))
        while remaining or seq == 0:
            size = min(len(remaining), self._packet_size - len(header))
            body, remaining = remaining[:size], remaining[size:]
            packet = header + body
            logger.log(LOG_LEVEL_TRAFFIC, "SEND: %s", packet.hex())
            self._connection.write_packet(packet.ljust(self._packet_size, b"\0"))
            header = struct.pack(">IB", self._channel_id, 0x7F & seq)
            seq += 1

        try:
            # Read response
            seq = 0
            response = b""
            last_ka = None
            while True:
                if event.is_set():
                    # Cancel
                    logger.debug("Sending cancel...")
                    self._send_cancel()

                recv = self._connection.read_packet()
                logger.log(LOG_LEVEL_TRAFFIC, "RECV: %s", recv.hex())

                r_channel = struct.unpack_from(">I", recv)[0]
                recv = recv[4:]
                if r_channel != self._channel_id:
                    raise ConnectionFailure("Wrong channel")

                if not response:  # Initialization packet
                    r_cmd, r_len = struct.unpack_from(">BH", recv)
                    recv = recv[3:]
                    if r_cmd == TYPE_INIT | cmd:
                        pass  # first data packet
                    elif r_cmd == TYPE_INIT | CTAPHID.KEEPALIVE:
                        try:
                            ka_status = STATUS(struct.unpack_from(">B", recv)[0])
                            logger.debug(f"Got keepalive status: {ka_status:02x}")
                        except ValueError:
                            raise ConnectionFailure("Invalid keepalive status")
                        if on_keepalive and ka_status != last_ka:
                            last_ka = ka_status
                            on_keepalive(ka_status)
                        continue
                    elif r_cmd == TYPE_INIT | CTAPHID.ERROR:
                        raise CtapError(struct.unpack_from(">B", recv)[0])
                    else:
                        raise CtapError(CtapError.ERR.INVALID_COMMAND)
                else:  # Continuation packet
                    r_seq = struct.unpack_from(">B", recv)[0]
                    recv = recv[1:]
                    if r_seq != seq:
                        raise ConnectionFailure("Wrong sequence number")
                    seq += 1

                response += recv
                if len(response) >= r_len:
                    break

            return response[:r_len]
        except KeyboardInterrupt:
            logger.debug("Keyboard interrupt, cancelling...")
            self._send_cancel()

            raise

    def wink(self) -> None:
        """Causes the authenticator to blink."""
        self.call(CTAPHID.WINK)

    def ping(self, msg: bytes = b"Hello FIDO") -> bytes:
        """Sends data to the authenticator, which echoes it back.

        :param msg: The data to send.
        :return: The response from the authenticator.
        """
        return self.call(CTAPHID.PING, msg)

    def lock(self, lock_time: int = 10) -> None:
        """Locks the channel."""
        self.call(CTAPHID.LOCK, struct.pack(">B", lock_time))

    def close(self) -> None:
        if self._connection:
            self._connection.close()
            self._connection = None

    @classmethod
    def list_devices(cls) -> Iterator[CtapHidDevice]:
        for d in list_descriptors():
            yield cls(d, open_connection(d))


def list_devices() -> Iterator[CtapHidDevice]:
    return CtapHidDevice.list_devices()


def open_device(path) -> CtapHidDevice:
    descriptor = get_descriptor(path)
    return CtapHidDevice(descriptor, open_connection(descriptor))