# shared channel config fields

payment_basepoint = attr.ib(type=OnlyPubkeyKeypair)

multisig_key = attr.ib(type=OnlyPubkeyKeypair)

htlc_basepoint = attr.ib(type=OnlyPubkeyKeypair)

delayed_basepoint = attr.ib(type=OnlyPubkeyKeypair)

revocation_basepoint = attr.ib(type=OnlyPubkeyKeypair)

to_self_delay = attr.ib(type=int)

dust_limit_sat = attr.ib(type=int)

max_htlc_value_in_flight_msat = attr.ib(type=int)

max_accepted_htlcs = attr.ib(type=int)

initial_msat = attr.ib(type=int)

# specific to "LOCAL" config

per_commitment_secret_seed = attr.ib(type=bytes)

funding_locked_received = attr.ib(type=bool)

was_announced = attr.ib(type=bool)

current_commitment_signature = attr.ib(type=bytes)

# specific to "REMOTE" config

htlc_minimum_msat = attr.ib(type=int)

next_per_commitment_point = attr.ib(type=bytes)

txid = attr.ib(type=str)

output_index = attr.ib(type=int)

def to_str(self):

success: bool

route: Optional['LNPaymentRoute'] = None

preimage: Optional[bytes] = None

failure_details: Optional[PaymentAttemptFailureDetails] = None

""" Taken from LND, see license in lnchannel.py. """

START_INDEX = 2 ** 48 - 1

def __init__(self, storage):

if len(storage) == 0:

storage['index'] = self.START_INDEX

storage['buckets'] = {}

self.storage = storage

self.buckets = storage['buckets']

def add_next_entry(self, hsh):

index = self.storage['index']

new_element = ShachainElement(index=index, secret=hsh)

bucket = count_trailing_zeros(index)

for i in range(0, bucket):

this_bucket = self.buckets[i]

e = shachain_derive(new_element, this_bucket.index)

if e != this_bucket:

raise Exception("hash is not derivable: {} {} {}".format(bh2u(e.secret), bh2u(this_bucket.secret), this_bucket.index))

self.buckets[bucket] = new_element

self.storage['index'] = index - 1

def retrieve_secret(self, index: int) -> bytes:

assert index <= self.START_INDEX, index

for i in range(0, 49):

bucket = self.buckets.get(str(i))

if bucket is None:

raise UnableToDeriveSecret()

try:

element = shachain_derive(bucket, index)

except UnableToDeriveSecret:

continue

return element.secret

raise UnableToDeriveSecret()

def __eq__(self, o):

return type(o) is RevocationStore and self.serialize() == o.serialize()

def __hash__(self):

""" BOLT-03 (where_to_put_secret) """

try:

return list(reversed(bin(index)[2:])).index("1")

except ValueError:

def get_prefix(index, pos):

mask = (1 << 64) - 1 - ((1 << pos) - 1)

return index & mask

from_index = element.index

zeros = count_trailing_zeros(from_index)

if from_index != get_prefix(to_index, zeros):

raise UnableToDeriveSecret("prefixes are different; index not derivable")

return ShachainElement(

get_per_commitment_secret_from_seed(element.secret, to_index, zeros),

"""Generate per commitment secret."""

per_commitment_secret = bytearray(seed)

for bitindex in range(bits - 1, -1, -1):

mask = 1 << bitindex

if i & mask:

per_commitment_secret[bitindex // 8] ^= 1 << (bitindex % 8)

per_commitment_secret = bytearray(sha256(per_commitment_secret))

bajts = bytes(per_commitment_secret)

assert type(secret) is int

basepoint_bytes = secret_to_pubkey(secret)

basepoint = secret + ecc.string_to_number(sha256(per_commitment_point + basepoint_bytes))

basepoint %= CURVE_ORDER

basepoint = ecc.ECPrivkey(basepoint_secret).get_public_key_bytes(compressed=True)

per_commitment_point = ecc.ECPrivkey(per_commitment_secret).get_public_key_bytes(compressed=True)

k1 = ecc.string_to_number(basepoint_secret) * ecc.string_to_number(sha256(basepoint + per_commitment_point))

k2 = ecc.string_to_number(per_commitment_secret) * ecc.string_to_number(sha256(per_commitment_point + basepoint))

sum = (k1 + k2) % ecc.CURVE_ORDER

assert type(amount_msat) is int

assert type(local_feerate) is int

assert type(revocationpubkey) is bytes

assert type(local_delayedpubkey) is bytes

script = bytes([opcodes.OP_IF]) \

+ bfh(push_script(bh2u(revocationpubkey))) \

+ bytes([opcodes.OP_ELSE]) \

+ bitcoin.add_number_to_script(to_self_delay) \

+ bytes([opcodes.OP_CHECKSEQUENCEVERIFY, opcodes.OP_DROP]) \

+ bfh(push_script(bh2u(local_delayedpubkey))) \

+ bytes([opcodes.OP_ENDIF, opcodes.OP_CHECKSIG])

p2wsh = bitcoin.redeem_script_to_address('p2wsh', bh2u(script))

weight = HTLC_SUCCESS_WEIGHT if success else HTLC_TIMEOUT_WEIGHT

fee = local_feerate * weight

fee = fee // 1000 * 1000

final_amount_sat = (amount_msat - fee) // 1000

assert final_amount_sat > 0, final_amount_sat

output = PartialTxOutput.from_address_and_value(p2wsh, final_amount_sat)

payment_preimage: bytes, witness_script: bytes) -> bytes:

assert type(remotehtlcsig) is bytes

assert type(localhtlcsig) is bytes

assert type(payment_preimage) is bytes

assert type(witness_script) is bytes

amount_msat: int, witness_script: str) -> List[PartialTxInput]:

