From f8273fc90439522b25e348c3f9ea27112adeb3d1 Mon Sep 17 00:00:00 2001 From: Amal Graafstra Date: Wed, 8 Jul 2026 11:25:34 -0700 Subject: [PATCH] Add Apex OTP importer Import TOTP credentials from the export spreadsheet into a VivoKey Apex OTP (ykneo-oath) applet over PC/SC. Includes the OATH protocol layer (SELECT/VALIDATE/LIST/PUT with 61xx reassembly), PBKDF2 password authentication, pre-flight input validation, ask-per-conflict handling, optional javacard-memory profiling, and hardware-free unit tests. --- .gitignore | 24 ++ README.md | 61 ++++ apex_otp_import.py | 709 +++++++++++++++++++++++++++++++++++++++++++++ docs/design.md | 231 +++++++++++++++ test_oath.py | 318 ++++++++++++++++++++ 5 files changed, 1343 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 apex_otp_import.py create mode 100644 docs/design.md create mode 100644 test_oath.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..05037e0 --- /dev/null +++ b/.gitignore @@ -0,0 +1,24 @@ +# Never commit OTP secrets, exports, or profiling logs. +*.xlsx +*.xls +*.csv +otp-keys-export.* +apex-otp-memlog.csv + +# QR screenshots / backup codes, in case any data is placed in the repo tree. +*.png +*.jpg +*_backup_codes.txt +*-backup-codes.txt +*recovery*.txt + +# Python +__pycache__/ +*.pyc +*.pyo +.venv/ +venv/ +.pytest_cache/ + +# CAP artifacts +*.cap diff --git a/README.md b/README.md new file mode 100644 index 0000000..5cd4aae --- /dev/null +++ b/README.md @@ -0,0 +1,61 @@ +# otp-import + +Import TOTP credentials from an exported spreadsheet into a VivoKey Apex OTP +(ykneo-oath) applet over PC/SC. + +The importer reads the curated columns of an export spreadsheet +(`otp-keys-export.xlsx`: Issuer, Account, Secret, Type, Algorithm, Digits, +Period) and writes each credential to the applet using the standard YKOATH +`PUT` command. It authenticates first if the applet is password-protected, asks +before overwriting credentials that already exist, and can optionally profile +per-credential on-chip memory consumption via the `javacard-memory` applet. + +The protocol layer is verified against the applet sources +(`com.vivokey.otp.YkneoOath` / `OathObj`) and `doc/Protocol.txt`. See +[docs/design.md](docs/design.md) for the full design and protocol notes. + +## Security + +This tool handles live TOTP secrets. **Keep your secrets out of this +repository.** The spreadsheet, any CSV, and QR screenshots are excluded by +`.gitignore`; keep the actual data in a private location (e.g. a synced +personal folder) and point the importer at it with `--input`. + +## Requirements + +- Python 3.9+ +- `pyscard`, `cryptography`, `openpyxl` +- A PC/SC reader with the Apex (or a JavaCard carrying the applet) presented + +## Usage + +``` +# 1. Validate the data and preview the APDUs (no card needed) +python apex_otp_import.py --input /path/to/otp-keys-export.xlsx --dry-run + +# 2. Read-only: list what is already on the applet +python apex_otp_import.py --input /path/to/otp-keys-export.xlsx --list + +# 3. Import (asks before overwriting existing credentials) +python apex_otp_import.py --input /path/to/otp-keys-export.xlsx + +# Non-interactive conflict handling +python apex_otp_import.py --input ... --overwrite-all +python apex_otp_import.py --input ... --skip-existing + +# Target a specific reader, supply a password, and profile memory +python apex_otp_import.py --input ... --reader VivoKey --password '...' --profile-memory +``` + +The input is validated before any reader is touched. Duplicate credential names, +invalid Base32, over-long names/keys, and unsupported algorithms halt the run +with a report so they can be fixed in the source data. + +## Tests + +Hardware-free unit tests cover the protocol encoding, password derivation, +input validation, and response reassembly: + +``` +python test_oath.py +``` diff --git a/apex_otp_import.py b/apex_otp_import.py new file mode 100644 index 0000000..f1c6d0e --- /dev/null +++ b/apex_otp_import.py @@ -0,0 +1,709 @@ +#!/usr/bin/env python3 +"""Import TOTP credentials from otp-keys-export.xlsx into a VivoKey Apex OTP +(ykneo-oath) applet over PC/SC. + +Protocol verified against the apex-totp (com.vivokey.otp.YkneoOath) and +javacard-memory applet sources. See +docs/superpowers/specs/2026-07-08-apex-otp-import-design.md. +""" + +import argparse +import base64 +import csv +import datetime +import getpass +import hashlib +import hmac +import os +import sys +from dataclasses import dataclass, field + +# ykneo-oath enforces a 14-byte minimum HMAC key; the Yubico host right-zero-pads +# short secrets to this length. Padding is HMAC-identical (HMAC zero-pads the key +# to the block size regardless), so TOTP output is unchanged. +HMAC_MIN_KEY = 14 +# The applet stores the HMAC key in a 64-byte ipad/opad buffer and rejects longer. +MAX_KEY = 64 +MAX_NAME = 64 +DEFAULT_PERIOD = 30 + +# ykneo-oath tags / type|algorithm nibbles (verified against YkneoOath/OathObj). +TAG_NAME = 0x71 +TAG_NAME_LIST = 0x72 +TAG_KEY = 0x73 +TAG_CHALLENGE = 0x74 +TAG_RESPONSE = 0x75 +TAG_VERSION = 0x79 +TAG_ALGORITHM = 0x7B + +HOTP_TYPE = 0x10 +TOTP_TYPE = 0x20 +HMAC_SHA1 = 0x01 +HMAC_SHA256 = 0x02 +HMAC_SHA512 = 0x03 # rejected by the applet; used only to flag bad input + + +# --------------------------------------------------------------------------- +# Credential field helpers +# --------------------------------------------------------------------------- + +def decode_secret(secret: str) -> bytes: + """Base32-decode an OATH secret, tolerating lowercase, spaces and missing + padding.""" + s = secret.strip().replace(" ", "").upper() + s += "=" * ((8 - len(s) % 8) % 8) + return base64.b32decode(s) + + +def pad_key(raw: bytes) -> bytes: + """Right-zero-pad a raw HMAC key to the applet's 14-byte minimum.""" + if len(raw) >= HMAC_MIN_KEY: + return raw + return raw + b"\x00" * (HMAC_MIN_KEY - len(raw)) + + +def format_cred_id(issuer: str, account: str, period: int = DEFAULT_PERIOD) -> str: + """Build the on-applet credential name: '[period/]issuer:account'. + The period prefix is only added when it differs from the 30s default.""" + label = f"{issuer}:{account}" if issuer else account + if period and period != DEFAULT_PERIOD: + return f"{period}/{label}" + return label + + +@dataclass +class Credential: + issuer: str + account: str + secret: str # Base32 text + oath_type: int = TOTP_TYPE + algorithm: int = HMAC_SHA1 + digits: int = 6 + period: int = DEFAULT_PERIOD + source: str = "" + + @property + def cred_id(self) -> str: + return format_cred_id(self.issuer, self.account, self.period) + + @property + def type_algo(self) -> int: + return self.oath_type | self.algorithm + + def raw_key(self) -> bytes: + """Base32-decode the secret (raises on invalid Base32).""" + return decode_secret(self.secret) + + def put_data(self) -> bytes: + return build_put_data(self.cred_id, self.type_algo, self.digits, + pad_key(self.raw_key())) + + +_TYPE_MAP = {"TOTP": TOTP_TYPE, "HOTP": HOTP_TYPE} +_ALGO_MAP = {"SHA1": HMAC_SHA1, "SHA256": HMAC_SHA256, "SHA512": HMAC_SHA512} + + +def _match_columns(header): + """Map the OTP Keys header row to column indices by keyword (order-agnostic).""" + col = {} + for i, h in enumerate(header): + if h is None: + continue + k = str(h).strip().lower() + if "issuer" in k: col.setdefault("issuer", i) + elif "account" in k: col.setdefault("account", i) + elif "secret" in k: col.setdefault("secret", i) + elif k.startswith("type"): col.setdefault("type", i) + elif "algorithm" in k: col.setdefault("algorithm", i) + elif "digit" in k: col.setdefault("digits", i) + elif "period" in k: col.setdefault("period", i) + elif "source" in k: col.setdefault("source", i) + return col + + +def load_credentials(xlsx_path, sheet="OTP Keys"): + """Load credentials from the curated columns of the export spreadsheet.""" + from openpyxl import load_workbook + wb = load_workbook(xlsx_path, read_only=True, data_only=True) + ws = wb[sheet] + rows = ws.iter_rows(values_only=True) + header = next(rows) + col = _match_columns(header) + + def cell(row, key, default=None): + i = col.get(key) + return row[i] if i is not None and i < len(row) else default + + def as_int(v, default): + try: + return int(v) + except (TypeError, ValueError): + return default + + creds = [] + for row in rows: + if row is None: + continue + secret = cell(row, "secret") + issuer = cell(row, "issuer") or "" + account = cell(row, "account") or "" + if not secret and not issuer and not account: + continue # skip blank rows + creds.append(Credential( + issuer=str(issuer).strip(), + account=str(account).strip(), + secret=str(secret or "").strip(), + oath_type=_TYPE_MAP.get(str(cell(row, "type", "TOTP")).strip().upper(), TOTP_TYPE), + algorithm=_ALGO_MAP.get(str(cell(row, "algorithm", "SHA1")).strip().upper(), HMAC_SHA1), + digits=as_int(cell(row, "digits"), 6), + period=as_int(cell(row, "period"), DEFAULT_PERIOD), + source=str(cell(row, "source", "") or "").strip(), + )) + wb.close() + return creds + + +def validate_input(creds): + """Return a list of human-readable problems that must be fixed in the source + data before importing. An empty list means the input is clean.""" + problems = [] + seen = {} + for c in creds: + cid = c.cred_id + # duplicate credential ids (OATH keys by name, so a dup silently clobbers) + if cid in seen: + problems.append( + f"Duplicate credential name '{cid}': {seen[cid]} and {c.source} " + f"map to the same applet entry (different secrets would collide)." + ) + else: + seen[cid] = c.source + + # name length + if len(cid.encode("utf-8")) > MAX_NAME: + problems.append(f"Name too long (>{MAX_NAME} bytes): '{cid}' ({c.source}).") + + # algorithm supported by the applet + if c.algorithm not in (HMAC_SHA1, HMAC_SHA256): + problems.append( + f"Unsupported algorithm 0x{c.algorithm:02X} (SHA512?) for '{cid}' " + f"({c.source}); the applet accepts only SHA1/SHA256." + ) + + # secret decodes and fits + try: + raw = c.raw_key() + except Exception as e: + problems.append(f"Invalid Base32 secret for '{cid}' ({c.source}): {e}.") + continue + if len(raw) == 0: + problems.append(f"Empty secret for '{cid}' ({c.source}).") + elif len(raw) > MAX_KEY: + problems.append( + f"Key too long ({len(raw)}>{MAX_KEY} bytes) for '{cid}' ({c.source})." + ) + return problems + + +# --------------------------------------------------------------------------- +# BER-TLV helpers +# --------------------------------------------------------------------------- + +def tlv(tag: int, value: bytes) -> bytes: + """Encode a single-byte-tag BER-TLV with 1/2/3-byte length forms.""" + n = len(value) + if n < 0x80: + length = bytes([n]) + elif n <= 0xFF: + length = bytes([0x81, n]) + else: + length = bytes([0x82, (n >> 8) & 0xFF, n & 0xFF]) + return bytes([tag]) + length + value + + +_HMAC_HASH = {HMAC_SHA1: "sha1", HMAC_SHA256: "sha256"} + + +def derive_key(password: str, salt: bytes) -> bytes: + """Derive the 16-byte OATH access key from a password, per the ykneo-oath + convention: PBKDF2-HMAC-SHA1(password, salt = device id from SELECT, 1000 + rounds). PBKDF2 always uses SHA1 here, regardless of the challenge algo.""" + return hashlib.pbkdf2_hmac("sha1", password.encode("utf-8"), salt, 1000, 16) + + +def compute_response(key: bytes, challenge: bytes, algo: int = HMAC_SHA1) -> bytes: + """HMAC(key, challenge) using the algorithm named by the SELECT ALGORITHM + tag (SHA1 in practice).""" + return hmac.new(key, challenge, _HMAC_HASH[algo]).digest() + + +def build_put_data(name: str, type_algo: int, digits: int, key: bytes) -> bytes: + """PUT payload: NAME(0x71) + KEY(0x73, [type|algo, digits, key...]). + `key` must already be padded (see pad_key). PROPERTY/IMF are omitted + (TOTP, no touch).""" + return ( + tlv(TAG_NAME, name.encode("utf-8")) + + tlv(TAG_KEY, bytes([type_algo, digits]) + key) + ) + + +def build_validate_data(response: bytes, client_challenge: bytes) -> bytes: + """VALIDATE payload: RESPONSE(0x75, our HMAC of the device challenge) + + CHALLENGE(0x74, our fresh challenge for mutual auth).""" + return tlv(TAG_RESPONSE, response) + tlv(TAG_CHALLENGE, client_challenge) + + +INS_SEND_REMAINING = 0xA5 + + +class CardChannel: + """Thin wrapper over a pyscard connection that transceives APDUs and + transparently follows 61xx status words with SEND REMAINING.""" + + def __init__(self, connection): + self.conn = connection + + def send(self, cla, ins, p1, p2, data: bytes = b"", le=None): + """Transmit one APDU; returns (response_bytes, sw) with sw = (sw1<<8)|sw2. + Reassembles chained 61xx responses.""" + apdu = [cla, ins, p1, p2] + if data: + apdu += [len(data)] + list(data) + if le is not None: + apdu += [le] + resp, sw1, sw2 = self.conn.transmit(apdu) + out = bytes(resp) + while sw1 == 0x61: + resp, sw1, sw2 = self.conn.transmit( + [0x00, INS_SEND_REMAINING, 0x00, 0x00, sw2]) + out += bytes(resp) + return out, (sw1 << 8) | sw2 + + +def parse_tlvs(data: bytes): + """Parse a concatenation of single-byte-tag BER-TLVs into an ordered list + of (tag, value) tuples. Preserves repeated tags (e.g. LIST 0x72 entries).""" + out = [] + i = 0 + n = len(data) + while i < n: + tag = data[i] + i += 1 + first = data[i] + if first < 0x80: + length = first + i += 1 + elif first == 0x81: + length = data[i + 1] + i += 2 + elif first == 0x82: + length = (data[i + 1] << 8) | data[i + 2] + i += 3 + else: + raise ValueError(f"unsupported TLV length byte 0x{first:02X}") + out.append((tag, data[i:i + length])) + i += length + return out + + +# --------------------------------------------------------------------------- +# OATH session (ykneo-oath) over a CardChannel +# --------------------------------------------------------------------------- + +# Applet-instance AID first, then the shorter Yubico prefixes (the applet +# prefix-selects, so any of these reaches it). +OATH_AIDS = [ + bytes.fromhex("A0000005272101014150455801"), + bytes.fromhex("A000000527210101"), + bytes.fromhex("A0000005272101"), +] + +INS_SELECT = 0xA4 +INS_PUT = 0x01 +INS_LIST = 0xA1 +INS_VALIDATE = 0xA3 + +_SW_MESSAGES = { + 0x9000: "OK", + 0x6A84: "applet full (no space)", + 0x6982: "authentication required", + 0x6A80: "wrong data / incorrect password", + 0x6984: "no such object", + 0x6D00: "instruction not supported (wrong applet?)", + 0x6E00: "class not supported", +} + + +def sw_message(sw: int) -> str: + return _SW_MESSAGES.get(sw, f"error SW={sw:04X}") + + +class OathError(Exception): + def __init__(self, sw: int, message: str = ""): + self.sw = sw + super().__init__(message or sw_message(sw)) + + +class OathSession: + """ykneo-oath session: SELECT (with AID fallback), password VALIDATE, LIST, + PUT. All wire encoding verified against the applet source.""" + + def __init__(self, channel: "CardChannel", rng=os.urandom, aids=OATH_AIDS): + self.ch = channel + self.rng = rng + self.aids = aids + self.version = None + self.device_id = None + self.challenge = None + self.algorithm = HMAC_SHA1 + self._key = None + + def select(self): + last_sw = None + for aid in self.aids: + data, sw = self.ch.send(0x00, INS_SELECT, 0x04, 0x00, data=aid, le=0x00) + if sw == 0x9000 and data and data[0] == TAG_VERSION: + self._parse_select(data) + return self + last_sw = sw + raise OathError(last_sw or 0, "OATH applet not found (SELECT failed)") + + def _parse_select(self, data): + self.challenge = None + for tag, val in parse_tlvs(data): + if tag == TAG_VERSION: + self.version = tuple(val) + elif tag == TAG_NAME: + self.device_id = val + elif tag == TAG_CHALLENGE: + self.challenge = val + elif tag == TAG_ALGORITHM: + self.algorithm = val[0] if val else HMAC_SHA1 + + @property + def locked(self) -> bool: + return self.challenge is not None + + def validate(self, password: str): + """Mutually authenticate against a password-protected applet.""" + if not self.locked: + return + key = derive_key(password, self.device_id) + our_response = compute_response(key, self.challenge, self.algorithm) + client_challenge = self.rng(8) + data, sw = self.ch.send( + 0x00, INS_VALIDATE, 0x00, 0x00, + data=build_validate_data(our_response, client_challenge), le=0x00) + if sw != 0x9000: + raise OathError(sw, "authentication failed (incorrect password?)") + device_response = dict(parse_tlvs(data)).get(TAG_RESPONSE) + if device_response != compute_response(key, client_challenge, self.algorithm): + raise OathError(sw, "mutual authentication check failed") + self._key = key + return + + def list_names(self): + data, sw = self.ch.send(0x00, INS_LIST, 0x00, 0x00, le=0x00) + if sw != 0x9000: + raise OathError(sw, "LIST failed") + names = [] + for tag, val in parse_tlvs(data): + if tag == TAG_NAME_LIST and len(val) >= 1: + names.append(val[1:].decode("utf-8", "replace")) # drop type byte + return names + + def put(self, cred: "Credential"): + data, sw = self.ch.send(0x00, INS_PUT, 0x00, 0x00, data=cred.put_data()) + if sw != 0x9000: + raise OathError(sw) + return + + +# --------------------------------------------------------------------------- +# Optional javacard-memory applet (for capacity checks / profiling) +# --------------------------------------------------------------------------- + +MEMORY_AID = bytes.fromhex("A0000008466D656D6F727901") +INS_GET_BATCH = 0x01 + + +def parse_memory(data: bytes) -> dict: + """Parse the 12-byte SELECT response of the javacard-memory applet.""" + return { + "available_persistent": int.from_bytes(data[0:4], "big"), + "total_persistent": int.from_bytes(data[4:8], "big"), + "transient_reset": int.from_bytes(data[8:10], "big"), + "transient_deselect": int.from_bytes(data[10:12], "big"), + } + + +class MemorySession: + """Optional javacard-memory applet. select() returns None if absent.""" + + def __init__(self, channel: "CardChannel"): + self.ch = channel + + def select(self): + data, sw = self.ch.send(0x00, INS_SELECT, 0x04, 0x00, data=MEMORY_AID, le=0x00) + if sw != 0x9000 or len(data) < 12: + return None + return parse_memory(data) + + def available(self): + """SELECT the applet and return available persistent bytes (or None).""" + m = self.select() + return m["available_persistent"] if m else None + + def batch(self): + data, sw = self.ch.send(0x00, INS_GET_BATCH, 0x00, 0x00, le=0x00) + return data.hex() if sw == 0x9000 else "" + + +# --------------------------------------------------------------------------- +# CLI / orchestration (hardware I/O) +# --------------------------------------------------------------------------- + +DEFAULT_INPUT = "otp-keys-export.xlsx" +DEFAULT_MEMLOG = "apex-otp-memlog.csv" +MEM_PER_CRED_ESTIMATE = 250 # rough EEPROM per credential (3x64B arrays + overhead) + + +def get_uid(connection) -> str: + """Contactless/contact UID via the PC/SC GET DATA pseudo-APDU.""" + try: + resp, sw1, sw2 = connection.transmit([0xFF, 0xCA, 0x00, 0x00, 0x00]) + if sw1 == 0x90: + return bytes(resp).hex().upper() + except Exception: + pass + return "" + + +def _readers_with_cards(): + from smartcard.System import readers + out = [] + for r in readers(): + try: + conn = r.createConnection() + conn.connect() + out.append((str(r), conn)) + except Exception: + pass + return out + + +def pick_reader(reader_filter=None): + """Return (reader_name, connection) for the single Apex holding an OATH + applet, prompting if several are present. (None, None) if none found.""" + candidates = _readers_with_cards() + if reader_filter: + candidates = [(n, c) for (n, c) in candidates + if reader_filter.lower() in n.lower()] + oath = [] + for name, conn in candidates: + try: + OathSession(CardChannel(conn)).select() + oath.append((name, conn)) + except Exception: + try: + conn.disconnect() + except Exception: + pass + if not oath: + print("No Apex OTP applet found on any reader with a card.") + print("Place your Apex on a reader (or insert the test card) and re-run.") + return None, None + if len(oath) == 1: + return oath[0] + print("Multiple readers present an OATH applet:") + for i, (name, _) in enumerate(oath): + print(f" [{i}] {name}") + idx = int(input("Choose target reader index: ").strip()) + chosen = oath[idx] + for j, (_, conn) in enumerate(oath): + if j != idx: + try: + conn.disconnect() + except Exception: + pass + return chosen + + +class ConflictPolicy: + """Decide overwrite/skip for a credential already on the applet.""" + + def __init__(self, args): + self.always = ("overwrite" if args.overwrite_all + else "skip" if args.skip_existing else None) + + def decide(self, name): + if self.always: + return self.always + if not sys.stdin.isatty(): + raise SystemExit( + f"'{name}' already exists on the applet and no " + f"--overwrite-all/--skip-existing was given (non-interactive).") + while True: + ans = input(f" '{name}' exists. [o]verwrite / [s]kip / " + f"[a]ll overwrite / [x] skip all: ").strip().lower() + if ans == "o": + return "overwrite" + if ans == "s": + return "skip" + if ans == "a": + self.always = "overwrite" + return "overwrite" + if ans == "x": + self.always = "skip" + return "skip" + + +def _put_apdu_hex(cred): + data = cred.put_data() + return (bytes([0x00, INS_PUT, 0x00, 0x00, len(data)]) + data).hex().upper() + + +def parse_args(argv=None): + p = argparse.ArgumentParser( + description="Import TOTP credentials from the export spreadsheet into a " + "VivoKey Apex OTP (ykneo-oath) applet over PC/SC.") + p.add_argument("--input", default=DEFAULT_INPUT, + help=f"export .xlsx (default: {DEFAULT_INPUT})") + p.add_argument("--reader", help="substring to select a specific PC/SC reader") + p.add_argument("--dry-run", action="store_true", + help="validate and print the APDUs; no card I/O") + p.add_argument("--list", action="store_true", + help="read-only: list credentials already on the applet") + g = p.add_mutually_exclusive_group() + g.add_argument("--overwrite-all", action="store_true", + help="overwrite any credential that already exists") + g.add_argument("--skip-existing", action="store_true", + help="skip any credential that already exists") + p.add_argument("--profile-memory", action="store_true", + help="record per-credential EEPROM consumption to the mem-log") + p.add_argument("--memlog", default=DEFAULT_MEMLOG, + help=f"profiling CSV path (default: {DEFAULT_MEMLOG})") + p.add_argument("--password", help="applet password (else prompted if locked)") + return p.parse_args(argv) + + +def _reauth(oath, pw): + oath.select() + if oath.locked: + oath.validate(pw) + + +def main(argv=None): + args = parse_args(argv) + + # 1. Load + validate BEFORE touching any reader. + creds = load_credentials(args.input) + problems = validate_input(creds) + if problems: + print(f"Input validation FAILED ({len(problems)} problem(s)) - " + f"fix the spreadsheet and re-run:") + for p in problems: + print(" -", p) + return 2 + print(f"Loaded and validated {len(creds)} credentials from {args.input}.") + + # 2. Dry-run: no card. + if args.dry_run: + print("\n--dry-run: PUT APDUs that WOULD be sent (no card I/O):") + for c in creds: + print(f" {c.cred_id}") + print(f" {_put_apdu_hex(c)}") + return 0 + + # 3. Reader/card detection. + reader, connection = pick_reader(args.reader) + if connection is None: + return 1 + channel = CardChannel(connection) + + # 4. Capacity pre-check (selects the memory applet, before the OATH session). + if not args.list: + avail = MemorySession(channel).available() + if avail is not None: + est = MEM_PER_CRED_ESTIMATE * len(creds) + msg = (f"Memory applet: {avail} bytes free " + f"(rough estimate ~{est} bytes for {len(creds)} credentials).") + print("WARNING: " + msg + " They may not all fit." if avail < est else msg) + + # 5. OATH session. + oath = OathSession(channel) + oath.select() + print(f"OATH applet on '{reader}': v{'.'.join(map(str, oath.version))}, " + f"{'LOCKED' if oath.locked else 'unlocked'}.") + pw = None + if oath.locked: + pw = args.password or getpass.getpass("Applet password: ") + oath.validate(pw) + print("Authenticated OK.") + existing = set(oath.list_names()) + + if args.list: + print(f"\n{len(existing)} credential(s) currently on the applet:") + for n in sorted(existing): + print(" ", n) + return 0 + + # 6. Import. + policy = ConflictPolicy(args) + writer = uid = batch = None + memlog_fh = None + if args.profile_memory: + uid = get_uid(connection) + ms = MemorySession(channel) + ms.select() + batch = ms.batch() + _reauth(oath, pw) + new_file = not os.path.exists(args.memlog) + memlog_fh = open(args.memlog, "a", newline="", encoding="utf-8") + writer = csv.writer(memlog_fh) + if new_file: + writer.writerow(["timestamp_iso", "card_uid", "mem_batch", "key_name", + "avail_before", "avail_after", "consumed_bytes"]) + + ok = skipped = failed = 0 + try: + for c in creds: + if c.cred_id in existing and policy.decide(c.cred_id) == "skip": + skipped += 1 + print(f" skip {c.cred_id} (already present)") + continue + try: + if args.profile_memory: + before = MemorySession(channel).available() + _reauth(oath, pw) + oath.put(c) + after = MemorySession(channel).available() + _reauth(oath, pw) + consumed = (before - after) if (before is not None + and after is not None) else "" + writer.writerow([datetime.datetime.now().isoformat(timespec="seconds"), + uid, batch, c.cred_id, before, after, consumed]) + print(f" ok {c.cred_id} (consumed ~{consumed} B)") + else: + oath.put(c) + print(f" ok {c.cred_id}") + existing.add(c.cred_id) + ok += 1 + except OathError as e: + failed += 1 + print(f" FAIL {c.cred_id}: {e}") + if e.sw == 0x6A84: + print(" Applet reports FULL (0x6A84) - stopping.") + break + finally: + if memlog_fh: + memlog_fh.close() + + print(f"\nDone: {ok} written, {skipped} skipped, {failed} failed.") + if args.profile_memory: + print(f"Memory profile appended to {args.memlog}.") + return 0 if failed == 0 else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/docs/design.md b/docs/design.md new file mode 100644 index 0000000..177d5dd --- /dev/null +++ b/docs/design.md @@ -0,0 +1,231 @@ +# Apex OTP Importer — Design + +Date: 2026-07-08 +Status: Draft for review + +## Goal + +A single runnable Python script that reads the curated OTP export +(`otp-keys-export.xlsx`, sheet `OTP Keys`) and writes every credential to a +VivoKey Apex's OTP (OATH) applet over PC/SC, quickly and repeatably. It must +support password-protected OATH applets (authenticate before writing) and, +optionally, profile per-credential on-chip memory consumption. + +## Context and constraints + +- Host: Windows 11, Python 3.11. Installed and used: `pyscard`, `cryptography`, + `openpyxl`. No Yubico libraries (`ykman`/`yubikit`) — the OATH protocol is + implemented directly, so there is no added dependency and no reliance on + Yubico device enumeration (Apex is not a Yubico device). +- Transport: PC/SC. Multiple contactless readers are typically present + (ACR1252 CL, OMNIKEY 5022, Identiv uTrust 3700 F CL, VivoKey Smart Reader). +- Input source of truth: the **curated columns** of the spreadsheet + (Issuer, Account, Secret, Type, Algorithm, Digits, Period) — NOT the + `otpauth://` URI column. The curated Account values were cleaned during export + (e.g. Heroku → `amal@vivokey.com`, not the QR label "Amal Graafstra"). + +## Verified protocol facts + +Verified directly against the applet sources in the local repos +`apex-totp` (ykneo-oath, `com.vivokey.otp.YkneoOath` / `OathObj`) and +`javacard-memory` (`de.chrz.jcmemory.JCMemoryApplet`), plus +`apex-totp/doc/Protocol.txt`. + +### OATH applet + +- Applet AID (from `apex-totp/build.xml`): + `A0 00 00 05 27 21 01 01 41 50 45 58 01` (`41 50 45 58` = "APEX"). + SELECT strategy: try this full AID first, then fall back to + `A0 00 00 05 27 21 01 01` (8 B, per repo docs/tests) and + `A0 00 00 05 27 21 01` (7 B, canonical Yubico). Use the first that returns a + valid SELECT response (version tag `0x79`). Prefix selection works because + Yubico Authenticator already talks to Apex with the short AID. +- INS: PUT `0x01`, DELETE `0x02`, SET CODE `0x03`, RESET `0x04` (P1P2 must be + `0xDEAD`), RENAME `0x05`, LIST `0xA1`, CALCULATE `0xA2`, VALIDATE `0xA3`, + CALCULATE ALL `0xA4`, SEND REMAINING `0xA5`. +- Tags: NAME `0x71`, NAME_LIST `0x72`, KEY `0x73`, CHALLENGE `0x74`, + RESPONSE `0x75`, T_RESPONSE `0x76`, NO_RESPONSE `0x77`, PROPERTY `0x78`, + VERSION `0x79`, IMF `0x7A`, ALGORITHM `0x7B`. +- Type|algo byte: type in high nibble (HOTP `0x10`, TOTP `0x20`), algo in low + nibble (SHA1 `0x01`, SHA256 `0x02`). **SHA512 is rejected by this applet.** + All our credentials are TOTP/SHA1 → `0x21`. +- SELECT response TLV: VERSION(`0x79`) + NAME(`0x71`, 8-byte device id/salt) and, + if a password is set, CHALLENGE(`0x74`, 8 bytes) + ALGORITHM(`0x7B`, 1 byte). + Presence of the CHALLENGE tag ⇒ applet is locked. +- PUT data: `NAME(0x71,len,name)` + `KEY(0x73, len=2+keylen, [typeAlgo, digits, key…])`. + PROPERTY/IMF omitted (no touch, TOTP only). Name ≤ 64 B, key ≤ 64 B. + The applet XORs exactly `keylen` bytes into the 64-byte HMAC ipad/opad, so a + short secret yields the same HMAC as a zero-padded one; we still right-pad + short secrets to 14 B to match the Yubico host (HMAC-identical). +- LIST response (repeats): `NAME_LIST(0x72, len=namelen+1, [type, name…])`. +- Long responses (LIST/CALCULATE ALL) chunk with SW `61xx`; fetch the rest with + SEND REMAINING (`0xA5`) until `9000`. +- Response codes: success `9000`; applet full `6A84`; auth required `6982`; + wrong syntax / wrong data (incl. wrong password on VALIDATE) `6A80`; + no such object `6984`. +- Auth state is stored in a CLEAR_ON_DESELECT buffer: any deselect (e.g. a + memory-applet SELECT) drops authentication and requires a fresh VALIDATE. + +### Password authentication (VALIDATE) + +Per `doc/Protocol.txt` and `OathObj`: + +1. `key = PBKDF2-HMAC-SHA1(password_utf8, salt = 8-byte device id from SELECT, + iterations = 1000, dklen = 16)`. PBKDF2 always uses HMAC-SHA1 here, + independent of the credential/challenge algorithm. +2. The challenge-response HMAC uses the algorithm named by the SELECT + `ALGORITHM` (`0x7B`) tag (low nibble: SHA1 `0x01` / SHA256 `0x02`), defaulting + to SHA1 — which is what the Yubico host sets in practice (20-byte response). + `deviceResponse = HMAC(key, deviceChallenge)` where `deviceChallenge` is the + SELECT `0x74` value. +3. `clientChallenge = os.urandom(8)`. +4. Send VALIDATE (`0xA3`): `RESPONSE(0x75, deviceResponse)` + + `CHALLENGE(0x74, clientChallenge[8])`. The response length must equal the + applet's digest length or it is rejected — hence honoring the ALGORITHM tag. +5. Applet returns `RESPONSE(0x75, HMAC(key, clientChallenge))`; the host verifies + it to confirm mutual auth. Wrong password → `6A80`. + +The derived key is stable across re-SELECTs (device id only changes on RESET), +so it is derived once and reused when re-validation is needed during profiling. + +### Memory applet (optional) + +- Applet AID: `A0 00 00 08 46 6D 65 6D 6F 72 79 01` (`46 6D…79` = "Fmemory"). +- SELECT returns 12 bytes: availPersistent(4 B BE), totalBaseline(4 B BE), + transientReset(2 B), transientDeselect(2 B). Apex Flex baseline = 84336 B. +- `GET_BATCH` = `00 01 00 00 00` → 4-byte batch id. +- Each SELECT runs `requestObjectDeletion()` (GC) then reports free memory. + Because selecting it deselects OATH, profiling must re-VALIDATE afterward. + +## Architecture + +One runnable file `apex_otp_import.py`, organized into focused units: + +- `Tlv` helpers — encode/parse BER-TLV with 1/2/3-byte lengths (`0x81`/`0x82`). +- `CardChannel` — thin wrapper over a pyscard connection: `transmit(apdu)` plus + `send(cla, ins, p1, p2, data, le)` that transparently follows `61xx` with + SEND REMAINING and returns `(data, sw)`. +- `OathSession` — SELECT (with AID fallback), parse device id/challenge/locked, + `derive_key(password)`, `validate(key)`, `list()`, `put(cred)`, + and status-word → exception mapping. Pure protocol; unit-testable with a fake + channel. +- `MemorySession` — SELECT the memory applet, parse availPersistent + baseline, + `get_batch()`. Absent applet handled gracefully. +- `Credential` — normalized record: issuer, account, secret(bytes), oath_type, + algorithm, digits, period, and `cred_id` (`issuer:account`, with `period/` + prefix only if period ≠ 30). +- `load_credentials(xlsx_path)` — read the `OTP Keys` sheet from curated columns. +- `validate_input(creds)` — pre-flight checks; returns a list of problems. +- `pick_reader(...)` — enumerate readers, detect cards, SELECT OATH, choose. +- `main()` — CLI + orchestration in the order below. + +Companion `test_oath.py` — hardware-free unit tests. + +## Execution flow + +1. **Load + validate input first (before any reader access).** + `load_credentials` → `validate_input`. If any problems, print them all and + exit non-zero. No interactive fixing — the user edits the spreadsheet. +2. **Reader/card detection.** For each reader with a card inserted, try the OATH + SELECT. 0 responders → error "Place your Apex on a reader and re-run", exit. + 1 → use it. >1 → list reader names and prompt for the target. +3. **Session.** SELECT OATH. If locked, `getpass` the password and VALIDATE + (map `6A80` → "incorrect password", exit). LIST existing credential names. +4. **Optional capacity pre-check.** If the memory applet is present, read + availPersistent and warn (no halt) if `availPersistent < ~250 B × N`. +5. **Import.** For each credential: if its `cred_id` is already on the applet, + ask per conflict (overwrite / skip; with all/none shortcuts). Otherwise PUT. + Map `6A84` → "applet full" and stop further PUTs (report remainder as + skipped). `--profile-memory` wraps each PUT in before/after memory reads + (re-VALIDATE between, since the memory SELECT deselects OATH) and appends a + log row. +6. **Report.** Per-credential OK / skipped / failed summary, plus mem-log path + when profiling. + +## Pre-flight validation rules (halt on any) + +- Duplicate `cred_id` within the input (e.g. the two + `SAW:michelle@dangerousthings.com` entries with different secrets). +- Secret missing or not valid Base32 after uppercasing and removing spaces. +- Decoded key length > 64 bytes. +- `cred_id` (UTF-8) length > 64 bytes. +- Algorithm SHA512 (unsupported by the applet), or a non-TOTP/HOTP type. + +Each problem is reported with the offending issuer/account and source row. + +## Conflict handling (on-device) + +Interactive per name already present on the applet: `[o]verwrite` / +`[s]kip` / `[a]` overwrite-all / `[x]` skip-all. Non-interactive overrides: +`--overwrite-all`, `--skip-existing`. + +## Memory profiling (`--profile-memory`, opt-in) + +Per credential: memory SELECT → availBefore; OATH SELECT + re-VALIDATE (if +locked) → PUT; memory SELECT → availAfter; `consumed = availBefore - availAfter`. +Append to `apex-otp-memlog.csv` with columns: +`timestamp_iso, card_uid, mem_batch, key_name, avail_before, avail_after, consumed_bytes`. +`timestamp_iso` is the host clock; `card_uid` from `FF CA 00 00 00`; the key +secret is never logged. Block-granular allocation means `consumed_bytes` is a +close estimate, not byte-exact. + +## CLI + +``` +apex_otp_import.py [--input PATH] [--reader SUBSTR] [--dry-run] [--list] + [--overwrite-all | --skip-existing] [--profile-memory] + [--memlog PATH] [--password PASS] +``` + +- `--dry-run` — load, validate, build every APDU, hex-print them; no card I/O. +- `--list` — read-only: connect, authenticate if needed, print on-applet creds. +- `--password` — supply the applet password non-interactively (else `getpass`). + +## Error handling + +Central status-word → message map: `6A84` applet full, `6982` auth required, +`6A80` wrong data / incorrect password, `6984` no such object, `61xx` handled by +SEND REMAINING, `6D00`/`6E00` wrong INS/CLA (wrong applet or AID). Card +exceptions (no card, reader gone) surface as clear, actionable messages. + +## Testing (`test_oath.py`, no hardware) + +- TLV encode/parse round-trips incl. multi-byte lengths. +- `cred_id` formatting (period prefix rule). +- Byte-exact PUT payload for a known credential (type/algo, digits, padded key). +- Byte-exact VALIDATE payload and mutual-response verification against a known + key/challenge. +- Base32 decode + 14-byte right-pad; HMAC-equivalence of padded vs unpadded. +- PBKDF2/HMAC vectors; RFC 6238 TOTP HMAC sanity check. +- `validate_input` catches each halt condition (dup, bad Base32, oversize name, + oversize key, SHA512). +- `61xx` reassembly via a fake channel that chunks a long LIST response. + +## Out of scope (YAGNI) + +Setting/changing/removing the applet password (authenticate-only); HOTP import; +touch-required; RENAME/DELETE; CSV/text input; on-device code calculation +(beyond optional post-import verification, not included). + +## Files + +Code and docs live in the `otp-import` repo +(`C:\Users\amal\Documents\Repos\otp-import`, remote +`git.dngr.us/VivoKey/otp-import`): `apex_otp_import.py`, `test_oath.py`, +`docs/design.md`, `README.md`, `.gitignore`. + +The secret data stays **out of the repo**: the export spreadsheet +(`otp-keys-export.xlsx`), the profiling CSV (`apex-otp-memlog.csv`, created only +when profiling), and QR screenshots remain in the private Nextcloud folder and +are excluded by `.gitignore`. The importer is pointed at the data with +`--input`. + +## Risks / notes + +- The instance AID on a Fidesmo-provisioned Apex may differ from `build.xml`; + mitigated by the AID fallback list and validating the SELECT response shape. +- Contactless UID from `FF CA` may be random per tap on some configurations; it + is logged as-is for correlation, not treated as a stable serial. +- Keep the card still: the whole import runs in one OATH session; a mid-import + deselect drops auth and (without profiling's re-VALIDATE) would fail PUTs with + `6982`. diff --git a/test_oath.py b/test_oath.py new file mode 100644 index 0000000..5192446 --- /dev/null +++ b/test_oath.py @@ -0,0 +1,318 @@ +"""Hardware-free unit tests for apex_otp_import (OATH protocol layer + input).""" +import os +import tempfile +import unittest + +from openpyxl import Workbook + +from apex_otp_import import ( + tlv, parse_tlvs, decode_secret, pad_key, format_cred_id, + build_put_data, build_validate_data, + derive_key, compute_response, + Credential, validate_input, load_credentials, + CardChannel, OathSession, OathError, parse_memory, + TOTP_TYPE, HOTP_TYPE, HMAC_SHA1, HMAC_SHA256, HMAC_SHA512, +) + + +class FakeConn: + """Fake pyscard connection: replays a scripted list of (data, sw1, sw2).""" + def __init__(self, script): + self.script = list(script) + self.sent = [] + + def transmit(self, apdu): + self.sent.append(apdu) + data, sw1, sw2 = self.script.pop(0) + return list(data), sw1, sw2 + + +def cred(issuer="Iss", account="acct", secret="GEZDGNBVGY3TQOJQ", **kw): + return Credential(issuer=issuer, account=account, secret=secret, **kw) + + +class TestTlv(unittest.TestCase): + def test_short_form_length(self): + self.assertEqual(tlv(0x71, b"AB"), bytes([0x71, 0x02, 0x41, 0x42])) + + def test_boundary_127_is_short_form(self): + self.assertEqual(tlv(0x73, b"\x00" * 127)[:2], bytes([0x73, 0x7F])) + + def test_extended_form_128(self): + # length 128 uses 0x81 + self.assertEqual(tlv(0x73, b"\x00" * 128)[:3], bytes([0x73, 0x81, 0x80])) + + +class TestParseTlvs(unittest.TestCase): + def test_parses_ordered_pairs(self): + data = bytes([0x79, 0x03, 0x00, 0x01, 0x02, 0x71, 0x02, 0xAA, 0xBB]) + self.assertEqual( + parse_tlvs(data), + [(0x79, b"\x00\x01\x02"), (0x71, b"\xAA\xBB")], + ) + + def test_preserves_repeated_tags(self): + # LIST returns repeated 0x72 entries + data = tlv(0x72, b"\x21A") + tlv(0x72, b"\x21B") + self.assertEqual(parse_tlvs(data), [(0x72, b"\x21A"), (0x72, b"\x21B")]) + + def test_round_trips_extended_length(self): + val = bytes(range(256)) * 2 # 512 bytes -> 0x82 form + self.assertEqual(parse_tlvs(tlv(0x73, val)), [(0x73, val)]) + + +class TestSecret(unittest.TestCase): + def test_rfc6238_vector_decodes_to_known_bytes(self): + # RFC 6238 seed "12345678901234567890" == this Base32 + self.assertEqual( + decode_secret("GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ"), + b"12345678901234567890", + ) + + def test_lowercase_and_spaces_and_missing_padding(self): + self.assertEqual(decode_secret("ge zd gn bv"), decode_secret("GEZDGNBV")) + + def test_pad_short_key_to_14_zero_filled(self): + raw = decode_secret("JBSWY3DPEHPK3PXP") # 16 chars -> 10 bytes + self.assertEqual(len(raw), 10) + padded = pad_key(raw) + self.assertEqual(len(padded), 14) + self.assertEqual(padded[:10], raw) + self.assertEqual(padded[10:], b"\x00\x00\x00\x00") + + def test_pad_leaves_long_enough_key_unchanged(self): + raw = b"\x01" * 20 + self.assertEqual(pad_key(raw), raw) + + +class TestCredId(unittest.TestCase): + def test_default_period_is_issuer_colon_account(self): + self.assertEqual(format_cred_id("Google", "amal@amal.net", 30), + "Google:amal@amal.net") + + def test_non_default_period_prefixed(self): + self.assertEqual(format_cred_id("Steam", "user", 15), "15/Steam:user") + + def test_empty_issuer_is_just_account(self): + self.assertEqual(format_cred_id("", "amal", 30), "amal") + + +class TestPutData(unittest.TestCase): + def test_byte_exact_put_payload(self): + key = b"\x01" * 14 + data = build_put_data("AB", TOTP_TYPE | HMAC_SHA1, 6, key) + expected = ( + bytes([0x71, 0x02, 0x41, 0x42]) # NAME "AB" + + bytes([0x73, 0x10, 0x21, 0x06]) + key # KEY: len=2+14, typealgo, digits + ) + self.assertEqual(data, expected) + + def test_totp_sha1_type_byte_is_0x21(self): + self.assertEqual(TOTP_TYPE | HMAC_SHA1, 0x21) + + +class TestValidateData(unittest.TestCase): + def test_byte_exact_validate_payload(self): + resp = b"\xAA" * 20 + chal = b"\xBB" * 8 + data = build_validate_data(resp, chal) + self.assertEqual(data, bytes([0x75, 0x14]) + resp + bytes([0x74, 0x08]) + chal) + + +class TestPasswordAuth(unittest.TestCase): + SALT = bytes([0x19, 0x83, 0x57, 0x1E, 0xC3, 0xE5, 0xAB, 0xD7]) + + def test_derive_key_pbkdf2_1000_rounds_16_bytes(self): + key = derive_key("test1234", self.SALT) + self.assertEqual(key.hex(), "aa69a3f71eec1660849d7bb16c313e32") + self.assertEqual(len(key), 16) + + def test_compute_response_matches_rfc2202_hmac_sha1(self): + # RFC 2202 HMAC-SHA1 test case 1 + resp = compute_response(b"\x0b" * 20, b"Hi There", HMAC_SHA1) + self.assertEqual(resp.hex(), "b617318655057264e28bc0b6fb378c8ef146be00") + + +class TestValidateInput(unittest.TestCase): + def test_clean_input_has_no_problems(self): + creds = [cred(account="a@x.com"), cred(account="b@x.com")] + self.assertEqual(validate_input(creds), []) + + def test_duplicate_cred_id_is_reported(self): + # same issuer+account, different secret -> the SAW/michelle case + creds = [ + cred(issuer="SAW", account="m@d.com", secret="GEZDGNBVGY3TQOJQ", source="a.png"), + cred(issuer="SAW", account="m@d.com", secret="MZXW6YTBOI======", source="b.png"), + ] + problems = validate_input(creds) + self.assertEqual(len(problems), 1) + self.assertIn("SAW:m@d.com", problems[0]) + + def test_invalid_base32_is_reported(self): + problems = validate_input([cred(secret="not-base32!")]) + self.assertTrue(any("base32" in p.lower() for p in problems)) + + def test_oversize_name_is_reported(self): + problems = validate_input([cred(issuer="I" * 40, account="a" * 40)]) + self.assertTrue(any("64" in p for p in problems)) + + def test_sha512_algorithm_is_reported(self): + problems = validate_input([cred(algorithm=HMAC_SHA512)]) + self.assertTrue(any("sha512" in p.lower() or "algorithm" in p.lower() + for p in problems)) + + +class TestLoadCredentials(unittest.TestCase): + def _make_xlsx(self, rows): + wb = Workbook() + ws = wb.active + ws.title = "OTP Keys" + ws.append(["Issuer", "Account", "Secret (Base32)", "Type", "Algorithm", + "Digits", "Period (s)", "QR Label", "otpauth:// URI", "Source File"]) + for r in rows: + ws.append(r) + fd, path = tempfile.mkstemp(suffix=".xlsx") + os.close(fd) + wb.save(path) + return path + + def test_maps_columns_and_string_enums(self): + path = self._make_xlsx([ + ["Google", "a@x.com", "GEZDGNBVGY3TQOJQ", "TOTP", "SHA1", 6, 30, "lbl", "uri", "g.png"], + ["Steam", "user", "MZXW6YTBOI======", "TOTP", "SHA256", 8, 15, "lbl", "uri", "s.png"], + ]) + try: + creds = load_credentials(path) + finally: + os.remove(path) + self.assertEqual(len(creds), 2) + g = creds[0] + self.assertEqual((g.issuer, g.account, g.secret), ("Google", "a@x.com", "GEZDGNBVGY3TQOJQ")) + self.assertEqual((g.oath_type, g.algorithm, g.digits, g.period), (TOTP_TYPE, HMAC_SHA1, 6, 30)) + self.assertEqual(g.source, "g.png") + self.assertEqual((creds[1].algorithm, creds[1].digits, creds[1].period), (HMAC_SHA256, 8, 15)) + + +class TestCardChannel(unittest.TestCase): + def test_reassembles_61xx_via_send_remaining(self): + conn = FakeConn([ + (b"\x72\x02\x21A", 0x61, 0x04), # first chunk + "4 more" + (b"\x72\x02\x21B", 0x90, 0x00), # remainder + OK + ]) + ch = CardChannel(conn) + data, sw = ch.send(0x00, 0xA1, 0x00, 0x00, le=0x00) # LIST + self.assertEqual(sw, 0x9000) + self.assertEqual(data, b"\x72\x02\x21A\x72\x02\x21B") + # second APDU must be SEND REMAINING (INS 0xA5) + self.assertEqual(conn.sent[1][:2], [0x00, 0xA5]) + + def test_single_response_returns_sw(self): + conn = FakeConn([(b"", 0x90, 0x00)]) + data, sw = CardChannel(conn).send(0x00, 0x01, 0x00, 0x00, data=b"\x71\x01A") + self.assertEqual(sw, 0x9000) + self.assertEqual(data, b"") + + +DEV_ID = bytes(range(8)) # 00..07 +VERSION_TLV = tlv(0x79, b"\x00\x01\x02") + + +def select_unlocked(): + return VERSION_TLV + tlv(0x71, DEV_ID) + + +def select_locked(challenge): + return (VERSION_TLV + tlv(0x71, DEV_ID) + + tlv(0x74, challenge) + tlv(0x7B, bytes([HMAC_SHA1]))) + + +class TestOathSessionSelect(unittest.TestCase): + def test_parses_unlocked_select(self): + s = OathSession(CardChannel(FakeConn([(select_unlocked(), 0x90, 0x00)]))) + s.select() + self.assertFalse(s.locked) + self.assertEqual(s.device_id, DEV_ID) + self.assertEqual(s.version, (0, 1, 2)) + + def test_parses_locked_select(self): + chal = b"\xC1" * 8 + s = OathSession(CardChannel(FakeConn([(select_locked(chal), 0x90, 0x00)]))) + s.select() + self.assertTrue(s.locked) + self.assertEqual(s.challenge, chal) + self.assertEqual(s.algorithm, HMAC_SHA1) + + def test_falls_back_to_next_aid_on_failure(self): + conn = FakeConn([(b"", 0x6A, 0x82), # first AID: not found + (select_unlocked(), 0x90, 0x00)]) # second AID: ok + s = OathSession(CardChannel(conn)) + s.select() + self.assertFalse(s.locked) + self.assertEqual(len(conn.sent), 2) # tried two AIDs + + +class TestOathSessionList(unittest.TestCase): + def test_list_names_parses_entries(self): + listing = (tlv(0x72, bytes([0x21]) + b"Google:a@x.com") + + tlv(0x72, bytes([0x21]) + b"AWS:b")) + conn = FakeConn([(select_unlocked(), 0x90, 0x00), (listing, 0x90, 0x00)]) + s = OathSession(CardChannel(conn)); s.select() + self.assertEqual(s.list_names(), ["Google:a@x.com", "AWS:b"]) + + +class TestOathSessionPut(unittest.TestCase): + def _cred(self): + return Credential("Iss", "acct", "GEZDGNBVGY3TQOJQ") + + def test_put_success_sends_put_apdu(self): + conn = FakeConn([(select_unlocked(), 0x90, 0x00), (b"", 0x90, 0x00)]) + s = OathSession(CardChannel(conn)); s.select() + s.put(self._cred()) + self.assertEqual(conn.sent[1][1], 0x01) # INS PUT + self.assertEqual(bytes(conn.sent[1][5:]), self._cred().put_data()) + + def test_put_full_raises_with_sw(self): + conn = FakeConn([(select_unlocked(), 0x90, 0x00), (b"", 0x6A, 0x84)]) + s = OathSession(CardChannel(conn)); s.select() + with self.assertRaises(OathError) as ctx: + s.put(self._cred()) + self.assertEqual(ctx.exception.sw, 0x6A84) + + +class TestParseMemory(unittest.TestCase): + def test_parses_available_and_total(self): + # avail=84000, total=84336 (0x00014970), tRes=0x0100, tDes=0x0080 + data = (84000).to_bytes(4, "big") + bytes.fromhex("00014970") \ + + bytes.fromhex("0100") + bytes.fromhex("0080") + m = parse_memory(data) + self.assertEqual(m["available_persistent"], 84000) + self.assertEqual(m["total_persistent"], 84336) + + +class TestOathSessionValidate(unittest.TestCase): + def test_validate_success_sends_correct_response(self): + from apex_otp_import import derive_key, compute_response + chal = b"\xC1" * 8 + client = b"\xD2" * 8 + key = derive_key("hunter2", DEV_ID) + device_reply = tlv(0x75, compute_response(key, client)) + conn = FakeConn([(select_locked(chal), 0x90, 0x00), + (device_reply, 0x90, 0x00)]) + s = OathSession(CardChannel(conn), rng=lambda n: client) + s.select() + s.validate("hunter2") # must not raise + # our response to the device must be HMAC(key, device challenge) + sent = bytes(conn.sent[1]) + self.assertIn(compute_response(key, chal), sent) + + def test_validate_wrong_password_raises(self): + conn = FakeConn([(select_locked(b"\xC1" * 8), 0x90, 0x00), + (b"", 0x6A, 0x80)]) + s = OathSession(CardChannel(conn), rng=lambda n: b"\xD2" * 8) + s.select() + with self.assertRaises(OathError): + s.validate("wrong") + + +if __name__ == "__main__": + unittest.main()