#!/usr/bin/env python3 """ wallet_tools.py — read-only public wallet-data helpers for the Hermes Telegram bot. Pure stdlib (urllib). No private keys, no credential capture, no network fallback beyond well-known public read-only RPC/explorers. Nothing here stores secrets. Chains supported (read-only RPC/explorer APIs): solana ~ api.mainnet.solana.com + solscan.io evm ~ quicknode/blockscout / solscan token api bitcoin ~ mempool.space/api/address/{addr} Usage: python3 tools/wallet_tools.py --selftest python3 tools/wallet_tools.py --fetch
python3 -c "import tools.wallet_tools as w; print(w.fetch_balance('...'))" """ from __future__ import annotations import base64 import binascii import json import re import urllib.request import urllib.error from typing import Any, Dict, List, Optional # ---- well-known, public read-only endpoints (no keys) ---- SOLANA_RPC = "https://api.mainnet.solana.com" SOLSCAN_API = "https://public-api.solscan.io" BLOCKSCOUT_ETH = "https://eth.blockscout.com" BLOCKSCOUT_BASE = "https://nd-.nd.ethersapp.io" # placeholder; not used MEMPOOL_SPACE = "https://mempool.space/api" # Bare minimum solscan signature for token holdings SOLSCAN_TOKEN_SUMMARY = SOLSCAN_API + "/token/%s/token-holders-count" # Chain heuristics _EVM_RE = re.compile(r"^0x[a-fA-F0-9]{40}$") _SOL_RE = re.compile(r"^[1-9a-km-zA-KM-Z]{32,44}$") _BTC_RE = re.compile(r"^[13][1-9A-HJ-NP-Za-km-z]{25,35}$") _BECH32_RE = re.compile(r"^bc1[qpzry9x8gf2tvdw0s3jn54khce6mua7l]+$") def detect_chain(address: str) -> Dict[str, Any]: """Guess the chain a public address belongs to.""" a = address.strip() if _EVM_RE.match(a): return {"detected": "evm", "address": a, "valid": True} if _SOL_RE.match(a): return {"detected": "solana", "address": a, "valid": True} if _BTC_RE.match(a): return {"detected": "bitcoin", "address": a, "valid": True} if _BECH32_RE.match(a): return {"detected": "bitcoin_bech32", "address": a, "valid": True} return {"detected": None, "address": a, "valid": False, "reason": "unknown format"} def decode(data: str) -> Dict[str, Any]: """Decode hex / utf-8 / base64 payloads (read-only).""" t = data.strip() if not t: return {"ok": False, "payload": "", "reason": "empty input"} if len(t) < 4: return {"ok": False, "payload": t, "reason": "too short"} # hex try: if t.lower().startswith("0x"): t = t[2:] if len(t) % 2 == 0 and re.fullmatch(r"[0-9a-fA-F]+", t): return {"ok": True, "payload": bytes.fromhex(t).decode("utf-8", errors="replace")} except Exception: pass # base64 try: dec = base64.b64decode(t, validate=True) return {"ok": True, "payload": dec.decode("utf-8", errors="replace")} except Exception: pass # plain utf-8 (maybe) return {"ok": False, "payload": t, "reason": "not hex/base64"} def _get(url: str, timeout: float = 6.0) -> Optional[Dict[str, Any]]: try: req = urllib.request.Request(url, headers={"User-Agent": "hermes-bot-miniapp/1.0"}) with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read().decode("utf-8", errors="replace")) except Exception as e: return {"error": f"{type(e).__name__}: {e}"} def _rpc_get(url: str, timeout: float = 6.0) -> Dict[str, Any]: try: req = urllib.request.Request(url, headers={"Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read().decode("utf-8", errors="replace")) except Exception as e: return {"error": f"{type(e).__name__}: {e}"} def fetch_balance(address: str, chain: Optional[str] = None) -> Dict[str, Any]: """Read-only balance fetch for solana / evm / bitcoin.""" a = address.strip() _det = detect_chain(a) chain = (_det.get("detected") or "").lower() if _det.get("valid") else None if chain == "evm": out = {"address": a, "chain": "evm", "balance": {}} res = _rpc_get(f"https://eth.blockscout.com/api/v2/address/{a}/balance") if "error" not in res: try: out["balance"]["eth"] = float(res.get("balance", 0)) / 1e18 except Exception: pass tok = _rpc_get(f"https://eth.blockscout.com/api/v2/address/{a}/tokens") out["tokens"] = (tok or {}).get("tokens", []) return out if chain == "solana": out = {"address": a, "chain": "solana", "balance": {}} res = _rpc_get(f'{SOLANA_RPC}?jsonrpc=2.0&method=getBalance¶ms=["{a}"],"commitment":"finalized"') if "error" not in res: try: out["balance"]["sol"] = float(res.get("result", {}).get("value", 0)) / 1e9 except Exception: pass return out if chain == "bitcoin" or chain == "bitcoin_bech32": out = {"address": a, "chain": "bitcoin", "balance": {}} res = _get(f"{MEMPOOL_SPACE}/address/{a}") if "error" not in res: cs = res.get("chain_stats") or {} unspent = (cs.get("funded_txo_sum") or 0) - (cs.get("spent_txo_sum") or 0) out["balance"] = {"sat": int(unspent), "btc": round(int(unspent) / 1e8, 8), "tx_count": cs.get("tx_count")} out["balance"]["raw"] = cs return out return {"address": a, "chain": None, "error": "no reachable RPC for chain format", "balance": {}, "detected": detect_chain(a)} def fetch_assets_chains(address: str) -> Dict[str, Any]: """Fallback: fetch address summary from starting explorers; best-effort.""" a = address.strip() chain_info = detect_chain(a) if not chain_info.get("valid"): return {"status": "error", "address": a, "detected": chain_info.get("detected"), "chains": []} out = {"status": "ok", "address": a, "chains": []} det = chain_info["detected"] if det in ("evm", "bitcoin_bech32", "bitcoin"): bal = fetch_balance(a, det) out["chains"].append({"chain": det, "balance": bal.get("balance")}) return out if __name__ == "__main__": import sys argv = [a for a in sys.argv[1:] if not a.startswith("--")] if argv and argv[0] == "selftest": ok = True print("SELFTEST:", detect_chain("0x863DF6BFa4469f3ead0bE8f9F2AAE51c91A907b4")) print("SELFTEST:", detect_chain("1A1zP1eP5QGefi2DMPTfTL5SLmv7DivfNa")) print("SELFTEST:", detect_chain("0xNOTAVALID")) print("SELFTEST:", decode("0x536f756c6c6573")["payload"]) print("SELFTEST:", decode("Y29sbGVjdA==").get("payload")) print("SELFTEST:", decode("plain text")["payload"]) print("SELFTEST OK") elif argv and argv[0] == "fetch": print(json.dumps(fetch_assets_chains(argv[1]), indent=2)) else: print("Sign request: pending")