esetheyers
4b158e662d
Link another Caudal (phone, USB) with a code or a QR instead of the password, choosing the server's root certificate as a file; read Bitcoin straight from Bitcoin Core or Knots, without an Electrum server; Android app (beta) and a Windows build script. Fixed: Save did nothing for expenses, refunds and dividends entered by hand.
225 lines
11 KiB
Python
225 lines
11 KiB
Python
"""
|
|
core_falso.py — Un Bitcoin Core de mentira para las pruebas.
|
|
|
|
Contesta por HTTP a las órdenes de RPC que usa cartera/core.py, sobre la misma
|
|
cadena inventada que nodo_falso.py. Imita lo que importa de un nodo de verdad:
|
|
|
|
· solo sabe de las direcciones que se le han importado Y repasado: lo que
|
|
se importa sin repasar no enseña su pasado;
|
|
· el repaso se puede dejar «en marcha» (repaso_lento) para probar la espera;
|
|
· se puede «reiniciar» (cambia su arranque y corta el repaso a medias).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import re
|
|
import threading
|
|
import time
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
from cartera import bitcoin
|
|
from cartera.core import con_suma
|
|
|
|
USUARIO, CLAVE = "caudal", "secreto-de-rpc"
|
|
DERIVADAS = 25 # direcciones por rama que este nodo de mentira llega a seguir
|
|
_DESC = re.compile(r"^(sh\(wpkh|wpkh|pkh|tr)\(([a-zA-Z0-9]+)/([01])/\*\)+$")
|
|
_TIPO = {"wpkh": "p2wpkh", "tr": "p2tr", "sh(wpkh": "p2sh-p2wpkh", "pkh": "p2pkh"}
|
|
|
|
|
|
class ErrorRpc(Exception):
|
|
def __init__(self, codigo: int, mensaje: str):
|
|
super().__init__(mensaje)
|
|
self.codigo = codigo
|
|
|
|
|
|
class CoreFalso:
|
|
def __init__(self, cadena, podado: bool = False, con_monedero: bool = True):
|
|
self.cadena = cadena
|
|
self.podado = podado
|
|
self.con_monedero = con_monedero
|
|
self.arranque = time.time() - 5000
|
|
self.cargados: set[str] = set()
|
|
self.existentes: set[str] = set()
|
|
self.scripts: dict[str, str] = {} # script → dirección (lo importado)
|
|
self.repasados: set[str] = set() # scripts cuyo pasado ya se ha mirado
|
|
self.etiquetas: dict[str, str] = {} # dirección → etiqueta
|
|
self.repaso_lento = False
|
|
self.repasando = False
|
|
self.soltar = threading.Event()
|
|
self.ordenes: list[str] = []
|
|
self.repasos = 0
|
|
nodo = self
|
|
|
|
class Manejador(BaseHTTPRequestHandler):
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
def do_POST(self):
|
|
cuerpo = self.rfile.read(int(self.headers.get("Content-Length") or 0))
|
|
esperada = "Basic " + base64.b64encode(f"{USUARIO}:{CLAVE}".encode()).decode()
|
|
if self.headers.get("Authorization") != esperada:
|
|
self.send_response(401)
|
|
self.end_headers()
|
|
return
|
|
peticion = json.loads(cuerpo)
|
|
monedero = self.path.rsplit("/", 1)[1] if self.path.startswith("/wallet/") else None
|
|
lote = isinstance(peticion, list)
|
|
respuestas = [nodo._responder(p, monedero) for p in (peticion if lote else [peticion])]
|
|
texto = json.dumps(respuestas if lote else respuestas[0]).encode()
|
|
self.send_response(200 if lote or not respuestas[0]["error"] else 500)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(texto)))
|
|
self.end_headers()
|
|
self.wfile.write(texto)
|
|
|
|
self.http = ThreadingHTTPServer(("127.0.0.1", 0), Manejador)
|
|
self.http.daemon_threads = True
|
|
self.puerto = self.http.server_address[1]
|
|
threading.Thread(target=self.http.serve_forever, daemon=True).start()
|
|
|
|
def parar(self):
|
|
self.soltar.set()
|
|
self.http.shutdown()
|
|
self.http.server_close()
|
|
|
|
def servidor_config(self, **cambios) -> dict:
|
|
return {"nombre": "core de prueba", "host": "127.0.0.1", "puerto": self.puerto, "tls": False, "tor": False,
|
|
"clase": "core", "usuario": USUARIO, "clave": CLAVE} | cambios
|
|
|
|
def reiniciar(self):
|
|
"""Como apagar y encender el nodo: el repaso a medias se pierde."""
|
|
self.arranque = time.time()
|
|
self.repasando = False
|
|
self.soltar.set()
|
|
self.soltar = threading.Event()
|
|
|
|
# ── órdenes ──
|
|
def _responder(self, p: dict, monedero) -> dict:
|
|
self.ordenes.append(p.get("method"))
|
|
try:
|
|
return {"result": self._ejecutar(p.get("method"), p.get("params") or [], monedero),
|
|
"error": None, "id": p.get("id")}
|
|
except ErrorRpc as e:
|
|
return {"result": None, "error": {"code": e.codigo, "message": str(e)}, "id": p.get("id")}
|
|
|
|
@staticmethod
|
|
def _arg(params, nombre: str, posicion: int, defecto=None):
|
|
if isinstance(params, dict):
|
|
return params.get(nombre, defecto)
|
|
return params[posicion] if len(params) > posicion else defecto
|
|
|
|
def _mias(self) -> dict[str, dict]:
|
|
"""Las transacciones que tocan algo ya repasado."""
|
|
salida = {}
|
|
for txid, tx in self.cadena.txs.items():
|
|
tocada = any(s["script"] in self.repasados for s in tx["salidas"])
|
|
if not tocada and not tx["coinbase"]:
|
|
tocada = any(self.cadena._script_de(e["txid"], e["indice"]) in self.repasados for e in tx["entradas"])
|
|
if tocada:
|
|
salida[txid] = tx
|
|
return salida
|
|
|
|
def _ejecutar(self, metodo: str, params, monedero):
|
|
del_monedero = {"getwalletinfo", "importdescriptors", "rescanblockchain", "listlabels",
|
|
"getaddressesbylabel", "setlabel", "listsinceblock", "gettransaction"}
|
|
if metodo in del_monedero | {"listwallets", "loadwallet", "createwallet"} and not self.con_monedero:
|
|
raise ErrorRpc(-32601, "Method not found")
|
|
if metodo in del_monedero and monedero not in self.cargados:
|
|
raise ErrorRpc(-18, "Requested wallet does not exist or is not loaded")
|
|
if metodo == "getblockchaininfo":
|
|
return {"chain": "main", "blocks": self.cadena.altura(), "pruned": self.podado}
|
|
if metodo == "getnetworkinfo":
|
|
return {"subversion": "/Satoshi:31.0.0/"}
|
|
if metodo == "uptime":
|
|
return int(time.time() - self.arranque)
|
|
if metodo == "getblockhash":
|
|
return f"{int(params[0]):064x}"
|
|
if metodo == "getblockheader":
|
|
return self.cadena.cabecera(int(params[0], 16))
|
|
if metodo == "getrawtransaction":
|
|
tx = self.cadena.txs.get(params[0])
|
|
if not tx:
|
|
raise ErrorRpc(-5, "No such mempool or blockchain transaction.")
|
|
return tx["raw"]
|
|
if metodo == "listwallets":
|
|
return sorted(self.cargados)
|
|
if metodo == "loadwallet":
|
|
if params[0] not in self.existentes:
|
|
raise ErrorRpc(-18, "Wallet file verification failed. Path does not exist.")
|
|
self.cargados.add(params[0])
|
|
return {"name": params[0]}
|
|
if metodo == "createwallet":
|
|
nombre = self._arg(params, "wallet_name", 0)
|
|
if not self._arg(params, "disable_private_keys", 1) or not self._arg(params, "blank", 2):
|
|
raise ErrorRpc(-8, "la prueba exige un monedero sin claves privadas y vacío")
|
|
self.existentes.add(nombre)
|
|
self.cargados.add(nombre)
|
|
return {"name": nombre}
|
|
if metodo == "getwalletinfo":
|
|
return {"walletname": monedero, "scanning": {"duration": 3, "progress": 0.37} if self.repasando else False}
|
|
if metodo == "importdescriptors":
|
|
return [self._importar(d) for d in params[0]]
|
|
if metodo == "rescanblockchain":
|
|
return self._repasar()
|
|
if metodo == "listlabels":
|
|
return sorted(set(self.etiquetas.values()))
|
|
if metodo == "getaddressesbylabel":
|
|
encontradas = {d: {"purpose": "receive"} for d, e in self.etiquetas.items() if e == params[0]}
|
|
if not encontradas:
|
|
raise ErrorRpc(-11, "No addresses with label " + params[0])
|
|
return encontradas
|
|
if metodo == "setlabel":
|
|
if params[0] not in self.scripts.values():
|
|
raise ErrorRpc(-5, "la dirección no es de este monedero")
|
|
self.etiquetas[params[0]] = params[1]
|
|
return None
|
|
if metodo == "listsinceblock":
|
|
if isinstance(params, list) and params:
|
|
raise ErrorRpc(-8, "la prueba espera parámetros con nombre")
|
|
return {"transactions": [{"txid": t, "category": "receive", "confirmations": 1 if tx["altura"] > 0 else 0}
|
|
for t, tx in self._mias().items()], "removed": [], "lastblock": "00"}
|
|
if metodo == "gettransaction":
|
|
tx = self._mias().get(self._arg(params, "txid", 0))
|
|
if not tx:
|
|
raise ErrorRpc(-5, "Invalid or non-wallet transaction id")
|
|
confirmada = tx["altura"] > 0
|
|
return {"hex": tx["raw"], "confirmations": 1 if confirmada else 0} | (
|
|
{"blockheight": tx["altura"]} if confirmada else {})
|
|
raise ErrorRpc(-32601, "Method not found")
|
|
|
|
def _importar(self, d: dict) -> dict:
|
|
descriptor, _, suma = str(d.get("desc") or "").partition("#")
|
|
if con_suma(descriptor) != f"{descriptor}#{suma}":
|
|
return {"success": False, "error": {"code": -5, "message": "Provided checksum does not match"}}
|
|
if d.get("timestamp") != "now":
|
|
return {"success": False, "error": {"code": -8, "message": "la prueba espera importar sin repasar"}}
|
|
if descriptor.startswith("addr("):
|
|
direccion = descriptor[5:-1]
|
|
self.scripts[bitcoin.script_de_direccion(direccion)[0].hex()] = direccion
|
|
return {"success": True}
|
|
m = _DESC.match(descriptor)
|
|
if not m or not m.group(2).startswith(("xpub", "tpub")) or not d.get("range"):
|
|
return {"success": False, "error": {"code": -5, "message": "descriptor que la prueba no entiende"}}
|
|
extendida = bitcoin.ClaveExtendida(m.group(2))
|
|
desde, hasta = d["range"]
|
|
# Derivar mil direcciones en Python puro tarda: la prueba se queda con las primeras
|
|
for x in bitcoin.direcciones(extendida, _TIPO[m.group(1)], int(m.group(3)), desde, min(hasta - desde + 1, DERIVADAS)):
|
|
self.scripts[x["script"]] = x["direccion"]
|
|
return {"success": True}
|
|
|
|
def _repasar(self) -> dict:
|
|
if self.podado:
|
|
raise ErrorRpc(-1, "Can't rescan beyond pruned data.")
|
|
self.repasos += 1
|
|
pendientes = set(self.scripts)
|
|
if self.repaso_lento:
|
|
self.repasando = True
|
|
soltar = self.soltar
|
|
soltar.wait(30)
|
|
if soltar is not self.soltar: # el nodo se «reinició» a mitad
|
|
raise ErrorRpc(-1, "Rescan aborted.")
|
|
self.repasando = False
|
|
self.repasados |= pendientes
|
|
return {"start_height": 0, "stop_height": self.cadena.altura()}
|