TOTP Generator (python, written by Codex)

Codex wrote this python project from the task below. It installed and ran on a clean Ubuntu 22.04 machine as written.

Task given to the agent:

TASK: TOTP Generator

Write a program that generates and validates Time-based One-Time Passwords (TOTP) compatible with RFC 6238, supporting secret key management, QR code URI generation, and multi-account storage.

FUNCTIONAL REQUIREMENTS:
- Support subcommands: generate (create a new TOTP secret), code (show current OTP code), verify (check if a code is valid), and list (show all stored accounts)
- generate: Create a cryptographically random base32-encoded secret key of configurable length (--length flag, default 20 bytes)
- code: Display the current 6-digit TOTP code for a given account, with a countdown timer showing seconds until the code expires (30-second default period)
- verify: Accept a 6-digit code and check if it matches the current or adjacent time windows (configurable drift tolerance via --drift flag, default 1 window)
- Support configurable TOTP parameters: digit count (6 or 8), time period (30 or 60 seconds), and hash algorithm (SHA-1, SHA-256, SHA-512)
- Generate otpauth:// URIs compatible with authenticator apps (Google Authenticator, Authy)
- Store account secrets in an encrypted local JSON file using a master password
- Support multiple accounts with labels (--account flag with issuer:username format)
- Print the current code, remaining seconds, and next code to console
- If no arguments are given, generate a demo account with a random secret, display the current code and the otpauth:// URI, verify the current code, then show what happens with an incorrect code
- Handle errors: invalid base32 secrets, expired codes, duplicate account names, wrong master password

Create a complete Python project for a clean Ubuntu 22.04 machine with only Python 3.10+ installed. Include:
- Source code
- requirements.txt with all dependencies (direct and transitive) pinned to exact versions
- README.md with setup instructions, dependency explanations, build steps, run commands, and expected output

[]

# TOTP Generator (Python)

RFC 6238-compatible TOTP generator/verifier with encrypted multi-account storage.

## Requirements

- Ubuntu 22.04
- Python 3.10+

## Dependencies (Pinned)

- `cryptography==44.0.0`
- `cffi==1.17.1`
- `pycparser==2.22`

`requirements.txt` includes direct and transitive dependencies explicitly pinned.

## Setup

```bash
python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
```

## Run

```bash
python src/main.py generate --master "<password>" --account "Issuer:user@example.com" [--length 20] [--digits 6|8] [--period 30|60] [--algorithm SHA-1|SHA-256|SHA-512] [--store ./totp_accounts.enc.json]
python src/main.py code --master "<password>" --account "Issuer:user@example.com" [--store ./totp_accounts.enc.json]
python src/main.py verify --master "<password>" --account "Issuer:user@example.com" --code 123456 [--drift 1] [--store ./totp_accounts.enc.json]
python src/main.py list --master "<password>" [--store ./totp_accounts.enc.json]
```

## Features

- Subcommands: `generate`, `code`, `verify`, `list`
- Base32 secrets with configurable byte length
- TOTP code generation for 6/8 digits, 30/60s period, SHA-1/256/512
- Drift-tolerant verification
- `otpauth://` URI generation
- Encrypted JSON storage using PBKDF2 + AES-GCM
- Duplicate account and wrong password handling
- No-args demo mode

cryptography==44.0.0
cffi==1.17.1
pycparser==2.22

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import base64
import hashlib
import hmac
import json
import secrets
import struct
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from urllib.parse import urlencode, quote

from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from cryptography.hazmat.primitives import hashes


DEFAULT_STORE = Path("totp_accounts.enc.json").resolve()
PBKDF2_ITERS = 150_000


def now_iso() -> str:
    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")


def normalize_algorithm(raw: str) -> tuple[str, str]:
    value = raw.strip().upper()
    mapping = {
        "SHA-1": ("SHA-1", "sha1"),
        "SHA-256": ("SHA-256", "sha256"),
        "SHA-512": ("SHA-512", "sha512"),
    }
    if value not in mapping:
        raise ValueError("Unsupported algorithm. Use SHA-1, SHA-256, or SHA-512.")
    return mapping[value]


