""" Multicast DNS Service Discovery for Python, v0.14-wmcbrine

Copyright 2003 Paul Scott-Murphy, 2014 William McBrine

This module provides a framework for the use of DNS Service Discovery

using IP multicast.

This library is free software; you can redistribute it and/or

modify it under the terms of the GNU Lesser General Public

License as published by the Free Software Foundation; either

version 2.1 of the License, or (at your option) any later version.

This library is distributed in the hope that it will be useful,

but WITHOUT ANY WARRANTY; without even the implied warranty of

MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU

Lesser General Public License for more details.

You should have received a copy of the GNU Lesser General Public

License along with this library; if not, write to the Free Software

Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301

USA

"""

import enum

import errno

import ipaddress

import socket

import struct

import sys

from typing import Any, List, Optional, Tuple, Union, cast

import ifaddr

from .._logger import log

from ..const import _IPPROTO_IPV6, _MDNS_ADDR, _MDNS_ADDR6, _MDNS_PORT

@enum.unique

class InterfaceChoice(enum.Enum):

Default = 1

All = 2

InterfacesType = Union[List[Union[str, int, Tuple[Tuple[str, int, int], int]]], InterfaceChoice]

@enum.unique

class ServiceStateChange(enum.Enum):

Added = 1

Removed = 2

Updated = 3

@enum.unique

class IPVersion(enum.Enum):

V4Only = 1

V6Only = 2

All = 3

# 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

return socket.inet_pton(address_family, address)

def get_all_addresses() -> List[str]:

return list({addr.ip for iface in ifaddr.get_adapters() for addr in iface.ips if addr.is_IPv4})

def get_all_addresses_v6() -> List[Tuple[Tuple[str, int, int], int]]:

# IPv6 multicast uses positive indexes for interfaces

# TODO: What about multi-address interfaces?

return list(

{(addr.ip, iface.index) for iface in ifaddr.get_adapters() for addr in iface.ips if addr.is_IPv6}

)

def ip6_to_address_and_index(adapters: List[Any], ip: str) -> Tuple[Tuple[str, int, int], int]:

if '%' in ip:

ip = ip[: ip.index('%')] # Strip scope_id.

ipaddr = ipaddress.ip_address(ip)

for adapter in adapters:

for adapter_ip in adapter.ips:

# IPv6 addresses are represented as tuples

if isinstance(adapter_ip.ip, tuple) and ipaddress.ip_address(adapter_ip.ip[0]) == ipaddr:

return (cast(Tuple[str, int, int], adapter_ip.ip), cast(int, adapter.index))

raise RuntimeError('No adapter found for IP address %s' % ip)

def interface_index_to_ip6_address(adapters: List[Any], index: int) -> Tuple[str, int, int]:

for adapter in adapters:

if adapter.index == index:

for adapter_ip in adapter.ips:

# IPv6 addresses are represented as tuples

if isinstance(adapter_ip.ip, tuple):

return cast(Tuple[str, int, int], adapter_ip.ip)

raise RuntimeError('No adapter found for index %s' % index)

def ip6_addresses_to_indexes(

interfaces: List[Union[str, int, Tuple[Tuple[str, int, int], int]]]

) -> List[Tuple[Tuple[str, int, int], int]]:

"""Convert IPv6 interface addresses to interface indexes.

IPv4 addresses are ignored.

:param interfaces: List of IP addresses and indexes.

:returns: List of indexes.

"""

result = []

adapters = ifaddr.get_adapters()

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:

result.append(ip6_to_address_and_index(adapters, iface))

return result

def normalize_interface_choice(

choice: InterfacesType, ip_version: IPVersion = IPVersion.V4Only

) -> List[Union[str, Tuple[Tuple[str, int, int], int]]]:

"""Convert the interfaces choice into internal representation.

:param choice: `InterfaceChoice` or list of interface addresses or indexes (IPv6 only).

:param ip_address: IP version to use (ignored if `choice` is a list).

:returns: List of IP addresses (for IPv4) and indexes (for IPv6).

"""

result: List[Union[str, Tuple[Tuple[str, int, int], int]]] = []

if choice is InterfaceChoice.Default:

if ip_version != IPVersion.V4Only:

# IPv6 multicast uses interface 0 to mean the default

result.append((('', 0, 0), 0))

if ip_version != IPVersion.V6Only:

result.append('0.0.0.0')

elif choice is InterfaceChoice.All:

if ip_version != IPVersion.V4Only:

result.extend(get_all_addresses_v6())

if ip_version != IPVersion.V6Only:

result.extend(get_all_addresses())

if not result:

