Merge pull request #10312 from SomberNight/202511_pr10230_1

SomberNight · web-flow · commit ee80065bb8ff · 2025-11-18T16:41:19.000Z
lnonion: immutable OnionPacket and OnionHopsDataSingle
diff --git a/electrum/lnmsg.py b/electrum/lnmsg.py
@@ -1,7 +1,8 @@
 import os
 import csv
 import io
-from typing import Callable, Tuple, Any, Dict, List, Sequence, Union, Optional
+from typing import Callable, Tuple, Any, Dict, List, Sequence, Union, Optional, Mapping
+from types import MappingProxyType
 from collections import OrderedDict
 
 from .lnutil import OnionFailureCodeMetaFlag
@@ -289,7 +290,7 @@ def _write_tlv_record(*, fd: io.BytesIO, tlv_type: int, tlv_val: bytes) -> None:
     _write_primitive_field(fd=fd, field_type="byte", count=tlv_len, value=tlv_val)
 
 
-def _resolve_field_count(field_count_str: str, *, vars_dict: dict, allow_any=False) -> Union[int, str]:
+def _resolve_field_count(field_count_str: str, *, vars_dict: Mapping, allow_any=False) -> Union[int, str]:
     """Returns an evaluated field count, typically an int.
     If allow_any is True, the return value can be a str with value=="...".
     """
@@ -403,7 +404,7 @@ def write_field(
             fd: io.BytesIO,
             field_type: str,
             count: Union[int, str],
-            value: Union[List[Dict[str, Any]], Dict[str, Any]]
+            value: Union[Sequence[Mapping[str, Any]], Mapping[str, Any]],
     ) -> None:
         assert fd
 
