src/services/wallet.py

"""HD wallet address derivation — public-only (xpub).

The tracker only ever holds the EXTENDED PUBLIC KEY (xpub), never the seed.
From the xpub we derive a unique deposit address per user (m/44'/60'/0'/0/<i>).
Anyone with the xpub can generate addresses and watch balances, but CANNOT
spend funds — spending requires the 12-word seed that only you hold offline.

Network: works for both Ethereum mainnet and Base (both EVM, same keys).
"""
from __future__ import annotations

import os

from bip32 import BIP32
from eth_keys import keys
from eth_utils import keccak, to_checksum_address

# The account-level xpub (m/44'/60'/0') loaded from the WALLET_XPUB env var.
# Set as an HF Space secret. Public-only — safe to store server-side.
_XPUB: str = os.environ.get("WALLET_XPUB", "")

# The BIP32 object, lazily built once and cached (parsing is cheap but why repeat it).
_BIP: BIP32 | None = None

# Standard BIP44 external-chain prefix: addresses live at m/0/<index> relative to
# the account-level xpub (which already encodes m/44'/60'/0').
_CHAIN = "m/0"


def is_configured() -> bool:
    """True if WALLET_XPUB is set and parseable as a valid xpub."""
    if not _XPUB:
        return False
    try:
        _load()
        return True
    except Exception:
        return False


def _load() -> BIP32:
    """Parse and cache the xpub. Raises if missing/invalid."""
    global _BIP
    if _BIP is not None:
        return _BIP
    if not _XPUB:
        raise RuntimeError("WALLET_XPUB is not set")
    _BIP = BIP32.from_xpub(_XPUB)
    return _BIP


def derive_deposit_address(index: int) -> str:
    """Derive the checksummed deposit address for user <index>.

    index = the user's permanent slot (0, 1, 2, ...). Each index maps to one
    unique address that never changes for that user.
    """
    bip = _load()
    pub_compressed = bip.get_pubkey_from_path(f"{_CHAIN}/{index}")
    # eth_keys gives us the 64-byte (x||y) form keccak expects
    raw = keys.PublicKey.from_compressed_bytes(pub_compressed).to_bytes()
    return to_checksum_address("0x" + keccak(raw).hex()[-40:])


def derive_many(start: int, count: int) -> list[str]:
    """Derive <count> consecutive addresses starting at <start>."""
    return [derive_deposit_address(start + i) for i in range(count)]