raise RuntimeError(

'No interfaces to listen on, check that any interfaces have IP version %s' % ip_version

)

elif isinstance(choice, list):

# First, take IPv4 addresses.

result = [i for i in choice if isinstance(i, str) and ipaddress.ip_address(i).version == 4]

# Unlike IP_ADD_MEMBERSHIP, IPV6_JOIN_GROUP requires interface indexes.

result += ip6_addresses_to_indexes(choice)

else:

raise TypeError("choice must be a list or InterfaceChoice, got %r" % choice)

return result

def disable_ipv6_only_or_raise(s: socket.socket) -> None:

"""Make V6 sockets work for both V4 and V6 (required for Windows)."""

try:

s.setsockopt(_IPPROTO_IPV6, socket.IPV6_V6ONLY, False)

except OSError:

log.error('Support for dual V4-V6 sockets is not present, use IPVersion.V4 or IPVersion.V6')

raise

def set_so_reuseport_if_available(s: socket.socket) -> None:

"""Set SO_REUSEADDR on a socket if available."""

# SO_REUSEADDR should be equivalent to SO_REUSEPORT for

# multicast UDP sockets (p 731, "TCP/IP Illustrated,

# Volume 2"), but some BSD-derived systems require

# SO_REUSEPORT to be specified explicitly. Also, not all

# versions of Python have SO_REUSEPORT available.

# Catch OSError and socket.error for kernel versions <3.9 because lacking

# SO_REUSEPORT support.

if not hasattr(socket, 'SO_REUSEPORT'):

return

try:

s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) # pylint: disable=no-member

except OSError as err:

if err.errno != errno.ENOPROTOOPT:

raise

def set_mdns_port_socket_options_for_ip_version(

s: socket.socket, bind_addr: Union[Tuple[str], Tuple[str, int, int]], ip_version: IPVersion

) -> None:

"""Set ttl/hops and loop for mdns port."""

if ip_version != IPVersion.V6Only:

ttl = struct.pack(b'B', 255)

loop = struct.pack(b'B', 1)

# OpenBSD needs the ttl and loop values for the IP_MULTICAST_TTL and

# IP_MULTICAST_LOOP socket options as an unsigned char.

try:

s.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, ttl)

s.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_LOOP, loop)

except OSError as e:

if bind_addr[0] != '' or get_errno(e) != errno.EINVAL: # Fails to set on MacOS

raise

if ip_version != IPVersion.V4Only:

# However, char doesn't work here (at least on Linux)

s.setsockopt(_IPPROTO_IPV6, socket.IPV6_MULTICAST_HOPS, 255)

s.setsockopt(_IPPROTO_IPV6, socket.IPV6_MULTICAST_LOOP, True)

def new_socket(

bind_addr: Union[Tuple[str], Tuple[str, int, int]],

port: int = _MDNS_PORT,

ip_version: IPVersion = IPVersion.V4Only,

apple_p2p: bool = False,

) -> Optional[socket.socket]:

log.debug(

'Creating new socket with port %s, ip_version %s, apple_p2p %s and bind_addr %r',

port,

ip_version,

apple_p2p,

bind_addr,

)

socket_family = socket.AF_INET if ip_version == IPVersion.V4Only else socket.AF_INET6

s = socket.socket(socket_family, socket.SOCK_DGRAM)

if ip_version == IPVersion.All:

disable_ipv6_only_or_raise(s)

s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

set_so_reuseport_if_available(s)

if port == _MDNS_PORT:

set_mdns_port_socket_options_for_ip_version(s, bind_addr, ip_version)

if apple_p2p:

# SO_RECV_ANYIF = 0x1104

# https://opensource.apple.com/source/xnu/xnu-4570.41.2/bsd/sys/socket.h

s.setsockopt(socket.SOL_SOCKET, 0x1104, 1)

bind_tup = (bind_addr[0], port, *bind_addr[1:])

try:

s.bind(bind_tup)

except OSError as ex:

if ex.errno == errno.EADDRNOTAVAIL:

log.warning(

'Address not available when binding to %s, ' 'it is expected to happen on some systems',

bind_tup,

)

return None

raise

log.debug('Created socket %s', s)

return s

def add_multicast_member(

listen_socket: socket.socket,

interface: Union[str, Tuple[Tuple[str, int, int], int]],

) -> bool:

# This is based on assumptions in normalize_interface_choice

is_v6 = isinstance(interface, tuple)

err_einval = {errno.EINVAL}

if sys.platform == 'win32':

# No WSAEINVAL definition in typeshed

err_einval |= {cast(Any, errno).WSAEINVAL} # pylint: disable=no-member

