fixes

lnbits/core/views/blockexplorer_api.py

@@ -186,15 +186,14 @@ async def ws_address(websocket: WebSocket, address: str) -> None:
     if not settings.lnbits_blockexplorer_enabled:
         await websocket.close(code=1008)
         return
-    try:
-        scripthash_from_address(address)  # validate
-    except ValueError as e:
-        await websocket.close(code=1008, reason=str(e))
-        return
     await websocket.accept()
 
     queue: asyncio.Queue[OnchainAddressEvent] = asyncio.Queue()
-    task_manager.register_ws_address_queue(address, queue)
+    try:
+        task_manager.register_ws_address_queue(address, queue)
+    except ValueError as e:
+        await websocket.close(code=1008, reason=str(e))
+        return
     try:
         await _ws_address_loop(websocket, address, queue)
     finally:

lnbits/settings.py

@@ -1092,9 +1092,6 @@ class EnvSettings(LNbitsSettings):
     lnbits_max_extensions: int = Field(default=0, ge=0)
     task_heart_beat_verbose: bool = Field(default=False)
     task_heart_beat_interval: int = Field(default=30)
-    lnbits_blockexplorer_electrum_url: str = Field(
-        default="ssl://electrum.blockstream.info:50002"
-    )
 
     @property
     def has_default_extension_path(self) -> bool:

lnbits/task_manager.py

@@ -14,6 +14,7 @@ from lnbits.utils.electrum import (
     OnchainAddressEvent,
     OnchainTxEvent,
     TransactionTracker,
+    scripthash_from_address,
 )
 
 
@@ -192,7 +193,11 @@ class TaskManager:
     def register_ws_address_queue(
         self, address: str, queue: asyncio.Queue[OnchainAddressEvent]
     ) -> None:
-        """Register a per-connection queue for a watched address."""
+        """Register a per-connection queue for a watched address.
+
+        Raises ValueError if the address is invalid.
+        """
+        scripthash_from_address(address)
         self._ws_address_queues.setdefault(address, []).append(queue)
         self.track_address(address)
 

lnbits/utils/electrum.py

@@ -16,11 +16,55 @@ from collections.abc import Callable, Coroutine
 from typing import Any
 from urllib.parse import urlparse
 
-from bech32 import bech32_decode, convertbits
+from bech32 import (
+    CHARSET,
+    bech32_hrp_expand,
+    bech32_polymod,
+    convertbits,
+)
 from bech32 import encode as bech32_segwit_encode
 from loguru import logger
 from pydantic import BaseModel
 
+_BECH32M_CONST = 0x2BC830A3  # BIP-350
+
+
+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:
+        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)
+
 
 class ElectrumError(Exception):
     pass
@@ -55,14 +99,7 @@ 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")):
-        _, data = bech32_decode(address)
-        if data is None:
-            raise ValueError(f"Invalid bech32 address: {address!r}")
-        witness_version = data[0]
-        bits = convertbits(data[1:], 5, 8, False)
-        if bits is None:
-            raise ValueError(f"Invalid bech32 witness program in address: {address!r}")
-        witness_prog = bytes(bits)
+        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:
@@ -118,7 +155,7 @@ def _scriptpubkey_info(script: bytes) -> tuple[str, str | None]:
         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", bech32_segwit_encode("bc", 1, list(script[2:]))
+        return "witness_v1_taproot", _bech32m_encode("bc", 1, script[2:])
     # P2PK
     if n in (35, 67) and script[-1] == 0xAC:
         return "pubkey", None