esetheyers/caudal
1
Fork 0
Code Issues Pull requests Projects Releases 10 Wiki Activity
caudal/programa/cartera/bitcoin.py
esetheyers 416b3b31c5 Caudal 0.1.1
2026-10-13 08:01:06 -05:00

550 lines
21 KiB
Python
Raw Permalink Blame History

"""
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")
Reference in a new issue View git blame Copy permalink