log.debug('Adding %r (socket %d) to multicast group', interface, listen_socket.fileno())

try:

if is_v6:

try:

mdns_addr6_bytes = socket.inet_pton(socket.AF_INET6, _MDNS_ADDR6)

except OSError:

log.info(

'Unable to translate IPv6 address when adding %s to multicast group, '

'this can happen if IPv6 is disabled on the system',

interface,

)

return False

iface_bin = struct.pack('@I', cast(int, interface[1]))

_value = mdns_addr6_bytes + iface_bin

listen_socket.setsockopt(_IPPROTO_IPV6, socket.IPV6_JOIN_GROUP, _value)

else:

_value = socket.inet_aton(_MDNS_ADDR) + socket.inet_aton(cast(str, interface))

listen_socket.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, _value)

except OSError as e:

_errno = get_errno(e)

if _errno == errno.EADDRINUSE:

log.info(

'Address in use when adding %s to multicast group, '

'it is expected to happen on some systems',

interface,

)

return False

if _errno == errno.EADDRNOTAVAIL:

log.info(

'Address not available when adding %s to multicast '

'group, it is expected to happen on some systems',

interface,

)

return False

if _errno in err_einval:

log.info('Interface of %s does not support multicast, ' 'it is expected in WSL', interface)

return False

if _errno == errno.ENOPROTOOPT:

log.info(

'Failed to set socket option on %s, this can happen if '

'the network adapter is in a disconnected state',

interface,

)

return False

if is_v6 and _errno == errno.ENODEV:

log.info(

'Address in use when adding %s to multicast group, '

'it is expected to happen when the device does not have ipv6',

interface,

)

return False

raise

return True

def new_respond_socket(

interface: Union[str, Tuple[Tuple[str, int, int], int]],

apple_p2p: bool = False,

) -> Optional[socket.socket]:

is_v6 = isinstance(interface, tuple)

respond_socket = new_socket(

ip_version=(IPVersion.V6Only if is_v6 else IPVersion.V4Only),

apple_p2p=apple_p2p,

bind_addr=cast(Tuple[Tuple[str, int, int], int], interface)[0] if is_v6 else (cast(str, interface),),

)

if not respond_socket:

return None

log.debug('Configuring socket %s with multicast interface %s', respond_socket, interface)

if is_v6:

iface_bin = struct.pack('@I', cast(int, interface[1]))

respond_socket.setsockopt(_IPPROTO_IPV6, socket.IPV6_MULTICAST_IF, iface_bin)

else:

respond_socket.setsockopt(

socket.IPPROTO_IP, socket.IP_MULTICAST_IF, socket.inet_aton(cast(str, interface))

)

return respond_socket

def create_sockets(

interfaces: InterfacesType = InterfaceChoice.All,

unicast: bool = False,

ip_version: IPVersion = IPVersion.V4Only,

apple_p2p: bool = False,

) -> Tuple[Optional[socket.socket], List[socket.socket]]:

if unicast:

listen_socket = None

else:

listen_socket = new_socket(ip_version=ip_version, apple_p2p=apple_p2p, bind_addr=('',))

normalized_interfaces = normalize_interface_choice(interfaces, ip_version)

# If we are using InterfaceChoice.Default we can use

# a single socket to listen and respond.

if not unicast and interfaces is InterfaceChoice.Default:

for i in normalized_interfaces:

add_multicast_member(cast(socket.socket, listen_socket), i)

return listen_socket, [cast(socket.socket, listen_socket)]

respond_sockets = []

for i in normalized_interfaces:

if not unicast:

if add_multicast_member(cast(socket.socket, listen_socket), i):

respond_socket = new_respond_socket(i, apple_p2p=apple_p2p)

else:

respond_socket = None

else:

respond_socket = new_socket(

port=0,

ip_version=ip_version,

apple_p2p=apple_p2p,

bind_addr=i[0] if isinstance(i, tuple) else (i,),

)

if respond_socket is not None:

respond_sockets.append(respond_socket)

return listen_socket, respond_sockets

def get_errno(e: Exception) -> int:

assert isinstance(e, socket.error)

return cast(int, e.args[0])

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:

"""Auto detect the IP version when it is not provided."""

if isinstance(interfaces, list):

has_v6 = any(

isinstance(i, int) or (isinstance(i, str) and ipaddress.ip_address(i).version == 6)

for i in interfaces

)

has_v4 = any(isinstance(i, str) and ipaddress.ip_address(i).version == 4 for i in interfaces)

if has_v4 and has_v6:

return IPVersion.All

if has_v6:

return IPVersion.V6Only

return IPVersion.V4Only