caudal/programa/pruebas/nodo_falso.py

"""
nodo_falso.py — Un servidor Electrum de mentira para las pruebas.

Habla el mismo protocolo que electrs (mensajes JSON por línea, con lotes) sobre
una cadena de bloques inventada que se monta en memoria: así se puede probar
todo el camino —derivar direcciones, pedir saldos, leer transacciones y
proponer operaciones— sin tocar internet ni el nodo de verdad.
"""

from __future__ import annotations

import json
import socket
import threading

from cartera.bitcoin import huella_script, sha256

MOMENTO_BASE = 1_735_689_600        # 1 de enero de 2025


def _varint(n: int) -> bytes:
    if n < 0xFD:
        return bytes([n])
    if n <= 0xFFFF:
        return b"\xfd" + n.to_bytes(2, "little")
    return b"\xfe" + n.to_bytes(4, "little")


def construir(entradas, salidas, coinbase: bool = False, testigos: bool = False) -> str:
    """Transacción en hexadecimal. entradas: [(txid, indice)]; salidas: [(sats, script_hex)]."""
    if coinbase:
        entradas = [("00" * 32, 0xFFFFFFFF)]
    cuerpo = b"\x01\x00\x00\x00"
    if testigos:
        cuerpo += b"\x00\x01"
    cuerpo += _varint(len(entradas))
    for txid, indice in entradas:
        cuerpo += bytes.fromhex(txid)[::-1] + (indice & 0xFFFFFFFF).to_bytes(4, "little")
        cuerpo += b"\x00" + b"\xff\xff\xff\xff"
    cuerpo += _varint(len(salidas))
    for sats, script in salidas:
        script = bytes.fromhex(script)
        cuerpo += int(sats).to_bytes(8, "little") + _varint(len(script)) + script
    if testigos:
        for _ in entradas:
            cuerpo += b"\x01\x04" + b"\xde\xad\xbe\xef"      # un testigo cualquiera
    cuerpo += b"\x00\x00\x00\x00"
    return cuerpo.hex()


class Cadena:
    """Las transacciones inventadas y lo que se deduce de ellas."""

    def __init__(self):
        self.txs: dict[str, dict] = {}          # txid → {raw, altura, salidas, entradas}

    def añadir(self, hexadecimal: str, altura: int) -> str:
        from cartera.bitcoin import leer_transaccion
        tx = leer_transaccion(hexadecimal)
        self.txs[tx["txid"]] = {"raw": hexadecimal, "altura": altura, **tx}
        return tx["txid"]

    def _script_de(self, txid: str, indice: int):
        tx = self.txs.get(txid)
        if not tx:
            return None
        return next((s["script"] for s in tx["salidas"] if s["indice"] == indice), None)

    def historial(self, huella: str) -> list[dict]:
        salida = []
        for txid, tx in self.txs.items():
            tocada = any(huella_script(bytes.fromhex(s["script"])) == huella for s in tx["salidas"])
            if not tocada and not tx["coinbase"]:
                for e in tx["entradas"]:
                    script = self._script_de(e["txid"], e["indice"])
                    if script and huella_script(bytes.fromhex(script)) == huella:
                        tocada = True
                        break
            if tocada:
                salida.append({"tx_hash": txid, "height": tx["altura"]})
        return sorted(salida, key=lambda t: (t["height"], t["tx_hash"]))

    def saldo(self, huella: str) -> dict:
        gastados = {(e["txid"], e["indice"]) for tx in self.txs.values() for e in tx["entradas"]}
        confirmado = sin_confirmar = 0
        for txid, tx in self.txs.items():
            for s in tx["salidas"]:
                if huella_script(bytes.fromhex(s["script"])) != huella:
                    continue
                if (txid, s["indice"]) in gastados:
                    continue
                if tx["altura"] > 0:
                    confirmado += s["satoshis"]
                else:
                    sin_confirmar += s["satoshis"]
        return {"confirmed": confirmado, "unconfirmed": sin_confirmar}

    def cabecera(self, altura: int) -> str:
        cuerpo = bytearray(80)
        cuerpo[68:72] = (MOMENTO_BASE + altura * 600).to_bytes(4, "little")
        cuerpo[:4] = sha256(str(altura).encode())[:4]
        return bytes(cuerpo).hex()

    def altura(self) -> int:
        return max((tx["altura"] for tx in self.txs.values()), default=0)


class NodoFalso(threading.Thread):
    """Escucha en 127.0.0.1 y responde como un servidor Electrum."""

    def __init__(self, cadena: Cadena):
        super().__init__(daemon=True)
        self.cadena = cadena
        self.peticiones: list[str] = []
        self.sock = socket.socket()
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.sock.bind(("127.0.0.1", 0))
        self.sock.listen(8)
        self.puerto = self.sock.getsockname()[1]
        self._parar = False

    def servidor_config(self, nombre="prueba") -> dict:
        return {"nombre": nombre, "host": "127.0.0.1", "puerto": self.puerto, "tls": False, "tor": False}

    def run(self):
        while not self._parar:
            try:
                cliente, _ = self.sock.accept()
            except OSError:
                return
            threading.Thread(target=self._atender, args=(cliente,), daemon=True).start()

    def parar(self):
        self._parar = True
        try:
            self.sock.close()
        except OSError:
            pass

    def _atender(self, cliente: socket.socket):
        resto = b""
        with cliente:
            while True:
                try:
                    trozo = cliente.recv(65536)
                except OSError:
                    return
                if not trozo:
                    return
                resto += trozo
                while b"\n" in resto:
                    linea, resto = resto.split(b"\n", 1)
                    if not linea.strip():
                        continue
                    peticion = json.loads(linea)
                    lote = isinstance(peticion, list)
                    respuestas = [self._responder(p) for p in (peticion if lote else [peticion])]
                    cuerpo = json.dumps(respuestas if lote else respuestas[0]) + "\n"
                    try:
                        cliente.sendall(cuerpo.encode())
                    except OSError:
                        return

    def _responder(self, p: dict) -> dict:
        metodo, params = p.get("method"), p.get("params") or []
        self.peticiones.append(metodo)
        try:
            resultado = self._ejecutar(metodo, params)
        except KeyError as e:
            return {"jsonrpc": "2.0", "id": p.get("id"), "error": {"code": -32601, "message": str(e)}}
        return {"jsonrpc": "2.0", "id": p.get("id"), "result": resultado}

    def _ejecutar(self, metodo: str, params: list):
        if metodo == "server.version":
            return ["ElectrsDePrueba 0.10.0", "1.4"]
        if metodo == "blockchain.scripthash.get_history":
            return self.cadena.historial(params[0])
        if metodo == "blockchain.scripthash.get_balance":
            return self.cadena.saldo(params[0])
        if metodo == "blockchain.transaction.get":
            tx = self.cadena.txs.get(params[0])
            if not tx:
                raise KeyError("no such transaction")
            return tx["raw"]
        if metodo == "blockchain.block.header":
            return self.cadena.cabecera(int(params[0]))
        if metodo == "blockchain.headers.subscribe":
            return {"height": self.cadena.altura(), "hex": self.cadena.cabecera(self.cadena.altura())}
        raise KeyError(f"método desconocido: {metodo}")