lightning: rate-limit incoming channels

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

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."""

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

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
+

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())

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,