assert type(htlc_output_txid) is str

assert type(htlc_output_index) is int

assert type(amount_msat) is int

assert type(witness_script) is str

txin = PartialTxInput(prevout=TxOutpoint(txid=bfh(htlc_output_txid), out_idx=htlc_output_index),

nsequence=0)

txin.witness_script = bfh(witness_script)

txin.script_sig = b''

txin._trusted_value_sats = amount_msat // 1000

c_inputs = [txin]

assert type(cltv_expiry) is int

c_outputs = [output]

tx = PartialTransaction.from_io(inputs, c_outputs, locktime=cltv_expiry, version=2)

local_htlcpubkey: bytes, payment_hash: bytes) -> bytes:

assert type(revocation_pubkey) is bytes

assert type(remote_htlcpubkey) is bytes

assert type(local_htlcpubkey) is bytes

assert type(payment_hash) is bytes

return bytes([opcodes.OP_DUP, opcodes.OP_HASH160]) + bfh(push_script(bh2u(bitcoin.hash_160(revocation_pubkey))))\

+ bytes([opcodes.OP_EQUAL, opcodes.OP_IF, opcodes.OP_CHECKSIG, opcodes.OP_ELSE]) \

+ bfh(push_script(bh2u(remote_htlcpubkey)))\

+ bytes([opcodes.OP_SWAP, opcodes.OP_SIZE]) + bitcoin.add_number_to_script(32) + bytes([opcodes.OP_EQUAL, opcodes.OP_NOTIF, opcodes.OP_DROP])\

+ bitcoin.add_number_to_script(2) + bytes([opcodes.OP_SWAP]) + bfh(push_script(bh2u(local_htlcpubkey))) + bitcoin.add_number_to_script(2)\

+ bytes([opcodes.OP_CHECKMULTISIG, opcodes.OP_ELSE, opcodes.OP_HASH160])\

local_htlcpubkey: bytes, payment_hash: bytes, cltv_expiry: int) -> bytes:

for i in [revocation_pubkey, remote_htlcpubkey, local_htlcpubkey, payment_hash]:

assert type(i) is bytes

assert type(cltv_expiry) is int

return bytes([opcodes.OP_DUP, opcodes.OP_HASH160]) \

+ bfh(push_script(bh2u(bitcoin.hash_160(revocation_pubkey)))) \

+ bytes([opcodes.OP_EQUAL, opcodes.OP_IF, opcodes.OP_CHECKSIG, opcodes.OP_ELSE]) \

+ bfh(push_script(bh2u(remote_htlcpubkey))) \

+ bytes([opcodes.OP_SWAP, opcodes.OP_SIZE]) \

+ bitcoin.add_number_to_script(32) \

+ bytes([opcodes.OP_EQUAL, opcodes.OP_IF, opcodes.OP_HASH160]) \

+ bfh(push_script(bh2u(crypto.ripemd(payment_hash)))) \

+ bytes([opcodes.OP_EQUALVERIFY]) \

+ bitcoin.add_number_to_script(2) \

+ bytes([opcodes.OP_SWAP]) \

