Add QR-to-CSV utility and switch canonical format to CSV
Add qr_to_sheet.py: decode saved OTP QR screenshots and append new credentials to the canonical CSV, deduplicated on the otpauth:// data. Add otpauth.py (parse_otpauth / canonical_key / build_otpauth), shared by both tools. The canonical format becomes a minimal CSV (issuer, account, otpauth_uri, source_file); load_credentials reads CSV or xlsx and derives crypto from the immutable otpauth_uri while taking issuer/account from their editable label columns.
This commit is contained in:
48
README.md
48
README.md
@@ -1,14 +1,40 @@
|
|||||||
# otp-import
|
# otp-import
|
||||||
|
|
||||||
Import TOTP credentials from an exported spreadsheet into a VivoKey Apex OTP
|
Turn saved OTP QR-code screenshots into credentials on a VivoKey Apex OTP
|
||||||
(ykneo-oath) applet over PC/SC.
|
(ykneo-oath) applet, via a canonical CSV.
|
||||||
|
|
||||||
The importer reads the curated columns of an export spreadsheet
|
## Pipeline
|
||||||
(`otp-keys-export.xlsx`: Issuer, Account, Secret, Type, Algorithm, Digits,
|
|
||||||
Period) and writes each credential to the applet using the standard YKOATH
|
1. **QR -> CSV** (`qr_to_sheet.py`): decode QR screenshots and append new
|
||||||
`PUT` command. It authenticates first if the applet is password-protected, asks
|
credentials to `otp-keys.csv`, deduplicated on the `otpauth://` data.
|
||||||
before overwriting credentials that already exist, and can optionally profile
|
2. **Edit labels (optional)**: adjust the `issuer` / `account` columns to taste.
|
||||||
per-credential on-chip memory consumption via the `javacard-memory` applet.
|
Never edit the `otpauth_uri` column - it is the immutable source of truth.
|
||||||
|
3. **CSV -> applet** (`apex_otp_import.py`): write to the Apex, validating for
|
||||||
|
anything you may have broken (duplicate names, invalid characters, overlength).
|
||||||
|
|
||||||
|
The CSV has four columns: `issuer`, `account` (editable labels), `otpauth_uri`
|
||||||
|
(do not edit), and `source_file`. The secret and other crypto parameters live
|
||||||
|
inside `otpauth_uri` and are always taken from there.
|
||||||
|
|
||||||
|
```
|
||||||
|
# Populate/refresh the CSV from a folder of QR screenshots
|
||||||
|
python qr_to_sheet.py --images /path/to/qr-screenshots --csv otp-keys.csv
|
||||||
|
python qr_to_sheet.py --images ... --csv ... --dry-run # preview, no write
|
||||||
|
```
|
||||||
|
|
||||||
|
Undecodable images (stylized/colored QRs, or text-only screenshots) are listed
|
||||||
|
by filename so you can add them by hand.
|
||||||
|
|
||||||
|
## Importing
|
||||||
|
|
||||||
|
Import TOTP credentials from the CSV (or a legacy `.xlsx`) into the applet.
|
||||||
|
|
||||||
|
The importer reads `otp-keys.csv` (or a legacy `.xlsx`), takes the credential
|
||||||
|
name from the `issuer`/`account` columns and the crypto from `otpauth_uri`, and
|
||||||
|
writes each credential 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
|
The protocol layer is verified against the applet sources
|
||||||
(`com.vivokey.otp.YkneoOath` / `OathObj`) and `doc/Protocol.txt`. See
|
(`com.vivokey.otp.YkneoOath` / `OathObj`) and `doc/Protocol.txt`. See
|
||||||
@@ -31,13 +57,13 @@ personal folder) and point the importer at it with `--input`.
|
|||||||
|
|
||||||
```
|
```
|
||||||
# 1. Validate the data and preview the APDUs (no card needed)
|
# 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
|
python apex_otp_import.py --input /path/to/otp-keys.csv --dry-run
|
||||||
|
|
||||||
# 2. Read-only: list what is already on the applet
|
# 2. Read-only: list what is already on the applet
|
||||||
python apex_otp_import.py --input /path/to/otp-keys-export.xlsx --list
|
python apex_otp_import.py --input /path/to/otp-keys.csv --list
|
||||||
|
|
||||||
# 3. Import (asks before overwriting existing credentials)
|
# 3. Import (asks before overwriting existing credentials)
|
||||||
python apex_otp_import.py --input /path/to/otp-keys-export.xlsx
|
python apex_otp_import.py --input /path/to/otp-keys.csv
|
||||||
|
|
||||||
# Non-interactive conflict handling
|
# Non-interactive conflict handling
|
||||||
python apex_otp_import.py --input ... --overwrite-all
|
python apex_otp_import.py --input ... --overwrite-all
|
||||||
|
|||||||
@@ -18,6 +18,8 @@ import os
|
|||||||
import sys
|
import sys
|
||||||
from dataclasses import dataclass, field
|
from dataclasses import dataclass, field
|
||||||
|
|
||||||
|
from otpauth import parse_otpauth
|
||||||
|
|
||||||
# ykneo-oath enforces a 14-byte minimum HMAC key; the Yubico host right-zero-pads
|
# 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
|
# short secrets to this length. Padding is HMAC-identical (HMAC zero-pads the key
|
||||||
# to the block size regardless), so TOTP output is unchanged.
|
# to the block size regardless), so TOTP output is unchanged.
|
||||||
@@ -105,13 +107,14 @@ _ALGO_MAP = {"SHA1": HMAC_SHA1, "SHA256": HMAC_SHA256, "SHA512": HMAC_SHA512}
|
|||||||
|
|
||||||
|
|
||||||
def _match_columns(header):
|
def _match_columns(header):
|
||||||
"""Map the OTP Keys header row to column indices by keyword (order-agnostic)."""
|
"""Map a header row to column indices by keyword (order-agnostic)."""
|
||||||
col = {}
|
col = {}
|
||||||
for i, h in enumerate(header):
|
for i, h in enumerate(header):
|
||||||
if h is None:
|
if h is None:
|
||||||
continue
|
continue
|
||||||
k = str(h).strip().lower()
|
k = str(h).strip().lower()
|
||||||
if "issuer" in k: col.setdefault("issuer", i)
|
if "otpauth" in k: col.setdefault("otpauth", i)
|
||||||
|
elif "issuer" in k: col.setdefault("issuer", i)
|
||||||
elif "account" in k: col.setdefault("account", i)
|
elif "account" in k: col.setdefault("account", i)
|
||||||
elif "secret" in k: col.setdefault("secret", i)
|
elif "secret" in k: col.setdefault("secret", i)
|
||||||
elif k.startswith("type"): col.setdefault("type", i)
|
elif k.startswith("type"): col.setdefault("type", i)
|
||||||
@@ -122,45 +125,68 @@ def _match_columns(header):
|
|||||||
return col
|
return col
|
||||||
|
|
||||||
|
|
||||||
def load_credentials(xlsx_path, sheet="OTP Keys"):
|
def _as_int(v, default):
|
||||||
"""Load credentials from the curated columns of the export spreadsheet."""
|
try:
|
||||||
from openpyxl import load_workbook
|
return int(v)
|
||||||
wb = load_workbook(xlsx_path, read_only=True, data_only=True)
|
except (TypeError, ValueError):
|
||||||
ws = wb[sheet]
|
return default
|
||||||
rows = ws.iter_rows(values_only=True)
|
|
||||||
header = next(rows)
|
|
||||||
|
def _read_field_dicts(path, sheet):
|
||||||
|
"""Read a CSV or XLSX into a list of {field: value} dicts keyed by the
|
||||||
|
matched column names."""
|
||||||
|
ext = os.path.splitext(path)[1].lower()
|
||||||
|
if ext == ".csv":
|
||||||
|
with open(path, newline="", encoding="utf-8") as f:
|
||||||
|
rows = list(csv.reader(f))
|
||||||
|
else:
|
||||||
|
from openpyxl import load_workbook
|
||||||
|
wb = load_workbook(path, read_only=True, data_only=True)
|
||||||
|
rows = list(wb[sheet].iter_rows(values_only=True))
|
||||||
|
wb.close()
|
||||||
|
if not rows:
|
||||||
|
return []
|
||||||
|
header, data = rows[0], rows[1:]
|
||||||
col = _match_columns(header)
|
col = _match_columns(header)
|
||||||
|
out = []
|
||||||
def cell(row, key, default=None):
|
for row in data:
|
||||||
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:
|
if row is None:
|
||||||
continue
|
continue
|
||||||
secret = cell(row, "secret")
|
out.append({k: (row[i] if i is not None and i < len(row) else None)
|
||||||
issuer = cell(row, "issuer") or ""
|
for k, i in col.items()})
|
||||||
account = cell(row, "account") or ""
|
return out
|
||||||
if not secret and not issuer and not account:
|
|
||||||
continue # skip blank rows
|
|
||||||
|
def load_credentials(path, sheet="OTP Keys"):
|
||||||
|
"""Load credentials from a CSV or XLSX. Issuer/account come from their
|
||||||
|
(editable) columns; the crypto (secret/type/algorithm/digits/period) is
|
||||||
|
derived from the immutable otpauth_uri when present, else from the
|
||||||
|
individual crypto columns (legacy)."""
|
||||||
|
creds = []
|
||||||
|
for f in _read_field_dicts(path, sheet):
|
||||||
|
issuer = str(f.get("issuer") or "").strip()
|
||||||
|
account = str(f.get("account") or "").strip()
|
||||||
|
otp = str(f.get("otpauth") or "").strip()
|
||||||
|
secret_col = f.get("secret")
|
||||||
|
if not (otp or secret_col or issuer or account):
|
||||||
|
continue # blank row
|
||||||
|
if otp.startswith("otpauth://"):
|
||||||
|
o = parse_otpauth(otp)
|
||||||
|
secret = o.secret
|
||||||
|
oath_type = _TYPE_MAP.get(o.type.upper(), TOTP_TYPE)
|
||||||
|
algorithm = _ALGO_MAP.get(o.algorithm.upper(), HMAC_SHA1)
|
||||||
|
digits, period = o.digits, o.period
|
||||||
|
else:
|
||||||
|
secret = str(secret_col or "").strip()
|
||||||
|
oath_type = _TYPE_MAP.get(str(f.get("type") or "TOTP").strip().upper(), TOTP_TYPE)
|
||||||
|
algorithm = _ALGO_MAP.get(str(f.get("algorithm") or "SHA1").strip().upper(), HMAC_SHA1)
|
||||||
|
digits = _as_int(f.get("digits"), 6)
|
||||||
|
period = _as_int(f.get("period"), DEFAULT_PERIOD)
|
||||||
creds.append(Credential(
|
creds.append(Credential(
|
||||||
issuer=str(issuer).strip(),
|
issuer=issuer, account=account, secret=secret,
|
||||||
account=str(account).strip(),
|
oath_type=oath_type, algorithm=algorithm, digits=digits, period=period,
|
||||||
secret=str(secret or "").strip(),
|
source=str(f.get("source") 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
|
return creds
|
||||||
|
|
||||||
|
|
||||||
@@ -528,7 +554,7 @@ class MemorySession:
|
|||||||
# CLI / orchestration (hardware I/O)
|
# CLI / orchestration (hardware I/O)
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
DEFAULT_INPUT = "otp-keys-export.xlsx"
|
DEFAULT_INPUT = "otp-keys.csv"
|
||||||
DEFAULT_MEMLOG = "apex-otp-memlog.csv"
|
DEFAULT_MEMLOG = "apex-otp-memlog.csv"
|
||||||
MEM_PER_CRED_ESTIMATE = 250 # rough EEPROM per credential (3x64B arrays + overhead)
|
MEM_PER_CRED_ESTIMATE = 250 # rough EEPROM per credential (3x64B arrays + overhead)
|
||||||
|
|
||||||
|
|||||||
119
docs/qr-to-csv-design.md
Normal file
119
docs/qr-to-csv-design.md
Normal file
@@ -0,0 +1,119 @@
|
|||||||
|
# QR-to-CSV Utility — Design
|
||||||
|
|
||||||
|
Date: 2026-07-08
|
||||||
|
Status: Draft for review
|
||||||
|
|
||||||
|
## Goal and workflow
|
||||||
|
|
||||||
|
A companion utility that turns saved OTP QR-code screenshots into rows of the
|
||||||
|
canonical CSV that the importer consumes. The overall flow is a one-way
|
||||||
|
pipeline, not a hand-maintained file:
|
||||||
|
|
||||||
|
1. **QR -> CSV** (`qr_to_sheet.py`): decode QR images, deduplicate on the
|
||||||
|
`otpauth://` data, and append only genuinely-new entries to the CSV.
|
||||||
|
2. **Optional user edits**: the user may adjust the *label* columns (issuer,
|
||||||
|
account). They are instructed never to touch the `otpauth://` data.
|
||||||
|
3. **CSV -> applet** (`apex_otp_import.py`): push to the Apex, validating for
|
||||||
|
human-introduced problems (duplicate names, invalid characters, overlength).
|
||||||
|
|
||||||
|
The two stages guard different risks: stage 1 prevents re-importing the same QR;
|
||||||
|
stage 3 catches bad edits. The `otpauth://` string is the immutable source of
|
||||||
|
truth throughout.
|
||||||
|
|
||||||
|
## Canonical format: CSV
|
||||||
|
|
||||||
|
Replaces the `.xlsx` as the canonical format (open, plain-text, git-diffable,
|
||||||
|
trivial to append and dedup). Minimal schema:
|
||||||
|
|
||||||
|
| Column | Editable? | Purpose |
|
||||||
|
| ------ | --------- | ------- |
|
||||||
|
| `issuer` | yes (label) | Issuer shown on the applet; seeded from the otpauth. |
|
||||||
|
| `account` | yes (label) | Account shown on the applet; seeded from the otpauth label. |
|
||||||
|
| `otpauth_uri` | **no** | The raw decoded `otpauth://` string. Dedup key (stage 1) and crypto source (stage 3). |
|
||||||
|
| `source_file` | informational | Filename the QR was decoded from. |
|
||||||
|
|
||||||
|
The secret remains visible (inside `otpauth_uri`), so nothing is hidden. The
|
||||||
|
credential name written to the applet is `issuer:account` from the editable
|
||||||
|
columns; the secret/type/algorithm/digits/period come from parsing
|
||||||
|
`otpauth_uri`.
|
||||||
|
|
||||||
|
A one-time conversion turns the existing `otp-keys-export.xlsx` "OTP Keys" sheet
|
||||||
|
into `otp-keys.csv` (issuer, account, otpauth_uri, source_file), so no data is
|
||||||
|
lost and the tools operate on the CSV going forward.
|
||||||
|
|
||||||
|
## Shared module: `otpauth.py`
|
||||||
|
|
||||||
|
Used by both tools:
|
||||||
|
|
||||||
|
- `parse_otpauth(uri) -> OtpAuth`: parse `otpauth://TYPE/LABEL?params` into
|
||||||
|
`type` (totp/hotp), `issuer` (from the `issuer` param, else the label prefix
|
||||||
|
before `:`), `account` (label after the `issuer:` prefix, else the whole
|
||||||
|
label), `secret` (uppercased, spaces removed), `algorithm` (default SHA1),
|
||||||
|
`digits` (default 6), `period` (default 30). Percent-decodes label/issuer.
|
||||||
|
- `canonical_key(otp) -> tuple`: a normalized dedup key
|
||||||
|
`(type, issuer, account, secret, algorithm, digits, period)` with defaults
|
||||||
|
filled and the secret uppercased. Two scans of the same credential — even with
|
||||||
|
the issuer given via the label vs. the `issuer` param — produce the same key;
|
||||||
|
a different secret produces a different key.
|
||||||
|
- `build_otpauth(...) -> str`: reconstruct a URI (used by the xlsx conversion
|
||||||
|
for any legacy row lacking a stored URI).
|
||||||
|
|
||||||
|
## `qr_to_sheet.py`
|
||||||
|
|
||||||
|
CLI: `qr_to_sheet.py [--images DIR] [--csv PATH] [--dry-run]`
|
||||||
|
|
||||||
|
1. Enumerate QR images in `--images` (default: the data folder) — `*.png`,
|
||||||
|
`*.jpg`.
|
||||||
|
2. Decode each with OpenCV using the robust multi-pass strategy (direct ->
|
||||||
|
grayscale -> upscale 2x/3x/4x/0.5x -> Otsu/adaptive threshold).
|
||||||
|
3. `parse_otpauth` each decoded `otpauth://totp/...`. An `otpauth-migration://`
|
||||||
|
payload is reported as unsupported (out of scope), not parsed.
|
||||||
|
4. Load the existing CSV (if any); build the set of `canonical_key` values from
|
||||||
|
every row's `otpauth_uri`.
|
||||||
|
5. For each decoded QR: if its `canonical_key` is already present (in the CSV or
|
||||||
|
already added this run), skip it as a duplicate; otherwise append a row
|
||||||
|
(`issuer`, `account` from the parse; `otpauth_uri` = the raw decoded string;
|
||||||
|
`source_file` = image filename).
|
||||||
|
6. Never rewrite or reorder existing rows — append only.
|
||||||
|
7. Print a summary: added, skipped-as-duplicate, and **undecodable** images
|
||||||
|
(stylized/colored QRs or text-only screenshots) listed by filename for
|
||||||
|
manual entry. `--dry-run` reports without writing.
|
||||||
|
|
||||||
|
## Importer changes (`apex_otp_import.py`)
|
||||||
|
|
||||||
|
- `load_credentials` accepts CSV or XLSX, chosen by file extension.
|
||||||
|
- Crypto (secret/type/algorithm/digits/period) is derived by
|
||||||
|
`parse_otpauth(otpauth_uri)` when an `otpauth_uri` is present; `issuer` and
|
||||||
|
`account` come from their columns. When no `otpauth_uri` is present (legacy
|
||||||
|
xlsx), it falls back to the individual crypto columns as today.
|
||||||
|
- `validate_input` is unchanged in intent and still catches the stage-3 risks
|
||||||
|
(duplicate `issuer:account`, control/`:`/`/` characters, overlength names,
|
||||||
|
bad digits/period, invalid Base32, unsupported algorithm).
|
||||||
|
|
||||||
|
## Testing (hardware-free, TDD)
|
||||||
|
|
||||||
|
- `parse_otpauth`: issuer from param vs. label prefix; percent-decoding;
|
||||||
|
defaults for algorithm/digits/period; account extraction.
|
||||||
|
- `canonical_key`: equal for label-issuer vs. param-issuer forms of the same
|
||||||
|
credential; differs when the secret differs; secret case-insensitive.
|
||||||
|
- Dedup: given existing CSV rows and a batch of candidates, only the new
|
||||||
|
canonical keys are appended; intra-run duplicates collapse.
|
||||||
|
- CSV read/append round-trip; append preserves existing rows verbatim.
|
||||||
|
- `load_credentials` from CSV: crypto comes from `otpauth_uri`, labels from the
|
||||||
|
editable columns; legacy fallback path still works.
|
||||||
|
|
||||||
|
QR image decoding is the untestable I/O edge, validated against the real PNGs.
|
||||||
|
|
||||||
|
## Out of scope (YAGNI)
|
||||||
|
|
||||||
|
`otpauth-migration://` (Google Authenticator batch export protobuf) — reported
|
||||||
|
as unsupported; OCR of text-only screenshots; editing or removing existing rows;
|
||||||
|
backup/recovery `.txt` codes.
|
||||||
|
|
||||||
|
## Files (in the `otp-import` repo)
|
||||||
|
|
||||||
|
New: `qr_to_sheet.py`, `otpauth.py`, `test_qr.py` (parse/canonical/dedup/CSV
|
||||||
|
tests), `docs/qr-to-csv-design.md`. Changed: `apex_otp_import.py`
|
||||||
|
(`load_credentials` CSV + otpauth-authoritative), `test_oath.py` (CSV-load
|
||||||
|
cases), `README.md`. One-time output: `otp-keys.csv` (git-ignored, lives with
|
||||||
|
the private data, not in the repo).
|
||||||
73
otpauth.py
Normal file
73
otpauth.py
Normal file
@@ -0,0 +1,73 @@
|
|||||||
|
"""Parse and canonicalize otpauth:// URIs. Shared by the QR-to-CSV tool and the
|
||||||
|
importer; the otpauth:// string is the immutable source of truth for a
|
||||||
|
credential."""
|
||||||
|
|
||||||
|
from dataclasses import dataclass
|
||||||
|
from urllib.parse import urlparse, parse_qs, unquote, quote
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class OtpAuth:
|
||||||
|
type: str # "totp" / "hotp"
|
||||||
|
issuer: str
|
||||||
|
account: str
|
||||||
|
secret: str # Base32, uppercased, spaces removed
|
||||||
|
algorithm: str # "SHA1" / "SHA256" / "SHA512"
|
||||||
|
digits: int
|
||||||
|
period: int
|
||||||
|
|
||||||
|
|
||||||
|
def parse_otpauth(uri: str) -> OtpAuth:
|
||||||
|
"""Parse an otpauth://TYPE/LABEL?params URI. Issuer comes from the `issuer`
|
||||||
|
parameter when present, else the label prefix before ':'. Account is the
|
||||||
|
label after that prefix."""
|
||||||
|
u = urlparse(uri)
|
||||||
|
otype = u.netloc.lower()
|
||||||
|
label = unquote(u.path.lstrip("/"))
|
||||||
|
q = parse_qs(u.query)
|
||||||
|
|
||||||
|
def g(k, d=""):
|
||||||
|
return q.get(k, [d])[0]
|
||||||
|
|
||||||
|
issuer = unquote(g("issuer"))
|
||||||
|
if ":" in label:
|
||||||
|
label_issuer, account = label.split(":", 1)
|
||||||
|
if not issuer:
|
||||||
|
issuer = label_issuer.strip()
|
||||||
|
else:
|
||||||
|
account = label
|
||||||
|
return OtpAuth(
|
||||||
|
type=otype,
|
||||||
|
issuer=issuer.strip(),
|
||||||
|
account=account.strip(),
|
||||||
|
secret=g("secret").upper().replace(" ", ""),
|
||||||
|
algorithm=(g("algorithm", "SHA1") or "SHA1").upper(),
|
||||||
|
digits=int(g("digits", "6") or "6"),
|
||||||
|
period=int(g("period", "30") or "30"),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def canonical_key(otp: OtpAuth) -> tuple:
|
||||||
|
"""Normalized dedup key for the full otpauth data. Equal for two scans of the
|
||||||
|
same credential (including label-issuer vs. param-issuer forms); different
|
||||||
|
when any crypto-relevant field differs."""
|
||||||
|
return (
|
||||||
|
otp.type.lower(),
|
||||||
|
otp.issuer.strip(),
|
||||||
|
otp.account.strip(),
|
||||||
|
otp.secret.upper().replace(" ", ""),
|
||||||
|
otp.algorithm.upper(),
|
||||||
|
otp.digits,
|
||||||
|
otp.period,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def build_otpauth(issuer: str, account: str, secret: str, otype: str = "totp",
|
||||||
|
algorithm: str = "SHA1", digits: int = 6, period: int = 30) -> str:
|
||||||
|
"""Reconstruct an otpauth:// URI (used to seed URIs for legacy rows)."""
|
||||||
|
label = quote(f"{issuer}:{account}" if issuer else account)
|
||||||
|
params = [f"secret={secret}"]
|
||||||
|
if issuer:
|
||||||
|
params.append(f"issuer={quote(issuer)}")
|
||||||
|
params += [f"algorithm={algorithm}", f"digits={digits}", f"period={period}"]
|
||||||
|
return f"otpauth://{otype}/{label}?" + "&".join(params)
|
||||||
180
qr_to_sheet.py
Normal file
180
qr_to_sheet.py
Normal file
@@ -0,0 +1,180 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""Decode saved OTP QR-code screenshots into rows of the canonical CSV.
|
||||||
|
|
||||||
|
Stage 1 of the pipeline (see docs/qr-to-csv-design.md): decode each QR, dedup on
|
||||||
|
the otpauth:// data, and append only new credentials. The otpauth:// string is
|
||||||
|
the immutable source of truth; the issuer/account columns are editable labels
|
||||||
|
seeded from it.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import argparse
|
||||||
|
import csv
|
||||||
|
import glob
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
|
||||||
|
from otpauth import parse_otpauth, canonical_key
|
||||||
|
|
||||||
|
CSV_COLUMNS = ["issuer", "account", "otpauth_uri", "source_file"]
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# CSV + dedup (pure)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def read_csv_rows(path):
|
||||||
|
"""Read the canonical CSV into a list of dict rows ([] if it doesn't exist)."""
|
||||||
|
if not os.path.exists(path):
|
||||||
|
return []
|
||||||
|
with open(path, newline="", encoding="utf-8") as f:
|
||||||
|
return list(csv.DictReader(f))
|
||||||
|
|
||||||
|
|
||||||
|
def append_csv(path, rows):
|
||||||
|
"""Append rows to the CSV, writing the header first if the file is new.
|
||||||
|
Existing rows are never rewritten or reordered."""
|
||||||
|
new_file = not os.path.exists(path)
|
||||||
|
with open(path, "a", newline="", encoding="utf-8") as f:
|
||||||
|
w = csv.DictWriter(f, fieldnames=CSV_COLUMNS)
|
||||||
|
if new_file:
|
||||||
|
w.writeheader()
|
||||||
|
for r in rows:
|
||||||
|
w.writerow({k: r.get(k, "") for k in CSV_COLUMNS})
|
||||||
|
|
||||||
|
|
||||||
|
def existing_keys(rows):
|
||||||
|
"""Canonical dedup keys for every row that carries an otpauth_uri."""
|
||||||
|
keys = set()
|
||||||
|
for r in rows:
|
||||||
|
uri = (r.get("otpauth_uri") or "").strip()
|
||||||
|
if uri:
|
||||||
|
keys.add(canonical_key(parse_otpauth(uri)))
|
||||||
|
return keys
|
||||||
|
|
||||||
|
|
||||||
|
def dedup_new(candidates, existing):
|
||||||
|
"""From (otpauth_uri, source_file) candidates, return (new_rows, dup_count).
|
||||||
|
Skips any whose canonical otpauth is already present or seen earlier in the
|
||||||
|
batch."""
|
||||||
|
seen = set(existing)
|
||||||
|
new_rows = []
|
||||||
|
dups = 0
|
||||||
|
for uri, source in candidates:
|
||||||
|
key = canonical_key(parse_otpauth(uri))
|
||||||
|
if key in seen:
|
||||||
|
dups += 1
|
||||||
|
continue
|
||||||
|
seen.add(key)
|
||||||
|
o = parse_otpauth(uri)
|
||||||
|
new_rows.append({
|
||||||
|
"issuer": o.issuer,
|
||||||
|
"account": o.account,
|
||||||
|
"otpauth_uri": uri,
|
||||||
|
"source_file": source,
|
||||||
|
})
|
||||||
|
return new_rows, dups
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# QR decoding (I/O)
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
def decode_qr_image(path):
|
||||||
|
"""Decode a QR image to its payload string, or None. Multi-pass to cope with
|
||||||
|
scaling and thresholding differences across screenshots."""
|
||||||
|
import cv2
|
||||||
|
detector = cv2.QRCodeDetector()
|
||||||
|
img = cv2.imread(path)
|
||||||
|
if img is None:
|
||||||
|
return None
|
||||||
|
gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY) if img.ndim == 3 else img
|
||||||
|
attempts = [img, gray]
|
||||||
|
for scale in (2, 3, 4, 0.5):
|
||||||
|
h, w = gray.shape[:2]
|
||||||
|
attempts.append(cv2.resize(gray, (int(w * scale), int(h * scale)),
|
||||||
|
interpolation=cv2.INTER_CUBIC))
|
||||||
|
_, otsu = cv2.threshold(gray, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)
|
||||||
|
attempts.append(otsu)
|
||||||
|
attempts.append(cv2.adaptiveThreshold(gray, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C,
|
||||||
|
cv2.THRESH_BINARY, 31, 5))
|
||||||
|
for a in attempts:
|
||||||
|
try:
|
||||||
|
data, _, _ = detector.detectAndDecode(a)
|
||||||
|
except cv2.error:
|
||||||
|
continue
|
||||||
|
if data:
|
||||||
|
return data
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def scan_images(image_dir):
|
||||||
|
"""Return sorted image paths under image_dir."""
|
||||||
|
paths = []
|
||||||
|
for ext in ("*.png", "*.jpg", "*.jpeg", "*.PNG", "*.JPG"):
|
||||||
|
paths.extend(glob.glob(os.path.join(image_dir, ext)))
|
||||||
|
return sorted(set(paths))
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# CLI
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
DEFAULT_CSV = "otp-keys.csv"
|
||||||
|
|
||||||
|
|
||||||
|
def main(argv=None):
|
||||||
|
p = argparse.ArgumentParser(
|
||||||
|
description="Decode OTP QR screenshots and append new credentials to the "
|
||||||
|
"canonical CSV (deduplicated on the otpauth:// data).")
|
||||||
|
p.add_argument("--images", default=".", help="directory of QR images")
|
||||||
|
p.add_argument("--csv", default=DEFAULT_CSV, help=f"CSV path (default: {DEFAULT_CSV})")
|
||||||
|
p.add_argument("--dry-run", action="store_true", help="report without writing")
|
||||||
|
args = p.parse_args(argv)
|
||||||
|
|
||||||
|
images = scan_images(args.images)
|
||||||
|
if not images:
|
||||||
|
print(f"No QR images found in {args.images}")
|
||||||
|
return 1
|
||||||
|
|
||||||
|
candidates = [] # (otpauth_uri, source_file)
|
||||||
|
undecodable = []
|
||||||
|
unsupported = []
|
||||||
|
for path in images:
|
||||||
|
name = os.path.basename(path)
|
||||||
|
data = decode_qr_image(path)
|
||||||
|
if not data:
|
||||||
|
undecodable.append(name)
|
||||||
|
elif data.startswith("otpauth://"):
|
||||||
|
candidates.append((data, name))
|
||||||
|
elif data.startswith("otpauth-migration://"):
|
||||||
|
unsupported.append(name)
|
||||||
|
else:
|
||||||
|
undecodable.append(name)
|
||||||
|
|
||||||
|
keys = existing_keys(read_csv_rows(args.csv))
|
||||||
|
new_rows, dups = dedup_new(candidates, keys)
|
||||||
|
|
||||||
|
print(f"Scanned {len(images)} image(s): {len(candidates)} otpauth QR(s) decoded.")
|
||||||
|
print(f" {len(new_rows)} new, {dups} duplicate(s) skipped.")
|
||||||
|
if unsupported:
|
||||||
|
print(f" {len(unsupported)} otpauth-migration:// (unsupported): "
|
||||||
|
+ ", ".join(unsupported))
|
||||||
|
if undecodable:
|
||||||
|
print(f" {len(undecodable)} could not be decoded (add manually): "
|
||||||
|
+ ", ".join(undecodable))
|
||||||
|
|
||||||
|
if args.dry_run:
|
||||||
|
for r in new_rows:
|
||||||
|
print(f" + {r['issuer']}:{r['account']} ({r['source_file']})")
|
||||||
|
return 0
|
||||||
|
|
||||||
|
if new_rows:
|
||||||
|
append_csv(args.csv, new_rows)
|
||||||
|
print(f"Appended {len(new_rows)} row(s) to {args.csv}.")
|
||||||
|
else:
|
||||||
|
print("Nothing to append.")
|
||||||
|
return 0
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
sys.exit(main())
|
||||||
18
test_oath.py
18
test_oath.py
@@ -228,6 +228,24 @@ class TestLoadCredentials(unittest.TestCase):
|
|||||||
self.assertEqual(g.source, "g.png")
|
self.assertEqual(g.source, "g.png")
|
||||||
self.assertEqual((creds[1].algorithm, creds[1].digits, creds[1].period), (HMAC_SHA256, 8, 15))
|
self.assertEqual((creds[1].algorithm, creds[1].digits, creds[1].period), (HMAC_SHA256, 8, 15))
|
||||||
|
|
||||||
|
def test_csv_takes_labels_from_columns_and_crypto_from_otpauth(self):
|
||||||
|
import csv
|
||||||
|
d = tempfile.mkdtemp()
|
||||||
|
path = os.path.join(d, "otp-keys.csv")
|
||||||
|
with open(path, "w", newline="", encoding="utf-8") as f:
|
||||||
|
w = csv.writer(f)
|
||||||
|
w.writerow(["issuer", "account", "otpauth_uri", "source_file"])
|
||||||
|
# otpauth label is "Amal Graafstra"; the editable account differs
|
||||||
|
w.writerow(["Heroku", "amal@vivokey.com",
|
||||||
|
"otpauth://totp/Heroku:Amal%20Graafstra?secret=JBSWY3DP"
|
||||||
|
"&issuer=Heroku&algorithm=SHA256&digits=8&period=60", "h.png"])
|
||||||
|
creds = load_credentials(path)
|
||||||
|
c = creds[0]
|
||||||
|
self.assertEqual((c.issuer, c.account), ("Heroku", "amal@vivokey.com")) # from columns
|
||||||
|
self.assertEqual(c.secret, "JBSWY3DP") # from otpauth
|
||||||
|
self.assertEqual((c.oath_type, c.algorithm, c.digits, c.period),
|
||||||
|
(TOTP_TYPE, HMAC_SHA256, 8, 60)) # from otpauth
|
||||||
|
|
||||||
|
|
||||||
class TestCardChannel(unittest.TestCase):
|
class TestCardChannel(unittest.TestCase):
|
||||||
def test_reassembles_61xx_via_send_remaining(self):
|
def test_reassembles_61xx_via_send_remaining(self):
|
||||||
|
|||||||
87
test_qr.py
Normal file
87
test_qr.py
Normal file
@@ -0,0 +1,87 @@
|
|||||||
|
"""Hardware-free tests for the QR->CSV utility (otpauth parsing + dedup)."""
|
||||||
|
import os
|
||||||
|
import tempfile
|
||||||
|
import unittest
|
||||||
|
|
||||||
|
from otpauth import parse_otpauth, canonical_key
|
||||||
|
from qr_to_sheet import existing_keys, dedup_new, append_csv, read_csv_rows
|
||||||
|
|
||||||
|
|
||||||
|
class TestParseOtpauth(unittest.TestCase):
|
||||||
|
def test_basic_totp(self):
|
||||||
|
o = parse_otpauth("otpauth://totp/Google:amal@amal.net?secret=abcde&issuer=Google")
|
||||||
|
self.assertEqual(o.type, "totp")
|
||||||
|
self.assertEqual(o.issuer, "Google")
|
||||||
|
self.assertEqual(o.account, "amal@amal.net")
|
||||||
|
self.assertEqual(o.secret, "ABCDE") # uppercased
|
||||||
|
self.assertEqual((o.algorithm, o.digits, o.period), ("SHA1", 6, 30))
|
||||||
|
|
||||||
|
def test_percent_decoding(self):
|
||||||
|
o = parse_otpauth("otpauth://totp/Google%3Aamal%40amal.net?secret=AA&issuer=Google")
|
||||||
|
self.assertEqual(o.account, "amal@amal.net")
|
||||||
|
|
||||||
|
def test_issuer_from_label_when_no_param(self):
|
||||||
|
o = parse_otpauth("otpauth://totp/SAW:michelle@d.com?secret=AA")
|
||||||
|
self.assertEqual(o.issuer, "SAW")
|
||||||
|
self.assertEqual(o.account, "michelle@d.com")
|
||||||
|
|
||||||
|
def test_label_without_issuer_prefix(self):
|
||||||
|
o = parse_otpauth("otpauth://totp/amalg?secret=AA&issuer=Porkbun")
|
||||||
|
self.assertEqual(o.issuer, "Porkbun")
|
||||||
|
self.assertEqual(o.account, "amalg")
|
||||||
|
|
||||||
|
def test_explicit_params(self):
|
||||||
|
o = parse_otpauth("otpauth://totp/X:y?secret=AA&algorithm=SHA256&digits=8&period=60")
|
||||||
|
self.assertEqual((o.algorithm, o.digits, o.period), ("SHA256", 8, 60))
|
||||||
|
|
||||||
|
|
||||||
|
class TestCanonicalKey(unittest.TestCase):
|
||||||
|
def test_label_issuer_equals_param_issuer(self):
|
||||||
|
a = parse_otpauth("otpauth://totp/Google:amal@amal.net?secret=JBSWY3DP&issuer=Google")
|
||||||
|
b = parse_otpauth("otpauth://totp/amal@amal.net?secret=JBSWY3DP&issuer=Google")
|
||||||
|
self.assertEqual(canonical_key(a), canonical_key(b))
|
||||||
|
|
||||||
|
def test_secret_case_insensitive(self):
|
||||||
|
a = parse_otpauth("otpauth://totp/X:y?secret=jbswy3dp&issuer=X")
|
||||||
|
b = parse_otpauth("otpauth://totp/X:y?secret=JBSWY3DP&issuer=X")
|
||||||
|
self.assertEqual(canonical_key(a), canonical_key(b))
|
||||||
|
|
||||||
|
def test_different_secret_differs(self):
|
||||||
|
a = parse_otpauth("otpauth://totp/X:y?secret=AAAA&issuer=X")
|
||||||
|
b = parse_otpauth("otpauth://totp/X:y?secret=BBBB&issuer=X")
|
||||||
|
self.assertNotEqual(canonical_key(a), canonical_key(b))
|
||||||
|
|
||||||
|
|
||||||
|
A = "otpauth://totp/X:a?secret=AAAA&issuer=X"
|
||||||
|
B = "otpauth://totp/X:b?secret=BBBB&issuer=X"
|
||||||
|
|
||||||
|
|
||||||
|
class TestDedup(unittest.TestCase):
|
||||||
|
def test_dedups_against_existing_and_within_batch(self):
|
||||||
|
existing = [{"issuer": "X", "account": "a", "otpauth_uri": A, "source_file": "a.png"}]
|
||||||
|
keys = existing_keys(existing)
|
||||||
|
new_rows, dups = dedup_new([(A, "a2.png"), (B, "b.png"), (B, "b2.png")], keys)
|
||||||
|
self.assertEqual([r["otpauth_uri"] for r in new_rows], [B])
|
||||||
|
self.assertEqual(new_rows[0]["issuer"], "X")
|
||||||
|
self.assertEqual(new_rows[0]["account"], "b")
|
||||||
|
self.assertEqual(dups, 2)
|
||||||
|
|
||||||
|
|
||||||
|
class TestCsvIO(unittest.TestCase):
|
||||||
|
def test_append_creates_then_appends_preserving_rows(self):
|
||||||
|
d = tempfile.mkdtemp()
|
||||||
|
p = os.path.join(d, "otp-keys.csv")
|
||||||
|
append_csv(p, [{"issuer": "X", "account": "a", "otpauth_uri": "u1", "source_file": "a.png"}])
|
||||||
|
append_csv(p, [{"issuer": "Y", "account": "b", "otpauth_uri": "u2", "source_file": "b.png"}])
|
||||||
|
rows = read_csv_rows(p)
|
||||||
|
self.assertEqual(len(rows), 2)
|
||||||
|
self.assertEqual(rows[0]["issuer"], "X")
|
||||||
|
self.assertEqual(rows[1]["account"], "b")
|
||||||
|
self.assertEqual(rows[1]["otpauth_uri"], "u2")
|
||||||
|
|
||||||
|
def test_read_missing_file_is_empty(self):
|
||||||
|
self.assertEqual(read_csv_rows(os.path.join(tempfile.mkdtemp(), "nope.csv")), [])
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
Reference in New Issue
Block a user