invoices.py (7365B)
1 import time 2 from typing import TYPE_CHECKING, List, Optional, Union, Dict, Any 3 from decimal import Decimal 4 5 import attr 6 7 from .json_db import StoredObject 8 from .i18n import _ 9 from .util import age 10 from .lnaddr import lndecode, LnAddr 11 from . import constants 12 from .bitcoin import COIN, TOTAL_COIN_SUPPLY_LIMIT_IN_BTC 13 from .transaction import PartialTxOutput 14 15 if TYPE_CHECKING: 16 from .paymentrequest import PaymentRequest 17 18 # convention: 'invoices' = outgoing , 'request' = incoming 19 20 # types of payment requests 21 PR_TYPE_ONCHAIN = 0 22 PR_TYPE_LN = 2 23 24 # status of payment requests 25 PR_UNPAID = 0 26 PR_EXPIRED = 1 27 PR_UNKNOWN = 2 # sent but not propagated 28 PR_PAID = 3 # send and propagated 29 PR_INFLIGHT = 4 # unconfirmed 30 PR_FAILED = 5 31 PR_ROUTING = 6 32 PR_UNCONFIRMED = 7 33 34 pr_color = { 35 PR_UNPAID: (.7, .7, .7, 1), 36 PR_PAID: (.2, .9, .2, 1), 37 PR_UNKNOWN: (.7, .7, .7, 1), 38 PR_EXPIRED: (.9, .2, .2, 1), 39 PR_INFLIGHT: (.9, .6, .3, 1), 40 PR_FAILED: (.9, .2, .2, 1), 41 PR_ROUTING: (.9, .6, .3, 1), 42 PR_UNCONFIRMED: (.9, .6, .3, 1), 43 } 44 45 pr_tooltips = { 46 PR_UNPAID:_('Unpaid'), 47 PR_PAID:_('Paid'), 48 PR_UNKNOWN:_('Unknown'), 49 PR_EXPIRED:_('Expired'), 50 PR_INFLIGHT:_('In progress'), 51 PR_FAILED:_('Failed'), 52 PR_ROUTING: _('Computing route...'), 53 PR_UNCONFIRMED: _('Unconfirmed'), 54 } 55 56 PR_DEFAULT_EXPIRATION_WHEN_CREATING = 24*60*60 # 1 day 57 pr_expiration_values = { 58 0: _('Never'), 59 10*60: _('10 minutes'), 60 60*60: _('1 hour'), 61 24*60*60: _('1 day'), 62 7*24*60*60: _('1 week'), 63 } 64 assert PR_DEFAULT_EXPIRATION_WHEN_CREATING in pr_expiration_values 65 66 67 def _decode_outputs(outputs) -> List[PartialTxOutput]: 68 ret = [] 69 for output in outputs: 70 if not isinstance(output, PartialTxOutput): 71 output = PartialTxOutput.from_legacy_tuple(*output) 72 ret.append(output) 73 return ret 74 75 76 # hack: BOLT-11 is not really clear on what an expiry of 0 means. 77 # It probably interprets it as 0 seconds, so already expired... 78 # Our higher level invoices code however uses 0 for "never". 79 # Hence set some high expiration here 80 LN_EXPIRY_NEVER = 100 * 365 * 24 * 60 * 60 # 100 years 81 82 @attr.s 83 class Invoice(StoredObject): 84 type = attr.ib(type=int, kw_only=True) 85 86 message: str 87 exp: int 88 time: int 89 90 def is_lightning(self): 91 return self.type == PR_TYPE_LN 92 93 def get_status_str(self, status): 94 status_str = pr_tooltips[status] 95 if status == PR_UNPAID: 96 if self.exp > 0 and self.exp != LN_EXPIRY_NEVER: 97 expiration = self.exp + self.time 98 status_str = _('Expires') + ' ' + age(expiration, include_seconds=True) 99 return status_str 100 101 def get_amount_sat(self) -> Union[int, Decimal, str, None]: 102 """Returns a decimal satoshi amount, or '!' or None.""" 103 raise NotImplementedError() 104 105 @classmethod 106 def from_json(cls, x: dict) -> 'Invoice': 107 # note: these raise if x has extra fields 108 if x.get('type') == PR_TYPE_LN: 109 return LNInvoice(**x) 110 else: 111 return OnchainInvoice(**x) 112 113 114 @attr.s 115 class OnchainInvoice(Invoice): 116 message = attr.ib(type=str, kw_only=True) 117 amount_sat = attr.ib(kw_only=True) # type: Union[int, str] # in satoshis. can be '!' 118 exp = attr.ib(type=int, kw_only=True, validator=attr.validators.instance_of(int)) 119 time = attr.ib(type=int, kw_only=True, validator=attr.validators.instance_of(int)) 120 id = attr.ib(type=str, kw_only=True) 121 outputs = attr.ib(kw_only=True, converter=_decode_outputs) # type: List[PartialTxOutput] 122 bip70 = attr.ib(type=str, kw_only=True) # type: Optional[str] 123 requestor = attr.ib(type=str, kw_only=True) # type: Optional[str] 124 height = attr.ib(type=int, kw_only=True, validator=attr.validators.instance_of(int)) 125 126 def get_address(self) -> str: 127 """returns the first address, to be displayed in GUI""" 128 return self.outputs[0].address 129 130 def get_amount_sat(self) -> Union[int, str]: 131 return self.amount_sat or 0 132 133 @amount_sat.validator 134 def _validate_amount(self, attribute, value): 135 if isinstance(value, int): 136 if not (0 <= value <= TOTAL_COIN_SUPPLY_LIMIT_IN_BTC * COIN): 137 raise ValueError(f"amount is out-of-bounds: {value!r} sat") 138 elif isinstance(value, str): 139 if value != "!": 140 raise ValueError(f"unexpected amount: {value!r}") 141 else: 142 raise ValueError(f"unexpected amount: {value!r}") 143 144 @classmethod 145 def from_bip70_payreq(cls, pr: 'PaymentRequest', height:int) -> 'OnchainInvoice': 146 return OnchainInvoice( 147 type=PR_TYPE_ONCHAIN, 148 amount_sat=pr.get_amount(), 149 outputs=pr.get_outputs(), 150 message=pr.get_memo(), 151 id=pr.get_id(), 152 time=pr.get_time(), 153 exp=pr.get_expiration_date() - pr.get_time(), 154 bip70=pr.raw.hex(), 155 requestor=pr.get_requestor(), 156 height=height, 157 ) 158 159 @attr.s 160 class LNInvoice(Invoice): 161 invoice = attr.ib(type=str) 162 amount_msat = attr.ib(kw_only=True) # type: Optional[int] # needed for zero amt invoices 163 164 __lnaddr = None 165 166 @invoice.validator 167 def _validate_invoice_str(self, attribute, value): 168 lndecode(value) # this checks the str can be decoded 169 170 @amount_msat.validator 171 def _validate_amount(self, attribute, value): 172 if value is None: 173 return 174 if isinstance(value, int): 175 if not (0 <= value <= TOTAL_COIN_SUPPLY_LIMIT_IN_BTC * COIN * 1000): 176 raise ValueError(f"amount is out-of-bounds: {value!r} msat") 177 else: 178 raise ValueError(f"unexpected amount: {value!r}") 179 180 @property 181 def _lnaddr(self) -> LnAddr: 182 if self.__lnaddr is None: 183 self.__lnaddr = lndecode(self.invoice) 184 return self.__lnaddr 185 186 @property 187 def rhash(self) -> str: 188 return self._lnaddr.paymenthash.hex() 189 190 def get_amount_msat(self) -> Optional[int]: 191 amount_btc = self._lnaddr.amount 192 amount = int(amount_btc * COIN * 1000) if amount_btc else None 193 return amount or self.amount_msat 194 195 def get_amount_sat(self) -> Union[Decimal, None]: 196 amount_msat = self.get_amount_msat() 197 if amount_msat is None: 198 return None 199 return Decimal(amount_msat) / 1000 200 201 @property 202 def exp(self) -> int: 203 return self._lnaddr.get_expiry() 204 205 @property 206 def time(self) -> int: 207 return self._lnaddr.date 208 209 @property 210 def message(self) -> str: 211 return self._lnaddr.get_description() 212 213 @classmethod 214 def from_bech32(cls, invoice: str) -> 'LNInvoice': 215 amount_msat = lndecode(invoice).get_amount_msat() 216 return LNInvoice( 217 type=PR_TYPE_LN, 218 invoice=invoice, 219 amount_msat=amount_msat, 220 ) 221 222 def to_debug_json(self) -> Dict[str, Any]: 223 d = self.to_json() 224 d.update({ 225 'pubkey': self._lnaddr.pubkey.serialize().hex(), 226 'amount_BTC': str(self._lnaddr.amount), 227 'rhash': self._lnaddr.paymenthash.hex(), 228 'description': self._lnaddr.get_description(), 229 'exp': self._lnaddr.get_expiry(), 230 'time': self._lnaddr.date, 231 # 'tags': str(lnaddr.tags), 232 }) 233 return d 234