diff --git a/zeroconf/__init__.py b/zeroconf/__init__.py index 055d0439e..5c04eb26e 100644 --- a/zeroconf/__init__.py +++ b/zeroconf/__init__.py @@ -67,7 +67,6 @@ from ._utils.name import service_type_name # noqa # import needed for backwards compat from ._utils.net import ( # noqa # import needed for backwards compat add_multicast_member, - can_send_to, autodetect_ip_version, create_sockets, get_all_addresses_v6, diff --git a/zeroconf/_core.py b/zeroconf/_core.py index 15852ffb2..74b1828f0 100644 --- a/zeroconf/_core.py +++ b/zeroconf/_core.py @@ -836,12 +836,13 @@ def _async_send_transport( v6_flow_scope: Union[Tuple[()], Tuple[int, int]] = (), ) -> None: s = transport.get_extra_info('socket') + ipv6_socket = s.family == socket.AF_INET6 if addr is None: - real_addr = _MDNS_ADDR6 if s.family == socket.AF_INET6 else _MDNS_ADDR + real_addr = _MDNS_ADDR6 if ipv6_socket else _MDNS_ADDR else: real_addr = addr - if not can_send_to(s, real_addr): - return + if not can_send_to(ipv6_socket, real_addr): + return log.debug( 'Sending to (%s, %d) via [socket %s (%s)] (%d bytes #%d) %r as %r...', real_addr, @@ -855,7 +856,7 @@ def _async_send_transport( ) # Get flowinfo and scopeid for the IPV6 socket to create a complete IPv6 # address tuple: https://docs.python.org/3.6/library/socket.html#socket-families - if s.family == socket.AF_INET6 and not v6_flow_scope: + if ipv6_socket and not v6_flow_scope: _, _, sock_flowinfo, sock_scopeid = s.getsockname() v6_flow_scope = (sock_flowinfo, sock_scopeid) transport.sendto(packet, (real_addr, port or _MDNS_PORT, *v6_flow_scope)) diff --git a/zeroconf/_utils/net.py b/zeroconf/_utils/net.py index bfae9db46..c53ec9786 100644 --- a/zeroconf/_utils/net.py +++ b/zeroconf/_utils/net.py @@ -379,9 +379,13 @@ def get_errno(e: Exception) -> int: return cast(int, e.args[0]) -def can_send_to(sock: socket.socket, address: str) -> bool: - addr = ipaddress.ip_address(address) - return cast(bool, addr.version == 6 if sock.family == socket.AF_INET6 else addr.version == 4) +def can_send_to(ipv6_socket: bool, address: str) -> bool: + """Check if the address type matches the socket type. + + This function does not validate if the address is a valid + ipv6 or ipv4 address. + """ + return ":" in address if ipv6_socket else ":" not in address def autodetect_ip_version(interfaces: InterfacesType) -> IPVersion: