"""
Bitcoin Wallet Manager Module
Handles HD wallet creation, derivation, and encrypted storage in Redis.
Supports BIP39 mnemonics and BIP84 derivation paths (m/84'/0'/0'/0/x) for Native SegWit.
"""
import os
import jsonutil as json
import base64
import hashlib
import logging
from typing import Optional, Dict, Any, List
from datetime import datetime
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.exceptions import InvalidTag
from btc_networks import get_btc_network
from wallet_key_utils import ensure_master_key
logger = logging.getLogger(__name__)
BTC_WALLET_KEY_PREFIX = "btc_wallet:"
BTC_WALLET_INDEX_PREFIX = "btc_wallet_index:"
BTC_WALLET_MASTER_KEY_ENV = "BTC_WALLET_MASTER_KEY"
BTC_WALLET_MASTER_REDIS_KEY = "wallet_master_key:btc"
[docs]
class BTCWalletManager:
"""
Manages Bitcoin HD wallets with encrypted storage in Redis.
Features:
- BIP39 mnemonic generation and import
- BIP84 derivation (m/84'/0'/0'/0/x) for Native SegWit
- AES-256-GCM encryption for seeds at rest
- Per-user wallet isolation
- Address caching for derived addresses
Accepts an async Redis client via method parameters for v3 compatibility.
"""
[docs]
def __init__(self):
"""Construct a wallet manager with no master key loaded yet.
Leaves ``_master_key`` as ``None`` so the AES-256-GCM master key is fetched or
generated lazily on first use via ``_ensure_master_key`` (which needs an async
Redis client that is not available at construction time). Performs no I/O.
A single module-level singleton ``btc_wallet_manager`` is instantiated at the
bottom of this file and imported by ``tools/btc_wallet_tools.py``; this method is
invoked once at import time.
"""
self._master_key: Optional[bytes] = None
async def _ensure_master_key(self, redis_client) -> None:
"""Lazily load, or first-time generate and persist, the wallet master key.
Delegates to the shared ``ensure_master_key`` helper in ``wallet_key_utils``,
which prefers the in-memory cached key, then the ``BTC_WALLET_MASTER_KEY``
environment variable, then the Redis key ``wallet_master_key:btc``, generating
and storing a fresh key on first run. The resolved key is cached on
``self._master_key`` and is the root from which every per-wallet AES key is
derived in ``_derive_wallet_key``, so all encryption and decryption paths call
this first. Reads from and may write to Redis via the supplied client.
Called at the top of every state-touching method here (``create_wallet``,
``import_wif``, ``derive_address``, ``get_private_key``, ``get_decrypted_seed``).
Args:
redis_client: Async Redis client used to read or persist the master key.
Returns:
None.
"""
self._master_key = await ensure_master_key(
self._master_key,
redis_client,
BTC_WALLET_MASTER_REDIS_KEY,
BTC_WALLET_MASTER_KEY_ENV,
)
def _derive_wallet_key(self, user_id: str, wallet_name: str) -> bytes:
"""Derive a per-wallet 32-byte AES key from the master key via PBKDF2.
Mixes the loaded ``_master_key`` with a per-wallet salt built from the user id
and wallet name (100,000 SHA-256 PBKDF2 iterations) so that every wallet gets a
distinct encryption key and a leak of one wallet key does not compromise others.
The returned key feeds ``_encrypt``/``_decrypt`` to protect the seed/WIF at rest.
Requires ``_ensure_master_key`` to have run first so ``_master_key`` is set; does
no I/O of its own.
Called by the seed-handling methods in this class (``create_wallet``,
``import_wif``, ``derive_address``, ``get_private_key``, ``get_decrypted_seed``).
Args:
user_id (str): Owning user's id, mixed into the salt for isolation.
wallet_name (str): Wallet name, mixed into the salt.
Returns:
bytes: A 32-byte key suitable for AES-256-GCM.
"""
salt = f"btc:{user_id}:{wallet_name}".encode("utf-8")
derived = hashlib.pbkdf2_hmac(
"sha256", self._master_key, salt, iterations=100000, dklen=32
)
return derived
def _encrypt(self, key: bytes, plaintext: str) -> str:
"""Encrypt a secret string with AES-256-GCM under a per-wallet key.
Generates a fresh random 12-byte nonce, encrypts the UTF-8 plaintext with the
supplied key, and returns the nonce prepended to the ciphertext, URL-safe base64
encoded for safe storage as a Redis string value. The nonce-prepend layout is the
exact inverse of what ``_decrypt`` expects. Used to protect mnemonics and WIF keys
before they are written under the ``btc_wallet:`` Redis keys. Touches no Redis or
network state itself; only consumes OS randomness for the nonce.
Called by ``create_wallet`` and ``import_wif`` in this class.
Args:
key (bytes): 32-byte AES key from ``_derive_wallet_key``.
plaintext (str): The secret to encrypt (mnemonic or WIF).
Returns:
str: URL-safe base64 of ``nonce + ciphertext``.
"""
aesgcm = AESGCM(key)
nonce = os.urandom(12)
ciphertext = aesgcm.encrypt(nonce, plaintext.encode("utf-8"), None)
return base64.urlsafe_b64encode(nonce + ciphertext).decode("utf-8")
def _decrypt(self, key: bytes, encrypted: str) -> Optional[str]:
"""Decrypt a base64 AES-256-GCM blob produced by ``_encrypt``.
Reverses the encoding done in ``_encrypt``: base64-decodes the input, splits off
the leading 12-byte nonce, then authenticates and decrypts the remainder. GCM
verifies integrity, so a wrong key or tampered ciphertext raises ``InvalidTag``;
any failure is logged and swallowed, returning ``None`` rather than propagating,
so callers treat a bad decrypt as a missing/unrecoverable secret. Touches no Redis
or network state.
Called by ``derive_address``, ``get_private_key``, and ``get_decrypted_seed`` in
this class to recover a stored mnemonic or WIF.
Args:
key (bytes): 32-byte AES key from ``_derive_wallet_key``.
encrypted (str): URL-safe base64 of ``nonce + ciphertext``.
Returns:
Optional[str]: The decrypted UTF-8 plaintext, or ``None`` if decryption fails.
"""
try:
aesgcm = AESGCM(key)
data = base64.urlsafe_b64decode(encrypted)
nonce = data[:12]
ciphertext = data[12:]
plaintext = aesgcm.decrypt(nonce, ciphertext, None)
return plaintext.decode("utf-8")
except (InvalidTag, Exception) as e:
logger.error(f"Decryption failed: {e}")
return None
[docs]
def generate_mnemonic(self, strength: int = 128) -> str:
"""Generate a fresh BIP39 mnemonic seed phrase.
Lazily imports ``bitcoinlib.mnemonic.Mnemonic`` and produces a new random phrase
at the requested entropy strength (128 bits yields a 12-word phrase, 256 bits a
24-word phrase). The result is the human-recoverable backup that ``create_wallet``
then encrypts and stores; this method itself performs no Redis or network I/O.
Called by the ``_create_btc_wallet`` tool handler in
``tools/btc_wallet_tools.py`` when a user asks to create a new HD wallet without
supplying their own phrase.
Args:
strength (int): Entropy in bits (e.g. 128 or 256). Defaults to 128.
Returns:
str: A space-separated BIP39 mnemonic phrase.
"""
from bitcoinlib.mnemonic import Mnemonic
m = Mnemonic()
return m.generate(strength=strength)
[docs]
def validate_mnemonic(self, mnemonic: str) -> bool:
"""Check whether a string is a valid BIP39 mnemonic (wordlist plus checksum).
Lazily imports ``bitcoinlib.mnemonic.Mnemonic`` and runs its checksum/wordlist
validation, catching any exception and returning ``False`` so malformed input is
reported as invalid rather than raising. Used as a guard before a user-supplied
phrase is accepted and encrypted. Performs no Redis or network I/O.
Called internally by ``create_wallet`` and by the ``_create_btc_wallet`` tool
handler in ``tools/btc_wallet_tools.py`` to reject bad phrases early.
Args:
mnemonic (str): Candidate seed phrase to validate.
Returns:
bool: ``True`` if the phrase is a valid BIP39 mnemonic, else ``False``.
"""
from bitcoinlib.mnemonic import Mnemonic
try:
m = Mnemonic()
return m.check(mnemonic)
except Exception:
return False
[docs]
async def create_wallet(
self,
user_id: str,
wallet_name: str,
mnemonic: str,
redis_client,
network: str = "bitcoin",
) -> Dict[str, Any]:
"""Create and persist a new HD wallet from a BIP39 mnemonic.
Validates and normalizes the wallet name, ensures the master key is loaded, and
rejects duplicates by checking for an existing wallet first. It then validates the
mnemonic, resolves the network via ``get_btc_network``, and uses
``bitcoinlib.HDKey`` to derive the first Native SegWit (BIP84) bech32 receive
address at ``m/84'/<coin_type>'/0'/0/0`` (coin type ``1'`` for testnets, ``0'`` for
mainnet). The mnemonic is encrypted with a per-wallet key via ``_encrypt`` and the
wallet record is written to Redis under ``btc_wallet:<user>:<name>`` while the name
is added to the per-user index set ``btc_wallet_index:<user>``. The returned dict
deliberately excludes any secret material.
Touches Redis (``SET`` plus ``SADD``) and depends on the loaded master key; the
plaintext mnemonic exists only transiently in memory. Called by the
``_create_btc_wallet`` tool handler in ``tools/btc_wallet_tools.py``.
Args:
user_id (str): Owning user's id; namespaces the Redis keys.
wallet_name (str): 1-32 char alphanumeric name (``_`` and ``-`` allowed).
mnemonic (str): A valid BIP39 mnemonic to import as the wallet seed.
redis_client: Async Redis client for the master key and wallet storage.
network (str): Network name or alias (default ``bitcoin``).
Returns:
Dict[str, Any]: Public wallet info with ``wallet_name``, first ``address``,
display ``network``, and ``created_at``.
Raises:
ValueError: If the name is invalid, the wallet already exists, the mnemonic is
invalid, or the network is unknown.
"""
wallet_name = wallet_name.strip()
if not wallet_name or len(wallet_name) > 32:
raise ValueError("Wallet name must be 1-32 characters")
if not wallet_name.replace("_", "").replace("-", "").isalnum():
raise ValueError("Wallet name must be alphanumeric (with _ or -)")
await self._ensure_master_key(redis_client)
existing = await self.get_wallet(user_id, wallet_name, redis_client)
if existing:
raise ValueError(f"Wallet '{wallet_name}' already exists")
if not self.validate_mnemonic(mnemonic):
raise ValueError("Invalid mnemonic phrase")
net_config = get_btc_network(network)
if not net_config:
raise ValueError(f"Unknown network: {network}")
from bitcoinlib.keys import HDKey
hdkey = HDKey.from_passphrase(mnemonic, network=net_config.network_name)
coin_type = "1'" if net_config.is_testnet else "0'"
path = f"m/84'/{coin_type}/0'/0/0"
derived = hdkey.subkey_for_path(path)
address = derived.address(encoding="bech32")
encryption_key = self._derive_wallet_key(user_id, wallet_name)
encrypted_mnemonic = self._encrypt(encryption_key, mnemonic)
wallet_data = {
"wallet_name": wallet_name,
"type": "hd",
"network": net_config.network_name,
"encrypted_seed": encrypted_mnemonic,
"addresses": {"0": address},
"created_at": datetime.utcnow().isoformat(),
}
redis_key = f"{BTC_WALLET_KEY_PREFIX}{user_id}:{wallet_name}"
await redis_client.set(redis_key, json.dumps(wallet_data))
index_key = f"{BTC_WALLET_INDEX_PREFIX}{user_id}"
await redis_client.sadd(index_key, wallet_name)
return {
"wallet_name": wallet_name,
"address": address,
"network": net_config.name,
"created_at": wallet_data["created_at"],
}
[docs]
async def import_wif(
self,
user_id: str,
wallet_name: str,
wif: str,
redis_client,
network: str = "bitcoin",
) -> Dict[str, Any]:
"""Import a single private key (WIF) as a simple, non-HD wallet.
Validates and de-duplicates the wallet name, ensures the master key is loaded, and
resolves the network via ``get_btc_network``. It then parses the WIF with
``bitcoinlib.Key`` to derive the corresponding bech32 address, encrypts the WIF
with a per-wallet key via ``_encrypt``, and stores a record of ``type`` ``simple``
(one fixed address at index 0) under ``btc_wallet:<user>:<name>``, adding the name
to the per-user index set ``btc_wallet_index:<user>``. Unlike ``create_wallet`` this
wallet cannot derive further addresses.
Touches Redis (``SET`` plus ``SADD``) and depends on the loaded master key. Called
by the ``_import_btc_wallet`` tool handler in ``tools/btc_wallet_tools.py``.
Args:
user_id (str): Owning user's id; namespaces the Redis keys.
wallet_name (str): 1-32 char wallet name.
wif (str): The Wallet Import Format private key to store.
redis_client: Async Redis client for the master key and wallet storage.
network (str): Network name or alias (default ``bitcoin``).
Returns:
Dict[str, Any]: Public wallet info with ``wallet_name``, ``address``, display
``network``, and ``created_at``.
Raises:
ValueError: If the name is invalid, the wallet already exists, the network is
unknown, or the WIF key cannot be parsed.
"""
wallet_name = wallet_name.strip()
if not wallet_name or len(wallet_name) > 32:
raise ValueError("Wallet name must be 1-32 characters")
await self._ensure_master_key(redis_client)
existing = await self.get_wallet(user_id, wallet_name, redis_client)
if existing:
raise ValueError(f"Wallet '{wallet_name}' already exists")
net_config = get_btc_network(network)
if not net_config:
raise ValueError(f"Unknown network: {network}")
from bitcoinlib.keys import Key
try:
key = Key(wif, network=net_config.network_name)
address = key.address(encoding="bech32")
except Exception as e:
raise ValueError(f"Invalid WIF key: {e}")
encryption_key = self._derive_wallet_key(user_id, wallet_name)
encrypted_wif = self._encrypt(encryption_key, wif)
wallet_data = {
"wallet_name": wallet_name,
"type": "simple",
"network": net_config.network_name,
"encrypted_seed": encrypted_wif,
"addresses": {"0": address},
"created_at": datetime.utcnow().isoformat(),
}
redis_key = f"{BTC_WALLET_KEY_PREFIX}{user_id}:{wallet_name}"
await redis_client.set(redis_key, json.dumps(wallet_data))
index_key = f"{BTC_WALLET_INDEX_PREFIX}{user_id}"
await redis_client.sadd(index_key, wallet_name)
return {
"wallet_name": wallet_name,
"address": address,
"network": net_config.name,
"created_at": wallet_data["created_at"],
}
[docs]
async def get_wallet(
self, user_id: str, wallet_name: str, redis_client
) -> Optional[Dict[str, Any]]:
"""Fetch a wallet's public metadata from Redis, with the secret stripped out.
Reads the JSON record at ``btc_wallet:<user>:<name>`` and, before returning,
removes the ``encrypted_seed`` field so callers never see secret material through
this path (the decrypted seed is only reachable via the dedicated
``get_decrypted_seed``/``get_private_key`` methods). Returns ``None`` when no such
wallet key exists. Performs a single Redis ``GET``.
Called internally as a duplicate-existence check by ``create_wallet`` and
``import_wif``, by ``list_wallets`` to expand each indexed name, and by tool
handlers (``_derive_btc_address``, ``_export_btc_wallet``) in
``tools/btc_wallet_tools.py``.
Args:
user_id (str): Owning user's id.
wallet_name (str): Wallet name to look up.
redis_client: Async Redis client.
Returns:
Optional[Dict[str, Any]]: The wallet record without ``encrypted_seed``, or
``None`` if it does not exist.
"""
redis_key = f"{BTC_WALLET_KEY_PREFIX}{user_id}:{wallet_name}"
data = await redis_client.get(redis_key)
if data:
wallet = json.loads(data)
wallet.pop("encrypted_seed", None)
return wallet
return None
[docs]
async def list_wallets(self, user_id: str, redis_client) -> List[Dict[str, Any]]:
"""List all of a user's wallets as public metadata, sorted by creation time.
Reads the per-user index set ``btc_wallet_index:<user>`` to get the wallet names,
decoding any bytes members, then calls ``get_wallet`` for each to load its
secret-stripped record, skipping any that have gone missing. The combined list is
sorted by ``created_at``. Touches Redis with one ``SMEMBERS`` plus one ``GET`` per
wallet.
Called by the ``_list_btc_wallets`` tool handler in
``tools/btc_wallet_tools.py``.
Args:
user_id (str): Owning user's id.
redis_client: Async Redis client.
Returns:
List[Dict[str, Any]]: Public wallet records (no secrets), oldest first.
"""
index_key = f"{BTC_WALLET_INDEX_PREFIX}{user_id}"
wallet_names = await redis_client.smembers(index_key)
wallets = []
for name in wallet_names:
if isinstance(name, bytes):
name = name.decode("utf-8")
wallet = await self.get_wallet(user_id, name, redis_client)
if wallet:
wallets.append(wallet)
return sorted(wallets, key=lambda w: w.get("created_at", ""))
[docs]
async def derive_address(
self,
user_id: str,
wallet_name: str,
index: int,
redis_client,
address_type: str = "bech32",
) -> Optional[str]:
"""Derive (and cache) a receive address at a given index for an HD wallet.
Ensures the master key is loaded, then reads the raw wallet record from
``btc_wallet:<user>:<name>``. If the address for the ``index:address_type`` cache
key is already stored it is returned immediately. For ``simple`` (WIF-imported)
wallets only index 0 is valid and the stored address is returned. Otherwise it
decrypts the mnemonic via ``_decrypt``, rebuilds the ``bitcoinlib.HDKey``, and
derives the BIP84 subkey at ``m/84'/<coin_type>'/0'/0/<index>``, encoding as
bech32, p2sh, or legacy per ``address_type``. The newly derived address is written
back into the wallet's ``addresses`` cache in Redis so subsequent calls are cheap.
Reads and conditionally writes Redis (``GET`` then ``SET``) and depends on the
loaded master key; the mnemonic exists only transiently in memory. Called by
several tool handlers in ``tools/btc_wallet_tools.py`` (``_derive_btc_address``,
``_check_btc_balance``, ``_list_btc_utxos``, ``_send_btc``).
Args:
user_id (str): Owning user's id.
wallet_name (str): Wallet name to derive from.
index (int): Address index along the external chain.
redis_client: Async Redis client.
address_type (str): ``bech32`` (default), ``p2sh``, or legacy.
Returns:
Optional[str]: The derived address, or ``None`` if the wallet is missing or the
seed cannot be decrypted.
Raises:
ValueError: If a non-zero index is requested for a ``simple`` wallet.
"""
await self._ensure_master_key(redis_client)
redis_key = f"{BTC_WALLET_KEY_PREFIX}{user_id}:{wallet_name}"
data = await redis_client.get(redis_key)
if not data:
return None
wallet = json.loads(data)
cache_key = f"{index}:{address_type}"
if cache_key in wallet.get("addresses", {}):
return wallet["addresses"][cache_key]
if wallet["type"] == "simple":
if index != 0:
raise ValueError("Simple wallets only have one address (index 0)")
return wallet["addresses"].get("0")
encryption_key = self._derive_wallet_key(user_id, wallet_name)
mnemonic = self._decrypt(encryption_key, wallet["encrypted_seed"])
if not mnemonic:
return None
from bitcoinlib.keys import HDKey
network = wallet.get("network", "bitcoin")
net_config = get_btc_network(network)
hdkey = HDKey.from_passphrase(mnemonic, network=net_config.network_name)
coin_type = "1'" if net_config.is_testnet else "0'"
path = f"m/84'/{coin_type}/0'/0/{index}"
derived = hdkey.subkey_for_path(path)
if address_type == "bech32":
address = derived.address(encoding="bech32")
elif address_type == "p2sh":
address = derived.address(script_type="p2sh")
else:
address = derived.address()
wallet["addresses"][cache_key] = address
await redis_client.set(redis_key, json.dumps(wallet))
return address
[docs]
async def get_private_key(
self, user_id: str, wallet_name: str, redis_client, index: int = 0
) -> Optional[str]:
"""Recover the WIF private key for a wallet address, decrypting the stored seed.
Ensures the master key is loaded, reads the raw record from
``btc_wallet:<user>:<name>``, and decrypts the stored secret via ``_decrypt``. For
``simple`` wallets the decrypted value is itself the WIF and is returned directly;
for HD wallets it rebuilds the ``bitcoinlib.HDKey``, derives the BIP84 subkey at
``m/84'/<coin_type>'/0'/0/<index>``, and returns that subkey's WIF. This is the
signing-key path, so its output must never be logged or persisted by callers.
Reads Redis (``GET``) and depends on the loaded master key. Called by the
``_send_btc`` tool handler in ``tools/btc_wallet_tools.py`` to sign transactions.
Args:
user_id (str): Owning user's id.
wallet_name (str): Wallet name.
redis_client: Async Redis client.
index (int): Address index whose key is wanted (default 0).
Returns:
Optional[str]: The WIF private key, or ``None`` if the wallet is missing or the
seed cannot be decrypted.
"""
await self._ensure_master_key(redis_client)
redis_key = f"{BTC_WALLET_KEY_PREFIX}{user_id}:{wallet_name}"
data = await redis_client.get(redis_key)
if not data:
return None
wallet = json.loads(data)
encryption_key = self._derive_wallet_key(user_id, wallet_name)
seed = self._decrypt(encryption_key, wallet["encrypted_seed"])
if not seed:
return None
if wallet["type"] == "simple":
return seed
from bitcoinlib.keys import HDKey
network = wallet.get("network", "bitcoin")
net_config = get_btc_network(network)
hdkey = HDKey.from_passphrase(seed, network=net_config.network_name)
coin_type = "1'" if net_config.is_testnet else "0'"
path = f"m/84'/{coin_type}/0'/0/{index}"
derived = hdkey.subkey_for_path(path)
return derived.wif()
[docs]
async def get_decrypted_seed(
self, user_id: str, wallet_name: str, redis_client
) -> Optional[str]:
"""Decrypt and return a wallet's raw seed material (mnemonic or WIF).
Ensures the master key is loaded, reads the raw record from
``btc_wallet:<user>:<name>``, and returns the result of ``_decrypt`` over the stored
``encrypted_seed`` without any further derivation, so the caller gets the original
backup secret (a BIP39 mnemonic for HD wallets or the WIF for simple ones). This is
the most sensitive accessor and exists for explicit user-initiated export only.
Reads Redis (``GET``) and depends on the loaded master key. Called by the
``_export_btc_wallet`` tool handler in ``tools/btc_wallet_tools.py``.
Args:
user_id (str): Owning user's id.
wallet_name (str): Wallet name.
redis_client: Async Redis client.
Returns:
Optional[str]: The decrypted seed/WIF, or ``None`` if the wallet is missing or
decryption fails.
"""
await self._ensure_master_key(redis_client)
redis_key = f"{BTC_WALLET_KEY_PREFIX}{user_id}:{wallet_name}"
data = await redis_client.get(redis_key)
if not data:
return None
wallet = json.loads(data)
encryption_key = self._derive_wallet_key(user_id, wallet_name)
return self._decrypt(encryption_key, wallet["encrypted_seed"])
[docs]
async def delete_wallet(self, user_id: str, wallet_name: str, redis_client) -> bool:
"""Permanently delete a wallet and its index entry from Redis.
Checks whether the wallet key ``btc_wallet:<user>:<name>`` exists and, if so,
deletes it and removes the name from the per-user index set
``btc_wallet_index:<user>``. If the wallet was never present it short-circuits and
reports failure without touching the index. This destroys the only copy of the
encrypted seed, so the underlying funds become unrecoverable unless the user has
backed up the phrase. Touches Redis with ``EXISTS``, ``DELETE``, and ``SREM``.
Called by the ``_delete_btc_wallet`` tool handler in
``tools/btc_wallet_tools.py``.
Args:
user_id (str): Owning user's id.
wallet_name (str): Wallet name to delete.
redis_client: Async Redis client.
Returns:
bool: ``True`` if a wallet was found and removed, ``False`` if it did not exist.
"""
redis_key = f"{BTC_WALLET_KEY_PREFIX}{user_id}:{wallet_name}"
if not await redis_client.exists(redis_key):
return False
await redis_client.delete(redis_key)
index_key = f"{BTC_WALLET_INDEX_PREFIX}{user_id}"
await redis_client.srem(index_key, wallet_name)
return True
btc_wallet_manager = BTCWalletManager()