+ bfh(push_script(bh2u(local_htlcpubkey))) \

+ bitcoin.add_number_to_script(2) \

+ bytes([opcodes.OP_CHECKMULTISIG, opcodes.OP_ELSE, opcodes.OP_DROP]) \

+ bitcoin.add_number_to_script(cltv_expiry) \

local_htlc_pubkey: bytes, payment_hash: bytes, cltv_expiry: Optional[int]) -> bytes:

if is_received_htlc:

return make_received_htlc(revocation_pubkey=remote_revocation_pubkey,

remote_htlcpubkey=remote_htlc_pubkey,

local_htlcpubkey=local_htlc_pubkey,

payment_hash=payment_hash,

cltv_expiry=cltv_expiry)

else:

return make_offered_htlc(revocation_pubkey=remote_revocation_pubkey,

remote_htlcpubkey=remote_htlc_pubkey,

local_htlcpubkey=local_htlc_pubkey,

Union[LocalConfig, RemoteConfig]]:

conf = chan.config[LOCAL] if for_us else chan.config[REMOTE]

other_conf = chan.config[LOCAL] if not for_us else chan.config[REMOTE]

htlc_direction: 'Direction', ctx: Transaction,

htlc: 'UpdateAddHtlc') -> Set[int]:

amount_msat, cltv_expiry, payment_hash = htlc.amount_msat, htlc.cltv_expiry, htlc.payment_hash

for_us = subject == LOCAL

conf, other_conf = get_ordered_channel_configs(chan=chan, for_us=for_us)

other_revocation_pubkey = derive_blinded_pubkey(other_conf.revocation_basepoint.pubkey, pcp)

other_htlc_pubkey = derive_pubkey(other_conf.htlc_basepoint.pubkey, pcp)

htlc_pubkey = derive_pubkey(conf.htlc_basepoint.pubkey, pcp)

preimage_script = make_htlc_output_witness_script(is_received_htlc=htlc_direction == RECEIVED,

remote_revocation_pubkey=other_revocation_pubkey,

remote_htlc_pubkey=other_htlc_pubkey,

local_htlc_pubkey=htlc_pubkey,

payment_hash=payment_hash,

cltv_expiry=cltv_expiry)

htlc_address = redeem_script_to_address('p2wsh', bh2u(preimage_script))

candidates = ctx.get_output_idxs_from_address(htlc_address)

return {output_idx for output_idx in candidates

subject: 'HTLCOwner', ctn: int) -> Dict[Tuple['Direction', 'UpdateAddHtlc'], Tuple[int, int]]:

"""Returns a dict from (htlc_dir, htlc) to (ctx_output_idx, htlc_relative_idx)"""

htlc_to_ctx_output_idx_map = {} # type: Dict[Tuple[Direction, UpdateAddHtlc], int]

unclaimed_ctx_output_idxs = set(range(len(ctx.outputs())))

offered_htlcs = chan.included_htlcs(subject, SENT, ctn=ctn)

offered_htlcs.sort(key=lambda htlc: htlc.cltv_expiry)

received_htlcs = chan.included_htlcs(subject, RECEIVED, ctn=ctn)

received_htlcs.sort(key=lambda htlc: htlc.cltv_expiry)

for direction, htlcs in zip([SENT, RECEIVED], [offered_htlcs, received_htlcs]):

for htlc in htlcs:

cands = sorted(possible_output_idxs_of_htlc_in_ctx(chan=chan,

pcp=pcp,

subject=subject,

htlc_direction=direction,

ctx=ctx,

htlc=htlc))

for ctx_output_idx in cands:

if ctx_output_idx in unclaimed_ctx_output_idxs:

unclaimed_ctx_output_idxs.discard(ctx_output_idx)

htlc_to_ctx_output_idx_map[(direction, htlc)] = ctx_output_idx

break

# calc htlc_relative_idx

inverse_map = {ctx_output_idx: (direction, htlc)

for ((direction, htlc), ctx_output_idx) in htlc_to_ctx_output_idx_map.items()}

return {inverse_map[ctx_output_idx]: (ctx_output_idx, htlc_relative_idx)

htlc_direction: 'Direction', commit: Transaction, ctx_output_idx: int,

htlc: 'UpdateAddHtlc', name: str = None) -> Tuple[bytes, PartialTransaction]:

amount_msat, cltv_expiry, payment_hash = htlc.amount_msat, htlc.cltv_expiry, htlc.payment_hash

