""" btc.py — Comprobar tus bitcoins contra tu propio nodo. Qué hace, en cristiano: 1. Guardas las claves públicas extendidas (xpub / ypub / zpub) de tus monederos. No son claves privadas: solo sirven para VER, nunca para gastar. 2. La aplicación deriva las direcciones de cada monedero (las que ya has usado y las siguientes, hasta 20 seguidas sin estrenar) y le pregunta a tu servidor Electrum cuánto hay en ellas. 3. Con eso, historial_btc.py compara lo que hay con lo que tienes apuntado y te propone qué apuntar de cada transacción: lo nuevo cada día (la comprobación diaria) o todo el pasado («Revisar todo el historial»). Aquí vive lo básico: la configuración, las direcciones de cada monedero y la consulta al servidor. Las propuestas se hacen en historial_btc.py. Si el monedero no enseña su clave extendida, se pueden dar sus direcciones una a una: las de recibir Y las de cambio, porque la vuelta de un pago que no esté en la lista parecería bitcoin que se fue fuera. La comprobación diaria no trae el pasado: al dar de alta un monedero, todo lo que ya hay se marca como «visto» y solo se avisa de lo que llegue a partir de entonces. El pasado se revisa, cuando se quiera, en el historial completo. Cuidado con la privacidad: quien responda a estas consultas sabe qué direcciones son tuyas. Por eso el servidor debe ser el tuyo, y se llega a él por Tor. """ from __future__ import annotations import datetime as dt import json import os import tempfile import uuid from pathlib import Path from . import bitcoin from .electrum import Cliente, ErrorElectrum, conectar from . import almacen FICHERO = "bitcoin.json" HUECO = 20 # direcciones seguidas sin usar que dan por terminada una rama MAX_DIRECCIONES = 2000 # tope de seguridad por rama MAX_VISTAS = 5000 # txids recordados por monedero MAX_SUELTAS = 5000 # direcciones sueltas por monedero TANDA = 50 # direcciones por consulta al servidor SATOSHI = 100_000_000 class ErrorBitcoin(RuntimeError): """Problema de configuración o de consulta que se le enseña al usuario.""" # ── Configuración ──────────────────────────────────────────────────────────── def ruta_config(carpeta_datos: Path) -> Path: return carpeta_datos.parent / FICHERO def cargar(carpeta_datos: Path) -> dict: try: c = json.loads(almacen.leer_texto(ruta_config(carpeta_datos))) except (OSError, ValueError): c = {} if not isinstance(c, dict): c = {} c.setdefault("servidores", []) c.setdefault("monederos", []) c.setdefault("proxy", {"host": "127.0.0.1", "puerto": 9050}) c.setdefault("comprobar_al_abrir", True) c.setdefault("ultima", None) return c def guardar(carpeta_datos: Path, config: dict) -> None: ruta = ruta_config(carpeta_datos) # La marca de fecha decide quién gana al sincronizar (ver sincronizar.py) config = {**config, "modificado": dt.datetime.now().isoformat(timespec="seconds")} # Las claves extendidas delatan todo tu historial: el fichero solo lo lee su dueño tmp = tempfile.NamedTemporaryFile("wb", dir=ruta.parent, prefix=".bitcoin-", suffix=".tmp", delete=False) try: tmp.write(almacen.sellar(json.dumps(config, ensure_ascii=False, indent=1).encode("utf-8"))) tmp.flush() os.fsync(tmp.fileno()) tmp.close() os.chmod(tmp.name, 0o600) os.replace(tmp.name, ruta) except OSError: try: os.unlink(tmp.name) except OSError: pass raise def proxy_de(config: dict) -> tuple[str, int]: p = config.get("proxy") or {} return str(p.get("host") or "127.0.0.1"), int(p.get("puerto") or 9050) def normalizar_servidor(s: dict) -> dict: host = str(s.get("host") or "").strip() if not host: raise ErrorBitcoin("Pon la dirección del servidor (un .onion, un nombre o una IP).") if "://" in host: host = host.split("://", 1)[1] host = host.strip("/").split("/")[0] if host.count(":") == 1: # admite «maquina:50001» host, _, puerto_pegado = host.partition(":") s = {**s, "puerto": s.get("puerto") or puerto_pegado} core = s.get("clase") == "core" try: puerto = int(s.get("puerto") or (8332 if core else 50001)) except (TypeError, ValueError): raise ErrorBitcoin("El puerto tiene que ser un número (50001 sin cifrar, 50002 con TLS; " "8332 si es el nodo directamente).") from None if not 1 <= puerto <= 65535: raise ErrorBitcoin("El puerto está fuera de rango.") normal = {"nombre": str(s.get("nombre") or host)[:60], "host": host, "puerto": puerto, "tls": bool(s.get("tls")), "tor": bool(s.get("tor", host.endswith(".onion")))} if core: # El nodo directamente (Bitcoin Core o Knots), con su usuario y contraseña de RPC: ver core.py normal |= {"clase": "core", "usuario": str(s.get("usuario") or "")[:120], "clave": str(s.get("clave") or "")[:200]} return normal def sin_claves(servidores: list[dict]) -> list[dict]: """La lista de servidores como se le enseña a la pantalla: la contraseña de RPC no sale del programa (solo se dice si la hay).""" return [{k: v for k, v in s.items() if k != "clave"} | ({"con_clave": bool(s.get("clave"))} if s.get("clase") == "core" else {}) for s in servidores] def con_su_clave(servidor: dict, guardados: list[dict]) -> dict: """Si la pantalla devuelve un nodo sin contraseña (no la tiene), se le pone la que ya estaba guardada para ese mismo nodo y usuario.""" if servidor.get("clase") != "core" or servidor.get("clave"): return servidor for g in guardados: if g.get("clase") == "core" and g.get("clave") and all( g.get(k) == servidor.get(k) for k in ("host", "puerto", "usuario")): return {**servidor, "clave": g["clave"]} return servidor def vigilable(m: dict) -> bool: """Monedero que se puede consultar: con clave extendida o con direcciones sueltas.""" return bool(m.get("clave") or m.get("direcciones")) def parte_de(m: dict) -> float: """Qué parte de lo que hay en el monedero es tuya (1 = todo; 0,5 = la mitad, si lo compartes con otra persona).""" try: p = float(m.get("parte") or 1) except (TypeError, ValueError): return 1.0 return p if 0 < p <= 1 else 1.0 def _leer_parte(m: dict, anterior: dict) -> float: v = m.get("parte") if v in (None, ""): return parte_de(anterior) if anterior else 1.0 try: p = float(str(v).replace(",", ".")) except ValueError: raise ErrorBitcoin("Tu parte del monedero tiene que ser un porcentaje.") from None if not 0 < p <= 1: raise ErrorBitcoin("Tu parte del monedero tiene que estar entre el 1 % y el 100 %.") return round(p, 4) def leer_direcciones(texto) -> list[str]: """Lista limpia de direcciones: acepta una lista o un texto con una por línea (o separadas por comas o espacios). Sin repetidas y en el orden dado.""" if isinstance(texto, list): trozos = [str(t) for t in texto] else: trozos = str(texto or "").replace(",", " ").replace(";", " ").split() vistas, salida = set(), [] for t in trozos: t = t.strip().strip('"\'') if t.lower().startswith("bitcoin:"): # un enlace «bitcoin:bc1…?amount=…» t = t[8:].split("?")[0] if t.lower().startswith(("bc1", "tb1")): t = t.lower() if t and t not in vistas: vistas.add(t) salida.append(t) return salida def _lugar_de(m: dict, anterior: dict) -> str: """El lugar del monedero dentro de su activo (ver lugares.py). Si cambia de activo, el lugar viejo no vale: se le asignará uno al leer el nodo.""" if m.get("lugar") is not None: return str(m.get("lugar") or "")[:40] if anterior and anterior.get("activo") == m.get("activo"): return str(anterior.get("lugar") or "") return "" def _normalizar_sueltas(m: dict, existentes: list[dict]) -> dict: direcciones = leer_direcciones(m.get("direcciones")) if not direcciones: raise ErrorBitcoin("Pega al menos una dirección del monedero.") if len(direcciones) > MAX_SUELTAS: raise ErrorBitcoin(f"Como mucho {MAX_SUELTAS} direcciones por monedero.") redes, malas = set(), [] for d in direcciones: try: redes.add(bitcoin.script_de_direccion(d)[1]) except ValueError as e: malas.append(f"{d[:14]}…: {e}" if len(d) > 16 else f"{d}: {e}") if malas: raise ErrorBitcoin("Revisa estas direcciones: " + "; ".join(malas[:5]) + (f" (y {len(malas) - 5} más)" if len(malas) > 5 else "") + ".") if len(redes) > 1: raise ErrorBitcoin("Mezclas direcciones de la red principal y de testnet.") if not m.get("activo"): raise ErrorBitcoin("Elige a qué activo de la cartera corresponde este monedero.") mid = m.get("id") or "w" + uuid.uuid4().hex[:12] mias = set(direcciones) for o in existentes: if o.get("id") != mid and mias & set(o.get("direcciones") or []): raise ErrorBitcoin(f"Alguna de esas direcciones ya está en «{o.get('nombre')}».") anterior = next((o for o in existentes if o.get("id") == mid), {}) return {"id": mid, "nombre": str(m.get("nombre") or "Monedero")[:60], "direcciones": direcciones, "tipo": "sueltas", "red": redes.pop(), "activo": str(m["activo"]), "lugar": _lugar_de(m, anterior), "parte": _leer_parte(m, anterior), "actividad": str(m.get("actividad") or "")[:40], "inicial": bool(anterior.get("inicial", True) if anterior else m.get("inicial", True)), "vistas": list(anterior.get("vistas") or []), "saldo": anterior.get("saldo"), "comprobado": anterior.get("comprobado"), "direcciones_usadas": anterior.get("direcciones_usadas"), "revision": anterior.get("revision") if anterior.get("direcciones") == direcciones else None} def normalizar_monedero(m: dict, existentes: list[dict]) -> dict: if m.get("direcciones") is not None and not m.get("clave"): return _normalizar_sueltas(m, existentes) clave = str(m.get("clave") or "").strip().replace(" ", "") if not clave: raise ErrorBitcoin("Pega la clave pública extendida del monedero (empieza por xpub, ypub o zpub).") try: extendida = bitcoin.ClaveExtendida(clave) except ValueError as e: raise ErrorBitcoin(f"Esa clave no vale: {e}.") from None tipo = m.get("tipo") or extendida.tipo_sugerido if tipo not in bitcoin.TIPOS: raise ErrorBitcoin("Tipo de direcciones desconocido.") if not m.get("activo"): raise ErrorBitcoin("Elige a qué activo de la cartera corresponde este monedero.") mid = m.get("id") or "w" + uuid.uuid4().hex[:12] repetido = next((o for o in existentes if o.get("clave") == clave and o.get("id") != mid), None) if repetido: raise ErrorBitcoin(f"Esa misma clave ya está guardada como «{repetido.get('nombre')}».") anterior = next((o for o in existentes if o.get("id") == mid), {}) return {"id": mid, "nombre": str(m.get("nombre") or "Monedero")[:60], "clave": clave, "tipo": tipo, "red": extendida.red, "activo": str(m["activo"]), "lugar": _lugar_de(m, anterior), "parte": _leer_parte(m, anterior), "hueco": max(5, min(100, int(m.get("hueco") or HUECO))), "actividad": str(m.get("actividad") or "")[:40], # Un monedero nuevo arranca con todo su pasado dado por apuntado "inicial": bool(anterior.get("inicial", True) if anterior else m.get("inicial", True)), "vistas": list(anterior.get("vistas") or []), "saldo": anterior.get("saldo"), "comprobado": anterior.get("comprobado"), "direcciones_usadas": anterior.get("direcciones_usadas"), "revision": anterior.get("revision") if anterior.get("clave") == clave else None} # ── Escaneo de un monedero ─────────────────────────────────────────────────── def _rama(cliente: Cliente, extendida, tipo: str, cambio: int, hueco: int) -> tuple[dict, list[dict], bool]: """Recorre una rama hasta encontrar «hueco» direcciones seguidas sin usar. Devuelve también si se ha llegado al tope de seguridad sin terminar, que querría decir que el saldo enseñado se queda corto. """ scripts: dict[str, dict] = {} historial: list[dict] = [] desde, vacias = 0, 0 tope = min(MAX_DIRECCIONES, cliente.tope_rama or MAX_DIRECCIONES) while vacias < hueco and desde < tope: tanda = bitcoin.direcciones(extendida, tipo, cambio, desde, hueco) respuestas = cliente.historiales([d["huella"] for d in tanda]) for d, h in zip(tanda, respuestas): scripts[d["script"]] = d if h: vacias = 0 d["usada"] = True for t in h: historial.append({"txid": t.get("tx_hash"), "altura": int(t.get("height") or 0)}) else: vacias += 1 if vacias >= hueco: break desde += hueco return scripts, historial, desde >= tope and vacias < hueco def _escanear_sueltas(cliente: Cliente, monedero: dict) -> dict: scripts: dict[str, dict] = {} for i, d in enumerate(monedero.get("direcciones") or []): script, _ = bitcoin.script_de_direccion(d) scripts[script.hex()] = {"cambio": None, "indice": i, "script": script.hex(), "huella": bitcoin.huella_script(script), "direccion": d} lista = list(scripts.values()) historial = [] for i in range(0, len(lista), TANDA): tanda = lista[i:i + TANDA] for d, h in zip(tanda, cliente.historiales([x["huella"] for x in tanda])): if h: d["usada"] = True historial += [{"txid": t.get("tx_hash"), "altura": int(t.get("height") or 0)} for t in h] usadas = [d for d in lista if d.get("usada")] sats = 0 for i in range(0, len(usadas), TANDA): saldos = cliente.saldos([d["huella"] for d in usadas[i:i + TANDA]]) sats += sum(int(s.get("confirmed") or 0) + int(s.get("unconfirmed") or 0) for s in saldos if isinstance(s, dict)) return {"scripts": scripts, "historial": historial, "sats": sats, "usadas": len(usadas), "cortada": False} def escanear(cliente: Cliente, monedero: dict) -> dict: """Direcciones, saldo e historial de un monedero.""" cliente.preparar(monedero) if not monedero.get("clave") and monedero.get("direcciones"): return _escanear_sueltas(cliente, monedero) extendida = bitcoin.ClaveExtendida(monedero["clave"]) tipo = monedero.get("tipo") or extendida.tipo_sugerido hueco = int(monedero.get("hueco") or HUECO) scripts, historial, cortada = {}, [], False for cambio in (0, 1): s, h, corte = _rama(cliente, extendida, tipo, cambio, hueco) scripts.update(s) historial += h cortada = cortada or corte usadas = [d for d in scripts.values() if d.get("usada")] saldos = cliente.saldos([d["huella"] for d in usadas]) if usadas else [] sats = sum(int(s.get("confirmed") or 0) + int(s.get("unconfirmed") or 0) for s in saldos if isinstance(s, dict)) return {"scripts": scripts, "historial": historial, "sats": sats, "usadas": len(usadas), "cortada": cortada} def revisar(cliente: Cliente, monedero: dict) -> dict: """Qué tiene de verdad cada monedero, para saber si está bien dado de alta. Con clave extendida se prueban LOS CUATRO tipos de dirección (no solo el elegido): si el elegido no tiene nada y otro sí, el tipo está mal. Con direcciones sueltas se dice cuáles no tienen ninguna transacción (una errata o una dirección que no era de este monedero).""" fecha = dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds") cliente.preparar(monedero) if not monedero.get("clave") and monedero.get("direcciones"): r = _escanear_sueltas(cliente, monedero) usadas = {d["direccion"] for d in r["scripts"].values() if d.get("usada")} return {"fecha": fecha, "clase": "sueltas", "total": len(monedero["direcciones"]), "usadas": len(usadas), "transacciones": len({t["txid"] for t in r["historial"]}), "sin_uso": [d for d in monedero["direcciones"] if d not in usadas]} extendida = bitcoin.ClaveExtendida(monedero["clave"]) hueco = int(monedero.get("hueco") or HUECO) tipos = {} for tipo in bitcoin.TIPOS: usadas, txids = 0, set() for cambio in (0, 1): scripts, historial, _ = _rama(cliente, extendida, tipo, cambio, hueco) usadas += sum(1 for d in scripts.values() if d.get("usada")) txids |= {t["txid"] for t in historial if t["txid"]} tipos[tipo] = {"usadas": usadas, "transacciones": len(txids)} return {"fecha": fecha, "clase": "clave", "elegido": monedero.get("tipo") or extendida.tipo_sugerido, "tipos": tipos} def tipo_de_partida(extendida) -> str: """El tipo de direcciones que se supone a una clave cuando no hay forma de saberlo. El prefijo «ypub» o «zpub» lo dice; «xpub» no dice nada: muchos monederos dan una xpub y usan direcciones bc1q, así que se parte de esas.""" return "p2wpkh" if extendida.tipo_sugerido == "p2pkh" else extendida.tipo_sugerido def detectar(cliente: Cliente, clave: str) -> tuple[str, bool]: """(tipo de direcciones, si se ha visto en el nodo). Prueba los cuatro tipos y se queda con el que tenga historial; el prefijo de la clave solo decide por cuál empezar y qué proponer si el monedero está sin estrenar.""" extendida = bitcoin.ClaveExtendida(clave) cliente.preparar({"clave": clave}) partida = tipo_de_partida(extendida) for tipo in dict.fromkeys((partida, extendida.tipo_sugerido, *bitcoin.TIPOS)): muestra = bitcoin.direcciones(extendida, tipo, 0, 0, 5) if any(cliente.historiales([d["huella"] for d in muestra])): return tipo, True return partida, False def detectar_tipo(cliente: Cliente, clave: str) -> str: return detectar(cliente, clave)[0] # ── Fechas de los bloques ──────────────────────────────────────────────────── def _fechas(cliente: Cliente, alturas: list[int]) -> dict[int, str]: utiles = sorted({a for a in alturas if a > 0}) if not utiles: return {} fechas = {} for altura, cabecera in zip(utiles, cliente.cabeceras(utiles)): try: momento = bitcoin.momento_de_cabecera(cabecera) fechas[altura] = dt.datetime.fromtimestamp(momento, dt.timezone.utc).date().isoformat() except (ValueError, TypeError, OSError): continue return fechas def probar(config: dict, servidor: dict) -> dict: """Comprueba que se llega al servidor y dice qué versión y qué altura tiene.""" s = con_su_clave(normalizar_servidor(servidor), config.get("servidores") or []) cliente, _, _ = conectar([s], proxy_de(config)) try: return {"servidor": sin_claves([s])[0], "version": cliente.version, "altura": cliente.altura(), "podado": bool(getattr(cliente, "podado", False))} finally: cliente.cerrar()