def parse_digits(raw: str) -> int:
    v = int(raw)
    if v not in (6, 8):
        raise ValueError("Digits must be 6 or 8.")
    return v


def parse_period(raw: str) -> int:
    v = int(raw)
    if v not in (30, 60):
        raise ValueError("Period must be 30 or 60.")
    return v


def parse_drift(raw: str) -> int:
    v = int(raw)
    if v < 0 or v > 10:
        raise ValueError("Drift must be in range 0..10.")
    return v


def validate_account_label(label: str) -> str:
    if ":" not in label:
        raise ValueError("Account must be in issuer:username format.")
    return label


def generate_base32_secret(length: int) -> str:
    if length < 10 or length > 128:
        raise ValueError("Length must be between 10 and 128 bytes.")
    return base64.b32encode(secrets.token_bytes(length)).decode("ascii").rstrip("=")


def decode_base32(secret: str) -> bytes:
    clean = "".join(secret.split()).upper().strip("=")
    if not clean:
        raise ValueError("Invalid base32 secret.")
    padding = "=" * ((8 - len(clean) % 8) % 8)
    try:
        return base64.b32decode(clean + padding, casefold=True)
    except Exception as exc:  # pragma: no cover
        raise ValueError("Invalid base32 secret.") from exc


def totp_at(secret_b32: str, digits: int, algorithm_hashlib: str, counter: int) -> str:
    secret = decode_base32(secret_b32)
    msg = struct.pack(">Q", counter)
    digest = hmac.new(secret, msg, getattr(hashlib, algorithm_hashlib)).digest()
    offset = digest[-1] & 0x0F
    code_int = ((digest[offset] & 0x7F) << 24) | ((digest[offset + 1] & 0xFF) << 16) | ((digest[offset + 2] & 0xFF) << 8) | (digest[offset + 3] & 0xFF)
    return str(code_int % (10**digits)).zfill(digits)


@dataclass
class TotpView:
    current: str
    next_code: str
    remaining: int
    counter: int


def totp_now(secret_b32: str, digits: int, period: int, algorithm_hashlib: str) -> TotpView:
    now = int(time.time())
    counter = now // period
    remaining = period - (now % period)
    return TotpView(
        current=totp_at(secret_b32, digits, algorithm_hashlib, counter),
        next_code=totp_at(secret_b32, digits, algorithm_hashlib, counter + 1),
        remaining=remaining,
        counter=counter,
    )


def verify_totp(secret_b32: str, code: str, digits: int, period: int, algorithm_hashlib: str, drift: int) -> tuple[bool, int | None]:
    if not code.isdigit() or len(code) not in (6, 8):
        raise ValueError("Code must be 6 or 8 digits.")
    now = int(time.time())
    counter = now // period
    for offset in range(-drift, drift + 1):
        if totp_at(secret_b32, digits, algorithm_hashlib, counter + offset) == code:
            return True, offset
    return False, None


def make_otpauth_uri(account: str, secret: str, digits: int, period: int, algo_label: str) -> str:
    issuer, username = account.split(":", 1)
    label = quote(f"{issuer}:{username}")
    query = urlencode(
        {
            "secret": secret,
            "issuer": issuer,
            "algorithm": algo_label.replace("-", ""),
            "digits": digits,
            "period": period,
        }
    )
    return f"otpauth://totp/{label}?{query}"


def derive_key(master_password: str, salt: bytes) -> bytes:
    kdf = PBKDF2HMAC(algorithm=hashes.SHA256(), length=32, salt=salt, iterations=PBKDF2_ITERS)
    return kdf.derive(master_password.encode("utf-8"))