for_us = subject == LOCAL

conf, other_conf = get_ordered_channel_configs(chan=chan, for_us=for_us)

delayedpubkey = derive_pubkey(conf.delayed_basepoint.pubkey, pcp)

other_revocation_pubkey = derive_blinded_pubkey(other_conf.revocation_basepoint.pubkey, pcp)

other_htlc_pubkey = derive_pubkey(other_conf.htlc_basepoint.pubkey, pcp)

htlc_pubkey = derive_pubkey(conf.htlc_basepoint.pubkey, pcp)

# HTLC-success for the HTLC spending from a received HTLC output

# if we do not receive, and the commitment tx is not for us, they receive, so it is also an HTLC-success

is_htlc_success = htlc_direction == RECEIVED

witness_script_of_htlc_tx_output, htlc_tx_output = make_htlc_tx_output(

amount_msat = amount_msat,

local_feerate = chan.get_next_feerate(LOCAL if for_us else REMOTE),

revocationpubkey=other_revocation_pubkey,

local_delayedpubkey=delayedpubkey,

success = is_htlc_success,

to_self_delay = other_conf.to_self_delay)

preimage_script = make_htlc_output_witness_script(is_received_htlc=is_htlc_success,

remote_revocation_pubkey=other_revocation_pubkey,

remote_htlc_pubkey=other_htlc_pubkey,

local_htlc_pubkey=htlc_pubkey,

payment_hash=payment_hash,

cltv_expiry=cltv_expiry)

htlc_tx_inputs = make_htlc_tx_inputs(

commit.txid(), ctx_output_idx,

amount_msat=amount_msat,

witness_script=bh2u(preimage_script))

if is_htlc_success:

cltv_expiry = 0

htlc_tx = make_htlc_tx(cltv_expiry=cltv_expiry, inputs=htlc_tx_inputs, output=htlc_tx_output)

funding_pos: int, funding_txid: str, funding_sat: int) -> PartialTxInput:

pubkeys = sorted([bh2u(local_funding_pubkey), bh2u(remote_funding_pubkey)])

# commitment tx input

prevout = TxOutpoint(txid=bfh(funding_txid), out_idx=funding_pos)

c_input = PartialTxInput(prevout=prevout)

c_input.script_type = 'p2wsh'

c_input.pubkeys = [bfh(pk) for pk in pubkeys]

c_input.num_sig = 2

c_input._trusted_value_sats = funding_sat

LOCAL = 1

REMOTE = -LOCAL

def inverted(self):

local_script: str, remote_script: str, htlcs: List[ScriptHtlc], dust_limit_sat: int) -> Tuple[List[PartialTxOutput], List[PartialTxOutput]]:

to_local_amt = local_amount_msat - fees_per_participant[LOCAL]

