mirror of
https://github.com/Nighthawk42/bnet_auth_tool.git
synced 2026-08-30 14:32:26 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Nighthawk42andClaude Opus 4.8
|
c0a5dc49bf | ||
|
Nighthawk
|
6580432070 | ||
|
Nighthawk42andClaude Opus 4.8
|
24c6fb393a | ||
|
Nighthawk42andClaude Opus 4.8
|
c416e98162 | ||
|
Nighthawk42andClaude Opus 4.8
|
023ccd9f74 | ||
|
Nighthawk
|
0db5fa287e | ||
|
weslenng
|
70e5d0fc15 | ||
|
Nighthawk
|
e0ee0067f0 | ||
|
Nighthawk
|
9f1860a0a4 | ||
|
Nighthawk
|
9cd1dcbbdf | ||
|
Nighthawk
|
fbd349d653 | ||
|
Nighthawk
|
e13b27d2ac | ||
|
Nighthawk
|
a6c065acc6 | ||
|
Nighthawk
|
0199878312 | ||
|
Nighthawk
|
fa02a8f4b7 | ||
|
Nighthawk
|
351c9e59b6 | ||
|
Nighthawk
|
50c7a64c85 | ||
|
Nighthawk
|
cd8e6c315d | ||
|
Nighthawk
|
9dc551c434 | ||
|
Nighthawk
|
2fdee4de71 | ||
|
Nighthawk
|
0e4797d8ca |
@@ -0,0 +1,5 @@
|
||||
blank_issues_enabled: false
|
||||
contact_links:
|
||||
- name: Official Blizzard Support
|
||||
url: https://us.battle.net/support/en/
|
||||
about: If you are locked out of your Battle.net account, contact Blizzard. The project maintainer cannot recover your account.
|
||||
@@ -0,0 +1,55 @@
|
||||
name: "Bug Report / API Endpoint Update"
|
||||
description: "Report a bug or share updated Blizzard endpoint details. Account-recovery requests will be closed."
|
||||
title: "[Bug]: "
|
||||
labels: ["bug"]
|
||||
body:
|
||||
- type: markdown
|
||||
attributes:
|
||||
value: |
|
||||
Thanks for contributing!
|
||||
|
||||
* **Account recovery:** the maintainer has no access to Blizzard's backend and
|
||||
cannot recover accounts. Contact Blizzard Customer Support instead.
|
||||
* **Online flows are unverified** against the live Blizzard API and may be blocked
|
||||
at any time. The offline vault / TOTP / migration features are the supported core.
|
||||
|
||||
- type: checkboxes
|
||||
id: acknowledgment
|
||||
attributes:
|
||||
label: Acknowledgment
|
||||
options:
|
||||
- label: "I understand the maintainer CANNOT recover my Battle.net account."
|
||||
required: true
|
||||
- label: "This is not an account-recovery request."
|
||||
required: true
|
||||
|
||||
- type: dropdown
|
||||
id: area
|
||||
attributes:
|
||||
label: Affected area
|
||||
options:
|
||||
- "Vault / encryption (offline)"
|
||||
- "TOTP / QR reconstruction (offline)"
|
||||
- "Legacy migration (offline)"
|
||||
- "Attach / retrieve (online, unverified)"
|
||||
- "Other"
|
||||
validations:
|
||||
required: true
|
||||
|
||||
- type: textarea
|
||||
id: details
|
||||
attributes:
|
||||
label: Description
|
||||
description: "What happened, what you expected, and steps to reproduce. Include the tool version (`bnet-auth --version`) and your OS."
|
||||
placeholder: "Steps to reproduce..."
|
||||
validations:
|
||||
required: true
|
||||
|
||||
- type: textarea
|
||||
id: endpoint-data
|
||||
attributes:
|
||||
label: Updated endpoint data (optional)
|
||||
description: "If you've captured new Blizzard API endpoints/headers/payloads, share them here so settings.yaml can be updated."
|
||||
placeholder: "e.g., the device endpoint now lives at /v3/..."
|
||||
validations:
|
||||
required: false
|
||||
+66
@@ -0,0 +1,66 @@
|
||||
# ---------------------------------------------------------------------------
|
||||
# Secrets — NEVER commit authenticator material
|
||||
# ---------------------------------------------------------------------------
|
||||
# Legacy per-serial backups dumped by old versions
|
||||
battlenet_authenticator_*.json
|
||||
reconstructed_*.json
|
||||
# QR codes contain the raw TOTP secret
|
||||
*.png
|
||||
!docs/**/*.png
|
||||
# Encrypted vault / exported secrets
|
||||
*.vault
|
||||
vault.json
|
||||
secrets.yaml
|
||||
# User-edited local settings copy (may contain custom endpoints)
|
||||
/settings.local.yaml
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Python
|
||||
# ---------------------------------------------------------------------------
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
*$py.class
|
||||
*.egg-info/
|
||||
.eggs/
|
||||
build/
|
||||
dist/
|
||||
wheels/
|
||||
*.egg
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# PyInstaller / flet pack build artifacts
|
||||
# ---------------------------------------------------------------------------
|
||||
*.spec
|
||||
*.exe
|
||||
*.zip
|
||||
/release/
|
||||
# flet pack scratch dirs
|
||||
*.flet
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Virtual environments / tooling
|
||||
# ---------------------------------------------------------------------------
|
||||
.venv/
|
||||
venv/
|
||||
env/
|
||||
.uv/
|
||||
.python-version
|
||||
# uv.lock is intentionally committed for reproducible installs
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test / lint caches
|
||||
# ---------------------------------------------------------------------------
|
||||
.pytest_cache/
|
||||
.ruff_cache/
|
||||
.coverage
|
||||
htmlcov/
|
||||
.mypy_cache/
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Editors / OS
|
||||
# ---------------------------------------------------------------------------
|
||||
.idea/
|
||||
.vscode/
|
||||
*.swp
|
||||
.DS_Store
|
||||
Thumbs.db
|
||||
@@ -0,0 +1,63 @@
|
||||
# AGENTS.md
|
||||
|
||||
Guidance for AI agents and human contributors working on `bnet_auth_tool`.
|
||||
|
||||
## What this is
|
||||
|
||||
A Python CLI for managing Battle.net software authenticators. Two halves:
|
||||
|
||||
- **Offline (the important, fully-tested half):** an encrypted vault of authenticator
|
||||
secrets, TOTP/QR reconstruction, and migration of legacy backups.
|
||||
- **Online (unverified):** attach/retrieve flows against Blizzard's identity API. These
|
||||
may be blocked by Blizzard at any time. **Do not claim they are "fixed"** — keep the
|
||||
tempered "unverified" framing in the README and `--help`/menu text.
|
||||
|
||||
## Architecture
|
||||
|
||||
Source lives under `src/bnet_auth_tool/` (a proper package; there is no top-level script).
|
||||
|
||||
| Module | Responsibility |
|
||||
| --- | --- |
|
||||
| `config.py` | Load bundled `settings.yaml` + user override; resolve config/data dirs (`platformdirs`). Typed `Settings`/`ApiConfig`/`CryptoConfig`/`TotpConfig`. |
|
||||
| `crypto.py` | `EncryptionManager`: scrypt+AES‑256‑GCM encrypt; decrypt dispatches on a versioned header (scrypt / PBKDF2 600k / legacy PBKDF2 100k). |
|
||||
| `storage.py` | `Vault`: single encrypted file keyed by serial; add/list/get/remove. |
|
||||
| `fileio.py` | Atomic, `0600`-hardened writes (`atomic_write_bytes`, `_harden`). |
|
||||
| `api.py` | `BattleNetAuthenticator` online client; leak-safe error handling. |
|
||||
| `totp.py` | hex→base32, `otpauth://` URL builder, QR PNG. |
|
||||
| `migrate.py` | Discover + import legacy `battlenet_authenticator_*.json` files into the vault. |
|
||||
| `cli.py` | Interactive menu **and** argparse subcommands; `main()` is the entry point. |
|
||||
| `gui.py` | Optional Flet desktop GUI (`bnet-auth-gui`); imperative, same vault/crypto as the CLI. Behind the `gui` extra. |
|
||||
| `errors.py` | Exception hierarchy rooted at `BnetAuthError`. |
|
||||
|
||||
`settings.yaml` is shipped inside the package and copied to the user's config dir on first
|
||||
run. Endpoints/KDF/TOTP params are config, not code — change them there.
|
||||
|
||||
## Hard constraints (do not break)
|
||||
|
||||
1. **Legacy decryption must keep working.** Files encrypted by v1.x — including pre‑v1.3
|
||||
files with *no* `kdf_iterations` field — must still decrypt. Covered by `tests/test_crypto.py`.
|
||||
2. **Secrets are sensitive.** Never log raw device secrets, restore codes, session tokens,
|
||||
or full server error bodies. Vault/QR files are written `0600` and atomically.
|
||||
3. **Online flows stay labelled unverified.** No optimistic "it works now" claims.
|
||||
|
||||
## Conventions
|
||||
|
||||
- Python **3.9+**; `from __future__ import annotations` in every module (so `X | None`
|
||||
annotations are fine).
|
||||
- Lint/format with **ruff**; config in `pyproject.toml`.
|
||||
- Keep crypto parameters fast in tests (small scrypt `n`) — see `tests/conftest.py`.
|
||||
- The GUI targets **Flet 0.85+** in imperative style: `ft.run(main)`, `page.show_dialog()` /
|
||||
`page.pop_dialog()`, `page.clipboard.set()`, `page.window.width`. It holds no logic of its
|
||||
own — all crypto/storage/TOTP goes through the same modules as the CLI.
|
||||
|
||||
## Workflow
|
||||
|
||||
```bash
|
||||
uv sync --extra dev
|
||||
uv run pytest
|
||||
uv run ruff check .
|
||||
uv run bnet-auth --help
|
||||
```
|
||||
|
||||
When changing the encryption format, bump the `format` header in `crypto.py` and add a
|
||||
back-compat test rather than mutating the existing decrypt paths.
|
||||
@@ -0,0 +1,5 @@
|
||||
# CLAUDE.md
|
||||
|
||||
This project's agent guidance lives in [AGENTS.md](AGENTS.md). Read it before making
|
||||
changes — it covers the architecture, the hard constraints (legacy-decryption back-compat,
|
||||
secret hygiene, and the "online flows are unverified" framing), and the dev workflow.
|
||||
@@ -1,6 +1,6 @@
|
||||
MIT License
|
||||
|
||||
Copyright (c) 2024 Nighthawk
|
||||
Copyright (c) 2024-2026 Nighthawk42
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
of this software and associated documentation files (the "Software"), to deal
|
||||
|
||||
@@ -1,112 +1,169 @@
|
||||
# Battle.net Authenticator Tool
|
||||
|
||||
Version: 1.3.0
|
||||
|
||||
A Python-based command-line tool for managing Battle.net software authenticators. This tool allows you to:
|
||||
|
||||
* Attach a new software authenticator to your Battle.net account.
|
||||
* Retrieve the secret details of an *existing* software authenticator using its Serial Number and Restore Code.
|
||||
* Generate standard TOTP (Time-Based One-Time Password) configuration (Base32 secret, `otpauth://` URL) and a QR code compatible with common authenticator apps (like Google Authenticator, Authy, Microsoft Authenticator, etc.).
|
||||
* Optionally encrypt the saved authenticator details using strong AES-256-GCM encryption derived from a user-provided passphrase.
|
||||
* Load previously saved authenticator details (plain or encrypted) to regenerate the TOTP URL and QR code.
|
||||
* Encrypt previously saved plain-text authenticator files.
|
||||
* Decrypt previously encrypted authenticator files (for viewing or saving as plain text).
|
||||
|
||||
**Disclaimer:** This tool interacts with your Battle.net account and handles sensitive security information (authenticator secrets). Use it responsibly and at your own risk. Ensure you understand the security implications and securely manage any generated files and passphrases. The author is not responsible for any damage or loss resulting from the use of this tool.
|
||||
|
||||
## Features
|
||||
|
||||
* **Attach New Authenticator:** Guides through attaching a new virtual authenticator.
|
||||
* **Retrieve Existing Secret:** Recovers the secret key if you have the Serial and Restore Code.
|
||||
* **Standard TOTP Output:** Generates Base32 secrets and `otpauth://` URLs compatible with RFC 6238 (SHA1, 8 Digits, 30s period for Battle.net).
|
||||
* **QR Code Generation:** Creates `.png` QR codes for easy import into authenticator apps.
|
||||
* **Secure File Encryption (Optional):** Uses AES-256-GCM with PBKDF2 (increased to 600k iterations in v1.3.0) for strong protection of saved secrets.
|
||||
* **File Management:** Load, reconstruct, encrypt, and decrypt saved authenticator files (`.json`).
|
||||
* **Backward Compatibility:** Can decrypt files encrypted with older versions (v1.2) that used fewer PBKDF2 iterations (100k).
|
||||
* **Region Support:** Works with session tokens from various Battle.net regions (US, EU, KR, TW, CN detected).
|
||||
|
||||
## Security Warning
|
||||
|
||||
* **Backup Your Data:** The `.json` file generated by this tool contains your authenticator's Serial, Restore Code, and the critical Device Secret. **Losing this file (especially if unencrypted) and the Restore Code means you could lose access to your authenticator.** Back up this file securely (e.g., encrypted external drive, password manager).
|
||||
* **Protect Your Passphrase:** If you choose to encrypt the `.json` file, your passphrase is the *only* way to decrypt it. **There is no recovery for a lost passphrase.** Choose a strong, unique passphrase and store it securely.
|
||||
* **Secure QR Codes:** The generated `.png` QR code also contains your secret key. Treat it as securely as the `.json` file. Delete it after successfully importing it into your authenticator app(s).
|
||||
* **Session Token Exposure:** The process requires obtaining a temporary session token from your browser. Ensure you do this in a secure environment and log out afterwards if using a public computer.
|
||||
|
||||
## Important Notice for Users Upgrading from v1.2
|
||||
|
||||
Version `1.3.0` introduces a significant improvement to the security of *newly encrypted* files by increasing the **PBKDF2 iteration count**. This makes brute-force attacks against the encryption passphrase much harder.
|
||||
|
||||
**Compatibility:**
|
||||
|
||||
* **✅ v1.3.0 CAN decrypt files encrypted by v1.2:** The new version automatically detects if a file is missing the iteration count field and assumes the old count for decryption. Your old encrypted files will work fine with v1.3.0.
|
||||
* **❌ v1.2 CANNOT decrypt files encrypted by v1.3.0:** If you encrypt a file using v1.3.0 (either by attaching/retrieving and choosing encrypt, or using the "Encrypt existing" option), the older v1.2 script will *not* be able to decrypt it due to the mismatch in iteration counts.
|
||||
|
||||
**Recommendation:**
|
||||
|
||||
* **Upgrade:** All users should upgrade to v1.3.0 or later for the improved security and compatibility handling.
|
||||
* **(Optional) Re-encrypt:** For maximum security benefit on your existing files, you can:
|
||||
1. Use v1.3.0 to **decrypt** your old `.json` file (using option 5 and saving to a *new* plain file).
|
||||
2. Use v1.3.0 to **encrypt** that newly saved plain file (using option 4). This will re-encrypt it with the stronger 600k iterations.
|
||||
3. Securely delete the intermediate plain text file.
|
||||
|
||||
## Requirements
|
||||
|
||||
* Python 3.7+
|
||||
* Required Python libraries (install via pip):
|
||||
* `requests`
|
||||
* `cryptography`
|
||||
* `qrcode[pil]` (This installs both `qrcode` and the `Pillow` imaging library)
|
||||
|
||||
## Installation
|
||||
|
||||
**Recommended:** Download the pre-compiled executable from the [Releases page](https://github.com/Nighthawk42/bnet_auth_tool/releases/). This avoids needing Python or manual library installation.
|
||||
|
||||
**Manual (using Python):**
|
||||
|
||||
1. Ensure Python 3.7+ and `pip` are installed and accessible from your command line.
|
||||
2. Clone the repository or download the source code (`.zip`).
|
||||
```bash
|
||||
git clone https://github.com/Nighthawk42/bnet_auth_tool.git
|
||||
cd bnet_auth_tool
|
||||
```
|
||||
3. Install the required libraries:
|
||||
```bash
|
||||
pip install -r requirements.txt
|
||||
# Or: pip install requests cryptography "qrcode[pil]"
|
||||
```
|
||||
|
||||
## Usage
|
||||
|
||||
1. Open your terminal or command prompt.
|
||||
2. Navigate to the directory where you placed the script or executable.
|
||||
3. Run the tool:
|
||||
* If using the Python script:
|
||||
```bash
|
||||
python bnet_auth_tool.py
|
||||
# or potentially: python3 bnet_auth_tool.py
|
||||
```
|
||||
* If using the executable (Windows example):
|
||||
```bash
|
||||
bnet_auth_tool.exe
|
||||
```
|
||||
4. The tool will display a menu with available actions:
|
||||
* **Attach a new authenticator:** Guides you through getting a session token and attaches a new virtual authenticator, saving the details.
|
||||
* **Retrieve existing device secret:** Guides you through getting a session token and uses your existing Serial/Restore code to retrieve the secret, saving the details.
|
||||
* **Reconstruct TOTP from JSON:** Loads a saved `.json` file (plain or encrypted, prompts for passphrase if needed) and displays the TOTP info / generates a QR code.
|
||||
* **Encrypt existing plain JSON file(s):** Finds unencrypted `.json` files in the directory, prompts you to select which ones to encrypt, and asks for a passphrase. *Overwrites the original file.*
|
||||
* **Decrypt an encrypted JSON file:** Prompts you to select an encrypted `.json` file, asks for the passphrase, and then offers to display the decrypted data or save it to a *new* plain-text `.json` file.
|
||||
* **Exit:** Closes the tool.
|
||||
5. Follow the on-screen prompts for each action. Pay close attention to instructions for obtaining the session token and handling passphrases.
|
||||
> [!IMPORTANT]
|
||||
> **Work in Progress:** This rewrite is actively under development. Please report bugs via issues, and code contributions via Pull Requests are highly encouraged!
|
||||
>
|
||||
> **Development Note:** Parts of this codebase were generated/optimized using Claude. If you are ideologically opposed to LLM-assisted development, you are entirely free to skip using this software or fork it and strip it out yourself.
|
||||
|
||||
---
|
||||
|
||||
## Output Files
|
||||
> Back up Battle.net authenticator TOTP secrets in an encrypted local vault and export
|
||||
> them to any authenticator app — via CLI or an optional desktop GUI. Offline features
|
||||
> fully work; online attach/retrieve is unverified against Blizzard's API.
|
||||
|
||||
* **`.json` File:** (`battlenet_authenticator_SERIAL.json`)
|
||||
* Contains the Serial Number, Restore Code, raw hexadecimal Device Secret, Base32 secret, `otpauth://` URL, and a timestamp. (Encrypted files also contain salt, nonce, and iteration count).
|
||||
* This file is crucial for backup and recovery.
|
||||
* Can be saved as plain text or encrypted (recommended).
|
||||
* **`.png` File:** (`battlenet_authenticator_SERIAL.png` or `reconstructed_SERIAL.png`)
|
||||
* A QR code image containing the `otpauth://` URL.
|
||||
* Scan this with your authenticator app to add the key.
|
||||
* Securely delete after successful import.
|
||||
A tool for managing **Battle.net software authenticators**, available as both a CLI
|
||||
(`bnet-auth`) and an optional desktop GUI (`bnet-auth-gui`). It can attach/retrieve
|
||||
authenticator secrets online, and — most importantly — keep your TOTP backups in a single
|
||||
**encrypted local vault** so you can re-import them into any standard authenticator app
|
||||
(Aegis, Bitwarden, 1Password, Google Authenticator, …).
|
||||
|
||||
> [!IMPORTANT]
|
||||
> **Online status is unverified.** The attach/retrieve flows depend on
|
||||
> Blizzard's identity API, which has changed before and may be blocked again.
|
||||
> They are kept here and made configurable, but are **not** guaranteed to work
|
||||
> against the live backend. The **offline** vault / TOTP / migration features are
|
||||
> fully tested and work regardless of Blizzard's API.
|
||||
|
||||
---
|
||||
|
||||
## What it does
|
||||
|
||||
* **Attach a new authenticator** (online, unverified) and store the secret in your vault.
|
||||
* **Retrieve an existing device secret** (online, unverified) from a serial + restore code.
|
||||
* **Reconstruct TOTP** keys and **QR codes** (RFC 6238: SHA1, 8 digits, 30s) from the vault.
|
||||
* **Encrypted vault** — all authenticators live in one AES‑256‑GCM file protected by a
|
||||
passphrase (scrypt key derivation), stored in your OS user-data directory.
|
||||
* **Migrate legacy backups** — import old `battlenet_authenticator_*.json` files (plaintext
|
||||
or encrypted with the older PBKDF2 scheme) into the vault.
|
||||
* **Optional GUI** — a small Flet desktop app (`bnet-auth-gui`) with live TOTP codes over
|
||||
the same vault, for users who'd rather not use the terminal.
|
||||
|
||||
## Security model
|
||||
|
||||
| Aspect | Detail |
|
||||
| --- | --- |
|
||||
| Cipher | AES‑256‑GCM (authenticated encryption) |
|
||||
| KDF (new files) | **scrypt** (memory-hard), parameters in `settings.yaml` |
|
||||
| KDF (legacy files) | PBKDF2‑HMAC‑SHA256 (100k and 600k) — still decryptable |
|
||||
| Storage | single encrypted vault in the per-user data dir |
|
||||
| File permissions | vault and QR PNGs written `0600` (owner-only) on POSIX |
|
||||
| Writes | atomic (temp file + replace) so a crash can't truncate the vault |
|
||||
|
||||
> Your vault passphrase is the **only** way to decrypt your secrets. There is no
|
||||
> recovery if you lose it. QR-code PNGs contain the raw secret — delete them after import.
|
||||
|
||||
## Install
|
||||
|
||||
### Pre-built Windows executables
|
||||
|
||||
Grab `bnet-auth.exe` (CLI) and `bnet-auth-gui.exe` (GUI) from the
|
||||
[Releases page](https://github.com/Nighthawk42/bnet_auth_tool/releases) — no Python needed.
|
||||
Both share the same encrypted vault.
|
||||
|
||||
### With [uv](https://docs.astral.sh/uv/) (recommended)
|
||||
|
||||
```bash
|
||||
uv tool install . # install the `bnet-auth` command
|
||||
# or, for development:
|
||||
uv sync --extra dev
|
||||
```
|
||||
|
||||
### With pip
|
||||
|
||||
```bash
|
||||
pip install .
|
||||
# or, using the fallback dependency list:
|
||||
pip install -r requirements.txt && pip install .
|
||||
```
|
||||
|
||||
Requires **Python 3.9+**.
|
||||
|
||||
### GUI (optional)
|
||||
|
||||
Prefer a window over a terminal? Install the optional Flet GUI:
|
||||
|
||||
```bash
|
||||
uv tool install ".[gui]" # or: pip install ".[gui]"
|
||||
bnet-auth-gui # or: python -m bnet_auth_tool.gui
|
||||
```
|
||||
|
||||
The GUI uses the **same** encrypted vault, settings, and crypto as the CLI — unlock the
|
||||
vault, see live rotating TOTP codes, copy a code, view a QR, import legacy backups, and
|
||||
(online, unverified) attach/retrieve. It's purely an alternative front-end.
|
||||
|
||||
## Usage (CLI)
|
||||
|
||||
Run with no arguments for the interactive menu:
|
||||
|
||||
```bash
|
||||
bnet-auth
|
||||
```
|
||||
|
||||
Or use scriptable subcommands:
|
||||
|
||||
```bash
|
||||
bnet-auth list # list authenticators in the vault
|
||||
bnet-auth reconstruct US-1234-... # print TOTP details + optional QR
|
||||
bnet-auth migrate --dir . # import legacy JSON backups from a folder
|
||||
bnet-auth paths # show config / data / vault locations
|
||||
bnet-auth attach # online (unverified)
|
||||
bnet-auth retrieve # online (unverified)
|
||||
```
|
||||
|
||||
### Migrating from older versions
|
||||
|
||||
Older releases dropped one `battlenet_authenticator_<serial>.json` per authenticator into
|
||||
the working directory. To pull them into the encrypted vault:
|
||||
|
||||
```bash
|
||||
cd /folder/with/old/json/files
|
||||
bnet-auth migrate --dir .
|
||||
```
|
||||
|
||||
You'll be prompted for the vault passphrase (creating it on first run) and for each
|
||||
encrypted legacy file's passphrase. After verifying the vault, **securely delete the old
|
||||
plaintext files**.
|
||||
|
||||
## Configuration
|
||||
|
||||
A user-editable `settings.yaml` is created on first run in your config directory
|
||||
(`bnet-auth paths` shows where). Edit it to change API endpoints, regions, KDF
|
||||
parameters, or TOTP output — handy if Blizzard moves an endpoint again. Any key you omit
|
||||
falls back to the bundled default.
|
||||
|
||||
## Development
|
||||
|
||||
```bash
|
||||
uv sync --extra dev
|
||||
uv run pytest # tests
|
||||
uv run ruff check . # lint
|
||||
```
|
||||
|
||||
See [`CLAUDE.md`](CLAUDE.md) / [`AGENTS.md`](AGENTS.md) for the architecture overview.
|
||||
|
||||
### Building the executables
|
||||
|
||||
```bash
|
||||
uv pip install -e ".[build]"
|
||||
python packaging/build_exes.py # both exes -> dist/, zip -> release/
|
||||
python packaging/build_exes.py --skip-gui # CLI only
|
||||
```
|
||||
|
||||
The CLI is frozen with PyInstaller and the GUI with `flet pack` (PyInstaller plus the
|
||||
bundled Flet desktop runtime). Build outputs (`dist/`, `build/`, `release/`, `*.exe`,
|
||||
`*.zip`, `*.spec`) are gitignored.
|
||||
|
||||
## Account recovery
|
||||
|
||||
This project has **zero** association with Blizzard and no access to their backend. If you
|
||||
are locked out of your account, contact
|
||||
[Blizzard Customer Support](https://us.battle.net/support/en/) — the maintainer cannot
|
||||
recover accounts.
|
||||
|
||||
## License
|
||||
|
||||
[MIT](LICENSE) © 2024-2026 Nighthawk42
|
||||
|
||||
## Donations
|
||||
|
||||
[](https://ko-fi.com/P5P21QRW51)
|
||||
|
||||
-1109
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,172 @@
|
||||
"""Build standalone Windows executables for the CLI and GUI.
|
||||
|
||||
* CLI -> PyInstaller (one-file, console).
|
||||
* GUI -> ``flet pack`` (PyInstaller under the hood, but with the Flet desktop
|
||||
runtime bundled — plain PyInstaller cannot ship a working Flet app).
|
||||
|
||||
Both bundle the package's ``settings.yaml`` as data so the frozen binaries can
|
||||
read their default configuration via ``importlib.resources``.
|
||||
|
||||
Usage (from the repo root, in an env with the ``build`` extra installed)::
|
||||
|
||||
uv pip install -e ".[build]"
|
||||
python packaging/build_exes.py # build both + zip
|
||||
python packaging/build_exes.py --skip-gui # CLI only
|
||||
python packaging/build_exes.py --no-zip # don't produce the release zip
|
||||
|
||||
Artifacts land in ``dist/`` and the zip in ``release/``. All of that is
|
||||
gitignored.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
PKG_DIR = REPO_ROOT / "src" / "bnet_auth_tool"
|
||||
SETTINGS = PKG_DIR / "settings.yaml"
|
||||
DIST = REPO_ROOT / "dist"
|
||||
# `flet pack` wipes its --distpath, so give the GUI its own and copy the result
|
||||
# into dist/ afterwards. That way CLI and GUI builds never clobber each other.
|
||||
GUI_DIST = REPO_ROOT / "build" / "gui-dist"
|
||||
RELEASE = REPO_ROOT / "release"
|
||||
ENTRY_CLI = REPO_ROOT / "packaging" / "pyi_entry_cli.py"
|
||||
ENTRY_GUI = REPO_ROOT / "packaging" / "pyi_entry_gui.py"
|
||||
|
||||
CLI_NAME = "bnet-auth"
|
||||
GUI_NAME = "bnet-auth-gui"
|
||||
DATA_SEP = ";" if os.name == "nt" else ":"
|
||||
EXE_SUFFIX = ".exe" if os.name == "nt" else ""
|
||||
|
||||
|
||||
def _version() -> str:
|
||||
sys.path.insert(0, str(REPO_ROOT / "src"))
|
||||
from bnet_auth_tool import __version__
|
||||
|
||||
return __version__
|
||||
|
||||
|
||||
def _release_label(version: str) -> str:
|
||||
# PEP 440 "2.0.0a0" -> human "2.0.0-alpha" for tags/zip names.
|
||||
return version.replace("a0", "-alpha").replace("b0", "-beta").replace("rc0", "-rc")
|
||||
|
||||
|
||||
def _run(cmd: list[str]) -> None:
|
||||
print(f"\n$ {' '.join(cmd)}\n", flush=True)
|
||||
subprocess.run(cmd, check=True, cwd=REPO_ROOT)
|
||||
|
||||
|
||||
def build_cli() -> Path:
|
||||
_run(
|
||||
[
|
||||
sys.executable,
|
||||
"-m",
|
||||
"PyInstaller",
|
||||
"--onefile",
|
||||
"--console",
|
||||
"--clean",
|
||||
"--noconfirm",
|
||||
"--name",
|
||||
CLI_NAME,
|
||||
"--add-data",
|
||||
f"{SETTINGS}{DATA_SEP}bnet_auth_tool",
|
||||
str(ENTRY_CLI),
|
||||
]
|
||||
)
|
||||
return DIST / f"{CLI_NAME}{EXE_SUFFIX}"
|
||||
|
||||
|
||||
def build_gui() -> Path:
|
||||
# `flet pack` wraps PyInstaller and bundles the Flet desktop runtime.
|
||||
# Its --add-data uses "source:destination"; pass a *relative* source so the
|
||||
# drive-letter colon in an absolute Windows path doesn't confuse the parser.
|
||||
settings_rel = SETTINGS.relative_to(REPO_ROOT).as_posix()
|
||||
_run(
|
||||
[
|
||||
sys.executable,
|
||||
"-m",
|
||||
"flet.cli",
|
||||
"pack",
|
||||
"-y", # don't prompt to delete the build dir
|
||||
str(ENTRY_GUI),
|
||||
"--name",
|
||||
GUI_NAME,
|
||||
"--distpath",
|
||||
str(GUI_DIST),
|
||||
"--add-data",
|
||||
f"{settings_rel}:bnet_auth_tool",
|
||||
]
|
||||
)
|
||||
# Copy the GUI exe into the shared dist/ so it sits next to the CLI exe.
|
||||
built = GUI_DIST / f"{GUI_NAME}{EXE_SUFFIX}"
|
||||
DIST.mkdir(exist_ok=True)
|
||||
dest = DIST / f"{GUI_NAME}{EXE_SUFFIX}"
|
||||
if built.exists():
|
||||
shutil.copy2(built, dest)
|
||||
return dest
|
||||
|
||||
|
||||
def make_zip(exes: list[Path], version: str) -> Path:
|
||||
RELEASE.mkdir(exist_ok=True)
|
||||
label = _release_label(version)
|
||||
plat = "windows-x64" if os.name == "nt" else os.name
|
||||
zip_path = RELEASE / f"bnet-auth-tool-{label}-{plat}.zip"
|
||||
|
||||
readme = (
|
||||
f"Battle.net Authenticator Tool {label}\n"
|
||||
f"{'=' * 40}\n\n"
|
||||
"bnet-auth.exe - command-line interface (run in a terminal)\n"
|
||||
"bnet-auth-gui.exe - desktop GUI (double-click)\n\n"
|
||||
"Both share one encrypted vault in your user data dir. ALPHA build:\n"
|
||||
"online attach/retrieve is unverified against Blizzard's API; the\n"
|
||||
"offline vault/TOTP/migration features are the supported core.\n\n"
|
||||
"Source & docs: https://github.com/Nighthawk42/bnet_auth_tool\n"
|
||||
)
|
||||
|
||||
with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
|
||||
for exe in exes:
|
||||
zf.write(exe, exe.name)
|
||||
zf.writestr("README.txt", readme)
|
||||
return zip_path
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description="Build CLI/GUI executables.")
|
||||
parser.add_argument("--skip-gui", action="store_true", help="build the CLI only")
|
||||
parser.add_argument("--skip-cli", action="store_true", help="build the GUI only")
|
||||
parser.add_argument("--no-zip", action="store_true", help="skip the release zip")
|
||||
args = parser.parse_args()
|
||||
|
||||
version = _version()
|
||||
print(f"Building bnet-auth-tool {version} ({_release_label(version)})")
|
||||
|
||||
exes: list[Path] = []
|
||||
if not args.skip_cli:
|
||||
exes.append(build_cli())
|
||||
if not args.skip_gui:
|
||||
exes.append(build_gui())
|
||||
|
||||
missing = [str(p) for p in exes if not p.exists()]
|
||||
if missing:
|
||||
print(f"\nERROR: expected artifacts not found: {missing}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
print("\nBuilt:")
|
||||
for exe in exes:
|
||||
print(f" {exe} ({exe.stat().st_size / 1_000_000:.1f} MB)")
|
||||
|
||||
if not args.no_zip and exes:
|
||||
zip_path = make_zip(exes, version)
|
||||
print(f"\nPackaged: {zip_path}")
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,12 @@
|
||||
"""PyInstaller entry point for the CLI.
|
||||
|
||||
A standalone script (not a module) so PyInstaller has a concrete file to freeze;
|
||||
it simply imports and runs the installed package's CLI entry point.
|
||||
"""
|
||||
|
||||
import sys
|
||||
|
||||
from bnet_auth_tool.cli import main
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,10 @@
|
||||
"""PyInstaller / flet-pack entry point for the GUI.
|
||||
|
||||
A standalone script so the packager has a concrete file to freeze; it imports
|
||||
and runs the installed package's GUI entry point.
|
||||
"""
|
||||
|
||||
from bnet_auth_tool.gui import main
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,69 @@
|
||||
[build-system]
|
||||
requires = ["hatchling"]
|
||||
build-backend = "hatchling.build"
|
||||
|
||||
[project]
|
||||
name = "bnet-auth-tool"
|
||||
version = "2.0.0a0"
|
||||
description = "Manage Battle.net software authenticators (CLI + optional GUI): attach/retrieve secrets and back up TOTP keys in an encrypted local vault."
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.9"
|
||||
license = { text = "MIT" }
|
||||
authors = [{ name = "Nighthawk42" }]
|
||||
keywords = ["battle.net", "blizzard", "authenticator", "totp", "2fa", "mfa", "otp"]
|
||||
classifiers = [
|
||||
"Development Status :: 4 - Beta",
|
||||
"Environment :: Console",
|
||||
"Intended Audience :: End Users/Desktop",
|
||||
"License :: OSI Approved :: MIT License",
|
||||
"Operating System :: OS Independent",
|
||||
"Programming Language :: Python :: 3",
|
||||
"Topic :: Security :: Cryptography",
|
||||
"Topic :: Utilities",
|
||||
]
|
||||
|
||||
dependencies = [
|
||||
"requests>=2.31",
|
||||
"qrcode[pil]>=7.4",
|
||||
"cryptography>=42.0",
|
||||
"PyYAML>=6.0",
|
||||
"platformdirs>=4.0",
|
||||
]
|
||||
|
||||
[project.optional-dependencies]
|
||||
gui = [
|
||||
"flet>=0.28",
|
||||
]
|
||||
dev = [
|
||||
"pytest>=8.0",
|
||||
"ruff>=0.5",
|
||||
]
|
||||
build = [
|
||||
"flet[all]>=0.28",
|
||||
"pyinstaller>=6.0",
|
||||
]
|
||||
|
||||
[project.urls]
|
||||
Homepage = "https://github.com/Nighthawk42/bnet_auth_tool"
|
||||
Repository = "https://github.com/Nighthawk42/bnet_auth_tool"
|
||||
Issues = "https://github.com/Nighthawk42/bnet_auth_tool/issues"
|
||||
|
||||
[project.scripts]
|
||||
bnet-auth = "bnet_auth_tool.cli:main"
|
||||
bnet-auth-gui = "bnet_auth_tool.gui:main"
|
||||
|
||||
[tool.hatch.build.targets.wheel]
|
||||
packages = ["src/bnet_auth_tool"]
|
||||
|
||||
[tool.ruff]
|
||||
line-length = 100
|
||||
target-version = "py39"
|
||||
src = ["src", "tests"]
|
||||
|
||||
[tool.ruff.lint]
|
||||
select = ["E", "F", "I", "UP", "B", "SIM", "C4"]
|
||||
ignore = ["E501"]
|
||||
|
||||
[tool.pytest.ini_options]
|
||||
testpaths = ["tests"]
|
||||
addopts = "-ra"
|
||||
+22
-4
@@ -1,4 +1,22 @@
|
||||
cryptography
|
||||
qrcode
|
||||
Pillow
|
||||
requests
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fallback dependency list for `pip install -r requirements.txt`.
|
||||
#
|
||||
# The canonical source of truth is pyproject.toml; this file mirrors its
|
||||
# runtime dependencies for environments without uv/PEP 621 support.
|
||||
# For development, prefer: uv sync --extra dev (or) pip install -e .[dev]
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# HTTP client for the (online) Battle.net authenticator API
|
||||
requests>=2.31
|
||||
|
||||
# QR-code generation (Pillow is pulled in by the [pil] extra for PNG output)
|
||||
qrcode[pil]>=7.4
|
||||
|
||||
# AES-256-GCM encryption + scrypt/PBKDF2 key derivation for the local vault
|
||||
cryptography>=42.0
|
||||
|
||||
# YAML parsing for the user-editable settings file (endpoints, KDF, TOTP)
|
||||
PyYAML>=6.0
|
||||
|
||||
# Cross-platform resolution of the per-user data/config directories
|
||||
platformdirs>=4.0
|
||||
|
||||
@@ -0,0 +1,11 @@
|
||||
"""Battle.net software authenticator tool.
|
||||
|
||||
Attach/retrieve Battle.net authenticator secrets (online) and manage TOTP
|
||||
backups in an encrypted local vault (offline). See :mod:`bnet_auth_tool.cli`.
|
||||
"""
|
||||
|
||||
__version__ = "2.0.0a0"
|
||||
__author__ = "Nighthawk42"
|
||||
__license__ = "MIT"
|
||||
|
||||
__all__ = ["__version__", "__author__", "__license__"]
|
||||
@@ -0,0 +1,10 @@
|
||||
"""Enable ``python -m bnet_auth_tool``."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
|
||||
from .cli import main
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,157 @@
|
||||
"""Online Battle.net authenticator client.
|
||||
|
||||
.. warning::
|
||||
These flows depend on Blizzard's identity API, which has changed before and
|
||||
may be blocked again. They are **unverified** against the live backend.
|
||||
Endpoints are configurable in ``settings.yaml`` so they can be re-mapped
|
||||
without code changes. Offline vault/TOTP features do not use this module.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from typing import Any
|
||||
|
||||
import requests
|
||||
|
||||
from . import __version__
|
||||
from .config import ApiConfig
|
||||
from .errors import AuthenticatorError
|
||||
|
||||
# Server error bodies can echo back submitted material; cap what we surface.
|
||||
_MAX_ERROR_BODY = 200
|
||||
|
||||
|
||||
class BattleNetAuthenticator:
|
||||
"""Thin client for attach / retrieve authenticator operations."""
|
||||
|
||||
def __init__(self, api: ApiConfig):
|
||||
self._api = api
|
||||
self._session = requests.Session()
|
||||
self._session.headers.update(
|
||||
{"User-Agent": f"bnet-auth-tool/{__version__}"}
|
||||
)
|
||||
|
||||
# -- internals ---------------------------------------------------------- #
|
||||
def _request(
|
||||
self,
|
||||
method: str,
|
||||
url: str,
|
||||
*,
|
||||
headers: dict[str, str] | None = None,
|
||||
data: Any | None = None,
|
||||
json_payload: dict[str, Any] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
try:
|
||||
response = self._session.request(
|
||||
method,
|
||||
url,
|
||||
headers=headers,
|
||||
data=data,
|
||||
json=json_payload,
|
||||
timeout=self._api.timeout,
|
||||
)
|
||||
response.raise_for_status()
|
||||
|
||||
if response.status_code == 204 or not response.content:
|
||||
return {}
|
||||
content_type = response.headers.get("Content-Type", "")
|
||||
if "application/json" not in content_type:
|
||||
raise AuthenticatorError(
|
||||
f"Unexpected content type '{content_type}' from {url}."
|
||||
)
|
||||
return response.json()
|
||||
|
||||
except requests.exceptions.HTTPError as exc:
|
||||
raise self._http_error(exc, url) from exc
|
||||
except requests.exceptions.RequestException as exc:
|
||||
raise AuthenticatorError(f"Request to {url} failed: {exc}") from exc
|
||||
except json.JSONDecodeError as exc:
|
||||
raise AuthenticatorError(f"Could not decode JSON from {url}: {exc}") from exc
|
||||
|
||||
@staticmethod
|
||||
def _http_error(exc: requests.exceptions.HTTPError, url: str) -> AuthenticatorError:
|
||||
status = exc.response.status_code
|
||||
|
||||
# Blizzard MFA errors come back as {errorCode, message}; surface those
|
||||
# rather than the raw body (which may contain submitted secrets).
|
||||
blz_detail = ""
|
||||
try:
|
||||
payload = exc.response.json()
|
||||
if isinstance(payload, dict) and (payload.get("errorCode") or payload.get("message")):
|
||||
blz_detail = (
|
||||
f" [errorCode={payload.get('errorCode')} "
|
||||
f"message={payload.get('message')}]"
|
||||
)
|
||||
except (ValueError, json.JSONDecodeError):
|
||||
# Only include a short, generic snippet of an opaque body.
|
||||
snippet = (exc.response.text or "").strip().replace("\n", " ")[:_MAX_ERROR_BODY]
|
||||
if snippet:
|
||||
blz_detail = f" (body: {snippet})"
|
||||
|
||||
hint = ""
|
||||
if status == 404 and not blz_detail:
|
||||
hint = (
|
||||
" Hint: route not found — the endpoint version may have changed. "
|
||||
"Update api.* in settings.yaml."
|
||||
)
|
||||
elif status in (401, 403):
|
||||
hint = " Hint: authorization failed (token/scope rejected)."
|
||||
|
||||
return AuthenticatorError(f"HTTP {status} from {url}.{blz_detail}{hint}")
|
||||
|
||||
# -- public flows ------------------------------------------------------- #
|
||||
def get_bearer_token(self, session_token: str) -> None:
|
||||
payload = {
|
||||
"client_id": self._api.client_id,
|
||||
"grant_type": "client_sso",
|
||||
"scope": "auth.authenticator",
|
||||
"token": session_token,
|
||||
}
|
||||
headers = {"content-type": "application/x-www-form-urlencoded; charset=utf-8"}
|
||||
response = self._request("POST", self._api.sso_url, headers=headers, data=payload)
|
||||
|
||||
access_token = response.get("access_token")
|
||||
if not access_token:
|
||||
raise AuthenticatorError("Bearer token not found in SSO response.")
|
||||
self._session.headers["Authorization"] = f"Bearer {access_token}"
|
||||
|
||||
def attach_authenticator(self) -> dict[str, Any]:
|
||||
if "Authorization" not in self._session.headers:
|
||||
raise AuthenticatorError("Bearer token not set; call get_bearer_token first.")
|
||||
|
||||
response = self._request(
|
||||
"POST", self._api.attach_url, headers={"accept": "application/json"}
|
||||
)
|
||||
self._reject_healup(response)
|
||||
|
||||
missing = [k for k in ("serial", "restoreCode", "deviceSecret") if k not in response]
|
||||
if missing:
|
||||
raise AuthenticatorError(f"API response missing keys: {missing}")
|
||||
return response
|
||||
|
||||
def retrieve_device_secret(
|
||||
self, account_identifier: str, serial: str, restore_code: str
|
||||
) -> dict[str, Any]:
|
||||
# v2/device authenticates via accountIdentifier + serial + restoreCode
|
||||
# (no bearer token required), mirroring restoreAuthenticator.
|
||||
payload = {
|
||||
"accountIdentifier": account_identifier.strip(),
|
||||
"serial": serial.strip(),
|
||||
"restoreCode": restore_code.strip(),
|
||||
}
|
||||
response = self._request("POST", self._api.device_url, json_payload=payload)
|
||||
self._reject_healup(response)
|
||||
|
||||
if "deviceSecret" not in response:
|
||||
raise AuthenticatorError("API response missing 'deviceSecret'.")
|
||||
return response
|
||||
|
||||
@staticmethod
|
||||
def _reject_healup(response: dict[str, Any]) -> None:
|
||||
if response.get("requireHealup"):
|
||||
raise AuthenticatorError(
|
||||
"Server returned requireHealup=true: the account requires a 'heal up' "
|
||||
"step before credentials can be issued. The official app handles this "
|
||||
"flow; this tool cannot."
|
||||
)
|
||||
@@ -0,0 +1,415 @@
|
||||
"""Command-line interface: interactive menu plus scriptable subcommands.
|
||||
|
||||
Run without arguments for the interactive menu, or use a subcommand, e.g.::
|
||||
|
||||
bnet-auth list
|
||||
bnet-auth migrate --dir .
|
||||
bnet-auth reconstruct US-1234-...
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import getpass
|
||||
import sys
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
from . import __author__, __license__, __version__
|
||||
from .api import BattleNetAuthenticator
|
||||
from .config import (
|
||||
Settings,
|
||||
config_dir,
|
||||
data_dir,
|
||||
ensure_user_settings,
|
||||
load_settings,
|
||||
user_settings_path,
|
||||
vault_path,
|
||||
)
|
||||
from .crypto import EncryptionManager
|
||||
from .errors import BnetAuthError
|
||||
from .migrate import discover_legacy_files, migrate_files
|
||||
from .storage import Vault
|
||||
from .totp import build_totp_url, generate_qr_code, hex_secret_to_base32
|
||||
|
||||
TITLE = "Battle.net Authenticator Tool"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Small I/O helpers
|
||||
# --------------------------------------------------------------------------- #
|
||||
def _print_header() -> None:
|
||||
print(f"\n=== {TITLE} ===")
|
||||
print(f"Version {__version__} · Author {__author__} · License {__license__}")
|
||||
print(f"Vault: {vault_path()}")
|
||||
print("-" * 60)
|
||||
|
||||
|
||||
def _prompt(text: str) -> str | None:
|
||||
try:
|
||||
return input(text).strip()
|
||||
except (EOFError, KeyboardInterrupt):
|
||||
print("\nCancelled.")
|
||||
return None
|
||||
|
||||
|
||||
def _confirm(text: str) -> bool:
|
||||
answer = _prompt(f"{text} (y/n): ")
|
||||
return (answer or "").lower() == "y"
|
||||
|
||||
|
||||
def _prompt_passphrase(message: str, *, confirm: bool) -> str | None:
|
||||
try:
|
||||
while True:
|
||||
passphrase = getpass.getpass(message)
|
||||
if not passphrase:
|
||||
print("Passphrase cannot be empty.")
|
||||
continue
|
||||
if not confirm:
|
||||
return passphrase
|
||||
if passphrase == getpass.getpass("Confirm passphrase: "):
|
||||
return passphrase
|
||||
print("Passphrases do not match. Try again.")
|
||||
except (EOFError, KeyboardInterrupt):
|
||||
print("\nCancelled.")
|
||||
return None
|
||||
|
||||
|
||||
def _open_vault(settings: Settings, *, for_write: bool) -> Vault | None:
|
||||
"""Open the vault, prompting for the master passphrase.
|
||||
|
||||
Creating a new vault requires passphrase confirmation; opening an existing
|
||||
one does not.
|
||||
"""
|
||||
exists = vault_path().is_file()
|
||||
if not exists and not for_write:
|
||||
print(f"No vault yet at {vault_path()}.")
|
||||
return None
|
||||
|
||||
if not exists:
|
||||
print("No vault exists. Creating a new encrypted vault.")
|
||||
passphrase = _prompt_passphrase("Vault passphrase: ", confirm=not exists)
|
||||
if passphrase is None:
|
||||
return None
|
||||
|
||||
manager = EncryptionManager(passphrase, settings.crypto)
|
||||
try:
|
||||
return Vault(manager).load()
|
||||
except BnetAuthError as exc:
|
||||
print(f"Error opening vault: {exc}", file=sys.stderr)
|
||||
return None
|
||||
|
||||
|
||||
def _store_entry(vault: Vault, device_info: dict, settings: Settings) -> dict:
|
||||
"""Build a vault entry from raw device info, store it, and return it."""
|
||||
serial = device_info["serial"]
|
||||
base32_secret = hex_secret_to_base32(device_info["deviceSecret"])
|
||||
totp_url = build_totp_url(serial, base32_secret, settings.totp)
|
||||
entry = {
|
||||
"serial": serial,
|
||||
"restoreCode": device_info.get("restoreCode"),
|
||||
"deviceSecret": device_info["deviceSecret"],
|
||||
"base32Secret": base32_secret,
|
||||
"totpUrl": totp_url,
|
||||
"addedAt": datetime.now(timezone.utc).isoformat(timespec="seconds"),
|
||||
}
|
||||
vault.add(entry, overwrite=True)
|
||||
vault.save()
|
||||
return entry
|
||||
|
||||
|
||||
def _print_entry(entry: dict) -> None:
|
||||
print("\n" + "-" * 40)
|
||||
print(f" Serial: {entry.get('serial')}")
|
||||
print(f" Restore Code: {entry.get('restoreCode')}")
|
||||
print(f" Base32: {entry.get('base32Secret')}")
|
||||
print(f" otpauth URL: {entry.get('totpUrl')}")
|
||||
print(" TOTP params: SHA1, 8 digits, 30s period")
|
||||
print("-" * 40)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Online flows (UNVERIFIED against live Blizzard backend)
|
||||
# --------------------------------------------------------------------------- #
|
||||
_SESSION_TOKEN_HELP = """
|
||||
--- How to get a Session Token ---
|
||||
1. In a private browser window go to:
|
||||
https://account.battle.net/login/en/?ref=localhost
|
||||
2. Log in; you'll land on a 'Page Not Found' at localhost.
|
||||
3. From the URL, copy the token that looks like 'ST=US-...' (the value after ST=).
|
||||
----------------------------------"""
|
||||
|
||||
|
||||
def _get_session_token(settings: Settings) -> str | None:
|
||||
print(_SESSION_TOKEN_HELP)
|
||||
token = _prompt("Enter your Session Token (or blank to cancel): ")
|
||||
if not token:
|
||||
return None
|
||||
prefixes = settings.api.region_prefixes
|
||||
if prefixes and (not any(token.startswith(p) for p in prefixes) or len(token) < 20):
|
||||
print("Warning: token format looks unusual; ensure you copied the full value.")
|
||||
return token
|
||||
|
||||
|
||||
def action_attach(settings: Settings) -> None:
|
||||
print("\nNOTE: online flows are unverified against the current Blizzard API.")
|
||||
token = _get_session_token(settings)
|
||||
if not token:
|
||||
return
|
||||
vault = _open_vault(settings, for_write=True)
|
||||
if vault is None:
|
||||
return
|
||||
client = BattleNetAuthenticator(settings.api)
|
||||
try:
|
||||
print("Requesting bearer token...")
|
||||
client.get_bearer_token(token)
|
||||
print("Attaching a new authenticator...")
|
||||
device_info = client.attach_authenticator()
|
||||
except BnetAuthError as exc:
|
||||
print(f"\nAttach failed: {exc}", file=sys.stderr)
|
||||
return
|
||||
entry = _store_entry(vault, device_info, settings)
|
||||
print("Authenticator attached and saved to the vault.")
|
||||
_print_entry(entry)
|
||||
_maybe_export_qr(entry)
|
||||
|
||||
|
||||
def action_retrieve(settings: Settings) -> None:
|
||||
print("\nNOTE: online flows are unverified against the current Blizzard API.")
|
||||
account = _prompt("Account email or phone number: ")
|
||||
serial = _prompt("Authenticator serial: ")
|
||||
restore_code = _prompt("Authenticator restore code: ")
|
||||
if not (account and serial and restore_code):
|
||||
print("Account identifier, serial, and restore code are all required.")
|
||||
return
|
||||
vault = _open_vault(settings, for_write=True)
|
||||
if vault is None:
|
||||
return
|
||||
client = BattleNetAuthenticator(settings.api)
|
||||
try:
|
||||
print(f"Retrieving secret for serial {serial}...")
|
||||
retrieved = client.retrieve_device_secret(account, serial, restore_code)
|
||||
except BnetAuthError as exc:
|
||||
print(f"\nRetrieve failed: {exc}", file=sys.stderr)
|
||||
return
|
||||
device_info = {
|
||||
"serial": serial,
|
||||
"restoreCode": restore_code,
|
||||
"deviceSecret": retrieved["deviceSecret"],
|
||||
}
|
||||
entry = _store_entry(vault, device_info, settings)
|
||||
print("Device secret retrieved and saved to the vault.")
|
||||
_print_entry(entry)
|
||||
_maybe_export_qr(entry)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Offline flows
|
||||
# --------------------------------------------------------------------------- #
|
||||
def action_list(settings: Settings) -> None:
|
||||
vault = _open_vault(settings, for_write=False)
|
||||
if vault is None:
|
||||
return
|
||||
if len(vault) == 0:
|
||||
print("Vault is empty.")
|
||||
return
|
||||
print(f"\n{len(vault)} authenticator(s) in the vault:")
|
||||
for entry in vault.list():
|
||||
print(f" - {entry.get('serial')} (added {entry.get('addedAt', 'unknown')})")
|
||||
|
||||
|
||||
def action_reconstruct(settings: Settings, serial: str | None = None) -> None:
|
||||
vault = _open_vault(settings, for_write=False)
|
||||
if vault is None:
|
||||
return
|
||||
serials = vault.serials()
|
||||
if not serials:
|
||||
print("Vault is empty.")
|
||||
return
|
||||
if serial is None:
|
||||
serial = _choose(serials, "Select an authenticator")
|
||||
if serial is None:
|
||||
return
|
||||
entry = vault.get(serial)
|
||||
if entry is None:
|
||||
print(f"No entry for serial '{serial}'.")
|
||||
return
|
||||
if not entry.get("totpUrl"):
|
||||
base32 = entry.get("base32Secret") or hex_secret_to_base32(entry.get("deviceSecret", ""))
|
||||
entry["totpUrl"] = build_totp_url(serial, base32, settings.totp)
|
||||
_print_entry(entry)
|
||||
_maybe_export_qr(entry)
|
||||
|
||||
|
||||
def action_migrate(settings: Settings, directory: Path, overwrite: bool) -> None:
|
||||
files = discover_legacy_files(directory)
|
||||
if not files:
|
||||
print(f"No legacy authenticator JSON files found in {directory}.")
|
||||
return
|
||||
print(f"\nFound {len(files)} legacy file(s) in {directory}:")
|
||||
for f in files:
|
||||
print(f" - {f.name}")
|
||||
vault = _open_vault(settings, for_write=True)
|
||||
if vault is None:
|
||||
return
|
||||
|
||||
def provider(path: Path) -> str | None:
|
||||
print(f"\n'{path.name}' is encrypted.")
|
||||
return _prompt_passphrase(f"Passphrase for '{path.name}': ", confirm=False)
|
||||
|
||||
outcomes = migrate_files(files, vault, settings, provider, overwrite=overwrite)
|
||||
vault.save()
|
||||
|
||||
print("\nMigration summary:")
|
||||
for o in outcomes:
|
||||
suffix = f" ({o.detail})" if o.detail else ""
|
||||
print(f" [{o.status}] {o.path.name} -> {o.serial or '?'}{suffix}")
|
||||
imported = sum(1 for o in outcomes if o.status in ("imported", "replaced"))
|
||||
print(f"\n{imported} entr(y/ies) now in the vault.")
|
||||
if imported:
|
||||
print("Securely delete the original plaintext files once you've verified the vault.")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Shared helpers
|
||||
# --------------------------------------------------------------------------- #
|
||||
def _choose(options: list[str], prompt: str) -> str | None:
|
||||
for i, opt in enumerate(options, 1):
|
||||
print(f" {i}. {opt}")
|
||||
raw = _prompt(f"{prompt} (number, or blank to cancel): ")
|
||||
if not raw:
|
||||
return None
|
||||
try:
|
||||
idx = int(raw) - 1
|
||||
except ValueError:
|
||||
print("Invalid selection.")
|
||||
return None
|
||||
if 0 <= idx < len(options):
|
||||
return options[idx]
|
||||
print("Invalid selection.")
|
||||
return None
|
||||
|
||||
|
||||
def _maybe_export_qr(entry: dict) -> None:
|
||||
if not entry.get("totpUrl"):
|
||||
return
|
||||
if not _confirm("\nGenerate a QR-code PNG (contains your secret)?"):
|
||||
return
|
||||
out = Path.cwd() / f"bnet_{entry['serial']}.png"
|
||||
try:
|
||||
path = generate_qr_code(entry["totpUrl"], out)
|
||||
except Exception as exc: # noqa: BLE001 - qrcode/Pillow failure shouldn't crash CLI
|
||||
print(f"Could not generate QR code: {exc}", file=sys.stderr)
|
||||
return
|
||||
print(f"QR code written to {path}")
|
||||
print("WARNING: this PNG contains your TOTP secret. Delete it after importing.")
|
||||
|
||||
|
||||
def action_paths(settings: Settings) -> None:
|
||||
print(f"\nConfig dir: {config_dir()}")
|
||||
print(f"Settings file: {user_settings_path()}")
|
||||
print(f"Data dir: {data_dir()}")
|
||||
print(f"Vault file: {vault_path()}")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Interactive menu
|
||||
# --------------------------------------------------------------------------- #
|
||||
_MENU = [
|
||||
("Attach a new authenticator (online)", lambda s: action_attach(s)),
|
||||
("Retrieve an existing device secret (online)", lambda s: action_retrieve(s)),
|
||||
("Reconstruct TOTP / QR from the vault", lambda s: action_reconstruct(s)),
|
||||
("List authenticators in the vault", lambda s: action_list(s)),
|
||||
("Migrate legacy JSON files into the vault", lambda s: action_migrate(s, Path.cwd(), False)),
|
||||
("Show file paths", lambda s: action_paths(s)),
|
||||
]
|
||||
|
||||
|
||||
def interactive(settings: Settings) -> None:
|
||||
_print_header()
|
||||
while True:
|
||||
print("\nChoose an action:")
|
||||
for i, (label, _) in enumerate(_MENU, 1):
|
||||
print(f" {i}. {label}")
|
||||
print(f" {len(_MENU) + 1}. Exit")
|
||||
|
||||
choice = _prompt("Enter choice: ")
|
||||
if choice is None:
|
||||
break
|
||||
if choice == str(len(_MENU) + 1) or choice.lower() in ("exit", "q"):
|
||||
break
|
||||
try:
|
||||
idx = int(choice) - 1
|
||||
except ValueError:
|
||||
print("Invalid choice.")
|
||||
continue
|
||||
if 0 <= idx < len(_MENU):
|
||||
try:
|
||||
_MENU[idx][1](settings)
|
||||
except BnetAuthError as exc:
|
||||
print(f"\nError: {exc}", file=sys.stderr)
|
||||
else:
|
||||
print("Invalid choice.")
|
||||
print("\nExiting. Keep your vault and passphrase backed up securely.")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Entry point
|
||||
# --------------------------------------------------------------------------- #
|
||||
def _build_parser() -> argparse.ArgumentParser:
|
||||
parser = argparse.ArgumentParser(prog="bnet-auth", description=TITLE)
|
||||
parser.add_argument("--version", action="version", version=f"%(prog)s {__version__}")
|
||||
sub = parser.add_subparsers(dest="command")
|
||||
|
||||
sub.add_parser("attach", help="attach a new authenticator (online, unverified)")
|
||||
sub.add_parser("retrieve", help="retrieve an existing device secret (online, unverified)")
|
||||
sub.add_parser("list", help="list authenticators stored in the vault")
|
||||
|
||||
p_recon = sub.add_parser("reconstruct", help="reconstruct TOTP/QR from the vault")
|
||||
p_recon.add_argument("serial", nargs="?", help="authenticator serial (prompts if omitted)")
|
||||
|
||||
p_mig = sub.add_parser("migrate", help="import legacy JSON backups into the vault")
|
||||
p_mig.add_argument("--dir", default=".", help="directory to scan (default: current)")
|
||||
p_mig.add_argument("--overwrite", action="store_true", help="replace existing vault entries")
|
||||
|
||||
sub.add_parser("paths", help="show config/data/vault locations")
|
||||
return parser
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
parser = _build_parser()
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
ensure_user_settings()
|
||||
try:
|
||||
settings = load_settings()
|
||||
except BnetAuthError as exc:
|
||||
print(f"Configuration error: {exc}", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
try:
|
||||
if args.command == "attach":
|
||||
action_attach(settings)
|
||||
elif args.command == "retrieve":
|
||||
action_retrieve(settings)
|
||||
elif args.command == "list":
|
||||
action_list(settings)
|
||||
elif args.command == "reconstruct":
|
||||
action_reconstruct(settings, args.serial)
|
||||
elif args.command == "migrate":
|
||||
action_migrate(settings, Path(args.dir), args.overwrite)
|
||||
elif args.command == "paths":
|
||||
action_paths(settings)
|
||||
else:
|
||||
interactive(settings)
|
||||
except KeyboardInterrupt:
|
||||
print("\nInterrupted.")
|
||||
return 130
|
||||
except BnetAuthError as exc:
|
||||
print(f"\nError: {exc}", file=sys.stderr)
|
||||
return 1
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,173 @@
|
||||
"""Configuration loading and per-user path resolution.
|
||||
|
||||
The bundled ``settings.yaml`` (shipped inside the package) holds defaults. On
|
||||
first use a copy is written to the user's config directory; values found there
|
||||
are overlaid on top of the defaults, so users can edit endpoints/KDF/TOTP
|
||||
parameters without touching the installed package.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from importlib import resources
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import yaml
|
||||
from platformdirs import user_config_dir, user_data_dir
|
||||
|
||||
from .errors import ConfigError
|
||||
|
||||
APP_NAME = "bnet_auth_tool"
|
||||
APP_AUTHOR = "Nighthawk42"
|
||||
SETTINGS_FILENAME = "settings.yaml"
|
||||
VAULT_FILENAME = "vault.json"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Path helpers
|
||||
# --------------------------------------------------------------------------- #
|
||||
def config_dir() -> Path:
|
||||
"""Directory holding the user-editable settings file."""
|
||||
return Path(user_config_dir(APP_NAME, APP_AUTHOR))
|
||||
|
||||
|
||||
def data_dir() -> Path:
|
||||
"""Directory holding the encrypted vault."""
|
||||
return Path(user_data_dir(APP_NAME, APP_AUTHOR))
|
||||
|
||||
|
||||
def user_settings_path() -> Path:
|
||||
return config_dir() / SETTINGS_FILENAME
|
||||
|
||||
|
||||
def vault_path() -> Path:
|
||||
return data_dir() / VAULT_FILENAME
|
||||
|
||||
|
||||
def _bundled_settings_text() -> str:
|
||||
return resources.files(APP_NAME).joinpath(SETTINGS_FILENAME).read_text(encoding="utf-8")
|
||||
|
||||
|
||||
def ensure_user_settings() -> Path:
|
||||
"""Copy the bundled defaults into the user config dir if absent.
|
||||
|
||||
Returns the path to the user settings file.
|
||||
"""
|
||||
dest = user_settings_path()
|
||||
if not dest.exists():
|
||||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||||
dest.write_text(_bundled_settings_text(), encoding="utf-8")
|
||||
return dest
|
||||
|
||||
|
||||
def _deep_merge(base: dict[str, Any], overlay: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Recursively overlay ``overlay`` onto ``base`` (returns a new dict)."""
|
||||
result = dict(base)
|
||||
for key, value in overlay.items():
|
||||
if isinstance(value, dict) and isinstance(result.get(key), dict):
|
||||
result[key] = _deep_merge(result[key], value)
|
||||
else:
|
||||
result[key] = value
|
||||
return result
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Typed config sections
|
||||
# --------------------------------------------------------------------------- #
|
||||
@dataclass(frozen=True)
|
||||
class ApiConfig:
|
||||
host: str
|
||||
attach_path: str
|
||||
device_path: str
|
||||
sso_url: str
|
||||
client_id: str
|
||||
timeout: int
|
||||
region_prefixes: list[str] = field(default_factory=list)
|
||||
|
||||
@property
|
||||
def attach_url(self) -> str:
|
||||
return f"{self.host}{self.attach_path}"
|
||||
|
||||
@property
|
||||
def device_url(self) -> str:
|
||||
return f"{self.host}{self.device_path}"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class TotpConfig:
|
||||
algorithm: str
|
||||
digits: int
|
||||
period: int
|
||||
issuer: str
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CryptoConfig:
|
||||
kdf: str
|
||||
scrypt_n: int
|
||||
scrypt_r: int
|
||||
scrypt_p: int
|
||||
pbkdf2_iterations: int
|
||||
pbkdf2_legacy_iterations: int
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Settings:
|
||||
api: ApiConfig
|
||||
totp: TotpConfig
|
||||
crypto: CryptoConfig
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, raw: dict[str, Any]) -> Settings:
|
||||
try:
|
||||
api = raw["api"]
|
||||
totp = raw["totp"]
|
||||
crypto = raw["crypto"]
|
||||
scrypt = crypto["scrypt"]
|
||||
pbkdf2 = crypto["pbkdf2"]
|
||||
return cls(
|
||||
api=ApiConfig(
|
||||
host=api["host"].rstrip("/"),
|
||||
attach_path=api["attach_path"],
|
||||
device_path=api["device_path"],
|
||||
sso_url=api["sso_url"],
|
||||
client_id=api["client_id"],
|
||||
timeout=int(api.get("timeout", 20)),
|
||||
region_prefixes=list(api.get("region_prefixes", [])),
|
||||
),
|
||||
totp=TotpConfig(
|
||||
algorithm=str(totp["algorithm"]).upper(),
|
||||
digits=int(totp["digits"]),
|
||||
period=int(totp["period"]),
|
||||
issuer=str(totp["issuer"]),
|
||||
),
|
||||
crypto=CryptoConfig(
|
||||
kdf=str(crypto.get("kdf", "scrypt")).lower(),
|
||||
scrypt_n=int(scrypt["n"]),
|
||||
scrypt_r=int(scrypt["r"]),
|
||||
scrypt_p=int(scrypt["p"]),
|
||||
pbkdf2_iterations=int(pbkdf2["iterations"]),
|
||||
pbkdf2_legacy_iterations=int(pbkdf2["legacy_iterations"]),
|
||||
),
|
||||
)
|
||||
except (KeyError, TypeError, ValueError) as exc:
|
||||
raise ConfigError(f"Invalid settings file: {exc}") from exc
|
||||
|
||||
|
||||
def load_settings() -> Settings:
|
||||
"""Load bundled defaults overlaid with the user's settings file."""
|
||||
defaults = yaml.safe_load(_bundled_settings_text()) or {}
|
||||
merged = defaults
|
||||
|
||||
user_path = user_settings_path()
|
||||
if user_path.exists():
|
||||
try:
|
||||
overlay = yaml.safe_load(user_path.read_text(encoding="utf-8")) or {}
|
||||
except yaml.YAMLError as exc:
|
||||
raise ConfigError(f"Could not parse {user_path}: {exc}") from exc
|
||||
if not isinstance(overlay, dict):
|
||||
raise ConfigError(f"{user_path} must contain a YAML mapping at the top level.")
|
||||
merged = _deep_merge(defaults, overlay)
|
||||
|
||||
return Settings.from_dict(merged)
|
||||
@@ -0,0 +1,171 @@
|
||||
"""Authenticated encryption for vault and backup data.
|
||||
|
||||
New data is encrypted with scrypt (memory-hard KDF) + AES-256-GCM and a
|
||||
versioned, self-describing header. Decryption auto-detects the scheme so that
|
||||
files produced by older versions keep working:
|
||||
|
||||
* ``format: 2`` packages declare their own ``kdf`` ("scrypt" or "pbkdf2").
|
||||
* Legacy v1.x packages have no ``format``/``kdf`` key — they are PBKDF2-HMAC-
|
||||
SHA256, using the embedded ``kdf_iterations`` when present, or the configured
|
||||
legacy iteration count (100k) when the field is missing entirely.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import binascii
|
||||
import json
|
||||
import os
|
||||
from typing import Any
|
||||
|
||||
from cryptography.exceptions import InvalidTag
|
||||
from cryptography.hazmat.primitives import hashes
|
||||
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
||||
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
|
||||
from cryptography.hazmat.primitives.kdf.scrypt import Scrypt
|
||||
|
||||
from .config import CryptoConfig
|
||||
from .errors import DecryptionError, EncryptionError
|
||||
|
||||
FORMAT_VERSION = 2
|
||||
SALT_SIZE = 16
|
||||
NONCE_SIZE = 12
|
||||
AES_KEY_SIZE = 32 # AES-256
|
||||
|
||||
|
||||
def _b64e(raw: bytes) -> str:
|
||||
return base64.b64encode(raw).decode("ascii")
|
||||
|
||||
|
||||
def _b64d(text: str) -> bytes:
|
||||
return base64.b64decode(text)
|
||||
|
||||
|
||||
class EncryptionManager:
|
||||
"""Encrypts/decrypts JSON-serialisable mappings with a passphrase."""
|
||||
|
||||
def __init__(self, passphrase: str, crypto: CryptoConfig):
|
||||
if not passphrase:
|
||||
raise ValueError("Passphrase cannot be empty.")
|
||||
# Keep the passphrase as a mutable bytearray so it can be scrubbed.
|
||||
self._passphrase = bytearray(passphrase.encode("utf-8"))
|
||||
self._cfg = crypto
|
||||
|
||||
# -- key derivation ----------------------------------------------------- #
|
||||
def _derive_scrypt(self, salt: bytes, n: int, r: int, p: int) -> bytes:
|
||||
kdf = Scrypt(salt=salt, length=AES_KEY_SIZE, n=n, r=r, p=p)
|
||||
return kdf.derive(bytes(self._passphrase))
|
||||
|
||||
def _derive_pbkdf2(self, salt: bytes, iterations: int) -> bytes:
|
||||
kdf = PBKDF2HMAC(
|
||||
algorithm=hashes.SHA256(),
|
||||
length=AES_KEY_SIZE,
|
||||
salt=salt,
|
||||
iterations=iterations,
|
||||
)
|
||||
return kdf.derive(bytes(self._passphrase))
|
||||
|
||||
# -- public API --------------------------------------------------------- #
|
||||
def encrypt(self, data: dict[str, Any]) -> bytes:
|
||||
"""Encrypt a mapping, returning the JSON package as UTF-8 bytes."""
|
||||
try:
|
||||
plaintext = json.dumps(data, ensure_ascii=False).encode("utf-8")
|
||||
salt = os.urandom(SALT_SIZE)
|
||||
nonce = os.urandom(NONCE_SIZE)
|
||||
|
||||
package: dict[str, Any] = {
|
||||
"format": FORMAT_VERSION,
|
||||
"kdf": self._cfg.kdf,
|
||||
"salt": _b64e(salt),
|
||||
"nonce": _b64e(nonce),
|
||||
}
|
||||
|
||||
if self._cfg.kdf == "scrypt":
|
||||
n, r, p = self._cfg.scrypt_n, self._cfg.scrypt_r, self._cfg.scrypt_p
|
||||
key = self._derive_scrypt(salt, n, r, p)
|
||||
package["scrypt"] = {"n": n, "r": r, "p": p}
|
||||
elif self._cfg.kdf == "pbkdf2":
|
||||
iterations = self._cfg.pbkdf2_iterations
|
||||
key = self._derive_pbkdf2(salt, iterations)
|
||||
package["kdf_iterations"] = iterations
|
||||
else:
|
||||
raise EncryptionError(f"Unsupported KDF: {self._cfg.kdf!r}")
|
||||
|
||||
ciphertext = AESGCM(key).encrypt(nonce, plaintext, None)
|
||||
package["ciphertext"] = _b64e(ciphertext)
|
||||
return json.dumps(package, indent=2).encode("utf-8")
|
||||
except EncryptionError:
|
||||
raise
|
||||
except Exception as exc: # noqa: BLE001 - wrap any crypto/serialisation failure
|
||||
raise EncryptionError(f"Encryption failed: {exc}") from exc
|
||||
|
||||
def decrypt(self, encrypted_bytes: bytes) -> dict[str, Any]:
|
||||
"""Decrypt a package produced by any supported version of this tool."""
|
||||
try:
|
||||
package = json.loads(encrypted_bytes.decode("utf-8"))
|
||||
salt = _b64d(package["salt"])
|
||||
nonce = _b64d(package["nonce"])
|
||||
ciphertext = _b64d(package["ciphertext"])
|
||||
key = self._derive_key_for(package, salt)
|
||||
except InvalidTag: # pragma: no cover - raised below, kept for clarity
|
||||
raise
|
||||
except (KeyError, ValueError, TypeError, binascii.Error, json.JSONDecodeError) as exc:
|
||||
raise DecryptionError(
|
||||
f"Decryption failed: invalid data format or content. {exc}"
|
||||
) from exc
|
||||
|
||||
try:
|
||||
plaintext = AESGCM(key).decrypt(nonce, ciphertext, None)
|
||||
return json.loads(plaintext.decode("utf-8"))
|
||||
except InvalidTag as exc:
|
||||
raise DecryptionError(
|
||||
"Decryption failed: authentication tag mismatch. "
|
||||
"Check the passphrase or data integrity."
|
||||
) from exc
|
||||
except (ValueError, json.JSONDecodeError) as exc:
|
||||
raise DecryptionError(f"Decryption failed: corrupt plaintext. {exc}") from exc
|
||||
|
||||
def _derive_key_for(self, package: dict[str, Any], salt: bytes) -> bytes:
|
||||
"""Pick the KDF based on the package header and derive the AES key."""
|
||||
kdf = package.get("kdf")
|
||||
|
||||
if kdf == "scrypt":
|
||||
params = package.get("scrypt", {})
|
||||
return self._derive_scrypt(
|
||||
salt,
|
||||
int(params["n"]),
|
||||
int(params["r"]),
|
||||
int(params["p"]),
|
||||
)
|
||||
|
||||
if kdf == "pbkdf2" or "format" in package:
|
||||
iterations = int(package.get("kdf_iterations", self._cfg.pbkdf2_iterations))
|
||||
return self._derive_pbkdf2(salt, iterations)
|
||||
|
||||
# Legacy v1.x package: no format/kdf header -> PBKDF2.
|
||||
if "kdf_iterations" in package:
|
||||
iterations = int(package["kdf_iterations"])
|
||||
else:
|
||||
iterations = self._cfg.pbkdf2_legacy_iterations
|
||||
return self._derive_pbkdf2(salt, iterations)
|
||||
|
||||
# -- hygiene ------------------------------------------------------------ #
|
||||
def close(self) -> None:
|
||||
"""Best-effort scrub of the in-memory passphrase."""
|
||||
for i in range(len(self._passphrase)):
|
||||
self._passphrase[i] = 0
|
||||
|
||||
def __enter__(self) -> EncryptionManager:
|
||||
return self
|
||||
|
||||
def __exit__(self, *_exc: object) -> None:
|
||||
self.close()
|
||||
|
||||
|
||||
def looks_encrypted(raw: bytes) -> bool:
|
||||
"""Heuristic: does ``raw`` look like one of our encryption packages?"""
|
||||
try:
|
||||
package = json.loads(raw.decode("utf-8"))
|
||||
except (UnicodeDecodeError, json.JSONDecodeError):
|
||||
return False
|
||||
return isinstance(package, dict) and {"salt", "nonce", "ciphertext"} <= package.keys()
|
||||
@@ -0,0 +1,27 @@
|
||||
"""Exception hierarchy for bnet_auth_tool."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
|
||||
class BnetAuthError(Exception):
|
||||
"""Base class for all errors raised by this package."""
|
||||
|
||||
|
||||
class AuthenticatorError(BnetAuthError):
|
||||
"""A problem talking to the Battle.net authenticator API."""
|
||||
|
||||
|
||||
class EncryptionError(BnetAuthError):
|
||||
"""Encryption of vault/backup data failed."""
|
||||
|
||||
|
||||
class DecryptionError(BnetAuthError):
|
||||
"""Decryption failed (wrong passphrase, corrupt data, or bad format)."""
|
||||
|
||||
|
||||
class ConfigError(BnetAuthError):
|
||||
"""The settings file is missing required values or is malformed."""
|
||||
|
||||
|
||||
class StorageError(BnetAuthError):
|
||||
"""The vault could not be read or written."""
|
||||
@@ -0,0 +1,52 @@
|
||||
"""Filesystem helpers: atomic, permission-hardened writes.
|
||||
|
||||
Authenticator material is sensitive, so written files are created with owner-
|
||||
only permissions (``0o600``) where the platform supports it, and writes are
|
||||
atomic (temp file + ``os.replace``) so a crash mid-write cannot truncate an
|
||||
existing vault.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import os
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
# Owner read/write only.
|
||||
SECRET_FILE_MODE = 0o600
|
||||
|
||||
|
||||
def _harden(path: Path) -> None:
|
||||
"""Best-effort chmod to owner-only; a no-op on platforms without POSIX perms."""
|
||||
# Windows / restricted filesystems: ACLs differ; nothing portable to do.
|
||||
with contextlib.suppress(OSError, NotImplementedError):
|
||||
os.chmod(path, SECRET_FILE_MODE)
|
||||
|
||||
|
||||
def atomic_write_bytes(path: Path, data: bytes, *, secret: bool = True) -> None:
|
||||
"""Atomically write ``data`` to ``path``.
|
||||
|
||||
Writes to a temp file in the same directory, fsyncs it, hardens its
|
||||
permissions, then atomically replaces the destination.
|
||||
"""
|
||||
path = Path(path)
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
fd, tmp_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=str(path.parent))
|
||||
tmp_path = Path(tmp_name)
|
||||
try:
|
||||
with os.fdopen(fd, "wb") as fh:
|
||||
fh.write(data)
|
||||
fh.flush()
|
||||
os.fsync(fh.fileno())
|
||||
if secret:
|
||||
_harden(tmp_path)
|
||||
os.replace(tmp_path, path)
|
||||
except BaseException:
|
||||
tmp_path.unlink(missing_ok=True)
|
||||
raise
|
||||
|
||||
|
||||
def atomic_write_text(path: Path, text: str, *, secret: bool = True) -> None:
|
||||
atomic_write_bytes(path, text.encode("utf-8"), secret=secret)
|
||||
@@ -0,0 +1,572 @@
|
||||
"""Flet desktop GUI — a friendly front-end over the same vault as the CLI.
|
||||
|
||||
Run with ``bnet-auth-gui`` (after ``pip install '.[gui]'``) or
|
||||
``python -m bnet_auth_tool.gui``. The GUI uses the *same* encrypted vault,
|
||||
settings, and crypto as the command line; it is purely an alternative UI.
|
||||
|
||||
Online attach/retrieve are included but, like the CLI, are **unverified**
|
||||
against Blizzard's live API. The offline vault/TOTP features are the core.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
try:
|
||||
import flet as ft
|
||||
except ImportError as exc: # pragma: no cover - only hit without the gui extra
|
||||
raise SystemExit(
|
||||
"The GUI needs Flet. Install it with: pip install 'bnet-auth-tool[gui]'"
|
||||
) from exc
|
||||
|
||||
from . import __version__
|
||||
from .api import BattleNetAuthenticator
|
||||
from .config import Settings, ensure_user_settings, load_settings, vault_path
|
||||
from .crypto import EncryptionManager
|
||||
from .errors import BnetAuthError
|
||||
from .migrate import discover_legacy_files, migrate_files
|
||||
from .storage import Vault
|
||||
from .totp import (
|
||||
build_totp_url,
|
||||
current_code,
|
||||
generate_qr_code,
|
||||
hex_secret_to_base32,
|
||||
seconds_remaining,
|
||||
)
|
||||
|
||||
WINDOW_W, WINDOW_H = 560, 720
|
||||
_UNVERIFIED = "Online flows are unverified against Blizzard's API and may fail."
|
||||
|
||||
|
||||
class AuthApp:
|
||||
"""Holds GUI state for one window: the page, settings, and open vault."""
|
||||
|
||||
def __init__(self, page: ft.Page, settings: Settings):
|
||||
self.page = page
|
||||
self.settings = settings
|
||||
self.vault: Vault | None = None
|
||||
self.passphrase: str | None = None
|
||||
self._ticking = False
|
||||
self._code_cells: dict[str, tuple[ft.Text, ft.Text]] = {}
|
||||
self._tmpdir = Path(tempfile.mkdtemp(prefix="bnet_qr_"))
|
||||
|
||||
# -- window setup ------------------------------------------------------- #
|
||||
def start(self) -> None:
|
||||
self.page.title = "Battle.net Authenticator Tool"
|
||||
self.page.theme_mode = ft.ThemeMode.DARK
|
||||
self.page.padding = 24
|
||||
self.page.window.width = WINDOW_W
|
||||
self.page.window.height = WINDOW_H
|
||||
self.page.window.min_width = 420
|
||||
self.page.window.min_height = 520
|
||||
self._show_lock()
|
||||
|
||||
# -- small helpers ------------------------------------------------------ #
|
||||
def _swap(self, *controls: ft.Control) -> None:
|
||||
self._ticking = False
|
||||
self.page.controls.clear()
|
||||
self.page.add(*controls)
|
||||
|
||||
def _toast(self, message: str) -> None:
|
||||
self.page.show_dialog(
|
||||
ft.AlertDialog(
|
||||
content=ft.Text(message),
|
||||
actions=[ft.TextButton("OK", on_click=lambda e: self.page.pop_dialog())],
|
||||
)
|
||||
)
|
||||
|
||||
def _header(self, subtitle: str) -> ft.Control:
|
||||
return ft.Column(
|
||||
spacing=2,
|
||||
controls=[
|
||||
ft.Text("Battle.net Authenticator", size=22, weight=ft.FontWeight.BOLD),
|
||||
ft.Text(subtitle, size=12, color=ft.Colors.ON_SURFACE_VARIANT),
|
||||
],
|
||||
)
|
||||
|
||||
# ------------------------------------------------------------------ #
|
||||
# Lock screen
|
||||
# ------------------------------------------------------------------ #
|
||||
def _show_lock(self) -> None:
|
||||
exists = vault_path().is_file()
|
||||
title = "Unlock your vault" if exists else "Create a new vault"
|
||||
|
||||
pass_field = ft.TextField(
|
||||
label="Vault passphrase",
|
||||
password=True,
|
||||
can_reveal_password=True,
|
||||
autofocus=True,
|
||||
on_submit=lambda e: do_unlock(e),
|
||||
)
|
||||
confirm_field = ft.TextField(
|
||||
label="Confirm passphrase",
|
||||
password=True,
|
||||
can_reveal_password=True,
|
||||
visible=not exists,
|
||||
on_submit=lambda e: do_unlock(e),
|
||||
)
|
||||
error = ft.Text(color=ft.Colors.ERROR, visible=False)
|
||||
|
||||
def do_unlock(_e: object) -> None:
|
||||
error.visible = False
|
||||
pw = pass_field.value or ""
|
||||
if not pw:
|
||||
return self._set_error(error, "Passphrase cannot be empty.")
|
||||
if not exists and pw != (confirm_field.value or ""):
|
||||
return self._set_error(error, "Passphrases do not match.")
|
||||
try:
|
||||
manager = EncryptionManager(pw, self.settings.crypto)
|
||||
vault = Vault(manager).load()
|
||||
except BnetAuthError as exc:
|
||||
return self._set_error(error, f"Could not open vault: {exc}")
|
||||
self.vault = vault
|
||||
self.passphrase = pw
|
||||
self._show_vault()
|
||||
|
||||
button_label = "Unlock" if exists else "Create vault"
|
||||
self._swap(
|
||||
ft.Column(
|
||||
spacing=18,
|
||||
controls=[
|
||||
self._header(title),
|
||||
ft.Text(
|
||||
f"Vault file: {vault_path()}",
|
||||
size=11,
|
||||
color=ft.Colors.ON_SURFACE_VARIANT,
|
||||
selectable=True,
|
||||
),
|
||||
pass_field,
|
||||
confirm_field,
|
||||
error,
|
||||
ft.FilledButton(button_label, icon=ft.Icons.LOCK_OPEN, on_click=do_unlock),
|
||||
ft.Text(f"v{__version__}", size=10, color=ft.Colors.ON_SURFACE_VARIANT),
|
||||
],
|
||||
)
|
||||
)
|
||||
|
||||
def _set_error(self, control: ft.Text, message: str) -> None:
|
||||
control.value = message
|
||||
control.visible = True
|
||||
self.page.update()
|
||||
|
||||
# ------------------------------------------------------------------ #
|
||||
# Vault screen
|
||||
# ------------------------------------------------------------------ #
|
||||
def _show_vault(self) -> None:
|
||||
assert self.vault is not None
|
||||
self._code_cells = {}
|
||||
|
||||
toolbar = ft.Row(
|
||||
alignment=ft.MainAxisAlignment.SPACE_BETWEEN,
|
||||
controls=[
|
||||
self._header(f"{len(self.vault)} authenticator(s)"),
|
||||
ft.Row(
|
||||
controls=[
|
||||
ft.IconButton(
|
||||
ft.Icons.ADD, tooltip="Add (online)", on_click=self._open_add_menu
|
||||
),
|
||||
ft.IconButton(
|
||||
ft.Icons.UPLOAD_FILE,
|
||||
tooltip="Import legacy files",
|
||||
on_click=lambda e: self._show_migrate(),
|
||||
),
|
||||
ft.IconButton(
|
||||
ft.Icons.LOCK, tooltip="Lock", on_click=lambda e: self._lock()
|
||||
),
|
||||
]
|
||||
),
|
||||
],
|
||||
)
|
||||
|
||||
rows = [self._entry_row(entry) for entry in self._sorted_entries()]
|
||||
body: ft.Control
|
||||
if rows:
|
||||
body = ft.ListView(controls=rows, spacing=10, expand=True)
|
||||
else:
|
||||
body = ft.Container(
|
||||
content=ft.Text(
|
||||
"Vault is empty. Use + to add online, or import legacy backups.",
|
||||
color=ft.Colors.ON_SURFACE_VARIANT,
|
||||
),
|
||||
alignment=ft.Alignment.CENTER,
|
||||
expand=True,
|
||||
)
|
||||
|
||||
self.page.controls.clear()
|
||||
self.page.add(ft.Column(expand=True, spacing=16, controls=[toolbar, ft.Divider(), body]))
|
||||
self._start_ticker()
|
||||
|
||||
def _sorted_entries(self) -> list[dict]:
|
||||
assert self.vault is not None
|
||||
return sorted(self.vault.list(), key=lambda e: e.get("serial", ""))
|
||||
|
||||
def _entry_row(self, entry: dict) -> ft.Control:
|
||||
serial = entry.get("serial", "?")
|
||||
code_text = ft.Text("--------", size=26, weight=ft.FontWeight.BOLD, font_family="monospace")
|
||||
time_text = ft.Text("", size=11, color=ft.Colors.ON_SURFACE_VARIANT)
|
||||
self._code_cells[serial] = (code_text, time_text)
|
||||
self._update_code(entry)
|
||||
|
||||
return ft.Card(
|
||||
content=ft.Container(
|
||||
padding=14,
|
||||
content=ft.Row(
|
||||
alignment=ft.MainAxisAlignment.SPACE_BETWEEN,
|
||||
controls=[
|
||||
ft.Column(
|
||||
spacing=2,
|
||||
controls=[
|
||||
ft.Text(serial, weight=ft.FontWeight.W_500),
|
||||
ft.Row(spacing=10, controls=[code_text, time_text]),
|
||||
],
|
||||
),
|
||||
ft.Row(
|
||||
controls=[
|
||||
ft.IconButton(
|
||||
ft.Icons.COPY,
|
||||
tooltip="Copy code",
|
||||
on_click=lambda e, en=entry: self._copy_code(en),
|
||||
),
|
||||
ft.IconButton(
|
||||
ft.Icons.INFO_OUTLINE,
|
||||
tooltip="Details / QR",
|
||||
on_click=lambda e, en=entry: self._show_details(en),
|
||||
),
|
||||
]
|
||||
),
|
||||
],
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
# -- live code ticker --------------------------------------------------- #
|
||||
def _start_ticker(self) -> None:
|
||||
self._ticking = True
|
||||
self.page.run_task(self._tick)
|
||||
|
||||
async def _tick(self) -> None:
|
||||
while self._ticking:
|
||||
try:
|
||||
for entry in self._sorted_entries():
|
||||
self._update_code(entry)
|
||||
self.page.update()
|
||||
except Exception: # noqa: BLE001 - never let the ticker kill the app
|
||||
pass
|
||||
await asyncio.sleep(1)
|
||||
|
||||
def _update_code(self, entry: dict) -> None:
|
||||
cells = self._code_cells.get(entry.get("serial", ""))
|
||||
if not cells:
|
||||
return
|
||||
code_text, time_text = cells
|
||||
secret = entry.get("base32Secret")
|
||||
if not secret:
|
||||
code_text.value = "no secret"
|
||||
return
|
||||
try:
|
||||
code_text.value = current_code(secret, self.settings.totp)
|
||||
time_text.value = f"{seconds_remaining(self.settings.totp)}s"
|
||||
except ValueError:
|
||||
code_text.value = "bad secret"
|
||||
|
||||
def _copy_code(self, entry: dict) -> None:
|
||||
secret = entry.get("base32Secret")
|
||||
if not secret:
|
||||
return
|
||||
try:
|
||||
self.page.clipboard.set(current_code(secret, self.settings.totp))
|
||||
self._toast("Current code copied to clipboard.")
|
||||
except ValueError as exc:
|
||||
self._toast(str(exc))
|
||||
|
||||
# ------------------------------------------------------------------ #
|
||||
# Detail dialog (QR / copy / delete)
|
||||
# ------------------------------------------------------------------ #
|
||||
def _show_details(self, entry: dict) -> None:
|
||||
serial = entry.get("serial", "?")
|
||||
totp_url = entry.get("totpUrl") or build_totp_url(
|
||||
serial, entry.get("base32Secret", ""), self.settings.totp
|
||||
)
|
||||
|
||||
rows: list[ft.Control] = [
|
||||
self._field("Serial", serial),
|
||||
self._field("Restore code", entry.get("restoreCode") or "—"),
|
||||
self._field("Base32 secret", entry.get("base32Secret") or "—"),
|
||||
]
|
||||
|
||||
qr_holder = ft.Column(horizontal_alignment=ft.CrossAxisAlignment.CENTER)
|
||||
|
||||
def show_qr(_e: object) -> None:
|
||||
try:
|
||||
path = generate_qr_code(totp_url, self._tmpdir / f"{serial}.png")
|
||||
except Exception as exc: # noqa: BLE001
|
||||
qr_holder.controls = [ft.Text(f"QR error: {exc}", color=ft.Colors.ERROR)]
|
||||
self.page.update()
|
||||
return
|
||||
qr_holder.controls = [
|
||||
ft.Image(src=str(path), width=220, height=220),
|
||||
ft.Text(
|
||||
"This QR contains your secret — close after scanning.",
|
||||
size=11,
|
||||
color=ft.Colors.ERROR,
|
||||
),
|
||||
]
|
||||
self.page.update()
|
||||
|
||||
dialog = ft.AlertDialog(
|
||||
title=ft.Text(f"Authenticator {serial}"),
|
||||
content=ft.Column(
|
||||
tight=True,
|
||||
width=360,
|
||||
scroll=ft.ScrollMode.AUTO,
|
||||
controls=[*rows, ft.Divider(), qr_holder],
|
||||
),
|
||||
actions=[
|
||||
ft.TextButton("Show QR", icon=ft.Icons.QR_CODE, on_click=show_qr),
|
||||
ft.TextButton(
|
||||
"Delete",
|
||||
icon=ft.Icons.DELETE,
|
||||
style=ft.ButtonStyle(color=ft.Colors.ERROR),
|
||||
on_click=lambda e: self._confirm_delete(entry),
|
||||
),
|
||||
ft.TextButton("Close", on_click=lambda e: self.page.pop_dialog()),
|
||||
],
|
||||
)
|
||||
self.page.show_dialog(dialog)
|
||||
|
||||
def _field(self, label: str, value: str) -> ft.Control:
|
||||
return ft.Column(
|
||||
spacing=0,
|
||||
controls=[
|
||||
ft.Text(label, size=11, color=ft.Colors.ON_SURFACE_VARIANT),
|
||||
ft.Text(value, selectable=True, font_family="monospace"),
|
||||
],
|
||||
)
|
||||
|
||||
def _confirm_delete(self, entry: dict) -> None:
|
||||
serial = entry.get("serial", "?")
|
||||
|
||||
def really_delete(_e: object) -> None:
|
||||
assert self.vault is not None
|
||||
self.vault.remove(serial)
|
||||
self.vault.save()
|
||||
self.page.pop_dialog()
|
||||
self._show_vault()
|
||||
|
||||
self.page.show_dialog(
|
||||
ft.AlertDialog(
|
||||
modal=True,
|
||||
title=ft.Text(f"Remove {serial}?"),
|
||||
content=ft.Text(
|
||||
"This deletes the entry from the vault. Make sure you have a backup "
|
||||
"of the secret — there is no undo."
|
||||
),
|
||||
actions=[
|
||||
ft.TextButton(
|
||||
"Delete",
|
||||
style=ft.ButtonStyle(color=ft.Colors.ERROR),
|
||||
on_click=really_delete,
|
||||
),
|
||||
ft.TextButton("Cancel", on_click=lambda e: self.page.pop_dialog()),
|
||||
],
|
||||
)
|
||||
)
|
||||
|
||||
# ------------------------------------------------------------------ #
|
||||
# Add (online) — attach / retrieve
|
||||
# ------------------------------------------------------------------ #
|
||||
def _open_add_menu(self, _e: object) -> None:
|
||||
self.page.show_dialog(
|
||||
ft.AlertDialog(
|
||||
title=ft.Text("Add authenticator (online)"),
|
||||
content=ft.Text(_UNVERIFIED),
|
||||
actions=[
|
||||
ft.TextButton("Attach new", on_click=lambda e: self._show_attach()),
|
||||
ft.TextButton("Retrieve existing", on_click=lambda e: self._show_retrieve()),
|
||||
ft.TextButton("Cancel", on_click=lambda e: self.page.pop_dialog()),
|
||||
],
|
||||
)
|
||||
)
|
||||
|
||||
def _show_attach(self) -> None:
|
||||
token = ft.TextField(label="Session token (ST=...)", autofocus=True)
|
||||
error = ft.Text(color=ft.Colors.ERROR, visible=False)
|
||||
|
||||
def run(_e: object) -> None:
|
||||
value = (token.value or "").strip()
|
||||
if not value:
|
||||
return self._set_error(error, "Session token is required.")
|
||||
client = BattleNetAuthenticator(self.settings.api)
|
||||
try:
|
||||
client.get_bearer_token(value)
|
||||
device = client.attach_authenticator()
|
||||
except BnetAuthError as exc:
|
||||
return self._set_error(error, str(exc))
|
||||
self._store_device(device)
|
||||
|
||||
self.page.show_dialog(
|
||||
ft.AlertDialog(
|
||||
title=ft.Text("Attach new authenticator"),
|
||||
content=ft.Column(
|
||||
tight=True,
|
||||
width=360,
|
||||
controls=[
|
||||
ft.Text(_UNVERIFIED, size=11, color=ft.Colors.ON_SURFACE_VARIANT),
|
||||
token,
|
||||
error,
|
||||
],
|
||||
),
|
||||
actions=[
|
||||
ft.TextButton("Attach", on_click=run),
|
||||
ft.TextButton("Cancel", on_click=lambda e: self.page.pop_dialog()),
|
||||
],
|
||||
)
|
||||
)
|
||||
|
||||
def _show_retrieve(self) -> None:
|
||||
account = ft.TextField(label="Account email or phone", autofocus=True)
|
||||
serial = ft.TextField(label="Serial")
|
||||
restore = ft.TextField(label="Restore code")
|
||||
error = ft.Text(color=ft.Colors.ERROR, visible=False)
|
||||
|
||||
def run(_e: object) -> None:
|
||||
if not (account.value and serial.value and restore.value):
|
||||
return self._set_error(error, "All three fields are required.")
|
||||
client = BattleNetAuthenticator(self.settings.api)
|
||||
try:
|
||||
retrieved = client.retrieve_device_secret(
|
||||
account.value, serial.value, restore.value
|
||||
)
|
||||
except BnetAuthError as exc:
|
||||
return self._set_error(error, str(exc))
|
||||
self._store_device(
|
||||
{
|
||||
"serial": serial.value.strip(),
|
||||
"restoreCode": restore.value.strip(),
|
||||
"deviceSecret": retrieved["deviceSecret"],
|
||||
}
|
||||
)
|
||||
|
||||
self.page.show_dialog(
|
||||
ft.AlertDialog(
|
||||
title=ft.Text("Retrieve existing secret"),
|
||||
content=ft.Column(
|
||||
tight=True,
|
||||
width=360,
|
||||
controls=[
|
||||
ft.Text(_UNVERIFIED, size=11, color=ft.Colors.ON_SURFACE_VARIANT),
|
||||
account,
|
||||
serial,
|
||||
restore,
|
||||
error,
|
||||
],
|
||||
),
|
||||
actions=[
|
||||
ft.TextButton("Retrieve", on_click=run),
|
||||
ft.TextButton("Cancel", on_click=lambda e: self.page.pop_dialog()),
|
||||
],
|
||||
)
|
||||
)
|
||||
|
||||
def _store_device(self, device: dict) -> None:
|
||||
assert self.vault is not None
|
||||
serial = device["serial"]
|
||||
base32_secret = hex_secret_to_base32(device["deviceSecret"])
|
||||
entry = {
|
||||
"serial": serial,
|
||||
"restoreCode": device.get("restoreCode"),
|
||||
"deviceSecret": device["deviceSecret"],
|
||||
"base32Secret": base32_secret,
|
||||
"totpUrl": build_totp_url(serial, base32_secret, self.settings.totp),
|
||||
}
|
||||
self.vault.add(entry, overwrite=True)
|
||||
self.vault.save()
|
||||
self.page.pop_dialog()
|
||||
self._show_vault()
|
||||
|
||||
# ------------------------------------------------------------------ #
|
||||
# Migrate legacy files
|
||||
# ------------------------------------------------------------------ #
|
||||
def _show_migrate(self) -> None:
|
||||
directory = ft.TextField(label="Folder to scan", value=str(Path.cwd()))
|
||||
legacy_pw = ft.TextField(
|
||||
label="Passphrase for encrypted legacy files (optional)",
|
||||
password=True,
|
||||
can_reveal_password=True,
|
||||
)
|
||||
result = ft.Text(visible=False)
|
||||
|
||||
def run(_e: object) -> None:
|
||||
assert self.vault is not None
|
||||
files = discover_legacy_files(Path(directory.value or "."))
|
||||
if not files:
|
||||
result.value = "No legacy authenticator JSON files found."
|
||||
result.visible = True
|
||||
return self.page.update()
|
||||
def provider(_path: Path) -> str | None:
|
||||
return legacy_pw.value or None
|
||||
|
||||
outcomes = migrate_files(files, self.vault, self.settings, provider, overwrite=False)
|
||||
self.vault.save()
|
||||
imported = sum(1 for o in outcomes if o.status in ("imported", "replaced"))
|
||||
skipped = sum(1 for o in outcomes if o.status == "skipped")
|
||||
errored = sum(1 for o in outcomes if o.status == "error")
|
||||
result.value = (
|
||||
f"Imported {imported}, skipped {skipped}, errors {errored}. "
|
||||
"Securely delete the originals once verified."
|
||||
)
|
||||
result.visible = True
|
||||
self.page.update()
|
||||
|
||||
self.page.show_dialog(
|
||||
ft.AlertDialog(
|
||||
title=ft.Text("Import legacy backups"),
|
||||
content=ft.Column(
|
||||
tight=True,
|
||||
width=400,
|
||||
controls=[
|
||||
ft.Text(
|
||||
"Scans a folder for old battlenet_authenticator_*.json files "
|
||||
"and imports them into the vault.",
|
||||
size=11,
|
||||
color=ft.Colors.ON_SURFACE_VARIANT,
|
||||
),
|
||||
directory,
|
||||
legacy_pw,
|
||||
result,
|
||||
],
|
||||
),
|
||||
actions=[
|
||||
ft.TextButton("Import", on_click=run),
|
||||
ft.TextButton("Done", on_click=lambda e: (self.page.pop_dialog(), self._show_vault())),
|
||||
],
|
||||
)
|
||||
)
|
||||
|
||||
# ------------------------------------------------------------------ #
|
||||
def _lock(self) -> None:
|
||||
self._ticking = False
|
||||
if self.passphrase is not None:
|
||||
self.passphrase = None
|
||||
self.vault = None
|
||||
self._show_lock()
|
||||
|
||||
|
||||
def _app(page: ft.Page) -> None:
|
||||
ensure_user_settings()
|
||||
try:
|
||||
settings = load_settings()
|
||||
except BnetAuthError as exc:
|
||||
page.add(ft.Text(f"Configuration error: {exc}", color=ft.Colors.ERROR))
|
||||
return
|
||||
AuthApp(page, settings).start()
|
||||
|
||||
|
||||
def main() -> None:
|
||||
ft.run(_app)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,121 @@
|
||||
"""Conversion tool: import legacy loose JSON backups into the vault.
|
||||
|
||||
Older versions of this tool wrote one ``battlenet_authenticator_<serial>.json``
|
||||
per authenticator into the working directory — either plaintext or encrypted
|
||||
with PBKDF2 (100k legacy or 600k). This module discovers those files, decrypts
|
||||
them if needed, normalises them, and imports them into the encrypted vault.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable, Optional
|
||||
|
||||
from .config import Settings
|
||||
from .crypto import EncryptionManager, looks_encrypted
|
||||
from .errors import DecryptionError
|
||||
from .storage import Vault
|
||||
from .totp import build_totp_url, hex_secret_to_base32
|
||||
|
||||
# Callback that, given a file path, returns the passphrase for it (or None to
|
||||
# skip). Lets the CLI prompt interactively without coupling migrate to I/O.
|
||||
PassphraseProvider = Callable[[Path], Optional[str]]
|
||||
|
||||
_LEGACY_GLOBS = ("battlenet_authenticator_*.json", "reconstructed_*.json")
|
||||
|
||||
|
||||
@dataclass
|
||||
class MigrationOutcome:
|
||||
path: Path
|
||||
serial: str | None = None
|
||||
status: str = "imported" # imported | skipped | replaced | error
|
||||
detail: str = ""
|
||||
|
||||
|
||||
def discover_legacy_files(directory: Path) -> list[Path]:
|
||||
"""Find loose authenticator JSON files in ``directory`` (non-recursive)."""
|
||||
directory = Path(directory)
|
||||
found: set[Path] = set()
|
||||
for pattern in _LEGACY_GLOBS:
|
||||
found.update(p for p in directory.glob(pattern) if p.is_file())
|
||||
return sorted(found)
|
||||
|
||||
|
||||
def normalise_entry(data: dict[str, Any], settings: Settings) -> dict[str, Any]:
|
||||
"""Coerce a raw legacy record into a canonical vault entry.
|
||||
|
||||
Derives ``base32Secret``/``totpUrl`` from ``deviceSecret`` when missing.
|
||||
"""
|
||||
serial = data.get("serial")
|
||||
if not serial:
|
||||
raise ValueError("record has no 'serial'")
|
||||
|
||||
base32_secret = data.get("base32Secret")
|
||||
device_secret = data.get("deviceSecret")
|
||||
if not base32_secret and device_secret:
|
||||
base32_secret = hex_secret_to_base32(device_secret)
|
||||
|
||||
totp_url = data.get("totpUrl")
|
||||
if not totp_url and base32_secret:
|
||||
totp_url = build_totp_url(serial, base32_secret, settings.totp)
|
||||
|
||||
entry = {
|
||||
"serial": serial,
|
||||
"restoreCode": data.get("restoreCode"),
|
||||
"deviceSecret": device_secret,
|
||||
"base32Secret": base32_secret,
|
||||
"totpUrl": totp_url,
|
||||
}
|
||||
if "timestamp" in data:
|
||||
entry["addedAt"] = data["timestamp"]
|
||||
return entry
|
||||
|
||||
|
||||
def _read_record(
|
||||
path: Path, settings: Settings, passphrase_provider: PassphraseProvider
|
||||
) -> dict[str, Any] | None:
|
||||
"""Return the decoded record from a legacy file, or None to skip it."""
|
||||
import json
|
||||
|
||||
raw = path.read_bytes()
|
||||
if looks_encrypted(raw):
|
||||
passphrase = passphrase_provider(path)
|
||||
if not passphrase:
|
||||
return None
|
||||
with EncryptionManager(passphrase, settings.crypto) as manager:
|
||||
return manager.decrypt(raw)
|
||||
return json.loads(raw.decode("utf-8"))
|
||||
|
||||
|
||||
def migrate_files(
|
||||
files: list[Path],
|
||||
vault: Vault,
|
||||
settings: Settings,
|
||||
passphrase_provider: PassphraseProvider,
|
||||
*,
|
||||
overwrite: bool = False,
|
||||
) -> list[MigrationOutcome]:
|
||||
"""Import each file into ``vault`` (caller is responsible for ``vault.save()``)."""
|
||||
import json
|
||||
|
||||
outcomes: list[MigrationOutcome] = []
|
||||
for path in files:
|
||||
try:
|
||||
record = _read_record(path, settings, passphrase_provider)
|
||||
if record is None:
|
||||
outcomes.append(MigrationOutcome(path, status="skipped", detail="no passphrase"))
|
||||
continue
|
||||
entry = normalise_entry(record, settings)
|
||||
existed = vault.get(entry["serial"]) is not None
|
||||
added = vault.add(entry, overwrite=overwrite)
|
||||
if not added:
|
||||
outcomes.append(
|
||||
MigrationOutcome(path, entry["serial"], "skipped", "already in vault")
|
||||
)
|
||||
else:
|
||||
status = "replaced" if existed else "imported"
|
||||
outcomes.append(MigrationOutcome(path, entry["serial"], status))
|
||||
except (DecryptionError, ValueError, KeyError, json.JSONDecodeError, OSError) as exc:
|
||||
outcomes.append(MigrationOutcome(path, status="error", detail=str(exc)))
|
||||
return outcomes
|
||||
@@ -0,0 +1,54 @@
|
||||
# ===========================================================================
|
||||
# bnet_auth_tool settings
|
||||
# ===========================================================================
|
||||
# This is the bundled default configuration. On first run a copy is written to
|
||||
# your per-user config directory (shown by `bnet-auth paths`); edit THAT copy
|
||||
# to change endpoints without touching the installed package. Any key omitted
|
||||
# from your copy falls back to the value here.
|
||||
# ===========================================================================
|
||||
|
||||
# --- Online Battle.net authenticator REST API -----------------------------
|
||||
# These endpoints are reverse-engineered from the official client and are NOT
|
||||
# guaranteed stable. If Blizzard moves them, update the URLs below — that is
|
||||
# the whole point of keeping them in YAML.
|
||||
api:
|
||||
# GLOBAL region host (US/EU/KR/PTR). CN uses authenticator.api.battle.net.
|
||||
host: "https://authenticator-rest-api.bnet-identity.blizzard.net"
|
||||
# setupAuthenticator — attach a new authenticator (stays on v1).
|
||||
attach_path: "/v1/authenticator"
|
||||
# restoreAuthenticator — retrieve an existing device secret (moved to v2,
|
||||
# now requires accountIdentifier alongside serial + restoreCode).
|
||||
device_path: "/v2/authenticator/device"
|
||||
# OAuth SSO token exchange.
|
||||
sso_url: "https://oauth.battle.net/oauth/sso"
|
||||
# Official Battle.net app client id used for the client_sso grant.
|
||||
client_id: "baedda12fe054e4abdfc3ad7bdea970a"
|
||||
# Network timeout (seconds) for each request.
|
||||
timeout: 20
|
||||
# Recognised region prefixes for session-token validation.
|
||||
region_prefixes: ["US-", "EU-", "KR-", "TW-", "CN-"]
|
||||
|
||||
# --- TOTP output parameters (Battle.net uses non-default digits) -----------
|
||||
totp:
|
||||
algorithm: "SHA1"
|
||||
digits: 8
|
||||
period: 30
|
||||
issuer: "Battle.net"
|
||||
|
||||
# --- Key derivation / encryption ------------------------------------------
|
||||
# New files are encrypted with scrypt + AES-256-GCM. Legacy PBKDF2 files
|
||||
# (including pre-v1.3 files missing an iteration count) remain decryptable.
|
||||
crypto:
|
||||
# Active KDF for newly encrypted data: "scrypt" or "pbkdf2".
|
||||
kdf: "scrypt"
|
||||
scrypt:
|
||||
# n must be a power of two. 2**17 ≈ 128 MiB with r=8, p=1.
|
||||
n: 131072
|
||||
r: 8
|
||||
p: 1
|
||||
pbkdf2:
|
||||
# Used when kdf == pbkdf2, and as the default for decrypting modern
|
||||
# PBKDF2 files that carry an explicit iteration count.
|
||||
iterations: 600000
|
||||
# Assumed iteration count for legacy files missing the field.
|
||||
legacy_iterations: 100000
|
||||
@@ -0,0 +1,102 @@
|
||||
"""Encrypted vault: a single file holding all saved authenticators.
|
||||
|
||||
The vault lives in the per-user data directory (see :func:`config.vault_path`)
|
||||
and is encrypted as a whole with :class:`crypto.EncryptionManager`. Each entry
|
||||
is keyed by authenticator serial. Writes are atomic and permission-hardened.
|
||||
|
||||
Vault plaintext shape::
|
||||
|
||||
{
|
||||
"version": 1,
|
||||
"entries": {
|
||||
"<serial>": {
|
||||
"serial": "...", "restoreCode": "...", "deviceSecret": "...",
|
||||
"base32Secret": "...", "totpUrl": "...", "addedAt": "<iso8601>"
|
||||
},
|
||||
...
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from .config import Settings, vault_path
|
||||
from .crypto import EncryptionManager
|
||||
from .errors import StorageError
|
||||
from .fileio import atomic_write_bytes
|
||||
|
||||
VAULT_VERSION = 1
|
||||
|
||||
|
||||
class Vault:
|
||||
"""An on-disk encrypted collection of authenticator entries."""
|
||||
|
||||
def __init__(self, manager: EncryptionManager, path: Path | None = None):
|
||||
self._manager = manager
|
||||
self._path = Path(path) if path else vault_path()
|
||||
self._entries: dict[str, dict[str, Any]] = {}
|
||||
|
||||
# -- lifecycle ---------------------------------------------------------- #
|
||||
@property
|
||||
def path(self) -> Path:
|
||||
return self._path
|
||||
|
||||
def exists(self) -> bool:
|
||||
return self._path.is_file()
|
||||
|
||||
def load(self) -> Vault:
|
||||
"""Decrypt the vault from disk into memory. New if the file is absent."""
|
||||
if not self.exists():
|
||||
self._entries = {}
|
||||
return self
|
||||
raw = self._path.read_bytes()
|
||||
data = self._manager.decrypt(raw)
|
||||
entries = data.get("entries", {})
|
||||
if not isinstance(entries, dict):
|
||||
raise StorageError("Vault is malformed: 'entries' is not a mapping.")
|
||||
self._entries = entries
|
||||
return self
|
||||
|
||||
def save(self) -> None:
|
||||
"""Encrypt and atomically persist the vault to disk."""
|
||||
payload = {"version": VAULT_VERSION, "entries": self._entries}
|
||||
atomic_write_bytes(self._path, self._manager.encrypt(payload), secret=True)
|
||||
|
||||
# -- entry operations --------------------------------------------------- #
|
||||
def list(self) -> list[dict[str, Any]]:
|
||||
return [dict(v) for v in self._entries.values()]
|
||||
|
||||
def serials(self) -> list[str]:
|
||||
return sorted(self._entries.keys())
|
||||
|
||||
def get(self, serial: str) -> dict[str, Any] | None:
|
||||
entry = self._entries.get(serial)
|
||||
return dict(entry) if entry is not None else None
|
||||
|
||||
def add(self, entry: dict[str, Any], *, overwrite: bool = False) -> bool:
|
||||
"""Add/replace an entry. Returns False if it exists and overwrite=False."""
|
||||
serial = entry.get("serial")
|
||||
if not serial:
|
||||
raise StorageError("Cannot store an entry without a 'serial'.")
|
||||
if serial in self._entries and not overwrite:
|
||||
return False
|
||||
record = dict(entry)
|
||||
record.setdefault("addedAt", datetime.now(timezone.utc).isoformat(timespec="seconds"))
|
||||
self._entries[serial] = record
|
||||
return True
|
||||
|
||||
def remove(self, serial: str) -> bool:
|
||||
return self._entries.pop(serial, None) is not None
|
||||
|
||||
def __len__(self) -> int:
|
||||
return len(self._entries)
|
||||
|
||||
|
||||
def open_vault(passphrase: str, settings: Settings, path: Path | None = None) -> Vault:
|
||||
"""Convenience: build an EncryptionManager and load (or init) the vault."""
|
||||
manager = EncryptionManager(passphrase, settings.crypto)
|
||||
return Vault(manager, path).load()
|
||||
@@ -0,0 +1,85 @@
|
||||
"""TOTP helpers: secret conversion, otpauth URL building, codes, and QR codes."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import binascii
|
||||
import hashlib
|
||||
import hmac
|
||||
import struct
|
||||
import time
|
||||
from pathlib import Path
|
||||
from urllib.parse import quote, urlencode
|
||||
|
||||
from .config import TotpConfig
|
||||
from .fileio import _harden
|
||||
|
||||
_HASHES = {"SHA1": hashlib.sha1, "SHA256": hashlib.sha256, "SHA512": hashlib.sha512}
|
||||
|
||||
|
||||
def hex_secret_to_base32(hex_secret: str) -> str:
|
||||
"""Convert a raw hex device secret to an unpadded Base32 TOTP secret."""
|
||||
try:
|
||||
secret_bytes = binascii.unhexlify(hex_secret)
|
||||
except (binascii.Error, TypeError) as exc:
|
||||
raise ValueError(f"Invalid hex secret: {exc}") from exc
|
||||
return base64.b32encode(secret_bytes).decode("ascii").rstrip("=")
|
||||
|
||||
|
||||
def build_totp_url(serial: str, base32_secret: str, totp: TotpConfig) -> str:
|
||||
"""Build an ``otpauth://totp/`` URL for the given serial and secret."""
|
||||
# Keep the issuer:account separator as a literal colon (otpauth convention).
|
||||
label = quote(f"{totp.issuer}:{serial}", safe=":")
|
||||
params = urlencode(
|
||||
{
|
||||
"secret": base32_secret,
|
||||
"issuer": totp.issuer,
|
||||
"digits": totp.digits,
|
||||
"algorithm": totp.algorithm,
|
||||
"period": totp.period,
|
||||
}
|
||||
)
|
||||
return f"otpauth://totp/{label}?{params}"
|
||||
|
||||
|
||||
def _b32_to_bytes(base32_secret: str) -> bytes:
|
||||
"""Decode a (possibly unpadded) Base32 TOTP secret to raw bytes."""
|
||||
cleaned = base32_secret.strip().replace(" ", "").upper()
|
||||
padding = "=" * (-len(cleaned) % 8)
|
||||
try:
|
||||
return base64.b32decode(cleaned + padding)
|
||||
except binascii.Error as exc:
|
||||
raise ValueError(f"Invalid Base32 secret: {exc}") from exc
|
||||
|
||||
|
||||
def current_code(base32_secret: str, totp: TotpConfig, *, at: float | None = None) -> str:
|
||||
"""Compute the current RFC 6238 TOTP code (zero-padded to ``totp.digits``)."""
|
||||
key = _b32_to_bytes(base32_secret)
|
||||
counter = int((time.time() if at is None else at) // totp.period)
|
||||
digest_fn = _HASHES.get(totp.algorithm.upper(), hashlib.sha1)
|
||||
mac = hmac.new(key, struct.pack(">Q", counter), digest_fn).digest()
|
||||
offset = mac[-1] & 0x0F
|
||||
binary = struct.unpack(">I", mac[offset : offset + 4])[0] & 0x7FFFFFFF
|
||||
return str(binary % (10**totp.digits)).zfill(totp.digits)
|
||||
|
||||
|
||||
def seconds_remaining(totp: TotpConfig, *, at: float | None = None) -> int:
|
||||
"""Seconds until the current TOTP window rolls over."""
|
||||
now = time.time() if at is None else at
|
||||
return totp.period - int(now % totp.period)
|
||||
|
||||
|
||||
def generate_qr_code(totp_url: str, out_path: Path) -> Path:
|
||||
"""Render ``totp_url`` to a PNG QR code at ``out_path`` (perm-hardened)."""
|
||||
import qrcode
|
||||
|
||||
out_path = Path(out_path)
|
||||
out_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
qr = qrcode.QRCode(error_correction=qrcode.constants.ERROR_CORRECT_L, box_size=10, border=4)
|
||||
qr.add_data(totp_url)
|
||||
qr.make(fit=True)
|
||||
img = qr.make_image(fill_color="black", back_color="white")
|
||||
img.save(str(out_path))
|
||||
_harden(out_path)
|
||||
return out_path
|
||||
@@ -0,0 +1,37 @@
|
||||
"""Shared test fixtures."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from bnet_auth_tool.config import (
|
||||
ApiConfig,
|
||||
CryptoConfig,
|
||||
Settings,
|
||||
TotpConfig,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def settings() -> Settings:
|
||||
"""Settings with cheap scrypt params so tests stay fast."""
|
||||
return Settings(
|
||||
api=ApiConfig(
|
||||
host="https://example.test",
|
||||
attach_path="/v1/authenticator",
|
||||
device_path="/v2/authenticator/device",
|
||||
sso_url="https://oauth.example.test/sso",
|
||||
client_id="test-client",
|
||||
timeout=5,
|
||||
region_prefixes=["US-", "EU-"],
|
||||
),
|
||||
totp=TotpConfig(algorithm="SHA1", digits=8, period=30, issuer="Battle.net"),
|
||||
crypto=CryptoConfig(
|
||||
kdf="scrypt",
|
||||
scrypt_n=1024, # small N for fast tests
|
||||
scrypt_r=8,
|
||||
scrypt_p=1,
|
||||
pbkdf2_iterations=1000,
|
||||
pbkdf2_legacy_iterations=100,
|
||||
),
|
||||
)
|
||||
@@ -0,0 +1,89 @@
|
||||
"""Crypto round-trips and legacy-format back-compat."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import json
|
||||
import os
|
||||
|
||||
import pytest
|
||||
from cryptography.hazmat.primitives import hashes
|
||||
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
||||
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
|
||||
|
||||
from bnet_auth_tool.crypto import EncryptionManager, looks_encrypted
|
||||
from bnet_auth_tool.errors import DecryptionError
|
||||
|
||||
SAMPLE = {"serial": "US-1234", "deviceSecret": "deadbeef", "nested": {"a": [1, 2, 3]}}
|
||||
|
||||
|
||||
def _legacy_pbkdf2_package(data: dict, passphrase: str, iterations: int, *, include_iters: bool):
|
||||
"""Reproduce the v1.x encryption package format for back-compat tests."""
|
||||
salt = os.urandom(16)
|
||||
nonce = os.urandom(12)
|
||||
kdf = PBKDF2HMAC(algorithm=hashes.SHA256(), length=32, salt=salt, iterations=iterations)
|
||||
key = kdf.derive(passphrase.encode("utf-8"))
|
||||
ciphertext = AESGCM(key).encrypt(nonce, json.dumps(data).encode("utf-8"), None)
|
||||
package = {
|
||||
"salt": base64.b64encode(salt).decode(),
|
||||
"nonce": base64.b64encode(nonce).decode(),
|
||||
"ciphertext": base64.b64encode(ciphertext).decode(),
|
||||
}
|
||||
if include_iters:
|
||||
package["kdf_iterations"] = iterations
|
||||
return json.dumps(package).encode("utf-8")
|
||||
|
||||
|
||||
def test_scrypt_round_trip(settings):
|
||||
mgr = EncryptionManager("correct horse", settings.crypto)
|
||||
blob = mgr.encrypt(SAMPLE)
|
||||
assert looks_encrypted(blob)
|
||||
assert json.loads(blob)["kdf"] == "scrypt"
|
||||
assert mgr.decrypt(blob) == SAMPLE
|
||||
|
||||
|
||||
def test_pbkdf2_round_trip(settings):
|
||||
cfg = settings.crypto
|
||||
object.__setattr__(cfg, "kdf", "pbkdf2")
|
||||
mgr = EncryptionManager("pw", cfg)
|
||||
blob = mgr.encrypt(SAMPLE)
|
||||
assert json.loads(blob)["kdf"] == "pbkdf2"
|
||||
assert mgr.decrypt(blob) == SAMPLE
|
||||
|
||||
|
||||
def test_wrong_passphrase_raises(settings):
|
||||
blob = EncryptionManager("right", settings.crypto).encrypt(SAMPLE)
|
||||
with pytest.raises(DecryptionError):
|
||||
EncryptionManager("wrong", settings.crypto).decrypt(blob)
|
||||
|
||||
|
||||
def test_legacy_with_explicit_iterations(settings):
|
||||
# A modern PBKDF2 file embeds its own count; decrypt must honour it
|
||||
# regardless of the configured default. Small count keeps the test fast.
|
||||
blob = _legacy_pbkdf2_package(SAMPLE, "pw", 7777, include_iters=True)
|
||||
mgr = EncryptionManager("pw", settings.crypto)
|
||||
assert mgr.decrypt(blob) == SAMPLE
|
||||
|
||||
|
||||
def test_legacy_missing_iterations_uses_legacy_count(settings):
|
||||
# Simulate a pre-v1.3 file: no kdf_iterations field. The manager must assume
|
||||
# the configured legacy iteration count to decrypt it.
|
||||
iters = settings.crypto.pbkdf2_legacy_iterations
|
||||
blob = _legacy_pbkdf2_package(SAMPLE, "pw", iters, include_iters=False)
|
||||
mgr = EncryptionManager("pw", settings.crypto)
|
||||
assert mgr.decrypt(blob) == SAMPLE
|
||||
|
||||
|
||||
def test_corrupt_data_raises(settings):
|
||||
with pytest.raises(DecryptionError):
|
||||
EncryptionManager("pw", settings.crypto).decrypt(b"not json")
|
||||
|
||||
|
||||
def test_passphrase_scrub(settings):
|
||||
mgr = EncryptionManager("secret", settings.crypto)
|
||||
mgr.close()
|
||||
assert all(b == 0 for b in mgr._passphrase)
|
||||
|
||||
|
||||
def test_looks_encrypted_false_for_plain():
|
||||
assert not looks_encrypted(json.dumps(SAMPLE).encode())
|
||||
@@ -0,0 +1,87 @@
|
||||
"""Legacy file discovery and migration into the vault."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
|
||||
from bnet_auth_tool.crypto import EncryptionManager
|
||||
from bnet_auth_tool.migrate import (
|
||||
discover_legacy_files,
|
||||
migrate_files,
|
||||
normalise_entry,
|
||||
)
|
||||
from bnet_auth_tool.storage import Vault
|
||||
|
||||
PLAIN = {
|
||||
"serial": "US-1111",
|
||||
"restoreCode": "AAAA-BBBB",
|
||||
"deviceSecret": "deadbeef",
|
||||
"base32Secret": "32W353Y",
|
||||
"totpUrl": "otpauth://totp/Battle.net:US-1111?secret=32W353Y",
|
||||
"timestamp": "2024-01-01T00:00:00+00:00",
|
||||
}
|
||||
|
||||
|
||||
def _vault(tmp_path, settings, passphrase="pw"):
|
||||
return Vault(EncryptionManager(passphrase, settings.crypto), tmp_path / "vault.json").load()
|
||||
|
||||
|
||||
def test_discover_matches_expected_names(tmp_path):
|
||||
(tmp_path / "battlenet_authenticator_US-1.json").write_text("{}")
|
||||
(tmp_path / "reconstructed_US-2.json").write_text("{}")
|
||||
(tmp_path / "unrelated.json").write_text("{}")
|
||||
names = {p.name for p in discover_legacy_files(tmp_path)}
|
||||
assert names == {"battlenet_authenticator_US-1.json", "reconstructed_US-2.json"}
|
||||
|
||||
|
||||
def test_normalise_derives_missing_fields(settings):
|
||||
entry = normalise_entry({"serial": "US-9", "deviceSecret": "deadbeef"}, settings)
|
||||
assert entry["base32Secret"] == "32W353Y"
|
||||
assert entry["totpUrl"].startswith("otpauth://totp/Battle.net:US-9")
|
||||
|
||||
|
||||
def test_migrate_plaintext_file(tmp_path, settings):
|
||||
f = tmp_path / "battlenet_authenticator_US-1111.json"
|
||||
f.write_text(json.dumps(PLAIN))
|
||||
vault = _vault(tmp_path, settings)
|
||||
|
||||
outcomes = migrate_files([f], vault, settings, lambda p: None)
|
||||
vault.save()
|
||||
|
||||
assert outcomes[0].status == "imported"
|
||||
assert vault.get("US-1111")["deviceSecret"] == "deadbeef"
|
||||
assert vault.get("US-1111")["addedAt"] == PLAIN["timestamp"]
|
||||
|
||||
|
||||
def test_migrate_encrypted_file(tmp_path, settings):
|
||||
f = tmp_path / "battlenet_authenticator_US-2222.json"
|
||||
blob = EncryptionManager("filepw", settings.crypto).encrypt({**PLAIN, "serial": "US-2222"})
|
||||
f.write_bytes(blob)
|
||||
vault = _vault(tmp_path, settings)
|
||||
|
||||
outcomes = migrate_files([f], vault, settings, lambda p: "filepw")
|
||||
vault.save()
|
||||
|
||||
assert outcomes[0].status == "imported"
|
||||
assert vault.get("US-2222") is not None
|
||||
|
||||
|
||||
def test_migrate_skips_without_passphrase(tmp_path, settings):
|
||||
f = tmp_path / "battlenet_authenticator_US-3333.json"
|
||||
f.write_bytes(EncryptionManager("x", settings.crypto).encrypt({**PLAIN, "serial": "US-3333"}))
|
||||
vault = _vault(tmp_path, settings)
|
||||
|
||||
outcomes = migrate_files([f], vault, settings, lambda p: None)
|
||||
assert outcomes[0].status == "skipped"
|
||||
assert len(vault) == 0
|
||||
|
||||
|
||||
def test_migrate_no_overwrite_existing(tmp_path, settings):
|
||||
f = tmp_path / "battlenet_authenticator_US-1111.json"
|
||||
f.write_text(json.dumps(PLAIN))
|
||||
vault = _vault(tmp_path, settings)
|
||||
vault.add({"serial": "US-1111", "deviceSecret": "old"})
|
||||
|
||||
outcomes = migrate_files([f], vault, settings, lambda p: None)
|
||||
assert outcomes[0].status == "skipped"
|
||||
assert vault.get("US-1111")["deviceSecret"] == "old"
|
||||
@@ -0,0 +1,75 @@
|
||||
"""Vault persistence, round-trip, and permissions."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import stat
|
||||
|
||||
import pytest
|
||||
|
||||
from bnet_auth_tool.crypto import EncryptionManager
|
||||
from bnet_auth_tool.errors import DecryptionError
|
||||
from bnet_auth_tool.storage import Vault
|
||||
|
||||
ENTRY = {
|
||||
"serial": "US-1234",
|
||||
"restoreCode": "ABCD-EFGH",
|
||||
"deviceSecret": "deadbeef",
|
||||
"base32Secret": "32W353Y",
|
||||
"totpUrl": "otpauth://totp/Battle.net:US-1234?secret=32W353Y",
|
||||
}
|
||||
|
||||
|
||||
def _vault(tmp_path, passphrase, settings):
|
||||
mgr = EncryptionManager(passphrase, settings.crypto)
|
||||
return Vault(mgr, tmp_path / "vault.json")
|
||||
|
||||
|
||||
def test_add_save_load_round_trip(tmp_path, settings):
|
||||
v = _vault(tmp_path, "pw", settings).load()
|
||||
assert v.add(ENTRY) is True
|
||||
v.save()
|
||||
|
||||
reopened = _vault(tmp_path, "pw", settings).load()
|
||||
assert reopened.serials() == ["US-1234"]
|
||||
assert reopened.get("US-1234")["deviceSecret"] == "deadbeef"
|
||||
assert "addedAt" in reopened.get("US-1234")
|
||||
|
||||
|
||||
def test_add_no_overwrite(tmp_path, settings):
|
||||
v = _vault(tmp_path, "pw", settings).load()
|
||||
v.add(ENTRY)
|
||||
assert v.add(ENTRY) is False
|
||||
assert v.add({**ENTRY, "deviceSecret": "new"}, overwrite=True) is True
|
||||
assert v.get("US-1234")["deviceSecret"] == "new"
|
||||
|
||||
|
||||
def test_remove(tmp_path, settings):
|
||||
v = _vault(tmp_path, "pw", settings).load()
|
||||
v.add(ENTRY)
|
||||
assert v.remove("US-1234") is True
|
||||
assert v.remove("US-1234") is False
|
||||
assert len(v) == 0
|
||||
|
||||
|
||||
def test_wrong_passphrase_fails_to_load(tmp_path, settings):
|
||||
v = _vault(tmp_path, "pw", settings).load()
|
||||
v.add(ENTRY)
|
||||
v.save()
|
||||
with pytest.raises(DecryptionError):
|
||||
_vault(tmp_path, "wrong", settings).load()
|
||||
|
||||
|
||||
@pytest.mark.skipif(os.name == "nt", reason="POSIX permission semantics")
|
||||
def test_vault_file_is_owner_only(tmp_path, settings):
|
||||
v = _vault(tmp_path, "pw", settings).load()
|
||||
v.add(ENTRY)
|
||||
v.save()
|
||||
mode = stat.S_IMODE(os.stat(v.path).st_mode)
|
||||
assert mode == 0o600
|
||||
|
||||
|
||||
def test_empty_vault_loads(tmp_path, settings):
|
||||
v = _vault(tmp_path, "pw", settings).load()
|
||||
assert len(v) == 0
|
||||
assert v.serials() == []
|
||||
@@ -0,0 +1,61 @@
|
||||
"""TOTP secret conversion and otpauth URL building."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from urllib.parse import parse_qs, urlparse
|
||||
|
||||
import pytest
|
||||
|
||||
from bnet_auth_tool.config import TotpConfig
|
||||
from bnet_auth_tool.totp import (
|
||||
build_totp_url,
|
||||
current_code,
|
||||
hex_secret_to_base32,
|
||||
seconds_remaining,
|
||||
)
|
||||
|
||||
# RFC 6238 Appendix B reference: ASCII secret "12345678901234567890".
|
||||
_RFC_SECRET_B32 = "GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ"
|
||||
_RFC_TOTP = TotpConfig(algorithm="SHA1", digits=8, period=30, issuer="Test")
|
||||
|
||||
|
||||
def test_hex_to_base32_known_value():
|
||||
# "deadbeef" -> bytes de ad be ef -> base32 (unpadded)
|
||||
assert hex_secret_to_base32("deadbeef") == "32W353Y"
|
||||
|
||||
|
||||
def test_hex_to_base32_rejects_bad_hex():
|
||||
with pytest.raises(ValueError):
|
||||
hex_secret_to_base32("nothex!!")
|
||||
|
||||
|
||||
def test_build_totp_url(settings):
|
||||
url = build_totp_url("US-1234", "JBSWY3DPEHPK3PXP", settings.totp)
|
||||
parsed = urlparse(url)
|
||||
assert parsed.scheme == "otpauth"
|
||||
assert parsed.netloc == "totp"
|
||||
assert parsed.path == "/Battle.net:US-1234"
|
||||
qs = parse_qs(parsed.query)
|
||||
assert qs["secret"] == ["JBSWY3DPEHPK3PXP"]
|
||||
assert qs["issuer"] == ["Battle.net"]
|
||||
assert qs["digits"] == ["8"]
|
||||
assert qs["algorithm"] == ["SHA1"]
|
||||
assert qs["period"] == ["30"]
|
||||
|
||||
|
||||
def test_current_code_rfc6238_vector():
|
||||
# RFC 6238 Appendix B: SHA1, 8 digits, T=59s -> 94287082.
|
||||
assert current_code(_RFC_SECRET_B32, _RFC_TOTP, at=59) == "94287082"
|
||||
# T=1111111109 -> 07081804.
|
||||
assert current_code(_RFC_SECRET_B32, _RFC_TOTP, at=1111111109) == "07081804"
|
||||
|
||||
|
||||
def test_current_code_accepts_unpadded_secret():
|
||||
# Battle.net secrets are stored unpadded; must still decode.
|
||||
assert current_code("32W353Y", _RFC_TOTP, at=0).isdigit()
|
||||
|
||||
|
||||
def test_seconds_remaining():
|
||||
assert seconds_remaining(_RFC_TOTP, at=0) == 30
|
||||
assert seconds_remaining(_RFC_TOTP, at=29) == 1
|
||||
assert seconds_remaining(_RFC_TOTP, at=30) == 30
|
||||
Reference in New Issue
Block a user