@@ -421,10 +422,10 @@ def write_field(
             return
 
         if count == 1:
-            assert isinstance(value, dict) or isinstance(value, list)
-            values = [value] if isinstance(value, dict) else value
+            assert isinstance(value, (MappingProxyType, dict)) or isinstance(value, (list, tuple)), type(value)
+            values = [value] if isinstance(value, (MappingProxyType, dict)) else value
         else:
-            assert isinstance(value, list), f'{field_type=}, expected value of type list for {count=}'
+            assert isinstance(value, (tuple, list)), f'{field_type=}, expected value of type list/tuple for {count=}'
             values = value
 
         if count == '...':
diff --git a/electrum/lnonion.py b/electrum/lnonion.py
@@ -25,8 +25,10 @@
 
 import io
 import hashlib
-from typing import Sequence, List, Tuple, NamedTuple, TYPE_CHECKING, Dict, Any, Optional, Union
+from typing import Sequence, List, Tuple, NamedTuple, TYPE_CHECKING, Dict, Any, Optional, Union, Mapping
 from enum import IntEnum
+from dataclasses import dataclass, field, replace
+from types import MappingProxyType
 
 import electrum_ecc as ecc
 
@@ -53,18 +55,22 @@ class InvalidOnionPubkey(Exception): pass
 class InvalidPayloadSize(Exception): pass
 
 
-class OnionHopsDataSingle:  # called HopData in lnd
+@dataclass(frozen=True, kw_only=True)
+class OnionHopsDataSingle:
+    payload: Mapping = field(default_factory=lambda: MappingProxyType({}))
+    hmac: Optional[bytes] = None
+    tlv_stream_name: str = 'payload'
+    blind_fields: Mapping = field(default_factory=lambda: MappingProxyType({}))
+    _raw_bytes_payload: Optional[bytes] = None
 
-    def __init__(self, *, payload: dict = None, tlv_stream_name: str = 'payload', blind_fields: dict = None):
-        if payload is None:
-            payload = {}
-        self.payload = payload
-        self.hmac = None
-        self.tlv_stream_name = tlv_stream_name
-        if blind_fields is None:
-            blind_fields = {}
-        self.blind_fields = blind_fields
-        self._raw_bytes_payload = None  # used in unit tests
+    def __post_init__(self):
+        # make all fields immutable recursively
+        object.__setattr__(self, 'payload', util.make_object_immutable(self.payload))
+        object.__setattr__(self, 'blind_fields', util.make_object_immutable(self.blind_fields))
+        assert isinstance(self.payload, MappingProxyType)
+        assert isinstance(self.blind_fields, MappingProxyType)
+        assert isinstance(self.tlv_stream_name, str)
+        assert (isinstance(self.hmac, bytes) and len(self.hmac) == PER_HOP_HMAC_SIZE) or self.hmac is None
 
     def to_bytes(self) -> bytes:
         hmac_ = self.hmac if self.hmac is not None else bytes(PER_HOP_HMAC_SIZE)
@@ -101,32 +107,35 @@ def from_fd(cls, fd: io.BytesIO, *, tlv_stream_name: str = 'payload') -> 'OnionH
             hop_payload = fd.read(hop_payload_length)
             if hop_payload_length != len(hop_payload):
                 raise Exception(f"unexpected EOF")
-            ret = OnionHopsDataSingle(tlv_stream_name=tlv_stream_name)
-            ret.payload = OnionWireSerializer.read_tlv_stream(fd=io.BytesIO(hop_payload),
-                                                              tlv_stream_name=tlv_stream_name)
-            ret.hmac = fd.read(PER_HOP_HMAC_SIZE)
-            assert len(ret.hmac) == PER_HOP_HMAC_SIZE
+            payload = OnionWireSerializer.read_tlv_stream(fd=io.BytesIO(hop_payload),
+                                                          tlv_stream_name=tlv_stream_name)
+            ret = OnionHopsDataSingle(
+                tlv_stream_name=tlv_stream_name,
+                payload=payload,
+                hmac=fd.read(PER_HOP_HMAC_SIZE)
+            )
             return ret
 
     def __repr__(self):
-        return f"<OnionHopsDataSingle. payload={self.payload}. hmac={self.hmac}>"
+        return f"<OnionHopsDataSingle. {self.payload=}. {self.hmac=}>"
 
 
+@dataclass(frozen=True, kw_only=True)
 class OnionPacket:
-
-    def __init__(self, *, public_key: bytes, hops_data: bytes, hmac: bytes, version: int = 0):
-        assert len(public_key) == 33
-        assert len(hops_data) in [HOPS_DATA_SIZE, TRAMPOLINE_HOPS_DATA_SIZE, ONION_MESSAGE_LARGE_SIZE]
-        assert len(hmac) == PER_HOP_HMAC_SIZE
-        self.version = version
-        self.public_key = public_key
-        self.hops_data = hops_data  # also called RoutingInfo in bolt-04
-        self.hmac = hmac
-        if not ecc.ECPubkey.is_pubkey_bytes(public_key):
+    public_key: bytes
+    hops_data: bytes  # also called RoutingInfo in bolt-04
+    hmac: bytes
+    version: int = 0
+    # for debugging our own onions:
+    _debug_hops_data: Optional[Sequence[OnionHopsDataSingle]] = None
+    _debug_route: Optional['LNPaymentRoute'] = None
+
+    def __post_init__(self):
+        assert len(self.public_key) == 33
+        assert len(self.hops_data) in [HOPS_DATA_SIZE, TRAMPOLINE_HOPS_DATA_SIZE, ONION_MESSAGE_LARGE_SIZE]
+        assert len(self.hmac) == PER_HOP_HMAC_SIZE
+        if not ecc.ECPubkey.is_pubkey_bytes(self.public_key):
             raise InvalidOnionPubkey()
-        # for debugging our own onions:
-        self._debug_hops_data = None  # type: Optional[Sequence[OnionHopsDataSingle]]
-        self._debug_route = None      # type: Optional[LNPaymentRoute]
 
     def to_bytes(self) -> bytes:
         ret = bytes([self.version])
@@ -138,7 +147,7 @@ def to_bytes(self) -> bytes:
         return ret
 
     @classmethod
-    def from_bytes(cls, b: bytes):
+    def from_bytes(cls, b: bytes) -> 'OnionPacket':
         if len(b) - 66 not in [HOPS_DATA_SIZE, TRAMPOLINE_HOPS_DATA_SIZE, ONION_MESSAGE_LARGE_SIZE]:
             raise Exception('unexpected length {}'.format(len(b)))
         return OnionPacket(
@@ -187,7 +196,7 @@ def get_blinded_node_id(node_id: bytes, shared_secret: bytes):
 def new_onion_packet(
     payment_path_pubkeys: Sequence[bytes],
     session_key: bytes,
-    hops_data: Sequence[OnionHopsDataSingle],
+    hops_data: List[OnionHopsDataSingle],
     *,
     associated_data: bytes = b'',
     trampoline: bool = False,
@@ -226,7 +235,7 @@ def new_onion_packet(
     for i in range(num_hops-1, -1, -1):
         rho_key = get_bolt04_onion_key(b'rho', hop_shared_secrets[i])
         mu_key = get_bolt04_onion_key(b'mu', hop_shared_secrets[i])
-        hops_data[i].hmac = next_hmac
+        hops_data[i] = replace(hops_data[i], hmac=next_hmac)
         stream_bytes = generate_cipher_stream(rho_key, data_size)
         hop_data_bytes = hops_data[i].to_bytes()
         mix_header = mix_header[:-len(hop_data_bytes)]
diff --git a/electrum/lnworker.py b/electrum/lnworker.py
@@ -12,6 +12,7 @@
     Optional, Sequence, Tuple, List, Set, Dict, TYPE_CHECKING, NamedTuple, Mapping, Any, Iterable, AsyncGenerator,
     Callable, Awaitable
 )
+from types import MappingProxyType
 import threading
 import socket
 from functools import partial
@@ -3723,13 +3724,14 @@ def create_onion_for_route(
         # if we are forwarding a trampoline payment, add trampoline onion
         if trampoline_onion:
             self.logger.info(f'adding trampoline onion to final payload')
-            trampoline_payload = hops_data[-1].payload
+            trampoline_payload = dict(hops_data[-1].payload)
             trampoline_payload["trampoline_onion_packet"] = {
                 "version": trampoline_onion.version,
                 "public_key": trampoline_onion.public_key,
                 "hops_data": trampoline_onion.hops_data,
                 "hmac": trampoline_onion.hmac
             }
+            hops_data[-1] = dataclasses.replace(hops_data[-1], payload=trampoline_payload)
             if t_hops_data := trampoline_onion._debug_hops_data:  # None if trampoline-forwarding
                 t_route = trampoline_onion._debug_route
                 assert t_route is not None
diff --git a/electrum/onion_message.py b/electrum/onion_message.py
@@ -27,9 +27,11 @@
 import os
 import threading
 import time
+import dataclasses
 from random import random
+from types import MappingProxyType
 
-from typing import TYPE_CHECKING, Optional, Sequence, NamedTuple
+from typing import TYPE_CHECKING, Optional, Sequence, NamedTuple, List
 
 import electrum_ecc as ecc
 
@@ -139,7 +141,7 @@ def is_onion_message_node(node_id: bytes, node_info: Optional['NodeInfo']) -> bo
 
 
 def encrypt_onionmsg_tlv_hops_data(
-        hops_data: Sequence[OnionHopsDataSingle],
+        hops_data: List[OnionHopsDataSingle],
         hop_shared_secrets: Sequence[bytes]
 ) -> None:
     """encrypt unencrypted onionmsg_tlv.encrypted_recipient_data for hops with blind_fields"""
@@ -148,7 +150,9 @@ def encrypt_onionmsg_tlv_hops_data(
         if hops_data[i].tlv_stream_name == 'onionmsg_tlv' and 'encrypted_recipient_data' not in hops_data[i].payload:
             # construct encrypted_recipient_data from blind_fields
             encrypted_recipient_data = encrypt_onionmsg_data_tlv(shared_secret=hop_shared_secrets[i], **hops_data[i].blind_fields)
-            hops_data[i].payload['encrypted_recipient_data'] = {'encrypted_recipient_data': encrypted_recipient_data}
+            new_payload = dict(hops_data[i].payload)
+            new_payload['encrypted_recipient_data'] = {'encrypted_recipient_data': encrypted_recipient_data}
+            hops_data[i] = dataclasses.replace(hops_data[i], payload=new_payload)
 
 
 def create_onion_message_route_to(lnwallet: 'LNWallet', node_id: bytes) -> Sequence[PathEdge]:
@@ -280,7 +284,7 @@ def send_onion_message_to(
                         hops_data = [
                             OnionHopsDataSingle(
                                 tlv_stream_name='onionmsg_tlv',
-                                blind_fields={'next_node_id': {'node_id': x.end_node}}
+                                blind_fields={'next_node_id': {'node_id': x.end_node}},
                             ) for x in path[:-1]
                         ]
 
@@ -290,7 +294,7 @@ def send_onion_message_to(
                             blind_fields={
                                 'next_node_id': {'node_id': introduction_point},
                                 'next_path_key_override': {'path_key': blinded_path['first_path_key']},
-                            }
+                            },
                         )
                         hops_data.append(final_hop_pre_ip)
 
@@ -299,9 +303,11 @@ def send_onion_message_to(
                             encrypted_recipient_data = encrypt_onionmsg_data_tlv(
                                 shared_secret=hop_shared_secrets[i],
                                 **hops_data[i].blind_fields)
-                            hops_data[i].payload['encrypted_recipient_data'] = {
+                            payload = dict(hops_data[i].payload)
+                            payload['encrypted_recipient_data'] = {
                                 'encrypted_recipient_data': encrypted_recipient_data
                             }
+                            hops_data[i] = dataclasses.replace(hops_data[i], payload=payload)
 
                         path_key = ecc.ECPrivkey(session_key).get_public_key_bytes()
 
@@ -345,13 +351,13 @@ def send_onion_message_to(
             hops_data = [
                 OnionHopsDataSingle(
                     tlv_stream_name='onionmsg_tlv',
-                    blind_fields={'next_node_id': {'node_id': x.end_node}}
+                    blind_fields={'next_node_id': {'node_id': x.end_node}},
                 ) for x in path[1:]
             ]
 
         final_hop = OnionHopsDataSingle(
             tlv_stream_name='onionmsg_tlv',
-            payload=destination_payload
+            payload=destination_payload,
         )
 
         hops_data.append(final_hop)
diff --git a/electrum/trampoline.py b/electrum/trampoline.py
@@ -1,7 +1,9 @@
 import io
 import os
 import random
+import dataclasses
 from typing import Mapping, Tuple, Optional, List, Iterable, Sequence, Set, Any
+from types import MappingProxyType
 
 from .lnutil import LnFeatures, PaymentFeeBudget, FeeBudgetExceeded
 from .lnonion import (
@@ -302,12 +304,12 @@ def create_trampoline_onion(
     for i in range(num_hops):
         route_edge = route[i]
         assert route_edge.is_trampoline()
-        payload = hops_data[i].payload
+        payload = dict(hops_data[i].payload)
         if i < num_hops - 1:
             payload.pop('short_channel_id')
             next_edge = route[i+1]
             assert next_edge.is_trampoline()
-            hops_data[i].payload["outgoing_node_id"] = {"outgoing_node_id": next_edge.node_id}
+            payload["outgoing_node_id"] = {"outgoing_node_id": next_edge.node_id}
         # only for final
         if i == num_hops - 1:
             payload["payment_data"] = {
@@ -322,10 +324,11 @@ def create_trampoline_onion(
                 "payment_secret": payment_secret,
                 "total_msat": total_msat
             }
+        hops_data[i] = dataclasses.replace(hops_data[i], payload=payload)
 
     if (index := routing_info_payload_index) is not None:
         # fill the remaining payload space with available routing hints (r_tags)
-        payload: dict = hops_data[index].payload
+        payload = dict(hops_data[index].payload)
         # try different r_tag order on each attempt
         invoice_routing_info = random_shuffled_copy(route[index].invoice_routing_info)
         remaining_payload_space = TRAMPOLINE_HOPS_DATA_SIZE \
@@ -341,12 +344,16 @@ def create_trampoline_onion(
             remaining_payload_space -= r_tag_size
         # add the chosen r_tags to the payload
         payload["invoice_routing_info"] = {"invoice_routing_info": b''.join(routing_info_to_use)}
+        hops_data[index] = dataclasses.replace(hops_data[index], payload=payload)
         _logger.debug(f"Using {len(routing_info_to_use)} of {len(invoice_routing_info)} r_tags")
 
     trampoline_session_key = os.urandom(32)
     trampoline_onion = new_onion_packet(payment_path_pubkeys, trampoline_session_key, hops_data, associated_data=payment_hash, trampoline=True)
-    trampoline_onion._debug_hops_data = hops_data
-    trampoline_onion._debug_route = route
+    trampoline_onion = dataclasses.replace(
+        trampoline_onion,
+        _debug_hops_data=hops_data,
+        _debug_route=route,
+    )
     return trampoline_onion, amount_msat, cltv_abs
 
 
diff --git a/electrum/util.py b/electrum/util.py
@@ -32,6 +32,7 @@
     NamedTuple, Union, TYPE_CHECKING, Tuple, Optional, Callable, Any, Sequence, Dict, Generic, TypeVar, List, Iterable,
     Set, Awaitable
 )
+from types import MappingProxyType
 from datetime import datetime, timezone, timedelta
 import decimal
 from decimal import Decimal
@@ -1875,6 +1876,21 @@ def __setitem__(self, key, *args, **kwargs):
         return ret
 
 
+def make_object_immutable(obj):
+    """Makes the passed object immutable recursively."""
+    allowed_types = (
+        dict, MappingProxyType, list, tuple, set, frozenset, str, int, float, bool, bytes, type(None)
+    )
+    assert isinstance(obj, allowed_types), f"{type(obj)=} cannot be made immutable"
+    if isinstance(obj, (dict, MappingProxyType)):
+        return MappingProxyType({k: make_object_immutable(v) for k, v in obj.items()})
+    elif isinstance(obj, (list, tuple)):
+        return tuple(make_object_immutable(item) for item in obj)
+    elif isinstance(obj, (set, frozenset)):
+        return frozenset(make_object_immutable(item) for item in obj)
+    return obj
+
+
 def multisig_type(wallet_type):
     """If wallet_type is mofn multi-sig, return [m, n],
     otherwise return None."""
diff --git a/tests/test_lnrouter.py b/tests/test_lnrouter.py
diff --git a/tests/test_onion_message.py b/tests/test_onion_message.py
