lnmsg/lnonion: follow-up prev: "hide" MappingProxyType usage

electrum/lnmsg.py

@@ -1,7 +1,7 @@
 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
 
@@ -290,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: Union[dict, MappingProxyType], 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=="...".
     """
@@ -404,12 +404,7 @@ class LNSerializer:
             fd: io.BytesIO,
             field_type: str,
             count: Union[int, str],
-            value: Union[
+            value: Union[Sequence[Mapping[str, Any]], Mapping[str, Any]],
-                List[Union[MappingProxyType[str, Any], Dict[str, Any]]],
-                Tuple[Union[MappingProxyType[str, Any], Dict[str, Any]], ...],
-                Dict[str, Any],
-                MappingProxyType[str, Any],
-            ],
     ) -> None:
         assert fd
 

electrum/lnonion.py

@@ -25,7 +25,7 @@
 
 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
@@ -57,10 +57,10 @@ class InvalidPayloadSize(Exception): pass
 
 @dataclass(frozen=True, kw_only=True)
 class OnionHopsDataSingle:
-    payload: MappingProxyType = field(default_factory=lambda: MappingProxyType({}))
+    payload: Mapping = field(default_factory=lambda: MappingProxyType({}))
     hmac: Optional[bytes] = None
     tlv_stream_name: str = 'payload'
-    blind_fields: MappingProxyType = field(default_factory=lambda: MappingProxyType({}))
+    blind_fields: Mapping = field(default_factory=lambda: MappingProxyType({}))
     _raw_bytes_payload: Optional[bytes] = None
 
     def __post_init__(self):
@@ -111,7 +111,7 @@ class OnionHopsDataSingle:
                                                           tlv_stream_name=tlv_stream_name)
             ret = OnionHopsDataSingle(
                 tlv_stream_name=tlv_stream_name,
-                payload=MappingProxyType(payload),
+                payload=payload,
                 hmac=fd.read(PER_HOP_HMAC_SIZE)
             )
             return ret
@@ -303,7 +303,7 @@ def calc_hops_data_for_payment(
         "total_msat": total_msat,
         "amount_msat": amt
     }
-    hops_data = [OnionHopsDataSingle(payload=MappingProxyType(hop_payload))]
+    hops_data = [OnionHopsDataSingle(payload=hop_payload)]
     # payloads, backwards from last hop (but excluding the first edge):
     for edge_index in range(len(route) - 1, 0, -1):
         route_edge = route[edge_index]
@@ -313,7 +313,7 @@ def calc_hops_data_for_payment(
             "short_channel_id": {"short_channel_id": route_edge.short_channel_id},
         }
         hops_data.append(
-            OnionHopsDataSingle(payload=MappingProxyType(hop_payload)))
+            OnionHopsDataSingle(payload=hop_payload))
         amt += route_edge.fee_for_edge(amt)
         cltv_abs += route_edge.cltv_delta
     hops_data.reverse()

electrum/lnworker.py

@@ -3731,7 +3731,7 @@ class LNWallet(LNWorker):
                 "hops_data": trampoline_onion.hops_data,
                 "hmac": trampoline_onion.hmac
             }
-            hops_data[-1] = dataclasses.replace(hops_data[-1], payload=MappingProxyType(trampoline_payload))
+            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

electrum/onion_message.py

@@ -152,7 +152,7 @@ def encrypt_onionmsg_tlv_hops_data(
             encrypted_recipient_data = encrypt_onionmsg_data_tlv(shared_secret=hop_shared_secrets[i], **hops_data[i].blind_fields)
             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=MappingProxyType(new_payload))
+            hops_data[i] = dataclasses.replace(hops_data[i], payload=new_payload)
 
 
 def create_onion_message_route_to(lnwallet: 'LNWallet', node_id: bytes) -> Sequence[PathEdge]:
@@ -284,17 +284,17 @@ def send_onion_message_to(
                         hops_data = [
                             OnionHopsDataSingle(
                                 tlv_stream_name='onionmsg_tlv',
-                                blind_fields=MappingProxyType({'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 pre-ip, add next_path_key_override
                         final_hop_pre_ip = OnionHopsDataSingle(
                             tlv_stream_name='onionmsg_tlv',
-                            blind_fields=MappingProxyType({
+                            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)
 
@@ -307,7 +307,7 @@ def send_onion_message_to(
                             payload['encrypted_recipient_data'] = {
                                 'encrypted_recipient_data': encrypted_recipient_data
                             }
-                            hops_data[i] = dataclasses.replace(hops_data[i], payload=MappingProxyType(payload))
+                            hops_data[i] = dataclasses.replace(hops_data[i], payload=payload)
 
                         path_key = ecc.ECPrivkey(session_key).get_public_key_bytes()
 
@@ -320,7 +320,7 @@ def send_onion_message_to(
                 }
                 if i == len(remaining_blinded_path) - 1:  # final hop
                     payload.update(destination_payload)
-                hop = OnionHopsDataSingle(tlv_stream_name='onionmsg_tlv', payload=MappingProxyType(payload))
+                hop = OnionHopsDataSingle(tlv_stream_name='onionmsg_tlv', payload=payload)
                 hops_data.append(hop)
 
             payment_path_pubkeys = blinded_node_ids + blinded_path_blinded_ids
@@ -351,13 +351,13 @@ def send_onion_message_to(
             hops_data = [
                 OnionHopsDataSingle(
                     tlv_stream_name='onionmsg_tlv',
-                    blind_fields=MappingProxyType({'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=MappingProxyType(destination_payload),
+            payload=destination_payload,
         )
 
         hops_data.append(final_hop)

electrum/trampoline.py

@@ -324,7 +324,7 @@ def create_trampoline_onion(
                 "payment_secret": payment_secret,
                 "total_msat": total_msat
             }
-        hops_data[i] = dataclasses.replace(hops_data[i], payload=MappingProxyType(payload))
+        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)
@@ -344,7 +344,7 @@ 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=MappingProxyType(payload))
+        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)

tests/test_lnrouter.py

@@ -430,26 +430,26 @@ class Test_LNRouter(ElectrumTestCase):
         session_key = bfh('4141414141414141414141414141414141414141414141414141414141414141')
         associated_data = bfh('4242424242424242424242424242424242424242424242424242424242424242')
         hops_data = [
-            OnionHopsDataSingle(payload=MappingProxyType({
+            OnionHopsDataSingle(payload={
                 'amt_to_forward': {'amt_to_forward': 15000},
                 'outgoing_cltv_value': {'outgoing_cltv_value': 1500},
-                'short_channel_id': {'short_channel_id': bfh('0000000000000001')}})),
+                'short_channel_id': {'short_channel_id': bfh('0000000000000001')}}),
-            OnionHopsDataSingle(payload=MappingProxyType({
+            OnionHopsDataSingle(payload={
                 'amt_to_forward': {'amt_to_forward': 14000},
                 'outgoing_cltv_value': {'outgoing_cltv_value': 1400},
-                'short_channel_id': {'short_channel_id': bfh('0000000000000002')}})),
+                'short_channel_id': {'short_channel_id': bfh('0000000000000002')}}),
-            OnionHopsDataSingle(payload=MappingProxyType({
+            OnionHopsDataSingle(payload={
                 'amt_to_forward': {'amt_to_forward': 12500},
                 'outgoing_cltv_value': {'outgoing_cltv_value': 1250},
-                'short_channel_id': {'short_channel_id': bfh('0000000000000003')}})),
+                'short_channel_id': {'short_channel_id': bfh('0000000000000003')}}),
-            OnionHopsDataSingle(payload=MappingProxyType({
+            OnionHopsDataSingle(payload={
                 'amt_to_forward': {'amt_to_forward': 10000},
                 'outgoing_cltv_value': {'outgoing_cltv_value': 1000},
-                'short_channel_id': {'short_channel_id': bfh('0000000000000004')}})),
+                'short_channel_id': {'short_channel_id': bfh('0000000000000004')}}),
-            OnionHopsDataSingle(payload=MappingProxyType({
+            OnionHopsDataSingle(payload={
                 'amt_to_forward': {'amt_to_forward': 10000},
                 'outgoing_cltv_value': {'outgoing_cltv_value': 1000},
-                'payment_data': {'payment_secret': bfh('24a33562c54507a9334e79f0dc4f17d407e6d7c61f0e2f3d0d38599502f61704'), 'total_msat': 10000}})),
+                'payment_data': {'payment_secret': bfh('24a33562c54507a9334e79f0dc4f17d407e6d7c61f0e2f3d0d38599502f61704'), 'total_msat': 10000}}),
         ]
         packet = new_onion_packet(payment_path_pubkeys, session_key, hops_data, associated_data=associated_data)
         self.assertEqual(bfh('0002eec7245d6b7d2ccb30380bfbe2a3648cd7a942653f5aa340edcea1f283686619f7f3416a5aa36dc7eeb3ec6d421e9615471ab858ba970cd3cceb768b44e692be2f390c0b7fe70122abae84d7801db070dfb1638cd8d263072206dbed0234f6505e21e282abd8587124c572aad8de04610a136d6c71a7648c0ef66f1b3655d8a9eea1f92349132c93befbd6c37dbfc55615814ae09e4cbef721c01b487007811bbbfdc1fc7bd869aeb70eb08b4140ff5f501394b3653ada2a3b36a263535ea421d26818afb278df46abcec093305b715cac22b0b03645f8f4797cf2987b1bf4bfdd9ed8648ed42ed1a831fc36ccd45416a132580281ddac4e7470e4d2afd675baad9282ec6335403a73e1391427e330996c834db93848b4ae29dd975f678b2f5155ad6865ca23190725d4b7238fb44f0e3762dd59091b45c97d45df8164a15d9ca0329ec76f957b0a0e49ae372154620708df5c0fa991f0dd12b6bff1ebaf9e2376bb64bc24713f7c57da569bcd9c43a50c088416564b786a87d1f40936a051a3dbfe023bd867a5e66148b61cdd24a79f8c18682150e55aa6969ce9becf51f7c69e72deafcd0659f6be4f78463eaef8716e56615c77b3fbea8190806359909dcbec13c1592523b3d2985ec3e83d42cb7286a66a22f58704ddf6979ceb6883ab4ad8ac99d30251035189ffd514e03ce1576844513d66965d4adfc2523f4eee0dede229ab96303e31348c72bc0c8c816c666a904e5ccbabadf5a919720438f4a14dbd4a802f8d4b942f0ca8572f59644c9ac1912c8c8efefc4afa7f19e27411d46b7541c55985e28ce5cd7620b335fea51de55fa00ef977e8522181ad19e5e04f93bcfc83a36edd7e96fe48e846f2e54fe7a7090fe8e46ba72123e1cdee0667777c38c4930e50401074d8ab31a9717457fcefaa46323003af553bee2b49ea7f907eb2ff3301463e64a8c53975c853bbdd2956b9001b5ce1562264963fce84201daaf752de6df7ca31291226969c9851d1fc4ea88ca67d38c38587c2cdd8bc4d3f7bdf705497a1e054246f684554b3b8dfac43194f1eadec7f83b711e663b5645bde6d7f8cefb59758303599fed25c3b4d2e4499d439c915910dd283b3e7118320f1c6e7385009fbcb9ae79bab72a85e644182b4dafc0a173241f2ae68ae6a504f17f102da1e91de4548c7f5bc1c107354519077a4e83407f0d6a8f0975b4ac0c2c7b30637a998dda27b56b56245371296b816876b859677bcf3473a07e0f300e788fdd60c51b1626b46050b182457c6d716994847aaef667ca45b2cede550c92d336ff29ce6effd933b875f81381cda6e59e9727e728a58c0b3e74035beeeb639ab7463744322bf40138b81895e9a8e8850c9513782dc7a79f04380c216cb177951d8940d576486b887a232fcd382adcbd639e70af0c1a08bcf1405496606fce4645aef10d769dc0c010a8a433d8cd24d5943843a89cdbc8d16531db027b312ab2c03a7f1fdb7f2bcb128639c49e86705c948137fd42d0080fda4be4e9ee812057c7974acbf0162730d3b647b355ac1a5adbb2993832eba443b7c9b5a0ae1fc00a6c0c2b0b65b9019690565739d6439bf602066a3a9bd9c67b83606de51792d25ae517cbbdf6e1827fa0e8b2b5c6023cbb1e9f0e10b786dc6fa154e282fd9c90b8d46ca685d0f4434760035073c92d131564b6845ef57457488add4f709073bbb41f5f31f8226904875a9fd9e1b7a2901e71426104d7a298a05af0d4ab549fbd69c539ebe64949a9b6088f16e2e4bc827c305cb8d64536b8364dc3d5f7519c3b431faa38b47a958cf0c6dcabf205280693abf747c262f44cd6ffa11b32fc38d4f9c3631d554d8b57389f1390ac65c06357843ee6d9f289bb054ef25de45c5149c090fe6ddcd4095696dcc9a5cfc09c8bdfd5b83a153'),

tests/test_onion_message.py

@@ -74,34 +74,34 @@ class TestOnionMessage(ElectrumTestCase):
         hops_data = [
             OnionHopsDataSingle(
                 tlv_stream_name='onionmsg_tlv',
-                blind_fields=MappingProxyType({
+                blind_fields={
                     'next_node_id': {'node_id': bfh(ALICE_TLVS['next_node_id'])},
                     'next_path_key_override': {'path_key': bfh(ALICE_TLVS['next_path_key_override'])},
-                }
+                },
-            )),
+            ),
             OnionHopsDataSingle(
                 tlv_stream_name='onionmsg_tlv',
-                blind_fields=MappingProxyType({
+                blind_fields={
                     'next_node_id': {'node_id': bfh(BOB_TLVS['next_node_id'])},
                     'unknown_tag_561': {'data': bfh(BOB_TLVS['unknown_tag_561'])},
-                }
+                },
-            )),
+            ),
             OnionHopsDataSingle(
                 tlv_stream_name='onionmsg_tlv',
-                blind_fields=MappingProxyType({
+                blind_fields={
                     'padding': {'padding': bfh(CAROL_TLVS['padding'])},
                     'next_node_id': {'node_id': bfh(CAROL_TLVS['next_node_id'])},
-                }
+                },
-            )),
+            ),
             OnionHopsDataSingle(
                 tlv_stream_name='onionmsg_tlv',
-                payload=MappingProxyType({'message': {'text': bfh(test_vectors['onionmessage']['unknown_tag_1'])}}),
+                payload={'message': {'text': bfh(test_vectors['onionmessage']['unknown_tag_1'])}},
-                blind_fields=MappingProxyType({
+                blind_fields={
                     'padding': {'padding': bfh(DAVE_TLVS['padding'])},
                     'path_id': {'data': bfh(DAVE_TLVS['path_id'])},
                     'unknown_tag_65535': {'data': bfh(DAVE_TLVS['unknown_tag_65535'])},
-                }
+                },
-            ))
+            )
         ]
 
         encrypt_onionmsg_tlv_hops_data(hops_data, hop_shared_secrets)
@@ -119,11 +119,11 @@ class TestOnionMessage(ElectrumTestCase):
             return [
                 OnionHopsDataSingle(
                     tlv_stream_name='onionmsg_tlv',
-                    payload=MappingProxyType({'message': {'text': message.encode('utf-8')}}),
+                    payload={'message': {'text': message.encode('utf-8')}},
-                    blind_fields=MappingProxyType({
+                    blind_fields={
                         'path_id': {'data': bfh('deadbeefbadc0ffeedeadbeefbadc0ffeedeadbeefbadc0ffeedeadbeefbadc0')},
-                    }
+                    },
-                    ))
+                ),
             ]
         hops_data = hops_data_for_message('short_message')  # fit in HOPS_DATA_SIZE
         encrypt_onionmsg_tlv_hops_data(hops_data, hop_shared_secrets)
@@ -234,18 +234,18 @@ class TestOnionMessage(ElectrumTestCase):
         hops_data = [
             OnionHopsDataSingle(
                 tlv_stream_name='onionmsg_tlv',
-                blind_fields=MappingProxyType({
+                blind_fields={
                     'next_node_id': {'node_id': BOB_PUBKEY},
                     'next_path_key_override': {'path_key': bfh(ALICE_TLVS['next_path_key_override'])},
-                }
+                },
-            )),
+            ),
         ]
         # encrypt encrypted_data_tlv here
         for i in range(len(hops_data)):
             encrypted_recipient_data = encrypt_onionmsg_data_tlv(shared_secret=hop_shared_secrets[i], **hops_data[i].blind_fields)
             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=MappingProxyType(new_payload))
+            hops_data[i] = dataclasses.replace(hops_data[i], payload=new_payload)
 
         blinded_path_blinded_ids = []
         for i, x in enumerate(blinded_path_to_dave.get('path')):
@@ -257,7 +257,7 @@ class TestOnionMessage(ElectrumTestCase):
             hops_data.append(
                 OnionHopsDataSingle(
                     tlv_stream_name='onionmsg_tlv',
-                    payload=MappingProxyType(payload))
+                    payload=payload),
             )
         payment_path_pubkeys = blinded_node_ids + blinded_path_blinded_ids
         hop_shared_secrets, _ = get_shared_secrets_along_route(payment_path_pubkeys, SESSION_KEY)