Add BOLT11DecodeException and InvalidBitcoinURI wraps for missing uncaught exceptions.

electrum/bip21.py

@@ -7,7 +7,7 @@
 from . import bitcoin
 from .util import format_satoshis_plain
 from .bitcoin import COIN, TOTAL_COIN_SUPPLY_LIMIT_IN_BTC
-from .bolt11 import decode_bolt11_invoice, BOLT11DecodeException
+from .bolt11 import decode_bolt11_invoice, BOLT11InvoiceException
 
 # note: when checking against these, use .lower() to support case-insensitivity
 BITCOIN_BIP21_URI_SCHEME = 'bitcoin'
@@ -35,7 +35,11 @@ def parse_bip21_URI(uri: str) -> dict:
             raise InvalidBitcoinURI("Not a bitcoin address")
         return {'address': uri}
 
-    u = urllib.parse.urlparse(uri)
+    try:
+        u = urllib.parse.urlparse(uri)
+    except ValueError as e:
+        raise InvalidBitcoinURI("failed to parse uri") from e
+
     if u.scheme.lower() != BITCOIN_BIP21_URI_SCHEME:
         raise InvalidBitcoinURI("Not a bitcoin URI")
     address = u.path
@@ -94,7 +98,7 @@ def parse_bip21_URI(uri: str) -> dict:
     if 'lightning' in out:
         try:
             lnaddr = decode_bolt11_invoice(out['lightning'])
-        except BOLT11DecodeException as e:
+        except BOLT11InvoiceException as e:
             raise InvalidBitcoinURI(f"Failed to decode 'lightning' field: {e!r}") from e
         amount_sat = out.get('amount')
         if amount_sat:

electrum/bolt11.py

@@ -7,13 +7,13 @@
 from hashlib import sha256
 from binascii import hexlify
 from decimal import Decimal
-from typing import Optional, TYPE_CHECKING, Type, Dict, Any, Sequence, Tuple
+from typing import Optional, TYPE_CHECKING, Type, Dict, Any, Sequence, Tuple, List
 import random
 
 import electrum_ecc as ecc
 
 from .bitcoin import hash160_to_b58_address, b58_address_to_hash160, TOTAL_COIN_SUPPLY_LIMIT_IN_BTC
-from .segwit_addr import bech32_encode, bech32_decode, CHARSET, CHARSET_INVERSE, convertbits
+from .segwit_addr import bech32_encode, bech32_decode, CHARSET, CHARSET_INVERSE, convertbits, INVALID_BECH32
 from . import segwit_addr
 from . import constants
 from .constants import AbstractNet
@@ -22,6 +22,8 @@
 if TYPE_CHECKING:
     from .lnutil import LnFeatures
 
+TIMESTAMP_SANE_MAX = 2**35 - 1
+
 
 class BOLT11InvoiceException(Exception): pass
 class BOLT11DecodeException(BOLT11InvoiceException): pass
@@ -96,13 +98,24 @@ def encode_fallback_addr(fallback: str, net: Type[AbstractNet]) -> Sequence[int]
 
 
 def parse_fallback_addr(data5: Sequence[int], net: Type[AbstractNet]) -> Optional[str]:
+    """Returns None if the fallback address cannot be parsed."""
+    if not data5:
+        return None
     wver = data5[0]
-    data8 = bytes(convertbits(data5[1:], 5, 8, False))
+    data8 = convertbits(data5[1:], 5, 8, False)
+    if data8 is None:  # invalid padding
+        return None
+    data8 = bytes(data8)
     if wver == 17:
+        if len(data8) != 20:  # hash160
+            return None
         addr = hash160_to_b58_address(data8, net.ADDRTYPE_P2PKH)
     elif wver == 18:
+        if len(data8) != 20:  # hash160
+            return None
         addr = hash160_to_b58_address(data8, net.ADDRTYPE_P2SH)
     elif wver <= 16:
+        # note: encode_segwit_address checks the witness program length
         addr = segwit_addr.encode_segwit_address(net.SEGWIT_HRP, wver, data8)
     else:
         return None
@@ -255,8 +268,15 @@ def encode_bolt11_invoice(addr: 'BOLT11Addr', privkey) -> str:
 
 
 class BOLT11Addr:
-    def __init__(self, *, paymenthash: bytes = None, amount=None, net: Type[AbstractNet] = None, tags=None, date=None,
-                 payment_secret: bytes = None):
+    def __init__(
+        self, *,
+        paymenthash: bytes = None,
+        amount: Optional[int | Decimal] = None,
+        net: Type[AbstractNet] = None,
+        tags: Optional[List[Tuple[str, Any]]] = None,
+        date: Optional[int | float] = None,
+        payment_secret: bytes = None
+    ):
         self.date = int(time.time()) if not date else int(date)
         self.tags = [] if not tags else tags
         self.unknown_tags = []
