diff --git a/README.rst b/README.rst index 2f97a4b..4b0c942 100644 --- a/README.rst +++ b/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 `_ 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 ----------------------------- diff --git a/docs/examples.rst b/docs/examples.rst index 6aef32f..753c408 100644 --- a/docs/examples.rst +++ b/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. diff --git a/examples/filters/filters_example.py b/examples/filters/filters_example.py new file mode 100644 index 0000000..877dffa --- /dev/null +++ b/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() diff --git a/multiaddr/__init__.py b/multiaddr/__init__.py index 456004f..016bc00 100755 --- a/multiaddr/__init__.py +++ b/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", diff --git a/multiaddr/filters.py b/multiaddr/filters.py new file mode 100644 index 0000000..13327e1 --- /dev/null +++ b/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 diff --git a/newsfragments/117.feature.rst b/newsfragments/117.feature.rst new file mode 100644 index 0000000..6c0e971 --- /dev/null +++ b/newsfragments/117.feature.rst @@ -0,0 +1 @@ +Add ``Filters`` / ``Action`` for accept/deny IP filtering (go-multiaddr parity). diff --git a/tests/test_filters.py b/tests/test_filters.py new file mode 100644 index 0000000..30f3eea --- /dev/null +++ b/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 diff --git a/tests/test_package_exports.py b/tests/test_package_exports.py index 04a1928..f9ecba0 100644 --- a/tests/test_package_exports.py +++ b/tests/test_package_exports.py @@ -1,6 +1,8 @@ import multiaddr EXPECTED_EXPORTS = { + "Action", + "Filters", "PROTOCOLS", "P_DNS", "P_DNS4",