diff --git a/app/wallet/__init__.py b/app/wallet/__init__.py new file mode 100644 index 0000000..4dd3433 --- /dev/null +++ b/app/wallet/__init__.py @@ -0,0 +1,25 @@ +"""Wallet subsystem. + +Phase 3B of AUDIT-2026-Q3.md. + +Re-exports the canonical public API of WalletManagerV2. The bulk of the +implementation lives in app.wallet.manager (moved verbatim from +app.wallet_manager_v2 on 2026-07-07). +""" +from app.wallet.manager import ( # noqa: F401 + ChainMeta, + PaymentRecord, + PaymentType, + ShamirSecretSharing, + WalletEncryption, + WalletManagerFacade, + WalletManagerV2, + WalletPurpose, + WalletRecord, + WalletRotationSchedule, + WalletStatus, + WalletTier, + ZKAddressVerifier, + get_wallet_manager_v2, + import_legacy_wallets, +) diff --git a/app/wallet/manager.py b/app/wallet/manager.py new file mode 100644 index 0000000..86c8c9b --- /dev/null +++ b/app/wallet/manager.py @@ -0,0 +1,2378 @@ +""" +RMI Wallet Manager v2 - Enterprise Multi-Chain Wallet Management +=============================================================== +A production-grade wallet management system for crypto intelligence platforms. + +Chains: 95+ across EVM, Solana, Monero, Cosmos, Polkadot, Cardano, Algorand, +Stellar, Ripple, Near, Aptos, Sui, TRON, Bitcoin variants, Litecoin, Dogecoin, +Dash, Zcash, Tezos, Nano, TON, and more. + +Features: + • HD Wallet Generation (BIP39/BIP44/BIP49/BIP84) - 95+ chains via bip_utils + • Monero Support - ed25519-blake2b keygen + stealth address derivation + • Chain Registry - metadata, validation, derivation paths per chain + • Key Rotation & Expiry - automatic scheduled rotation + • Multi-signature Support - threshold signatures for high-value wallets + • Payment Integration - x402, premium subscriptions, marketplace + • Balance Monitoring - real-time balance tracking across all chains + • Transaction History - unified tx view with categorization + • Alert System - low balance, large tx, suspicious activity alerts + • Wallet Labels & Organization - tags, groups, notes + • Cold/Hot Wallet Separation - security tier management + • API Access - programmatic wallet management for bots/agents + • Audit Trail - complete history of all wallet operations + • Backup & Recovery - Shamir's Secret Sharing for mnemonic recovery + • ZK Ownership Proof - prove wallet ownership without exposing private key + +Security: + - AES-256-GCM encryption with Argon2id key derivation (memory-hardened) + - Shamir's Secret Sharing (M-of-N) for backup mnemonic recovery + - HSM-compatible key storage interface + - Rate-limited wallet operations + - IP-restricted access for sensitive operations + - Private keys NEVER stored in plaintext - encrypted at rest, cleared from memory + - ZK address ownership verification (no private key exposure) + - GPG vault integration for master encryption password + +Author: RMI Development +Date: 2026-06-02 +""" + +import base64 +import hashlib +import json +import logging +import os +import secrets +import time +from dataclasses import asdict, dataclass, field +from datetime import UTC, datetime, timedelta +from enum import StrEnum +from typing import Any + +logger = logging.getLogger("wallet_manager_v2") + +# ── Crypto Imports ─────────────────────────────────────────────── + +try: + from cryptography.hazmat.primitives import hashes # noqa: F401 + from cryptography.hazmat.primitives.ciphers.aead import AESGCM + from cryptography.hazmat.primitives.kdf.argon2 import Argon2id + from cryptography.hazmat.primitives.kdf.hkdf import HKDF # noqa: F401 + + _HAS_CRYPTO = True +except ImportError: + _HAS_CRYPTO = False + logger.warning("cryptography not installed - encryption disabled") + +try: + from bip_utils import ( + Bip32Ed25519Slip, # noqa: F401 + Bip32Secp256k1, # noqa: F401 + Bip39MnemonicGenerator, + Bip39SeedGenerator, + Bip39WordsNum, + Bip44, + Bip44Coins, + Bip49, + Bip49Coins, + Bip84, + Bip84Coins, + Monero, # noqa: F401 + MoneroCoins, # noqa: F401 + ) + + _HAS_BIP_UTILS = True +except ImportError: + _HAS_BIP_UTILS = False + logger.warning("bip_utils not installed - using fallback generation") + +try: + import base58 + + _HAS_BASE58 = True +except ImportError: + _HAS_BASE58 = False + +try: + from nacl.bindings import crypto_sign_ed25519_sk_to_seed # noqa: F401 + from nacl.signing import SigningKey as NaClSigningKey + + _HAS_NACL = True +except ImportError: + _HAS_NACL = False + +try: + import monero # noqa: F401 + from monero import ed25519 as monero_ed25519 # noqa: F401 + from monero.address import Address as MoneroAddress # noqa: F401 + from monero.seed import Seed as MoneroSeed # noqa: F401 + + _HAS_MONERO = True +except ImportError: + _HAS_MONERO = False + logger.info("monero-python not fully available - basic XMR support only") + +# ── Enums ───────────────────────────────────────────────────────── + + +class WalletStatus(StrEnum): + ACTIVE = "active" + ROTATED = "rotated" + FROZEN = "frozen" + EXPIRED = "expired" + ARCHIVED = "archived" + COMPROMISED = "compromised" + + +class WalletTier(StrEnum): + HOT = "hot" + WARM = "warm" + COLD = "cold" + VAULT = "vault" + + +class WalletPurpose(StrEnum): + PAYMENTS = "payments" + SUBSCRIPTIONS = "subscriptions" + OPERATIONS = "operations" + TREASURY = "treasury" + USER_ESCROW = "user_escrow" + BOT_TRADING = "bot_trading" + AIRDROPS = "airdrops" + DEVELOPER = "developer" + MARKETING = "marketing" + RESERVE = "reserve" + + +class PaymentType(StrEnum): + X402 = "x402" + SUBSCRIPTION = "subscription" + ONE_TIME = "one_time" + MARKETPLACE = "marketplace" + REFUND = "refund" + WITHDRAWAL = "withdrawal" + DEPOSIT = "deposit" + FEE = "fee" + REWARD = "reward" + + +# ── Chain Registry ──────────────────────────────────────────────── +# Maps chain keys to full metadata for 20+ major non-EVM chains. +# EVM chains all share the same key derivation (secp256k1) and address +# format (0x...), so they're grouped under a single "evm" entry with +# sub-chains listed separately. + + +@dataclass +class ChainMeta: + """Metadata for a single blockchain.""" + + key: str # Short key (e.g. "sol", "xmr") + name: str # Human name (e.g. "Solana") + native_symbol: str # Native currency ticker + native_decimals: int = 18 + coin_class: Any = None # bip_utils coin class (Bip44Coins.*) + derivation: str = "bip44" # bip44, bip49, bip84, ed25519_slip, monero + address_prefix: str = "" # For address validation regex + address_pattern: str = "" # Python regex for address format + testnet_coin_class: Any = None + curve: str = "secp256k1" # secp256k1, ed25519, ed25519_blake2b + hd_path_template: str = "m/44'/{coin_type}'/{account}'/{change}/{index}" + sub_chains: list[str] = field(default_factory=list) # For EVM grouping + icon: str = "₿" # Display icon + rpc_endpoint: str = "" # Default RPC (env-overridable) + explorer_url: str = "" # Block explorer base URL + min_confirmation_blocks: int = 1 + supports_tokens: bool = True + supports_nfts: bool = False + + +# ── Complete Chain Registry ─────────────────────────────────────── + +CHAIN_REGISTRY: dict[str, ChainMeta] = {} + + +def _build_registry(): + """Build the comprehensive chain registry.""" + reg = {} + + # ── EVM Group (9 chains, all secp256k1, all 0x addresses) ── + evm_chains = [ + ("eth", "Ethereum", 60, Bip44Coins.ETHEREUM), + ("base", "Base", 8453, None), # Base uses ETH derivation + ("polygon", "Polygon", 137, Bip44Coins.POLYGON), + ("arbitrum", "Arbitrum One", 42161, Bip44Coins.ARBITRUM), + ("optimism", "Optimism", 10, Bip44Coins.OPTIMISM), + ("avalanche", "Avalanche C-Chain", 43114, Bip44Coins.AVAX_C_CHAIN), + ("bsc", "BNB Chain", 56, Bip44Coins.BINANCE_SMART_CHAIN), + ("fantom", "Fantom", 250, Bip44Coins.FANTOM_OPERA), + ("gnosis", "Gnosis", 100, Bip44Coins.ETHEREUM), # Gnosis uses ETH derivation + ] + evm_subs = [c[0] for c in evm_chains] + for key, name, _chain_id, coin_class in evm_chains: + reg[key] = ChainMeta( + key=key, + name=name, + native_symbol="ETH" if key != "bsc" else "BNB", + native_decimals=18, + derivation="bip44", + coin_class=coin_class if coin_class else Bip44Coins.ETHEREUM, + address_pattern=r"^0x[a-fA-F0-9]{40}$", + curve="secp256k1", + hd_path_template="m/44'/{coin_type}'/{account}'/{change}/{index}", + sub_chains=evm_subs, + icon="⟠", + explorer_url=f"https://{'etherscan.io' if key == 'eth' else key + ('scan.com' if key in ['bsc', 'polygon'] else '.blockscout.com')}", + ) + + # ── Solana ── + reg["sol"] = ChainMeta( + key="sol", + name="Solana", + native_symbol="SOL", + native_decimals=9, + coin_class=Bip44Coins.SOLANA, + derivation="bip44_ed25519", + address_pattern=r"^[1-9A-HJ-NP-Za-km-z]{32,44}$", + curve="ed25519", + hd_path_template="m/44'/501'/{account}'/{change}'/{index}", + icon="◎", + explorer_url="https://solscan.io", + supports_nfts=True, + ) + + # ── Monero ── + reg["xmr"] = ChainMeta( + key="xmr", + name="Monero", + native_symbol="XMR", + native_decimals=12, + coin_class=None, # Monero uses custom crypto + derivation="monero", + address_pattern=r"^[48][0-9AB][1-9A-HJ-NP-Za-km-z]{93}$", + curve="ed25519_blake2b", + hd_path_template="m/44'/128'/{account}'/{change}/{index}", + icon="ɱ", + explorer_url="https://xmrchain.net", + supports_tokens=False, + ) + + # ── Cosmos / IBC Ecosystem ── + cosmos_chains = [ + ("atom", "Cosmos Hub", Bip44Coins.COSMOS, 118, "uatom"), + ("osmo", "Osmosis", Bip44Coins.OSMOSIS, 118, "uosmo"), + ("juno", "Juno", Bip44Coins.COSMOS, 118, "ujuno"), # Uses Cosmos derivation + ("secret", "Secret Network", Bip44Coins.SECRET_NETWORK_NEW, 529, "uscrt"), + ("inj", "Injective", Bip44Coins.INJECTIVE, 60, "inj"), + ] + for key, name, coin_cls, _coin_type, _denom in cosmos_chains: + reg[key] = ChainMeta( + key=key, + name=name, + native_symbol=key.upper(), + native_decimals=6, + coin_class=coin_cls, + derivation="bip44_cosmos", + address_pattern=r"^[a-z]+1[a-z0-9]{38,58}$", + curve="secp256k1", + hd_path_template="m/44'/{coin_type}'/{account}'/{change}/{index}", + icon="⚛", + explorer_url=f"https://www.mintscan.io/{key if key != 'secret' else 'secret'}", + ) + + # ── Polkadot / Substrate ── + for key, name, coin_cls in [ + ("dot", "Polkadot", Bip44Coins.POLKADOT_ED25519_SLIP), + ("ksm", "Kusama", Bip44Coins.KUSAMA_ED25519_SLIP), + ]: + reg[key] = ChainMeta( + key=key, + name=name, + native_symbol=key.upper(), + native_decimals=10 if key == "dot" else 12, + coin_class=coin_cls, + derivation="bip44_ed25519_slip", + address_pattern=r"^[1-9A-HJ-NP-Za-km-z]{47,48}$", + curve="ed25519", + hd_path_template="m/44'/354'/{account}'/{change}'/{index}" + if key == "dot" + else "m/44'/434'/{account}'/{change}'/{index}", + icon="⬡", + explorer_url=f"https://{key}.subscan.io", + ) + + # ── Cardano ── + reg["ada"] = ChainMeta( + key="ada", + name="Cardano", + native_symbol="ADA", + native_decimals=6, + coin_class=Bip44Coins.CARDANO_BYRON_ICARUS, + derivation="bip44_ed25519", + address_pattern=r"^addr1[a-z0-9]{50,100}$|^Ae2[a-zA-Z0-9]{50,}$", + curve="ed25519", + hd_path_template="m/44'/1815'/{account}'/{change}/{index}", + icon="₳", + explorer_url="https://cardanoscan.io", + supports_nfts=True, + ) + + # ── Algorand ── + reg["algo"] = ChainMeta( + key="algo", + name="Algorand", + native_symbol="ALGO", + native_decimals=6, + coin_class=Bip44Coins.ALGORAND, + derivation="bip44_ed25519", + address_pattern=r"^[A-Z0-9]{58}$", + curve="ed25519", + hd_path_template="m/44'/283'/{account}'/{change}'/{index}", + icon="🅰", + explorer_url="https://algoexplorer.io", + ) + + # ── Stellar ── + reg["xlm"] = ChainMeta( + key="xlm", + name="Stellar", + native_symbol="XLM", + native_decimals=7, + coin_class=Bip44Coins.STELLAR, + derivation="bip44_ed25519", + address_pattern=r"^G[A-Z0-9]{55}$", + curve="ed25519", + hd_path_template="m/44'/148'/{account}'/{change}'/{index}", + icon="⭐", + explorer_url="https://stellar.expert", + ) + + # ── Ripple ── + reg["xrp"] = ChainMeta( + key="xrp", + name="Ripple", + native_symbol="XRP", + native_decimals=6, + coin_class=Bip44Coins.RIPPLE, + derivation="bip44_secp256k1", + address_pattern=r"^r[rpshnaf39wBUDNEGHJKLM4PQRST7VWXYZ2bcdeCg65jkm8oFqi1tuvAxyz]{27,34}$", + curve="secp256k1", + hd_path_template="m/44'/144'/{account}'/{change}/{index}", + icon="💧", + explorer_url="https://xrpscan.com", + ) + + # ── Near ── + reg["near"] = ChainMeta( + key="near", + name="Near Protocol", + native_symbol="NEAR", + native_decimals=24, + coin_class=Bip44Coins.NEAR_PROTOCOL, + derivation="bip44_ed25519_slip", + address_pattern=r"^[a-z0-9._-]{2,64}\.near$|^[0-9a-fA-F]{64}$", + curve="ed25519", + hd_path_template="m/44'/397'/{account}'/{change}'/{index}", + icon="🌐", + explorer_url="https://nearblocks.io", + ) + + # ── Aptos ── + reg["apt"] = ChainMeta( + key="apt", + name="Aptos", + native_symbol="APT", + native_decimals=8, + coin_class=Bip44Coins.APTOS, + derivation="bip44_ed25519", + address_pattern=r"^0x[a-fA-F0-9]{64}$", + curve="ed25519", + hd_path_template="m/44'/637'/{account}'/{change}'/{index}", + icon="🏗", + explorer_url="https://aptoscan.com", + ) + + # ── Sui ── + reg["sui"] = ChainMeta( + key="sui", + name="Sui", + native_symbol="SUI", + native_decimals=9, + coin_class=Bip44Coins.SUI, + derivation="bip44_ed25519", + address_pattern=r"^0x[a-fA-F0-9]{64}$", + curve="ed25519", + hd_path_template="m/44'/784'/{account}'/{change}'/{index}", + icon="💧", + explorer_url="https://suiscan.xyz", + ) + + # ── TRON ── + reg["trx"] = ChainMeta( + key="trx", + name="TRON", + native_symbol="TRX", + native_decimals=6, + coin_class=Bip44Coins.TRON, + derivation="bip44_secp256k1", + address_pattern=r"^T[A-Za-z1-9]{33}$", + curve="secp256k1", + hd_path_template="m/44'/195'/{account}'/{change}/{index}", + icon="🔷", + explorer_url="https://tronscan.org", + ) + + # ── Bitcoin (multiple address formats) ── + reg["btc"] = ChainMeta( + key="btc", + name="Bitcoin (Legacy)", + native_symbol="BTC", + native_decimals=8, + coin_class=Bip44Coins.BITCOIN, + derivation="bip44", + address_pattern=r"^[13][a-km-zA-HJ-NP-Z1-9]{25,34}$", + curve="secp256k1", + hd_path_template="m/44'/0'/{account}'/{change}/{index}", + icon="₿", + explorer_url="https://mempool.space", + supports_tokens=False, + ) + reg["btc-segwit"] = ChainMeta( + key="btc-segwit", + name="Bitcoin (SegWit)", + native_symbol="BTC", + native_decimals=8, + coin_class=None, + derivation="bip49", + address_pattern=r"^3[a-km-zA-HJ-NP-Z1-9]{25,34}$", + curve="secp256k1", + hd_path_template="m/49'/0'/{account}'/{change}/{index}", + icon="₿", + supports_tokens=False, + ) + reg["btc-native-segwit"] = ChainMeta( + key="btc-native-segwit", + name="Bitcoin (Native SegWit)", + native_symbol="BTC", + native_decimals=8, + coin_class=None, + derivation="bip84", + address_pattern=r"^bc1[a-z0-9]{39,59}$", + curve="secp256k1", + hd_path_template="m/84'/0'/{account}'/{change}/{index}", + icon="₿", + supports_tokens=False, + ) + + # ── Bitcoin Forks ── + for key, name, coin_cls in [ + ("ltc", "Litecoin", Bip44Coins.LITECOIN), + ("doge", "Dogecoin", Bip44Coins.DOGECOIN), + ("dash", "Dash", Bip44Coins.DASH), + ("bch", "Bitcoin Cash", Bip44Coins.BITCOIN_CASH), + ("zec", "Zcash", Bip44Coins.ZCASH), + ]: + reg[key] = ChainMeta( + key=key, + name=name, + native_symbol=key.upper(), + native_decimals=8, + coin_class=coin_cls, + derivation="bip44", + address_pattern=r"^[a-zA-Z0-9]{26,42}$", + curve="secp256k1", + hd_path_template="m/44'/{coin_type}'/{account}'/{change}/{index}", + icon="🪙", + supports_tokens=False, + ) + + # ── Tezos ── + reg["xtz"] = ChainMeta( + key="xtz", + name="Tezos", + native_symbol="XTZ", + native_decimals=6, + coin_class=Bip44Coins.TEZOS, + derivation="bip44_ed25519", + address_pattern=r"^tz[1-3][1-9A-HJ-NP-Za-km-z]{33}$", + curve="ed25519", + hd_path_template="m/44'/1729'/{account}'/{change}/{index}", + icon="ꜩ", + explorer_url="https://tzkt.io", + ) + + # ── Nano ── + reg["nano"] = ChainMeta( + key="nano", + name="Nano", + native_symbol="XNO", + native_decimals=30, + coin_class=Bip44Coins.NANO, + derivation="bip44_ed25519_blake2b", + address_pattern=r"^(nano|xrb)_[13][a-z0-9]{59}$", + curve="ed25519_blake2b", + hd_path_template="m/44'/165'/{account}'/{change}'/{index}", + icon="⚡", + explorer_url="https://nanocrawler.cc", + supports_tokens=False, + ) + + # ── TON ── + reg["ton"] = ChainMeta( + key="ton", + name="TON", + native_symbol="TON", + native_decimals=9, + coin_class=Bip44Coins.TON, + derivation="bip44_ed25519", + address_pattern=r"^(EQ|UQ)[a-zA-Z0-9_-]{46}$", + curve="ed25519", + hd_path_template="m/44'/607'/{account}'/{change}/{index}", + icon="💎", + explorer_url="https://tonscan.org", + ) + + # ── Band Protocol ── + reg["band"] = ChainMeta( + key="band", + name="Band Protocol", + native_symbol="BAND", + native_decimals=6, + coin_class=Bip44Coins.BAND_PROTOCOL, + derivation="bip44_cosmos", + address_pattern=r"^band[a-z0-9]{38,58}$", + curve="secp256k1", + hd_path_template="m/44'/494'/{account}'/{change}'/{index}", + icon="📡", + ) + + return reg + + +CHAIN_REGISTRY = _build_registry() + + +# ── Data Models ───────────────────────────────────────────────── + + +@dataclass +class WalletRecord: + """Complete wallet record with metadata.""" + + wallet_id: str + chain: str + address: str + public_key: str = "" + + # Identity + name: str = "" + description: str = "" + purpose: str = WalletPurpose.OPERATIONS.value + tier: str = WalletTier.WARM.value + status: str = WalletStatus.ACTIVE.value + + # Organization + tags: list[str] = field(default_factory=list) + group: str = "default" + labels: dict[str, str] = field(default_factory=dict) + + # Security + created_at: str = "" + rotated_at: str | None = None + expires_at: str | None = None + last_used: str | None = None + use_count: int = 0 + + # Balance tracking + balance_raw: str = "0" + balance_decimal: float = 0.0 + balance_usd: float = 0.0 + token_balances: dict[str, dict] = field(default_factory=dict) + + # Transaction tracking + total_received: float = 0.0 + total_sent: float = 0.0 + tx_count: int = 0 + last_tx_hash: str = "" + last_tx_time: str | None = None + + # Payment integration + x402_enabled: bool = False + x402_price_usd: float = 0.0 + subscription_enabled: bool = False + subscription_tiers: list[str] = field(default_factory=list) + + # Verification + verified_at: str | None = None + verified_by: str = "" + chain_registered: bool = False # x402 chain registration check + + # Alerts + alert_threshold_usd: float = 100.0 + low_balance_threshold: float = 0.0 + + # Audit + created_by: str = "" + notes: str = "" + version: int = 1 + + def to_dict(self) -> dict: + return asdict(self) + + def to_safe_dict(self) -> dict: + """Return without sensitive data.""" + d = self.to_dict() + d.pop("public_key", None) + return d + + +@dataclass +class PaymentRecord: + """Payment transaction record.""" + + payment_id: str + wallet_id: str + wallet_address: str + chain: str + payment_type: str + + amount: float = 0.0 + amount_usd: float = 0.0 + token: str = "" + + from_address: str = "" + to_address: str = "" + tx_hash: str = "" + block_number: int = 0 + + status: str = "pending" + confirmations: int = 0 + + user_id: str = "" + user_email: str = "" + tool_id: str = "" + tool_name: str = "" + + x402_resource: str = "" + x402_facet: str = "" + + created_at: str = "" + confirmed_at: str | None = None + + metadata: dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict: + return asdict(self) + + +@dataclass +class WalletRotationSchedule: + """Schedule for automatic wallet rotation.""" + + wallet_id: str + chain: str + rotate_every_days: int = 90 + auto_rotate: bool = False + notify_before_days: int = 7 + last_rotated: str | None = None + next_rotation: str | None = None + + def to_dict(self) -> dict: + return asdict(self) + + +# ── Encryption Layer ────────────────────────────────────────────── + + +class WalletEncryption: + """ + AES-256-GCM encryption with memory-hardened Argon2id key derivation. + + Security guarantees: + - Argon2id with high memory cost (128MB) for brute-force resistance + - AES-256-GCM authenticated encryption (confidentiality + integrity) + - Unique random salt + nonce per encryption (no reuse) + - Keys derived per-operation, never cached + """ + + # Memory-hard parameters (OWASP 2026 recommendations) + ARGON_TIME = 4 # Iterations + ARGON_MEMORY = 131072 # 128 MB + ARGON_PARALLELISM = 4 # Threads + + @staticmethod + def _derive_key(password: str, salt: bytes) -> bytes: + """Derive encryption key using Argon2id.""" + if _HAS_CRYPTO: + kdf = Argon2id( + salt=salt, + length=32, + iterations=WalletEncryption.ARGON_TIME, + lanes=WalletEncryption.ARGON_PARALLELISM, + memory_cost=WalletEncryption.ARGON_MEMORY, + ) + return kdf.derive(password.encode()) + else: + import hashlib + + return hashlib.pbkdf2_hmac("sha256", password.encode(), salt, 600000, 32) + + @staticmethod + def encrypt(plaintext: str, password: str) -> str: + """Encrypt plaintext. Returns 'ENC_V2:' or 'DEV:'.""" + if not password: + raise ValueError("Encryption requires a password") + + if not _HAS_CRYPTO: + logger.warning("cryptography not installed - using dev fallback") + return "DEV:" + base64.b64encode(plaintext.encode()).decode() + + salt = os.urandom(16) + key = WalletEncryption._derive_key(password, salt) + aesgcm = AESGCM(key) + nonce = os.urandom(12) + ciphertext = aesgcm.encrypt(nonce, plaintext.encode(), None) + + combined = salt + nonce + ciphertext + # Wipe key from memory + key = b"\x00" * 32 + return "ENC_V2:" + base64.b64encode(combined).decode() + + @staticmethod + def decrypt(ciphertext: str, password: str) -> str: + """Decrypt ciphertext. Raises on wrong password or corruption.""" + if ciphertext.startswith("DEV:"): + return base64.b64decode(ciphertext[4:]).decode() + + prefix = "ENC_V2:" if ciphertext.startswith("ENC_V2:") else "ENC:" + offset = len(prefix) + + if not _HAS_CRYPTO: + raise RuntimeError("cryptography library required for decryption") + + combined = base64.b64decode(ciphertext[offset:]) + salt = combined[:16] + nonce = combined[16:28] + ct = combined[28:] + + key = WalletEncryption._derive_key(password, salt) + aesgcm = AESGCM(key) + plaintext = aesgcm.decrypt(nonce, ct, None) + + result = plaintext.decode() + # Wipe key from memory + key = b"\x00" * 32 + return result + + +# ── Shamir's Secret Sharing ────────────────────────────────────── + + +class ShamirSecretSharing: + """ + M-of-N secret sharing for mnemonic recovery. + + Split a mnemonic into N shares, any M of which can reconstruct it. + Uses finite field arithmetic over GF(256) with simple polynomial evaluation. + + Production note: For real deployment, use the sss-python library or + the cryptography.hazmat.primitives.secret_sharing module. + This implementation is a self-contained, auditable reference. + """ + + @staticmethod + def split(secret_bytes: bytes, threshold: int, total_shares: int) -> list[tuple[int, bytes]]: + """ + Split secret into total_shares shares, threshold needed to recover. + + Returns list of (x, y) tuples where x is the share index (1-based) + and y is the share value. Each share is the same length as the secret. + """ + if threshold > total_shares: + raise ValueError("threshold must be <= total_shares") + if threshold < 2: + raise ValueError("threshold must be >= 2") + if len(secret_bytes) == 0: + raise ValueError("secret cannot be empty") + + # Generate random coefficients for polynomial of degree (threshold-1) + coeffs = [secret_bytes] # coeffs[0] = secret + for i in range(1, threshold): # noqa: B007 + coeffs.append(os.urandom(len(secret_bytes))) + + # Evaluate polynomial at points x=1..total_shares + shares = [] + for x in range(1, total_shares + 1): + # y = coeffs[0] + coeffs[1]*x + coeffs[2]*x^2 + ... + y = bytearray(len(secret_bytes)) + for i, coeff in enumerate(coeffs): + if i == 0: + for j in range(len(y)): + y[j] ^= coeff[j] # XOR identity + else: + multiplier = ShamirSecretSharing._gf_pow(x, i) + for j in range(len(y)): + y[j] ^= ShamirSecretSharing._gf_mul(coeff[j], multiplier) + shares.append((x, bytes(y))) + + return shares + + @staticmethod + def combine(shares: list[tuple[int, bytes]]) -> bytes: + """Reconstruct secret from M shares using Lagrange interpolation.""" + if len(shares) < 2: + raise ValueError("need at least 2 shares") + + secret_len = len(shares[0][1]) + result = bytearray(secret_len) + + for i, (xi, yi) in enumerate(shares): + # Compute Lagrange basis polynomial L_i(0) + numerator = 1 + denominator = 1 + for j, (xj, _) in enumerate(shares): + if i != j: + numerator = ShamirSecretSharing._gf_mul(numerator, xj) + denominator = ShamirSecretSharing._gf_mul(denominator, ShamirSecretSharing._gf_sub(xj, xi)) + + lagrange_coeff = ShamirSecretSharing._gf_div(numerator, denominator) + + for k in range(secret_len): + result[k] ^= ShamirSecretSharing._gf_mul(yi[k], lagrange_coeff) + + return bytes(result) + + # GF(256) arithmetic using Rijndael's finite field + @staticmethod + def _gf_mul(a: int, b: int) -> int: + """Multiply in GF(256).""" + p = 0 + for _ in range(8): + if b & 1: + p ^= a + hi_bit = a & 0x80 + a = (a << 1) & 0xFF + if hi_bit: + a ^= 0x1B # Rijndael's irreducible polynomial + b >>= 1 + return p + + @staticmethod + def _gf_pow(base: int, exp: int) -> int: + """Exponentiation in GF(256).""" + result = 1 + for _ in range(exp): + result = ShamirSecretSharing._gf_mul(result, base) + return result + + @staticmethod + def _gf_sub(a: int, b: int) -> int: + """Subtraction (same as addition/XOR in GF(256)).""" + return a ^ b + + @staticmethod + def _gf_div(a: int, b: int) -> int: + """Division in GF(256).""" + if b == 0: + raise ZeroDivisionError("division by zero in GF(256)") + # Find multiplicative inverse of b + inverse = ShamirSecretSharing._gf_inverse(b) + return ShamirSecretSharing._gf_mul(a, inverse) + + @staticmethod + def _gf_inverse(a: int) -> int: + """Multiplicative inverse in GF(256) using extended Euclidean algorithm.""" + if a == 0: + raise ValueError("cannot invert zero") + # Fermat's little theorem: a^(254) = a^(-1) in GF(256) + result = 1 + base = a + for _ in range(7): # 254 = 11111110 in binary + result = ShamirSecretSharing._gf_mul(result, base) + base = ShamirSecretSharing._gf_mul(base, base) + return result + + +# ── Zero-Knowledge Address Ownership ───────────────────────────── + + +class ZKAddressVerifier: + """ + Prove wallet ownership without exposing the private key. + + Uses a simple challenge-response protocol: + 1. Verifier sends random challenge + 2. Prover signs challenge with private key + 3. Verifier checks signature against public key/address + + For production, would integrate with: + - EIP-712 typed data signing (EVM) + - ed25519-dalek signatures (Solana, Cosmos) + - Monero's MLSAG ring signatures + """ + + @staticmethod + def generate_challenge() -> str: + """Generate a random challenge for ownership verification.""" + return secrets.token_hex(32) + + @staticmethod + def verify_evm_ownership(address: str, challenge: str, signature: str) -> bool: + """ + Verify EVM wallet ownership via personal_sign. + + In production: use web3.eth.account.recover_message() or similar. + Currently returns True if signature is plausible (correct length). + """ + # Basic validation + if not address.startswith("0x") or len(address) != 42: + return False + # Placeholder - real verification needs web3.py + return signature.startswith("0x") and len(signature) >= 130 + + @staticmethod + def verify_ed25519_ownership(public_key_hex: str, challenge: str, signature_hex: str) -> bool: + """Verify ed25519 ownership.""" + if _HAS_NACL: + try: + from nacl.signing import VerifyKey + + vk = VerifyKey(bytes.fromhex(public_key_hex)) + vk.verify(challenge.encode(), bytes.fromhex(signature_hex)) + return True + except Exception: + return False + return True # Placeholder when nacl not available + + +# ── Core Wallet Manager ───────────────────────────────────────── + + +class WalletManagerV2: + """ + Enterprise wallet management system - 95+ chains. + Handles generation, rotation, monitoring, payments, and recovery. + """ + + # Persistent storage - under /app which is volume-mounted from host + VAULT_PATH = "/app/data/wallets/vault_v2.json" + KEYSTORE_PATH = "/app/data/wallets/keystore_v2.enc" + PAYMENTS_PATH = "/app/data/wallets/payments_v2.jsonl" + SHARES_PATH = "/app/data/wallets/shamir_shares/" + + def __init__(self, encryption_password: str = ""): + self.encryption_password = encryption_password or os.getenv("WALLET_VAULT_PASSWORD", "") + self._ensure_dirs() + self._wallets: dict[str, WalletRecord] = {} + self._payments: list[PaymentRecord] = [] + self._load_vault() + self._load_payments() + + def _ensure_dirs(self): + """Ensure wallet directories exist with secure permissions.""" + for p in [os.path.dirname(self.VAULT_PATH), self.SHARES_PATH]: + os.makedirs(p, mode=0o700, exist_ok=True) + + def _load_vault(self): + """Load wallet vault from disk.""" + if os.path.exists(self.VAULT_PATH): + try: + with open(self.VAULT_PATH) as f: + data = json.load(f) + for wdata in data.get("wallets", []): + wr = WalletRecord(**wdata) + self._wallets[wr.wallet_id] = wr + logger.info(f"Vault loaded: {len(self._wallets)} wallets") + except Exception as e: + logger.error(f"Vault load error: {e}") + else: + logger.info("No vault found - starting fresh") + + def _save_vault(self): + """Save wallet vault to disk.""" + data = { + "version": "3.0", + "saved_at": datetime.now(UTC).isoformat(), + "wallet_count": len(self._wallets), + "chains": sorted({w.chain for w in self._wallets.values()}), + "wallets": [w.to_dict() for w in self._wallets.values()], + } + try: + with open(self.VAULT_PATH, "w") as f: + json.dump(data, f, indent=2) + os.chmod(self.VAULT_PATH, 0o600) + except Exception as e: + logger.error(f"Vault save error: {e}") + + def _load_payments(self): + """Load payment records from disk.""" + if os.path.exists(self.PAYMENTS_PATH): + try: + with open(self.PAYMENTS_PATH) as f: + for line in f: + line = line.strip() + if line: + self._payments.append(PaymentRecord(**json.loads(line))) + except Exception as e: + logger.error(f"Payments load error: {e}") + + # ── Chain Registry Access ───────────────────────────────── + + @staticmethod + def get_chain_meta(chain: str) -> ChainMeta | None: + """Get chain metadata.""" + return CHAIN_REGISTRY.get(chain) + + @staticmethod + def list_chains() -> list[str]: + """List all supported chain keys.""" + return sorted(CHAIN_REGISTRY.keys()) + + @staticmethod + def get_chains_by_group() -> dict[str, list[str]]: + """Get chains grouped by ecosystem.""" + groups = { + "evm": [ + k + for k, v in CHAIN_REGISTRY.items() + if v.curve == "secp256k1" + and v.key + in [ + "eth", + "base", + "polygon", + "arbitrum", + "optimism", + "avalanche", + "bsc", + "fantom", + "gnosis", + ] + ], + "cosmos": ["atom", "osmo", "juno", "secret", "inj"], + "bitcoin": [ + "btc", + "btc-segwit", + "btc-native-segwit", + "ltc", + "doge", + "dash", + "bch", + "zec", + ], + "privacy": ["xmr"], + "l1_alt": [ + "sol", + "dot", + "ksm", + "ada", + "algo", + "xlm", + "xrp", + "near", + "apt", + "sui", + "trx", + "xtz", + "nano", + "ton", + "band", + ], + } + return groups + + # ── Wallet Generation ───────────────────────────────────── + + def generate_wallet( + self, + chain: str, + name: str = "", + purpose: str = WalletPurpose.OPERATIONS.value, + tier: str = WalletTier.WARM.value, + tags: list[str] | None = None, + group: str = "default", + created_by: str = "", + ) -> WalletRecord: + """Generate a new wallet for any supported chain.""" + + chain_meta = CHAIN_REGISTRY.get(chain) + if not chain_meta: + raise ValueError(f"Unsupported chain: {chain}. Supported: {sorted(CHAIN_REGISTRY.keys())}") + + wallet_id = f"wal_{chain}_{int(time.time())}_{secrets.token_hex(4)}" + + # Generate keys using chain-appropriate method + if chain_meta.derivation in ("bip44", "bip49", "bip84") and chain_meta.curve == "secp256k1": + address, public_key, mnemonic = self._generate_bip_wallet(chain_meta) + elif chain_meta.derivation == "bip44_ed25519" or chain_meta.derivation == "bip44_ed25519_slip": + address, public_key, mnemonic = self._generate_ed25519_wallet(chain_meta) + elif chain_meta.derivation == "bip44_cosmos": + address, public_key, mnemonic = self._generate_cosmos_wallet(chain_meta) + elif chain_meta.derivation == "monero": + address, public_key, mnemonic = self._generate_monero_wallet(chain_meta) + elif chain_meta.derivation == "bip44_secp256k1": + address, public_key, mnemonic = self._generate_bip_wallet(chain_meta) + elif chain_meta.derivation == "bip44_ed25519_blake2b": + address, public_key, mnemonic = self._generate_ed25519_wallet(chain_meta) + else: + address, public_key, mnemonic = self._generate_bip_wallet(chain_meta) + + wallet = WalletRecord( + wallet_id=wallet_id, + chain=chain, + address=address, + public_key=public_key, + name=name or f"{chain_meta.name} Wallet", + purpose=purpose, + tier=tier, + tags=tags or [], + group=group, + created_at=datetime.now(UTC).isoformat(), + created_by=created_by, + ) + + # Store mnemonic encrypted + if mnemonic and self.encryption_password: + self._store_mnemonic(wallet_id, mnemonic) + wallet.labels["mnemonic"] = "encrypted" + + self._wallets[wallet_id] = wallet + self._save_vault() + + logger.info(f"Wallet generated: {wallet_id} ({chain}) - {address}") + return wallet + + def _generate_bip_wallet(self, chain_meta: ChainMeta) -> tuple[str, str, str]: + """Generate BIP44/49/84 wallet.""" + if not _HAS_BIP_UTILS: + return self._generate_fallback(chain_meta) + + mnemonic = Bip39MnemonicGenerator().FromWordsNumber(Bip39WordsNum.WORDS_NUM_24).ToStr() + seed = Bip39SeedGenerator(mnemonic).Generate() + + if chain_meta.derivation == "bip49" and Bip49Coins: + bip_ctx = Bip49.FromSeed(seed, Bip49Coins.BITCOIN) + elif chain_meta.derivation == "bip84" and Bip84Coins: + bip_ctx = Bip84.FromSeed(seed, Bip84Coins.BITCOIN) + elif chain_meta.coin_class: + bip_ctx = Bip44.FromSeed(seed, chain_meta.coin_class) + else: + bip_ctx = Bip44.FromSeed(seed, Bip44Coins.ETHEREUM) + + address = bip_ctx.PublicKey().ToAddress() + pub_key = bip_ctx.PublicKey().RawCompressed().ToHex() + + return address, pub_key, mnemonic + + def _generate_ed25519_wallet(self, chain_meta: ChainMeta) -> tuple[str, str, str]: + """Generate ed25519-based wallet (Solana, Cardano, Algorand, Stellar, Near, etc.).""" + if not _HAS_BIP_UTILS: + return self._generate_fallback(chain_meta) + + mnemonic = Bip39MnemonicGenerator().FromWordsNumber(Bip39WordsNum.WORDS_NUM_24).ToStr() + seed = Bip39SeedGenerator(mnemonic).Generate() + + if chain_meta.coin_class: + bip_ctx = Bip44.FromSeed(seed, chain_meta.coin_class) + else: + bip_ctx = Bip44.FromSeed(seed, Bip44Coins.SOLANA) + + address = bip_ctx.PublicKey().ToAddress() + pub_key = bip_ctx.PublicKey().RawCompressed().ToHex() + + return address, pub_key, mnemonic + + def _generate_cosmos_wallet(self, chain_meta: ChainMeta) -> tuple[str, str, str]: + """Generate Cosmos/IBC wallet.""" + if not _HAS_BIP_UTILS: + return self._generate_fallback(chain_meta) + + mnemonic = Bip39MnemonicGenerator().FromWordsNumber(Bip39WordsNum.WORDS_NUM_24).ToStr() + seed = Bip39SeedGenerator(mnemonic).Generate() + + if chain_meta.coin_class: + bip_ctx = Bip44.FromSeed(seed, chain_meta.coin_class) + else: + bip_ctx = Bip44.FromSeed(seed, Bip44Coins.COSMOS) + + address = bip_ctx.PublicKey().ToAddress() + pub_key = bip_ctx.PublicKey().RawCompressed().ToHex() + + return address, pub_key, mnemonic + + def _generate_monero_wallet(self, chain_meta: ChainMeta) -> tuple[str, str, str]: + """Generate Monero wallet using ed25519-blake2b.""" + mnemonic = "" + + if _HAS_BIP_UTILS: + try: + # Monero uses 25-word mnemonics + mnemonic = Bip39MnemonicGenerator().FromWordsNumber(Bip39WordsNum.WORDS_NUM_24).ToStr() + seed = Bip39SeedGenerator(mnemonic).Generate() + + # Use bip_utils Monero support + monero_ctx = Bip44.FromSeed(seed, Bip44Coins.MONERO_ED25519_SLIP) + address = monero_ctx.PublicKey().ToAddress() + pub_key = monero_ctx.PublicKey().RawCompressed().ToHex() + return address, pub_key, mnemonic + except Exception as e: + logger.warning(f"bip_utils Monero failed ({e}), trying fallback") + + # Fallback: raw ed25519-blake2b keypair + if _HAS_NACL: + sk = NaClSigningKey.generate() + pub_key = sk.verify_key.encode().hex() + + # Monero-style address (simplified) + # Real Monero addresses require keccak + base58 encoding + h = hashlib.sha3_256() + h.update(bytes.fromhex(pub_key)) + raw_addr = h.digest() + address = "4" + base58.b58encode(raw_addr[:64]).decode()[:94] if _HAS_BASE58 else "4" + raw_addr.hex()[:94] + + return address, pub_key, mnemonic + + # Ultimate fallback + priv = secrets.token_hex(32) + pub = hashlib.sha256(priv.encode()).hexdigest() + return "4" + pub[:94], pub, mnemonic + + def _generate_fallback(self, chain_meta: ChainMeta) -> tuple[str, str, str]: + """Fallback wallet generation (dev only, not for production).""" + logger.warning(f"Using fallback keygen for {chain_meta.key} - NOT SECURE FOR PRODUCTION") + priv = secrets.token_hex(32) + pub = hashlib.sha256(priv.encode()).hexdigest() + + prefix_map = { + "eth": "0x", + "base": "0x", + "polygon": "0x", + "arbitrum": "0x", + "optimism": "0x", + "avalanche": "0x", + "bsc": "0x", + "fantom": "0x", + "gnosis": "0x", + "sol": "", + "trx": "T", + "xrp": "r", + } + prefix = prefix_map.get(chain_meta.key, chain_meta.key + "_") + addr = prefix + pub[:40] + + return addr, pub, "" + + def generate_hd_wallet( + self, + chain: str, + mnemonic: str = "", + account_index: int = 0, + address_index: int = 0, + name: str = "", + purpose: str = WalletPurpose.OPERATIONS.value, + tier: str = WalletTier.WARM.value, + tags: list[str] | None = None, + group: str = "default", + created_by: str = "", + ) -> WalletRecord: + """Generate HD wallet from mnemonic or create new one.""" + + if not mnemonic and _HAS_BIP_UTILS: + mnemonic = Bip39MnemonicGenerator().FromWordsNumber(Bip39WordsNum.WORDS_NUM_24).ToStr() + + wallet = self.generate_wallet( + chain=chain, + name=name, + purpose=purpose, + tier=tier, + tags=tags, + group=group, + created_by=created_by, + ) + + if mnemonic: + self._store_mnemonic(wallet.wallet_id, mnemonic) + + wallet.labels["hd_account"] = str(account_index) + wallet.labels["hd_address"] = str(address_index) + + self._save_vault() + return wallet + + def generate_batch( + self, + chains: list[str], + name_prefix: str = "", + purpose: str = WalletPurpose.PAYMENTS.value, + tier: str = WalletTier.WARM.value, + group: str = "x402", + created_by: str = "", + ) -> list[WalletRecord]: + """Batch generate wallets for multiple chains.""" + wallets = [] + for chain in chains: + try: + wallet = self.generate_wallet( + chain=chain, + name=f"{name_prefix} {chain.upper()}".strip() if name_prefix else f"{chain.upper()} Payment", + purpose=purpose, + tier=tier, + group=group, + created_by=created_by, + ) + wallets.append(wallet) + except Exception as e: + logger.error(f"Failed to generate {chain} wallet: {e}") + return wallets + + # ── Key Storage ─────────────────────────────────────────── + + def _store_mnemonic(self, wallet_id: str, mnemonic: str): + """Store encrypted mnemonic in keystore.""" + if not self.encryption_password: + logger.warning("No encryption password - mnemonic not stored securely") + return + + encrypted = WalletEncryption.encrypt(mnemonic, self.encryption_password) + + keystore = {} + if os.path.exists(self.KEYSTORE_PATH): + try: + with open(self.KEYSTORE_PATH) as f: + keystore = json.load(f) + except Exception: + pass + + keystore[wallet_id] = { + "encrypted_mnemonic": encrypted, + "stored_at": datetime.now(UTC).isoformat(), + "encryption_version": "V2_Argon2id", + } + + with open(self.KEYSTORE_PATH, "w") as f: + json.dump(keystore, f, indent=2) + os.chmod(self.KEYSTORE_PATH, 0o600) + + def get_mnemonic(self, wallet_id: str) -> str | None: + """Retrieve and decrypt mnemonic.""" + if not os.path.exists(self.KEYSTORE_PATH): + return None + + try: + with open(self.KEYSTORE_PATH) as f: + keystore = json.load(f) + except Exception: + return None + + entry = keystore.get(wallet_id) + if not entry: + return None + + return WalletEncryption.decrypt(entry["encrypted_mnemonic"], self.encryption_password) + + def create_shamir_backup(self, wallet_id: str, threshold: int = 3, total: int = 5) -> list[str]: + """Create Shamir's Secret Sharing backup shares for a wallet's mnemonic.""" + mnemonic = self.get_mnemonic(wallet_id) + if not mnemonic: + raise ValueError("No mnemonic stored for this wallet") + + shares = ShamirSecretSharing.split(mnemonic.encode(), threshold, total) + + share_dir = os.path.join(self.SHARES_PATH, wallet_id) + os.makedirs(share_dir, mode=0o700, exist_ok=True) + + share_paths = [] + for x, share_bytes in shares: + path = os.path.join(share_dir, f"share_{x}_of_{total}.enc") + enc_share = WalletEncryption.encrypt(share_bytes.hex(), self.encryption_password) + with open(path, "w") as f: + json.dump( + { + "wallet_id": wallet_id, + "share_index": x, + "total_shares": total, + "threshold": threshold, + "encrypted_share": enc_share, + "created_at": datetime.now(UTC).isoformat(), + }, + f, + indent=2, + ) + os.chmod(path, 0o600) + share_paths.append(path) + + # Mark in wallet labels + wallet = self._wallets.get(wallet_id) + if wallet: + wallet.labels["shamir_backup"] = f"{threshold}-of-{total}" + wallet.labels["shamir_path"] = share_dir + self._save_vault() + + logger.info(f"Shamir backup created for {wallet_id}: {threshold}-of-{total}") + return share_paths + + def recover_from_shamir(self, wallet_id: str, share_paths: list[str]) -> str: + """Recover mnemonic from Shamir shares.""" + shares = [] + for path in share_paths: + with open(path) as f: + data = json.load(f) + share_hex = WalletEncryption.decrypt(data["encrypted_share"], self.encryption_password) + shares.append((data["share_index"], bytes.fromhex(share_hex))) + + mnemonic = ShamirSecretSharing.combine(shares).decode() + self._store_mnemonic(wallet_id, mnemonic) + return mnemonic + + # ── Wallet Operations ───────────────────────────────────── + + def get_wallet(self, wallet_id: str) -> WalletRecord | None: + """Get wallet by ID.""" + return self._wallets.get(wallet_id) + + def get_wallet_by_address(self, chain: str, address: str) -> WalletRecord | None: + """Find wallet by chain + address.""" + for w in self._wallets.values(): + if w.chain == chain and w.address.lower() == address.lower(): + return w + return None + + def list_wallets( + self, + chain: str = "", + purpose: str = "", + tier: str = "", + status: str = "", + group: str = "", + tags: list[str] | None = None, + x402_enabled: bool | None = None, + ) -> list[WalletRecord]: + """List wallets with filtering.""" + results = [] + for w in self._wallets.values(): + if chain and w.chain != chain: + continue + if purpose and w.purpose != purpose: + continue + if tier and w.tier != tier: + continue + if status and w.status != status: + continue + if group and w.group != group: + continue + if tags and not any(t in w.tags for t in tags): + continue + if x402_enabled is not None and w.x402_enabled != x402_enabled: + continue + results.append(w) + return results + + def update_wallet(self, wallet_id: str, updates: dict[str, Any]) -> WalletRecord | None: + """Update wallet metadata.""" + wallet = self._wallets.get(wallet_id) + if not wallet: + return None + + for key, value in updates.items(): + if hasattr(wallet, key): + setattr(wallet, key, value) + + wallet.version += 1 + self._save_vault() + return wallet + + def delete_wallet(self, wallet_id: str) -> bool: + """Archive wallet.""" + wallet = self._wallets.get(wallet_id) + if not wallet: + return False + + wallet.status = WalletStatus.ARCHIVED.value + wallet.labels["archived_at"] = datetime.now(UTC).isoformat() + self._save_vault() + return True + + # ── Rotation ────────────────────────────────────────────── + + def rotate_wallet( + self, + wallet_id: str, + transfer_balance: bool = False, + rotate_by: str = "", + ) -> WalletRecord | None: + """Rotate to a new wallet, marking old as rotated.""" + old_wallet = self._wallets.get(wallet_id) + if not old_wallet: + return None + + new_wallet = self.generate_wallet( + chain=old_wallet.chain, + name=old_wallet.name + " (v2)", + purpose=old_wallet.purpose, + tier=old_wallet.tier, + tags=[*old_wallet.tags, "rotated"], + group=old_wallet.group, + created_by=rotate_by, + ) + + old_wallet.status = WalletStatus.ROTATED.value + old_wallet.rotated_at = datetime.now(UTC).isoformat() + old_wallet.labels["rotated_to"] = new_wallet.wallet_id + + new_wallet.labels["rotated_from"] = old_wallet.wallet_id + new_wallet.labels["rotation_reason"] = "scheduled" + + self._save_vault() + logger.info(f"Wallet rotated: {wallet_id} -> {new_wallet.wallet_id}") + return new_wallet + + def schedule_rotation(self, wallet_id: str, days: int, auto: bool = False) -> WalletRotationSchedule | None: + """Schedule automatic rotation.""" + wallet = self._wallets.get(wallet_id) + if not wallet: + return None + + schedule = WalletRotationSchedule( + wallet_id=wallet_id, + chain=wallet.chain, + rotate_every_days=days, + auto_rotate=auto, + last_rotated=wallet.created_at, + next_rotation=(datetime.now(UTC) + timedelta(days=days)).isoformat(), + ) + + wallet.labels["rotation_schedule"] = json.dumps(schedule.to_dict()) + self._save_vault() + return schedule + + def check_rotations_due(self) -> list[WalletRotationSchedule]: + """Check which wallets are due for rotation.""" + due = [] + now = datetime.now(UTC) + + for w in self._wallets.values(): + if w.status != WalletStatus.ACTIVE.value: + continue + schedule_str = w.labels.get("rotation_schedule") + if not schedule_str: + continue + schedule = WalletRotationSchedule(**json.loads(schedule_str)) + if schedule.next_rotation: + next_rot = datetime.fromisoformat(schedule.next_rotation.replace("Z", "+00:00")) + if now >= next_rot: + due.append(schedule) + + return due + + # ── Balance & Monitoring ───────────────────────────────── + + def update_balance( + self, + wallet_id: str, + balance_raw: str, + balance_decimal: float, + balance_usd: float, + token_balances: dict | None = None, + ) -> bool: + """Update wallet balance.""" + wallet = self._wallets.get(wallet_id) + if not wallet: + return False + + wallet.balance_raw = balance_raw + wallet.balance_decimal = balance_decimal + wallet.balance_usd = balance_usd + if token_balances: + wallet.token_balances = token_balances + + self._save_vault() + return True + + def record_transaction( + self, + wallet_id: str, + tx_hash: str, + amount: float, + direction: str, + token: str = "", + usd_value: float = 0.0, + ) -> bool: + """Record a transaction for a wallet.""" + wallet = self._wallets.get(wallet_id) + if not wallet: + return False + + wallet.tx_count += 1 + wallet.last_tx_hash = tx_hash + wallet.last_tx_time = datetime.now(UTC).isoformat() + + if direction == "in": + wallet.total_received += amount + else: + wallet.total_sent += amount + + wallet.use_count += 1 + wallet.last_used = datetime.now(UTC).isoformat() + + self._save_vault() + return True + + # ── Payment Integration ─────────────────────────────────── + + def record_payment(self, payment: PaymentRecord) -> bool: + """Record a payment transaction.""" + self._payments.append(payment) + + try: + with open(self.PAYMENTS_PATH, "a") as f: + f.write(json.dumps(payment.to_dict()) + "\n") + except Exception as e: + logger.error(f"Payment record error: {e}") + + if payment.wallet_id: + wallet = self._wallets.get(payment.wallet_id) + if wallet: + if payment.payment_type in [ + PaymentType.DEPOSIT.value, + PaymentType.SUBSCRIPTION.value, + ]: + wallet.total_received += payment.amount_usd + elif payment.payment_type in [ + PaymentType.WITHDRAWAL.value, + PaymentType.REFUND.value, + ]: + wallet.total_sent += payment.amount_usd + wallet.tx_count += 1 + wallet.last_tx_time = payment.created_at + + self._save_vault() + return True + + def get_payments( + self, + wallet_id: str = "", + chain: str = "", + payment_type: str = "", + status: str = "", + user_id: str = "", + start_date: str = "", + end_date: str = "", + limit: int = 100, + ) -> list[PaymentRecord]: + """Query payment records.""" + results = [] + for p in reversed(self._payments): + if wallet_id and p.wallet_id != wallet_id: + continue + if chain and p.chain != chain: + continue + if payment_type and p.payment_type != payment_type: + continue + if status and p.status != status: + continue + if user_id and p.user_id != user_id: + continue + if start_date and p.created_at < start_date: + continue + if end_date and p.created_at > end_date: + continue + results.append(p) + if len(results) >= limit: + break + return results + + def enable_x402(self, wallet_id: str, price_usd: float) -> bool: + """Enable x402 payment processing for a wallet.""" + wallet = self._wallets.get(wallet_id) + if not wallet: + return False + + wallet.x402_enabled = True + wallet.x402_price_usd = price_usd + wallet.purpose = WalletPurpose.PAYMENTS.value + self._save_vault() + return True + + def enable_subscription(self, wallet_id: str, tiers: list[str]) -> bool: + """Enable subscription payments for a wallet.""" + wallet = self._wallets.get(wallet_id) + if not wallet: + return False + + wallet.subscription_enabled = True + wallet.subscription_tiers = tiers + wallet.purpose = WalletPurpose.SUBSCRIPTIONS.value + self._save_vault() + return True + + # ── x402 Chain Registration ────────────────────────────── + + def register_for_x402(self, wallet_id: str) -> bool: + """ + Register wallet with x402 payment system. + + Verifies the wallet can receive payments according to the chain's + standards, checks address format, and marks it as chain_registered. + """ + wallet = self._wallets.get(wallet_id) + if not wallet: + return False + + chain_meta = CHAIN_REGISTRY.get(wallet.chain) + if not chain_meta: + logger.warning(f"No chain metadata for {wallet.chain}") + return False + + # Validate address format + import re + + if chain_meta.address_pattern: + if not re.match(chain_meta.address_pattern, wallet.address): + logger.error(f"Address {wallet.address} doesn't match pattern for {wallet.chain}") + return False + + # Mark as verified and chain-registered + wallet.verified_at = datetime.now(UTC).isoformat() + wallet.verified_by = "wallet_manager_v2" + wallet.chain_registered = True + + # Enable x402 + wallet.x402_enabled = True + + self._save_vault() + logger.info(f"Wallet {wallet_id} ({wallet.chain}) registered for x402") + return True + + # ── Statistics ─────────────────────────────────────────── + + def get_stats(self) -> dict[str, Any]: + """Get comprehensive wallet statistics.""" + total_balance_usd = sum(w.balance_usd for w in self._wallets.values()) + + by_chain = {} + by_purpose = {} + by_tier = {} + by_status = {} + by_ecosystem = {} + + for w in self._wallets.values(): + by_chain[w.chain] = by_chain.get(w.chain, 0) + 1 + by_purpose[w.purpose] = by_purpose.get(w.purpose, 0) + 1 + by_tier[w.tier] = by_tier.get(w.tier, 0) + 1 + by_status[w.status] = by_status.get(w.status, 0) + 1 + + # Ecosystem grouping + meta = CHAIN_REGISTRY.get(w.chain) + eco = meta.curve if meta else "unknown" + by_ecosystem[eco] = by_ecosystem.get(eco, 0) + 1 + + total_payments = len(self._payments) + total_revenue = sum(p.amount_usd for p in self._payments if p.status == "confirmed") + + return { + "total_wallets": len(self._wallets), + "active_wallets": by_status.get(WalletStatus.ACTIVE.value, 0), + "total_balance_usd": round(total_balance_usd, 2), + "chains_supported": len(CHAIN_REGISTRY), + "chains_in_use": len(by_chain), + "by_chain": by_chain, + "by_ecosystem": by_ecosystem, + "by_purpose": by_purpose, + "by_tier": by_tier, + "by_status": by_status, + "total_payments": total_payments, + "total_revenue_usd": round(total_revenue, 2), + "x402_wallets": sum(1 for w in self._wallets.values() if w.x402_enabled), + "subscription_wallets": sum(1 for w in self._wallets.values() if w.subscription_enabled), + "rotated_wallets": sum(1 for w in self._wallets.values() if w.status == WalletStatus.ROTATED.value), + "last_updated": datetime.now(UTC).isoformat(), + } + + # ── Export / Import ─────────────────────────────────────── + + def export_safe(self) -> dict[str, Any]: + """Export wallet data without keys.""" + return { + "version": "3.0", + "exported_at": datetime.now(UTC).isoformat(), + "wallets": [w.to_safe_dict() for w in self._wallets.values()], + "chain_registry": {k: {"name": v.name, "symbol": v.native_symbol} for k, v in CHAIN_REGISTRY.items()}, + } + + def export_for_chain(self, chain: str) -> list[dict]: + """Export wallets for a specific chain.""" + return [w.to_safe_dict() for w in self._wallets.values() if w.chain == chain] + + # ── Alerts ──────────────────────────────────────────────── + + def check_alerts(self) -> list[dict]: + """Check all wallets for alert conditions.""" + alerts = [] + + for w in self._wallets.values(): + if w.status != WalletStatus.ACTIVE.value: + continue + + if w.low_balance_threshold > 0 and w.balance_usd < w.low_balance_threshold: + alerts.append( + { + "type": "low_balance", + "wallet_id": w.wallet_id, + "address": w.address, + "chain": w.chain, + "balance_usd": w.balance_usd, + "threshold": w.low_balance_threshold, + "severity": "warning", + } + ) + + schedule_str = w.labels.get("rotation_schedule") + if schedule_str: + schedule = WalletRotationSchedule(**json.loads(schedule_str)) + if schedule.next_rotation: + next_rot = datetime.fromisoformat(schedule.next_rotation.replace("Z", "+00:00")) + days_until = (next_rot - datetime.now(UTC)).days + + if days_until <= schedule.notify_before_days and days_until > 0: + alerts.append( + { + "type": "rotation_due", + "wallet_id": w.wallet_id, + "address": w.address, + "chain": w.chain, + "days_until": days_until, + "severity": "info", + } + ) + elif days_until <= 0: + alerts.append( + { + "type": "rotation_overdue", + "wallet_id": w.wallet_id, + "address": w.address, + "chain": w.chain, + "days_overdue": abs(days_until), + "severity": "critical", + } + ) + + return alerts + + # ── Auto-Sweep ─────────────────────────────────────────────── + + async def sweep_to_owner(self, chain: str = "", min_usd_threshold: float = 5.0) -> dict: + """ + Sweep USDC balances above threshold from x402 receiving wallets to owner wallets. + + EVM owner: 0x1E3AC01d0fdb976179790BDD02823196A92705C9 + Solana owner: Gix4P9AmwcZRGzr2hCEME5m2QAvY86dBfm8c7e7MpFzv + + USDC on Base: 0x833589fcd6edb6e08f4c7c32d4f71b54bda02913 + """ + owner_map = { + "evm": "0x1E3AC01d0fdb976179790BDD02823196A92705C9", + "sol": "Gix4P9AmwcZRGzr2hCEME5m2QAvY86dBfm8c7e7MpFzv", + } + evm_chains = { + "eth", + "base", + "polygon", + "arbitrum", + "optimism", + "avalanche", + "bsc", + "fantom", + "gnosis", + } + + result = { + "timestamp": datetime.now(UTC).isoformat(), + "chain_filter": chain or "all", + "threshold_usd": min_usd_threshold, + "swept": [], + "skipped": [], + "errors": [], + "total_swept_usd": 0.0, + } + + # Find x402-enabled wallets + wallets_to_check = [] + for w in self._wallets.values(): + if not w.x402_enabled: + continue + if chain and w.chain != chain: + continue + wallets_to_check.append(w) + + if not wallets_to_check: + logger.info("sweep_to_owner: no x402-enabled wallets found") + result["skipped"].append("no_x402_wallets") + return result + + for wallet in wallets_to_check: + try: + balance = wallet.balance_usd + if balance < min_usd_threshold: + result["skipped"].append( + { + "wallet_id": wallet.wallet_id, + "address": wallet.address, + "chain": wallet.chain, + "balance_usd": balance, + "reason": f"below_threshold ({balance} < {min_usd_threshold})", + } + ) + continue + + # Determine owner address and execute REAL on-chain sweep + if wallet.chain in evm_chains: + owner = owner_map["evm"] + rpc_url = ( + os.getenv("BASE_RPC_URL", "https://mainnet.base.org") + if wallet.chain == "base" + else os.getenv("ETH_RPC_URL", "https://eth.llamarpc.com") + ) + + # 1. Retrieve mnemonic (will fail if keys were never saved) + mnemonic = self.get_mnemonic(wallet.wallet_id) + if not mnemonic: + raise RuntimeError( + f"CRITICAL: Cannot sweep {wallet.wallet_id}. Mnemonic is missing. Keys were never persisted." + ) + + # 2. Derive private key + from eth_account import Account + + acct = Account.from_mnemonic(mnemonic) + if acct.address.lower() != wallet.address.lower(): + raise RuntimeError(f"Derived address {acct.address} does not match wallet {wallet.address}") + + # 3. Connect to Web3 + from web3 import Web3 + + w3 = Web3(Web3.HTTPProvider(rpc_url)) + if not w3.is_connected(): + raise RuntimeError(f"Failed to connect to RPC: {rpc_url}") + + # 4. Check actual on-chain balance + on_chain_balance_wei = w3.eth.get_balance(wallet.address) + if on_chain_balance_wei == 0: + raise RuntimeError("On-chain balance is 0. Local ledger is out of sync.") + + # Leave a small amount for future gas (0.0001 ETH) + gas_reserve_wei = w3.to_wei(0.0001, "ether") + if on_chain_balance_wei <= gas_reserve_wei: + raise RuntimeError("Balance too low to sweep after gas reserve.") + + sweep_amount_wei = on_chain_balance_wei - gas_reserve_wei + + # 5. Build and sign transaction + nonce = w3.eth.get_transaction_count(wallet.address) + gas_price = w3.eth.gas_price + tx = { + "nonce": nonce, + "to": Web3.to_checksum_address(owner), + "value": sweep_amount_wei, + "gas": 21000, + "gasPrice": gas_price, + "chainId": w3.eth.chain_id, + } + signed_tx = w3.eth.account.sign_transaction(tx, acct.key) + + # 6. Broadcast to network + tx_hash = w3.eth.send_raw_transaction(signed_tx.rawTransaction) + tx_hash_hex = tx_hash.hex() + logger.info(f"SUCCESS: Broadcasted native sweep tx {tx_hash_hex} for {wallet.wallet_id}") + + # 7. Update local ledger ONLY AFTER SUCCESSFUL BROADCAST + wallet.balance_raw = str(gas_reserve_wei) + wallet.balance_decimal = float(w3.from_wei(gas_reserve_wei, "ether")) + wallet.balance_usd = float(w3.from_wei(gas_reserve_wei, "ether")) * 2500.0 # Approx ETH price + wallet.total_sent += float(w3.from_wei(sweep_amount_wei, "ether")) + + result["swept"].append( + { + "wallet_id": wallet.wallet_id, + "address": wallet.address, + "chain": wallet.chain, + "amount_wei": sweep_amount_wei, + "tx_hash": tx_hash_hex, + "owner": owner, + "status": "broadcasted", + } + ) + result["total_swept_usd"] += float(w3.from_wei(sweep_amount_wei, "ether")) * 2500.0 + + elif wallet.chain == "sol": + owner = owner_map["sol"] + rpc_url = os.getenv("SOLANA_RPC_URL", "https://api.mainnet-beta.solana.com") + + # 1. Retrieve mnemonic + mnemonic = self.get_mnemonic(wallet.wallet_id) + if not mnemonic: + raise RuntimeError(f"CRITICAL: Cannot sweep {wallet.wallet_id}. Mnemonic is missing.") + + # 2. Derive private key using bip_utils + from bip_utils import Bip39SeedGenerator, Bip44, Bip44Coins + + seed_bytes = Bip39SeedGenerator(mnemonic).Generate() + bip44_mst_ctx = Bip44.FromSeed(seed_bytes, Bip44Coins.SOLANA) + bip44_def_ctx = bip44_mst_ctx.Purpose().Coin().Account(0).Change(0).AddressIndex(0) + private_key_bytes = bip44_def_ctx.PrivateKey().Raw().ToBytes() + + # 3. Create Solana Keypair + from solders.keypair import Keypair as SoldersKeypair + from solders.pubkey import Pubkey as SoldersPubkey + + keypair = SoldersKeypair.from_bytes(private_key_bytes) + if str(keypair.pubkey()) != wallet.address: + raise RuntimeError(f"Derived address {keypair.pubkey()} does not match wallet {wallet.address}") + + # 4. Connect to Solana RPC + from solana.rpc.async_api import AsyncClient + from solana.rpc.commitment import Confirmed + from solana.transaction import Transaction + from solders.system_program import TransferParams, transfer + + owner_pubkey = SoldersPubkey.from_string(owner) + + async with AsyncClient(rpc_url) as client: + # 5. Check actual on-chain balance + resp = await client.get_balance(keypair.pubkey(), commitment=Confirmed) + on_chain_balance_lamports = resp.value + + if on_chain_balance_lamports == 0: + raise RuntimeError("On-chain balance is 0. Local ledger is out of sync.") + + # Leave a small amount for rent/gas (e.g., 0.001 SOL = 1,000,000 lamports) + gas_reserve_lamports = 1_000_000 + if on_chain_balance_lamports <= gas_reserve_lamports: + raise RuntimeError("Balance too low to sweep after gas reserve.") + + sweep_amount_lamports = on_chain_balance_lamports - gas_reserve_lamports + + # 6. Build and sign transaction + transfer_ix = transfer( + TransferParams( + from_pubkey=keypair.pubkey(), + to_pubkey=owner_pubkey, + lamports=sweep_amount_lamports, + ) + ) + tx = Transaction().add(transfer_ix) + tx.sign(keypair) + + # 7. Broadcast to network + result_tx = await client.send_transaction(tx, keypair) + tx_sig = str(result_tx.value) + logger.info(f"SUCCESS: Broadcasted Solana sweep tx {tx_sig} for {wallet.wallet_id}") + + # 8. Update local ledger ONLY AFTER SUCCESSFUL BROADCAST + wallet.balance_raw = str(gas_reserve_lamports) + wallet.balance_decimal = gas_reserve_lamports / 1_000_000_000 + wallet.balance_usd = wallet.balance_decimal * 150.0 # Approx SOL price + wallet.total_sent += sweep_amount_lamports / 1_000_000_000 + + result["swept"].append( + { + "wallet_id": wallet.wallet_id, + "address": wallet.address, + "chain": wallet.chain, + "amount_lamports": sweep_amount_lamports, + "tx_signature": tx_sig, + "owner": owner, + "status": "broadcasted", + } + ) + result["total_swept_usd"] += (sweep_amount_lamports / 1_000_000_000) * 150.0 + else: + result["skipped"].append( + { + "wallet_id": wallet.wallet_id, + "address": wallet.address, + "chain": wallet.chain, + "balance_usd": balance, + "reason": "unsupported_chain_for_sweep", + } + ) + continue + + except Exception as e: + logger.error(f"sweep_to_owner error for {wallet.wallet_id}: {e}") + result["errors"].append( + { + "wallet_id": wallet.wallet_id, + "address": wallet.address, + "chain": wallet.chain, + "error": str(e), + } + ) + + self._save_vault() + result["total_swept_usd"] = round(result["total_swept_usd"], 2) + logger.info( + f"sweep_to_owner complete: {len(result['swept'])} swept, " + f"{len(result['skipped'])} skipped, {len(result['errors'])} errors" + ) + return result + + async def check_all_balances(self) -> list[dict]: + """ + Check balances on all x402-enabled wallets. + Returns alerts for any wallet below its low_balance_threshold. + Called by cron job every 30 minutes. + """ + alerts = [] + checked = 0 + below_threshold = 0 + + for w in self._wallets.values(): + if not w.x402_enabled: + continue + checked += 1 + + alert = { + "wallet_id": w.wallet_id, + "address": w.address, + "chain": w.chain, + "balance_usd": w.balance_usd, + "low_balance_threshold": w.low_balance_threshold, + "checked_at": datetime.now(UTC).isoformat(), + } + + if w.low_balance_threshold > 0 and w.balance_usd < w.low_balance_threshold: + alert["type"] = "low_balance" + alert["severity"] = "warning" + alert["message"] = ( + f"Wallet {w.wallet_id} ({w.chain}) balance ${w.balance_usd:.2f} " + f"is below threshold ${w.low_balance_threshold:.2f}" + ) + alerts.append(alert) + below_threshold += 1 + logger.warning(alert["message"]) + elif w.balance_usd <= 0: + alert["type"] = "zero_balance" + alert["severity"] = "info" + alert["message"] = f"Wallet {w.wallet_id} ({w.chain}) has zero balance" + alerts.append(alert) + else: + alert["type"] = "ok" + alert["severity"] = "info" + alert["message"] = f"Wallet {w.wallet_id} ({w.chain}) balance ${w.balance_usd:.2f} OK" + alerts.append(alert) + + summary = { + "checked": checked, + "below_threshold": below_threshold, + "total_alerts": len(alerts), + "alerts": alerts, + "checked_at": datetime.now(UTC).isoformat(), + } + + logger.info(f"check_all_balances: checked {checked} x402 wallets, {below_threshold} below threshold") + return summary + + async def get_revenue_wallet_status(self) -> dict: + """ + Return summary of all x402 receiving wallets with current balances, + chain info, and whether they need sweeping. + """ + wallets = [] + total_balance_usd = 0.0 + evm_chains = { + "eth", + "base", + "polygon", + "arbitrum", + "optimism", + "avalanche", + "bsc", + "fantom", + "gnosis", + } + sweep_threshold = 5.0 # Default sweep threshold + + for w in self._wallets.values(): + if not w.x402_enabled: + continue + + needs_sweep = w.balance_usd >= sweep_threshold + if w.chain in evm_chains: + owner = "0x1E3AC01d0fdb976179790BDD02823196A92705C9" + elif w.chain == "sol": + owner = "Gix4P9AmwcZRGzr2hCEME5m2QAvY86dBfm8c7e7MpFzv" + else: + owner = "" + + wallets.append( + { + "wallet_id": w.wallet_id, + "address": w.address, + "chain": w.chain, + "balance_usd": w.balance_usd, + "balance_raw": w.balance_raw, + "tier": w.tier, + "purpose": w.purpose, + "status": w.status, + "x402_price_usd": w.x402_price_usd, + "low_balance_threshold": w.low_balance_threshold, + "needs_sweep": needs_sweep, + "owner_address": owner, + "last_used": w.last_used, + "tx_count": w.tx_count, + } + ) + total_balance_usd += w.balance_usd + + need_sweep_count = sum(1 for w in wallets if w["needs_sweep"]) + below_threshold_count = sum( + 1 for w in wallets if w["low_balance_threshold"] > 0 and w["balance_usd"] < w["low_balance_threshold"] + ) + + return { + "timestamp": datetime.now(UTC).isoformat(), + "total_x402_wallets": len(wallets), + "total_balance_usd": round(total_balance_usd, 2), + "need_sweep_count": need_sweep_count, + "below_threshold_count": below_threshold_count, + "sweep_threshold_usd": sweep_threshold, + "wallets": wallets, + } + + +# ── Singleton ───────────────────────────────────────────────────── + +_wallet_manager_instance: WalletManagerV2 | None = None + + +def get_wallet_manager_v2(password: str = "") -> WalletManagerV2: + """Get or create wallet manager instance.""" + global _wallet_manager_instance + if _wallet_manager_instance is None: + _wallet_manager_instance = WalletManagerV2(password) + return _wallet_manager_instance + + +# ── Utility: Import legacy wallets ──────────────────────────────── + + +def import_legacy_wallets(manager: WalletManagerV2) -> int: + """Import wallets from legacy v1 JSON files into v2 vault.""" + imported = 0 + + legacy_files = [ + "/root/.rmi/wallets/x402_wallets.json", + "/root/.rmi/wallets/referral_wallets.json", + "/app/data/wallets/x402_wallets.json", + "/app/data/wallets/referral_wallets.json", + ] + + for path in legacy_files: + if not os.path.exists(path): + continue + + try: + with open(path) as f: + data = json.load(f) + + wallets_data = data.get("wallets", {}) + for chain_key, wdata in wallets_data.items(): + chain_map = { + "tron": "trx", + "bitcoin": "btc", + "eth": "eth", + "sol": "sol", + } + chain = chain_map.get(chain_key, chain_key) + + # Check if already imported + existing = manager.get_wallet_by_address(chain, wdata["address"]) + if existing: + logger.info(f"Legacy {chain_key} wallet already in vault") + continue + + wallet_id = f"wal_{chain}_legacy_{int(time.time())}_{secrets.token_hex(4)}" + wallet = WalletRecord( + wallet_id=wallet_id, + chain=chain, + address=wdata["address"], + public_key=wdata.get("public_key_hex", ""), + name=f"Legacy {chain_key.upper()} Wallet", + purpose=WalletPurpose.PAYMENTS.value + if path.endswith("x402_wallets.json") + else WalletPurpose.OPERATIONS.value, + tier=WalletTier.WARM.value, + tags=["legacy", "imported"], + created_at=data.get("created", datetime.now(UTC).isoformat()), + created_by="import_script", + ) + + manager._wallets[wallet_id] = wallet + imported += 1 + logger.info(f"Imported legacy wallet: {chain_key} -> {wallet_id}") + except Exception as e: + logger.error(f"Legacy import error ({path}): {e}") + + if imported > 0: + manager._save_vault() + logger.info(f"Imported {imported} legacy wallets") + + return imported + + +# ───────────────────────────────────────────────────────────────────────── +# WalletManager facade — thin class wrapper over the module-level state. +# Phase 3B of AUDIT-2026-Q3.md refactor. +# +# The bulk of wallet logic remains at module level for backward compat. +# This class provides a clean instance API for new code. +# ───────────────────────────────────────────────────────────────────────── + + +class WalletManagerFacade: + """Class-based facade for the v2 wallet manager. + + Delegates to the module-level get_wallet_manager_v2() singleton and + exposes the most common operations as instance methods. + """ + + def __init__(self, password: str = "") -> None: + self._password = password + self._manager = None + + @property + def manager(self): + if self._manager is None: + self._manager = get_wallet_manager_v2(self._password) + return self._manager + + def import_legacy_wallets(self) -> int: + return import_legacy_wallets(self.manager) + + def status(self) -> dict: + m = self.manager + return { + "wallets": len(m._wallets) if hasattr(m, "_wallets") else 0, + "chains": len(m._chain_registry) if hasattr(m, "_chain_registry") else 0, + } + + +__all__ = [ + "WalletManagerFacade", + "WalletManagerV2", + "WalletStatus", + "WalletTier", + "WalletPurpose", + "PaymentType", + "ChainMeta", + "WalletRecord", + "PaymentRecord", + "WalletRotationSchedule", + "WalletEncryption", + "ShamirSecretSharing", + "ZKAddressVerifier", + "get_wallet_manager_v2", + "import_legacy_wallets", + "_build_registry", +] diff --git a/app/wallet_manager_v2.py b/app/wallet_manager_v2.py index 90911c8..1207318 100644 --- a/app/wallet_manager_v2.py +++ b/app/wallet_manager_v2.py @@ -1,2321 +1,20 @@ -""" -RMI Wallet Manager v2 - Enterprise Multi-Chain Wallet Management -=============================================================== -A production-grade wallet management system for crypto intelligence platforms. - -Chains: 95+ across EVM, Solana, Monero, Cosmos, Polkadot, Cardano, Algorand, -Stellar, Ripple, Near, Aptos, Sui, TRON, Bitcoin variants, Litecoin, Dogecoin, -Dash, Zcash, Tezos, Nano, TON, and more. - -Features: - • HD Wallet Generation (BIP39/BIP44/BIP49/BIP84) - 95+ chains via bip_utils - • Monero Support - ed25519-blake2b keygen + stealth address derivation - • Chain Registry - metadata, validation, derivation paths per chain - • Key Rotation & Expiry - automatic scheduled rotation - • Multi-signature Support - threshold signatures for high-value wallets - • Payment Integration - x402, premium subscriptions, marketplace - • Balance Monitoring - real-time balance tracking across all chains - • Transaction History - unified tx view with categorization - • Alert System - low balance, large tx, suspicious activity alerts - • Wallet Labels & Organization - tags, groups, notes - • Cold/Hot Wallet Separation - security tier management - • API Access - programmatic wallet management for bots/agents - • Audit Trail - complete history of all wallet operations - • Backup & Recovery - Shamir's Secret Sharing for mnemonic recovery - • ZK Ownership Proof - prove wallet ownership without exposing private key - -Security: - - AES-256-GCM encryption with Argon2id key derivation (memory-hardened) - - Shamir's Secret Sharing (M-of-N) for backup mnemonic recovery - - HSM-compatible key storage interface - - Rate-limited wallet operations - - IP-restricted access for sensitive operations - - Private keys NEVER stored in plaintext - encrypted at rest, cleared from memory - - ZK address ownership verification (no private key exposure) - - GPG vault integration for master encryption password - -Author: RMI Development -Date: 2026-06-02 -""" - -import base64 -import hashlib -import json -import logging -import os -import secrets -import time -from dataclasses import asdict, dataclass, field -from datetime import UTC, datetime, timedelta -from enum import StrEnum -from typing import Any - -logger = logging.getLogger("wallet_manager_v2") - -# ── Crypto Imports ─────────────────────────────────────────────── - -try: - from cryptography.hazmat.primitives import hashes # noqa: F401 - from cryptography.hazmat.primitives.ciphers.aead import AESGCM - from cryptography.hazmat.primitives.kdf.argon2 import Argon2id - from cryptography.hazmat.primitives.kdf.hkdf import HKDF # noqa: F401 - - _HAS_CRYPTO = True -except ImportError: - _HAS_CRYPTO = False - logger.warning("cryptography not installed - encryption disabled") - -try: - from bip_utils import ( - Bip32Ed25519Slip, # noqa: F401 - Bip32Secp256k1, # noqa: F401 - Bip39MnemonicGenerator, - Bip39SeedGenerator, - Bip39WordsNum, - Bip44, - Bip44Coins, - Bip49, - Bip49Coins, - Bip84, - Bip84Coins, - Monero, # noqa: F401 - MoneroCoins, # noqa: F401 - ) - - _HAS_BIP_UTILS = True -except ImportError: - _HAS_BIP_UTILS = False - logger.warning("bip_utils not installed - using fallback generation") - -try: - import base58 - - _HAS_BASE58 = True -except ImportError: - _HAS_BASE58 = False - -try: - from nacl.bindings import crypto_sign_ed25519_sk_to_seed # noqa: F401 - from nacl.signing import SigningKey as NaClSigningKey - - _HAS_NACL = True -except ImportError: - _HAS_NACL = False - -try: - import monero # noqa: F401 - from monero import ed25519 as monero_ed25519 # noqa: F401 - from monero.address import Address as MoneroAddress # noqa: F401 - from monero.seed import Seed as MoneroSeed # noqa: F401 - - _HAS_MONERO = True -except ImportError: - _HAS_MONERO = False - logger.info("monero-python not fully available - basic XMR support only") - -# ── Enums ───────────────────────────────────────────────────────── - - -class WalletStatus(StrEnum): - ACTIVE = "active" - ROTATED = "rotated" - FROZEN = "frozen" - EXPIRED = "expired" - ARCHIVED = "archived" - COMPROMISED = "compromised" - - -class WalletTier(StrEnum): - HOT = "hot" - WARM = "warm" - COLD = "cold" - VAULT = "vault" - - -class WalletPurpose(StrEnum): - PAYMENTS = "payments" - SUBSCRIPTIONS = "subscriptions" - OPERATIONS = "operations" - TREASURY = "treasury" - USER_ESCROW = "user_escrow" - BOT_TRADING = "bot_trading" - AIRDROPS = "airdrops" - DEVELOPER = "developer" - MARKETING = "marketing" - RESERVE = "reserve" - - -class PaymentType(StrEnum): - X402 = "x402" - SUBSCRIPTION = "subscription" - ONE_TIME = "one_time" - MARKETPLACE = "marketplace" - REFUND = "refund" - WITHDRAWAL = "withdrawal" - DEPOSIT = "deposit" - FEE = "fee" - REWARD = "reward" - - -# ── Chain Registry ──────────────────────────────────────────────── -# Maps chain keys to full metadata for 20+ major non-EVM chains. -# EVM chains all share the same key derivation (secp256k1) and address -# format (0x...), so they're grouped under a single "evm" entry with -# sub-chains listed separately. - - -@dataclass -class ChainMeta: - """Metadata for a single blockchain.""" - - key: str # Short key (e.g. "sol", "xmr") - name: str # Human name (e.g. "Solana") - native_symbol: str # Native currency ticker - native_decimals: int = 18 - coin_class: Any = None # bip_utils coin class (Bip44Coins.*) - derivation: str = "bip44" # bip44, bip49, bip84, ed25519_slip, monero - address_prefix: str = "" # For address validation regex - address_pattern: str = "" # Python regex for address format - testnet_coin_class: Any = None - curve: str = "secp256k1" # secp256k1, ed25519, ed25519_blake2b - hd_path_template: str = "m/44'/{coin_type}'/{account}'/{change}/{index}" - sub_chains: list[str] = field(default_factory=list) # For EVM grouping - icon: str = "₿" # Display icon - rpc_endpoint: str = "" # Default RPC (env-overridable) - explorer_url: str = "" # Block explorer base URL - min_confirmation_blocks: int = 1 - supports_tokens: bool = True - supports_nfts: bool = False - - -# ── Complete Chain Registry ─────────────────────────────────────── - -CHAIN_REGISTRY: dict[str, ChainMeta] = {} - - -def _build_registry(): - """Build the comprehensive chain registry.""" - reg = {} - - # ── EVM Group (9 chains, all secp256k1, all 0x addresses) ── - evm_chains = [ - ("eth", "Ethereum", 60, Bip44Coins.ETHEREUM), - ("base", "Base", 8453, None), # Base uses ETH derivation - ("polygon", "Polygon", 137, Bip44Coins.POLYGON), - ("arbitrum", "Arbitrum One", 42161, Bip44Coins.ARBITRUM), - ("optimism", "Optimism", 10, Bip44Coins.OPTIMISM), - ("avalanche", "Avalanche C-Chain", 43114, Bip44Coins.AVAX_C_CHAIN), - ("bsc", "BNB Chain", 56, Bip44Coins.BINANCE_SMART_CHAIN), - ("fantom", "Fantom", 250, Bip44Coins.FANTOM_OPERA), - ("gnosis", "Gnosis", 100, Bip44Coins.ETHEREUM), # Gnosis uses ETH derivation - ] - evm_subs = [c[0] for c in evm_chains] - for key, name, _chain_id, coin_class in evm_chains: - reg[key] = ChainMeta( - key=key, - name=name, - native_symbol="ETH" if key != "bsc" else "BNB", - native_decimals=18, - derivation="bip44", - coin_class=coin_class if coin_class else Bip44Coins.ETHEREUM, - address_pattern=r"^0x[a-fA-F0-9]{40}$", - curve="secp256k1", - hd_path_template="m/44'/{coin_type}'/{account}'/{change}/{index}", - sub_chains=evm_subs, - icon="⟠", - explorer_url=f"https://{'etherscan.io' if key == 'eth' else key + ('scan.com' if key in ['bsc', 'polygon'] else '.blockscout.com')}", - ) - - # ── Solana ── - reg["sol"] = ChainMeta( - key="sol", - name="Solana", - native_symbol="SOL", - native_decimals=9, - coin_class=Bip44Coins.SOLANA, - derivation="bip44_ed25519", - address_pattern=r"^[1-9A-HJ-NP-Za-km-z]{32,44}$", - curve="ed25519", - hd_path_template="m/44'/501'/{account}'/{change}'/{index}", - icon="◎", - explorer_url="https://solscan.io", - supports_nfts=True, - ) - - # ── Monero ── - reg["xmr"] = ChainMeta( - key="xmr", - name="Monero", - native_symbol="XMR", - native_decimals=12, - coin_class=None, # Monero uses custom crypto - derivation="monero", - address_pattern=r"^[48][0-9AB][1-9A-HJ-NP-Za-km-z]{93}$", - curve="ed25519_blake2b", - hd_path_template="m/44'/128'/{account}'/{change}/{index}", - icon="ɱ", - explorer_url="https://xmrchain.net", - supports_tokens=False, - ) - - # ── Cosmos / IBC Ecosystem ── - cosmos_chains = [ - ("atom", "Cosmos Hub", Bip44Coins.COSMOS, 118, "uatom"), - ("osmo", "Osmosis", Bip44Coins.OSMOSIS, 118, "uosmo"), - ("juno", "Juno", Bip44Coins.COSMOS, 118, "ujuno"), # Uses Cosmos derivation - ("secret", "Secret Network", Bip44Coins.SECRET_NETWORK_NEW, 529, "uscrt"), - ("inj", "Injective", Bip44Coins.INJECTIVE, 60, "inj"), - ] - for key, name, coin_cls, _coin_type, _denom in cosmos_chains: - reg[key] = ChainMeta( - key=key, - name=name, - native_symbol=key.upper(), - native_decimals=6, - coin_class=coin_cls, - derivation="bip44_cosmos", - address_pattern=r"^[a-z]+1[a-z0-9]{38,58}$", - curve="secp256k1", - hd_path_template="m/44'/{coin_type}'/{account}'/{change}/{index}", - icon="⚛", - explorer_url=f"https://www.mintscan.io/{key if key != 'secret' else 'secret'}", - ) - - # ── Polkadot / Substrate ── - for key, name, coin_cls in [ - ("dot", "Polkadot", Bip44Coins.POLKADOT_ED25519_SLIP), - ("ksm", "Kusama", Bip44Coins.KUSAMA_ED25519_SLIP), - ]: - reg[key] = ChainMeta( - key=key, - name=name, - native_symbol=key.upper(), - native_decimals=10 if key == "dot" else 12, - coin_class=coin_cls, - derivation="bip44_ed25519_slip", - address_pattern=r"^[1-9A-HJ-NP-Za-km-z]{47,48}$", - curve="ed25519", - hd_path_template="m/44'/354'/{account}'/{change}'/{index}" - if key == "dot" - else "m/44'/434'/{account}'/{change}'/{index}", - icon="⬡", - explorer_url=f"https://{key}.subscan.io", - ) - - # ── Cardano ── - reg["ada"] = ChainMeta( - key="ada", - name="Cardano", - native_symbol="ADA", - native_decimals=6, - coin_class=Bip44Coins.CARDANO_BYRON_ICARUS, - derivation="bip44_ed25519", - address_pattern=r"^addr1[a-z0-9]{50,100}$|^Ae2[a-zA-Z0-9]{50,}$", - curve="ed25519", - hd_path_template="m/44'/1815'/{account}'/{change}/{index}", - icon="₳", - explorer_url="https://cardanoscan.io", - supports_nfts=True, - ) - - # ── Algorand ── - reg["algo"] = ChainMeta( - key="algo", - name="Algorand", - native_symbol="ALGO", - native_decimals=6, - coin_class=Bip44Coins.ALGORAND, - derivation="bip44_ed25519", - address_pattern=r"^[A-Z0-9]{58}$", - curve="ed25519", - hd_path_template="m/44'/283'/{account}'/{change}'/{index}", - icon="🅰", - explorer_url="https://algoexplorer.io", - ) - - # ── Stellar ── - reg["xlm"] = ChainMeta( - key="xlm", - name="Stellar", - native_symbol="XLM", - native_decimals=7, - coin_class=Bip44Coins.STELLAR, - derivation="bip44_ed25519", - address_pattern=r"^G[A-Z0-9]{55}$", - curve="ed25519", - hd_path_template="m/44'/148'/{account}'/{change}'/{index}", - icon="⭐", - explorer_url="https://stellar.expert", - ) - - # ── Ripple ── - reg["xrp"] = ChainMeta( - key="xrp", - name="Ripple", - native_symbol="XRP", - native_decimals=6, - coin_class=Bip44Coins.RIPPLE, - derivation="bip44_secp256k1", - address_pattern=r"^r[rpshnaf39wBUDNEGHJKLM4PQRST7VWXYZ2bcdeCg65jkm8oFqi1tuvAxyz]{27,34}$", - curve="secp256k1", - hd_path_template="m/44'/144'/{account}'/{change}/{index}", - icon="💧", - explorer_url="https://xrpscan.com", - ) - - # ── Near ── - reg["near"] = ChainMeta( - key="near", - name="Near Protocol", - native_symbol="NEAR", - native_decimals=24, - coin_class=Bip44Coins.NEAR_PROTOCOL, - derivation="bip44_ed25519_slip", - address_pattern=r"^[a-z0-9._-]{2,64}\.near$|^[0-9a-fA-F]{64}$", - curve="ed25519", - hd_path_template="m/44'/397'/{account}'/{change}'/{index}", - icon="🌐", - explorer_url="https://nearblocks.io", - ) - - # ── Aptos ── - reg["apt"] = ChainMeta( - key="apt", - name="Aptos", - native_symbol="APT", - native_decimals=8, - coin_class=Bip44Coins.APTOS, - derivation="bip44_ed25519", - address_pattern=r"^0x[a-fA-F0-9]{64}$", - curve="ed25519", - hd_path_template="m/44'/637'/{account}'/{change}'/{index}", - icon="🏗", - explorer_url="https://aptoscan.com", - ) - - # ── Sui ── - reg["sui"] = ChainMeta( - key="sui", - name="Sui", - native_symbol="SUI", - native_decimals=9, - coin_class=Bip44Coins.SUI, - derivation="bip44_ed25519", - address_pattern=r"^0x[a-fA-F0-9]{64}$", - curve="ed25519", - hd_path_template="m/44'/784'/{account}'/{change}'/{index}", - icon="💧", - explorer_url="https://suiscan.xyz", - ) - - # ── TRON ── - reg["trx"] = ChainMeta( - key="trx", - name="TRON", - native_symbol="TRX", - native_decimals=6, - coin_class=Bip44Coins.TRON, - derivation="bip44_secp256k1", - address_pattern=r"^T[A-Za-z1-9]{33}$", - curve="secp256k1", - hd_path_template="m/44'/195'/{account}'/{change}/{index}", - icon="🔷", - explorer_url="https://tronscan.org", - ) - - # ── Bitcoin (multiple address formats) ── - reg["btc"] = ChainMeta( - key="btc", - name="Bitcoin (Legacy)", - native_symbol="BTC", - native_decimals=8, - coin_class=Bip44Coins.BITCOIN, - derivation="bip44", - address_pattern=r"^[13][a-km-zA-HJ-NP-Z1-9]{25,34}$", - curve="secp256k1", - hd_path_template="m/44'/0'/{account}'/{change}/{index}", - icon="₿", - explorer_url="https://mempool.space", - supports_tokens=False, - ) - reg["btc-segwit"] = ChainMeta( - key="btc-segwit", - name="Bitcoin (SegWit)", - native_symbol="BTC", - native_decimals=8, - coin_class=None, - derivation="bip49", - address_pattern=r"^3[a-km-zA-HJ-NP-Z1-9]{25,34}$", - curve="secp256k1", - hd_path_template="m/49'/0'/{account}'/{change}/{index}", - icon="₿", - supports_tokens=False, - ) - reg["btc-native-segwit"] = ChainMeta( - key="btc-native-segwit", - name="Bitcoin (Native SegWit)", - native_symbol="BTC", - native_decimals=8, - coin_class=None, - derivation="bip84", - address_pattern=r"^bc1[a-z0-9]{39,59}$", - curve="secp256k1", - hd_path_template="m/84'/0'/{account}'/{change}/{index}", - icon="₿", - supports_tokens=False, - ) - - # ── Bitcoin Forks ── - for key, name, coin_cls in [ - ("ltc", "Litecoin", Bip44Coins.LITECOIN), - ("doge", "Dogecoin", Bip44Coins.DOGECOIN), - ("dash", "Dash", Bip44Coins.DASH), - ("bch", "Bitcoin Cash", Bip44Coins.BITCOIN_CASH), - ("zec", "Zcash", Bip44Coins.ZCASH), - ]: - reg[key] = ChainMeta( - key=key, - name=name, - native_symbol=key.upper(), - native_decimals=8, - coin_class=coin_cls, - derivation="bip44", - address_pattern=r"^[a-zA-Z0-9]{26,42}$", - curve="secp256k1", - hd_path_template="m/44'/{coin_type}'/{account}'/{change}/{index}", - icon="🪙", - supports_tokens=False, - ) - - # ── Tezos ── - reg["xtz"] = ChainMeta( - key="xtz", - name="Tezos", - native_symbol="XTZ", - native_decimals=6, - coin_class=Bip44Coins.TEZOS, - derivation="bip44_ed25519", - address_pattern=r"^tz[1-3][1-9A-HJ-NP-Za-km-z]{33}$", - curve="ed25519", - hd_path_template="m/44'/1729'/{account}'/{change}/{index}", - icon="ꜩ", - explorer_url="https://tzkt.io", - ) - - # ── Nano ── - reg["nano"] = ChainMeta( - key="nano", - name="Nano", - native_symbol="XNO", - native_decimals=30, - coin_class=Bip44Coins.NANO, - derivation="bip44_ed25519_blake2b", - address_pattern=r"^(nano|xrb)_[13][a-z0-9]{59}$", - curve="ed25519_blake2b", - hd_path_template="m/44'/165'/{account}'/{change}'/{index}", - icon="⚡", - explorer_url="https://nanocrawler.cc", - supports_tokens=False, - ) - - # ── TON ── - reg["ton"] = ChainMeta( - key="ton", - name="TON", - native_symbol="TON", - native_decimals=9, - coin_class=Bip44Coins.TON, - derivation="bip44_ed25519", - address_pattern=r"^(EQ|UQ)[a-zA-Z0-9_-]{46}$", - curve="ed25519", - hd_path_template="m/44'/607'/{account}'/{change}/{index}", - icon="💎", - explorer_url="https://tonscan.org", - ) - - # ── Band Protocol ── - reg["band"] = ChainMeta( - key="band", - name="Band Protocol", - native_symbol="BAND", - native_decimals=6, - coin_class=Bip44Coins.BAND_PROTOCOL, - derivation="bip44_cosmos", - address_pattern=r"^band[a-z0-9]{38,58}$", - curve="secp256k1", - hd_path_template="m/44'/494'/{account}'/{change}'/{index}", - icon="📡", - ) - - return reg - - -CHAIN_REGISTRY = _build_registry() - - -# ── Data Models ───────────────────────────────────────────────── - - -@dataclass -class WalletRecord: - """Complete wallet record with metadata.""" - - wallet_id: str - chain: str - address: str - public_key: str = "" - - # Identity - name: str = "" - description: str = "" - purpose: str = WalletPurpose.OPERATIONS.value - tier: str = WalletTier.WARM.value - status: str = WalletStatus.ACTIVE.value - - # Organization - tags: list[str] = field(default_factory=list) - group: str = "default" - labels: dict[str, str] = field(default_factory=dict) - - # Security - created_at: str = "" - rotated_at: str | None = None - expires_at: str | None = None - last_used: str | None = None - use_count: int = 0 - - # Balance tracking - balance_raw: str = "0" - balance_decimal: float = 0.0 - balance_usd: float = 0.0 - token_balances: dict[str, dict] = field(default_factory=dict) - - # Transaction tracking - total_received: float = 0.0 - total_sent: float = 0.0 - tx_count: int = 0 - last_tx_hash: str = "" - last_tx_time: str | None = None - - # Payment integration - x402_enabled: bool = False - x402_price_usd: float = 0.0 - subscription_enabled: bool = False - subscription_tiers: list[str] = field(default_factory=list) - - # Verification - verified_at: str | None = None - verified_by: str = "" - chain_registered: bool = False # x402 chain registration check - - # Alerts - alert_threshold_usd: float = 100.0 - low_balance_threshold: float = 0.0 - - # Audit - created_by: str = "" - notes: str = "" - version: int = 1 - - def to_dict(self) -> dict: - return asdict(self) - - def to_safe_dict(self) -> dict: - """Return without sensitive data.""" - d = self.to_dict() - d.pop("public_key", None) - return d - - -@dataclass -class PaymentRecord: - """Payment transaction record.""" - - payment_id: str - wallet_id: str - wallet_address: str - chain: str - payment_type: str - - amount: float = 0.0 - amount_usd: float = 0.0 - token: str = "" - - from_address: str = "" - to_address: str = "" - tx_hash: str = "" - block_number: int = 0 - - status: str = "pending" - confirmations: int = 0 - - user_id: str = "" - user_email: str = "" - tool_id: str = "" - tool_name: str = "" - - x402_resource: str = "" - x402_facet: str = "" - - created_at: str = "" - confirmed_at: str | None = None - - metadata: dict[str, Any] = field(default_factory=dict) - - def to_dict(self) -> dict: - return asdict(self) - - -@dataclass -class WalletRotationSchedule: - """Schedule for automatic wallet rotation.""" - - wallet_id: str - chain: str - rotate_every_days: int = 90 - auto_rotate: bool = False - notify_before_days: int = 7 - last_rotated: str | None = None - next_rotation: str | None = None - - def to_dict(self) -> dict: - return asdict(self) - - -# ── Encryption Layer ────────────────────────────────────────────── - - -class WalletEncryption: - """ - AES-256-GCM encryption with memory-hardened Argon2id key derivation. - - Security guarantees: - - Argon2id with high memory cost (128MB) for brute-force resistance - - AES-256-GCM authenticated encryption (confidentiality + integrity) - - Unique random salt + nonce per encryption (no reuse) - - Keys derived per-operation, never cached - """ - - # Memory-hard parameters (OWASP 2026 recommendations) - ARGON_TIME = 4 # Iterations - ARGON_MEMORY = 131072 # 128 MB - ARGON_PARALLELISM = 4 # Threads - - @staticmethod - def _derive_key(password: str, salt: bytes) -> bytes: - """Derive encryption key using Argon2id.""" - if _HAS_CRYPTO: - kdf = Argon2id( - salt=salt, - length=32, - iterations=WalletEncryption.ARGON_TIME, - lanes=WalletEncryption.ARGON_PARALLELISM, - memory_cost=WalletEncryption.ARGON_MEMORY, - ) - return kdf.derive(password.encode()) - else: - import hashlib - - return hashlib.pbkdf2_hmac("sha256", password.encode(), salt, 600000, 32) - - @staticmethod - def encrypt(plaintext: str, password: str) -> str: - """Encrypt plaintext. Returns 'ENC_V2:' or 'DEV:'.""" - if not password: - raise ValueError("Encryption requires a password") - - if not _HAS_CRYPTO: - logger.warning("cryptography not installed - using dev fallback") - return "DEV:" + base64.b64encode(plaintext.encode()).decode() - - salt = os.urandom(16) - key = WalletEncryption._derive_key(password, salt) - aesgcm = AESGCM(key) - nonce = os.urandom(12) - ciphertext = aesgcm.encrypt(nonce, plaintext.encode(), None) - - combined = salt + nonce + ciphertext - # Wipe key from memory - key = b"\x00" * 32 - return "ENC_V2:" + base64.b64encode(combined).decode() - - @staticmethod - def decrypt(ciphertext: str, password: str) -> str: - """Decrypt ciphertext. Raises on wrong password or corruption.""" - if ciphertext.startswith("DEV:"): - return base64.b64decode(ciphertext[4:]).decode() - - prefix = "ENC_V2:" if ciphertext.startswith("ENC_V2:") else "ENC:" - offset = len(prefix) - - if not _HAS_CRYPTO: - raise RuntimeError("cryptography library required for decryption") - - combined = base64.b64decode(ciphertext[offset:]) - salt = combined[:16] - nonce = combined[16:28] - ct = combined[28:] - - key = WalletEncryption._derive_key(password, salt) - aesgcm = AESGCM(key) - plaintext = aesgcm.decrypt(nonce, ct, None) - - result = plaintext.decode() - # Wipe key from memory - key = b"\x00" * 32 - return result - - -# ── Shamir's Secret Sharing ────────────────────────────────────── - - -class ShamirSecretSharing: - """ - M-of-N secret sharing for mnemonic recovery. - - Split a mnemonic into N shares, any M of which can reconstruct it. - Uses finite field arithmetic over GF(256) with simple polynomial evaluation. - - Production note: For real deployment, use the sss-python library or - the cryptography.hazmat.primitives.secret_sharing module. - This implementation is a self-contained, auditable reference. - """ - - @staticmethod - def split(secret_bytes: bytes, threshold: int, total_shares: int) -> list[tuple[int, bytes]]: - """ - Split secret into total_shares shares, threshold needed to recover. - - Returns list of (x, y) tuples where x is the share index (1-based) - and y is the share value. Each share is the same length as the secret. - """ - if threshold > total_shares: - raise ValueError("threshold must be <= total_shares") - if threshold < 2: - raise ValueError("threshold must be >= 2") - if len(secret_bytes) == 0: - raise ValueError("secret cannot be empty") - - # Generate random coefficients for polynomial of degree (threshold-1) - coeffs = [secret_bytes] # coeffs[0] = secret - for i in range(1, threshold): # noqa: B007 - coeffs.append(os.urandom(len(secret_bytes))) - - # Evaluate polynomial at points x=1..total_shares - shares = [] - for x in range(1, total_shares + 1): - # y = coeffs[0] + coeffs[1]*x + coeffs[2]*x^2 + ... - y = bytearray(len(secret_bytes)) - for i, coeff in enumerate(coeffs): - if i == 0: - for j in range(len(y)): - y[j] ^= coeff[j] # XOR identity - else: - multiplier = ShamirSecretSharing._gf_pow(x, i) - for j in range(len(y)): - y[j] ^= ShamirSecretSharing._gf_mul(coeff[j], multiplier) - shares.append((x, bytes(y))) - - return shares - - @staticmethod - def combine(shares: list[tuple[int, bytes]]) -> bytes: - """Reconstruct secret from M shares using Lagrange interpolation.""" - if len(shares) < 2: - raise ValueError("need at least 2 shares") - - secret_len = len(shares[0][1]) - result = bytearray(secret_len) - - for i, (xi, yi) in enumerate(shares): - # Compute Lagrange basis polynomial L_i(0) - numerator = 1 - denominator = 1 - for j, (xj, _) in enumerate(shares): - if i != j: - numerator = ShamirSecretSharing._gf_mul(numerator, xj) - denominator = ShamirSecretSharing._gf_mul(denominator, ShamirSecretSharing._gf_sub(xj, xi)) - - lagrange_coeff = ShamirSecretSharing._gf_div(numerator, denominator) - - for k in range(secret_len): - result[k] ^= ShamirSecretSharing._gf_mul(yi[k], lagrange_coeff) - - return bytes(result) - - # GF(256) arithmetic using Rijndael's finite field - @staticmethod - def _gf_mul(a: int, b: int) -> int: - """Multiply in GF(256).""" - p = 0 - for _ in range(8): - if b & 1: - p ^= a - hi_bit = a & 0x80 - a = (a << 1) & 0xFF - if hi_bit: - a ^= 0x1B # Rijndael's irreducible polynomial - b >>= 1 - return p - - @staticmethod - def _gf_pow(base: int, exp: int) -> int: - """Exponentiation in GF(256).""" - result = 1 - for _ in range(exp): - result = ShamirSecretSharing._gf_mul(result, base) - return result - - @staticmethod - def _gf_sub(a: int, b: int) -> int: - """Subtraction (same as addition/XOR in GF(256)).""" - return a ^ b - - @staticmethod - def _gf_div(a: int, b: int) -> int: - """Division in GF(256).""" - if b == 0: - raise ZeroDivisionError("division by zero in GF(256)") - # Find multiplicative inverse of b - inverse = ShamirSecretSharing._gf_inverse(b) - return ShamirSecretSharing._gf_mul(a, inverse) - - @staticmethod - def _gf_inverse(a: int) -> int: - """Multiplicative inverse in GF(256) using extended Euclidean algorithm.""" - if a == 0: - raise ValueError("cannot invert zero") - # Fermat's little theorem: a^(254) = a^(-1) in GF(256) - result = 1 - base = a - for _ in range(7): # 254 = 11111110 in binary - result = ShamirSecretSharing._gf_mul(result, base) - base = ShamirSecretSharing._gf_mul(base, base) - return result - - -# ── Zero-Knowledge Address Ownership ───────────────────────────── - - -class ZKAddressVerifier: - """ - Prove wallet ownership without exposing the private key. - - Uses a simple challenge-response protocol: - 1. Verifier sends random challenge - 2. Prover signs challenge with private key - 3. Verifier checks signature against public key/address - - For production, would integrate with: - - EIP-712 typed data signing (EVM) - - ed25519-dalek signatures (Solana, Cosmos) - - Monero's MLSAG ring signatures - """ - - @staticmethod - def generate_challenge() -> str: - """Generate a random challenge for ownership verification.""" - return secrets.token_hex(32) - - @staticmethod - def verify_evm_ownership(address: str, challenge: str, signature: str) -> bool: - """ - Verify EVM wallet ownership via personal_sign. - - In production: use web3.eth.account.recover_message() or similar. - Currently returns True if signature is plausible (correct length). - """ - # Basic validation - if not address.startswith("0x") or len(address) != 42: - return False - # Placeholder - real verification needs web3.py - return signature.startswith("0x") and len(signature) >= 130 - - @staticmethod - def verify_ed25519_ownership(public_key_hex: str, challenge: str, signature_hex: str) -> bool: - """Verify ed25519 ownership.""" - if _HAS_NACL: - try: - from nacl.signing import VerifyKey - - vk = VerifyKey(bytes.fromhex(public_key_hex)) - vk.verify(challenge.encode(), bytes.fromhex(signature_hex)) - return True - except Exception: - return False - return True # Placeholder when nacl not available - - -# ── Core Wallet Manager ───────────────────────────────────────── - - -class WalletManagerV2: - """ - Enterprise wallet management system - 95+ chains. - Handles generation, rotation, monitoring, payments, and recovery. - """ - - # Persistent storage - under /app which is volume-mounted from host - VAULT_PATH = "/app/data/wallets/vault_v2.json" - KEYSTORE_PATH = "/app/data/wallets/keystore_v2.enc" - PAYMENTS_PATH = "/app/data/wallets/payments_v2.jsonl" - SHARES_PATH = "/app/data/wallets/shamir_shares/" - - def __init__(self, encryption_password: str = ""): - self.encryption_password = encryption_password or os.getenv("WALLET_VAULT_PASSWORD", "") - self._ensure_dirs() - self._wallets: dict[str, WalletRecord] = {} - self._payments: list[PaymentRecord] = [] - self._load_vault() - self._load_payments() - - def _ensure_dirs(self): - """Ensure wallet directories exist with secure permissions.""" - for p in [os.path.dirname(self.VAULT_PATH), self.SHARES_PATH]: - os.makedirs(p, mode=0o700, exist_ok=True) - - def _load_vault(self): - """Load wallet vault from disk.""" - if os.path.exists(self.VAULT_PATH): - try: - with open(self.VAULT_PATH) as f: - data = json.load(f) - for wdata in data.get("wallets", []): - wr = WalletRecord(**wdata) - self._wallets[wr.wallet_id] = wr - logger.info(f"Vault loaded: {len(self._wallets)} wallets") - except Exception as e: - logger.error(f"Vault load error: {e}") - else: - logger.info("No vault found - starting fresh") - - def _save_vault(self): - """Save wallet vault to disk.""" - data = { - "version": "3.0", - "saved_at": datetime.now(UTC).isoformat(), - "wallet_count": len(self._wallets), - "chains": sorted({w.chain for w in self._wallets.values()}), - "wallets": [w.to_dict() for w in self._wallets.values()], - } - try: - with open(self.VAULT_PATH, "w") as f: - json.dump(data, f, indent=2) - os.chmod(self.VAULT_PATH, 0o600) - except Exception as e: - logger.error(f"Vault save error: {e}") - - def _load_payments(self): - """Load payment records from disk.""" - if os.path.exists(self.PAYMENTS_PATH): - try: - with open(self.PAYMENTS_PATH) as f: - for line in f: - line = line.strip() - if line: - self._payments.append(PaymentRecord(**json.loads(line))) - except Exception as e: - logger.error(f"Payments load error: {e}") - - # ── Chain Registry Access ───────────────────────────────── - - @staticmethod - def get_chain_meta(chain: str) -> ChainMeta | None: - """Get chain metadata.""" - return CHAIN_REGISTRY.get(chain) - - @staticmethod - def list_chains() -> list[str]: - """List all supported chain keys.""" - return sorted(CHAIN_REGISTRY.keys()) - - @staticmethod - def get_chains_by_group() -> dict[str, list[str]]: - """Get chains grouped by ecosystem.""" - groups = { - "evm": [ - k - for k, v in CHAIN_REGISTRY.items() - if v.curve == "secp256k1" - and v.key - in [ - "eth", - "base", - "polygon", - "arbitrum", - "optimism", - "avalanche", - "bsc", - "fantom", - "gnosis", - ] - ], - "cosmos": ["atom", "osmo", "juno", "secret", "inj"], - "bitcoin": [ - "btc", - "btc-segwit", - "btc-native-segwit", - "ltc", - "doge", - "dash", - "bch", - "zec", - ], - "privacy": ["xmr"], - "l1_alt": [ - "sol", - "dot", - "ksm", - "ada", - "algo", - "xlm", - "xrp", - "near", - "apt", - "sui", - "trx", - "xtz", - "nano", - "ton", - "band", - ], - } - return groups - - # ── Wallet Generation ───────────────────────────────────── - - def generate_wallet( - self, - chain: str, - name: str = "", - purpose: str = WalletPurpose.OPERATIONS.value, - tier: str = WalletTier.WARM.value, - tags: list[str] | None = None, - group: str = "default", - created_by: str = "", - ) -> WalletRecord: - """Generate a new wallet for any supported chain.""" - - chain_meta = CHAIN_REGISTRY.get(chain) - if not chain_meta: - raise ValueError(f"Unsupported chain: {chain}. Supported: {sorted(CHAIN_REGISTRY.keys())}") - - wallet_id = f"wal_{chain}_{int(time.time())}_{secrets.token_hex(4)}" - - # Generate keys using chain-appropriate method - if chain_meta.derivation in ("bip44", "bip49", "bip84") and chain_meta.curve == "secp256k1": - address, public_key, mnemonic = self._generate_bip_wallet(chain_meta) - elif chain_meta.derivation == "bip44_ed25519" or chain_meta.derivation == "bip44_ed25519_slip": - address, public_key, mnemonic = self._generate_ed25519_wallet(chain_meta) - elif chain_meta.derivation == "bip44_cosmos": - address, public_key, mnemonic = self._generate_cosmos_wallet(chain_meta) - elif chain_meta.derivation == "monero": - address, public_key, mnemonic = self._generate_monero_wallet(chain_meta) - elif chain_meta.derivation == "bip44_secp256k1": - address, public_key, mnemonic = self._generate_bip_wallet(chain_meta) - elif chain_meta.derivation == "bip44_ed25519_blake2b": - address, public_key, mnemonic = self._generate_ed25519_wallet(chain_meta) - else: - address, public_key, mnemonic = self._generate_bip_wallet(chain_meta) - - wallet = WalletRecord( - wallet_id=wallet_id, - chain=chain, - address=address, - public_key=public_key, - name=name or f"{chain_meta.name} Wallet", - purpose=purpose, - tier=tier, - tags=tags or [], - group=group, - created_at=datetime.now(UTC).isoformat(), - created_by=created_by, - ) - - # Store mnemonic encrypted - if mnemonic and self.encryption_password: - self._store_mnemonic(wallet_id, mnemonic) - wallet.labels["mnemonic"] = "encrypted" - - self._wallets[wallet_id] = wallet - self._save_vault() - - logger.info(f"Wallet generated: {wallet_id} ({chain}) - {address}") - return wallet - - def _generate_bip_wallet(self, chain_meta: ChainMeta) -> tuple[str, str, str]: - """Generate BIP44/49/84 wallet.""" - if not _HAS_BIP_UTILS: - return self._generate_fallback(chain_meta) - - mnemonic = Bip39MnemonicGenerator().FromWordsNumber(Bip39WordsNum.WORDS_NUM_24).ToStr() - seed = Bip39SeedGenerator(mnemonic).Generate() - - if chain_meta.derivation == "bip49" and Bip49Coins: - bip_ctx = Bip49.FromSeed(seed, Bip49Coins.BITCOIN) - elif chain_meta.derivation == "bip84" and Bip84Coins: - bip_ctx = Bip84.FromSeed(seed, Bip84Coins.BITCOIN) - elif chain_meta.coin_class: - bip_ctx = Bip44.FromSeed(seed, chain_meta.coin_class) - else: - bip_ctx = Bip44.FromSeed(seed, Bip44Coins.ETHEREUM) - - address = bip_ctx.PublicKey().ToAddress() - pub_key = bip_ctx.PublicKey().RawCompressed().ToHex() - - return address, pub_key, mnemonic - - def _generate_ed25519_wallet(self, chain_meta: ChainMeta) -> tuple[str, str, str]: - """Generate ed25519-based wallet (Solana, Cardano, Algorand, Stellar, Near, etc.).""" - if not _HAS_BIP_UTILS: - return self._generate_fallback(chain_meta) - - mnemonic = Bip39MnemonicGenerator().FromWordsNumber(Bip39WordsNum.WORDS_NUM_24).ToStr() - seed = Bip39SeedGenerator(mnemonic).Generate() - - if chain_meta.coin_class: - bip_ctx = Bip44.FromSeed(seed, chain_meta.coin_class) - else: - bip_ctx = Bip44.FromSeed(seed, Bip44Coins.SOLANA) - - address = bip_ctx.PublicKey().ToAddress() - pub_key = bip_ctx.PublicKey().RawCompressed().ToHex() - - return address, pub_key, mnemonic - - def _generate_cosmos_wallet(self, chain_meta: ChainMeta) -> tuple[str, str, str]: - """Generate Cosmos/IBC wallet.""" - if not _HAS_BIP_UTILS: - return self._generate_fallback(chain_meta) - - mnemonic = Bip39MnemonicGenerator().FromWordsNumber(Bip39WordsNum.WORDS_NUM_24).ToStr() - seed = Bip39SeedGenerator(mnemonic).Generate() - - if chain_meta.coin_class: - bip_ctx = Bip44.FromSeed(seed, chain_meta.coin_class) - else: - bip_ctx = Bip44.FromSeed(seed, Bip44Coins.COSMOS) - - address = bip_ctx.PublicKey().ToAddress() - pub_key = bip_ctx.PublicKey().RawCompressed().ToHex() - - return address, pub_key, mnemonic - - def _generate_monero_wallet(self, chain_meta: ChainMeta) -> tuple[str, str, str]: - """Generate Monero wallet using ed25519-blake2b.""" - mnemonic = "" - - if _HAS_BIP_UTILS: - try: - # Monero uses 25-word mnemonics - mnemonic = Bip39MnemonicGenerator().FromWordsNumber(Bip39WordsNum.WORDS_NUM_24).ToStr() - seed = Bip39SeedGenerator(mnemonic).Generate() - - # Use bip_utils Monero support - monero_ctx = Bip44.FromSeed(seed, Bip44Coins.MONERO_ED25519_SLIP) - address = monero_ctx.PublicKey().ToAddress() - pub_key = monero_ctx.PublicKey().RawCompressed().ToHex() - return address, pub_key, mnemonic - except Exception as e: - logger.warning(f"bip_utils Monero failed ({e}), trying fallback") - - # Fallback: raw ed25519-blake2b keypair - if _HAS_NACL: - sk = NaClSigningKey.generate() - pub_key = sk.verify_key.encode().hex() - - # Monero-style address (simplified) - # Real Monero addresses require keccak + base58 encoding - h = hashlib.sha3_256() - h.update(bytes.fromhex(pub_key)) - raw_addr = h.digest() - address = "4" + base58.b58encode(raw_addr[:64]).decode()[:94] if _HAS_BASE58 else "4" + raw_addr.hex()[:94] - - return address, pub_key, mnemonic - - # Ultimate fallback - priv = secrets.token_hex(32) - pub = hashlib.sha256(priv.encode()).hexdigest() - return "4" + pub[:94], pub, mnemonic - - def _generate_fallback(self, chain_meta: ChainMeta) -> tuple[str, str, str]: - """Fallback wallet generation (dev only, not for production).""" - logger.warning(f"Using fallback keygen for {chain_meta.key} - NOT SECURE FOR PRODUCTION") - priv = secrets.token_hex(32) - pub = hashlib.sha256(priv.encode()).hexdigest() - - prefix_map = { - "eth": "0x", - "base": "0x", - "polygon": "0x", - "arbitrum": "0x", - "optimism": "0x", - "avalanche": "0x", - "bsc": "0x", - "fantom": "0x", - "gnosis": "0x", - "sol": "", - "trx": "T", - "xrp": "r", - } - prefix = prefix_map.get(chain_meta.key, chain_meta.key + "_") - addr = prefix + pub[:40] - - return addr, pub, "" - - def generate_hd_wallet( - self, - chain: str, - mnemonic: str = "", - account_index: int = 0, - address_index: int = 0, - name: str = "", - purpose: str = WalletPurpose.OPERATIONS.value, - tier: str = WalletTier.WARM.value, - tags: list[str] | None = None, - group: str = "default", - created_by: str = "", - ) -> WalletRecord: - """Generate HD wallet from mnemonic or create new one.""" - - if not mnemonic and _HAS_BIP_UTILS: - mnemonic = Bip39MnemonicGenerator().FromWordsNumber(Bip39WordsNum.WORDS_NUM_24).ToStr() - - wallet = self.generate_wallet( - chain=chain, - name=name, - purpose=purpose, - tier=tier, - tags=tags, - group=group, - created_by=created_by, - ) - - if mnemonic: - self._store_mnemonic(wallet.wallet_id, mnemonic) - - wallet.labels["hd_account"] = str(account_index) - wallet.labels["hd_address"] = str(address_index) - - self._save_vault() - return wallet - - def generate_batch( - self, - chains: list[str], - name_prefix: str = "", - purpose: str = WalletPurpose.PAYMENTS.value, - tier: str = WalletTier.WARM.value, - group: str = "x402", - created_by: str = "", - ) -> list[WalletRecord]: - """Batch generate wallets for multiple chains.""" - wallets = [] - for chain in chains: - try: - wallet = self.generate_wallet( - chain=chain, - name=f"{name_prefix} {chain.upper()}".strip() if name_prefix else f"{chain.upper()} Payment", - purpose=purpose, - tier=tier, - group=group, - created_by=created_by, - ) - wallets.append(wallet) - except Exception as e: - logger.error(f"Failed to generate {chain} wallet: {e}") - return wallets - - # ── Key Storage ─────────────────────────────────────────── - - def _store_mnemonic(self, wallet_id: str, mnemonic: str): - """Store encrypted mnemonic in keystore.""" - if not self.encryption_password: - logger.warning("No encryption password - mnemonic not stored securely") - return - - encrypted = WalletEncryption.encrypt(mnemonic, self.encryption_password) - - keystore = {} - if os.path.exists(self.KEYSTORE_PATH): - try: - with open(self.KEYSTORE_PATH) as f: - keystore = json.load(f) - except Exception: - pass - - keystore[wallet_id] = { - "encrypted_mnemonic": encrypted, - "stored_at": datetime.now(UTC).isoformat(), - "encryption_version": "V2_Argon2id", - } - - with open(self.KEYSTORE_PATH, "w") as f: - json.dump(keystore, f, indent=2) - os.chmod(self.KEYSTORE_PATH, 0o600) - - def get_mnemonic(self, wallet_id: str) -> str | None: - """Retrieve and decrypt mnemonic.""" - if not os.path.exists(self.KEYSTORE_PATH): - return None - - try: - with open(self.KEYSTORE_PATH) as f: - keystore = json.load(f) - except Exception: - return None - - entry = keystore.get(wallet_id) - if not entry: - return None - - return WalletEncryption.decrypt(entry["encrypted_mnemonic"], self.encryption_password) - - def create_shamir_backup(self, wallet_id: str, threshold: int = 3, total: int = 5) -> list[str]: - """Create Shamir's Secret Sharing backup shares for a wallet's mnemonic.""" - mnemonic = self.get_mnemonic(wallet_id) - if not mnemonic: - raise ValueError("No mnemonic stored for this wallet") - - shares = ShamirSecretSharing.split(mnemonic.encode(), threshold, total) - - share_dir = os.path.join(self.SHARES_PATH, wallet_id) - os.makedirs(share_dir, mode=0o700, exist_ok=True) - - share_paths = [] - for x, share_bytes in shares: - path = os.path.join(share_dir, f"share_{x}_of_{total}.enc") - enc_share = WalletEncryption.encrypt(share_bytes.hex(), self.encryption_password) - with open(path, "w") as f: - json.dump( - { - "wallet_id": wallet_id, - "share_index": x, - "total_shares": total, - "threshold": threshold, - "encrypted_share": enc_share, - "created_at": datetime.now(UTC).isoformat(), - }, - f, - indent=2, - ) - os.chmod(path, 0o600) - share_paths.append(path) - - # Mark in wallet labels - wallet = self._wallets.get(wallet_id) - if wallet: - wallet.labels["shamir_backup"] = f"{threshold}-of-{total}" - wallet.labels["shamir_path"] = share_dir - self._save_vault() - - logger.info(f"Shamir backup created for {wallet_id}: {threshold}-of-{total}") - return share_paths - - def recover_from_shamir(self, wallet_id: str, share_paths: list[str]) -> str: - """Recover mnemonic from Shamir shares.""" - shares = [] - for path in share_paths: - with open(path) as f: - data = json.load(f) - share_hex = WalletEncryption.decrypt(data["encrypted_share"], self.encryption_password) - shares.append((data["share_index"], bytes.fromhex(share_hex))) - - mnemonic = ShamirSecretSharing.combine(shares).decode() - self._store_mnemonic(wallet_id, mnemonic) - return mnemonic - - # ── Wallet Operations ───────────────────────────────────── - - def get_wallet(self, wallet_id: str) -> WalletRecord | None: - """Get wallet by ID.""" - return self._wallets.get(wallet_id) - - def get_wallet_by_address(self, chain: str, address: str) -> WalletRecord | None: - """Find wallet by chain + address.""" - for w in self._wallets.values(): - if w.chain == chain and w.address.lower() == address.lower(): - return w - return None - - def list_wallets( - self, - chain: str = "", - purpose: str = "", - tier: str = "", - status: str = "", - group: str = "", - tags: list[str] | None = None, - x402_enabled: bool | None = None, - ) -> list[WalletRecord]: - """List wallets with filtering.""" - results = [] - for w in self._wallets.values(): - if chain and w.chain != chain: - continue - if purpose and w.purpose != purpose: - continue - if tier and w.tier != tier: - continue - if status and w.status != status: - continue - if group and w.group != group: - continue - if tags and not any(t in w.tags for t in tags): - continue - if x402_enabled is not None and w.x402_enabled != x402_enabled: - continue - results.append(w) - return results - - def update_wallet(self, wallet_id: str, updates: dict[str, Any]) -> WalletRecord | None: - """Update wallet metadata.""" - wallet = self._wallets.get(wallet_id) - if not wallet: - return None - - for key, value in updates.items(): - if hasattr(wallet, key): - setattr(wallet, key, value) - - wallet.version += 1 - self._save_vault() - return wallet - - def delete_wallet(self, wallet_id: str) -> bool: - """Archive wallet.""" - wallet = self._wallets.get(wallet_id) - if not wallet: - return False - - wallet.status = WalletStatus.ARCHIVED.value - wallet.labels["archived_at"] = datetime.now(UTC).isoformat() - self._save_vault() - return True - - # ── Rotation ────────────────────────────────────────────── - - def rotate_wallet( - self, - wallet_id: str, - transfer_balance: bool = False, - rotate_by: str = "", - ) -> WalletRecord | None: - """Rotate to a new wallet, marking old as rotated.""" - old_wallet = self._wallets.get(wallet_id) - if not old_wallet: - return None - - new_wallet = self.generate_wallet( - chain=old_wallet.chain, - name=old_wallet.name + " (v2)", - purpose=old_wallet.purpose, - tier=old_wallet.tier, - tags=[*old_wallet.tags, "rotated"], - group=old_wallet.group, - created_by=rotate_by, - ) - - old_wallet.status = WalletStatus.ROTATED.value - old_wallet.rotated_at = datetime.now(UTC).isoformat() - old_wallet.labels["rotated_to"] = new_wallet.wallet_id - - new_wallet.labels["rotated_from"] = old_wallet.wallet_id - new_wallet.labels["rotation_reason"] = "scheduled" - - self._save_vault() - logger.info(f"Wallet rotated: {wallet_id} -> {new_wallet.wallet_id}") - return new_wallet - - def schedule_rotation(self, wallet_id: str, days: int, auto: bool = False) -> WalletRotationSchedule | None: - """Schedule automatic rotation.""" - wallet = self._wallets.get(wallet_id) - if not wallet: - return None - - schedule = WalletRotationSchedule( - wallet_id=wallet_id, - chain=wallet.chain, - rotate_every_days=days, - auto_rotate=auto, - last_rotated=wallet.created_at, - next_rotation=(datetime.now(UTC) + timedelta(days=days)).isoformat(), - ) - - wallet.labels["rotation_schedule"] = json.dumps(schedule.to_dict()) - self._save_vault() - return schedule - - def check_rotations_due(self) -> list[WalletRotationSchedule]: - """Check which wallets are due for rotation.""" - due = [] - now = datetime.now(UTC) - - for w in self._wallets.values(): - if w.status != WalletStatus.ACTIVE.value: - continue - schedule_str = w.labels.get("rotation_schedule") - if not schedule_str: - continue - schedule = WalletRotationSchedule(**json.loads(schedule_str)) - if schedule.next_rotation: - next_rot = datetime.fromisoformat(schedule.next_rotation.replace("Z", "+00:00")) - if now >= next_rot: - due.append(schedule) - - return due - - # ── Balance & Monitoring ───────────────────────────────── - - def update_balance( - self, - wallet_id: str, - balance_raw: str, - balance_decimal: float, - balance_usd: float, - token_balances: dict | None = None, - ) -> bool: - """Update wallet balance.""" - wallet = self._wallets.get(wallet_id) - if not wallet: - return False - - wallet.balance_raw = balance_raw - wallet.balance_decimal = balance_decimal - wallet.balance_usd = balance_usd - if token_balances: - wallet.token_balances = token_balances - - self._save_vault() - return True - - def record_transaction( - self, - wallet_id: str, - tx_hash: str, - amount: float, - direction: str, - token: str = "", - usd_value: float = 0.0, - ) -> bool: - """Record a transaction for a wallet.""" - wallet = self._wallets.get(wallet_id) - if not wallet: - return False - - wallet.tx_count += 1 - wallet.last_tx_hash = tx_hash - wallet.last_tx_time = datetime.now(UTC).isoformat() - - if direction == "in": - wallet.total_received += amount - else: - wallet.total_sent += amount - - wallet.use_count += 1 - wallet.last_used = datetime.now(UTC).isoformat() - - self._save_vault() - return True - - # ── Payment Integration ─────────────────────────────────── - - def record_payment(self, payment: PaymentRecord) -> bool: - """Record a payment transaction.""" - self._payments.append(payment) - - try: - with open(self.PAYMENTS_PATH, "a") as f: - f.write(json.dumps(payment.to_dict()) + "\n") - except Exception as e: - logger.error(f"Payment record error: {e}") - - if payment.wallet_id: - wallet = self._wallets.get(payment.wallet_id) - if wallet: - if payment.payment_type in [ - PaymentType.DEPOSIT.value, - PaymentType.SUBSCRIPTION.value, - ]: - wallet.total_received += payment.amount_usd - elif payment.payment_type in [ - PaymentType.WITHDRAWAL.value, - PaymentType.REFUND.value, - ]: - wallet.total_sent += payment.amount_usd - wallet.tx_count += 1 - wallet.last_tx_time = payment.created_at - - self._save_vault() - return True - - def get_payments( - self, - wallet_id: str = "", - chain: str = "", - payment_type: str = "", - status: str = "", - user_id: str = "", - start_date: str = "", - end_date: str = "", - limit: int = 100, - ) -> list[PaymentRecord]: - """Query payment records.""" - results = [] - for p in reversed(self._payments): - if wallet_id and p.wallet_id != wallet_id: - continue - if chain and p.chain != chain: - continue - if payment_type and p.payment_type != payment_type: - continue - if status and p.status != status: - continue - if user_id and p.user_id != user_id: - continue - if start_date and p.created_at < start_date: - continue - if end_date and p.created_at > end_date: - continue - results.append(p) - if len(results) >= limit: - break - return results - - def enable_x402(self, wallet_id: str, price_usd: float) -> bool: - """Enable x402 payment processing for a wallet.""" - wallet = self._wallets.get(wallet_id) - if not wallet: - return False - - wallet.x402_enabled = True - wallet.x402_price_usd = price_usd - wallet.purpose = WalletPurpose.PAYMENTS.value - self._save_vault() - return True - - def enable_subscription(self, wallet_id: str, tiers: list[str]) -> bool: - """Enable subscription payments for a wallet.""" - wallet = self._wallets.get(wallet_id) - if not wallet: - return False - - wallet.subscription_enabled = True - wallet.subscription_tiers = tiers - wallet.purpose = WalletPurpose.SUBSCRIPTIONS.value - self._save_vault() - return True - - # ── x402 Chain Registration ────────────────────────────── - - def register_for_x402(self, wallet_id: str) -> bool: - """ - Register wallet with x402 payment system. - - Verifies the wallet can receive payments according to the chain's - standards, checks address format, and marks it as chain_registered. - """ - wallet = self._wallets.get(wallet_id) - if not wallet: - return False - - chain_meta = CHAIN_REGISTRY.get(wallet.chain) - if not chain_meta: - logger.warning(f"No chain metadata for {wallet.chain}") - return False - - # Validate address format - import re - - if chain_meta.address_pattern: - if not re.match(chain_meta.address_pattern, wallet.address): - logger.error(f"Address {wallet.address} doesn't match pattern for {wallet.chain}") - return False - - # Mark as verified and chain-registered - wallet.verified_at = datetime.now(UTC).isoformat() - wallet.verified_by = "wallet_manager_v2" - wallet.chain_registered = True - - # Enable x402 - wallet.x402_enabled = True - - self._save_vault() - logger.info(f"Wallet {wallet_id} ({wallet.chain}) registered for x402") - return True - - # ── Statistics ─────────────────────────────────────────── - - def get_stats(self) -> dict[str, Any]: - """Get comprehensive wallet statistics.""" - total_balance_usd = sum(w.balance_usd for w in self._wallets.values()) - - by_chain = {} - by_purpose = {} - by_tier = {} - by_status = {} - by_ecosystem = {} - - for w in self._wallets.values(): - by_chain[w.chain] = by_chain.get(w.chain, 0) + 1 - by_purpose[w.purpose] = by_purpose.get(w.purpose, 0) + 1 - by_tier[w.tier] = by_tier.get(w.tier, 0) + 1 - by_status[w.status] = by_status.get(w.status, 0) + 1 - - # Ecosystem grouping - meta = CHAIN_REGISTRY.get(w.chain) - eco = meta.curve if meta else "unknown" - by_ecosystem[eco] = by_ecosystem.get(eco, 0) + 1 - - total_payments = len(self._payments) - total_revenue = sum(p.amount_usd for p in self._payments if p.status == "confirmed") - - return { - "total_wallets": len(self._wallets), - "active_wallets": by_status.get(WalletStatus.ACTIVE.value, 0), - "total_balance_usd": round(total_balance_usd, 2), - "chains_supported": len(CHAIN_REGISTRY), - "chains_in_use": len(by_chain), - "by_chain": by_chain, - "by_ecosystem": by_ecosystem, - "by_purpose": by_purpose, - "by_tier": by_tier, - "by_status": by_status, - "total_payments": total_payments, - "total_revenue_usd": round(total_revenue, 2), - "x402_wallets": sum(1 for w in self._wallets.values() if w.x402_enabled), - "subscription_wallets": sum(1 for w in self._wallets.values() if w.subscription_enabled), - "rotated_wallets": sum(1 for w in self._wallets.values() if w.status == WalletStatus.ROTATED.value), - "last_updated": datetime.now(UTC).isoformat(), - } - - # ── Export / Import ─────────────────────────────────────── - - def export_safe(self) -> dict[str, Any]: - """Export wallet data without keys.""" - return { - "version": "3.0", - "exported_at": datetime.now(UTC).isoformat(), - "wallets": [w.to_safe_dict() for w in self._wallets.values()], - "chain_registry": {k: {"name": v.name, "symbol": v.native_symbol} for k, v in CHAIN_REGISTRY.items()}, - } - - def export_for_chain(self, chain: str) -> list[dict]: - """Export wallets for a specific chain.""" - return [w.to_safe_dict() for w in self._wallets.values() if w.chain == chain] - - # ── Alerts ──────────────────────────────────────────────── - - def check_alerts(self) -> list[dict]: - """Check all wallets for alert conditions.""" - alerts = [] - - for w in self._wallets.values(): - if w.status != WalletStatus.ACTIVE.value: - continue - - if w.low_balance_threshold > 0 and w.balance_usd < w.low_balance_threshold: - alerts.append( - { - "type": "low_balance", - "wallet_id": w.wallet_id, - "address": w.address, - "chain": w.chain, - "balance_usd": w.balance_usd, - "threshold": w.low_balance_threshold, - "severity": "warning", - } - ) - - schedule_str = w.labels.get("rotation_schedule") - if schedule_str: - schedule = WalletRotationSchedule(**json.loads(schedule_str)) - if schedule.next_rotation: - next_rot = datetime.fromisoformat(schedule.next_rotation.replace("Z", "+00:00")) - days_until = (next_rot - datetime.now(UTC)).days - - if days_until <= schedule.notify_before_days and days_until > 0: - alerts.append( - { - "type": "rotation_due", - "wallet_id": w.wallet_id, - "address": w.address, - "chain": w.chain, - "days_until": days_until, - "severity": "info", - } - ) - elif days_until <= 0: - alerts.append( - { - "type": "rotation_overdue", - "wallet_id": w.wallet_id, - "address": w.address, - "chain": w.chain, - "days_overdue": abs(days_until), - "severity": "critical", - } - ) - - return alerts - - # ── Auto-Sweep ─────────────────────────────────────────────── - - async def sweep_to_owner(self, chain: str = "", min_usd_threshold: float = 5.0) -> dict: - """ - Sweep USDC balances above threshold from x402 receiving wallets to owner wallets. - - EVM owner: 0x1E3AC01d0fdb976179790BDD02823196A92705C9 - Solana owner: Gix4P9AmwcZRGzr2hCEME5m2QAvY86dBfm8c7e7MpFzv - - USDC on Base: 0x833589fcd6edb6e08f4c7c32d4f71b54bda02913 - """ - owner_map = { - "evm": "0x1E3AC01d0fdb976179790BDD02823196A92705C9", - "sol": "Gix4P9AmwcZRGzr2hCEME5m2QAvY86dBfm8c7e7MpFzv", - } - evm_chains = { - "eth", - "base", - "polygon", - "arbitrum", - "optimism", - "avalanche", - "bsc", - "fantom", - "gnosis", - } - - result = { - "timestamp": datetime.now(UTC).isoformat(), - "chain_filter": chain or "all", - "threshold_usd": min_usd_threshold, - "swept": [], - "skipped": [], - "errors": [], - "total_swept_usd": 0.0, - } - - # Find x402-enabled wallets - wallets_to_check = [] - for w in self._wallets.values(): - if not w.x402_enabled: - continue - if chain and w.chain != chain: - continue - wallets_to_check.append(w) - - if not wallets_to_check: - logger.info("sweep_to_owner: no x402-enabled wallets found") - result["skipped"].append("no_x402_wallets") - return result - - for wallet in wallets_to_check: - try: - balance = wallet.balance_usd - if balance < min_usd_threshold: - result["skipped"].append( - { - "wallet_id": wallet.wallet_id, - "address": wallet.address, - "chain": wallet.chain, - "balance_usd": balance, - "reason": f"below_threshold ({balance} < {min_usd_threshold})", - } - ) - continue - - # Determine owner address and execute REAL on-chain sweep - if wallet.chain in evm_chains: - owner = owner_map["evm"] - rpc_url = ( - os.getenv("BASE_RPC_URL", "https://mainnet.base.org") - if wallet.chain == "base" - else os.getenv("ETH_RPC_URL", "https://eth.llamarpc.com") - ) - - # 1. Retrieve mnemonic (will fail if keys were never saved) - mnemonic = self.get_mnemonic(wallet.wallet_id) - if not mnemonic: - raise RuntimeError( - f"CRITICAL: Cannot sweep {wallet.wallet_id}. Mnemonic is missing. Keys were never persisted." - ) - - # 2. Derive private key - from eth_account import Account - - acct = Account.from_mnemonic(mnemonic) - if acct.address.lower() != wallet.address.lower(): - raise RuntimeError(f"Derived address {acct.address} does not match wallet {wallet.address}") - - # 3. Connect to Web3 - from web3 import Web3 - - w3 = Web3(Web3.HTTPProvider(rpc_url)) - if not w3.is_connected(): - raise RuntimeError(f"Failed to connect to RPC: {rpc_url}") - - # 4. Check actual on-chain balance - on_chain_balance_wei = w3.eth.get_balance(wallet.address) - if on_chain_balance_wei == 0: - raise RuntimeError("On-chain balance is 0. Local ledger is out of sync.") - - # Leave a small amount for future gas (0.0001 ETH) - gas_reserve_wei = w3.to_wei(0.0001, "ether") - if on_chain_balance_wei <= gas_reserve_wei: - raise RuntimeError("Balance too low to sweep after gas reserve.") - - sweep_amount_wei = on_chain_balance_wei - gas_reserve_wei - - # 5. Build and sign transaction - nonce = w3.eth.get_transaction_count(wallet.address) - gas_price = w3.eth.gas_price - tx = { - "nonce": nonce, - "to": Web3.to_checksum_address(owner), - "value": sweep_amount_wei, - "gas": 21000, - "gasPrice": gas_price, - "chainId": w3.eth.chain_id, - } - signed_tx = w3.eth.account.sign_transaction(tx, acct.key) - - # 6. Broadcast to network - tx_hash = w3.eth.send_raw_transaction(signed_tx.rawTransaction) - tx_hash_hex = tx_hash.hex() - logger.info(f"SUCCESS: Broadcasted native sweep tx {tx_hash_hex} for {wallet.wallet_id}") - - # 7. Update local ledger ONLY AFTER SUCCESSFUL BROADCAST - wallet.balance_raw = str(gas_reserve_wei) - wallet.balance_decimal = float(w3.from_wei(gas_reserve_wei, "ether")) - wallet.balance_usd = float(w3.from_wei(gas_reserve_wei, "ether")) * 2500.0 # Approx ETH price - wallet.total_sent += float(w3.from_wei(sweep_amount_wei, "ether")) - - result["swept"].append( - { - "wallet_id": wallet.wallet_id, - "address": wallet.address, - "chain": wallet.chain, - "amount_wei": sweep_amount_wei, - "tx_hash": tx_hash_hex, - "owner": owner, - "status": "broadcasted", - } - ) - result["total_swept_usd"] += float(w3.from_wei(sweep_amount_wei, "ether")) * 2500.0 - - elif wallet.chain == "sol": - owner = owner_map["sol"] - rpc_url = os.getenv("SOLANA_RPC_URL", "https://api.mainnet-beta.solana.com") - - # 1. Retrieve mnemonic - mnemonic = self.get_mnemonic(wallet.wallet_id) - if not mnemonic: - raise RuntimeError(f"CRITICAL: Cannot sweep {wallet.wallet_id}. Mnemonic is missing.") - - # 2. Derive private key using bip_utils - from bip_utils import Bip39SeedGenerator, Bip44, Bip44Coins - - seed_bytes = Bip39SeedGenerator(mnemonic).Generate() - bip44_mst_ctx = Bip44.FromSeed(seed_bytes, Bip44Coins.SOLANA) - bip44_def_ctx = bip44_mst_ctx.Purpose().Coin().Account(0).Change(0).AddressIndex(0) - private_key_bytes = bip44_def_ctx.PrivateKey().Raw().ToBytes() - - # 3. Create Solana Keypair - from solders.keypair import Keypair as SoldersKeypair - from solders.pubkey import Pubkey as SoldersPubkey - - keypair = SoldersKeypair.from_bytes(private_key_bytes) - if str(keypair.pubkey()) != wallet.address: - raise RuntimeError(f"Derived address {keypair.pubkey()} does not match wallet {wallet.address}") - - # 4. Connect to Solana RPC - from solana.rpc.async_api import AsyncClient - from solana.rpc.commitment import Confirmed - from solana.transaction import Transaction - from solders.system_program import TransferParams, transfer - - owner_pubkey = SoldersPubkey.from_string(owner) - - async with AsyncClient(rpc_url) as client: - # 5. Check actual on-chain balance - resp = await client.get_balance(keypair.pubkey(), commitment=Confirmed) - on_chain_balance_lamports = resp.value - - if on_chain_balance_lamports == 0: - raise RuntimeError("On-chain balance is 0. Local ledger is out of sync.") - - # Leave a small amount for rent/gas (e.g., 0.001 SOL = 1,000,000 lamports) - gas_reserve_lamports = 1_000_000 - if on_chain_balance_lamports <= gas_reserve_lamports: - raise RuntimeError("Balance too low to sweep after gas reserve.") - - sweep_amount_lamports = on_chain_balance_lamports - gas_reserve_lamports - - # 6. Build and sign transaction - transfer_ix = transfer( - TransferParams( - from_pubkey=keypair.pubkey(), - to_pubkey=owner_pubkey, - lamports=sweep_amount_lamports, - ) - ) - tx = Transaction().add(transfer_ix) - tx.sign(keypair) - - # 7. Broadcast to network - result_tx = await client.send_transaction(tx, keypair) - tx_sig = str(result_tx.value) - logger.info(f"SUCCESS: Broadcasted Solana sweep tx {tx_sig} for {wallet.wallet_id}") - - # 8. Update local ledger ONLY AFTER SUCCESSFUL BROADCAST - wallet.balance_raw = str(gas_reserve_lamports) - wallet.balance_decimal = gas_reserve_lamports / 1_000_000_000 - wallet.balance_usd = wallet.balance_decimal * 150.0 # Approx SOL price - wallet.total_sent += sweep_amount_lamports / 1_000_000_000 - - result["swept"].append( - { - "wallet_id": wallet.wallet_id, - "address": wallet.address, - "chain": wallet.chain, - "amount_lamports": sweep_amount_lamports, - "tx_signature": tx_sig, - "owner": owner, - "status": "broadcasted", - } - ) - result["total_swept_usd"] += (sweep_amount_lamports / 1_000_000_000) * 150.0 - else: - result["skipped"].append( - { - "wallet_id": wallet.wallet_id, - "address": wallet.address, - "chain": wallet.chain, - "balance_usd": balance, - "reason": "unsupported_chain_for_sweep", - } - ) - continue - - except Exception as e: - logger.error(f"sweep_to_owner error for {wallet.wallet_id}: {e}") - result["errors"].append( - { - "wallet_id": wallet.wallet_id, - "address": wallet.address, - "chain": wallet.chain, - "error": str(e), - } - ) - - self._save_vault() - result["total_swept_usd"] = round(result["total_swept_usd"], 2) - logger.info( - f"sweep_to_owner complete: {len(result['swept'])} swept, " - f"{len(result['skipped'])} skipped, {len(result['errors'])} errors" - ) - return result - - async def check_all_balances(self) -> list[dict]: - """ - Check balances on all x402-enabled wallets. - Returns alerts for any wallet below its low_balance_threshold. - Called by cron job every 30 minutes. - """ - alerts = [] - checked = 0 - below_threshold = 0 - - for w in self._wallets.values(): - if not w.x402_enabled: - continue - checked += 1 - - alert = { - "wallet_id": w.wallet_id, - "address": w.address, - "chain": w.chain, - "balance_usd": w.balance_usd, - "low_balance_threshold": w.low_balance_threshold, - "checked_at": datetime.now(UTC).isoformat(), - } - - if w.low_balance_threshold > 0 and w.balance_usd < w.low_balance_threshold: - alert["type"] = "low_balance" - alert["severity"] = "warning" - alert["message"] = ( - f"Wallet {w.wallet_id} ({w.chain}) balance ${w.balance_usd:.2f} " - f"is below threshold ${w.low_balance_threshold:.2f}" - ) - alerts.append(alert) - below_threshold += 1 - logger.warning(alert["message"]) - elif w.balance_usd <= 0: - alert["type"] = "zero_balance" - alert["severity"] = "info" - alert["message"] = f"Wallet {w.wallet_id} ({w.chain}) has zero balance" - alerts.append(alert) - else: - alert["type"] = "ok" - alert["severity"] = "info" - alert["message"] = f"Wallet {w.wallet_id} ({w.chain}) balance ${w.balance_usd:.2f} OK" - alerts.append(alert) - - summary = { - "checked": checked, - "below_threshold": below_threshold, - "total_alerts": len(alerts), - "alerts": alerts, - "checked_at": datetime.now(UTC).isoformat(), - } - - logger.info(f"check_all_balances: checked {checked} x402 wallets, {below_threshold} below threshold") - return summary - - async def get_revenue_wallet_status(self) -> dict: - """ - Return summary of all x402 receiving wallets with current balances, - chain info, and whether they need sweeping. - """ - wallets = [] - total_balance_usd = 0.0 - evm_chains = { - "eth", - "base", - "polygon", - "arbitrum", - "optimism", - "avalanche", - "bsc", - "fantom", - "gnosis", - } - sweep_threshold = 5.0 # Default sweep threshold - - for w in self._wallets.values(): - if not w.x402_enabled: - continue - - needs_sweep = w.balance_usd >= sweep_threshold - if w.chain in evm_chains: - owner = "0x1E3AC01d0fdb976179790BDD02823196A92705C9" - elif w.chain == "sol": - owner = "Gix4P9AmwcZRGzr2hCEME5m2QAvY86dBfm8c7e7MpFzv" - else: - owner = "" - - wallets.append( - { - "wallet_id": w.wallet_id, - "address": w.address, - "chain": w.chain, - "balance_usd": w.balance_usd, - "balance_raw": w.balance_raw, - "tier": w.tier, - "purpose": w.purpose, - "status": w.status, - "x402_price_usd": w.x402_price_usd, - "low_balance_threshold": w.low_balance_threshold, - "needs_sweep": needs_sweep, - "owner_address": owner, - "last_used": w.last_used, - "tx_count": w.tx_count, - } - ) - total_balance_usd += w.balance_usd - - need_sweep_count = sum(1 for w in wallets if w["needs_sweep"]) - below_threshold_count = sum( - 1 for w in wallets if w["low_balance_threshold"] > 0 and w["balance_usd"] < w["low_balance_threshold"] - ) - - return { - "timestamp": datetime.now(UTC).isoformat(), - "total_x402_wallets": len(wallets), - "total_balance_usd": round(total_balance_usd, 2), - "need_sweep_count": need_sweep_count, - "below_threshold_count": below_threshold_count, - "sweep_threshold_usd": sweep_threshold, - "wallets": wallets, - } - - -# ── Singleton ───────────────────────────────────────────────────── - -_wallet_manager_instance: WalletManagerV2 | None = None - - -def get_wallet_manager_v2(password: str = "") -> WalletManagerV2: - """Get or create wallet manager instance.""" - global _wallet_manager_instance - if _wallet_manager_instance is None: - _wallet_manager_instance = WalletManagerV2(password) - return _wallet_manager_instance - - -# ── Utility: Import legacy wallets ──────────────────────────────── - - -def import_legacy_wallets(manager: WalletManagerV2) -> int: - """Import wallets from legacy v1 JSON files into v2 vault.""" - imported = 0 - - legacy_files = [ - "/root/.rmi/wallets/x402_wallets.json", - "/root/.rmi/wallets/referral_wallets.json", - "/app/data/wallets/x402_wallets.json", - "/app/data/wallets/referral_wallets.json", - ] - - for path in legacy_files: - if not os.path.exists(path): - continue - - try: - with open(path) as f: - data = json.load(f) - - wallets_data = data.get("wallets", {}) - for chain_key, wdata in wallets_data.items(): - chain_map = { - "tron": "trx", - "bitcoin": "btc", - "eth": "eth", - "sol": "sol", - } - chain = chain_map.get(chain_key, chain_key) - - # Check if already imported - existing = manager.get_wallet_by_address(chain, wdata["address"]) - if existing: - logger.info(f"Legacy {chain_key} wallet already in vault") - continue - - wallet_id = f"wal_{chain}_legacy_{int(time.time())}_{secrets.token_hex(4)}" - wallet = WalletRecord( - wallet_id=wallet_id, - chain=chain, - address=wdata["address"], - public_key=wdata.get("public_key_hex", ""), - name=f"Legacy {chain_key.upper()} Wallet", - purpose=WalletPurpose.PAYMENTS.value - if path.endswith("x402_wallets.json") - else WalletPurpose.OPERATIONS.value, - tier=WalletTier.WARM.value, - tags=["legacy", "imported"], - created_at=data.get("created", datetime.now(UTC).isoformat()), - created_by="import_script", - ) - - manager._wallets[wallet_id] = wallet - imported += 1 - logger.info(f"Imported legacy wallet: {chain_key} -> {wallet_id}") - except Exception as e: - logger.error(f"Legacy import error ({path}): {e}") - - if imported > 0: - manager._save_vault() - logger.info(f"Imported {imported} legacy wallets") - - return imported +"""Backward-compat shim — moved to app.wallet.manager in P3B.""" +from app.wallet.manager import * # noqa: F401,F403 +from app.wallet.manager import ( # noqa: F401 + ChainMeta, + PaymentRecord, + PaymentType, + ShamirSecretSharing, + WalletEncryption, + WalletManagerFacade, + WalletManagerV2, + WalletPurpose, + WalletRecord, + WalletRotationSchedule, + WalletStatus, + WalletTier, + ZKAddressVerifier, + get_wallet_manager_v2, + import_legacy_wallets, + _build_registry, +)