def encrypt_store(plaintext_json: str, master_password: str) -> dict[str, Any]:
    salt = secrets.token_bytes(16)
    nonce = secrets.token_bytes(12)
    key = derive_key(master_password, salt)
    aesgcm = AESGCM(key)
    ciphertext = aesgcm.encrypt(nonce, plaintext_json.encode("utf-8"), None)
    return {
        "version": 1,
        "kdf": "PBKDF2-SHA256",
        "iterations": PBKDF2_ITERS,
        "salt": base64.b64encode(salt).decode("ascii"),
        "nonce": base64.b64encode(nonce).decode("ascii"),
        "data": base64.b64encode(ciphertext).decode("ascii"),
    }


def decrypt_store(blob: dict[str, Any], master_password: str) -> str:
    try:
        salt = base64.b64decode(blob["salt"])
        nonce = base64.b64decode(blob["nonce"])
        ciphertext = base64.b64decode(blob["data"])
        key = derive_key(master_password, salt)
        aesgcm = AESGCM(key)
        plaintext = aesgcm.decrypt(nonce, ciphertext, None)
        return plaintext.decode("utf-8")
    except Exception as exc:
        raise ValueError("Wrong master password or corrupted encrypted store.") from exc


def load_db(store_path: Path, master_password: str) -> dict[str, Any]:
    if not store_path.exists():
        return {"version": 1, "accounts": []}
    blob = json.loads(store_path.read_text(encoding="utf-8"))
    plaintext = decrypt_store(blob, master_password)
    db = json.loads(plaintext)
    if not isinstance(db.get("accounts"), list):
        raise ValueError("Encrypted store is malformed.")
    return db


def save_db(store_path: Path, master_password: str, db: dict[str, Any]) -> None:
    blob = encrypt_store(json.dumps(db, indent=2), master_password)
    store_path.parent.mkdir(parents=True, exist_ok=True)
    store_path.write_text(json.dumps(blob, indent=2) + "\n", encoding="utf-8")


def require_account(db: dict[str, Any], account: str) -> dict[str, Any]:
    for item in db["accounts"]:
        if item["label"] == account:
            return item
    raise ValueError(f"Account not found: {account}")


def command_generate(args: argparse.Namespace) -> int:
    master = args.master
    if not master:
        raise ValueError("Missing --master <password> for encrypted storage.")
    account = validate_account_label(args.account)
    digits = parse_digits(str(args.digits))
    period = parse_period(str(args.period))
    algo_label, algo_hash = normalize_algorithm(args.algorithm)
    secret = generate_base32_secret(args.length)
    store = Path(args.store).resolve()

    db = load_db(store, master)
    if any(a["label"] == account for a in db["accounts"]):
        raise ValueError(f"Duplicate account: {account}")

    record = {
        "label": account,
        "secret": secret,
        "digits": digits,
        "period": period,
        "algorithm": algo_label,
        "createdAt": now_iso(),
    }
    db["accounts"].append(record)
    save_db(store, master, db)

    uri = make_otpauth_uri(account, secret, digits, period, algo_label)
    view = totp_now(secret, digits, period, algo_hash)
    print(f"Account added : {account}")
    print(f"Store file    : {store}")
    print(f"Secret (base32): {secret}")
    print(f"otpauth URI   : {uri}")
    print(f"Current code  : {view.current}")
    print(f"Next code     : {view.next_code}")
    print(f"Expires in    : {view.remaining}s")
    return 0


def command_code(args: argparse.Namespace) -> int:
    if not args.master:
        raise ValueError("Missing --master <password> for encrypted storage.")
    account_label = validate_account_label(args.account)
    db = load_db(Path(args.store).resolve(), args.master)
    account = require_account(db, account_label)
    algo_label, algo_hash = normalize_algorithm(account["algorithm"])
    _ = algo_label
    view = totp_now(account["secret"], account["digits"], account["period"], algo_hash)
    print(f"Account       : {account['label']}")
    print(f"Current code  : {view.current}")
    print(f"Next code     : {view.next_code}")
    print(f"Expires in    : {view.remaining}s")
    return 0


