2424
2525import enum
2626import errno
27- from functools import lru_cache
2827import ipaddress
2928import socket
3029import struct
3130import sys
3231import warnings
3332from collections .abc import Iterable , Sequence
33+ from functools import lru_cache
3434from typing import Any , Union , cast
3535
3636import ifaddr
@@ -74,14 +74,17 @@ def _encode_address(address: str) -> bytes:
7474 address_family = socket .AF_INET6 if is_ipv6 else socket .AF_INET
7575 return socket .inet_pton (address_family , address )
7676
77+
7778@lru_cache (maxsize = 512 )
7879def _get_interface_name (iface : int ) -> str :
7980 """Get the interface name from the interface index."""
8081 return socket .if_indextoname (iface )
8182
8283
8384def get_all_addresses_ipv4 (adapters : Iterable [ifaddr .Adapter ]) -> list [str ]:
84- return list ({(addr .ip , iface .index ) for iface in ifaddr .get_adapters () for addr in iface .ips if addr .is_IPv4 }) # type: ignore[misc]
85+ return list (
86+ {(addr .ip , iface .index ) for iface in ifaddr .get_adapters () for addr in iface .ips if addr .is_IPv4 }
87+ ) # type: ignore[misc]
8588
8689
8790def get_all_addresses_ipv6 (adapters : Iterable [ifaddr .Adapter ]) -> list [tuple [tuple [str , int , int ], int ]]:
@@ -91,14 +94,18 @@ def get_all_addresses_ipv6(adapters: Iterable[ifaddr.Adapter]) -> list[tuple[tup
9194 {(addr .ip , iface .index ) for iface in adapters for addr in iface .ips if addr .is_IPv6 } # type: ignore[misc]
9295 )
9396
94- def get_all_addresses_ipv6_link_local (adapters : Iterable [ifaddr .Adapter ]) -> list [tuple [tuple [str , int , int ], int ]]:
97+
98+ def get_all_addresses_ipv6_link_local (
99+ adapters : Iterable [ifaddr .Adapter ],
100+ ) -> list [tuple [tuple [str , int , int ], int ]]:
95101 return [
96102 (ip .ip , iface .index )
97103 for iface in adapters
98104 for ip in iface .ips
99105 if ip .is_IPv6 and ip .ip [2 ] == iface .index
100106 ]
101107
108+
102109def get_all_addresses () -> list [str ]:
103110 warnings .warn (
104111 "get_all_addresses is deprecated, and will be removed in a future version. Use ifaddr"
@@ -119,14 +126,16 @@ def get_all_addresses_v6() -> list[tuple[tuple[str, int, int], int]]:
119126 return get_all_addresses_ipv6 (ifaddr .get_adapters ())
120127
121128
122- def ip_to_address_and_index (adapters : Iterable [ifaddr .Adapter ], ip : str ) -> tuple [tuple [str , int , int ], int ] | tuple [str , int ]:
129+ def ip_to_address_and_index (
130+ adapters : Iterable [ifaddr .Adapter ], ip : str
131+ ) -> tuple [tuple [str , int , int ], int ] | tuple [str , int ]:
123132 """Convert IP to IP address and interface index.
124133
125134 :param adapters: List of adapters.
126135 :param ip: IP address to convert. If the address is an IPv6 with scope id, the scope id is being validated.
127136 :returns: Tuple of IP address and interface index.
128137 """
129- addr , sep , scope_id = ip .partition ('%' )
138+ addr , sep , scope_id = ip .partition ("%" )
130139 if not sep :
131140 scope_id = None
132141
@@ -144,10 +153,7 @@ def ip_to_address_and_index(adapters: Iterable[ifaddr.Adapter], ip: str) -> tupl
144153 return (adapter_ip .ip , adapter .index )
145154
146155 # IPv4 addresses are represented as strings
147- if (
148- isinstance (adapter_ip .ip , str )
149- and ipaddress .ip_address (adapter_ip .ip ) == ipaddr
150- ):
156+ if isinstance (adapter_ip .ip , str ) and ipaddress .ip_address (adapter_ip .ip ) == ipaddr :
151157 return (adapter_ip .ip , adapter .index )
152158
153159 raise RuntimeError (f"No adapter found for IP address { ip } " )
@@ -347,11 +353,12 @@ def new_socket(
347353 log .debug ("Created socket %s" , s )
348354 return s
349355
356+
350357def _bind_to_interface (
351358 s : socket .socket ,
352359 ifaceidx : int ,
353360) -> None :
354- if hasattr (socket ,' SO_BINDTODEVICE' ):
361+ if hasattr (socket , " SO_BINDTODEVICE" ):
355362 iface = _get_interface_name (ifaceidx ).encode ()
356363 s .setsockopt (socket .SOL_SOCKET , socket .SO_BINDTODEVICE , iface )
357364 elif sys .platform != "win32" :
@@ -377,7 +384,7 @@ def add_multicast_member(
377384 err_einval |= {cast (Any , errno ).WSAEINVAL } # pylint: disable=no-member
378385 log .debug ("Adding %r (socket %d) to multicast group" , interface , listen_socket .fileno ())
379386 try :
380- #_bind_to_interface(listen_socket, interface[1])
387+ # _bind_to_interface(listen_socket, interface[1])
381388 if is_v6 :
382389 try :
383390 mdns_addr6_bytes = socket .inet_pton (socket .AF_INET6 , _MDNS_ADDR6 )
@@ -461,7 +468,7 @@ def new_respond_socket(
461468 # - Use ephemeral ports if in unicast mode
462469 # - Create socket according to the interface IP type (IPv4 or IPv6)
463470 respond_socket = new_socket (
464- #bind_addr=cast(tuple[tuple[str, int, int], int], interface)[0] if is_v6 else (cast(tuple[str, int], interface)[0],),
471+ # bind_addr=cast(tuple[tuple[str, int, int], int], interface)[0] if is_v6 else (cast(tuple[str, int], interface)[0],),
465472 bind_addr = ("::" , 0 , interface [1 ]),
466473 port = 0 if unicast else _MDNS_PORT ,
467474 ip_version = (IPVersion .V6Only if is_v6 else IPVersion .V4Only ),
0 commit comments