feat: add Filters for accept/deny IP filtering (fixes #117)

README.rst

@@ -226,6 +226,23 @@ Multiaddr supports DNS-based address resolution using the DNSADDR protocol. This
 
 For comprehensive examples including bootstrap node resolution, protocol comparison, and py-libp2p integration, see the `DNS examples <https://github.com/multiformats/py-multiaddr/tree/master/examples/dns>`_ in the examples directory.
 
+IP Filtering
+------------
+
+Accept/deny IP ranges for multiaddrs, similar to go-multiaddr ``Filters``:
+
+
+.. code-block:: python
+
+    from multiaddr import Action, Filters, Multiaddr
+
+    filters = Filters()
+    filters.add_filter("10.0.0.0/8", Action.DENY)
+    assert filters.addr_blocked(Multiaddr("/ip4/10.1.2.3/tcp/80"))
+    assert not filters.addr_blocked(Multiaddr("/ip4/8.8.8.8/tcp/53"))
+
+See ``examples/filters/filters_example.py`` for a printable demo.
+
 Thin Waist Address Validation
 -----------------------------
 

docs/examples.rst

@@ -117,9 +117,6 @@ This example shows:
    :language: python
    :caption: examples/tag_only/tag_only_examples.py
 
-Resolver Utility Examples
---------------------------
-
 WireGuard (``wg``)
 ------------------
 
@@ -134,6 +131,24 @@ This example shows:
    :language: python
    :caption: examples/wg/wg_examples.py
 
+IP Filters
+----------
+
+The `examples/filters/` directory demonstrates accept/deny IP filtering for multiaddrs.
+
+This example shows:
+
+* Creating a ``Filters`` set with a default action
+* Adding deny rules for private ranges
+* Checking whether addresses are blocked
+
+.. literalinclude:: ../examples/filters/filters_example.py
+   :language: python
+   :caption: examples/filters/filters_example.py
+
+Resolver Utility Examples
+--------------------------
+
 The `examples/resolver_utils/` directory demonstrates the utility functions ported from go-multiaddr-dns for working with DNS-based multiaddr resolution.
 
 This example shows:
@@ -180,4 +195,7 @@ All examples can be run directly with Python:
     # WireGuard examples
     python examples/wg/wg_examples.py
 
+    # IP Filters examples
+    python examples/filters/filters_example.py
+
 Note: Some examples require network connectivity and may take a few seconds to complete due to DNS resolution.

examples/filters/filters_example.py

@@ -0,0 +1,30 @@
+"""
+IP Filters example (accept/deny).
+
+Usage:
+    python examples/filters/filters_example.py
+"""
+
+from multiaddr import Action, Filters, Multiaddr
+
+
+def main() -> None:
+    print("=== Default ACCEPT with private range DENY ===")
+    filters = Filters()
+    filters.add_filter("10.0.0.0/8", Action.DENY)
+    filters.add_filter("192.168.0.0/16", Action.DENY)
+
+    samples = [
+        "/ip4/8.8.8.8/tcp/53",
+        "/ip4/10.0.0.5/tcp/80",
+        "/ip4/192.168.1.10/tcp/443",
+        "/unix/var/run/docker.sock",
+    ]
+    for addr in samples:
+        ma = Multiaddr(addr)
+        blocked = filters.addr_blocked(ma)
+        print(f"{addr} -> blocked={blocked}")
+
+
+if __name__ == "__main__":
+    main()

multiaddr/__init__.py

@@ -9,6 +9,7 @@
     ResolutionError,
     StringParseError,
 )
