class EnumChoice(click.Choice):
class _YkmanCommand(click.Command):
class _YkmanGroup(click.Group):
class YkmanContextObject(MutableMapping):
    def __init__(self, choices_enum, hidden=[]):
    def convert(self, value, param, ctx):
    def __init__(self, name=None, **attrs):
    def __init__(self, name=None, commands=None, **attrs):
    def list_commands(self, ctx):
def ykman_group(
    connections=[SmartCardConnection, OtpConnection, FidoConnection], *args, **kwargs
):
def ykman_command(interfaces, *args, **kwargs):
def click_callback(invoke_on_missing=False):
    def wrap(f):
        def inner(ctx, param, val):
def click_parse_format(ctx, param, val):
    def __init__(self):
    def add_resolver(self, key, f):
    def resolve(self):
    def __getitem__(self, key):
    def __setitem__(self, key, value):
    def __delitem__(self, key):
    def __len__(self):
    def __iter__(self):
def click_postpone_execution(f):
    def inner(*args, **kwargs):
def click_parse_b32_key(ctx, param, val):
def click_prompt(prompt, err=True, **kwargs):
def prompt_for_touch():
def prompt_timeout(timeout=0.5):
def cli_fail(message: str, code: int = 1) -> NoReturn: class ApduError(CommandError):
    """Thrown when an APDU response has the wrong SW code"""

    def __init__(self, data: bytes, sw: int):
        self.data = data
        self.sw = sw

    def __str__(self):
        return f"APDU error: SW=0x{self.sw:04x}" class SW(IntEnum):
    NO_INPUT_DATA = 0x6285
    VERIFY_FAIL_NO_RETRY = 0x63C0
    WRONG_LENGTH = 0x6700
    SECURITY_CONDITION_NOT_SATISFIED = 0x6982
    AUTH_METHOD_BLOCKED = 0x6983
    DATA_INVALID = 0x6984
    CONDITIONS_NOT_SATISFIED = 0x6985
    COMMAND_NOT_ALLOWED = 0x6986
    INCORRECT_PARAMETERS = 0x6A80
    FUNCTION_NOT_SUPPORTED = 0x6A81
    FILE_NOT_FOUND = 0x6A82
    NO_SPACE = 0x6A84
    REFERENCE_DATA_NOT_FOUND = 0x6A88
    WRONG_PARAMETERS_P1P2 = 0x6B00
    INVALID_INSTRUCTION = 0x6D00
    COMMAND_ABORTED = 0x6F00
    OK = 0x9000 class SmartCardConnection(Connection, metaclass=abc.ABCMeta):
    @property
    @abc.abstractmethod
    def transport(self) -> TRANSPORT:
        """Get the transport type of the connection (USB or NFC)"""

    @abc.abstractmethod
    def send_and_receive(self, apdu: bytes) -> Tuple[bytes, int]:
        """Sends a command APDU and returns the response""" class OathSession:
    def __init__(self, connection: SmartCardConnection):
        self.protocol = SmartCardProtocol(connection, INS_SEND_REMAINING)
        self._version, self._salt, self._challenge = _parse_select(
            self.protocol.select(AID.OATH)
        )
        self._has_key = self._challenge is not None
        self._device_id = _get_device_id(self._salt)
        self.protocol.enable_touch_workaround(self._version)

    @property
    def version(self) -> Version:
        return self._version

    @property
    def device_id(self) -> str:
        return self._device_id

    @property
    def has_key(self) -> bool:
        return self._has_key

    @property
    def locked(self) -> bool:
        return self._challenge is not None

    def reset(self) -> None:
        self.protocol.send_apdu(0, INS_RESET, 0xDE, 0xAD)
        _, self._salt, self._challenge = _parse_select(self.protocol.select(AID.OATH))
        self._has_key = False
        self._device_id = _get_device_id(self._salt)

    def derive_key(self, password: str) -> bytes:
        return _derive_key(self._salt, password)

    def validate(self, key: bytes) -> None:
        response = _hmac_sha1(key, self._challenge)
        challenge = os.urandom(8)
        data = Tlv(TAG_RESPONSE, response) + Tlv(TAG_CHALLENGE, challenge)
        resp = self.protocol.send_apdu(0, INS_VALIDATE, 0, 0, data)
        verification = _hmac_sha1(key, challenge)
        if not hmac.compare_digest(Tlv.unpack(TAG_RESPONSE, resp), verification):
            raise BadResponseError(
                "Response from validation does not match verification!"
            )
        self._challenge = None

    def set_key(self, key: bytes) -> None:
        challenge = os.urandom(8)
        response = _hmac_sha1(key, challenge)
        self.protocol.send_apdu(
            0,
            INS_SET_CODE,
            0,
            0,
            (
                Tlv(TAG_KEY, int2bytes(OATH_TYPE.TOTP | HASH_ALGORITHM.SHA1) + key)
                + Tlv(TAG_CHALLENGE, challenge)
                + Tlv(TAG_RESPONSE, response)
            ),
        )
        self._has_key = True

    def unset_key(self) -> None:
        self.protocol.send_apdu(0, INS_SET_CODE, 0, 0, Tlv(TAG_KEY))
        self._has_key = False

    def put_credential(
        self, credential_data: CredentialData, touch_required: bool = False
    ) -> Credential:
        d = credential_data
        cred_id = d.get_id()
        secret = _hmac_shorten_key(d.secret, d.hash_algorithm)
        secret = secret.ljust(HMAC_MINIMUM_KEY_SIZE, b"\0")
        data = Tlv(TAG_NAME, cred_id) + Tlv(
            TAG_KEY,
            struct.pack("<BB", d.oath_type | d.hash_algorithm, d.digits) + secret,
        )

        if touch_required:
            data += struct.pack(b">BB", TAG_PROPERTY, PROP_REQUIRE_TOUCH)

        if d.counter > 0:
            data += Tlv(TAG_IMF, struct.pack(">I", d.counter))

        self.protocol.send_apdu(0, INS_PUT, 0, 0, data)
        return Credential(
            self.device_id,
            cred_id,
            d.issuer,
            d.name,
            d.oath_type,
            d.period,
            touch_required,
        )

    def rename_credential(
        self, credential_id: bytes, name: str, issuer: Optional[str] = None
    ) -> bytes:
        require_version(self.version, (5, 3, 1))
        _, _, period = _parse_cred_id(credential_id, OATH_TYPE.TOTP)
        new_id = _format_cred_id(issuer, name, OATH_TYPE.TOTP, period)
        self.protocol.send_apdu(
            0, INS_RENAME, 0, 0, Tlv(TAG_NAME, credential_id) + Tlv(TAG_NAME, new_id)
        )
        return new_id

    def list_credentials(self) -> List[Credential]:
        creds = []
        for tlv in Tlv.parse_list(self.protocol.send_apdu(0, INS_LIST, 0, 0)):
            data = Tlv.unpack(TAG_NAME_LIST, tlv)
            oath_type = OATH_TYPE(MASK_TYPE & data[0])
            cred_id = data[1:]
            issuer, name, period = _parse_cred_id(cred_id, oath_type)
            creds.append(
                Credential(
                    self.device_id, cred_id, issuer, name, oath_type, period, None
                )
            )
        return creds

    def calculate(self, credential_id: bytes, challenge: bytes) -> bytes:
        resp = Tlv.unpack(
            TAG_RESPONSE,
            self.protocol.send_apdu(
                0,
                INS_CALCULATE,
                0,
                0,
                Tlv(TAG_NAME, credential_id) + Tlv(TAG_CHALLENGE, challenge),
            ),
        )
        return resp[1:]

    def delete_credential(self, credential_id: bytes) -> None:
        self.protocol.send_apdu(0, INS_DELETE, 0, 0, Tlv(TAG_NAME, credential_id))

    def calculate_all(
        self, timestamp: Optional[int] = None
    ) -> Mapping[Credential, Optional[Code]]:
        timestamp = int(timestamp or time())
        challenge = _get_challenge(timestamp, DEFAULT_PERIOD)

        entries = {}
        data = Tlv.parse_list(
            self.protocol.send_apdu(
                0, INS_CALCULATE_ALL, 0, 1, Tlv(TAG_CHALLENGE, challenge)
            )
        )
        while data:
            cred_id = Tlv.unpack(TAG_NAME, data.pop(0))
            tlv = data.pop(0)
            resp_tag = tlv.tag
            oath_type = OATH_TYPE.HOTP if resp_tag == TAG_HOTP else OATH_TYPE.TOTP
            touch = resp_tag == TAG_TOUCH
            issuer, name, period = _parse_cred_id(cred_id, oath_type)

            credential = Credential(
                self.device_id, cred_id, issuer, name, oath_type, period, touch
            )

            code = None  # Will be None for HOTP and touch
            if resp_tag == TAG_TRUNCATED:  # Only TOTP, no-touch here
                if period == DEFAULT_PERIOD:
                    code = _format_code(credential, timestamp, tlv.value)
                else:
                    # Non-standard period, recalculate
                    code = self.calculate_code(credential, timestamp)
            entries[credential] = code

        return entries

    def calculate_code(
        self, credential: Credential, timestamp: Optional[int] = None
    ) -> Code:
        if credential.device_id != self.device_id:
            raise ValueError("Credential does not belong to this YubiKey")

        timestamp = int(timestamp or time())
        if credential.oath_type == OATH_TYPE.TOTP:
            challenge = _get_challenge(timestamp, credential.period)
        else:  # HOTP
            challenge = b""

        response = Tlv.unpack(
            TAG_TRUNCATED,
            self.protocol.send_apdu(
                0,
                INS_CALCULATE,
                0,
                0x01,  # Truncate
                Tlv(TAG_NAME, credential.id) + Tlv(TAG_CHALLENGE, challenge),
            ),
        )
        return _format_code(credential, timestamp, response) class CredentialData:
    name: str
    oath_type: OATH_TYPE
    hash_algorithm: HASH_ALGORITHM
    secret: bytes
    digits: int = DEFAULT_DIGITS
    period: int = DEFAULT_PERIOD
    counter: int = DEFAULT_IMF
    issuer: Optional[str] = None

    @classmethod
    def parse_uri(cls, uri: str) -> "CredentialData":
        parsed = urlparse(uri.strip())
        if parsed.scheme != "otpauth":
            raise ValueError("Invalid URI scheme")

        if parsed.hostname is None:
            raise ValueError("Missing OATH type")
        oath_type = OATH_TYPE[parsed.hostname.upper()]

        params = dict((k, v[0]) for k, v in parse_qs(parsed.query).items())
        issuer = None
        name = unquote(parsed.path)[1:]  # Unquote and strip leading /
        if ":" in name:
            issuer, name = name.split(":", 1)

        return cls(
            name=name,
            oath_type=oath_type,
            hash_algorithm=HASH_ALGORITHM[params.get("algorithm", "SHA1").upper()],
            secret=parse_b32_key(params["secret"]),
            digits=int(params.get("digits", DEFAULT_DIGITS)),
            period=int(params.get("period", DEFAULT_PERIOD)),
            counter=int(params.get("counter", DEFAULT_IMF)),
            issuer=params.get("issuer", issuer),
        )

    def get_id(self) -> bytes:
        return _format_cred_id(self.issuer, self.name, self.oath_type, self.period) class OATH_TYPE(IntEnum):
    HOTP = 0x10
    TOTP = 0x20 class HASH_ALGORITHM(IntEnum):
    SHA1 = 0x01
    SHA256 = 0x02
    SHA512 = 0x03 def parse_b32_key(key: str):
    key = key.upper().replace(" ", "")
    key += "=" * (-len(key) % 8)  # Support unpadded
    return b32decode(key) def _format_cred_id(issuer, name, oath_type, period=DEFAULT_PERIOD):
    cred_id = ""
    if oath_type == OATH_TYPE.TOTP and period != DEFAULT_PERIOD:
        cred_id += "%d/" % period
    if issuer:
        cred_id += issuer + ":"
    cred_id += name
    return cred_id.encode() def is_steam(credential):
    return credential.oath_type == OATH_TYPE.TOTP and credential.issuer == "Steam" def calculate_steam(app, credential, timestamp=None):
    timestamp = int(timestamp or time())
    resp = app.calculate(credential.id, struct.pack(">q", timestamp // 30))
    offset = resp[-1] & 0x0F
    code = struct.unpack(">I", resp[offset : offset + 4])[0] & 0x7FFFFFFF
    chars = []
    for i in range(5):
        chars.append(STEAM_CHAR_TABLE[code % len(STEAM_CHAR_TABLE)])
        code //= len(STEAM_CHAR_TABLE)
    return "".join(chars) def is_hidden(credential):
    return credential.issuer == "_hidden" def is_fips_version(version: Version) -> bool:
    """True if a given firmware version indicates a YubiKey (4) FIPS"""
    return (4, 4, 0) <= version < (4, 5, 0) class AppData(Settings):
    _config_dir = XDG_DATA_HOME if USE_XDG else HOME_CONFIG
import click
import logging
from .util import (
    cli_fail,
    click_force_option,
    click_postpone_execution,
    click_callback,
    click_parse_b32_key,
    click_prompt,
    ykman_group,
    prompt_for_touch,
    prompt_timeout,
    EnumChoice,
)
from yubikit.core.smartcard import ApduError, SW, SmartCardConnection
from yubikit.oath import (
    OathSession,
    CredentialData,
    OATH_TYPE,
    HASH_ALGORITHM,
    parse_b32_key,
    _format_cred_id,
)
from ..oath import is_steam, calculate_steam, is_hidden
from ..device import is_fips_version
from ..settings import AppData
            settings.write()
        click.echo("This YubiKey is not password protected.")
    else:
        if not password:
            password = click_prompt("Enter the password", hide_input=True)
        key = session.derive_key(password)
        _validate(ctx, key, True)


def _clear_all_passwords(ctx, param, value):
    if not value or ctx.resilient_parsing:
        return

    settings = AppData("oath")
    if "keys" in settings:
        del settings["keys"]
        settings.write()
    click.echo("All passwords have been forgotten.")
    ctx.exit()


@access.command()
@click.pass_context
@click.option(
    "-a",
    "--all",
    is_flag=True,
    is_eager=True,
    expose_value=False,
    callback=_clear_all_passwords,
    help="Remove all stored passwords.",
)
def forget(ctx):
    """
    Remove a stored password from this computer.
    """
    session = ctx.obj["session"]
    device_id = session.device_id
    settings = ctx.obj["settings"]
    keys = settings.setdefault("keys", {})

    if device_id in keys:
        del keys[session.device_id]
        settings.write()
        click.echo("Password forgotten.")
    else:
        click.echo("No password stored for this YubiKey.")


click_remember_option = click.option(
    "-r",
    "--remember",
    is_flag=True,
    help="Remember the password on this machine.",
)

click_touch_option = click.option(
    "-t", "--touch", is_flag=True, help="Require touch on YubiKey to generate code."
)


click_show_hidden_option = click.option(
    "-H", "--show-hidden", is_flag=True, help="Include hidden accounts."
)


def _string_id(credential):
    return credential.id.decode("utf-8")


def _error_multiple_hits(ctx, hits):
    click.echo(
        "Error: Multiple matches, please make the query more specific.", err=True
    )
    click.echo("", err=True)
    for cred in hits:
        click.echo(_string_id(cred), err=True)
    ctx.exit(1)


def _search(creds, query, show_hidden):
    hits = []
    for c in creds:
        cred_id = _string_id(c)
        if not show_hidden and is_hidden(c):
            continue
        if cred_id == query:
            return [c]
        if query.lower() in cred_id.lower():
            hits.append(c)
    return hits


@oath.group()
def accounts():
    """Manage and use OATH accounts."""


@accounts.command()
@click.argument("name")
@click.argument("secret", callback=click_parse_b32_key, required=False)
@click.option(
    "-o",
    "--oath-type",
    type=EnumChoice(OATH_TYPE),
    default=OATH_TYPE.TOTP.name,
    help="Time-based (TOTP) or counter-based (HOTP) account.",
    show_default=True,
)
@click.option(
    "-d",
    "--digits",
    type=click.Choice(["6", "7", "8"]),
    default="6",
    help="Number of digits in generated code.",
    show_default=True,
)
@click.option(
    "-a",
    "--algorithm",

    type=EnumChoice(HASH_ALGORITHM),