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.
This commit is contained in:
2026-07-08 11:25:34 -07:00
commit f8273fc904
5 changed files with 1343 additions and 0 deletions

24
.gitignore vendored Normal file
View File

@@ -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

61
README.md Normal file
View File

@@ -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
```

709
apex_otp_import.py Normal file
View File

@@ -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())

231
docs/design.md Normal file
View File

@@ -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`.

318
test_oath.py Normal file
View File

@@ -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 <len>
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()