diff --git a/lnbits/core/views/blockexplorer_api.py b/lnbits/core/views/blockexplorer_api.py index 9d033514f..dbadfe9b4 100644 --- a/lnbits/core/views/blockexplorer_api.py +++ b/lnbits/core/views/blockexplorer_api.py @@ -22,6 +22,7 @@ from lnbits.utils.electrum import ( ElectrumError, FeeResponse, Transaction, + network_from_name, parse_block_header, parse_raw_tx, scripthash_from_address, @@ -52,7 +53,10 @@ async def _check_api_access( def _client() -> ElectrumClient: - return ElectrumClient(settings.lnbits_blockexplorer_electrum_url) + return ElectrumClient( + settings.lnbits_blockexplorer_electrum_url, + network=network_from_name(settings.lnbits_blockexplorer_network), + ) # ---- REST ---- @@ -110,7 +114,7 @@ async def api_tx(txid: str) -> Transaction: try: async with _client() as c: raw_hex = await c.get_transaction(txid) - return parse_raw_tx(raw_hex) + return parse_raw_tx(raw_hex, network=c.network) except ElectrumError as e: raise HTTPException(HTTPStatus.SERVICE_UNAVAILABLE, detail=str(e)) from e diff --git a/lnbits/settings.py b/lnbits/settings.py index 62ca61a1f..896f25edb 100644 --- a/lnbits/settings.py +++ b/lnbits/settings.py @@ -851,6 +851,8 @@ class BlockExplorerSettings(LNbitsSettings): lnbits_blockexplorer_electrum_url: str = Field( default="ssl://electrum.blockstream.info:50002" ) + # one of: main, test, regtest, signet (see embit.networks.NETWORKS) + lnbits_blockexplorer_network: str = Field(default="main") class AuthMethods(Enum): diff --git a/lnbits/static/i18n/en.js b/lnbits/static/i18n/en.js index b91837529..dba74f331 100644 --- a/lnbits/static/i18n/en.js +++ b/lnbits/static/i18n/en.js @@ -852,6 +852,9 @@ window.localisation.en = { electrum_server_url: 'Electrum Server URL', electrum_server_url_hint: 'e.g. ssl://electrum.blockstream.info:50002 or tcp://localhost:50001', + blockexplorer_network: 'Bitcoin Network', + blockexplorer_network_hint: + 'The network the Electrum server is connected to, used to render addresses correctly.', blockexplorer_search_label: 'Search by TXID or Address', blockexplorer_search_hint: '64-char hex = transaction ยท anything else = Bitcoin address', diff --git a/lnbits/templates/components/admin/blockexplorer.vue b/lnbits/templates/components/admin/blockexplorer.vue index 4f5ea1455..cd7f10e0d 100644 --- a/lnbits/templates/components/admin/blockexplorer.vue +++ b/lnbits/templates/components/admin/blockexplorer.vue @@ -55,6 +55,15 @@ :hint="$t('electrum_server_url_hint')" > +
+ +
diff --git a/lnbits/utils/electrum.py b/lnbits/utils/electrum.py index e2f7a067c..c92fcb1f0 100644 --- a/lnbits/utils/electrum.py +++ b/lnbits/utils/electrum.py @@ -16,54 +16,23 @@ from collections.abc import Callable, Coroutine from typing import Any from urllib.parse import urlparse -from bech32 import ( - CHARSET, - bech32_hrp_expand, - bech32_polymod, - convertbits, -) -from bech32 import encode as bech32_segwit_encode +from embit.networks import NETWORKS +from embit.script import Script +from embit.transaction import Transaction as EmbitTransaction from loguru import logger from pydantic import BaseModel -_BECH32M_CONST = 0x2BC830A3 # BIP-350 +DEFAULT_NETWORK = NETWORKS["main"] -def _segwit_addr_decode(address: str) -> tuple[int, bytes]: - """Decode a segwit address, supporting bech32 (v0) and bech32m (v1+).""" - lower = address.lower() - pos = lower.rfind("1") - if ( - pos < 1 - or pos + 7 > len(lower) - or not all(c in CHARSET for c in lower[pos + 1 :]) - ): - raise ValueError(f"Invalid bech32 address: {address!r}") - hrp = lower[:pos] - data = [CHARSET.find(c) for c in lower[pos + 1 :]] - const = bech32_polymod(bech32_hrp_expand(hrp) + data) - if const not in (1, _BECH32M_CONST): - raise ValueError(f"Invalid bech32 address: {address!r}") - payload = data[:-6] - witness_version = payload[0] - bits = convertbits(payload[1:], 5, 8, False) - if bits is None: - raise ValueError(f"Invalid bech32 witness program in address: {address!r}") - expected = 1 if witness_version == 0 else _BECH32M_CONST - if const != expected: +def network_from_name(name: str) -> dict: + """Look up an embit network dict (see embit.networks.NETWORKS) by name.""" + try: + return NETWORKS[name] + except KeyError as exc: raise ValueError( - f"Wrong bech32 variant for witness version {witness_version}: {address!r}" - ) - return witness_version, bytes(bits) - - -def _bech32m_encode(hrp: str, witver: int, witprog: bytes) -> str: - """Encode a segwit address with bech32m checksum (witness version 1+).""" - data = [witver] + (convertbits(list(witprog), 8, 5) or []) - values = bech32_hrp_expand(hrp) + data - polymod = bech32_polymod([*values, 0, 0, 0, 0, 0, 0]) ^ _BECH32M_CONST - checksum = [(polymod >> 5 * (5 - i)) & 31 for i in range(6)] - return hrp + "1" + "".join(CHARSET[d] for d in data + checksum) + f"Unknown network {name!r}, expected one of {list(NETWORKS)}" + ) from exc class ElectrumError(Exception): @@ -75,94 +44,42 @@ def scripthash_from_scriptpubkey(scriptpubkey: bytes) -> str: return hashlib.sha256(scriptpubkey).digest()[::-1].hex() -_B58_ALPHABET = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz" - - -def _b58decode_check(s: str) -> bytes: - n = 0 - for c in s: - n = n * 58 + _B58_ALPHABET.index(c) - nz = len(s) - len(s.lstrip("1")) - buf: list[int] = [] - while n: - n, rem = divmod(n, 256) - buf.insert(0, rem) - raw = bytes([0] * nz + buf) - payload, chk = raw[:-4], raw[-4:] - expected = hashlib.sha256(hashlib.sha256(payload).digest()).digest()[:4] - if chk != expected: - raise ValueError(f"Bad base58check checksum: {s!r}") - return payload # byte 0 = version, bytes 1:21 = hash160 - - def address_to_scriptpubkey(address: str) -> bytes: """Convert a Bitcoin address (P2PKH/P2SH/P2WPKH/P2WSH/P2TR) to scriptPubKey.""" - lower = address.lower() - if lower.startswith(("bc1", "tb1", "bcrt1")): - witness_version, witness_prog = _segwit_addr_decode(address) - ver_op = 0x00 if witness_version == 0 else (0x50 + witness_version) - return bytes([ver_op, len(witness_prog)]) + witness_prog - else: - payload = _b58decode_check(address) - version, hash160 = payload[0], payload[1:] - if version in (0x00, 0x6F, 0x41): # P2PKH mainnet/testnet/regtest - return bytes([0x76, 0xA9, 0x14]) + hash160 + bytes([0x88, 0xAC]) - if version in (0x05, 0xC4, 0x3A): # P2SH mainnet/testnet/regtest - return bytes([0xA9, 0x14]) + hash160 + bytes([0x87]) - raise ValueError(f"Unknown address version byte: {version:#04x}") + try: + return Script.from_address(address).data + except Exception as exc: + raise ValueError(f"Invalid address: {address!r}") from exc def scripthash_from_address(address: str) -> str: return scripthash_from_scriptpubkey(address_to_scriptpubkey(address)) -def _b58encode_check(payload: bytes) -> str: - chk = hashlib.sha256(hashlib.sha256(payload).digest()).digest()[:4] - n = int.from_bytes(payload + chk, "big") - chars: list[str] = [] - while n: - n, rem = divmod(n, 58) - chars.insert(0, _B58_ALPHABET[rem]) - nz = len(payload) - len(payload.lstrip(b"\x00")) - return _B58_ALPHABET[0] * nz + "".join(chars) +_SCRIPT_TYPE_NAMES = { + "p2pkh": "pubkeyhash", + "p2sh": "scripthash", + "p2wpkh": "witness_v0_keyhash", + "p2wsh": "witness_v0_scripthash", + "p2tr": "witness_v1_taproot", +} -def _read_varint(data: bytes, i: int) -> tuple[int, int]: - b = data[i] - if b < 0xFD: - return b, i + 1 - if b == 0xFD: - return struct.unpack_from(" tuple[str, str | None]: +def _scriptpubkey_info(spk: bytes, network: dict) -> tuple[str, str | None]: """Return (type, address_or_None) for a scriptPubKey.""" - n = len(script) - # P2PKH - if n == 25 and script[:3] == b"\x76\xa9\x14" and script[23:] == b"\x88\xac": - return "pubkeyhash", _b58encode_check(b"\x00" + script[3:23]) - # P2SH - if n == 23 and script[0] == 0xA9 and script[1] == 0x14 and script[22] == 0x87: - return "scripthash", _b58encode_check(b"\x05" + script[2:22]) - # P2WPKH - if n == 22 and script[0] == 0x00 and script[1] == 0x14: - return "witness_v0_keyhash", bech32_segwit_encode("bc", 0, list(script[2:])) - # P2WSH - if n == 34 and script[0] == 0x00 and script[1] == 0x20: - return "witness_v0_scripthash", bech32_segwit_encode("bc", 0, list(script[2:])) - # P2TR - if n == 34 and script[0] == 0x51 and script[1] == 0x20: - return "witness_v1_taproot", _bech32m_encode("bc", 1, script[2:]) - # P2PK - if n in (35, 67) and script[-1] == 0xAC: + n = len(spk) + # P2PK (not classified by embit) + if n in (35, 67) and spk[-1] == 0xAC: return "pubkey", None - # OP_RETURN - if n >= 1 and script[0] == 0x6A: + # OP_RETURN (not classified by embit) + if n >= 1 and spk[0] == 0x6A: return "nulldata", None - return "nonstandard", None + + script = Script(spk) + script_type = script.script_type() + if script_type is None: + return "nonstandard", None + return _SCRIPT_TYPE_NAMES[script_type], script.address(network) # --------------------------------------------------------------------------- @@ -322,93 +239,56 @@ def parse_block_header(header_hex: str, height: int) -> BlockInfo: ) -def parse_raw_tx(hex_str: str) -> Transaction: +def parse_raw_tx(hex_str: str, network: dict | None = None) -> Transaction: """Parse a raw transaction hex string into a Transaction model.""" + network = network or DEFAULT_NETWORK data = bytes.fromhex(hex_str) - i = 0 + tx = EmbitTransaction.parse(data) - version = struct.unpack_from(" i + 1 and data[i] == 0x00 and data[i + 1] == 0x01 - if segwit: - i += 2 - - vin_start = i - vin_count, i = _read_varint(data, i) vin: list[TxInput] = [] - for _ in range(vin_count): - prev_txid = data[i : i + 32][::-1].hex() - i += 32 - prev_vout = struct.unpack_from(" None: parsed = urlparse(url) self.host = parsed.hostname or "" @@ -458,6 +339,7 @@ class ElectrumClient: self.client_name = client_name self.protocol_version = protocol_version self.ping_interval = ping_interval + self.network = network or DEFAULT_NETWORK self._counter = itertools.count(1) self._pending: dict[int, asyncio.Future[Any]] = {} self._subscriptions: dict[str, list[Callable[[list[Any]], Any]]] = {}