""" bitcoin.py — Lo mínimo de Bitcoin para mirar tus monederos: derivar las direcciones de una clave pública extendida (xpub / ypub / zpub) y convertirlas en la «huella de script» que entiende un servidor Electrum. Aquí NO hay claves privadas ni se firma nada: solo se calculan direcciones a partir de una clave pública, que es justo lo que hace un monedero de solo lectura. Todo con la biblioteca estándar de Python (sin instalar nada): • RIPEMD-160 propio, porque las versiones nuevas de OpenSSL ya no lo traen. • Aritmética de la curva secp256k1 (sumar puntos) para derivar hijos. • Codificación Base58Check (direcciones 1… y 3…) y Bech32/Bech32m (bc1…). Referencias: BIP32 (derivación), BIP44/49/84/86 (tipos de dirección), BIP173 y BIP350 (bech32 y bech32m). """ from __future__ import annotations import hashlib import hmac # ── secp256k1 ──────────────────────────────────────────────────────────────── P = 2**256 - 2**32 - 977 N = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEBAAEDCE6AF48A03BBFD25E8CD0364141 GX = 0x79BE667EF9DCBBAC55A06295CE870B07029BFCDB2DCE28D959F2815B16F81798 GY = 0x483ADA7726A3C4655DA4FBFC0E1108A8FD17B448A68554199C47D08FFB10D4B8 G = (GX, GY) Punto = tuple[int, int] | None def _suma(p1: Punto, p2: Punto) -> Punto: """Suma de dos puntos de la curva (None es el punto del infinito).""" if p1 is None: return p2 if p2 is None: return p1 x1, y1 = p1 x2, y2 = p2 if x1 == x2 and (y1 + y2) % P == 0: return None if p1 == p2: lam = 3 * x1 * x1 * pow(2 * y1, P - 2, P) % P else: lam = (y2 - y1) * pow(x2 - x1, P - 2, P) % P x3 = (lam * lam - x1 - x2) % P return (x3, (lam * (x1 - x3) - y1) % P) def _jac_doble(punto): x, y, z = punto if y == 0: return (0, 0, 0) yy = y * y % P s = 4 * x * yy % P m = 3 * x * x % P x3 = (m * m - 2 * s) % P return (x3, (m * (s - x3) - 8 * yy * yy) % P, 2 * y * z % P) def _jac_mas_afin(punto, q: Punto): x1, y1, z1 = punto if z1 == 0: return (q[0], q[1], 1) zz = z1 * z1 % P u2 = q[0] * zz % P s2 = q[1] * z1 * zz % P if u2 == x1: return _jac_doble(punto) if s2 == y1 else (0, 0, 0) h = (u2 - x1) % P r = (s2 - y1) % P hh = h * h % P hhh = h * hh % P v = x1 * hh % P x3 = (r * r - hhh - 2 * v) % P return (x3, (r * (v - x3) - y1 * hhh) % P, z1 * h % P) def _afin(punto) -> Punto: x, y, z = punto if z == 0: return None inv = pow(z, P - 2, P) return (x * inv * inv % P, y * pow(inv, 3, P) % P) _TABLA = None def _tabla_g() -> list: """G, 2G, 4G, 8G… para no tener que duplicar en cada multiplicación.""" global _TABLA if _TABLA is None: tabla, punto = [], G for _ in range(256): tabla.append(punto) punto = _afin(_jac_doble((punto[0], punto[1], 1))) _TABLA = tabla return _TABLA def _multiplica(k: int, p: Punto = G) -> Punto: if p == G: tabla = _tabla_g() r = (0, 0, 0) i = 0 while k: if k & 1: r = _jac_mas_afin(r, tabla[i]) k >>= 1 i += 1 return _afin(r) r = (0, 0, 0) while k: if k & 1: r = _jac_mas_afin(r, p) p = _afin(_jac_doble((p[0], p[1], 1))) k >>= 1 return _afin(r) def _descomprimir(clave: bytes) -> Punto: """De los 33 bytes comprimidos (02/03 + x) al punto (x, y).""" if len(clave) != 33 or clave[0] not in (2, 3): raise ValueError("clave pública comprimida no válida") x = int.from_bytes(clave[1:], "big") y2 = (pow(x, 3, P) + 7) % P y = pow(y2, (P + 1) // 4, P) if pow(y, 2, P) != y2: raise ValueError("clave pública fuera de la curva") if y % 2 != clave[0] % 2: y = P - y return (x, y) def _comprimir(p: Punto) -> bytes: if p is None: raise ValueError("punto del infinito") x, y = p return bytes([2 + (y & 1)]) + x.to_bytes(32, "big") # ── RIPEMD-160 ─────────────────────────────────────────────────────────────── # OpenSSL 3 lo desactivó, así que se calcula aquí. Implementación directa del # algoritmo (dos ramas de 80 pasos); los valores de las tablas son los del # estándar y las pruebas los comprueban contra vectores conocidos. _ML = (0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 7, 4, 13, 1, 10, 6, 15, 3, 12, 0, 9, 5, 2, 14, 11, 8, 3, 10, 14, 4, 9, 15, 8, 1, 2, 7, 0, 6, 13, 11, 5, 12, 1, 9, 11, 10, 0, 8, 12, 4, 13, 3, 7, 15, 14, 5, 6, 2, 4, 0, 5, 9, 7, 12, 2, 10, 14, 1, 3, 8, 11, 6, 15, 13) _MR = (5, 14, 7, 0, 9, 2, 11, 4, 13, 6, 15, 8, 1, 10, 3, 12, 6, 11, 3, 7, 0, 13, 5, 10, 14, 15, 8, 12, 4, 9, 1, 2, 15, 5, 1, 3, 7, 14, 6, 9, 11, 8, 12, 2, 10, 0, 4, 13, 8, 6, 4, 1, 3, 11, 15, 0, 5, 12, 2, 13, 9, 7, 10, 14, 12, 15, 10, 4, 1, 5, 8, 7, 6, 2, 13, 14, 0, 3, 9, 11) _RL = (11, 14, 15, 12, 5, 8, 7, 9, 11, 13, 14, 15, 6, 7, 9, 8, 7, 6, 8, 13, 11, 9, 7, 15, 7, 12, 15, 9, 11, 7, 13, 12, 11, 13, 6, 7, 14, 9, 13, 15, 14, 8, 13, 6, 5, 12, 7, 5, 11, 12, 14, 15, 14, 15, 9, 8, 9, 14, 5, 6, 8, 6, 5, 12, 9, 15, 5, 11, 6, 8, 13, 12, 5, 12, 13, 14, 11, 8, 5, 6) _RR = (8, 9, 9, 11, 13, 15, 15, 5, 7, 7, 8, 11, 14, 14, 12, 6, 9, 13, 15, 7, 12, 8, 9, 11, 7, 7, 12, 7, 6, 15, 13, 11, 9, 7, 15, 11, 8, 6, 6, 14, 12, 13, 5, 14, 13, 13, 7, 5, 15, 5, 8, 11, 14, 14, 6, 14, 6, 9, 12, 9, 12, 5, 15, 8, 8, 5, 12, 9, 12, 5, 14, 6, 8, 13, 6, 5, 15, 13, 11, 11) _KL = (0, 0x5A827999, 0x6ED9EBA1, 0x8F1BBCDC, 0xA953FD4E) _KR = (0x50A28BE6, 0x5C4DD124, 0x6D703EF3, 0x7A6D76E9, 0) def _f(ronda: int, x: int, y: int, z: int) -> int: if ronda == 0: return x ^ y ^ z if ronda == 1: return (x & y) | (~x & z) if ronda == 2: return (x | ~y) ^ z if ronda == 3: return (x & z) | (y & ~z) return x ^ (y | ~z) def _rol(x: int, i: int) -> int: return ((x << i) | (x >> (32 - i))) & 0xFFFFFFFF def _ripemd160_bloque(estado: tuple, bloque: bytes) -> tuple: x = [int.from_bytes(bloque[4 * i:4 * i + 4], "little") for i in range(16)] izq = der = estado for j in range(80): ronda = j >> 4 a, b, c, d, e = izq a = (_rol((a + _f(ronda, b, c, d) + x[_ML[j]] + _KL[ronda]) & 0xFFFFFFFF, _RL[j]) + e) & 0xFFFFFFFF izq = (e, a, b, _rol(c, 10), d) a, b, c, d, e = der a = (_rol((a + _f(4 - ronda, b, c, d) + x[_MR[j]] + _KR[ronda]) & 0xFFFFFFFF, _RR[j]) + e) & 0xFFFFFFFF der = (e, a, b, _rol(c, 10), d) h = estado return ((h[1] + izq[2] + der[3]) & 0xFFFFFFFF, (h[2] + izq[3] + der[4]) & 0xFFFFFFFF, (h[3] + izq[4] + der[0]) & 0xFFFFFFFF, (h[4] + izq[0] + der[1]) & 0xFFFFFFFF, (h[0] + izq[1] + der[2]) & 0xFFFFFFFF) def ripemd160(datos: bytes) -> bytes: try: # si OpenSSL lo trae, mejor (más rápido) return hashlib.new("ripemd160", datos).digest() except (ValueError, TypeError): return _ripemd160_puro(datos) def _ripemd160_puro(datos: bytes) -> bytes: estado = (0x67452301, 0xEFCDAB89, 0x98BADCFE, 0x10325476, 0xC3D2E1F0) relleno = datos + b"\x80" relleno += b"\x00" * ((56 - len(relleno) % 64) % 64) relleno += (len(datos) * 8).to_bytes(8, "little") for i in range(0, len(relleno), 64): estado = _ripemd160_bloque(estado, relleno[i:i + 64]) return b"".join(h.to_bytes(4, "little") for h in estado) def sha256(datos: bytes) -> bytes: return hashlib.sha256(datos).digest() def hash160(datos: bytes) -> bytes: return ripemd160(sha256(datos)) def _etiquetado(etiqueta: str, datos: bytes) -> bytes: """Hash con etiqueta de BIP340, usado por taproot.""" t = sha256(etiqueta.encode()) return sha256(t + t + datos) # ── Base58Check ────────────────────────────────────────────────────────────── _ALFABETO = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz" def b58_codificar(datos: bytes) -> str: n = int.from_bytes(datos, "big") salida = "" while n: n, resto = divmod(n, 58) salida = _ALFABETO[resto] + salida return "1" * (len(datos) - len(datos.lstrip(b"\x00"))) + salida def b58_descodificar(texto: str) -> bytes: n = 0 for c in texto: if c not in _ALFABETO: raise ValueError(f"carácter no válido en Base58: {c!r}") n = n * 58 + _ALFABETO.index(c) cuerpo = n.to_bytes((n.bit_length() + 7) // 8, "big") return b"\x00" * (len(texto) - len(texto.lstrip("1"))) + cuerpo def b58check_codificar(datos: bytes) -> str: return b58_codificar(datos + sha256(sha256(datos))[:4]) def b58check_descodificar(texto: str) -> bytes: bruto = b58_descodificar(texto) if len(bruto) < 5: raise ValueError("clave demasiado corta") cuerpo, suma = bruto[:-4], bruto[-4:] if sha256(sha256(cuerpo))[:4] != suma: raise ValueError("la clave tiene una errata (no cuadra su dígito de control)") return cuerpo # ── Bech32 y Bech32m ───────────────────────────────────────────────────────── _BECH = "qpzry9x8gf2tvdw0s3jn54khce6mua7l" def _polymod(valores) -> int: gen = (0x3B6A57B2, 0x26508E6D, 0x1EA119FA, 0x3D4233DD, 0x2A1462B3) chk = 1 for v in valores: b = chk >> 25 chk = ((chk & 0x1FFFFFF) << 5) ^ v for i in range(5): chk ^= gen[i] if ((b >> i) & 1) else 0 return chk def _expandir(hrp: str) -> list[int]: return [ord(c) >> 5 for c in hrp] + [0] + [ord(c) & 31 for c in hrp] def _convertir(datos, de: int, a: int, relleno: bool = True) -> list[int]: acc = bits = 0 salida = [] maximo = (1 << a) - 1 for valor in datos: acc = (acc << de) | valor bits += de while bits >= a: bits -= a salida.append((acc >> bits) & maximo) if relleno and bits: salida.append((acc << (a - bits)) & maximo) return salida def bech32(hrp: str, version: int, programa: bytes) -> str: datos = [version] + _convertir(programa, 8, 5) constante = 0x2BC830A3 if version else 1 # bech32m para v1+ (BIP350) pm = _polymod(_expandir(hrp) + datos + [0, 0, 0, 0, 0, 0]) ^ constante suma = [(pm >> 5 * (5 - i)) & 31 for i in range(6)] return hrp + "1" + "".join(_BECH[d] for d in datos + suma) def bech32_descodificar(texto: str) -> tuple[str, int, bytes]: """(hrp, versión, programa) de una dirección bc1…/tb1…; ValueError si está mal.""" if texto.lower() != texto and texto.upper() != texto: raise ValueError("mezcla mayúsculas y minúsculas") texto = texto.lower() corte = texto.rfind("1") if corte < 1 or corte + 7 > len(texto) or len(texto) > 90: raise ValueError("no tiene forma de dirección bech32") hrp, resto = texto[:corte], texto[corte + 1:] if any(c not in _BECH for c in resto): raise ValueError("tiene caracteres que no caben en una dirección bc1") datos = [_BECH.index(c) for c in resto] pm = _polymod(_expandir(hrp) + datos) version = datos[0] if pm != (0x2BC830A3 if version else 1): raise ValueError("tiene una errata (no cuadra su dígito de control)") programa = bytes(_convertir(datos[1:-6], 5, 8, relleno=False)) if version > 16 or not 2 <= len(programa) <= 40 or (version == 0 and len(programa) not in (20, 32)): raise ValueError("programa de testigo no válido") return hrp, version, programa def script_de_direccion(direccion: str) -> tuple[bytes, str]: """(script de salida, red) de una dirección cualquiera: 1…, 3…, bc1q…, bc1p… (y sus equivalentes de testnet). Sirve para vigilar direcciones sueltas de un monedero que no enseña su clave extendida.""" d = (direccion or "").strip() if not d: raise ValueError("dirección vacía") if d.lower().startswith(("bc1", "tb1", "bcrt1")): hrp, version, programa = bech32_descodificar(d) red = {"bc": "mainnet", "tb": "testnet"}.get(hrp) if red is None: raise ValueError("no es una dirección de bitcoin") op = 0 if version == 0 else 0x50 + version return bytes([op, len(programa)]) + programa, red try: cuerpo = b58check_descodificar(d) except ValueError: raise ValueError("tiene una errata o no es una dirección de bitcoin") from None if len(cuerpo) != 21: raise ValueError("no es una dirección de bitcoin") prefijo, h = cuerpo[0], cuerpo[1:] if prefijo in (0x00, 0x6F): return b"\x76\xa9\x14" + h + b"\x88\xac", "mainnet" if prefijo == 0x00 else "testnet" if prefijo in (0x05, 0xC4): return b"\xa9\x14" + h + b"\x87", "mainnet" if prefijo == 0x05 else "testnet" raise ValueError("no es una dirección de bitcoin") # ── Claves públicas extendidas ─────────────────────────────────────────────── # versión de la clave → (tipo de dirección que suele acompañarla, red) PRIVADAS = {0x0488ADE4, 0x049D7878, 0x04B2430C, 0x04358394, 0x044A4E28, 0x045F18BC} VERSIONES = { 0x0488B21E: ("p2pkh", "mainnet"), # xpub 0x049D7CB2: ("p2sh-p2wpkh", "mainnet"), # ypub 0x04B24746: ("p2wpkh", "mainnet"), # zpub 0x043587CF: ("p2pkh", "testnet"), # tpub 0x044A5262: ("p2sh-p2wpkh", "testnet"), # upub 0x045F1CF6: ("p2wpkh", "testnet"), # vpub } TIPOS = ("p2wpkh", "p2tr", "p2sh-p2wpkh", "p2pkh") NOMBRES_TIPO = { "p2wpkh": "SegWit nativo (bc1q…)", "p2tr": "Taproot (bc1p…)", "p2sh-p2wpkh": "SegWit compatible (3…)", "p2pkh": "Antiguo (1…)", } class ClaveExtendida: """Una clave pública de cuenta (xpub/ypub/zpub) lista para derivar hijos.""" __slots__ = ("clave", "cadena", "tipo_sugerido", "red", "profundidad") def __init__(self, texto: str): bruto = b58check_descodificar(texto.strip()) if len(bruto) != 78: raise ValueError("no parece una clave pública extendida") version = int.from_bytes(bruto[:4], "big") if version in PRIVADAS or (len(bruto) == 78 and bruto[45] == 0): raise ValueError("esto es una clave PRIVADA: pon la pública (xpub, ypub o zpub)") if version not in VERSIONES: raise ValueError("clave extendida de un tipo que no conozco") if bruto[45] not in (2, 3): raise ValueError("esto es una clave PRIVADA: pon la pública (xpub, ypub o zpub)") self.tipo_sugerido, self.red = VERSIONES[version] self.profundidad = bruto[4] self.cadena = bruto[13:45] self.clave = bruto[45:78] def hijo(self, indice: int) -> tuple[bytes, bytes]: """Derivación no endurecida (BIP32): devuelve (clave, cadena) del hijo.""" if not 0 <= indice < 0x80000000: raise ValueError("índice fuera de rango") return _hijo(self.clave, self.cadena, indice) def rama(self, cambio: int, indice: int) -> bytes: """Clave pública de .../cambio/indice (0 = recibir, 1 = cambio).""" clave, cadena = _hijo(self.clave, self.cadena, cambio) return _hijo(clave, cadena, indice)[0] def _hijo(clave: bytes, cadena: bytes, indice: int) -> tuple[bytes, bytes]: i = hmac.new(cadena, clave + indice.to_bytes(4, "big"), hashlib.sha512).digest() desplazamiento = int.from_bytes(i[:32], "big") if desplazamiento >= N: raise ValueError("índice imposible de derivar (probabilidad ínfima)") punto = _suma(_multiplica(desplazamiento), _descomprimir(clave)) if punto is None: raise ValueError("índice imposible de derivar (probabilidad ínfima)") return _comprimir(punto), i[32:] # ── De clave pública a script y dirección ──────────────────────────────────── def script_de(clave: bytes, tipo: str, red: str = "mainnet") -> bytes: """scriptPubKey (lo que mira el nodo) para una clave pública.""" if tipo == "p2wpkh": return b"\x00\x14" + hash160(clave) if tipo == "p2pkh": return b"\x76\xa9\x14" + hash160(clave) + b"\x88\xac" if tipo == "p2sh-p2wpkh": return b"\xa9\x14" + hash160(b"\x00\x14" + hash160(clave)) + b"\x87" if tipo == "p2tr": return b"\x51\x20" + _taproot(clave) raise ValueError(f"tipo de dirección desconocido: {tipo}") def _taproot(clave: bytes) -> bytes: """Clave de salida de taproot: la interna «ajustada» sin árbol de scripts (BIP86).""" interna = clave[1:] ajuste = int.from_bytes(_etiquetado("TapTweak", interna), "big") if ajuste >= N: raise ValueError("ajuste de taproot no válido") punto = _suma(_descomprimir(b"\x02" + interna), _multiplica(ajuste)) if punto is None: raise ValueError("ajuste de taproot no válido") return punto[0].to_bytes(32, "big") def direccion_de(clave: bytes, tipo: str, red: str = "mainnet") -> str: """Dirección legible, solo para enseñarla en pantalla.""" hrp = "bc" if red == "mainnet" else "tb" if tipo == "p2wpkh": return bech32(hrp, 0, hash160(clave)) if tipo == "p2tr": return bech32(hrp, 1, _taproot(clave)) prefijo = {"mainnet": {"p2pkh": 0x00, "p2sh-p2wpkh": 0x05}, "testnet": {"p2pkh": 0x6F, "p2sh-p2wpkh": 0xC4}}[red][tipo] cuerpo = hash160(clave) if tipo == "p2pkh" else hash160(b"\x00\x14" + hash160(clave)) return b58check_codificar(bytes([prefijo]) + cuerpo) def huella_script(script: bytes) -> str: """La «scripthash» de Electrum: sha256 del script, con los bytes al revés.""" return sha256(script)[::-1].hex() def direcciones(clave: ClaveExtendida, tipo: str, cambio: int, desde: int, cuantas: int) -> list[dict]: """Un tramo de direcciones de una rama, con su script y su huella.""" padre, cadena = _hijo(clave.clave, clave.cadena, cambio) salida = [] for i in range(desde, desde + cuantas): pub = _hijo(padre, cadena, i)[0] script = script_de(pub, tipo, clave.red) salida.append({"cambio": cambio, "indice": i, "script": script.hex(), "huella": huella_script(script), "direccion": direccion_de(pub, tipo, clave.red)}) return salida # ── Lectura de transacciones en crudo ──────────────────────────────────────── # Se pide la transacción en hexadecimal y se lee aquí, en vez de fiarlo al modo # «verbose» del servidor, que no todos los electrs ofrecen. def _varint(datos: bytes, i: int) -> tuple[int, int]: n = datos[i] if n < 0xFD: return n, i + 1 tamano = {0xFD: 2, 0xFE: 4, 0xFF: 8}[n] return int.from_bytes(datos[i + 1:i + 1 + tamano], "little"), i + 1 + tamano def leer_transaccion(hexadecimal: str) -> dict: """De la transacción en hexadecimal a {txid, entradas, salidas, coinbase}.""" datos = bytes.fromhex(hexadecimal) i = 4 # versión sin_testigos = bytearray(datos[:4]) testigos = len(datos) > 5 and datos[4] == 0x00 and datos[5] == 0x01 if testigos: i += 2 inicio = i cuantas, i = _varint(datos, i) entradas = [] for _ in range(cuantas): anterior = datos[i:i + 32][::-1].hex() indice = int.from_bytes(datos[i + 32:i + 36], "little") i += 36 largo, i = _varint(datos, i) i += largo + 4 # script de desbloqueo + secuencia entradas.append({"txid": anterior, "indice": indice}) cuantas, i = _varint(datos, i) salidas = [] for n in range(cuantas): valor = int.from_bytes(datos[i:i + 8], "little") i += 8 largo, i = _varint(datos, i) salidas.append({"indice": n, "satoshis": valor, "script": datos[i:i + largo].hex()}) i += largo sin_testigos += datos[inicio:i] if testigos: for _ in range(len(entradas)): piezas, i = _varint(datos, i) for _ in range(piezas): largo, i = _varint(datos, i) i += largo sin_testigos += datos[i:i + 4] # bloqueo temporal coinbase = len(entradas) == 1 and entradas[0]["txid"] == "00" * 32 return {"txid": sha256(sha256(bytes(sin_testigos)))[::-1].hex(), "entradas": entradas, "salidas": salidas, "coinbase": coinbase} def momento_de_cabecera(hexadecimal: str) -> int: """Marca de tiempo (segundos) de una cabecera de bloque de 80 bytes.""" datos = bytes.fromhex(hexadecimal) if len(datos) < 80: raise ValueError("cabecera de bloque incompleta") return int.from_bytes(datos[68:72], "little")