def command_verify(args: argparse.Namespace) -> int:
    if not args.master:
        raise ValueError("Missing --master <password> for encrypted storage.")
    account_label = validate_account_label(args.account)
    drift = parse_drift(str(args.drift))
    db = load_db(Path(args.store).resolve(), args.master)
    account = require_account(db, account_label)
    _, algo_hash = normalize_algorithm(account["algorithm"])
    ok, offset = verify_totp(
        account["secret"],
        args.code,
        account["digits"],
        account["period"],
        algo_hash,
        drift,
    )
    if not ok:
        print("Verification: INVALID (expired or incorrect code)")
        return 2
    if offset == 0:
        print("Verification: VALID (current window)")
    else:
        print(f"Verification: VALID (window offset {offset})")
    return 0


def command_list(args: argparse.Namespace) -> int:
    if not args.master:
        raise ValueError("Missing --master <password> for encrypted storage.")
    store = Path(args.store).resolve()
    db = load_db(store, args.master)
    if not db["accounts"]:
        print("No accounts stored.")
        return 0
    print(f"Accounts in {store}:")
    for account in db["accounts"]:
        print(
            f"- {account['label']} | {account['algorithm']} | digits={account['digits']} | period={account['period']}s"
        )
    return 0


def run_demo() -> int:
    demo_store = Path("totp_demo.enc.json").resolve()
    if demo_store.exists():
        demo_store.unlink()
    master = "demo-master-password"
    account = "DemoIssuer:demo.user@example.com"

    print("Running demo...\n")
    ns_gen = argparse.Namespace(
        master=master,
        store=str(demo_store),
        account=account,
        length=20,
        digits=6,
        period=30,
        algorithm="SHA-1",
    )
    command_generate(ns_gen)

    db = load_db(demo_store, master)
    acc = require_account(db, account)
    _, algo_hash = normalize_algorithm(acc["algorithm"])
    current = totp_now(acc["secret"], acc["digits"], acc["period"], algo_hash).current
    print(f"\nDemo verify with correct code ({current})")
    rc1 = command_verify(
        argparse.Namespace(master=master, store=str(demo_store), account=account, code=current, drift=1)
    )
    print("\nDemo verify with incorrect code (000000)")
    rc2 = command_verify(
        argparse.Namespace(master=master, store=str(demo_store), account=account, code="000000", drift=1)
    )
    return rc1 if rc1 != 0 else rc2


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="TOTP Generator (RFC 6238)")
    sub = parser.add_subparsers(dest="command")

    g = sub.add_parser("generate")
    g.add_argument("--master", required=True)
    g.add_argument("--account", required=True, help="issuer:username")
    g.add_argument("--length", type=int, default=20)
    g.add_argument("--digits", type=int, default=6)
    g.add_argument("--period", type=int, default=30)
    g.add_argument("--algorithm", default="SHA-1")
    g.add_argument("--store", default=str(DEFAULT_STORE))

    c = sub.add_parser("code")
    c.add_argument("--master", required=True)
    c.add_argument("--account", required=True)
    c.add_argument("--store", default=str(DEFAULT_STORE))

    v = sub.add_parser("verify")
    v.add_argument("--master", required=True)
    v.add_argument("--account", required=True)
    v.add_argument("--code", required=True)
    v.add_argument("--drift", type=int, default=1)
    v.add_argument("--store", default=str(DEFAULT_STORE))

    l = sub.add_parser("list")
    l.add_argument("--master", required=True)
    l.add_argument("--store", default=str(DEFAULT_STORE))

    return parser


def main() -> int:
    parser = build_parser()
    args = parser.parse_args()
    if not args.command:
        return run_demo()

    try:
        if args.command == "generate":
            return command_generate(args)
        if args.command == "code":
            return command_code(args)
        if args.command == "verify":
            return command_verify(args)
        if args.command == "list":
            return command_list(args)
        print("Error: unknown subcommand.")
        return 1
    except ValueError as exc:
        print(f"Error: {exc}")
        return 1
    except FileNotFoundError as exc:
        print(f"Error: {exc}")
        return 1
    except PermissionError:
        print("Error: permission denied.")
        return 1
    except Exception as exc:  # pragma: no cover
        print(f"Error: {exc}")
        return 1


if __name__ == "__main__":
    raise SystemExit(main())