""" 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 "Hace falta Tor encendido: en el teléfono, Orbot; en el ordenador, " "el servicio Tor o el Navegador Tor abierto.")) 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.""" tope_rama = None # sin límite propio de direcciones por rama (core.Cliente sí lo tiene) 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 preparar(self, monedero: dict) -> None: """Un servidor Electrum ya lo sabe todo de cualquier dirección: nada que preparar. (A un nodo pelado sí hay que avisarle: ver core.py.)""" 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: if s.get("clase") == "core": from .core import Cliente as ClienteCore # aquí, para no importarse en círculo cliente = ClienteCore(host, int(s.get("puerto") or 8332), bool(s.get("tls")), s.get("tor"), proxy, espera, str(s.get("usuario") or ""), str(s.get("clave") or "")) else: 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 ""))