401 lines
18 KiB
Python
401 lines
18 KiB
Python
"""
|
|
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}
|
|
try:
|
|
puerto = int(s.get("puerto") or 50001)
|
|
except (TypeError, ValueError):
|
|
raise ErrorBitcoin("El puerto tiene que ser un número (50001 sin cifrar, 50002 con TLS).") from None
|
|
if not 1 <= puerto <= 65535:
|
|
raise ErrorBitcoin("El puerto está fuera de rango.")
|
|
return {"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")))}
|
|
|
|
|
|
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
|
|
while vacias < hueco and desde < MAX_DIRECCIONES:
|
|
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 >= MAX_DIRECCIONES 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."""
|
|
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")
|
|
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)
|
|
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 = normalizar_servidor(servidor)
|
|
cliente, _, _ = conectar([s], proxy_de(config))
|
|
try:
|
|
return {"servidor": s, "version": cliente.version, "altura": cliente.altura()}
|
|
finally:
|
|
cliente.cerrar()
|