adding bolt11 lib and removing bolt11.py from the codebase (#1817)

* add latest bolt11 lib

decode exception handling for create_payment
fix json response for decode
bugfix hexing description hash
improvement on bolt11 lib
update to bolt11 2.0.1
fix clnrest
* bolt 2.0.4
* refactor core/crud.py

* catch bolt11 erxception clnrest

lnbits/bolt11.py

@@ -1,365 +1,7 @@
-import hashlib
+from bolt11 import (
-import re
+    Bolt11 as Invoice,  # noqa: F401
-import time
+)
-from decimal import Decimal
+from bolt11 import (
-from typing import List, NamedTuple, Optional
+    decode,  # noqa: F401
-
+    encode,  # noqa: F401
-import bitstring
+)
-import secp256k1
-from bech32 import CHARSET, bech32_decode, bech32_encode
-from ecdsa import SECP256k1, VerifyingKey
-from ecdsa.util import sigdecode_string
-
-
-class Route(NamedTuple):
-    pubkey: str
-    short_channel_id: str
-    base_fee_msat: int
-    ppm_fee: int
-    cltv: int
-
-
-class Invoice:
-    payment_hash: str
-    amount_msat: int = 0
-    description: Optional[str] = None
-    description_hash: Optional[str] = None
-    payee: Optional[str] = None
-    date: int
-    expiry: int = 3600
-    secret: Optional[str] = None
-    route_hints: List[Route] = []
-    min_final_cltv_expiry: int = 18
-
-
-def decode(pr: str) -> Invoice:
-    """bolt11 decoder,
-    based on https://github.com/rustyrussell/lightning-payencode/blob/master/lnaddr.py
-    """
-
-    hrp, decoded_data = bech32_decode(pr)
-    if hrp is None or decoded_data is None:
-        raise ValueError("Bad bech32 checksum")
-    if not hrp.startswith("ln"):
-        raise ValueError("Does not start with ln")
-
-    bitarray = _u5_to_bitarray(decoded_data)
-
-    # final signature 65 bytes, split it off.
-    if len(bitarray) < 65 * 8:
-        raise ValueError("Too short to contain signature")
-
-    # extract the signature
-    signature = bitarray[-65 * 8 :].tobytes()
-
-    # the tagged fields as a bitstream
-    data = bitstring.ConstBitStream(bitarray[: -65 * 8])
-
-    # build the invoice object
-    invoice = Invoice()
-
-    # decode the amount from the hrp
-    m = re.search(r"[^\d]+", hrp[2:])
-    if m:
-        amountstr = hrp[2 + m.end() :]
-        if amountstr != "":
-            invoice.amount_msat = _unshorten_amount(amountstr)
-
-    # pull out date
-    date_bin = data.read(35)
-    invoice.date = date_bin.uint  # type: ignore
-
-    while data.pos != data.len:
-        tag, tagdata, data = _pull_tagged(data)
-        data_length = len(tagdata or []) / 5
-
-        if tag == "d":
-            invoice.description = _trim_to_bytes(tagdata).decode()
-        elif tag == "h" and data_length == 52:
-            invoice.description_hash = _trim_to_bytes(tagdata).hex()
-        elif tag == "p" and data_length == 52:
-            invoice.payment_hash = _trim_to_bytes(tagdata).hex()
-        elif tag == "x":
-            invoice.expiry = tagdata.uint  # type: ignore
-        elif tag == "n":
-            invoice.payee = _trim_to_bytes(tagdata).hex()
-            # this won't work in most cases, we must extract the payee
-            # from the signature
-        elif tag == "s":
-            invoice.secret = _trim_to_bytes(tagdata).hex()
-        elif tag == "r":
-            s = bitstring.ConstBitStream(tagdata)
-            while s.pos + 264 + 64 + 32 + 32 + 16 < s.len:
-                route = Route(
-                    pubkey=s.read(264).tobytes().hex(),  # type: ignore
-                    short_channel_id=_readable_scid(s.read(64).intbe),  # type: ignore
-                    base_fee_msat=s.read(32).intbe,  # type: ignore
-                    ppm_fee=s.read(32).intbe,  # type: ignore
-                    cltv=s.read(16).intbe,  # type: ignore
-                )
-                invoice.route_hints.append(route)
-
-    # BOLT #11:
-    # A reader MUST check that the `signature` is valid (see the `n` tagged
-    # field specified below).
-    # A reader MUST use the `n` field to validate the signature instead of
-    # performing signature recovery if a valid `n` field is provided.
-    message = bytearray([ord(c) for c in hrp]) + data.tobytes()
-    sig = signature[0:64]
-    if invoice.payee:
-        key = VerifyingKey.from_string(bytes.fromhex(invoice.payee), curve=SECP256k1)
-        key.verify(sig, message, hashlib.sha256, sigdecode=sigdecode_string)
-    else:
-        keys = VerifyingKey.from_public_key_recovery(
-            sig, message, SECP256k1, hashlib.sha256
-        )
-        signaling_byte = signature[64]
-        key = keys[int(signaling_byte)]
-        invoice.payee = key.to_string("compressed").hex()
-
-    return invoice
-
-
-def encode(options):
-    """Convert options into LnAddr and pass it to the encoder"""
-    addr = LnAddr()
-    addr.currency = options["currency"]
-    addr.fallback = options["fallback"] if options["fallback"] else None
-    if options["amount"]:
-        addr.amount = options["amount"]
-    if options["timestamp"]:
-        addr.date = int(options["timestamp"])
-
-    addr.paymenthash = bytes.fromhex(options["paymenthash"])
-
-    if options["description"]:
-        addr.tags.append(("d", options["description"]))
-    if options["description_hash"]:
-        addr.tags.append(("h", options["description_hash"]))
-    if options["expires"]:
-        addr.tags.append(("x", options["expires"]))
-
-    if options["fallback"]:
-        addr.tags.append(("f", options["fallback"]))
-    if options["route"]:
-        for r in options["route"]:
-            splits = r.split("/")
-            route = []
-            while len(splits) >= 5:
-                route.append(
-                    (
-                        bytes.fromhex(splits[0]),
-                        bytes.fromhex(splits[1]),
-                        int(splits[2]),
-                        int(splits[3]),
-                        int(splits[4]),
-                    )
-                )
-                splits = splits[5:]
-            assert len(splits) == 0
-            addr.tags.append(("r", route))
-    return lnencode(addr, options["privkey"])
-
-
-def lnencode(addr, privkey):
-    if addr.amount:
-        amount = Decimal(str(addr.amount))
-        # We can only send down to millisatoshi.
-        if amount * 10**12 % 10:
-            raise ValueError(f"Cannot encode {addr.amount}: too many decimal places")
-
-        amount = addr.currency + shorten_amount(amount)
-    else:
-        amount = addr.currency if addr.currency else ""
-
-    hrp = f"ln{amount}0n"
-
-    # Start with the timestamp
-    data = bitstring.pack("uint:35", addr.date)
-
-    # Payment hash
-    data += tagged_bytes("p", addr.paymenthash)
-    tags_set = set()
-
-    for k, v in addr.tags:
-        # BOLT #11:
-        #
-        # A writer MUST NOT include more than one `d`, `h`, `n` or `x` fields,
-        if k in ("d", "h", "n", "x"):
-            if k in tags_set:
-                raise ValueError(f"Duplicate '{k}' tag")
-
-        if k == "r":
-            route = bitstring.BitArray()
-            for step in v:
-                pubkey, channel, feebase, feerate, cltv = step
-                route.append(
-                    bitstring.BitArray(pubkey)
-                    + bitstring.BitArray(channel)
-                    + bitstring.pack("intbe:32", feebase)
-                    + bitstring.pack("intbe:32", feerate)
-                    + bitstring.pack("intbe:16", cltv)
-                )
-            data += tagged("r", route)
-        elif k == "f":
-            # NOTE: there was an error fallback here that's now removed
-            continue
-        elif k == "d":
-            data += tagged_bytes("d", v.encode())
-        elif k == "x":
-            # Get minimal length by trimming leading 5 bits at a time.
-            expirybits = bitstring.pack("intbe:64", v)[4:64]
-            while expirybits.startswith("0b00000"):
-                expirybits = expirybits[5:]
-            data += tagged("x", expirybits)
-        elif k == "h":
-            data += tagged_bytes("h", v)
-        elif k == "n":
-            data += tagged_bytes("n", v)
-        else:
-            # FIXME: Support unknown tags?
-            raise ValueError(f"Unknown tag {k}")
-
-        tags_set.add(k)
-
-    # BOLT #11:
-    #
-    # A writer MUST include either a `d` or `h` field, and MUST NOT include
-    # both.
-    if "d" in tags_set and "h" in tags_set:
-        raise ValueError("Cannot include both 'd' and 'h'")
-    if "d" not in tags_set and "h" not in tags_set:
-        raise ValueError("Must include either 'd' or 'h'")
-
-    # We actually sign the hrp, then data (padded to 8 bits with zeroes).
-    privkey = secp256k1.PrivateKey(bytes.fromhex(privkey))
-    sig = privkey.ecdsa_sign_recoverable(
-        bytearray([ord(c) for c in hrp]) + data.tobytes()
-    )
-    # This doesn't actually serialize, but returns a pair of values :(
-    sig, recid = privkey.ecdsa_recoverable_serialize(sig)
-    data += bytes(sig) + bytes([recid])
-
-    return bech32_encode(hrp, bitarray_to_u5(data))
-
-
-class LnAddr:
-    def __init__(
-        self,
-        paymenthash=None,
-        amount=None,
-        currency="bc",
-        tags=None,
-        date=None,
-        fallback=None,
-    ):
-        self.date = int(time.time()) if not date else int(date)
-        self.tags = [] if not tags else tags
-        self.unknown_tags = []
-        self.paymenthash = paymenthash
-        self.signature = None
-        self.pubkey = None
-        self.fallback = fallback
-        self.currency = currency
-        self.amount = amount
-
-    def __str__(self):
-        assert self.pubkey, "LnAddr, pubkey must be set"
-        pubkey = bytes.hex(self.pubkey.serialize())
-        tags = ", ".join([f"{k}={v}" for k, v in self.tags])
-        return f"LnAddr[{pubkey}, amount={self.amount}{self.currency} tags=[{tags}]]"
-
-
-def shorten_amount(amount):
-    """Given an amount in bitcoin, shorten it"""
-    # Convert to pico initially
-    amount = int(amount * 10**12)
-    units = ["p", "n", "u", "m", ""]
-    unit = ""
-    for unit in units:
-        if amount % 1000 == 0:
-            amount //= 1000
-        else:
-            break
-    return str(amount) + unit
-
-
-def _unshorten_amount(amount: str) -> int:
-    """Given a shortened amount, return millisatoshis"""
-    # BOLT #11:
-    # The following `multiplier` letters are defined:
-    #
-    # * `m` (milli): multiply by 0.001
-    # * `u` (micro): multiply by 0.000001
-    # * `n` (nano): multiply by 0.000000001
-    # * `p` (pico): multiply by 0.000000000001
-    units = {"p": 10**12, "n": 10**9, "u": 10**6, "m": 10**3}
-    unit = str(amount)[-1]
-
-    # BOLT #11:
-    # A reader SHOULD fail if `amount` contains a non-digit, or is followed by
-    # anything except a `multiplier` in the table above.
-    if not re.fullmatch(r"\d+[pnum]?", str(amount)):
-        raise ValueError(f"Invalid amount '{amount}'")
-
-    if unit in units:
-        return int(int(amount[:-1]) * 100_000_000_000 / units[unit])
-    else:
-        return int(amount) * 100_000_000_000
-
-
-def _pull_tagged(stream):
-    tag = stream.read(5).uint
-    length = stream.read(5).uint * 32 + stream.read(5).uint
-    return (CHARSET[tag], stream.read(length * 5), stream)
-
-
-# Tagged field containing BitArray
-def tagged(char, bits):
-    # Tagged fields need to be zero-padded to 5 bits.
-    while bits.len % 5 != 0:
-        bits.append("0b0")
-    return (
-        bitstring.pack(
-            "uint:5, uint:5, uint:5",
-            CHARSET.find(char),
-            (bits.len / 5) / 32,
-            (bits.len / 5) % 32,
-        )
-        + bits
-    )
-
-
-def tagged_bytes(char, bits):
-    return tagged(char, bitstring.BitArray(bits))
-
-
-def _trim_to_bytes(barr):
-    # Adds a byte if necessary.
-    b = barr.tobytes()
-    if barr.len % 8 != 0:
-        return b[:-1]
-    return b
-
-
-def _readable_scid(short_channel_id: int) -> str:
-    blockheight = (short_channel_id >> 40) & 0xFFFFFF
-    transactionindex = (short_channel_id >> 16) & 0xFFFFFF
-    outputindex = short_channel_id & 0xFFFF
-    return f"{blockheight}x{transactionindex}x{outputindex}"
-
-
-def _u5_to_bitarray(arr: List[int]) -> bitstring.BitArray:
-    ret = bitstring.BitArray()
-    for a in arr:
-        ret += bitstring.pack("uint:5", a)
-    return ret
-
-
-def bitarray_to_u5(barr):
-    assert barr.len % 5 == 0
-    ret = []
-    s = bitstring.ConstBitStream(barr)
-    while s.pos != s.len:
-        ret.append(s.read(5).uint)  # type: ignore
-    return ret

lnbits/core/crud.py

@@ -5,8 +5,8 @@ from urllib.parse import urlparse
 from uuid import UUID, uuid4
 
 import shortuuid
+from bolt11.decode import decode
 
-from lnbits import bolt11
 from lnbits.core.db import db
 from lnbits.core.models import WalletType
 from lnbits.db import DB_TYPE, SQLITE, Connection, Database, Filters, Page
@@ -545,10 +545,11 @@ async def create_payment(
     previous_payment = await get_standalone_payment(checking_id, conn=conn)
     assert previous_payment is None, "Payment already exists"
 
-    try:
+    invoice = decode(payment_request)
-        invoice = bolt11.decode(payment_request)
+
+    if invoice.expiry:
         expiration_date = datetime.datetime.fromtimestamp(invoice.date + invoice.expiry)
-    except Exception:
+    else:
         # assume maximum bolt11 expiry of 31 days to be on the safe side
         expiration_date = datetime.datetime.now() + datetime.timedelta(days=31)
 

lnbits/core/services.py

@@ -176,6 +176,12 @@ async def pay_invoice(
     will regularly check for the payment.
     """
     invoice = bolt11.decode(payment_request)
+
+    if not invoice.amount_msat or not invoice.amount_msat > 0:
+        raise ValueError("Amountless invoices not supported.")
+    if max_sat and invoice.amount_msat > max_sat * 1000:
+        raise ValueError("Amount in invoice is too high.")
+
     fee_reserve_msat = fee_reserve(invoice.amount_msat)
     async with db.reuse_conn(conn) if conn else db.connect() as conn:
         temp_id = invoice.payment_hash

lnbits/core/views/api.py

@@ -16,11 +16,11 @@ from fastapi import (
     Depends,
     Header,
     Request,
-    Response,
     WebSocket,
     WebSocketDisconnect,
 )
 from fastapi.exceptions import HTTPException
+from fastapi.responses import JSONResponse
 from loguru import logger
 from sse_starlette.sse import EventSourceResponse
 from starlette.responses import RedirectResponse, StreamingResponse
@@ -622,29 +622,20 @@ async def api_lnurlscan(code: str, wallet: WalletTypeInfo = Depends(get_key_type
 
 
 @api_router.post("/api/v1/payments/decode", status_code=HTTPStatus.OK)
-async def api_payments_decode(data: DecodePayment, response: Response):
+async def api_payments_decode(data: DecodePayment) -> JSONResponse:
     payment_str = data.data
     try:
         if payment_str[:5] == "LNURL":
             url = lnurl.decode(payment_str)
-            return {"domain": url}
+            return JSONResponse({"domain": url})
         else:
             invoice = bolt11.decode(payment_str)
-            return {
+            return JSONResponse(invoice.data)
-                "payment_hash": invoice.payment_hash,
+    except Exception as exc:
-                "amount_msat": invoice.amount_msat,
+        return JSONResponse(
-                "description": invoice.description,
+            {"message": f"Failed to decode: {str(exc)}"},
-                "description_hash": invoice.description_hash,
+            status_code=HTTPStatus.BAD_REQUEST,
-                "payee": invoice.payee,
+        )
-                "date": invoice.date,
-                "expiry": invoice.expiry,
-                "secret": invoice.secret,
-                "route_hints": invoice.route_hints,
-                "min_final_cltv_expiry": invoice.min_final_cltv_expiry,
-            }
-    except Exception:
-        response.status_code = HTTPStatus.BAD_REQUEST
-        return {"message": "Failed to decode"}
 
 
 @api_router.post("/api/v1/lnurlauth")

lnbits/core/views/public_api.py

@@ -1,5 +1,4 @@
 import asyncio
-import datetime
 from http import HTTPStatus
 
 from fastapi import APIRouter, HTTPException
@@ -26,8 +25,7 @@ async def api_public_payment_longpolling(payment_hash):
 
     try:
         invoice = bolt11.decode(payment.bolt11)
-        expiration = datetime.datetime.fromtimestamp(invoice.date + invoice.expiry)
+        if invoice.has_expired():
-        if expiration < datetime.datetime.now():
             return {"status": "expired"}
     except Exception:
         raise HTTPException(

lnbits/wallets/corelightning.py

@@ -2,10 +2,11 @@ import asyncio
 import random
 from typing import Any, AsyncGenerator, Optional
 
+from bolt11.decode import decode as bolt11_decode
+from bolt11.exceptions import Bolt11Exception
 from loguru import logger
 from pyln.client import LightningRpc, RpcError
 
-from lnbits import bolt11 as lnbits_bolt11
 from lnbits.settings import settings
 
 from .base import (
@@ -95,12 +96,20 @@ class CoreLightningWallet(Wallet):
             return InvoiceResponse(False, None, None, str(e))
 
     async def pay_invoice(self, bolt11: str, fee_limit_msat: int) -> PaymentResponse:
-        invoice = lnbits_bolt11.decode(bolt11)
+        try:
+            invoice = bolt11_decode(bolt11)
+        except Bolt11Exception as exc:
+            return PaymentResponse(False, None, None, None, str(exc))
 
         previous_payment = await self.get_payment_status(invoice.payment_hash)
         if previous_payment.paid:
             return PaymentResponse(False, None, None, None, "invoice already paid")
 
+        if not invoice.amount_msat or invoice.amount_msat <= 0:
+            return PaymentResponse(
+                False, None, None, None, "CLN 0 amount invoice not supported"
+            )
+
         fee_limit_percent = fee_limit_msat / invoice.amount_msat * 100
         # so fee_limit_percent is applied even on payments with fee < 5000 millisatoshi
         # (which is default value of exemptfee)

lnbits/wallets/corelightningrest.py

@@ -4,9 +4,10 @@ import random
 from typing import AsyncGenerator, Dict, Optional
 
 import httpx
+from bolt11 import Bolt11Exception
+from bolt11.decode import decode
 from loguru import logger
 
-from lnbits import bolt11 as lnbits_bolt11
 from lnbits.settings import settings
 
 from .base import (
@@ -129,7 +130,14 @@ class CoreLightningRestWallet(Wallet):
         return InvoiceResponse(True, label, data["bolt11"], None)
 
     async def pay_invoice(self, bolt11: str, fee_limit_msat: int) -> PaymentResponse:
-        invoice = lnbits_bolt11.decode(bolt11)
+        try:
+            invoice = decode(bolt11)
+        except Bolt11Exception as exc:
+            return PaymentResponse(False, None, None, None, str(exc))
+
+        if not invoice.amount_msat or invoice.amount_msat <= 0:
+            error_message = "0 amount invoices are not allowed"
+            return PaymentResponse(False, None, None, None, error_message)
         fee_limit_percent = fee_limit_msat / invoice.amount_msat * 100
         r = await self.client.post(
             f"{self.url}/v1/pay",

lnbits/wallets/fake.py

@@ -2,13 +2,22 @@ import asyncio
 import hashlib
 import random
 from datetime import datetime
-from typing import AsyncGenerator, Dict, Optional
+from os import urandom
+from typing import AsyncGenerator, Optional
 
+from bolt11 import (
+    Bolt11,
+    Bolt11Exception,
+    MilliSatoshi,
+    TagChar,
+    Tags,
+    decode,
+    encode,
+)
 from loguru import logger
 
 from lnbits.settings import settings
 
-from ..bolt11 import Invoice, decode, encode
 from .base import (
     InvoiceResponse,
     PaymentResponse,
@@ -42,44 +51,55 @@ class FakeWallet(Wallet):
         memo: Optional[str] = None,
         description_hash: Optional[bytes] = None,
         unhashed_description: Optional[bytes] = None,
-        **kwargs,
+        expiry: Optional[int] = None,
+        payment_secret: Optional[bytes] = None,
+        **_,
     ) -> InvoiceResponse:
-        data: Dict = {
+        tags = Tags()
-            "out": False,
+
-            "amount": amount * 1000,
-            "currency": "bc",
-            "privkey": self.privkey,
-            "memo": memo,
-            "description_hash": b"",
-            "description": "",
-            "fallback": None,
-            "expires": kwargs.get("expiry"),
-            "timestamp": datetime.now().timestamp(),
-            "route": None,
-            "tags_set": [],
-        }
         if description_hash:
-            data["tags_set"] = ["h"]
+            tags.add(TagChar.description_hash, description_hash.hex())
-            data["description_hash"] = description_hash
         elif unhashed_description:
-            data["tags_set"] = ["h"]
+            tags.add(
-            data["description_hash"] = hashlib.sha256(unhashed_description).digest()
+                TagChar.description_hash,
+                hashlib.sha256(unhashed_description).hexdigest(),
+            )
         else:
-            data["tags_set"] = ["d"]
+            tags.add(TagChar.description, memo or "")
-            data["memo"] = memo
+
-            data["description"] = memo
+        if expiry:
-        randomHash = (
+            tags.add(TagChar.expire_time, expiry)
+
+        # random hash
+        checking_id = (
             self.privkey[:6]
             + hashlib.sha256(str(random.getrandbits(256)).encode()).hexdigest()[6:]
         )
-        data["paymenthash"] = randomHash
+
-        payment_request = encode(data)
+        tags.add(TagChar.payment_hash, checking_id)
-        checking_id = randomHash
+
+        if payment_secret:
+            secret = payment_secret.hex()
+        else:
+            secret = urandom(32).hex()
+        tags.add(TagChar.payment_secret, secret)
+
+        bolt11 = Bolt11(
+            currency="bc",
+            amount_msat=MilliSatoshi(amount * 1000),
+            date=int(datetime.now().timestamp()),
+            tags=tags,
+        )
+
+        payment_request = encode(bolt11, self.privkey)
 
         return InvoiceResponse(True, checking_id, payment_request)
 
     async def pay_invoice(self, bolt11: str, _: int) -> PaymentResponse:
+        try:
             invoice = decode(bolt11)
+        except Bolt11Exception as exc:
+            return PaymentResponse(ok=False, error_message=str(exc))
 
         if invoice.payment_hash[:6] == self.privkey[:6]:
             await self.queue.put(invoice)
@@ -97,5 +117,5 @@ class FakeWallet(Wallet):
 
     async def paid_invoices_stream(self) -> AsyncGenerator[str, None]:
         while True:
-            value: Invoice = await self.queue.get()
+            value: Bolt11 = await self.queue.get()
             yield value.payment_hash

poetry.lock

@@ -144,6 +144,25 @@ d = ["aiohttp (>=3.7.4)"]
 jupyter = ["ipython (>=7.8.0)", "tokenize-rt (>=3.2.0)"]
 uvloop = ["uvloop (>=0.15.2)"]
 
+[[package]]
+name = "bolt11"
+version = "2.0.5"
+description = "A library for encoding and decoding BOLT11 payment requests."
+optional = false
+python-versions = ">=3.8.1"
+files = [
+    {file = "bolt11-2.0.5-py3-none-any.whl", hash = "sha256:6791c2edee804a4a8a7d092c689f8d2c01212271a33963ede4a988b7a6ce1b81"},
+    {file = "bolt11-2.0.5.tar.gz", hash = "sha256:e6be2748b0c4a017900761f63d9944c1dde8f22fd2829006679a0e2346eaa47b"},
+]
+
+[package.dependencies]
+base58 = "*"
+bech32 = "*"
+bitstring = "*"
+click = "*"
+ecdsa = "*"
+secp256k1 = "*"
+
 [[package]]
 name = "cerberus"
 version = "1.3.4"
@@ -2418,4 +2437,4 @@ testing = ["big-O", "jaraco.functools", "jaraco.itertools", "more-itertools", "p
 [metadata]
 lock-version = "2.0"
 python-versions = "^3.10 | ^3.9"
-content-hash = "c21ec693cd8737a77199612b2608c413b1fd18c25befdefa86005f0e02bca28c"
+content-hash = "4fd361a6f46c9a1ad34b000ad13a56c076593398eab4f4bad157c608bfa6d6f4"

pyproject.toml

@@ -7,7 +7,6 @@ authors = ["Alan Bits <alan@lnbits.com>"]
 [tool.poetry.dependencies]
 python = "^3.10 | ^3.9"
 bech32 = "1.2.0"
-bitstring = "3.1.9"
 click = "8.1.7"
 ecdsa = "0.18.0"
 embit = "0.7.0"
@@ -39,6 +38,7 @@ websocket-client = "1.6.3"
 secp256k1 = "0.14.0"
 pycryptodomex = "3.19.0"
 packaging = "23.1"
+bolt11 = "^2.0.5"
 
 [tool.poetry.group.dev.dependencies]
 black = "^23.7.0"
@@ -90,6 +90,7 @@ module = [
   "shortuuid.*",
   "grpc.*",
   "lnurl.*",
+  "bolt11.*",
   "bitstring.*",
   "ecdsa.*",
   "psycopg2.*",

tests/core/views/test_api.py

@@ -379,8 +379,8 @@ async def test_create_invoice_with_description_hash(client, inkey_headers_to):
         "/api/v1/payments", json=data, headers=inkey_headers_to
     )
     invoice = response.json()
-    invoice_bolt11 = bolt11.decode(invoice["payment_request"])
 
+    invoice_bolt11 = bolt11.decode(invoice["payment_request"])
     assert invoice_bolt11.description_hash == descr_hash
     return invoice
 
@@ -392,8 +392,9 @@ async def test_create_invoice_with_description_hash(client, inkey_headers_to):
 @pytest.mark.asyncio
 async def test_create_invoice_with_unhashed_description(client, inkey_headers_to):
     data = await get_random_invoice_data()
-    descr_hash = hashlib.sha256("asdasdasd".encode()).hexdigest()
+    description = "test description"
-    data["unhashed_description"] = "asdasdasd".encode().hex()
+    descr_hash = hashlib.sha256(description.encode()).hexdigest()
+    data["unhashed_description"] = description.encode().hex()
 
     response = await client.post(
         "/api/v1/payments", json=data, headers=inkey_headers_to