Skip to content

feat: introduce InterfaceChoice.AllWithLoopback #1358

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 10 commits into from
5 changes: 5 additions & 0 deletions src/zeroconf/_core.py
Original file line number Diff line number Diff line change
Expand Up @@ -162,7 +162,12 @@ def __init__(
* `InterfaceChoice.All` is an alias for `InterfaceChoice.Default`
on Python versions before 3.8.

* `InterfaceChoice.AllWithLoopback` is the same as `InterfaceChoice.All`
on POSIX systems, but includes the loopback interfaces. This likely
only works on macOS/BSD.

Also listening on loopback (``::1``) doesn't work, use a real address.

:param ip_version: IP versions to support. If `choice` is a list, the default is detected
from it. Otherwise defaults to V4 only for backward compatibility.
:param apple_p2p: use AWDL interface (only macOS)
Expand Down
1 change: 0 additions & 1 deletion src/zeroconf/_dns.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,6 @@
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Union, cast

from ._exceptions import AbstractMethodException
from ._utils.net import _is_v6_address
from ._utils.time import current_time_millis
from .const import _CLASS_MASK, _CLASS_UNIQUE, _CLASSES, _TYPE_ANY, _TYPES

Expand Down
34 changes: 28 additions & 6 deletions src/zeroconf/_utils/net.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,8 +22,8 @@

import enum
import errno
import ipaddress
import socket
import ipaddress
import struct
import sys
from typing import Any, List, Optional, Sequence, Tuple, Union, cast
Expand All @@ -32,12 +32,14 @@

from .._logger import log
from ..const import _IPPROTO_IPV6, _MDNS_ADDR, _MDNS_ADDR6, _MDNS_PORT
from .ipaddress import _cached_ip_addresses


@enum.unique
class InterfaceChoice(enum.Enum):
Default = 1
All = 2
AllWithLoopback = 3


InterfacesType = Union[
Expand All @@ -62,10 +64,6 @@ class IPVersion(enum.Enum):
# utility functions


def _is_v6_address(addr: bytes) -> bool:
return len(addr) == 16


def _encode_address(address: str) -> bytes:
is_ipv6 = ":" in address
address_family = socket.AF_INET6 if is_ipv6 else socket.AF_INET
Expand Down Expand Up @@ -146,7 +144,11 @@ def ip6_addresses_to_indexes(
for iface in interfaces:
if isinstance(iface, int):
result.append((interface_index_to_ip6_address(adapters, iface), iface))
elif isinstance(iface, str) and ipaddress.ip_address(iface).version == 6:
elif (
isinstance(iface, str)
and (ip_address := _cached_ip_addresses(iface))
and ip_address.version == 6
):
result.append(ip6_to_address_and_index(adapters, iface))

return result
Expand All @@ -169,6 +171,26 @@ def normalize_interface_choice(
if ip_version != IPVersion.V6Only:
result.append("0.0.0.0")
elif choice is InterfaceChoice.All:
if ip_version != IPVersion.V4Only:
result.extend(
ip_tuple
for ip_tuple in get_all_addresses_v6()
if (ip_address := _cached_ip_addresses(ip_tuple[0][0]))
and not ip_address.is_loopback
)
if ip_version != IPVersion.V6Only:
result.extend(
ip
for ip in get_all_addresses()
if (ip_address := _cached_ip_addresses(ip))
and not ip_address.is_loopback
)
if not result:
raise RuntimeError(
"No interfaces to listen on, check that any interfaces have IP version %s"
% ip_version
)
elif choice is InterfaceChoice.AllWithLoopback:
if ip_version != IPVersion.V4Only:
result.extend(get_all_addresses_v6())
if ip_version != IPVersion.V6Only:
Expand Down
4 changes: 4 additions & 0 deletions src/zeroconf/asyncio.py
Original file line number Diff line number Diff line change
Expand Up @@ -161,6 +161,10 @@ def __init__(
* `InterfaceChoice.All` is an alias for `InterfaceChoice.Default`
on Python versions before 3.8.

* `InterfaceChoice.AllWithLoopback` is the same as `InterfaceChoice.All`
on POSIX systems, but includes the loopback interfaces. This likely
only works on macOS/BSD.

Also listening on loopback (``::1``) doesn't work, use a real address.
:param ip_version: IP versions to support. If `choice` is a list, the default is detected
from it. Otherwise defaults to V4 only for backward compatibility.
Expand Down
20 changes: 18 additions & 2 deletions tests/test_core.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,21 +63,29 @@ async def make_query():

class Framework(unittest.TestCase):
def test_launch_and_close(self):
rv = r.Zeroconf(interfaces=r.InterfaceChoice.AllWithLoopback)
rv.close()
rv = r.Zeroconf(interfaces=r.InterfaceChoice.All)
rv.close()
rv = r.Zeroconf(interfaces=r.InterfaceChoice.Default)
rv.close()

def test_launch_and_close_context_manager(self):
with r.Zeroconf(interfaces=r.InterfaceChoice.All) as rv:
with r.Zeroconf(interfaces=r.InterfaceChoice.AllWithLoopback) as rv:
assert rv.done is False
assert rv.done is True

with r.Zeroconf(interfaces=r.InterfaceChoice.All) as rv: # type: ignore[unreachable]
assert rv.done is False
assert rv.done is True

with r.Zeroconf(interfaces=r.InterfaceChoice.Default) as rv: # type: ignore[unreachable]
with r.Zeroconf(interfaces=r.InterfaceChoice.Default) as rv:
assert rv.done is False
assert rv.done is True

def test_launch_and_close_unicast(self):
rv = r.Zeroconf(interfaces=r.InterfaceChoice.AllWithLoopback, unicast=True)
rv.close()
rv = r.Zeroconf(interfaces=r.InterfaceChoice.All, unicast=True)
rv.close()
rv = r.Zeroconf(interfaces=r.InterfaceChoice.Default, unicast=True)
Expand All @@ -91,6 +99,10 @@ def test_close_multiple_times(self):
@unittest.skipIf(not has_working_ipv6(), "Requires IPv6")
@unittest.skipIf(os.environ.get("SKIP_IPV6"), "IPv6 tests disabled")
def test_launch_and_close_v4_v6(self):
rv = r.Zeroconf(
interfaces=r.InterfaceChoice.AllWithLoopback, ip_version=r.IPVersion.All
)
rv.close()
rv = r.Zeroconf(interfaces=r.InterfaceChoice.All, ip_version=r.IPVersion.All)
rv.close()
rv = r.Zeroconf(
Expand All @@ -101,6 +113,10 @@ def test_launch_and_close_v4_v6(self):
@unittest.skipIf(not has_working_ipv6(), "Requires IPv6")
@unittest.skipIf(os.environ.get("SKIP_IPV6"), "IPv6 tests disabled")
def test_launch_and_close_v6_only(self):
rv = r.Zeroconf(
interfaces=r.InterfaceChoice.AllWithLoopback, ip_version=r.IPVersion.V6Only
)
rv.close()
rv = r.Zeroconf(interfaces=r.InterfaceChoice.All, ip_version=r.IPVersion.V6Only)
rv.close()
rv = r.Zeroconf(
Expand Down
7 changes: 5 additions & 2 deletions tests/utils/test_net.py
Original file line number Diff line number Diff line change
Expand Up @@ -85,12 +85,15 @@ def test_ip6_addresses_to_indexes():
]


def test_normalize_interface_choice_errors():
@pytest.mark.parametrize(
"interface_choice", (r.InterfaceChoice.All, r.InterfaceChoice.AllWithLoopback)
)
def test_normalize_interface_choice_errors(interface_choice: r.InterfaceChoice) -> None:
"""Test we generate exception on invalid input."""
with patch("zeroconf._utils.net.get_all_addresses", return_value=[]), patch(
"zeroconf._utils.net.get_all_addresses_v6", return_value=[]
), pytest.raises(RuntimeError):
netutils.normalize_interface_choice(r.InterfaceChoice.All)
netutils.normalize_interface_choice(interface_choice)

with pytest.raises(TypeError):
netutils.normalize_interface_choice("1.2.3.4")
Expand Down
Loading