diff --git a/electrum/lnpeer.py b/electrum/lnpeer.py index 4b88f87d6e7..1008375719b 100644 --- a/electrum/lnpeer.py +++ b/electrum/lnpeer.py @@ -66,6 +66,9 @@ LN_P2P_NETWORK_TIMEOUT = 20 +class PeerWarning(Exception): pass + + class Peer(Logger, EventListener): # note: in general this class is NOT thread-safe. Most methods are assumed to be running on asyncio thread. @@ -295,6 +298,8 @@ def on_warning(self, payload): self.logger.info(f"remote peer sent warning [DO NOT TRUST THIS MESSAGE]: " f"{error_text_bytes_to_safe_str(err_bytes, max_len=None)}. chan_id={chan_id.hex()}. " f"{is_known_chan_id=}") + if is_known_chan_id: + self.ordered_message_queues[chan_id].put_nowait((None, {'warning': err_bytes})) def on_error(self, payload): chan_id = payload.get("channel_id") @@ -394,11 +399,12 @@ def on_pong(self, payload): async def wait_for_message(self, expected_name: str, channel_id: bytes): q = self.ordered_message_queues[channel_id] name, payload = await util.wait_for2(q.get(), LN_P2P_NETWORK_TIMEOUT) - # raise exceptions for errors, so that the caller sees them - if (err_bytes := payload.get("error")) is not None: + # raise exceptions for warnings and errors, so that the caller sees them + if (err_bytes := payload.get("error", payload.get("warning"))) is not None: err_text = error_text_bytes_to_safe_str(err_bytes) - raise GracefulDisconnect( - f"remote peer sent error [DO NOT TRUST THIS MESSAGE]: {err_text}") + msg_type = "error" if "error" in payload else "warning" + exc_str = f"remote peer sent {msg_type} [DO NOT TRUST THIS MESSAGE]: {err_text}" + raise GracefulDisconnect(exc_str) if "error" in payload else PeerWarning(exc_str) if name != expected_name: raise Exception(f"Received unexpected '{name}'") return payload @@ -1252,7 +1258,14 @@ def create_channel_storage( return chan_dict @non_blocking_msg_handler - async def on_open_channel(self, payload): + async def on_open_channel(self, payload: dict): + try: + with self.lnworker.incoming_channel_rate_limiter.rate_limit(self.pubkey): + await self._on_open_channel(payload) + except lnutil.IncomingChannelRateLimiter.ChannelOpeningRateLimited as e: + self.send_warning(payload["temporary_channel_id"], message=str(e), close_connection=False) + + async def _on_open_channel(self, payload): """Implements the channel acceptance flow. <- open_channel message diff --git a/electrum/lnutil.py b/electrum/lnutil.py index 8baff6182ae..811061fa914 100644 --- a/electrum/lnutil.py +++ b/electrum/lnutil.py @@ -3,10 +3,12 @@ # file LICENCE or http://www.opensource.org/licenses/mit-license.php from enum import IntFlag, IntEnum import enum -from typing import NamedTuple, List, Tuple, Mapping, Optional, TYPE_CHECKING, Union, Dict, Set, Sequence, FrozenSet +from typing import NamedTuple, List, Tuple, Mapping, Optional, TYPE_CHECKING, Union, Dict, Set, Sequence, FrozenSet, Iterable import sys import time from functools import lru_cache +from collections import defaultdict +from contextlib import contextmanager import electrum_ecc as ecc from electrum_ecc import CURVE_ORDER, ecdsa_sig64_from_der_sig @@ -36,6 +38,7 @@ from .lnchannel import Channel, AbstractChannel from .lnrouter import LNPaymentRoute from .lnonion import OnionRoutingFailure + from .lnworker import LNPeerManager from .simple_config import SimpleConfig @@ -1779,6 +1782,64 @@ def from_payload(cls, payload: dict) -> Optional['GossipForwardingMessage']: return cls(msg, scid, timestamp, sender_node_id) +class IncomingChannelRateLimiter: + """ + Rate limiting for new incoming channels. + If there previously was an active channel with the given peer before we allow it to bypass the global pending limit. + """ + MAX_UNFUNDED_INCOMING_CHANNELS_PER_PEER = 2 + MAX_UNFUNDED_INCOMING_CHANNELS_GLOBAL = 20 + + class ChannelOpeningRateLimited(Exception): pass + + def __init__(self, lnpeermgr: 'LNPeerManager'): + self._lnpeermgr = lnpeermgr + # stores the pending opening flows as they are not yet part of lnworker.channels + self._pending_channel_openings = defaultdict(int) # type: dict[bytes, int] + + @contextmanager + def rate_limit(self, node_id: bytes): + self._try_acquire(node_id) + try: + yield + finally: + self._release(node_id) + + def _try_acquire(self, node_id: bytes): + if self._count_unfunded_channels_with_peer(node_id) >= self.MAX_UNFUNDED_INCOMING_CHANNELS_PER_PEER: + raise self.ChannelOpeningRateLimited(f"per-peer rate limit ({self.MAX_UNFUNDED_INCOMING_CHANNELS_PER_PEER}) exceeded") + if not self._is_trusted_peer(node_id) \ + and self._count_unfunded_channels_global() >= self.MAX_UNFUNDED_INCOMING_CHANNELS_GLOBAL: + raise self.ChannelOpeningRateLimited(f"global rate limit ({self.MAX_UNFUNDED_INCOMING_CHANNELS_GLOBAL}) exceeded") + self._pending_channel_openings[node_id] += 1 + + def _release(self, node_id: bytes): + self._pending_channel_openings[node_id] -= 1 + if self._pending_channel_openings[node_id] <= 0: + del self._pending_channel_openings[node_id] + + @staticmethod + def _count_unfunded_channels(channels: Iterable['Channel']) -> int: + # count zeroconf as unfunded, flag gets dropped after 3 confirmations by LNWatcher. + # zeroconf channels aren't unbounded here as the LSP might even have an incentive + # to DOS us so we are unable to broadcast a fraud proof/preimage onchain? + return sum(1 for c in channels if (not c.is_funded() or c.is_zeroconf()) and not c.is_initiator()) + + def _is_trusted_peer(self, node_id: bytes) -> bool: + return any(c.get_latest_ctn(HTLCOwner.LOCAL) > 0 for c in self._lnpeermgr.channels_for_peer(node_id).values()) + + def _count_unfunded_channels_with_peer(self, node_id: bytes) -> int: + channels_with_peer = self._lnpeermgr.channels_for_peer(node_id).values() + count_unfunded_channels = self._count_unfunded_channels(channels_with_peer) + count_pending_channel_openings = self._pending_channel_openings.get(node_id, 0) + return count_unfunded_channels + count_pending_channel_openings + + def _count_unfunded_channels_global(self) -> int: + global_channels = self._lnpeermgr._lnwallet_or_lngossip.channels.values() + count_global_channels = self._count_unfunded_channels(global_channels) + return count_global_channels + sum(self._pending_channel_openings.values()) + + def list_enabled_ln_feature_bits(features: int) -> tuple[int, ...]: """Returns a list of enabled feature bits. If both opt and req are set, only req will be included in the result.""" diff --git a/electrum/lnworker.py b/electrum/lnworker.py index 6e65cf62685..68b5ea19f1b 100644 --- a/electrum/lnworker.py +++ b/electrum/lnworker.py @@ -1048,6 +1048,7 @@ def __init__(self, wallet: 'Abstract_Wallet', xprv, *, features: LnFeatures = No for channel_id, c in random_shuffled_copy(channels.items()): self._channels[bfh(channel_id)] = chan = Channel(c, lnworker=self) self.wallet.set_reserved_addresses_for_chan(chan, reserved=True) + self.incoming_channel_rate_limiter = lnutil.IncomingChannelRateLimiter(self.lnpeermgr) self._channel_backups = {} # type: Dict[bytes, ChannelBackup] # order is important: imported should overwrite onchain diff --git a/tests/lnhelpers.py b/tests/lnhelpers.py index a36f9335d96..094c50ebab3 100644 --- a/tests/lnhelpers.py +++ b/tests/lnhelpers.py @@ -23,7 +23,9 @@ from electrum.lnworker import LNWallet, PaySession from electrum.simple_config import SimpleConfig from electrum.fee_policy import FeeTimeEstimates, FEE_ETA_TARGETS -from electrum.wallet import Standard_Wallet +from electrum.wallet import Standard_Wallet, Abstract_Wallet +from electrum.transaction import PartialTransaction, PartialTxInput, PartialTxOutput, TxOutpoint +from electrum.address_synchronizer import TX_HEIGHT_UNCONFIRMED from . import restore_wallet_from_text__for_unittest @@ -51,6 +53,9 @@ def __init__(self, *, config: SimpleConfig): def get_local_height(self): return self.blockchain().height() + def get_server_height(self): + return self.blockchain().height() + def blockchain(self): return self._blockchain @@ -232,6 +237,26 @@ def prepare_lnwallets(elec_test_case: 'ElectrumTestCase', graph_definition) -> M return workers +def fund_wallet(wallet: 'Abstract_Wallet', *, value_sat: int) -> PartialTransaction: + """Give a mock wallet a spendable on-chain UTXO.""" + txin = PartialTxInput(prevout=TxOutpoint(txid=bytes(32), out_idx=0)) + txin._trusted_value_sats = value_sat + txin.script_type = 'p2wpkh' + txout = PartialTxOutput.from_address_and_value(wallet.get_receiving_address(), value_sat) + tx = PartialTransaction.from_io([txin], [txout], version=2) + wallet.adb.receive_tx_callback(tx, tx_height=TX_HEIGHT_UNCONFIRMED) + return tx + + +def force_state_transition(chanA: Channel, chanB: Channel) -> None: + chanB.receive_new_commitment(*chanA.sign_next_commitment()) + rev = chanB.revoke_current_commitment() + bob_sig, bob_htlc_sigs = chanB.sign_next_commitment() + chanA.receive_revocation(rev) + chanA.receive_new_commitment(bob_sig, bob_htlc_sigs) + chanB.receive_revocation(chanA.revoke_current_commitment()) + + def prepare_chans_and_peers_in_graph( elec_test_case: 'ElectrumTestCase', graph_definition=None, @@ -530,3 +555,9 @@ def create_test_channels( assert alice.channel_id == bob.channel_id return alice, bob + + +class PaymentDone(Exception): pass +class PaymentTimeout(Exception): pass +class SuccessfulTest(Exception): pass + diff --git a/tests/test_lnchannel.py b/tests/test_lnchannel.py index 9eedcd54f8d..c3acde1b3c6 100644 --- a/tests/test_lnchannel.py +++ b/tests/test_lnchannel.py @@ -42,7 +42,7 @@ from electrum.lnchannel import ChannelState, Channel from . import ElectrumTestCase -from .lnhelpers import create_test_channels +from .lnhelpers import create_test_channels, force_state_transition one_bitcoin_in_msat = bitcoin.COIN * 1000 @@ -1014,10 +1014,3 @@ class TestDustNoAnchors(TestDust): TEST_ANCHOR_CHANNELS = False -def force_state_transition(chanA: Channel, chanB: Channel) -> None: - chanB.receive_new_commitment(*chanA.sign_next_commitment()) - rev = chanB.revoke_current_commitment() - bob_sig, bob_htlc_sigs = chanB.sign_next_commitment() - chanA.receive_revocation(rev) - chanA.receive_new_commitment(bob_sig, bob_htlc_sigs) - chanB.receive_revocation(chanA.revoke_current_commitment()) diff --git a/tests/test_lnpeer.py b/tests/test_lnpeer.py index 19e02311552..c19eba28b67 100644 --- a/tests/test_lnpeer.py +++ b/tests/test_lnpeer.py @@ -1,62 +1,47 @@ import asyncio import dataclasses -import shutil import copy -import tempfile from decimal import Decimal import os -from contextlib import contextmanager -from collections import defaultdict import logging -import concurrent -from concurrent import futures -from functools import lru_cache from unittest import mock -from typing import Iterable, NamedTuple, Tuple, List, Dict, Sequence, Mapping +from typing import Tuple, List, Mapping from types import MappingProxyType import time import statistics -from aiorpcx import timeout_after, TaskTimeout -from electrum_ecc import ECPrivkey import electrum_ecc as ecc import electrum import electrum.trampoline from electrum import bitcoin from electrum import util -from electrum import constants -from electrum import bip32 -from electrum.network import Network, ProxySettings -from electrum import simple_config, lnutil +from electrum import lnutil from electrum.bolt11 import encode_bolt11_invoice, BOLT11Addr, decode_bolt11_invoice from electrum.bitcoin import COIN, sha256 from electrum.transaction import Transaction -from electrum.util import NetworkRetryManager, bfh, OldTaskGroup, EventListener, InvoiceError +from electrum.util import OldTaskGroup, InvoiceError from electrum.lnpeer import Peer from electrum.lntransport import LNPeerAddr from electrum.crypto import privkey_to_pubkey -from electrum.lnutil import Keypair, PaymentFailure, LnFeatures, HTLCOwner, PaymentFeeBudget, RECEIVED +from electrum.lnutil import PaymentFailure, LnFeatures, HTLCOwner, RECEIVED from electrum.lnchannel import ChannelState, PeerState, Channel -from electrum.lnrouter import LNPathFinder, PathEdge, LNPathInconsistent +from electrum.lnrouter import PathEdge, LNPathInconsistent from electrum.channel_db import ChannelDB, InvalidGossipMsg -from electrum.lnworker import LNWallet, NoPathFound, SentHtlcInfo, PaySession, LNPeerManager +from electrum.lnworker import LNWallet, NoPathFound, SentHtlcInfo from electrum.lnmsg import encode_msg, decode_msg from electrum import lnmsg -from electrum.logging import console_stderr_handler, Logger +from electrum.logging import console_stderr_handler from electrum.lnworker import PaymentInfo from electrum.lnonion import OnionFailureCode, OnionRoutingFailure, OnionHopsDataSingle, OnionPacket from electrum.lnutil import LOCAL, REMOTE, UpdateAddHtlc, RecvMPPResolution, RevocationStore from electrum.invoices import PR_PAID, PR_UNPAID, Invoice, LN_EXPIRY_NEVER from electrum.interface import GracefulDisconnect from electrum.simple_config import SimpleConfig -from electrum.fee_policy import FeeTimeEstimates, FEE_ETA_TARGETS from electrum.mpp_split import split_amount_normal -from electrum.wallet import Abstract_Wallet, Standard_Wallet -from .test_bitcoin import needs_test_with_all_chacha20_implementations -from . import ElectrumTestCase, restore_wallet_from_text__for_unittest, lnhelpers -from .lnhelpers import Graph, MockLNWallet, create_test_channels +from . import ElectrumTestCase, lnhelpers +from .lnhelpers import Graph, MockLNWallet, create_test_channels, SuccessfulTest, PaymentDone, PaymentTimeout high_fee_channel = { @@ -174,11 +159,6 @@ } -class PaymentDone(Exception): pass -class PaymentTimeout(Exception): pass -class SuccessfulTest(Exception): pass - - def inject_chan_into_gossipdb( *, channel_db: ChannelDB, diff --git a/tests/test_lnwallet.py b/tests/test_lnwallet.py index 4c3919f0774..ee97ae12945 100644 --- a/tests/test_lnwallet.py +++ b/tests/test_lnwallet.py @@ -6,20 +6,22 @@ from typing import Optional from electrum.address_synchronizer import TX_HEIGHT_LOCAL -from electrum import bitcoin +from electrum import lnutil import electrum.trampoline from electrum.lnutil import RECEIVED, MIN_FINAL_CLTV_DELTA_ACCEPTED, serialize_htlc_key, LnFeatures, HTLCOwner from electrum.logging import console_stderr_handler from electrum.lntransport import LNPeerAddr from electrum.invoices import LN_EXPIRY_NEVER, PR_UNPAID -from electrum.lnpeer import Peer +from electrum.lnpeer import Peer, PeerWarning from electrum.lnchannel import Channel, ChannelState from electrum.lnonion import OnionPacket, OnionRoutingFailure from electrum.crypto import sha256 from electrum.simple_config import SimpleConfig +from electrum.util import OldTaskGroup +from electrum.invoices import Invoice from . import ElectrumTestCase, lnhelpers -from .lnhelpers import create_test_channels +from .lnhelpers import create_test_channels, SuccessfulTest, force_state_transition class TestLNWallet(ElectrumTestCase): @@ -474,3 +476,67 @@ async def test_rebalance_channels(self): chan1.set_frozen_for_sending(True) # shouldn't matter, this channel will receive success, log = await alice.rebalance_channels(chan0, chan1, amount_msat=150_000_000) self.assertTrue(success, msg=log) + + async def test_channel_open_rate_limit(self): + alice = self.create_mock_lnwallet(name="alice") + bob = self.create_mock_lnwallet(name="bob") + carol = self.create_mock_lnwallet(name="carol") + dave = self.create_mock_lnwallet(name="dave") + alice.config.LIGHTNING_USE_RECOVERABLE_CHANNELS = False + bob.config.LIGHTNING_USE_RECOVERABLE_CHANNELS = False + alice.incoming_channel_rate_limiter.MAX_UNFUNDED_INCOMING_CHANNELS_GLOBAL = 3 + for lnw in (alice, bob, carol, dave): + lnhelpers.fund_wallet(lnw.wallet, value_sat=10_000_000) + + peer_loops = [] + def connect(w1, w2): + t1, t2 = lnhelpers.transport_pair(w1.node_keypair, w2.node_keypair, w1.name, w2.name) + peer_1 = lnhelpers.PeerInTests(w1, w2.node_keypair.pubkey, t1) + peer_2 = lnhelpers.PeerInTests(w2, w1.node_keypair.pubkey, t2) + w1.lnpeermgr._peers[w2.node_keypair.pubkey] = peer_1 + w2.lnpeermgr._peers[w1.node_keypair.pubkey] = peer_2 + peer_loops.extend([peer_1, peer_2]) + return peer_1, peer_2 + + bob_alice_peer, alice_bob_peer = connect(bob, alice) + carol_alice_peer, _ = connect(carol, alice) + dave_alice_peer, _ = connect(dave, alice) + + async def open_channel(initiator_peer, funding_sat=lnutil.MIN_FUNDING_SAT): + initiator_peer.lnworker.network._blockchain._height += 1 + return await initiator_peer.lnworker.open_channel_with_peer(peer=initiator_peer, funding_sat=funding_sat) + + async def run_test(): + chan1, _ = await open_channel(bob_alice_peer) + chan2, _ = await open_channel(bob_alice_peer) + # test per-peer rate limit + with self.assertLogs('electrum', level='ERROR') as logs: + with self.assertRaises(PeerWarning): + await open_channel(bob_alice_peer) # bob exhausted his allowed budget + self.assertTrue(any("remote peer sent warning [DO NOT TRUST THIS MESSAGE]: 'per-peer rate limit" in msg for msg in logs.output)) + + chan3, _ = await open_channel(carol_alice_peer) + # test global rate limit + with self.assertLogs('electrum', level='ERROR') as logs: + with self.assertRaises(PeerWarning): + await open_channel(dave_alice_peer) # global limit is full, dave cannot open channel + self.assertTrue(any("remote peer sent warning [DO NOT TRUST THIS MESSAGE]: 'global rate limit" in msg for msg in logs.output)) + + # alice can still open channels to bob, exceeding her own rate-limit(s) + await open_channel(alice_bob_peer) + + # a known peer is allowed to exceed the global rate limit + # inject a confirmed channel into alice so she knows dave + graph_def = {"alice": {"channels": {"dave": [{"local_balance_msat": 10_000_000_000,"remote_balance_msat": 50_000_000}],}}, "dave": {}} + g = lnhelpers.prepare_chans_and_peers_in_graph(self, graph_def, workers={'alice': alice, 'dave': dave}) + # bump the ctn of 'alice->dave' so the channel is considered active by alice + force_state_transition(g.channels[('alice', 'dave')][0], g.channels[('dave', 'alice')][0]) + await open_channel(dave_alice_peer) # dave should now be able to open a channel as he is known to alice + + raise SuccessfulTest + + with self.assertRaises(SuccessfulTest): + async with OldTaskGroup() as group: + await group.spawn(asyncio.wait_for(run_test(), timeout=5)) + for peer in peer_loops: + await group.spawn(peer.main_loop())