+from .filters import Action, Filters
 from .multiaddr import Multiaddr
 from .protocols import (
     P_DNS,
@@ -69,7 +70,9 @@
     "P_TCP",
     "P_UDP",
     "REGISTRY",
+    "Action",
     "BinaryParseError",
+    "Filters",
     "Multiaddr",
     "ParseError",
     "Protocol",

multiaddr/filters.py

@@ -0,0 +1,86 @@
+"""Accept/deny IP filtering for multiaddrs (go-multiaddr Filters parity)."""
+
+from __future__ import annotations
+
+import ipaddress
+from enum import Enum
+
+from .multiaddr import Multiaddr
+from .protocols import P_IP4, P_IP6
+
+
+class Action(Enum):
+    """Filter action applied to a matching network."""
+
+    ACCEPT = 1
+    DENY = 2
+
+
+def _extract_ip(ma: Multiaddr) -> ipaddress.IPv4Address | ipaddress.IPv6Address | None:
+    for proto in ma.protocols():
+        if proto.code in (P_IP4, P_IP6):
+            value = ma.value_for_protocol(proto.code)
+            if value is None:
+                return None
+            try:
+                return ipaddress.ip_address(value)
+            except ValueError:
+                return None
+    return None
+
+
+class Filters:
+    """Collection of accept/deny IP network rules.
+
+    The last matching filter wins. If no filter matches, ``default_action`` applies.
+    Non-IP multiaddrs are treated according to ``default_action``.
+    """
+
+    def __init__(self, default_action: Action = Action.ACCEPT) -> None:
+        self.default_action = default_action
+        self._filters: list[tuple[ipaddress.IPv4Network | ipaddress.IPv6Network, Action]] = []
+
+    def add_filter(
+        self,
+        network: str | ipaddress.IPv4Network | ipaddress.IPv6Network,
+        action: Action,
+    ) -> None:
+        """Add or replace a filter for the given network."""
+        if isinstance(network, str):
+            net: ipaddress.IPv4Network | ipaddress.IPv6Network = ipaddress.ip_network(
+                network, strict=False
+            )
+        else:
+            net = network
+
+        for idx, (existing, _) in enumerate(self._filters):
+            if existing == net:
+                self._filters[idx] = (net, action)
+                return
+        self._filters.append((net, action))
+
+    def remove_literal(self, network: str | ipaddress.IPv4Network | ipaddress.IPv6Network) -> bool:
+        """Remove the filter for an exact network match. Returns whether something was removed."""
+        if isinstance(network, str):
+            net: ipaddress.IPv4Network | ipaddress.IPv6Network = ipaddress.ip_network(
+                network, strict=False
+            )
+        else:
+            net = network
+        for idx, (existing, _) in enumerate(self._filters):
+            if existing == net:
+                del self._filters[idx]
+                return True
+        return False
+
+    def addr_blocked(self, ma: Multiaddr) -> bool:
+        """Return True if the multiaddr should be denied."""
+        ip = _extract_ip(ma)
+        if ip is None:
+            return self.default_action == Action.DENY
+
+        action = self.default_action
+        for network, filter_action in self._filters:
+            if ip in network:
+                action = filter_action
+        return action == Action.DENY

newsfragments/117.feature.rst

@@ -0,0 +1 @@
+Add ``Filters`` / ``Action`` for accept/deny IP filtering (go-multiaddr parity).

tests/test_filters.py

@@ -0,0 +1,55 @@
+import ipaddress
+
+from multiaddr import Action, Filters, Multiaddr
+
+
+def test_default_accept_allows_all():
+    filters = Filters()
+    assert filters.addr_blocked(Multiaddr("/ip4/1.2.3.4/tcp/80")) is False
+
+
+def test_default_deny_blocks_all():
+    filters = Filters(default_action=Action.DENY)
+    assert filters.addr_blocked(Multiaddr("/ip4/1.2.3.4/tcp/80")) is True
+
+
+def test_deny_network_blocks_matching_ip():
+    filters = Filters()
+    filters.add_filter("10.0.0.0/8", Action.DENY)
+    assert filters.addr_blocked(Multiaddr("/ip4/10.1.2.3/tcp/80")) is True
+    assert filters.addr_blocked(Multiaddr("/ip4/11.0.0.1/tcp/80")) is False
+
+
+def test_last_matching_filter_wins():
+    filters = Filters(default_action=Action.DENY)
+    filters.add_filter("10.0.0.0/8", Action.ACCEPT)
+    filters.add_filter("10.0.0.0/16", Action.DENY)
+    assert filters.addr_blocked(Multiaddr("/ip4/10.0.1.1/tcp/1")) is True
+    assert filters.addr_blocked(Multiaddr("/ip4/10.1.0.1/tcp/1")) is False
+
+
+def test_remove_literal():
+    filters = Filters()
+    filters.add_filter("192.168.0.0/16", Action.DENY)
+    assert filters.remove_literal("192.168.0.0/16") is True
+    assert filters.addr_blocked(Multiaddr("/ip4/192.168.1.1/tcp/80")) is False
+    assert filters.remove_literal("192.168.0.0/16") is False
+
+
+def test_add_filter_replaces_same_network():
+    filters = Filters()
+    filters.add_filter(ipaddress.ip_network("127.0.0.0/8"), Action.DENY)
+    filters.add_filter("127.0.0.0/8", Action.ACCEPT)
+    assert filters.addr_blocked(Multiaddr("/ip4/127.0.0.1/tcp/80")) is False
+
+
+def test_non_ip_uses_default_action():
+    filters = Filters(default_action=Action.DENY)
+    assert filters.addr_blocked(Multiaddr("/unix/tmp/socket")) is True
+
+
+def test_ipv6_filter():
+    filters = Filters()
+    filters.add_filter("fe80::/10", Action.DENY)
+    assert filters.addr_blocked(Multiaddr("/ip6/fe80::1/tcp/80")) is True
+    assert filters.addr_blocked(Multiaddr("/ip6/2001:db8::1/tcp/80")) is False

tests/test_package_exports.py

@@ -1,6 +1,8 @@
 import multiaddr
 
 EXPECTED_EXPORTS = {
+    "Action",
+    "Filters",
     "PROTOCOLS",
     "P_DNS",
     "P_DNS4",