to_local = PartialTxOutput(scriptpubkey=bfh(local_script), value=to_local_amt // 1000)

to_remote_amt = remote_amount_msat - fees_per_participant[REMOTE]

to_remote = PartialTxOutput(scriptpubkey=bfh(remote_script), value=to_remote_amt // 1000)

non_htlc_outputs = [to_local, to_remote]

htlc_outputs = []

for script, htlc in htlcs:

addr = bitcoin.redeem_script_to_address('p2wsh', bh2u(script))

htlc_outputs.append(PartialTxOutput(scriptpubkey=bfh(address_to_script(addr)),

value=htlc.amount_msat // 1000))

# trim outputs

c_outputs_filtered = list(filter(lambda x: x.value >= dust_limit_sat, non_htlc_outputs + htlc_outputs))

overall_weight = 500 + 172 * num_htlcs + 224

fee = feerate * overall_weight

fee = fee // 1000 * 1000

remote_payment_pubkey, funder_payment_basepoint,

fundee_payment_basepoint, revocation_pubkey,

delayed_pubkey, to_self_delay, funding_txid,

funding_pos, funding_sat, local_amount, remote_amount,

dust_limit_sat, fees_per_participant,

htlcs: List[ScriptHtlc]) -> PartialTransaction:

c_input = make_funding_input(local_funding_pubkey, remote_funding_pubkey,

funding_pos, funding_txid, funding_sat)

obs = get_obscured_ctn(ctn, funder_payment_basepoint, fundee_payment_basepoint)

locktime = (0x20 << 24) + (obs & 0xffffff)

sequence = (0x80 << 24) + (obs >> 24)

c_input.nsequence = sequence

c_inputs = [c_input]

# commitment tx outputs

local_address = make_commitment_output_to_local_address(revocation_pubkey, to_self_delay, delayed_pubkey)

remote_address = make_commitment_output_to_remote_address(remote_payment_pubkey)

# TODO trim htlc outputs here while also considering 2nd stage htlc transactions

# BOLT-03: "Transaction Input and Output Ordering

# Lexicographic ordering: see BIP69. In the case of identical HTLC outputs,

# the outputs are ordered in increasing cltv_expiry order."

# so we sort by cltv_expiry now; and the later BIP69-sort is assumed to be *stable*

htlcs = list(htlcs)

htlcs.sort(key=lambda x: x.htlc.cltv_expiry)

htlc_outputs, c_outputs_filtered = make_commitment_outputs(

fees_per_participant=fees_per_participant,

local_amount_msat=local_amount,

remote_amount_msat=remote_amount,

local_script=address_to_script(local_address),

remote_script=address_to_script(remote_address),

htlcs=htlcs,

dust_limit_sat=dust_limit_sat)

assert sum(x.value for x in c_outputs_filtered) <= funding_sat, (c_outputs_filtered, funding_sat)

# create commitment tx

tx = PartialTransaction.from_io(c_inputs, c_outputs_filtered, locktime=locktime, version=2)

revocation_pubkey: bytes, to_self_delay: int, delayed_pubkey: bytes) -> bytes:

local_script = bytes([opcodes.OP_IF]) + bfh(push_script(bh2u(revocation_pubkey))) + bytes([opcodes.OP_ELSE]) + bitcoin.add_number_to_script(to_self_delay) \

+ bytes([opcodes.OP_CHECKSEQUENCEVERIFY, opcodes.OP_DROP]) + bfh(push_script(bh2u(delayed_pubkey))) + bytes([opcodes.OP_ENDIF, opcodes.OP_CHECKSIG])

tx.sign({bh2u(local_config.multisig_key.pubkey): (local_config.multisig_key.privkey, True)})

sig = tx.inputs()[0].part_sigs[local_config.multisig_key.pubkey]

sig_64 = sig_string_from_der_sig(sig[:-1])

fundee_payment_basepoint: bytes) -> int:

tx.deserialize()

locktime = tx.locktime

sequence = tx.inputs()[txin_index].nsequence

obs = ((sequence & 0xffffff) << 24) + (locktime & 0xffffff)

funder_conf = chan.config[LOCAL] if chan.constraints.is_initiator else chan.config[REMOTE]

fundee_conf = chan.config[LOCAL] if not chan.constraints.is_initiator else chan.config[REMOTE]

return extract_ctn_from_tx(tx, txin_index=0,

funder_payment_basepoint=funder_conf.payment_basepoint.pubkey,

OPTION_DATA_LOSS_PROTECT_REQ = 1 << 0

OPTION_DATA_LOSS_PROTECT_OPT = 1 << 1

INITIAL_ROUTING_SYNC = 1 << 3

OPTION_UPFRONT_SHUTDOWN_SCRIPT_REQ = 1 << 4

OPTION_UPFRONT_SHUTDOWN_SCRIPT_OPT = 1 << 5

GOSSIP_QUERIES_REQ = 1 << 6

"""Ln Feature flags are assigned in pairs, one even, one odd. See BOLT-09.

if flag_bit % 2 == 0:

return flag_bit + 1

else:

def __init__(self, host: str, port: int, pubkey: bytes):

assert isinstance(host, str), repr(host)

assert isinstance(port, int), repr(port)

assert isinstance(pubkey, bytes), repr(pubkey)

try:

net_addr = NetAddress(host, port) # this validates host and port

except Exception as e:

raise ValueError(f"cannot construct LNPeerAddr: invalid host or port (host={host}, port={port})") from e

# note: not validating pubkey as it would be too expensive:

# if not ECPubkey.is_pubkey_bytes(pubkey): raise ValueError()

self.host = host

self.port = port

self.pubkey = pubkey

self._net_addr_str = str(net_addr)

def __str__(self):

return '{}@{}'.format(self.pubkey.hex(), self.net_addr_str())

def __repr__(self):

return f'<LNPeerAddr host={self.host} port={self.port} pubkey={self.pubkey.hex()}>'

def net_addr_str(self) -> str:

return self._net_addr_str

def __eq__(self, other):

if not isinstance(other, LNPeerAddr):

return False

return (self.host == other.host

and self.port == other.port

and self.pubkey == other.pubkey)

def __ne__(self, other):

return not (self == other)

def __hash__(self):

hrp, data_5bits = segwit_addr.bech32_decode(bech32_pubkey)

if hrp != 'ln':

raise Exception('unexpected hrp: {}'.format(hrp))

data_8bits = segwit_addr.convertbits(data_5bits, 5, 8, False)

# pad with zeroes

COMPRESSED_PUBKEY_LENGTH = 33

data_8bits = data_8bits + ((COMPRESSED_PUBKEY_LENGTH - len(data_8bits)) * [0])

funding_txid: str, funding_pos: int, funding_sat: int,

outputs: List[PartialTxOutput]) -> PartialTransaction:

c_input = make_funding_input(local_funding_pubkey, remote_funding_pubkey,

funding_pos, funding_txid, funding_sat)

c_input.nsequence = 0xFFFF_FFFF

tx = PartialTransaction.from_io([c_input], outputs, locktime=0, version=2)

ipv6 = re.compile(r'\[(?P<host>[:0-9]+)\](?P<port>:\d+)?$')

other = re.compile(r'(?P<host>[^:]+)(?P<port>:\d+)?$')

m = ipv6.match(host_port)

if not m:

m = other.match(host_port)

if not m:

raise ConnStringFormatError(_('Connection strings must be in <node_pubkey>@<host>:<port> format'))

host = m.group('host')

if m.group('port'):

port = m.group('port')[1:]

else:

port = '9735'

try:

int(port)

except ValueError:

raise ConnStringFormatError(_('Port number must be decimal'))

rest = None

try:

# connection string?

nodeid_hex, rest = connect_contents.split("@", 1)

except ValueError:

try:

# invoice?

invoice = lndecode(connect_contents)

nodeid_bytes = invoice.pubkey.serialize()

nodeid_hex = bh2u(nodeid_bytes)

except:

# node id as hex?

nodeid_hex = connect_contents

if rest == '':

raise ConnStringFormatError(_('At least a hostname must be supplied after the at symbol.'))

try:

node_id = bfh(nodeid_hex)

assert len(node_id) == 33, len(node_id)

except:

raise ConnStringFormatError(_('Invalid node ID, must be 33 bytes and hexadecimal'))

MULTISIG = 0

REVOCATION_BASE = 1

HTLC_BASE = 2

PAYMENT_BASE = 3

DELAY_BASE = 4

REVOCATION_ROOT = 5

def __repr__(self):

return f"<ShortChannelID: {format_short_channel_id(self)}>"

def __str__(self):

return format_short_channel_id(self)

@classmethod

def from_components(cls, block_height: int, tx_pos_in_block: int, output_index: int) -> 'ShortChannelID':

bh = block_height.to_bytes(3, byteorder='big')

tpos = tx_pos_in_block.to_bytes(3, byteorder='big')

oi = output_index.to_bytes(2, byteorder='big')

return ShortChannelID(bh + tpos + oi)

@classmethod

def normalize(cls, data: Union[None, str, bytes, 'ShortChannelID']) -> Optional['ShortChannelID']:

if isinstance(data, ShortChannelID) or data is None:

return data

if isinstance(data, str):

return ShortChannelID.fromhex(data)

if isinstance(data, bytes):

return ShortChannelID(data)

@property

def block_height(self) -> int:

return int.from_bytes(self[:3], byteorder='big')

@property

def txpos(self) -> int:

return int.from_bytes(self[3:6], byteorder='big')

@property

def output_index(self) -> int:

if not short_channel_id:

return _('Not yet available')

return str(int.from_bytes(short_channel_id[:3], 'big')) \

+ 'x' + str(int.from_bytes(short_channel_id[3:6], 'big')) \

# note: typing.NamedTuple cannot be used because we are overriding __new__

__slots__ = ()

def __new__(cls, *args, **kwargs):

# if you pass a hex-string as payment_hash, it is decoded to bytes.

# Bytes can't be saved to disk, so we save hex-strings.

if len(args) > 0:

args = list(args)

if type(args[1]) is str:

args[1] = bfh(args[1])

return super().__new__(cls, *args)

if type(kwargs['payment_hash']) is str:

kwargs['payment_hash'] = bfh(kwargs['payment_hash'])

if len(args) < 4 and 'htlc_id' not in kwargs:

kwargs['htlc_id'] = None