227 lines
9.8 KiB
Python
227 lines
9.8 KiB
Python
"""
|
|
electrum.py — Hablar con tu servidor Electrum (electrs o Fulcrum), por Tor si
|
|
hace falta.
|
|
|
|
El protocolo es sencillo: una conexión TCP por la que van y vienen mensajes
|
|
JSON, uno por línea. Aquí solo se piden cosas de solo lectura (saldos,
|
|
historial, transacciones) y nunca se envía nada tuyo: el servidor ve las
|
|
«huellas de script» que se consultan, por eso conviene que el servidor sea el
|
|
tuyo.
|
|
|
|
Para llegar a una dirección .onion se usa el SOCKS5 de Tor (127.0.0.1:9050),
|
|
implementado aquí mismo: no hace falta instalar nada.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import socket
|
|
import ssl
|
|
import threading
|
|
|
|
TIEMPO_ESPERA = 45 # segundos; por Tor la primera respuesta tarda
|
|
MAX_LINEA = 8_000_000 # una transacción grande en hexadecimal cabe de sobra
|
|
LOTE = 25 # peticiones por envío
|
|
|
|
|
|
# Cuando Caudal va en un servidor, quien se conecta al nodo es ese servidor y no el
|
|
# dispositivo desde el que se mira: lo pone servidor.arrancar y cambia lo que se explica
|
|
EN_SERVIDOR = False
|
|
AVISO_SERVIDOR = ("Recuerda que la conexión con el nodo la hace el servidor donde está Caudal, no el dispositivo "
|
|
"desde el que lo estás mirando: tener Tor o una VPN en tu móvil o en tu ordenador no cambia nada. "
|
|
"Añade una dirección a la que llegue el servidor (la de tu red local o la de su misma máquina).")
|
|
|
|
|
|
class ErrorElectrum(RuntimeError):
|
|
"""No se ha podido hablar con el servidor (o ha contestado un error)."""
|
|
|
|
|
|
def _socks5(destino: str, puerto: int, proxy_host: str, proxy_puerto: int, espera: float) -> socket.socket:
|
|
"""Conexión a través del SOCKS5 de Tor, hablando el protocolo a mano."""
|
|
try:
|
|
s = socket.create_connection((proxy_host, proxy_puerto), timeout=espera)
|
|
except OSError as e:
|
|
raise ErrorElectrum(f"no encuentro Tor en {proxy_host}:{proxy_puerto} ({e}). "
|
|
+ ("El servidor donde va Caudal no tiene Tor." if EN_SERVIDOR
|
|
else "¿Está Tor arrancado en este equipo?")) from None
|
|
s.settimeout(espera)
|
|
try:
|
|
s.sendall(b"\x05\x01\x00") # versión 5, sin autenticación
|
|
if _recibir(s, 2) != b"\x05\x00":
|
|
raise ErrorElectrum("Tor no acepta la conexión (SOCKS5)")
|
|
nombre = destino.encode("idna" if not destino.endswith(".onion") else "ascii")
|
|
if len(nombre) > 255:
|
|
raise ErrorElectrum("la dirección del servidor es demasiado larga")
|
|
s.sendall(b"\x05\x01\x00\x03" + bytes([len(nombre)]) + nombre + puerto.to_bytes(2, "big"))
|
|
respuesta = _recibir(s, 4)
|
|
if respuesta[1] != 0:
|
|
motivos = {1: "fallo general", 2: "no permitido", 3: "red inalcanzable",
|
|
4: "destino inalcanzable", 5: "conexión rechazada", 6: "expiró el plazo"}
|
|
raise ErrorElectrum(f"Tor no ha podido conectar: {motivos.get(respuesta[1], respuesta[1])}")
|
|
tipo = respuesta[3] # dirección que devuelve Tor
|
|
largo = {1: 4, 4: 16}.get(tipo) or _recibir(s, 1)[0]
|
|
_recibir(s, largo + 2) # dirección + puerto, que no usamos
|
|
return s
|
|
except ErrorElectrum:
|
|
s.close()
|
|
raise
|
|
except OSError as e:
|
|
s.close()
|
|
raise ErrorElectrum(f"error hablando con Tor: {e}") from None
|
|
|
|
|
|
def _recibir(s: socket.socket, cuantos: int) -> bytes:
|
|
trozos = b""
|
|
while len(trozos) < cuantos:
|
|
trozo = s.recv(cuantos - len(trozos))
|
|
if not trozo:
|
|
raise ErrorElectrum("la conexión se ha cortado")
|
|
trozos += trozo
|
|
return trozos
|
|
|
|
|
|
class Cliente:
|
|
"""Una conexión abierta a un servidor Electrum."""
|
|
|
|
def __init__(self, host: str, puerto: int, tls: bool = False, tor: bool | None = None,
|
|
proxy: tuple[str, int] = ("127.0.0.1", 9050), espera: float = TIEMPO_ESPERA):
|
|
self.host, self.puerto = host, puerto
|
|
self.tor = host.endswith(".onion") if tor is None else tor
|
|
self._lock = threading.Lock()
|
|
self._id = 0
|
|
self._resto = b""
|
|
if self.tor:
|
|
sock = _socks5(host, puerto, proxy[0], proxy[1], espera)
|
|
else:
|
|
try:
|
|
sock = socket.create_connection((host, puerto), timeout=espera)
|
|
except OSError as e:
|
|
raise ErrorElectrum(f"no contesta {host}:{puerto} ({e})") from None
|
|
if tls:
|
|
# Los servidores caseros usan certificados suyos: por eso no se
|
|
# comprueba la firma. Por .onion la propia Tor garantiza el destino.
|
|
contexto = ssl._create_unverified_context()
|
|
try:
|
|
sock = contexto.wrap_socket(sock, server_hostname=None if self.tor else host)
|
|
except OSError as e:
|
|
sock.close()
|
|
raise ErrorElectrum(f"error de cifrado con {host}:{puerto} ({e})") from None
|
|
sock.settimeout(espera)
|
|
self.sock = sock
|
|
self.version = self._saludo()
|
|
|
|
def _saludo(self) -> str:
|
|
r = self.llamada("server.version", ["Cartera", ["1.1", "1.4"]])
|
|
return r[0] if isinstance(r, list) and r else str(r)
|
|
|
|
# ── envío y recepción ──
|
|
def _linea(self) -> dict:
|
|
while b"\n" not in self._resto:
|
|
try:
|
|
trozo = self.sock.recv(65536)
|
|
except socket.timeout:
|
|
raise ErrorElectrum(f"{self.host} no responde (más de {TIEMPO_ESPERA} s)") from None
|
|
except OSError as e:
|
|
raise ErrorElectrum(f"conexión perdida con {self.host}: {e}") from None
|
|
if not trozo:
|
|
raise ErrorElectrum(f"{self.host} ha cerrado la conexión")
|
|
self._resto += trozo
|
|
if len(self._resto) > MAX_LINEA:
|
|
raise ErrorElectrum("respuesta desmesurada del servidor")
|
|
linea, self._resto = self._resto.split(b"\n", 1)
|
|
try:
|
|
return json.loads(linea)
|
|
except ValueError:
|
|
raise ErrorElectrum("el servidor ha contestado algo que no es JSON") from None
|
|
|
|
def _enviar(self, mensaje) -> None:
|
|
try:
|
|
self.sock.sendall((json.dumps(mensaje) + "\n").encode())
|
|
except OSError as e:
|
|
raise ErrorElectrum(f"no he podido escribir a {self.host}: {e}") from None
|
|
|
|
def llamada(self, metodo: str, params: list | None = None):
|
|
return self.varias([(metodo, params or [])])[0]
|
|
|
|
def varias(self, peticiones: list[tuple[str, list]]) -> list:
|
|
"""Varias peticiones de una tacada (el servidor las responde juntas)."""
|
|
if not peticiones:
|
|
return []
|
|
salida = []
|
|
with self._lock:
|
|
for i in range(0, len(peticiones), LOTE):
|
|
salida += self._tanda(peticiones[i:i + LOTE])
|
|
return salida
|
|
|
|
def _tanda(self, peticiones: list[tuple[str, list]]) -> list:
|
|
mensajes, esperados = [], {}
|
|
for metodo, params in peticiones:
|
|
self._id += 1
|
|
esperados[self._id] = len(mensajes)
|
|
mensajes.append({"jsonrpc": "2.0", "id": self._id, "method": metodo, "params": list(params)})
|
|
self._enviar(mensajes if len(mensajes) > 1 else mensajes[0])
|
|
recibidas: dict[int, object] = {}
|
|
while len(recibidas) < len(mensajes):
|
|
respuesta = self._linea()
|
|
for r in (respuesta if isinstance(respuesta, list) else [respuesta]):
|
|
if not isinstance(r, dict) or r.get("id") not in esperados:
|
|
continue # avisos y respuestas que no hemos pedido
|
|
if r.get("error"):
|
|
error = r["error"]
|
|
mensaje = error.get("message") if isinstance(error, dict) else str(error)
|
|
raise ErrorElectrum(f"el servidor rechaza la consulta: {mensaje}")
|
|
recibidas[r["id"]] = r.get("result")
|
|
return [recibidas[i] for i in sorted(esperados, key=lambda k: esperados[k])]
|
|
|
|
def cerrar(self) -> None:
|
|
try:
|
|
self.sock.close()
|
|
except OSError:
|
|
pass
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_):
|
|
self.cerrar()
|
|
|
|
# ── consultas ──
|
|
def historiales(self, huellas: list[str]) -> list[list[dict]]:
|
|
return self.varias([("blockchain.scripthash.get_history", [h]) for h in huellas])
|
|
|
|
def saldos(self, huellas: list[str]) -> list[dict]:
|
|
return self.varias([("blockchain.scripthash.get_balance", [h]) for h in huellas])
|
|
|
|
def transacciones(self, txids: list[str]) -> list[str]:
|
|
return self.varias([("blockchain.transaction.get", [t]) for t in txids])
|
|
|
|
def cabeceras(self, alturas: list[int]) -> list[str]:
|
|
return self.varias([("blockchain.block.header", [a]) for a in alturas])
|
|
|
|
def altura(self) -> int:
|
|
r = self.llamada("blockchain.headers.subscribe")
|
|
return int(r.get("height", 0)) if isinstance(r, dict) else 0
|
|
|
|
|
|
def conectar(servidores: list[dict], proxy: tuple[str, int] = ("127.0.0.1", 9050),
|
|
espera: float = TIEMPO_ESPERA) -> tuple[Cliente, dict, list[str]]:
|
|
"""Prueba los servidores en orden y se queda con el primero que responda.
|
|
|
|
Devuelve (cliente, servidor elegido, motivos por los que fallaron los otros).
|
|
"""
|
|
fallos = []
|
|
for s in servidores:
|
|
host = str(s.get("host") or "").strip()
|
|
if not host:
|
|
continue
|
|
try:
|
|
cliente = Cliente(host, int(s.get("puerto") or 50001), bool(s.get("tls")),
|
|
s.get("tor"), proxy, espera)
|
|
except ErrorElectrum as e:
|
|
fallos.append(f"{s.get('nombre') or host}: {e}")
|
|
continue
|
|
return cliente, s, fallos
|
|
if not fallos:
|
|
raise ErrorElectrum("No hay ningún servidor configurado.")
|
|
raise ErrorElectrum("No he podido conectar con ningún servidor.\n" + "\n".join(fallos)
|
|
+ (f"\n{AVISO_SERVIDOR}" if EN_SERVIDOR else ""))
|