mirror of
https://github.com/Nighthawk42/bnet_auth_tool.git
synced 2026-08-30 06:22:27 +00:00
Rewrite as a packaged tool with an encrypted vault (v2.0.0)
Replace the single-file v1.3.1 script with a proper src/bnet_auth_tool/ package: config (YAML settings + platformdirs), crypto (scrypt + AES-256-GCM with a versioned header; legacy PBKDF2 100k/600k still decrypt), storage (single encrypted vault), fileio (atomic 0600 writes), api, totp, migrate, and a cli with an interactive menu plus argparse subcommands. Security/audit fixes: drop catch-all excepts, stop leaking server bodies/tokens in errors, atomic permission-hardened writes, explicit Ctrl-C handling, versioned format header, best-effort passphrase scrubbing. Add packaging (pyproject for uv, organized requirements.txt fallback, uv.lock), .gitignore, settings.yaml, CLAUDE.md/AGENTS.md, 22 pytest tests, and ruff config. Online attach/retrieve is preserved but kept labelled unverified. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Nighthawk42
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
0db5fa287e
commit
023ccd9f74
@@ -1,34 +1,55 @@
|
||||
name: "Bug Report / API Endpoint Update"
|
||||
description: "Read before opening. Account recovery requests will be closed instantly."
|
||||
title: "[API Update]: "
|
||||
labels: ["help wanted"]
|
||||
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: |
|
||||
# ⚠️ STOP AND READ THIS ⚠️
|
||||
|
||||
**Blizzard changed their API endpoints. This tool is currently broken.**
|
||||
|
||||
* **Account Recovery:** I cannot help you. You must contact Blizzard Customer Support.
|
||||
* **Fixing the Tool:** This tool will remain broken until a developer sniffs out and provides the updated Blizzard endpoints/payload configurations.
|
||||
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: Mandatory Acknowledgment
|
||||
description: You must agree to these terms before submitting.
|
||||
label: Acknowledgment
|
||||
options:
|
||||
- label: "I understand that this tool is currently broken due to Blizzard's API changes."
|
||||
- label: "I understand the maintainer CANNOT recover my Battle.net account."
|
||||
required: true
|
||||
- label: "I understand that the repository maintainer CANNOT recover my Battle.net account."
|
||||
- label: "This is not an account-recovery request."
|
||||
required: true
|
||||
|
||||
- type: textarea
|
||||
id: technical-details
|
||||
- type: dropdown
|
||||
id: area
|
||||
attributes:
|
||||
label: Updated Endpoint Data / Technical Contribution
|
||||
description: "If you have intercepted the new Blizzard API traffic, please provide the new endpoints, request headers, or payload structure here."
|
||||
placeholder: "e.g., New authentication endpoint is found at..."
|
||||
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
|
||||
|
||||
+56
@@ -0,0 +1,56 @@
|
||||
# ---------------------------------------------------------------------------
|
||||
# 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
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 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,59 @@
|
||||
# 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. |
|
||||
| `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`.
|
||||
|
||||
## 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,153 +1,124 @@
|
||||
# Battle.net Authenticator Tool
|
||||
|
||||
---
|
||||
A command-line tool for managing **Battle.net software authenticators**. 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, …).
|
||||
|
||||
# 🛑 NOT MAINTAINED / BROKEN BY BLIZZARD 🛑
|
||||
|
||||
**This repository is functional ONLY as a local backup manager. Online features are dead.**
|
||||
|
||||
### 💻 What Happened?
|
||||
Blizzard recently modified their identity API endpoints and restricted authentication scopes. This change was implemented to prevent users from extracting their raw device secrets and forcing them to use the heavy, official Battle.net mobile app for 2FA.
|
||||
|
||||
This tool was built using completely legal, public API access. However, because Blizzard has locked down these endpoints to restrict user choice, **you can no longer attach new authenticators or retrieve secrets online using this script.**
|
||||
|
||||
### 🚨 Read Before Opening an Issue:
|
||||
* **Will this be fixed?** Only if a someone manages to legally map the new endpoints or payload schemas. Pull Requests are welcome.
|
||||
* **Are you locked out of your account?** I have zero association with Blizzard and zero access to their backend. **Do not open an issue.** You must contact [Blizzard Customer Support](https://us.battle.net/support/en/) directly to have the authenticator detached from your account.
|
||||
* **Spam Policy:** Any issue opened asking for a "fix," reporting an API connection failure, or asking for account help will be locked and deleted immediately.
|
||||
> [!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.
|
||||
|
||||
---
|
||||
|
||||
### 🔍 Legacy Local Functions (What Still Works)
|
||||
If you already ran this tool in the past and have your `battlenet_authenticator_SERIAL.json` backup file, the tool is 100% operational offline:
|
||||
## What it does
|
||||
|
||||
* **Reconstruct TOTP:** Generates your standard RFC 6238 TOTP keys (8 digits, 30s period) and outputs a QR code to import into **Aegis, Bitwarden, 1Password, or Google Authenticator**.
|
||||
* **Local Security:** Encrypt or decrypt your local JSON backups using strong AES-256-GCM encryption (upgraded to 600k PBKDF2 iterations in v1.3.0).
|
||||
* **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.
|
||||
|
||||
---
|
||||
## Security model
|
||||
|
||||
## 🛠️ Offline Usage
|
||||
| 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 |
|
||||
|
||||
If you are running the tool locally to regenerate keys from an existing backup:
|
||||
> 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
|
||||
|
||||
### With [uv](https://docs.astral.sh/uv/) (recommended)
|
||||
|
||||
```bash
|
||||
pip install requests cryptography "qrcode[pil]"
|
||||
python bnet_auth_tool.py
|
||||
uv tool install . # install the `bnet-auth` command
|
||||
# or, for development:
|
||||
uv sync --extra dev
|
||||
```
|
||||
|
||||
---
|
||||
# Version History
|
||||
### With pip
|
||||
|
||||
Version: 1.3.0
|
||||
```bash
|
||||
pip install .
|
||||
# or, using the fallback dependency list:
|
||||
pip install -r requirements.txt && pip install .
|
||||
```
|
||||
|
||||
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]"
|
||||
```
|
||||
Requires **Python 3.9+**.
|
||||
|
||||
## 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.
|
||||
Run with no arguments for the interactive menu:
|
||||
|
||||
---
|
||||
```bash
|
||||
bnet-auth
|
||||
```
|
||||
|
||||
## Output Files
|
||||
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.
|
||||
|
||||
## 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
|
||||
|
||||
* **`.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.
|
||||
|
||||
## Donations
|
||||
|
||||
[](https://ko-fi.com/P5P21QRW51)
|
||||
|
||||
@@ -1,654 +0,0 @@
|
||||
import json
|
||||
import base64
|
||||
import binascii
|
||||
import sys
|
||||
import os
|
||||
import getpass
|
||||
import platform
|
||||
from pathlib import Path
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any, Dict, Optional, Tuple, List
|
||||
|
||||
try:
|
||||
import requests
|
||||
import qrcode
|
||||
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
|
||||
from cryptography.hazmat.primitives import hashes
|
||||
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
||||
from cryptography.hazmat.backends import default_backend
|
||||
from cryptography.exceptions import InvalidTag
|
||||
except ImportError as e:
|
||||
print(f"Error: Missing required library. {e}")
|
||||
print("Please install dependencies using: pip install requests qrcode[pil] cryptography")
|
||||
sys.exit(1)
|
||||
|
||||
if platform.system() == "Windows":
|
||||
import ctypes
|
||||
|
||||
class AppConfig:
|
||||
TITLE = "Battle.net Authenticator Tool"
|
||||
VERSION = "1.3.1"
|
||||
AUTHOR = "Nighthawk42"
|
||||
LICENSE = "MIT"
|
||||
GITHUB_URL = "https://github.com/Nighthawk42/bnet_auth_tool"
|
||||
|
||||
# Authenticator REST API host (GLOBAL region: US/EU/KR/PTR).
|
||||
# CN uses "https://authenticator.api.battle.net" in the official app.
|
||||
API_HOST = "https://authenticator-rest-api.bnet-identity.blizzard.net"
|
||||
# Attach stays on v1 (setupAuthenticator: POST /v1/authenticator).
|
||||
ATTACH_URL = f"{API_HOST}/v1/authenticator"
|
||||
# Retrieve/restore moved to v2 and now requires accountIdentifier
|
||||
# (restoreAuthenticator: POST /v2/authenticator/device).
|
||||
DEVICE_URL = f"{API_HOST}/v2/authenticator/device"
|
||||
SSO_URL = "https://oauth.battle.net/oauth/sso"
|
||||
CLIENT_ID = "baedda12fe054e4abdfc3ad7bdea970a"
|
||||
|
||||
LEGACY_PBKDF2_ITERATIONS = 100_000
|
||||
DEFAULT_PBKDF2_ITERATIONS = 600_000
|
||||
SALT_SIZE = 16
|
||||
NONCE_SIZE = 12
|
||||
AES_KEY_SIZE = 32
|
||||
|
||||
class AuthenticatorError(Exception):
|
||||
pass
|
||||
|
||||
class EncryptionError(Exception):
|
||||
pass
|
||||
|
||||
class DecryptionError(Exception):
|
||||
pass
|
||||
|
||||
def set_console_title(title: str = AppConfig.TITLE) -> None:
|
||||
try:
|
||||
if platform.system() == "Windows":
|
||||
ctypes.windll.kernel32.SetConsoleTitleW(title)
|
||||
else:
|
||||
sys.stdout.write(f"\x1b]2;{title}\x07")
|
||||
sys.stdout.flush()
|
||||
except Exception as e:
|
||||
print(f"Warning: Could not set console title - {e}", file=sys.stderr)
|
||||
|
||||
def graceful_exit(exit_code: int = 0) -> None:
|
||||
print("\nExiting the program. Ensure you have securely backed up your data.")
|
||||
sys.exit(exit_code)
|
||||
|
||||
def print_header() -> None:
|
||||
print(r"""
|
||||
____ _ _ _ _
|
||||
| __ ) __ _ | |_ | |_ | | ___ _ __ ___ | |_
|
||||
| _ \ / _` || __|| __|| | / _ \ | '_ \ / _ \| __|
|
||||
| |_) || (_| || |_ | |_ | || __/ _ | | | || __/| |_
|
||||
|____/ \__,_| \__| \__||_| \___|(_)|_| |_| \___| \__|
|
||||
|
||||
|
||||
_ _ _ _ _ _
|
||||
/ \ _ _ | |_ | |__ ___ _ __ | |_ (_) ___ __ _ | |_ ___ _ __
|
||||
/ _ \ | | | || __|| '_ \ / _ \| '_ \ | __|| | / __| / _` || __| / _ \ | '__|
|
||||
/ ___ \ | |_| || |_ | | | || __/| | | || |_ | || (__ | (_| || |_ | (_) || |
|
||||
/_/ \_\ \__,_| \__||_| |_| \___||_| |_| \__||_| \___| \__,_| \__| \___/ |_|
|
||||
|
||||
|
||||
_____ _
|
||||
|_ _| ___ ___ | |
|
||||
| | / _ \ / _ \ | |
|
||||
| | | (_) || (_) || |
|
||||
|_| \___/ \___/ |_|
|
||||
|
||||
""")
|
||||
print(f"{AppConfig.TITLE}")
|
||||
print(f"Version: {AppConfig.VERSION}")
|
||||
print(f"Author: {AppConfig.AUTHOR}")
|
||||
print(f"License: {AppConfig.LICENSE}")
|
||||
print(f"Github: {AppConfig.GITHUB_URL}")
|
||||
print("-" * 40)
|
||||
|
||||
class EncryptionManager:
|
||||
def __init__(self, passphrase: str):
|
||||
if not passphrase:
|
||||
raise ValueError("Passphrase cannot be empty.")
|
||||
self.passphrase = passphrase.encode('utf-8')
|
||||
self.backend = default_backend()
|
||||
self.default_iterations = AppConfig.DEFAULT_PBKDF2_ITERATIONS
|
||||
|
||||
def _derive_key(self, salt: bytes, iterations: int) -> bytes:
|
||||
kdf = PBKDF2HMAC(
|
||||
algorithm=hashes.SHA256(),
|
||||
length=AppConfig.AES_KEY_SIZE,
|
||||
salt=salt,
|
||||
iterations=iterations,
|
||||
backend=self.backend
|
||||
)
|
||||
return kdf.derive(self.passphrase)
|
||||
|
||||
def encrypt(self, data: Dict[str, Any]) -> bytes:
|
||||
try:
|
||||
json_data_bytes = json.dumps(data, ensure_ascii=False).encode('utf-8')
|
||||
salt = os.urandom(AppConfig.SALT_SIZE)
|
||||
key = self._derive_key(salt, self.default_iterations)
|
||||
aesgcm = AESGCM(key)
|
||||
nonce = os.urandom(AppConfig.NONCE_SIZE)
|
||||
ciphertext = aesgcm.encrypt(nonce, json_data_bytes, None)
|
||||
|
||||
encrypted_package = {
|
||||
'salt': base64.b64encode(salt).decode('utf-8'),
|
||||
'nonce': base64.b64encode(nonce).decode('utf-8'),
|
||||
'ciphertext': base64.b64encode(ciphertext).decode('utf-8'),
|
||||
'kdf_iterations': self.default_iterations
|
||||
}
|
||||
return json.dumps(encrypted_package, indent=4).encode('utf-8')
|
||||
except Exception as e:
|
||||
raise EncryptionError(f"Encryption failed: {e}") from e
|
||||
|
||||
def decrypt(self, encrypted_bytes: bytes) -> Dict[str, Any]:
|
||||
missing_iterations_field = False
|
||||
try:
|
||||
encrypted_data = json.loads(encrypted_bytes.decode('utf-8'))
|
||||
salt = base64.b64decode(encrypted_data['salt'])
|
||||
nonce = base64.b64decode(encrypted_data['nonce'])
|
||||
ciphertext = base64.b64decode(encrypted_data['ciphertext'])
|
||||
|
||||
if 'kdf_iterations' in encrypted_data:
|
||||
stored_iterations = int(encrypted_data['kdf_iterations'])
|
||||
else:
|
||||
missing_iterations_field = True
|
||||
stored_iterations = AppConfig.LEGACY_PBKDF2_ITERATIONS
|
||||
print(f"Warning: 'kdf_iterations' field missing. Assuming legacy count ({AppConfig.LEGACY_PBKDF2_ITERATIONS}). Re-encrypt for better security.")
|
||||
|
||||
key = self._derive_key(salt, stored_iterations)
|
||||
aesgcm = AESGCM(key)
|
||||
decrypted_data_bytes = aesgcm.decrypt(nonce, ciphertext, None)
|
||||
return json.loads(decrypted_data_bytes.decode('utf-8'))
|
||||
except InvalidTag:
|
||||
raise DecryptionError("Decryption failed: Authentication tag mismatch. Check passphrase or data integrity.")
|
||||
except (KeyError, ValueError, TypeError, binascii.Error, json.JSONDecodeError) as e:
|
||||
extra_info = " (Note: Assumed legacy KDF iterations as field was missing)." if missing_iterations_field else ""
|
||||
raise DecryptionError(f"Decryption failed: Invalid data format or content. {e}{extra_info}") from e
|
||||
except Exception as e:
|
||||
raise DecryptionError(f"An unexpected error occurred during decryption: {e}") from e
|
||||
|
||||
class BattleNetAuthenticator:
|
||||
def __init__(self):
|
||||
self.bearer_token: Optional[str] = None
|
||||
self.session = requests.Session()
|
||||
self.session.headers.update({'User-Agent': f'{AppConfig.TITLE}/{AppConfig.VERSION}'})
|
||||
|
||||
def _make_request(self, method: str, url: str, headers: Optional[Dict] = None,
|
||||
data: Optional[Any] = None, json_payload: Optional[Dict] = None) -> Dict[str, Any]:
|
||||
try:
|
||||
request_headers = self.session.headers.copy()
|
||||
if headers:
|
||||
request_headers.update(headers)
|
||||
|
||||
response = self.session.request(method, url, headers=request_headers, data=data, json=json_payload, timeout=20)
|
||||
response.raise_for_status()
|
||||
|
||||
if response.status_code == 204:
|
||||
return {}
|
||||
|
||||
content_type = response.headers.get('Content-Type', '')
|
||||
if 'application/json' in content_type:
|
||||
return response.json()
|
||||
else:
|
||||
raise AuthenticatorError(f"Unexpected content type '{content_type}' received from {url}.")
|
||||
|
||||
except requests.exceptions.HTTPError as e:
|
||||
status = e.response.status_code
|
||||
raw_body = e.response.text or ""
|
||||
|
||||
# The MFA server returns errors as JSON {errorCode, message},
|
||||
# with codes shaped like BLZBNTARA1000xxxx. Surface them when present.
|
||||
blz_detail = ""
|
||||
try:
|
||||
err_json = e.response.json()
|
||||
if isinstance(err_json, dict) and (err_json.get("errorCode") or err_json.get("message")):
|
||||
blz_detail = f" [errorCode={err_json.get('errorCode')} message={err_json.get('message')}]"
|
||||
except (ValueError, json.JSONDecodeError):
|
||||
pass
|
||||
|
||||
# Hints about the root cause, since the server contract can change.
|
||||
# If the body carries a BLZ errorCode, the route exists and processed the
|
||||
# request (e.g. 404 BLZBNTARA...312 = authenticator not found) — not a moved route.
|
||||
hint = ""
|
||||
if status == 404 and not blz_detail:
|
||||
hint = (" Hint: route not found. The endpoint may have changed version"
|
||||
" (e.g. retrieve moved from /v1/authenticator/device to /v2/authenticator/device).")
|
||||
elif status in (401, 403):
|
||||
hint = (" Hint: authorization failed (token/scope rejected). For attach, the route is"
|
||||
" identical to the official app, so a 401/403 points to server-side gating of"
|
||||
" the client_sso flow for this client.")
|
||||
|
||||
error_details = f" Server Response: {raw_body[:500]}"
|
||||
raise AuthenticatorError(
|
||||
f"HTTP error {status} from {url}.{blz_detail}{hint}{error_details}"
|
||||
) from e
|
||||
except requests.exceptions.RequestException as e:
|
||||
raise AuthenticatorError(f"Request failed for {url}: {e}") from e
|
||||
except json.JSONDecodeError as e:
|
||||
raise AuthenticatorError(f"Failed to decode JSON response from {url}: {e}") from e
|
||||
|
||||
def get_bearer_token(self, session_token: str) -> None:
|
||||
payload = {
|
||||
"client_id": AppConfig.CLIENT_ID,
|
||||
"grant_type": "client_sso",
|
||||
"scope": "auth.authenticator",
|
||||
"token": session_token,
|
||||
}
|
||||
headers = {"content-type": "application/x-www-form-urlencoded; charset=utf-8"}
|
||||
|
||||
print("Requesting Bearer Token...")
|
||||
response_data = self._make_request("POST", AppConfig.SSO_URL, headers=headers, data=payload)
|
||||
|
||||
access_token = response_data.get("access_token")
|
||||
if not access_token:
|
||||
raise AuthenticatorError("Bearer token not found in SSO response.")
|
||||
|
||||
self.session.headers['Authorization'] = f"Bearer {access_token}"
|
||||
self.bearer_token = access_token
|
||||
print("Bearer Token obtained successfully.")
|
||||
|
||||
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.")
|
||||
|
||||
print("Attempting to attach a new authenticator...")
|
||||
response_data = self._make_request("POST", AppConfig.ATTACH_URL, headers={"accept": "application/json"})
|
||||
|
||||
if response_data.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."
|
||||
)
|
||||
|
||||
if not all(key in response_data for key in ["serial", "restoreCode", "deviceSecret"]):
|
||||
raise AuthenticatorError(f"API response missing expected keys. Got: {response_data.keys()}")
|
||||
|
||||
print("Authenticator attached successfully.")
|
||||
return response_data
|
||||
|
||||
def retrieve_device_secret(self, account_identifier: str, serial: str, restore_code: str) -> Dict[str, Any]:
|
||||
# The v2/device endpoint does NOT require a bearer/SSO: it authenticates via
|
||||
# accountIdentifier + serial + restoreCode (like restoreAuthenticator in the official app).
|
||||
# v2 contract: the server now requires accountIdentifier (account email or phone)
|
||||
# in addition to serial and restoreCode. Fields are only trimmed (as in the official app).
|
||||
payload = {
|
||||
"accountIdentifier": account_identifier,
|
||||
"serial": serial,
|
||||
"restoreCode": restore_code,
|
||||
}
|
||||
url = AppConfig.DEVICE_URL
|
||||
|
||||
print(f"Attempting to retrieve secret for serial {serial}...")
|
||||
response_data = self._make_request("POST", url, json_payload=payload)
|
||||
|
||||
if response_data.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."
|
||||
)
|
||||
|
||||
if "deviceSecret" not in response_data:
|
||||
raise AuthenticatorError(f"API response missing 'deviceSecret'. Got: {response_data.keys()}")
|
||||
|
||||
print("Device secret retrieved successfully.")
|
||||
return response_data
|
||||
|
||||
@staticmethod
|
||||
def save_json(filename: str, data: Dict[str, Any], encryption_manager: Optional[EncryptionManager] = None) -> None:
|
||||
file_path = Path(filename)
|
||||
if file_path.exists():
|
||||
while True:
|
||||
try:
|
||||
overwrite = input(f"'{filename}' already exists. Overwrite? (y/n): ").strip().lower()
|
||||
if overwrite == "y": break
|
||||
if overwrite == "n": print("Data not saved."); return
|
||||
print("Invalid input.")
|
||||
except EOFError:
|
||||
print("\nOperation cancelled."); return
|
||||
|
||||
try:
|
||||
if encryption_manager:
|
||||
encrypted_data = encryption_manager.encrypt(data)
|
||||
file_path.write_bytes(encrypted_data)
|
||||
print(f"Encrypted data saved to '{filename}'.")
|
||||
else:
|
||||
file_path.write_text(json.dumps(data, indent=4, ensure_ascii=False), encoding='utf-8')
|
||||
print(f"Data successfully saved to '{filename}'.")
|
||||
print("IMPORTANT: Securely back up this file and your passphrase if encrypted!")
|
||||
except IOError as e:
|
||||
raise IOError(f"Failed to write to '{filename}': {e}") from e
|
||||
|
||||
@staticmethod
|
||||
def load_json(filename: str, encryption_manager: Optional[EncryptionManager] = None) -> Dict[str, Any]:
|
||||
file_path = Path(filename)
|
||||
if not file_path.is_file():
|
||||
raise FileNotFoundError(f"File not found: '{filename}'")
|
||||
|
||||
try:
|
||||
if encryption_manager:
|
||||
encrypted_bytes = file_path.read_bytes()
|
||||
data = encryption_manager.decrypt(encrypted_bytes)
|
||||
print(f"Decrypted data loaded from '{filename}'.")
|
||||
return data
|
||||
else:
|
||||
data = json.loads(file_path.read_text(encoding='utf-8'))
|
||||
print(f"Data loaded from '{filename}'.")
|
||||
return data
|
||||
except IOError as e:
|
||||
raise IOError(f"Failed to read from '{filename}': {e}") from e
|
||||
|
||||
@staticmethod
|
||||
def convert_secret_to_base32(hex_secret: str) -> str:
|
||||
try:
|
||||
secret_bytes = binascii.unhexlify(hex_secret)
|
||||
return base64.b32encode(secret_bytes).decode("utf-8").rstrip("=")
|
||||
except (binascii.Error, TypeError) as e:
|
||||
raise ValueError(f"Failed to convert secret to Base32: Invalid hex input. ({e})") from e
|
||||
|
||||
@staticmethod
|
||||
def generate_qr_code(totp_url: str, filename_base: str) -> None:
|
||||
filename = f"{filename_base}.png"
|
||||
try:
|
||||
print(f"Generating QR code '{filename}'...")
|
||||
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(filename)
|
||||
print(f"QR code saved successfully as '{filename}'.")
|
||||
except IOError as e:
|
||||
raise IOError(f"Failed to save QR code image to '{filename}': {e}") from e
|
||||
except Exception as e:
|
||||
raise Exception(f"Error generating QR code image: {e}") from e
|
||||
|
||||
def _prompt_for_encryption() -> bool:
|
||||
print("\nEncryption adds a layer of security. You MUST remember your passphrase.")
|
||||
while True:
|
||||
try:
|
||||
choice = input("Encrypt the saved JSON file? (y/n): ").strip().lower()
|
||||
if choice in ['y', 'n']: return choice == 'y'
|
||||
print("Invalid input.")
|
||||
except (EOFError, KeyboardInterrupt):
|
||||
print("\nOperation cancelled.")
|
||||
return False
|
||||
|
||||
def _prompt_for_passphrase(prompt_message: str = "Enter encryption passphrase: ") -> Optional[EncryptionManager]:
|
||||
while True:
|
||||
try:
|
||||
passphrase = getpass.getpass(prompt_message)
|
||||
if not passphrase:
|
||||
print("Passphrase cannot be empty.")
|
||||
continue
|
||||
if passphrase == getpass.getpass("Confirm passphrase: "):
|
||||
return EncryptionManager(passphrase)
|
||||
else:
|
||||
print("Passphrases do not match.")
|
||||
except (EOFError, KeyboardInterrupt):
|
||||
print("\nOperation cancelled.")
|
||||
return None
|
||||
|
||||
def _get_session_token() -> Optional[str]:
|
||||
print("\n--- How to Get the Session Token ---")
|
||||
print("1. In a private browser window, navigate to: https://account.battle.net/login/en/?ref=localhost")
|
||||
print("2. Log in. You will land on an expected 'Page Not Found' on 'localhost'.")
|
||||
print("3. From the URL, copy the token value that looks like `ST=XX-...` (e.g., 'US-abc...').")
|
||||
print("-" * 36)
|
||||
try:
|
||||
session_token = input("Enter your Session Token (or 'exit'): ").strip()
|
||||
if session_token.lower() == "exit": return None
|
||||
if not session_token:
|
||||
print("Error: Session Token cannot be empty.")
|
||||
return None
|
||||
if not any(session_token.startswith(p) for p in ["US-", "EU-", "KR-", "TW-", "CN-"]) or len(session_token) < 20:
|
||||
print("Warning: Token format looks unusual. Ensure you copied the full value.")
|
||||
return session_token
|
||||
except (EOFError, KeyboardInterrupt):
|
||||
print("\nOperation cancelled.")
|
||||
return None
|
||||
|
||||
def _process_and_save_results(authenticator: BattleNetAuthenticator, device_info: Dict[str, Any], encryption_manager: Optional[EncryptionManager]) -> None:
|
||||
serial = device_info.get("serial")
|
||||
restore_code = device_info.get("restoreCode")
|
||||
device_secret = device_info.get("deviceSecret")
|
||||
|
||||
if not all([serial, restore_code, device_secret]):
|
||||
raise ValueError("Incomplete device information from API.")
|
||||
|
||||
print("\n" + "-" * 30)
|
||||
print("Authenticator Details:")
|
||||
print(f" Serial: {serial}")
|
||||
print(f" Restore Code: {restore_code}")
|
||||
print("-" * 30)
|
||||
|
||||
print("Generating TOTP Information...")
|
||||
base32_secret = authenticator.convert_secret_to_base32(device_secret)
|
||||
label = f"Battle.net:{serial}"
|
||||
totp_url = f"otpauth://totp/{label}?secret={base32_secret}&issuer=Battle.net&digits=8&algorithm=SHA1&period=30"
|
||||
|
||||
print("\n--- TOTP Key Details ---")
|
||||
print(f"Base32 Secret: {base32_secret}")
|
||||
print(f"TOTP URL: {totp_url}")
|
||||
print("\nApp Settings: Type=TOTP, Algorithm=SHA1, Digits=8, Period=30s")
|
||||
print("-" * 24)
|
||||
|
||||
data_to_save = {
|
||||
"serial": serial,
|
||||
"restoreCode": restore_code,
|
||||
"deviceSecret": device_secret,
|
||||
"base32Secret": base32_secret,
|
||||
"totpUrl": totp_url,
|
||||
"timestamp": datetime.now(timezone.utc).isoformat(timespec='seconds')
|
||||
}
|
||||
|
||||
filename_base = f"battlenet_authenticator_{serial}"
|
||||
json_filename = f"{filename_base}.json"
|
||||
|
||||
authenticator.save_json(json_filename, data_to_save, encryption_manager)
|
||||
authenticator.generate_qr_code(totp_url, filename_base)
|
||||
|
||||
def _handle_attach_action(authenticator: BattleNetAuthenticator) -> None:
|
||||
session_token = _get_session_token()
|
||||
if not session_token: return
|
||||
|
||||
encryption_manager = None
|
||||
if _prompt_for_encryption():
|
||||
encryption_manager = _prompt_for_passphrase("Enter passphrase to encrypt new file: ")
|
||||
if not encryption_manager: return
|
||||
|
||||
try:
|
||||
authenticator.get_bearer_token(session_token)
|
||||
device_info = authenticator.attach_authenticator()
|
||||
_process_and_save_results(authenticator, device_info, encryption_manager)
|
||||
except (AuthenticatorError, EncryptionError, IOError, ValueError, Exception) as e:
|
||||
print(f"\nError during attach process: {e}", file=sys.stderr)
|
||||
|
||||
def _handle_retrieve_action(authenticator: BattleNetAuthenticator) -> None:
|
||||
# Retrieve via v2/device needs no session token / bearer: the server
|
||||
# authenticates via accountIdentifier + serial + restoreCode.
|
||||
encryption_manager = None
|
||||
if _prompt_for_encryption():
|
||||
encryption_manager = _prompt_for_passphrase("Enter passphrase to encrypt retrieved file: ")
|
||||
if not encryption_manager: return
|
||||
|
||||
try:
|
||||
# The v2 endpoint requires the account identifier (email or phone), besides serial/restoreCode.
|
||||
account_identifier = input("Enter your account email or phone number: ").strip()
|
||||
serial = input("Enter the Authenticator Serial number: ").strip()
|
||||
restore_code = input("Enter the Authenticator Restore Code: ").strip()
|
||||
if not account_identifier or not serial or not restore_code:
|
||||
print("Error: Account identifier, Serial and Restore Code are required.")
|
||||
return
|
||||
|
||||
retrieved_info = authenticator.retrieve_device_secret(account_identifier, serial, restore_code)
|
||||
device_info = {"serial": serial, "restoreCode": restore_code, "deviceSecret": retrieved_info["deviceSecret"]}
|
||||
_process_and_save_results(authenticator, device_info, encryption_manager)
|
||||
except (AuthenticatorError, EncryptionError, IOError, ValueError, Exception) as e:
|
||||
print(f"\nError during retrieve process: {e}", file=sys.stderr)
|
||||
|
||||
def _select_json_file(prompt: str) -> Optional[Path]:
|
||||
json_files = sorted([p for p in Path('.').glob('*.json') if p.is_file()])
|
||||
if not json_files:
|
||||
print("No JSON files found in the current directory.")
|
||||
return None
|
||||
|
||||
print("\nFound the following JSON files:")
|
||||
for i, file in enumerate(json_files, 1):
|
||||
print(f"{i}. {file.name}")
|
||||
|
||||
while True:
|
||||
try:
|
||||
choice = input(f"{prompt} (number or 'c' to cancel): ").strip().lower()
|
||||
if choice == 'c': return None
|
||||
index = int(choice) - 1
|
||||
if 0 <= index < len(json_files): return json_files[index]
|
||||
else: print(f"Invalid selection. Enter a number between 1 and {len(json_files)}.")
|
||||
except (ValueError, EOFError, KeyboardInterrupt):
|
||||
print("\nInvalid input or operation cancelled.")
|
||||
return None
|
||||
|
||||
def _is_file_likely_encrypted(file_path: Path) -> bool:
|
||||
try:
|
||||
content = file_path.read_text(encoding='utf-8', errors='ignore')
|
||||
data = json.loads(content[:1024])
|
||||
return isinstance(data, dict) and all(k in data for k in ('salt', 'nonce', 'ciphertext'))
|
||||
except (IOError, json.JSONDecodeError, ValueError):
|
||||
return False
|
||||
|
||||
def _handle_reconstruct_action(authenticator: BattleNetAuthenticator) -> None:
|
||||
selected_file = _select_json_file("Select JSON file to reconstruct from")
|
||||
if not selected_file: return
|
||||
|
||||
encryption_manager: Optional[EncryptionManager] = None
|
||||
try:
|
||||
if _is_file_likely_encrypted(selected_file):
|
||||
print(f"File '{selected_file.name}' appears to be encrypted.")
|
||||
encryption_manager = _prompt_for_passphrase(f"Enter passphrase for '{selected_file.name}': ")
|
||||
if not encryption_manager: return
|
||||
|
||||
data = authenticator.load_json(str(selected_file), encryption_manager)
|
||||
except (FileNotFoundError, IOError, DecryptionError, json.JSONDecodeError) as e:
|
||||
print(f"\nError loading file: {e}", file=sys.stderr)
|
||||
return
|
||||
|
||||
serial = data.get("serial")
|
||||
if not serial:
|
||||
print("Could not find 'serial' in the JSON file.")
|
||||
return
|
||||
|
||||
totp_url = data.get("totpUrl")
|
||||
if not totp_url:
|
||||
base32_secret = data.get("base32Secret") or authenticator.convert_secret_to_base32(data.get("deviceSecret", ""))
|
||||
if not base32_secret:
|
||||
print("Error: Could not find or derive a secret from the JSON file.")
|
||||
return
|
||||
label = f"Battle.net:{serial}"
|
||||
totp_url = f"otpauth://totp/{label}?secret={base32_secret}&issuer=Battle.net&digits=8&algorithm=SHA1&period=30"
|
||||
print(f"Reconstructed TOTP URL: {totp_url}")
|
||||
|
||||
print("\n--- Reconstructed TOTP Details ---")
|
||||
print(f"URL: {totp_url}")
|
||||
print("App Settings: Algorithm=SHA1, Digits=8, Period=30s")
|
||||
|
||||
try:
|
||||
authenticator.generate_qr_code(totp_url, f"reconstructed_{serial}")
|
||||
except (IOError, Exception) as e:
|
||||
print(f"Error generating QR code: {e}", file=sys.stderr)
|
||||
|
||||
input("\nPress Enter to return to the main menu...")
|
||||
|
||||
def _handle_encrypt_files_action(authenticator: BattleNetAuthenticator) -> None:
|
||||
json_files = sorted([p for p in Path('.').glob('*.json') if p.is_file()])
|
||||
if not json_files:
|
||||
print("No JSON files found to encrypt."); return
|
||||
|
||||
plain_files = []
|
||||
print("\nChecking JSON files:")
|
||||
for file_path in json_files:
|
||||
if not _is_file_likely_encrypted(file_path):
|
||||
try:
|
||||
json.loads(file_path.read_text(encoding='utf-8'))
|
||||
plain_files.append(file_path)
|
||||
print(f" - {file_path.name} (Plain Text)")
|
||||
except (json.JSONDecodeError, IOError):
|
||||
print(f" - {file_path.name} (Not a valid plain JSON, skipping)")
|
||||
else:
|
||||
print(f" - {file_path.name} (Already Encrypted)")
|
||||
|
||||
if not plain_files:
|
||||
print("\nNo plain text JSON files found to encrypt."); return
|
||||
|
||||
encryption_manager = _prompt_for_passphrase("Enter passphrase for encryption: ")
|
||||
if not encryption_manager: return
|
||||
|
||||
success, fail = 0, 0
|
||||
for file_path in plain_files:
|
||||
print(f"\nEncrypting '{file_path.name}'...")
|
||||
try:
|
||||
plain_data = json.loads(file_path.read_text(encoding='utf-8'))
|
||||
authenticator.save_json(str(file_path), plain_data, encryption_manager)
|
||||
success += 1
|
||||
except (IOError, EncryptionError, json.JSONDecodeError) as e:
|
||||
print(f"Error encrypting '{file_path.name}': {e}", file=sys.stderr)
|
||||
fail += 1
|
||||
print(f"\nEncryption complete. {success} succeeded, {fail} failed.")
|
||||
|
||||
def _handle_decrypt_file_action(authenticator: BattleNetAuthenticator) -> None:
|
||||
selected_file = _select_json_file("Select JSON file to decrypt")
|
||||
if not selected_file: return
|
||||
|
||||
if not _is_file_likely_encrypted(selected_file):
|
||||
print(f"Warning: File '{selected_file.name}' may not be encrypted. Proceeding anyway.")
|
||||
|
||||
encryption_manager = _prompt_for_passphrase(f"Enter passphrase for '{selected_file.name}': ")
|
||||
if not encryption_manager: return
|
||||
|
||||
try:
|
||||
decrypted_data = authenticator.load_json(str(selected_file), encryption_manager)
|
||||
print("\nDecryption successful.")
|
||||
print(json.dumps(decrypted_data, indent=4, ensure_ascii=False))
|
||||
|
||||
if input("\nSave decrypted data to a new file? (y/n): ").strip().lower() == 'y':
|
||||
new_filename = input("Enter new filename (e.g., decrypted.json): ").strip()
|
||||
if new_filename:
|
||||
authenticator.save_json(new_filename, decrypted_data, None)
|
||||
else:
|
||||
print("Invalid filename. Save cancelled.")
|
||||
|
||||
except (IOError, DecryptionError, json.JSONDecodeError) as e:
|
||||
print(f"\nError during decryption: {e}", file=sys.stderr)
|
||||
|
||||
def interactive_cli() -> None:
|
||||
set_console_title()
|
||||
print_header()
|
||||
authenticator = BattleNetAuthenticator()
|
||||
|
||||
actions = {
|
||||
"1": ("Attach a new authenticator", _handle_attach_action),
|
||||
"2": ("Retrieve existing device secret", _handle_retrieve_action),
|
||||
"3": ("Reconstruct TOTP from JSON", _handle_reconstruct_action),
|
||||
"4": ("Encrypt existing plain JSON file(s)", _handle_encrypt_files_action),
|
||||
"5": ("Decrypt an encrypted JSON file", _handle_decrypt_file_action),
|
||||
"6": ("Exit", lambda _: graceful_exit()),
|
||||
}
|
||||
|
||||
while True:
|
||||
print("\nChoose an action:")
|
||||
for key, (desc, _) in actions.items():
|
||||
print(f"{key}. {desc}")
|
||||
|
||||
try:
|
||||
choice = input("Enter your choice: ").strip()
|
||||
if choice in actions:
|
||||
actions[choice][1](authenticator)
|
||||
else:
|
||||
print("Invalid choice.")
|
||||
except (EOFError, KeyboardInterrupt):
|
||||
graceful_exit()
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
interactive_cli()
|
||||
except Exception as e:
|
||||
print(f"\nFATAL ERROR: An unhandled exception occurred: {e}", file=sys.stderr)
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
graceful_exit(1)
|
||||
@@ -0,0 +1,61 @@
|
||||
[build-system]
|
||||
requires = ["hatchling"]
|
||||
build-backend = "hatchling.build"
|
||||
|
||||
[project]
|
||||
name = "bnet-auth-tool"
|
||||
version = "2.0.0"
|
||||
description = "Manage Battle.net software authenticators: 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]
|
||||
dev = [
|
||||
"pytest>=8.0",
|
||||
"ruff>=0.5",
|
||||
]
|
||||
|
||||
[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"
|
||||
|
||||
[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.0"
|
||||
__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,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,52 @@
|
||||
"""TOTP helpers: secret conversion, otpauth URL building, and QR codes."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import binascii
|
||||
from pathlib import Path
|
||||
from urllib.parse import quote, urlencode
|
||||
|
||||
from .config import TotpConfig
|
||||
from .fileio import _harden
|
||||
|
||||
|
||||
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 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,33 @@
|
||||
"""TOTP secret conversion and otpauth URL building."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from urllib.parse import parse_qs, urlparse
|
||||
|
||||
import pytest
|
||||
|
||||
from bnet_auth_tool.totp import build_totp_url, hex_secret_to_base32
|
||||
|
||||
|
||||
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"]
|
||||
Reference in New Issue
Block a user