@@ -265,19 +285,21 @@ def __init__(self, *, paymenthash: bytes = None, amount=None, net: Type[Abstract
         self.signature = None
         self.pubkey = None
         self.net = constants.net if net is None else net  # type: Type[AbstractNet]
-        self._amount = amount  # type: Optional[Decimal]  # in bitcoins
+        self.amount = amount  # type: Optional[int | Decimal]  # in bitcoins
 
     @property
     def amount(self) -> Optional[Decimal]:
         return self._amount
 
     @amount.setter
-    def amount(self, value):
-        if not (isinstance(value, Decimal) or value is None):
-            raise BOLT11InvoiceException(f"amount must be Decimal or None, not {value!r}")
+    def amount(self, value: Optional[int | Decimal]):
+        if not (isinstance(value, (int, Decimal)) or value is None):
+            raise BOLT11InvoiceException(f"amount must be Decimal, int or None, not {value!r}")
         if value is None:
             self._amount = None
             return
+        if isinstance(value, int):
+            value = Decimal(value)
         assert isinstance(value, Decimal)
         if value.is_nan() or not (0 <= value <= TOTAL_COIN_SUPPLY_LIMIT_IN_BTC):
             raise BOLT11InvoiceException(f"amount is out-of-bounds: {value!r} BTC")
@@ -286,6 +308,21 @@ def amount(self, value):
             raise BOLT11InvoiceException(f"Cannot encode {value!r}: too many decimal places")
         self._amount = value
 
+    @property
+    def date(self) -> int:
+        return self._date
+
+    @date.setter
+    def date(self, value: int | float):
+        if value is None or not isinstance(value, (int, float)):
+            raise BOLT11InvoiceException(f"date must be int or float, not {value!r}")
+        if isinstance(value, float):
+            # e.g. from time.time()
+            value = int(value)
+        if not 0 <= value <= TIMESTAMP_SANE_MAX:
+            raise BOLT11InvoiceException(f"date must be in [0; {TIMESTAMP_SANE_MAX!r}]: {value}")
+        self._date = value
+
     def get_amount_sat(self) -> Optional[Decimal]:
         # note that this has msat resolution potentially
         if self.amount is None:
@@ -405,16 +442,36 @@ def serialize(self):
 
 
 def decode_bolt11_invoice(invoice: str, *, verbose=False, net=None) -> BOLT11Addr:
-    """Parses a string into a BOLT11Addr object.
-    Can raise BOLT11DecodeException or IncompatibleOrInsaneFeatures.
+    """Parses a bech32 encoded string into a BOLT11Addr object.
+    Can raise BOLT11DecodeException.
     """
+
+    def _convertbits_tag(
+            tag: str,
+            data5: Sequence[int],
+            *args,
+            length_range: Optional[Tuple[int, int]] = None,
+    ) -> Sequence[int]:
+        if length_range is not None and not length_range[0] <= len(data5) <= length_range[1]:
+            raise BOLT11DecodeException(
+                f"Invalid data_length for tag '{tag}': {len(data5)} (expected {length_range})")
+        if (intseq := convertbits(data5, *args)) is None:
+            raise BOLT11DecodeException(f"Failed to decode tag '{tag}'")
+        return intseq
+
+    def _check_minimal_data5(tag: str, data5: Sequence[int]) -> None:
+        """ a field is 'minimal' if len>0 and the first int is non-zero"""
+        if len(data5) > 0 and data5[0] == 0:
+            raise BOLT11DecodeException(f"Non-minimal data_length for tag '{tag}'")
+
     if net is None:
         net = constants.net
     decoded_bech32 = bech32_decode(invoice, ignore_long_length=True)
+    if decoded_bech32 is INVALID_BECH32:
+        raise BOLT11DecodeException("Invalid bech32 checksum")
     hrp = decoded_bech32.hrp
     data5 = decoded_bech32.data  # "5" as in list of 5-bit integers
-    if decoded_bech32.encoding is None:
-        raise BOLT11DecodeException("Bad bech32 checksum")
+    assert data5 is not None
     if decoded_bech32.encoding != segwit_addr.Encoding.BECH32:
         raise BOLT11DecodeException("Bad bech32 encoding: must be using vanilla BECH32")
 
@@ -444,21 +501,30 @@ def decode_bolt11_invoice(invoice: str, *, verbose=False, net=None) -> BOLT11Add
     # A reader SHOULD indicate if amount is unspecified, otherwise it MUST
     # multiply `amount` by the `multiplier` value (if any) to derive the
     # amount required for payment.
-    if amountstr != '':
-        addr.amount = unshorten_amount(amountstr)
+    try:
+        if amountstr != '':
+            addr.amount = unshorten_amount(amountstr)
 
-    addr.date = int_from_data5(data5_remaining[:7])
-    data5_remaining = data5_remaining[7:]
+        addr.date = int_from_data5(data5_remaining[:7])
+        data5_remaining = data5_remaining[7:]
+    except BOLT11InvoiceException as e:
+        # raise as decode exception, as amount/date comes from the encoded invoice
+        raise BOLT11DecodeException(f"Failed to decode invoice: {e}") from e
 
     while data5_remaining:
-        tag, tagdata = pull_tagged(data5_remaining)  # mutates arg
+        try:
+            tag, tagdata = pull_tagged(data5_remaining)  # mutates arg
+        except ValueError as e:
+            raise BOLT11DecodeException(f"Corrupt tag data: {str(e)}")
 
         # BOLT #11:
         #
-        # A reader MUST skip over unknown fields, an `f` field with unknown
-        # `version`, or a `p`, `h`, or `n` field which does not have
-        # `data_length` 52, 52, or 53 respectively.
-        data_length = len(tagdata)
+        # A reader:
+        #   - MUST skip over `f` fields that use an unknown `version`.
+        #   - MUST fail the payment if any field with fixed `data_length` (`p`, `h`, `s`, `n`)
+        #     does not have the correct length (52, 52, 52, 53).
+        # note: the fixed-length check is done via _convertbits_tag(length_range=...) below.
+        #       Until bolts#1243 (2025-06) the spec instead required *skipping* such fields
 
         if tag == 'r':
             # BOLT #11:
@@ -471,7 +537,7 @@ def decode_bolt11_invoice(invoice: str, *, verbose=False, net=None) -> BOLT11Add
             #    * `feebase` (32 bits, big-endian)
             #    * `feerate` (32 bits, big-endian)
             #    * `cltv_expiry_delta` (16 bits, big-endian)
-            tagdata = convertbits(tagdata, 5, 8, False)
+            tagdata = _convertbits_tag(tag, tagdata, 5, 8, False)
             if not tagdata:
                 continue
             route = []
@@ -491,7 +557,7 @@ def decode_bolt11_invoice(invoice: str, *, verbose=False, net=None) -> BOLT11Add
             if route:
                 addr.tags.append(('r',route))
         elif tag == 't':
-            tagdata = convertbits(tagdata, 5, 8, False)
+            tagdata = _convertbits_tag(tag, tagdata, 5, 8, False)
             if not tagdata:
                 continue
             route = []
@@ -514,42 +580,47 @@ def decode_bolt11_invoice(invoice: str, *, verbose=False, net=None) -> BOLT11Add
                 # Incorrect version.
                 addr.unknown_tags.append((tag, tagdata))
                 continue
-
         elif tag == 'd':
-            addr.tags.append(('d', bytes(convertbits(tagdata, 5, 8, False)).decode('utf-8')))
-
+            if addr.get_tag('d') is not None:
+                raise BOLT11DecodeException("Unexpected multiple 'd' tags")
+            try:
+                addr.tags.append(('d', bytes(_convertbits_tag(tag, tagdata, 5, 8, False)).decode('utf-8')))
+            except UnicodeDecodeError as e:
+                raise BOLT11DecodeException(f"Invalid UTF-8 content in invoice: {str(e)}")
         elif tag == 'h':
-            if data_length != 52:
-                addr.unknown_tags.append((tag, tagdata))
-                continue
-            addr.tags.append(('h', bytes(convertbits(tagdata, 5, 8, False))))
-
+            if addr.get_tag('h') is not None:
+                raise BOLT11DecodeException("Unexpected multiple 'h' tags")
+            addr.tags.append(
+                ('h', bytes(_convertbits_tag(tag, tagdata, 5, 8, False, length_range=(52, 52))))
+            )
         elif tag == 'x':
+            # MUST use the minimum data_length possible
+            _check_minimal_data5(tag, tagdata)
             addr.tags.append(('x', int_from_data5(tagdata)))
-
         elif tag == 'p':
-            if data_length != 52:
-                addr.unknown_tags.append((tag, tagdata))
-                continue
-            addr.paymenthash = bytes(convertbits(tagdata, 5, 8, False))
-
+            # MUST include exactly one 'p' field
+            if addr.paymenthash is not None:
+                raise BOLT11DecodeException("Unexpected 'p' tag")
+            addr.paymenthash = bytes(_convertbits_tag(tag, tagdata, 5, 8, False, length_range=(52, 52)))
         elif tag == 's':
-            if data_length != 52:
-                addr.unknown_tags.append((tag, tagdata))
-                continue
-            addr.payment_secret = bytes(convertbits(tagdata, 5, 8, False))
-
+            # MUST include exactly one 's' field
+            if addr.payment_secret is not None:
+                raise BOLT11DecodeException("Unexpected 's' tag")
+            addr.payment_secret = bytes(_convertbits_tag(tag, tagdata, 5, 8, False, length_range=(52, 52)))
         elif tag == 'n':
-            if data_length != 53:
-                addr.unknown_tags.append((tag, tagdata))
-                continue
-            pubkeybytes = bytes(convertbits(tagdata, 5, 8, False))
+            # MAY include one n field
+            if addr.pubkey is not None:
+                raise BOLT11DecodeException("Unexpected 'n' tag")
+            pubkeybytes = bytes(_convertbits_tag(tag, tagdata, 5, 8, False, length_range=(53, 53)))
             addr.pubkey = pubkeybytes
-
         elif tag == 'c':
+            # MUST use the minimum data_length possible
+            _check_minimal_data5(tag, tagdata)
             addr.tags.append(('c', int_from_data5(tagdata)))
-
         elif tag == '9':
+            # MUST use the minimum data_length possible to encode the non-zero bits,
+            # with no 0 field-elements at the start
+            _check_minimal_data5(tag, tagdata)
             features = int_from_data5(tagdata)
             addr.tags.append(('9', features))
             # note: The features are not validated here in the parser,
@@ -560,6 +631,16 @@ def decode_bolt11_invoice(invoice: str, *, verbose=False, net=None) -> BOLT11Add
         else:
             addr.unknown_tags.append((tag, tagdata))
 
+    # MUST include exactly one 'p' field
+    if addr.paymenthash is None:
+        raise BOLT11DecodeException("Missing 'p' tag")
+    # MUST include exactly one 's' field
+    if addr.payment_secret is None:
+        raise BOLT11DecodeException("Missing 's' tag")
+    # MUST include either exactly one d or exactly one h field
+    if bool(addr.get_tag('d') is not None) + bool(addr.get_tag('h') is not None) != 1:
+        raise BOLT11DecodeException("Exactly one of 'd' tag or 'h' tag must be present in invoice")
+
     if verbose:
         print('hex of signature data (32 byte r, 32 byte s): {}'
               .format(hexlify(sigdecoded[0:64])))
@@ -575,20 +656,25 @@ def decode_bolt11_invoice(invoice: str, *, verbose=False, net=None) -> BOLT11Add
     # field specified below).
     addr.signature = sigdecoded[:65]
     hrp_hash = sha256(hrp.encode("ascii") + bytes(convertbits(data5, 5, 8, True))).digest()
-    if addr.pubkey:  # Specified by `n`
-        # BOLT #11:
-        #
-        # A reader MUST use the `n` field to validate the signature instead of
-        # performing signature recovery if a valid `n` field is provided.
-        if not ecc.ECPubkey(addr.pubkey).ecdsa_verify(sigdecoded[:64], hrp_hash):
-            raise BOLT11DecodeException("bad signature")
-        pubkey_copy = addr.pubkey
-
-        class WrappedBytesKey:
-            serialize = lambda: pubkey_copy
-
-        addr.pubkey = WrappedBytesKey
-    else: # Recover pubkey from signature.
-        addr.pubkey = SerializableKey(ecc.ECPubkey.from_ecdsa_sig64(sigdecoded[:64], sigdecoded[64], hrp_hash))
+    try:
+        if addr.pubkey:  # Specified by `n`
+            # BOLT #11:
+            #
+            # A reader MUST use the `n` field to validate the signature instead of
+            # performing signature recovery if a valid `n` field is provided.
+            if not ecc.ECPubkey(addr.pubkey).ecdsa_verify(sigdecoded[:64], hrp_hash):
+                raise BOLT11DecodeException("bad signature")
+            pubkey_copy = addr.pubkey
+
+            class WrappedBytesKey:
+                serialize = lambda: pubkey_copy
+
+            addr.pubkey = WrappedBytesKey
+        else: # Recover pubkey from signature.
+            addr.pubkey = SerializableKey(ecc.ECPubkey.from_ecdsa_sig64(sigdecoded[:64], sigdecoded[64], hrp_hash))
+    except Exception as e:
+        if isinstance(e, BOLT11DecodeException):
+            raise
+        raise BOLT11DecodeException(f"Invalid signature: {e}") from e
 
     return addr