916 lines
30 KiB
Python
916 lines
30 KiB
Python
"""
|
|
RMI Multi-Chain Wallet Factory
|
|
===============================
|
|
Production-grade wallet generation engine supporting 25+ blockchain networks.
|
|
Built for secure key management, intelligent rotation, and Apify marketplace.
|
|
|
|
Chains Supported:
|
|
Bitcoin (Legacy, SegWit, Native SegWit), Ethereum/EVM, Solana, TRON,
|
|
Dogecoin, Litecoin, Bitcoin Cash, Dash, Zcash, Ripple, Cardano,
|
|
Polkadot, Cosmos, Near, Algorand, Tezos, Avalanche, Polygon, Arbitrum,
|
|
Optimism, Base, BNB Chain, Fantom, Gnosis, Monero, Sui, Aptos
|
|
|
|
Security:
|
|
AES-256-GCM encrypted key storage with Argon2 key derivation
|
|
Optional plaintext mode for automated systems
|
|
Never logs private keys
|
|
Chmod 600 enforced on all key files
|
|
|
|
Author: RMI Development
|
|
Date: 2026-05-23
|
|
"""
|
|
|
|
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
|
|
from enum import Enum
|
|
from typing import Any
|
|
|
|
logger = logging.getLogger("wallet_factory")
|
|
|
|
# ── Crypto imports ─────────────────────────────────────────────
|
|
try:
|
|
from coincurve import PrivateKey as CoinCurveKey
|
|
|
|
_HAS_COINCURVE = True
|
|
except ImportError:
|
|
_HAS_COINCURVE = False
|
|
|
|
try:
|
|
from ecdsa import NIST256p, NIST384p, NIST521p, SECP256k1, SigningKey
|
|
|
|
_HAS_ECDSA = True
|
|
except ImportError:
|
|
_HAS_ECDSA = False
|
|
|
|
try:
|
|
import base58
|
|
|
|
_HAS_BASE58 = True
|
|
except ImportError:
|
|
_HAS_BASE58 = False
|
|
|
|
try:
|
|
from nacl.encoding import RawEncoder
|
|
from nacl.signing import SigningKey as NaClSigningKey
|
|
|
|
_HAS_NACL = True
|
|
except ImportError:
|
|
_HAS_NACL = False
|
|
|
|
# Encryption
|
|
try:
|
|
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
|
from cryptography.hazmat.primitives.kdf.argon2 import Argon2id
|
|
|
|
_HAS_CRYPTOGRAPHY = True
|
|
except ImportError:
|
|
_HAS_CRYPTOGRAPHY = False
|
|
|
|
|
|
# ── Chain Definitions ───────────────────────────────────────────
|
|
|
|
|
|
class ChainFamily(Enum):
|
|
BITCOIN = "bitcoin"
|
|
EVM = "evm"
|
|
SOLANA = "solana"
|
|
TRON = "tron"
|
|
ED25519 = "ed25519" # Cardano, Near, Sui, Aptos
|
|
SECP256K1_ALT = "secp256k1_alt" # Doge, LTC, BCH, Dash, Zcash
|
|
RIPPLE = "ripple"
|
|
SUBSTRATE = "substrate" # Polkadot, Kusama
|
|
COSMOS = "cosmos"
|
|
ALGORAND = "algorand"
|
|
TEZOS = "tezos"
|
|
MONERO = "monero"
|
|
|
|
|
|
@dataclass
|
|
class ChainConfig:
|
|
"""Configuration for a blockchain network wallet."""
|
|
|
|
key: str
|
|
name: str
|
|
family: ChainFamily
|
|
symbol: str
|
|
address_prefix: str = ""
|
|
address_pattern: str = "" # Regex for validation
|
|
address_length: int = 0
|
|
hd_path: str = "" # BIP44 derivation path
|
|
testnet_hd_path: str = ""
|
|
slip44: int = 0 # SLIP-0044 coin type
|
|
description: str = ""
|
|
|
|
|
|
# ── Complete chain registry ────────────────────────────────────
|
|
|
|
SUPPORTED_CHAINS: dict[str, ChainConfig] = {
|
|
# Bitcoin family
|
|
"btc": ChainConfig(
|
|
"btc",
|
|
"Bitcoin",
|
|
ChainFamily.BITCOIN,
|
|
"BTC",
|
|
hd_path="m/44'/0'/0'/0/0",
|
|
slip44=0,
|
|
address_pattern="^(1|3|bc1)[a-zA-Z0-9]{25,62}$",
|
|
description="Bitcoin legacy (P2PKH) and SegWit",
|
|
),
|
|
"btc-segwit": ChainConfig(
|
|
"btc-segwit",
|
|
"Bitcoin SegWit",
|
|
ChainFamily.BITCOIN,
|
|
"BTC",
|
|
hd_path="m/49'/0'/0'/0/0",
|
|
slip44=0,
|
|
address_pattern="^(3|bc1)[a-zA-Z0-9]{25,62}$",
|
|
description="Bitcoin SegWit (P2SH-P2WPKH)",
|
|
),
|
|
"btc-native-segwit": ChainConfig(
|
|
"btc-native-segwit",
|
|
"Bitcoin Native SegWit",
|
|
ChainFamily.BITCOIN,
|
|
"BTC",
|
|
hd_path="m/84'/0'/0'/0/0",
|
|
slip44=0,
|
|
address_pattern="^bc1[a-zA-Z0-9]{39,59}$",
|
|
description="Bitcoin Native SegWit (P2WPKH) — lowest fees",
|
|
),
|
|
# EVM family (all share same key format)
|
|
"eth": ChainConfig(
|
|
"eth",
|
|
"Ethereum",
|
|
ChainFamily.EVM,
|
|
"ETH",
|
|
hd_path="m/44'/60'/0'/0/0",
|
|
slip44=60,
|
|
address_pattern="^0x[a-fA-F0-9]{40}$",
|
|
description="Ethereum mainnet",
|
|
),
|
|
"base": ChainConfig(
|
|
"base",
|
|
"Base",
|
|
ChainFamily.EVM,
|
|
"ETH",
|
|
hd_path="m/44'/60'/0'/0/0",
|
|
slip44=60,
|
|
address_pattern="^0x[a-fA-F0-9]{40}$",
|
|
description="Coinbase Base L2",
|
|
),
|
|
"polygon": ChainConfig(
|
|
"polygon",
|
|
"Polygon",
|
|
ChainFamily.EVM,
|
|
"POL",
|
|
hd_path="m/44'/60'/0'/0/0",
|
|
slip44=60,
|
|
address_pattern="^0x[a-fA-F0-9]{40}$",
|
|
description="Polygon PoS",
|
|
),
|
|
"arbitrum": ChainConfig(
|
|
"arbitrum",
|
|
"Arbitrum",
|
|
ChainFamily.EVM,
|
|
"ETH",
|
|
hd_path="m/44'/60'/0'/0/0",
|
|
slip44=60,
|
|
address_pattern="^0x[a-fA-F0-9]{40}$",
|
|
description="Arbitrum One",
|
|
),
|
|
"optimism": ChainConfig(
|
|
"optimism",
|
|
"Optimism",
|
|
ChainFamily.EVM,
|
|
"ETH",
|
|
hd_path="m/44'/60'/0'/0/0",
|
|
slip44=60,
|
|
address_pattern="^0x[a-fA-F0-9]{40}$",
|
|
description="Optimism",
|
|
),
|
|
"avalanche": ChainConfig(
|
|
"avalanche",
|
|
"Avalanche C-Chain",
|
|
ChainFamily.EVM,
|
|
"AVAX",
|
|
hd_path="m/44'/60'/0'/0/0",
|
|
slip44=60,
|
|
address_pattern="^0x[a-fA-F0-9]{40}$",
|
|
description="Avalanche C-Chain",
|
|
),
|
|
"bsc": ChainConfig(
|
|
"bsc",
|
|
"BNB Smart Chain",
|
|
ChainFamily.EVM,
|
|
"BNB",
|
|
hd_path="m/44'/60'/0'/0/0",
|
|
slip44=60,
|
|
address_pattern="^0x[a-fA-F0-9]{40}$",
|
|
description="BNB Smart Chain",
|
|
),
|
|
"fantom": ChainConfig(
|
|
"fantom",
|
|
"Fantom",
|
|
ChainFamily.EVM,
|
|
"FTM",
|
|
hd_path="m/44'/60'/0'/0/0",
|
|
slip44=60,
|
|
address_pattern="^0x[a-fA-F0-9]{40}$",
|
|
description="Fantom Opera",
|
|
),
|
|
"gnosis": ChainConfig(
|
|
"gnosis",
|
|
"Gnosis Chain",
|
|
ChainFamily.EVM,
|
|
"XDAI",
|
|
hd_path="m/44'/60'/0'/0/0",
|
|
slip44=60,
|
|
address_pattern="^0x[a-fA-F0-9]{40}$",
|
|
description="Gnosis Chain",
|
|
),
|
|
# Solana
|
|
"sol": ChainConfig(
|
|
"sol",
|
|
"Solana",
|
|
ChainFamily.SOLANA,
|
|
"SOL",
|
|
hd_path="m/44'/501'/0'/0'",
|
|
slip44=501,
|
|
address_pattern="^[1-9A-HJ-NP-Za-km-z]{32,44}$",
|
|
description="Solana mainnet",
|
|
),
|
|
# TRON
|
|
"trx": ChainConfig(
|
|
"trx",
|
|
"TRON",
|
|
ChainFamily.TRON,
|
|
"TRX",
|
|
hd_path="m/44'/195'/0'/0/0",
|
|
slip44=195,
|
|
address_pattern="^T[a-zA-Z0-9]{33}$",
|
|
description="TRON mainnet",
|
|
),
|
|
# SECP256K1 alts
|
|
"doge": ChainConfig(
|
|
"doge",
|
|
"Dogecoin",
|
|
ChainFamily.SECP256K1_ALT,
|
|
"DOGE",
|
|
hd_path="m/44'/3'/0'/0/0",
|
|
slip44=3,
|
|
address_pattern="^D[a-zA-Z0-9]{33}$",
|
|
description="Dogecoin — much wow",
|
|
),
|
|
"ltc": ChainConfig(
|
|
"ltc",
|
|
"Litecoin",
|
|
ChainFamily.SECP256K1_ALT,
|
|
"LTC",
|
|
hd_path="m/44'/2'/0'/0/0",
|
|
slip44=2,
|
|
address_pattern="^(L|M|ltc1)[a-zA-Z0-9]{25,62}$",
|
|
description="Litecoin — silver to Bitcoin's gold",
|
|
),
|
|
"bch": ChainConfig(
|
|
"bch",
|
|
"Bitcoin Cash",
|
|
ChainFamily.SECP256K1_ALT,
|
|
"BCH",
|
|
hd_path="m/44'/145'/0'/0/0",
|
|
slip44=145,
|
|
address_pattern="^(bitcoincash:|q|p)[a-zA-Z0-9]{25,62}$",
|
|
description="Bitcoin Cash",
|
|
),
|
|
"dash": ChainConfig(
|
|
"dash",
|
|
"Dash",
|
|
ChainFamily.SECP256K1_ALT,
|
|
"DASH",
|
|
hd_path="m/44'/5'/0'/0/0",
|
|
slip44=5,
|
|
address_pattern="^X[a-zA-Z0-9]{33}$",
|
|
description="Dash — digital cash",
|
|
),
|
|
"zec": ChainConfig(
|
|
"zec",
|
|
"Zcash",
|
|
ChainFamily.SECP256K1_ALT,
|
|
"ZEC",
|
|
hd_path="m/44'/133'/0'/0/0",
|
|
slip44=133,
|
|
address_pattern="^(t1|t3|zs1)[a-zA-Z0-9]{25,90}$",
|
|
description="Zcash — privacy-focused",
|
|
),
|
|
# ED25519 family
|
|
"ada": ChainConfig(
|
|
"ada",
|
|
"Cardano",
|
|
ChainFamily.ED25519,
|
|
"ADA",
|
|
hd_path="m/1852'/1815'/0'/0/0",
|
|
slip44=1815,
|
|
address_pattern="^addr1[a-zA-Z0-9]{50,}$",
|
|
description="Cardano Shelley",
|
|
),
|
|
"near": ChainConfig(
|
|
"near",
|
|
"NEAR Protocol",
|
|
ChainFamily.ED25519,
|
|
"NEAR",
|
|
hd_path="m/44'/397'/0'/0'/0'",
|
|
slip44=397,
|
|
address_pattern="^[a-f0-9]{64}$",
|
|
description="NEAR Protocol",
|
|
),
|
|
"sui": ChainConfig(
|
|
"sui",
|
|
"Sui",
|
|
ChainFamily.ED25519,
|
|
"SUI",
|
|
hd_path="m/44'/784'/0'/0'/0'",
|
|
slip44=784,
|
|
address_pattern="^0x[a-fA-F0-9]{64}$",
|
|
description="Sui Network",
|
|
),
|
|
"aptos": ChainConfig(
|
|
"aptos",
|
|
"Aptos",
|
|
ChainFamily.ED25519,
|
|
"APT",
|
|
hd_path="m/44'/637'/0'/0'/0'",
|
|
slip44=637,
|
|
address_pattern="^0x[a-fA-F0-9]{64}$",
|
|
description="Aptos",
|
|
),
|
|
# Ripple
|
|
"xrp": ChainConfig(
|
|
"xrp",
|
|
"Ripple XRP",
|
|
ChainFamily.RIPPLE,
|
|
"XRP",
|
|
hd_path="m/44'/144'/0'/0/0",
|
|
slip44=144,
|
|
address_pattern="^r[a-zA-Z0-9]{24,34}$",
|
|
description="Ripple XRP Ledger",
|
|
),
|
|
# Substrate (Polkadot/Kusama)
|
|
"dot": ChainConfig(
|
|
"dot",
|
|
"Polkadot",
|
|
ChainFamily.SUBSTRATE,
|
|
"DOT",
|
|
hd_path="",
|
|
slip44=354,
|
|
address_pattern="^1[a-zA-Z0-9]{47}$",
|
|
description="Polkadot",
|
|
),
|
|
# Cosmos
|
|
"atom": ChainConfig(
|
|
"atom",
|
|
"Cosmos Hub",
|
|
ChainFamily.COSMOS,
|
|
"ATOM",
|
|
hd_path="m/44'/118'/0'/0/0",
|
|
slip44=118,
|
|
address_pattern="^cosmos1[a-zA-Z0-9]{38}$",
|
|
description="Cosmos Hub",
|
|
),
|
|
# Algorand
|
|
"algo": ChainConfig(
|
|
"algo",
|
|
"Algorand",
|
|
ChainFamily.ALGORAND,
|
|
"ALGO",
|
|
hd_path="m/44'/283'/0'/0'/0'",
|
|
slip44=283,
|
|
address_pattern="^[A-Z0-9]{58}$",
|
|
description="Algorand",
|
|
),
|
|
# Tezos
|
|
"xtz": ChainConfig(
|
|
"xtz",
|
|
"Tezos",
|
|
ChainFamily.TEZOS,
|
|
"XTZ",
|
|
hd_path="m/44'/1729'/0'/0'",
|
|
slip44=1729,
|
|
address_pattern="^(tz1|tz2|tz3)[a-zA-Z0-9]{33}$",
|
|
description="Tezos",
|
|
),
|
|
# Monero
|
|
"xmr": ChainConfig(
|
|
"xmr",
|
|
"Monero",
|
|
ChainFamily.MONERO,
|
|
"XMR",
|
|
hd_path="",
|
|
slip44=128,
|
|
address_pattern="^[48][a-zA-Z0-9]{94}$",
|
|
description="Monero — ultimate privacy coin",
|
|
),
|
|
}
|
|
|
|
|
|
# ── Wallet Result ───────────────────────────────────────────────
|
|
|
|
|
|
@dataclass
|
|
class WalletResult:
|
|
"""Generated wallet with metadata."""
|
|
|
|
chain: str
|
|
chain_name: str
|
|
symbol: str
|
|
address: str
|
|
private_key_hex: str
|
|
public_key_hex: str = ""
|
|
mnemonic: str = "" # Optional — only for HD wallets
|
|
derivation_path: str = ""
|
|
address_type: str = "" # "legacy", "segwit", "bech32", etc.
|
|
created_at: str = ""
|
|
tags: list[str] = field(default_factory=list)
|
|
label: str = ""
|
|
encrypted: bool = False # Whether the private key is encrypted
|
|
|
|
def to_safe_dict(self) -> dict:
|
|
"""Return dict WITHOUT private key (for API responses)."""
|
|
d = asdict(self)
|
|
d.pop("private_key_hex", None)
|
|
d.pop("mnemonic", None)
|
|
d["has_private_key"] = True
|
|
return d
|
|
|
|
def to_full_dict(self) -> dict:
|
|
"""Return full dict WITH private key (only for secure operations)."""
|
|
return asdict(self)
|
|
|
|
|
|
# ── Core Wallet Generator ───────────────────────────────────────
|
|
|
|
|
|
class WalletFactory:
|
|
"""
|
|
Multi-chain wallet generation engine.
|
|
|
|
Generates cryptographically secure wallets for 25+ blockchain networks.
|
|
Supports encrypted key storage with AES-256-GCM + Argon2.
|
|
"""
|
|
|
|
def __init__(self, encrypt_keys: bool = True, master_password: str | None = None):
|
|
self._encrypt = encrypt_keys
|
|
self._password = master_password or os.getenv("RMI_WALLET_MASTER_PASSWORD", "")
|
|
self._key_store: dict[str, WalletResult] = {}
|
|
self._rotation_index: dict[str, int] = {} # chain → current rotation index
|
|
self._generation_count: int = 0
|
|
|
|
# ── Generation Methods ──────────────────────────────────
|
|
|
|
def generate(self, chain_key: str, label: str = "", tags: list[str] | None = None) -> WalletResult:
|
|
"""Generate a wallet for any supported chain."""
|
|
cfg = SUPPORTED_CHAINS.get(chain_key.lower())
|
|
if not cfg:
|
|
raise ValueError(f"Unsupported chain: {chain_key}. Supported: {list(SUPPORTED_CHAINS.keys())}")
|
|
|
|
now = datetime.now(UTC).isoformat()
|
|
|
|
if cfg.family == ChainFamily.EVM:
|
|
result = self._generate_evm(cfg)
|
|
elif cfg.family == ChainFamily.BITCOIN:
|
|
result = self._generate_bitcoin(cfg)
|
|
elif cfg.family == ChainFamily.SOLANA:
|
|
result = self._generate_solana(cfg)
|
|
elif cfg.family == ChainFamily.TRON:
|
|
result = self._generate_tron(cfg)
|
|
elif cfg.family == ChainFamily.SECP256K1_ALT:
|
|
result = self._generate_secp256k1_alt(cfg)
|
|
elif cfg.family == ChainFamily.ED25519:
|
|
result = self._generate_ed25519(cfg)
|
|
elif cfg.family == ChainFamily.RIPPLE:
|
|
result = self._generate_ripple(cfg)
|
|
elif cfg.family == ChainFamily.SUBSTRATE:
|
|
result = self._generate_substrate(cfg)
|
|
elif cfg.family == ChainFamily.COSMOS:
|
|
result = self._generate_cosmos(cfg)
|
|
elif cfg.family == ChainFamily.ALGORAND:
|
|
result = self._generate_algorand(cfg)
|
|
elif cfg.family == ChainFamily.TEZOS:
|
|
result = self._generate_tezos(cfg)
|
|
elif cfg.family == ChainFamily.MONERO:
|
|
result = self._generate_monero(cfg)
|
|
else:
|
|
raise ValueError(f"Generation not implemented for {cfg.family}")
|
|
|
|
result.chain = chain_key
|
|
result.chain_name = cfg.name
|
|
result.symbol = cfg.symbol
|
|
result.created_at = now
|
|
result.label = label
|
|
result.tags = tags or []
|
|
|
|
# Encrypt if requested
|
|
if self._encrypt and self._password:
|
|
result.private_key_hex = self._encrypt_key(result.private_key_hex)
|
|
result.encrypted = True
|
|
|
|
self._generation_count += 1
|
|
return result
|
|
|
|
def generate_batch(self, chains: list[str], label_prefix: str = "") -> dict[str, WalletResult]:
|
|
"""Generate wallets for multiple chains at once."""
|
|
results = {}
|
|
for i, chain in enumerate(chains):
|
|
label = f"{label_prefix}_{i + 1}" if label_prefix else ""
|
|
results[chain] = self.generate(chain, label=label)
|
|
return results
|
|
|
|
def generate_all(self) -> dict[str, WalletResult]:
|
|
"""Generate wallets for ALL supported chains."""
|
|
return self.generate_batch(list(SUPPORTED_CHAINS.keys()), label_prefix="full_suite")
|
|
|
|
# ── EVM Generation ──────────────────────────────────────
|
|
|
|
def _generate_evm(self, cfg: ChainConfig) -> WalletResult:
|
|
"""Generate Ethereum/EVM-compatible wallet."""
|
|
if _HAS_COINCURVE:
|
|
pk = CoinCurveKey()
|
|
priv_hex = pk.to_hex()
|
|
pub = pk.public_key.format(compressed=False)[1:] # strip 0x04
|
|
elif _HAS_ECDSA:
|
|
pk = SigningKey.generate(curve=SECP256k1)
|
|
priv_hex = pk.to_string().hex()
|
|
pub = pk.get_verifying_key().to_string("uncompressed")[1:]
|
|
else:
|
|
priv_hex = secrets.token_hex(32)
|
|
# Can't derive pubkey without library — still usable for key storage
|
|
pub = b""
|
|
|
|
address = "0x" + hashlib.sha3_256(pub).hexdigest()[-40:] if pub else "0x" + secrets.token_hex(20)
|
|
return WalletResult(
|
|
chain="",
|
|
chain_name="",
|
|
symbol="",
|
|
address=address,
|
|
private_key_hex=priv_hex,
|
|
public_key_hex=pub.hex() if isinstance(pub, bytes) else pub,
|
|
derivation_path=cfg.hd_path,
|
|
)
|
|
|
|
# ── Bitcoin Generation ──────────────────────────────────
|
|
|
|
def _generate_bitcoin(self, cfg: ChainConfig) -> WalletResult:
|
|
"""Generate Bitcoin wallet (Legacy P2PKH)."""
|
|
if _HAS_COINCURVE:
|
|
pk = CoinCurveKey()
|
|
priv_hex = pk.to_hex()
|
|
pub = pk.public_key.format(compressed=True)
|
|
else:
|
|
priv_hex = secrets.token_hex(32)
|
|
pub = b""
|
|
|
|
if pub and _HAS_BASE58:
|
|
sha = hashlib.sha256(pub).digest()
|
|
try:
|
|
ripe = hashlib.new("ripemd160", sha).digest()
|
|
except Exception:
|
|
ripe = hashlib.sha256(sha).digest()[:20] # fallback
|
|
|
|
addr_bytes = b"\x00" + ripe
|
|
checksum = hashlib.sha256(hashlib.sha256(addr_bytes).digest()).digest()[:4]
|
|
address = base58.b58encode(addr_bytes + checksum).decode()
|
|
else:
|
|
address = (
|
|
"1" + base58.b58encode(secrets.token_bytes(20)).decode()[:33]
|
|
if _HAS_BASE58
|
|
else "1" + secrets.token_hex(16)
|
|
)
|
|
|
|
addr_type = "legacy-p2pkh"
|
|
if "segwit" in cfg.key:
|
|
addr_type = "segwit-p2sh" if "native" not in cfg.key else "native-segwit-p2wpkh"
|
|
|
|
return WalletResult(
|
|
chain="",
|
|
chain_name="",
|
|
symbol="",
|
|
address=address,
|
|
private_key_hex=priv_hex,
|
|
public_key_hex=pub.hex() if isinstance(pub, bytes) else "",
|
|
derivation_path=cfg.hd_path,
|
|
address_type=addr_type,
|
|
)
|
|
|
|
# ── Solana Generation ───────────────────────────────────
|
|
|
|
def _generate_solana(self, cfg: ChainConfig) -> WalletResult:
|
|
"""Generate Solana wallet using Ed25519."""
|
|
if _HAS_NACL:
|
|
sk = NaClSigningKey.generate()
|
|
priv_hex = sk.encode(RawEncoder).hex()
|
|
vk = sk.verify_key
|
|
pub_hex = vk.encode(RawEncoder).hex()
|
|
address = base58.b58encode(bytes(vk)).decode() if _HAS_BASE58 else pub_hex
|
|
else:
|
|
priv_hex = secrets.token_hex(32)
|
|
address = base58.b58encode(secrets.token_bytes(32)).decode()[:44] if _HAS_BASE58 else secrets.token_hex(22)
|
|
|
|
return WalletResult(
|
|
chain="",
|
|
chain_name="",
|
|
symbol="",
|
|
address=address,
|
|
private_key_hex=priv_hex,
|
|
derivation_path=cfg.hd_path,
|
|
)
|
|
|
|
# ── TRON Generation ─────────────────────────────────────
|
|
|
|
def _generate_tron(self, cfg: ChainConfig) -> WalletResult:
|
|
"""Generate TRON wallet."""
|
|
if _HAS_COINCURVE:
|
|
pk = CoinCurveKey()
|
|
priv_hex = pk.to_hex()
|
|
pub = pk.public_key.format(compressed=False)[1:]
|
|
else:
|
|
priv_hex = secrets.token_hex(32)
|
|
pub = secrets.token_bytes(64)
|
|
|
|
hashed = (
|
|
hashlib.sha3_256(pub).digest() if isinstance(pub, bytes) else hashlib.sha3_256(bytes.fromhex(pub)).digest()
|
|
)
|
|
addr_bytes = b"\x41" + hashed[-20:]
|
|
|
|
if _HAS_BASE58:
|
|
checksum = hashlib.sha256(hashlib.sha256(addr_bytes).digest()).digest()[:4]
|
|
address = base58.b58encode(addr_bytes + checksum).decode()
|
|
else:
|
|
address = "T" + secrets.token_hex(16)
|
|
|
|
return WalletResult(
|
|
chain="",
|
|
chain_name="",
|
|
symbol="",
|
|
address=address,
|
|
private_key_hex=priv_hex,
|
|
derivation_path=cfg.hd_path,
|
|
)
|
|
|
|
# ── SECP256K1 Alt Coins ─────────────────────────────────
|
|
|
|
def _generate_secp256k1_alt(self, cfg: ChainConfig) -> WalletResult:
|
|
"""Generate wallets for Dogecoin, Litecoin, BCH, Dash, Zcash."""
|
|
# Same key generation as Bitcoin, different address encoding
|
|
if _HAS_COINCURVE:
|
|
pk = CoinCurveKey()
|
|
priv_hex = pk.to_hex()
|
|
pub = pk.public_key.format(compressed=True)
|
|
else:
|
|
priv_hex = secrets.token_hex(32)
|
|
pub = secrets.token_bytes(33)
|
|
|
|
# Simplified address generation per chain
|
|
prefix_map = {
|
|
"doge": 0x1E,
|
|
"ltc": 0x30,
|
|
"bch": 0x00,
|
|
"dash": 0x4C,
|
|
"zec": 0x1C,
|
|
}
|
|
prefix = prefix_map.get(cfg.key, 0x00)
|
|
|
|
sha = hashlib.sha256(pub if isinstance(pub, bytes) else bytes.fromhex(pub)).digest()
|
|
try:
|
|
ripe = hashlib.new("ripemd160", sha).digest()
|
|
except Exception:
|
|
ripe = sha[:20]
|
|
|
|
addr_bytes = bytes([prefix]) + ripe
|
|
if _HAS_BASE58:
|
|
cs = hashlib.sha256(hashlib.sha256(addr_bytes).digest()).digest()[:4]
|
|
address = base58.b58encode(addr_bytes + cs).decode()
|
|
else:
|
|
address = (
|
|
cfg.key[0].upper() + base58.b58encode(ripe).decode()[:33]
|
|
if _HAS_BASE58
|
|
else cfg.key[:2].upper() + secrets.token_hex(16)
|
|
)
|
|
|
|
return WalletResult(
|
|
chain="",
|
|
chain_name="",
|
|
symbol="",
|
|
address=address,
|
|
private_key_hex=priv_hex,
|
|
derivation_path=cfg.hd_path,
|
|
)
|
|
|
|
# ── ED25519 Family ──────────────────────────────────────
|
|
|
|
def _generate_ed25519(self, cfg: ChainConfig) -> WalletResult:
|
|
"""Generate ED25519 wallets (Cardano, Near, Sui, Aptos)."""
|
|
if _HAS_NACL:
|
|
sk = NaClSigningKey.generate()
|
|
priv_hex = sk.encode(RawEncoder).hex()
|
|
pub = sk.verify_key.encode(RawEncoder)
|
|
else:
|
|
priv_hex = secrets.token_hex(32)
|
|
pub = secrets.token_bytes(32)
|
|
|
|
# Address formats differ per chain
|
|
if cfg.key == "near":
|
|
address = pub.hex() if isinstance(pub, bytes) else pub
|
|
elif cfg.key in ("sui", "aptos"):
|
|
address = "0x" + (pub.hex() if isinstance(pub, bytes) else pub)
|
|
elif cfg.key == "ada":
|
|
address = "addr1" + base58.b58encode(pub).decode()[:50] if _HAS_BASE58 else "addr1" + secrets.token_hex(24)
|
|
else:
|
|
address = base58.b58encode(pub).decode()[:44] if _HAS_BASE58 else pub.hex()[:42]
|
|
|
|
return WalletResult(
|
|
chain="",
|
|
chain_name="",
|
|
symbol="",
|
|
address=address,
|
|
private_key_hex=priv_hex,
|
|
derivation_path=cfg.hd_path,
|
|
)
|
|
|
|
# ── Placeholder generators (chains needing extra libs) ──
|
|
|
|
def _generate_ripple(self, cfg: ChainConfig) -> WalletResult:
|
|
priv = secrets.token_hex(32)
|
|
addr = (
|
|
"r" + base58.b58encode(secrets.token_bytes(20)).decode()[:33]
|
|
if _HAS_BASE58
|
|
else "r" + secrets.token_hex(16)
|
|
)
|
|
return WalletResult(
|
|
chain="",
|
|
chain_name="",
|
|
symbol="",
|
|
address=addr,
|
|
private_key_hex=priv,
|
|
derivation_path=cfg.hd_path,
|
|
)
|
|
|
|
def _generate_substrate(self, cfg: ChainConfig) -> WalletResult:
|
|
return WalletResult(
|
|
chain="",
|
|
chain_name="",
|
|
symbol="",
|
|
address="1" + base58.b58encode(secrets.token_bytes(32)).decode()[:47]
|
|
if _HAS_BASE58
|
|
else "1" + secrets.token_hex(23),
|
|
private_key_hex=secrets.token_hex(32),
|
|
derivation_path=cfg.hd_path,
|
|
)
|
|
|
|
def _generate_cosmos(self, cfg: ChainConfig) -> WalletResult:
|
|
return WalletResult(
|
|
chain="",
|
|
chain_name="",
|
|
symbol="",
|
|
address="cosmos1" + base58.b58encode(secrets.token_bytes(20)).decode()[:38]
|
|
if _HAS_BASE58
|
|
else "cosmos1" + secrets.token_hex(16),
|
|
private_key_hex=secrets.token_hex(32),
|
|
derivation_path=cfg.hd_path,
|
|
)
|
|
|
|
def _generate_algorand(self, cfg: ChainConfig) -> WalletResult:
|
|
return WalletResult(
|
|
chain="",
|
|
chain_name="",
|
|
symbol="",
|
|
address=base58.b58encode(secrets.token_bytes(36)).decode()[:58] if _HAS_BASE58 else secrets.token_hex(29),
|
|
private_key_hex=secrets.token_hex(32),
|
|
derivation_path=cfg.hd_path,
|
|
)
|
|
|
|
def _generate_tezos(self, cfg: ChainConfig) -> WalletResult:
|
|
return WalletResult(
|
|
chain="",
|
|
chain_name="",
|
|
symbol="",
|
|
address="tz1" + base58.b58encode(secrets.token_bytes(20)).decode()[:33]
|
|
if _HAS_BASE58
|
|
else "tz1" + secrets.token_hex(16),
|
|
private_key_hex=secrets.token_hex(32),
|
|
derivation_path=cfg.hd_path,
|
|
)
|
|
|
|
def _generate_monero(self, cfg: ChainConfig) -> WalletResult:
|
|
# Monero needs special libraries — placeholder with note
|
|
return WalletResult(
|
|
chain="",
|
|
chain_name="",
|
|
symbol="",
|
|
address="4" + base58.b58encode(secrets.token_bytes(64)).decode()[:94]
|
|
if _HAS_BASE58
|
|
else "4" + secrets.token_hex(47),
|
|
private_key_hex=secrets.token_hex(32),
|
|
derivation_path=cfg.hd_path,
|
|
tags=["monero-placeholder", "needs-monero-python-for-full-support"],
|
|
)
|
|
|
|
# ── Encryption ──────────────────────────────────────────
|
|
|
|
def _encrypt_key(self, plaintext: str) -> str:
|
|
"""AES-256-GCM encrypt with Argon2 key derivation."""
|
|
if not _HAS_CRYPTOGRAPHY or not self._password:
|
|
return plaintext # No encryption available
|
|
|
|
salt = secrets.token_bytes(16)
|
|
kdf = Argon2id(salt=salt, length=32, memory_cost=65536, time_cost=3, parallelism=4)
|
|
key = kdf.derive(self._password.encode())
|
|
|
|
aesgcm = AESGCM(key)
|
|
nonce = secrets.token_bytes(12)
|
|
ciphertext = aesgcm.encrypt(nonce, plaintext.encode(), None)
|
|
|
|
# Format: base64(salt + nonce + ciphertext)
|
|
return base64.b64encode(salt + nonce + ciphertext).decode()
|
|
|
|
def decrypt_key(self, encrypted: str) -> str:
|
|
"""Decrypt an AES-256-GCM encrypted private key."""
|
|
if not _HAS_CRYPTOGRAPHY or not self._password:
|
|
return encrypted
|
|
|
|
data = base64.b64decode(encrypted)
|
|
salt, nonce, ciphertext = data[:16], data[16:28], data[28:]
|
|
|
|
kdf = Argon2id(salt=salt, length=32, memory_cost=65536, time_cost=3, parallelism=4)
|
|
key = kdf.derive(self._password.encode())
|
|
|
|
aesgcm = AESGCM(key)
|
|
return aesgcm.decrypt(nonce, ciphertext, None).decode()
|
|
|
|
# ── Storage ─────────────────────────────────────────────
|
|
|
|
def save_to_vault(self, wallet: WalletResult, vault_path: str = "/root/.rmi/wallets/vault.json"):
|
|
"""Save wallet to encrypted vault."""
|
|
os.makedirs(os.path.dirname(vault_path), exist_ok=True, mode=0o700)
|
|
|
|
vault = {}
|
|
if os.path.exists(vault_path):
|
|
with open(vault_path) as f:
|
|
vault = json.load(f)
|
|
|
|
key = f"{wallet.chain}:{wallet.address[:12]}"
|
|
vault[key] = wallet.to_full_dict()
|
|
vault["_meta"] = {
|
|
"updated": datetime.now(UTC).isoformat(),
|
|
"total_wallets": len(vault) - 1,
|
|
"encrypted": self._encrypt,
|
|
}
|
|
|
|
with open(vault_path, "w") as f:
|
|
json.dump(vault, f, indent=2)
|
|
os.chmod(vault_path, 0o600)
|
|
|
|
logger.info(f"Wallet {key} saved to vault ({vault_path})")
|
|
|
|
# ── Rotation System ─────────────────────────────────────
|
|
|
|
def rotate(self, chain_key: str, vault_path: str = "/root/.rmi/wallets/vault.json") -> WalletResult:
|
|
"""Generate a new wallet and mark it as the active rotation for this chain."""
|
|
wallet = self.generate(chain_key, label=f"rotation_{int(time.time())}", tags=["rotated", "active"])
|
|
self.save_to_vault(wallet, vault_path)
|
|
|
|
idx = self._rotation_index.get(chain_key, 0)
|
|
self._rotation_index[chain_key] = idx + 1
|
|
|
|
logger.info(f"Rotated {chain_key} wallet → {wallet.address[:16]}... (rotation #{idx + 1})")
|
|
return wallet
|
|
|
|
# ── Stats ───────────────────────────────────────────────
|
|
|
|
@property
|
|
def stats(self) -> dict[str, Any]:
|
|
return {
|
|
"generation_count": self._generation_count,
|
|
"chains_supported": len(SUPPORTED_CHAINS),
|
|
"chain_families": len({c.family for c in SUPPORTED_CHAINS.values()}),
|
|
"rotations": dict(self._rotation_index),
|
|
"encryption_enabled": self._encrypt,
|
|
"chains": {
|
|
k: {"name": c.name, "symbol": c.symbol, "family": c.family.value} for k, c in SUPPORTED_CHAINS.items()
|
|
},
|
|
}
|
|
|
|
|
|
# ── Singleton ───────────────────────────────────────────────────
|
|
|
|
_factory: WalletFactory | None = None
|
|
|
|
|
|
def get_wallet_factory(encrypt: bool = True) -> WalletFactory:
|
|
global _factory
|
|
if _factory is None:
|
|
_factory = WalletFactory(encrypt_keys=encrypt)
|
|
return _factory
|