init proj skeleton
This commit is contained in:
409
src/hdwallet_recovery.py
Normal file
409
src/hdwallet_recovery.py
Normal file
@@ -0,0 +1,409 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
hdwallet_recovery.py (Python 3.12)
|
||||
|
||||
Subcommands:
|
||||
- fetchkey (online): download ASCII-armored PGP pubkey, print SHA256 + fingerprint
|
||||
- derive (default): derive ADDRESSES ONLY for ETH/SOL/BTC
|
||||
- optionally encrypt a secret payload using a PGP public key
|
||||
|
||||
Security invariant:
|
||||
- derive never performs network I/O
|
||||
- fetchkey refuses mnemonic/seed/passphrase inputs
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
import json
|
||||
import getpass
|
||||
import hashlib
|
||||
import urllib.request
|
||||
from dataclasses import dataclass, asdict
|
||||
from typing import Dict, List, Any, Optional
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# Dependency checks
|
||||
# -----------------------------------------------------------------------------
|
||||
|
||||
def _require(mod: str, pkg: str):
|
||||
try:
|
||||
__import__(mod)
|
||||
except ImportError:
|
||||
print(f"❌ Missing dependency: {pkg}", file=sys.stderr)
|
||||
print(f"Install with: pip install {pkg}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def require_for_fetchkey():
|
||||
_require("pgpy", "PGPy")
|
||||
|
||||
|
||||
def require_for_derive(export_private: bool, chains: List[str]):
|
||||
_require("bip_utils", "bip-utils")
|
||||
_require("pgpy", "PGPy")
|
||||
if export_private and ("solana" in chains):
|
||||
_require("nacl", "PyNaCl")
|
||||
_require("base58", "base58")
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# PGP helpers (PGPy)
|
||||
# -----------------------------------------------------------------------------
|
||||
|
||||
def pgp_load_pubkey(armored: str):
|
||||
import pgpy
|
||||
key, _ = pgpy.PGPKey.from_blob(armored)
|
||||
if not key.is_public:
|
||||
raise ValueError("Provided key is not a public key")
|
||||
return key
|
||||
|
||||
|
||||
def pgp_fingerprint(armored: str) -> str:
|
||||
return pgp_load_pubkey(armored).fingerprint
|
||||
|
||||
|
||||
def pgp_encrypt_ascii_armored(pubkey_armored: str, payload: Dict[str, Any]) -> str:
|
||||
import pgpy
|
||||
pub_key = pgp_load_pubkey(pubkey_armored)
|
||||
msg = pgpy.PGPMessage.new(json.dumps(payload, indent=2))
|
||||
enc = pub_key.encrypt(msg)
|
||||
return str(enc)
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# fetchkey (network allowed ONLY here)
|
||||
# -----------------------------------------------------------------------------
|
||||
|
||||
def fetch_ascii_armored_text(url: str, timeout: int = 15) -> str:
|
||||
req = urllib.request.Request(url, headers={"User-Agent": "hdwallet-recovery/1.0"})
|
||||
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
||||
data = resp.read()
|
||||
text = data.decode("utf-8", errors="strict")
|
||||
if "-----BEGIN PGP PUBLIC KEY BLOCK-----" not in text:
|
||||
raise ValueError("Downloaded content does not look like an ASCII-armored PGP public key")
|
||||
return text
|
||||
|
||||
|
||||
def sha256_hex_text(s: str) -> str:
|
||||
return hashlib.sha256(s.encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
def cmd_fetchkey(url: str, out_path: Optional[str], timeout: int):
|
||||
require_for_fetchkey()
|
||||
|
||||
armored = fetch_ascii_armored_text(url, timeout=timeout)
|
||||
s256 = sha256_hex_text(armored)
|
||||
fpr = pgp_fingerprint(armored)
|
||||
|
||||
if out_path:
|
||||
with open(out_path, "w", encoding="utf-8") as f:
|
||||
f.write(armored)
|
||||
|
||||
print("✅ Downloaded PGP public key")
|
||||
print(f"URL: {url}")
|
||||
if out_path:
|
||||
print(f"Saved: {out_path}")
|
||||
print(f"SHA256: {s256}")
|
||||
print(f"Fingerprint: {fpr}")
|
||||
|
||||
if not out_path:
|
||||
print("\n-----BEGIN DOWNLOADED KEY-----")
|
||||
print(armored.strip())
|
||||
print("-----END DOWNLOADED KEY-----")
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# Derivation
|
||||
# -----------------------------------------------------------------------------
|
||||
|
||||
@dataclass
|
||||
class AddrOut:
|
||||
index: int
|
||||
path: str
|
||||
address: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class BtcAddrOut:
|
||||
index: int
|
||||
path: str
|
||||
address_type: str
|
||||
address: str
|
||||
|
||||
|
||||
def _solana_phantom_export_from_seed32(seed32: bytes) -> Dict[str, Any]:
|
||||
# Phantom-compatible 64-byte secret key = seed32 || pubkey32 (ed25519).
|
||||
# PyNaCl SigningKey(seed32) supports 32-byte seeds.
|
||||
from nacl.signing import SigningKey
|
||||
import base58
|
||||
|
||||
sk = SigningKey(seed32)
|
||||
pub32 = sk.verify_key.encode()
|
||||
secret64 = seed32 + pub32
|
||||
|
||||
return {
|
||||
"phantom_base58": base58.b58encode(secret64).decode("ascii"),
|
||||
}
|
||||
|
||||
|
||||
def derive_addresses_and_maybe_secrets(seed_bytes: bytes, chains: List[str], count: int, export_private: bool) -> Dict[str, Any]:
|
||||
from bip_utils import (
|
||||
Bip44, Bip44Coins, Bip44Changes,
|
||||
Bip49, Bip49Coins,
|
||||
Bip84, Bip84Coins,
|
||||
)
|
||||
|
||||
out: Dict[str, Any] = {"addresses": {}}
|
||||
if export_private:
|
||||
out["secrets"] = {}
|
||||
|
||||
# ETH (BIP44 m/44'/60'/0'/0/i)
|
||||
if "ethereum" in chains:
|
||||
root = Bip44.FromSeed(seed_bytes, Bip44Coins.ETHEREUM)
|
||||
ctx = root.Purpose().Coin().Account(0).Change(Bip44Changes.CHAIN_EXT)
|
||||
|
||||
addrs: List[Dict[str, Any]] = []
|
||||
secrets: List[Dict[str, Any]] = []
|
||||
for i in range(count):
|
||||
node = ctx.AddressIndex(i)
|
||||
addrs.append({"index": i, "path": f"m/44'/60'/0'/0/{i}", "address": node.PublicKey().ToAddress()})
|
||||
if export_private:
|
||||
secrets.append({
|
||||
"index": i,
|
||||
"path": f"m/44'/60'/0'/0/{i}",
|
||||
"privkey_hex": node.PrivateKey().Raw().ToHex(),
|
||||
})
|
||||
|
||||
out["addresses"]["ethereum"] = addrs
|
||||
if export_private and secrets:
|
||||
out["secrets"]["ethereum"] = secrets
|
||||
|
||||
# SOL (BIP44 m/44'/501'/0'/0'/i')
|
||||
if "solana" in chains:
|
||||
root = Bip44.FromSeed(seed_bytes, Bip44Coins.SOLANA)
|
||||
ctx = root.Purpose().Coin().Account(0).Change(Bip44Changes.CHAIN_EXT)
|
||||
|
||||
addrs: List[Dict[str, Any]] = []
|
||||
secrets: List[Dict[str, Any]] = []
|
||||
for i in range(count):
|
||||
node = ctx.AddressIndex(i)
|
||||
addrs.append({"index": i, "path": f"m/44'/501'/0'/0'/{i}'", "address": node.PublicKey().ToAddress()})
|
||||
if export_private:
|
||||
seed32 = node.PrivateKey().Raw().ToBytes()
|
||||
if len(seed32) != 32:
|
||||
raise ValueError(f"Unexpected Solana private key length from bip_utils: {len(seed32)} bytes")
|
||||
secrets.append({
|
||||
"index": i,
|
||||
"path": f"m/44'/501'/0'/0'/{i}'",
|
||||
"phantom": _solana_phantom_export_from_seed32(seed32),
|
||||
})
|
||||
|
||||
out["addresses"]["solana"] = addrs
|
||||
if export_private and secrets:
|
||||
out["secrets"]["solana"] = secrets
|
||||
|
||||
# BTC (addresses only even with --export-private)
|
||||
if "bitcoin" in chains:
|
||||
addrs: List[Dict[str, Any]] = []
|
||||
|
||||
r44 = Bip44.FromSeed(seed_bytes, Bip44Coins.BITCOIN)
|
||||
c44 = r44.Purpose().Coin().Account(0).Change(Bip44Changes.CHAIN_EXT)
|
||||
|
||||
r49 = Bip49.FromSeed(seed_bytes, Bip49Coins.BITCOIN)
|
||||
c49 = r49.Purpose().Coin().Account(0).Change(Bip44Changes.CHAIN_EXT)
|
||||
|
||||
r84 = Bip84.FromSeed(seed_bytes, Bip84Coins.BITCOIN)
|
||||
c84 = r84.Purpose().Coin().Account(0).Change(Bip44Changes.CHAIN_EXT)
|
||||
|
||||
for i in range(count):
|
||||
n84 = c84.AddressIndex(i)
|
||||
n49 = c49.AddressIndex(i)
|
||||
n44 = c44.AddressIndex(i)
|
||||
|
||||
addrs.append({"index": i, "path": f"m/84'/0'/0'/0/{i}", "address_type": "native_segwit", "address": n84.PublicKey().ToAddress()})
|
||||
addrs.append({"index": i, "path": f"m/49'/0'/0'/0/{i}", "address_type": "segwit", "address": n49.PublicKey().ToAddress()})
|
||||
addrs.append({"index": i, "path": f"m/44'/0'/0'/0/{i}", "address_type": "legacy", "address": n44.PublicKey().ToAddress()})
|
||||
|
||||
out["addresses"]["bitcoin"] = addrs
|
||||
|
||||
# Don't create empty secrets dicts
|
||||
if export_private and "secrets" in out:
|
||||
out["secrets"] = {k: v for k, v in out["secrets"].items() if v}
|
||||
if not out["secrets"]:
|
||||
del out["secrets"]
|
||||
|
||||
return out
|
||||
|
||||
|
||||
def format_addresses_human(result: Dict[str, Any]) -> str:
|
||||
lines: List[str] = []
|
||||
lines.append("\n" + "=" * 80)
|
||||
lines.append("MULTI-CHAIN ADDRESS DERIVATION (ADDRESSES ONLY)")
|
||||
lines.append("=" * 80 + "\n")
|
||||
|
||||
for chain, addrs in result["addresses"].items():
|
||||
lines.append("─" * 80)
|
||||
lines.append(f"{chain.upper()} ADDRESSES")
|
||||
lines.append("─" * 80)
|
||||
|
||||
if chain == "bitcoin":
|
||||
by_type: Dict[str, List[Dict[str, Any]]] = {}
|
||||
for a in addrs:
|
||||
by_type.setdefault(a["address_type"], []).append(a)
|
||||
for t in ["native_segwit", "segwit", "legacy"]:
|
||||
if t in by_type:
|
||||
lines.append(f"\n {t.upper()}:")
|
||||
for a in by_type[t]:
|
||||
lines.append(f" [{a['index']}] {a['path']}")
|
||||
lines.append(f" → {a['address']}")
|
||||
else:
|
||||
for a in addrs:
|
||||
lines.append(f" [{a['index']}] {a['path']}")
|
||||
lines.append(f" → {a['address']}")
|
||||
lines.append("")
|
||||
|
||||
lines.append("=" * 80 + "\n")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def cmd_derive(args):
|
||||
require_for_derive(args.export_private, args.chains)
|
||||
|
||||
from bip_utils import Bip39MnemonicValidator, Bip39SeedGenerator
|
||||
|
||||
if args.export_private and not args.pgp_pubkey_file:
|
||||
raise ValueError("--export-private requires --pgp-pubkey-file (secrets must not go to stdout)")
|
||||
|
||||
mnemonic = None
|
||||
seed_hex = None
|
||||
|
||||
if args.interactive:
|
||||
mode = input("Enter 'm' for mnemonic or 's' for seed: ").strip().lower()
|
||||
if mode == "m":
|
||||
mnemonic = getpass.getpass("BIP39 mnemonic (hidden): ").strip()
|
||||
elif mode == "s":
|
||||
seed_hex = getpass.getpass("Seed hex (hidden): ").strip()
|
||||
else:
|
||||
raise ValueError("Invalid choice")
|
||||
elif args.mnemonic:
|
||||
mnemonic = args.mnemonic.strip()
|
||||
elif args.seed:
|
||||
seed_hex = args.seed.strip()
|
||||
else:
|
||||
raise ValueError("Missing input")
|
||||
|
||||
if mnemonic:
|
||||
if not Bip39MnemonicValidator().IsValid(mnemonic):
|
||||
raise ValueError("Invalid BIP39 mnemonic")
|
||||
seed_bytes = Bip39SeedGenerator(mnemonic).Generate(args.passphrase or "")
|
||||
else:
|
||||
b = bytes.fromhex(seed_hex)
|
||||
if len(b) != 64:
|
||||
raise ValueError(f"Seed must be 64 bytes (128 hex chars), got {len(b)}")
|
||||
seed_bytes = b
|
||||
|
||||
print(f"📍 Deriving {len(args.chains)} chain(s), {args.addresses} address(es) each...")
|
||||
result = derive_addresses_and_maybe_secrets(seed_bytes, args.chains, args.addresses, args.export_private)
|
||||
|
||||
# Always print addresses-only output (safe)
|
||||
if args.output == "json":
|
||||
out_text = json.dumps({"addresses": result["addresses"]}, indent=2)
|
||||
else:
|
||||
out_text = format_addresses_human({"addresses": result["addresses"]})
|
||||
print(out_text)
|
||||
|
||||
if args.file:
|
||||
with open(args.file, "w", encoding="utf-8") as f:
|
||||
f.write(out_text)
|
||||
print(f"✅ Addresses saved to {args.file}")
|
||||
|
||||
# Optional encrypted payload
|
||||
if args.pgp_pubkey_file:
|
||||
with open(args.pgp_pubkey_file, "r", encoding="utf-8") as f:
|
||||
pgp_pub = f.read()
|
||||
|
||||
payload: Dict[str, Any] = {
|
||||
"version": "v4",
|
||||
"purpose": "hdwallet recovery secret payload",
|
||||
}
|
||||
|
||||
# Always include mnemonic/seed_hex if present (your requirement)
|
||||
if mnemonic:
|
||||
payload["mnemonic"] = mnemonic
|
||||
else:
|
||||
payload["seed_hex"] = seed_hex
|
||||
|
||||
if args.export_private:
|
||||
payload["passphrase_set"] = bool(args.passphrase)
|
||||
payload["passphrase_hint"] = args.passphrase_hint or ""
|
||||
payload["note"] = "Private keys were derived from mnemonic/seed + (optional) passphrase. Passphrase value is intentionally omitted."
|
||||
payload["derived_private_keys"] = result.get("secrets", {})
|
||||
payload["addresses"] = result["addresses"] # include for verification
|
||||
else:
|
||||
payload["passphrase"] = args.passphrase or ""
|
||||
|
||||
armored = pgp_encrypt_ascii_armored(pgp_pub, payload)
|
||||
print("\n===== BEGIN PGP ENCRYPTED PAYLOAD =====")
|
||||
print(armored)
|
||||
print("===== END PGP ENCRYPTED PAYLOAD =====\n")
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# Main
|
||||
# -----------------------------------------------------------------------------
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="HD wallet recovery (addresses only) + fetchkey helper")
|
||||
subparsers = parser.add_subparsers(dest="cmd")
|
||||
|
||||
p_fetch = subparsers.add_parser("fetchkey", help="Download ASCII-armored PGP pubkey from URL")
|
||||
p_fetch.add_argument("url", help="URL to fetch (e.g., https://github.com/<user>.gpg)")
|
||||
p_fetch.add_argument("--out", help="Write key to file (recommended)", default=None)
|
||||
p_fetch.add_argument("--timeout", type=int, default=15, help="HTTP timeout seconds")
|
||||
|
||||
# derive mode args (default when no subcommand is given)
|
||||
parser.add_argument("--mnemonic", help="BIP39 mnemonic (12/24 words)")
|
||||
parser.add_argument("--seed", help="64-byte seed hex (128 hex chars)")
|
||||
parser.add_argument("--interactive", action="store_true", help="Prompt for input via stdin")
|
||||
parser.add_argument("--passphrase", default="", help="BIP39 passphrase (optional)")
|
||||
parser.add_argument("--passphrase-hint", default="", help="Hint/reminder for passphrase (stored only in encrypted payload when --export-private)")
|
||||
parser.add_argument("--chains", nargs="+", choices=["ethereum", "solana", "bitcoin"],
|
||||
default=["ethereum", "solana", "bitcoin"])
|
||||
parser.add_argument("--addresses", type=int, default=5)
|
||||
parser.add_argument("--output", choices=["text", "json"], default="text")
|
||||
parser.add_argument("--file", help="Save address output to file", default=None)
|
||||
parser.add_argument("--pgp-pubkey-file", help="ASCII-armored PGP public key file to encrypt secrets", default=None)
|
||||
parser.add_argument("--export-private", action="store_true",
|
||||
help="Encrypt derived private keys into the PGP payload (never printed). Requires --pgp-pubkey-file.")
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
try:
|
||||
if args.cmd == "fetchkey":
|
||||
forbidden = any([
|
||||
getattr(args, "mnemonic", None),
|
||||
getattr(args, "seed", None),
|
||||
getattr(args, "interactive", False),
|
||||
getattr(args, "passphrase", ""),
|
||||
getattr(args, "passphrase_hint", ""),
|
||||
getattr(args, "pgp_pubkey_file", None),
|
||||
getattr(args, "export_private", False),
|
||||
])
|
||||
if forbidden:
|
||||
raise ValueError("fetchkey mode must not be used with mnemonic/seed/passphrase/export-private options")
|
||||
cmd_fetchkey(args.url, args.out, args.timeout)
|
||||
return
|
||||
|
||||
if not (args.mnemonic or args.seed or args.interactive):
|
||||
raise ValueError("Provide --mnemonic or --seed or --interactive (or use subcommand fetchkey)")
|
||||
|
||||
cmd_derive(args)
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Error: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user