""" Pruebas de la conexión directa al nodo (cartera/core.py): Bitcoin Core o Knots por RPC, sin servidor Electrum. Contra un nodo de mentira (core_falso.py) que comparte cadena con el Electrum de mentira, para comprobar que por los dos caminos se ve exactamente lo mismo. python3 -m unittest discover -s pruebas -v """ from __future__ import annotations import socket import sys import threading import time import unittest from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from cartera import btc, bitcoin, core # noqa: E402 from cartera.electrum import Cliente as ClienteElectrum, ErrorElectrum, conectar # noqa: E402 from core_falso import CoreFalso # noqa: E402 from nodo_falso import construir # noqa: E402 from test_btc import AJENA, SATOSHI, XPUB_TAPROOT, ZPUB, ConNodo, script # noqa: E402 # El aviso de repaso llega tal cual o envuelto con el nombre del monedero (historial_btc) REPASO = (core.Repasando, btc.ErrorBitcoin) class TestDescriptores(unittest.TestCase): def test_la_suma_de_control_es_la_del_nodo(self): # Los dos ejemplos de la documentación de Bitcoin Core y del BIP 380 self.assertEqual(core.con_suma("raw(deadbeef)"), "raw(deadbeef)#89f8spxm") self.assertEqual(core.con_suma( "pkh([d34db33f/44'/0'/0']xpub6ERApfZwUNrhLCkDtcHTcxd75RbzS1ed54G1LkBUHQVHQKqhMkhgbmJbZRkrgZw4koxb5JaHWkY4ALHY2grBGRjaDMzQLcgJvLJuZZvRcEL/1/*)"), "pkh([d34db33f/44'/0'/0']xpub6ERApfZwUNrhLCkDtcHTcxd75RbzS1ed54G1LkBUHQVHQKqhMkhgbmJbZRkrgZw4koxb5JaHWkY4ALHY2grBGRjaDMzQLcgJvLJuZZvRcEL/1/*)#ml40v0wf") def test_una_zpub_se_le_da_al_nodo_escrita_como_xpub(self): estandar = core.clave_estandar(ZPUB) self.assertTrue(estandar.startswith("xpub")) a, b = bitcoin.ClaveExtendida(ZPUB), bitcoin.ClaveExtendida(estandar) self.assertEqual((a.clave, a.cadena), (b.clave, b.cadena)) # la misma clave, otro prefijo def test_con_clave_se_piden_los_cuatro_tipos_y_las_dos_ramas(self): pedidos = core.requeridos({"clave": ZPUB}) self.assertEqual(len(pedidos), 8) self.assertTrue(any(p["desc"].startswith("wpkh(xpub") and "/1/*)#" in p["desc"] for p in pedidos)) self.assertTrue(any(p["desc"].startswith("sh(wpkh(xpub") for p in pedidos)) self.assertEqual(len({p["marca"] for p in pedidos}), 8) def test_con_direcciones_sueltas_una_por_direccion(self): pedidos = core.requeridos({"direcciones": ["bc1qcr8te4kr609gcawutmrza0j4xv80jy8z306fyu"]}) self.assertEqual([(p["desc"].split("#")[0], p["rango"]) for p in pedidos], [("addr(bc1qcr8te4kr609gcawutmrza0j4xv80jy8z306fyu)", None)]) class ConCore(ConNodo): """La misma cadena inventada, vista además por un Bitcoin Core de mentira.""" def setUp(self): super().setUp() self.core = CoreFalso(self.cadena) self.addCleanup(self.core.parar) def config(self, monederos, **cambios) -> dict: c = super().config(monederos) c["servidores"] = [self.core.servidor_config(**cambios)] btc.guardar(self.carpeta, c) return c def cliente(self, **cambios) -> core.Cliente: return conectar([self.core.servidor_config(**cambios)])[0] def cuando_acabe(self, tarea, plazo: float = 5.0): """Repite hasta que el nodo deja de repasar (el repaso va en otro hilo).""" limite = time.time() + plazo while True: try: return tarea() except REPASO as e: if "repasando la cadena" not in str(e) or time.time() > limite: raise time.sleep(0.02) class TestLeerDelNodo(ConCore): def test_la_primera_vez_avisa_del_repaso_y_despues_lee_como_siempre(self): self.paga(ZPUB, "p2wpkh", 0, 0, 0.5, 100) self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}]) with self.assertRaisesRegex(REPASO, "repasando la cadena") as aviso: self.comprobar() self.assertIn("repasando la cadena", str(aviso.exception)) r = self.cuando_acabe(self.comprobar) self.assertEqual(r["ultima"]["cuadres"][0]["en_nodo"], 0.5) self.assertEqual(r["ultima"]["servidor"], "core de prueba") # lo nuevo ya no necesita repaso: el nodo lo va viendo él solo self.paga(ZPUB, "p2wpkh", 0, 1, 0.25, 200) r = self.comprobar() self.assertEqual([(p["clase"], p["accion"], p["cantidad"]) for p in r["partes"]], [("entrada", "compra", 0.25)]) self.assertEqual(self.core.repasos, 1) def test_el_monedero_del_nodo_no_tiene_claves_privadas(self): self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}]) self.cuando_acabe(self.comprobar) self.assertEqual(self.core.existentes, {"caudal"}) # el nodo de mentira rechaza otra cosa self.assertNotIn("getdescriptorinfo", self.core.ordenes) self.assertLessEqual(set(self.core.ordenes), set(core.METODOS)) # lo que hay que permitir en rpcwhitelist def test_por_el_nodo_y_por_electrum_se_ve_exactamente_lo_mismo(self): a = self.paga(ZPUB, "p2wpkh", 0, 0, 0.5, 100) self.paga(ZPUB, "p2wpkh", 0, 3, 0.2, 110) # un pago a otro con su vuelta, una recompensa de minería y algo todavía sin confirmar self.cadena.añadir(construir([(a, 0)], [(10_000_000, AJENA), (39_990_000, script(ZPUB, "p2wpkh", 1, 0))]), 120) self.cadena.añadir(construir([], [(312_500_000, script(ZPUB, "p2wpkh", 0, 5))], coinbase=True), 130) self.paga(ZPUB, "p2wpkh", 0, 6, 0.01, 0) extendida = bitcoin.ClaveExtendida(ZPUB) huellas = [d["huella"] for cambio in (0, 1) for d in bitcoin.direcciones(extendida, "p2wpkh", cambio, 0, 25)] nodo = self.cliente() self.cuando_acabe(lambda: nodo.preparar({"clave": ZPUB})) with ClienteElectrum("127.0.0.1", self.nodo.puerto) as electrum: self.assertEqual(nodo.historiales(huellas), electrum.historiales(huellas)) self.assertEqual(nodo.saldos(huellas), electrum.saldos(huellas)) self.assertEqual(nodo.transacciones([a]), electrum.transacciones([a])) self.assertEqual(nodo.cabeceras([100, 130]), electrum.cabeceras([100, 130])) self.assertEqual(nodo.altura(), electrum.altura()) m = {"clave": ZPUB, "tipo": "p2wpkh"} self.assertEqual(btc.escanear(nodo, m), btc.escanear(electrum, m)) self.assertGreater(sum(s["confirmed"] for s in nodo.saldos(huellas)), 3 * SATOSHI) def test_un_solo_repaso_vale_para_adivinar_el_tipo_revisar_y_leer(self): self.paga(XPUB_TAPROOT, "p2tr", 0, 0, 0.2, 100) nodo = self.cliente() self.assertEqual(self.cuando_acabe(lambda: btc.detectar(nodo, XPUB_TAPROOT)), ("p2tr", True)) nodo = self.cliente() revision = btc.revisar(nodo, {"clave": XPUB_TAPROOT, "tipo": "p2wpkh"}) self.assertEqual(revision["tipos"]["p2tr"], {"usadas": 1, "transacciones": 1}) self.assertEqual(btc.escanear(nodo, {"clave": XPUB_TAPROOT, "tipo": "p2tr"})["sats"], 20_000_000) self.assertEqual(self.core.repasos, 1) def test_las_direcciones_sueltas_tambien(self): direccion = bitcoin.direcciones(bitcoin.ClaveExtendida(ZPUB), "p2wpkh", 0, 2, 1)[0]["direccion"] self.paga(ZPUB, "p2wpkh", 0, 2, 0.3, 100) self.config([{"nombre": "Papel", "direcciones": direccion, "activo": "C1"}]) r = self.cuando_acabe(self.comprobar) self.assertEqual(r["ultima"]["cuadres"][0]["en_nodo"], 0.3) # añadir otra dirección obliga a repasar solo por ella otra = bitcoin.direcciones(bitcoin.ClaveExtendida(ZPUB), "p2wpkh", 0, 4, 1)[0]["direccion"] self.paga(ZPUB, "p2wpkh", 0, 4, 0.1, 110) nodo = self.cliente() r = self.cuando_acabe(lambda: btc.escanear(nodo, {"direcciones": [direccion, otra]})) self.assertEqual(r["sats"], 40_000_000) self.assertEqual(self.core.repasos, 2) def test_avisa_si_el_monedero_tiene_mas_direcciones_de_las_que_sigue_el_nodo(self): tope = core.Cliente.tope_rama core.Cliente.tope_rama = 20 try: for i in range(20): self.paga(ZPUB, "p2wpkh", 0, i, 0.01, 700 + i) nodo = self.cliente() r = self.cuando_acabe(lambda: btc.escanear(nodo, {"clave": ZPUB, "tipo": "p2wpkh"})) finally: core.Cliente.tope_rama = tope self.assertTrue(r["cortada"]) def test_una_transaccion_que_no_es_del_monedero_se_le_pide_al_nodo(self): ajena = self.cadena.añadir(construir([("cc" * 32, 0)], [(5_000, AJENA)]), 90) nodo = self.cliente() self.assertEqual(nodo.transacciones([ajena]), [self.cadena.txs[ajena]["raw"]]) with self.assertRaises(ErrorElectrum) as e: nodo.transacciones(["dd" * 32]) self.assertIn("txindex", str(e.exception)) class TestRepaso(ConCore): def test_mientras_repasa_dice_por_donde_va_y_no_ensena_cifras_a_medias(self): self.core.repaso_lento = True self.paga(ZPUB, "p2wpkh", 0, 0, 0.5, 100) self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}]) with self.assertRaisesRegex(REPASO, "repasando la cadena"): self.comprobar() for _ in range(100): # hasta que el nodo se pone a ello if self.core.repasando: break time.sleep(0.02) with self.assertRaisesRegex(REPASO, "repasando la cadena") as aviso: self.comprobar() self.assertIn("37 %", str(aviso.exception)) self.assertEqual(self.core.repasos, 1) # no lanza otro repaso encima self.core.soltar.set() self.assertEqual(self.cuando_acabe(self.comprobar)["ultima"]["cuadres"][0]["en_nodo"], 0.5) def _a_medias(self, arranque: int) -> dict: """Deja el nodo como si Caudal se hubiera cerrado con el repaso en marcha.""" monedero = {"clave": ZPUB, "tipo": "p2wpkh"} pedidos = core.requeridos(monedero) nodo = self.cliente() nodo._monedero() nodo._rpc("importdescriptors", [[{"desc": p["desc"], "timestamp": "now", "range": [0, 999]} for p in pedidos]], monedero=True) nodo._etiquetar([p["marca"] for p in pedidos], f"{core.REPASANDO}{arranque}") return monedero def test_si_caudal_se_cerro_pero_el_nodo_termino_no_repasa_otra_vez(self): self.paga(ZPUB, "p2wpkh", 0, 0, 0.5, 100) monedero = self._a_medias(int(self.core.arranque)) self.core.repasados |= set(self.core.scripts) # el nodo acabó por su cuenta self.assertEqual(btc.escanear(self.cliente(), monedero)["sats"], 50_000_000) self.assertEqual(self.core.repasos, 0) self.assertEqual(set(self.core.etiquetas.values()), {core.ETIQUETA}) def test_si_el_nodo_se_reinicio_a_medias_vuelve_a_repasar(self): self.paga(ZPUB, "p2wpkh", 0, 0, 0.5, 100) monedero = self._a_medias(int(self.core.arranque)) self.core.reiniciar() # el repaso se quedó a medias: no ha visto nada self.core.arranque += 1000 with self.assertRaisesRegex(REPASO, "repasando la cadena"): btc.escanear(self.cliente(), monedero) nodo = self.cliente() self.assertEqual(self.cuando_acabe(lambda: btc.escanear(nodo, monedero))["sats"], 50_000_000) self.assertEqual(self.core.repasos, 1) def test_un_nodo_podado_lo_dice_claro(self): self.core.podado = True self.config([{"nombre": "Sparrow", "clave": ZPUB, "activo": "C1"}]) with self.assertRaisesRegex(btc.ErrorBitcoin, "podado"): self.comprobar() self.assertEqual(self.core.repasos, 0) class TestConfiguracionDelNodo(ConCore): def test_probar_dice_version_y_altura(self): self.paga(ZPUB, "p2wpkh", 0, 0, 0.1, 432) r = btc.probar(btc.cargar(self.carpeta), self.core.servidor_config()) self.assertEqual((r["version"], r["altura"], r["podado"]), ("Satoshi:31.0.0", 432, False)) self.assertNotIn("clave", r["servidor"]) def test_el_puerto_de_partida_es_el_de_rpc(self): s = btc.normalizar_servidor({"host": "minodo.local", "clase": "core", "usuario": "u", "clave": "c"}) self.assertEqual((s["puerto"], s["clase"], s["usuario"], s["clave"]), (8332, "core", "u", "c")) self.assertNotIn("clase", btc.normalizar_servidor({"host": "minodo.local"})) def test_con_la_contrasena_mala_o_sin_monedero_lo_dice_claro(self): with self.assertRaises(ErrorElectrum) as e: self.cliente(clave="otra") self.assertIn("usuario y esa contraseña", str(e.exception)) self.core.con_monedero = False with self.assertRaises(ErrorElectrum) as e: self.cliente().preparar({"clave": ZPUB}) self.assertIn("parte de monedero", str(e.exception)) def test_si_no_es_un_nodo_lo_dice(self): otro = socket.socket() # algo que escucha pero no es un nodo otro.bind(("127.0.0.1", 0)) otro.listen(1) self.addCleanup(otro.close) def contestar(): try: cliente, _ = otro.accept() cliente.recv(65536) cliente.sendall(b"HTTP/1.0 200 OK\r\nContent-Length: 4\r\n\r\nhola") cliente.close() except OSError: pass threading.Thread(target=contestar, daemon=True).start() with self.assertRaises(ErrorElectrum) as e: conectar([self.core.servidor_config(puerto=otro.getsockname()[1])], espera=2) self.assertIn("no contesta como un nodo de Bitcoin", str(e.exception)) def test_la_contrasena_de_rpc_no_sale_a_la_pantalla_y_se_conserva_al_guardar(self): self.config([]) visto = self.app.btc_estado()["servidores"] self.assertEqual([(s["clase"], s["usuario"], s["con_clave"], "clave" in s) for s in visto], [("core", "caudal", True, False)]) # la pantalla devuelve la lista como la recibió (sin contraseña), con otro nombre self.app.btc_servidores({"servidores": [{**visto[0], "nombre": "El de casa"}]}) guardado = btc.cargar(self.carpeta)["servidores"][0] self.assertEqual((guardado["nombre"], guardado["clave"]), ("El de casa", "secreto-de-rpc")) self.assertEqual(self.app.btc_probar({"servidor": visto[0]})["version"], "Satoshi:31.0.0") # si se cambia de nodo, la contraseña vieja no se le manda al nuevo self.app.btc_servidores({"servidores": [{**visto[0], "host": "otro.local"}]}) self.assertEqual(btc.cargar(self.carpeta)["servidores"][0]["clave"], "") if __name__ == "__main__": unittest.main()