Merge pull request #1329 from qstokkink/add_peer_addr_freeze

Add peer address freeze

ipv8/peer.py

@@ -82,6 +82,11 @@ def __init__(self, key: Key | bytes, address: Address | None = None, intro: bool
         :param intro: is this peer suggested to us (otherwise it contacted us)
         """
         if not isinstance(key, Key):
+            if key.startswith((b"-----BEGIN EC PRIVATE KEY-----", b"LibNaCLSK:")):
+                msg = ("You attempted to initialize a Peer with PRIVATE key material."
+                       " This is normally ONLY done for your own Peer instance and internally handled by IPv8."
+                       " If this is truly what you want to do, initialize Peer with a PrivateKey subclass.")
+                raise ValueError(msg)
             self.key: Key = default_eccrypto.key_from_public_bin(key)
         else:
             self.key = cast(Key, key)
@@ -97,6 +102,11 @@ def __init__(self, key: Key | bytes, address: Address | None = None, intro: bool
         self.pings: deque = deque(maxlen=5)
         self.new_style_intro = False
 
+        self.address_frozen = False
+        """
+        Set this to True if you want to avoid this Peer's address being updated.
+        """
+
     @property
     def addresses(self) -> dict[type[Address], Address]:
         """
@@ -137,6 +147,8 @@ def add_address(self, value: Any) -> None:  # noqa: ANN401
          - Adding instances A(1), B(2) leads to addresses {A: A(1), B: B(2)}
          - Adding instances A(1), B(2), A(3) leads to addresses {A: A(3), B: B(2)}
         """
+        if self.address_frozen:
+            return
         self._addresses[value.__class__] = value
         self._update_preferred_address()
 
@@ -221,4 +233,4 @@ def __str__(self) -> str:
         """
         Represent this peer as a human-readable string.
         """
-        return 'Peer<%s:%d, %s>' % (self.address[0], self.address[1], b64encode(self.mid).decode('utf-8'))
+        return f"Peer<{self.address[0]}:{self.address[1]}, {b64encode(self.mid).decode()}>"

ipv8/test/test_peer.py

@@ -159,6 +159,18 @@ def test_set_address_addv6(self) -> None:
 
         self.assertEqual(peer.address, address)
 
+    def test_set_address_frozen(self) -> None:
+        """
+        Check if the add_address does not update frozen addresses.
+        """
+        old_address = ("6.7.8.9", 0)
+        peer = Peer(TestPeer.test_key, address=old_address)
+        peer.address_frozen = True
+
+        peer.add_address(UDPv4Address("1.2.3.4", 5))
+
+        self.assertEqual(peer.address, old_address)
+
     def test_address_order1(self) -> None:
         """
         Check if IPv6 is preferred over IPv4 (append out-of-order).