from __future__ import annotations

""" Multicast DNS Service Discovery for Python, v0.12

    Copyright (C) 2003, Paul Scott-Murphy

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

    using IP multicast.  It has been tested against the JRendezvous

    implementation from <a href="http://strangeberry.com">StrangeBerry</a>,

    and against the mDNSResponder from Mac OS X 10.3.8.

    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, see

    <http://www.gnu.org/licenses/>.

"""

"""0.12 update - allow selection of binding interface

         typo fix - Thanks A. M. Kuchlingi

         removed all use of word 'Rendezvous' - this is an API change"""

"""0.11 update - correction to comments for addListener method

                 support for new record types seen from OS X

                  - IPv6 address

                  - hostinfo

                 ignore unknown DNS record types

                 fixes to name decoding

                 works alongside other processes using port 5353 (e.g. Mac OS X)

                 tested against Mac OS X 10.3.2's mDNSResponder

                 corrections to removal of list entries for service browser"""

"""0.10 update - Jonathon Paisley contributed these corrections:

                 always multicast replies, even when query is unicast

                 correct a pointer encoding problem

                 can now write records in any order

                 traceback shown on failure

                 better TXT record parsing

                 server is now separate from name

                 can cancel a service browser

                 modified some unit tests to accommodate these changes"""

"""0.09 update - remove all records on service unregistration

                 fix DOS security problem with readName"""

"""0.08 update - changed licensing to LGPL"""

"""0.07 update - faster shutdown on engine

                 pointer encoding of outgoing names

                 ServiceBrowser now works

                 new unit tests"""

"""0.06 update - small improvements with unit tests

                 added defined exception types

                 new style objects

                 fixed hostname/interface problem

                 fixed socket timeout problem

                 fixed addServiceListener() typo bug

                 using select() for socket reads

                 tested on Debian unstable with Python 2.2.2"""

"""0.05 update - ensure case insensitivity on domain names

                 support for unicast DNS queries"""

"""0.04 update - added some unit tests

                 added __ne__ adjuncts where required

                 ensure names end in '.local.'

                 timeout on receiving socket for clean shutdown"""

__author__ = b"Paul Scott-Murphy"

__email__ = b"paul at scott dash murphy dot com"

__version__ = b"0.12"

import errno

import itertools

import select

import socket

import struct

import threading

import time

import traceback

from mercurial import pycompat

__all__ = ["Zeroconf", "ServiceInfo", "ServiceBrowser"]

# hook for threads

globals()['_GLOBAL_DONE'] = 0

# Some timing constants

_UNREGISTER_TIME = 125

_CHECK_TIME = 175

_REGISTER_TIME = 225

_LISTENER_TIME = 200

_BROWSER_TIME = 500

# Some DNS constants

_MDNS_ADDR = r'224.0.0.251'

_MDNS_PORT = 5353

_DNS_PORT = 53

_DNS_TTL = 60 * 60  # one hour default TTL

_MAX_MSG_TYPICAL = 1460  # unused

_MAX_MSG_ABSOLUTE = 8972

_FLAGS_QR_MASK = 0x8000  # query response mask

_FLAGS_QR_QUERY = 0x0000  # query

_FLAGS_QR_RESPONSE = 0x8000  # response

_FLAGS_AA = 0x0400  # Authoritative answer

_FLAGS_TC = 0x0200  # Truncated

_FLAGS_RD = 0x0100  # Recursion desired

_FLAGS_RA = 0x8000  # Recursion available

_FLAGS_Z = 0x0040  # Zero

_FLAGS_AD = 0x0020  # Authentic data

_FLAGS_CD = 0x0010  # Checking disabled

_CLASS_IN = 1

_CLASS_CS = 2

_CLASS_CH = 3

_CLASS_HS = 4

_CLASS_NONE = 254

_CLASS_ANY = 255

_CLASS_MASK = 0x7FFF

_CLASS_UNIQUE = 0x8000

_TYPE_A = 1

_TYPE_NS = 2

_TYPE_MD = 3

_TYPE_MF = 4

_TYPE_CNAME = 5

_TYPE_SOA = 6

_TYPE_MB = 7

_TYPE_MG = 8

_TYPE_MR = 9

_TYPE_NULL = 10

_TYPE_WKS = 11

_TYPE_PTR = 12

_TYPE_HINFO = 13

_TYPE_MINFO = 14

_TYPE_MX = 15

_TYPE_TXT = 16

_TYPE_AAAA = 28

_TYPE_SRV = 33

_TYPE_ANY = 255

# Mapping constants to names

_CLASSES = {

    _CLASS_IN: b"in",

    _CLASS_CS: b"cs",

    _CLASS_CH: b"ch",

    _CLASS_HS: b"hs",

    _CLASS_NONE: b"none",

    _CLASS_ANY: b"any",

}

_TYPES = {

    _TYPE_A: b"a",

    _TYPE_NS: b"ns",

    _TYPE_MD: b"md",

    _TYPE_MF: b"mf",

    _TYPE_CNAME: b"cname",

    _TYPE_SOA: b"soa",

    _TYPE_MB: b"mb",

    _TYPE_MG: b"mg",

    _TYPE_MR: b"mr",

    _TYPE_NULL: b"null",

    _TYPE_WKS: b"wks",

    _TYPE_PTR: b"ptr",

    _TYPE_HINFO: b"hinfo",

    _TYPE_MINFO: b"minfo",

    _TYPE_MX: b"mx",

    _TYPE_TXT: b"txt",

    _TYPE_AAAA: b"quada",

    _TYPE_SRV: b"srv",

    _TYPE_ANY: b"any",

}

# utility functions

def currentTimeMillis():

    """Current system time in milliseconds"""

    return time.time() * 1000

# Exceptions

class NonLocalNameException(Exception):

    pass

class NonUniqueNameException(Exception):

    pass

class NamePartTooLongException(Exception):

    pass

class AbstractMethodException(Exception):

    pass

class BadTypeInNameException(Exception):

    pass

class BadDomainName(Exception):

    def __init__(self, pos):

        Exception.__init__(self, b"at position %s" % pos)

class BadDomainNameCircular(BadDomainName):

    pass

# implementation classes

_SOL_IP = socket.SOL_IP

if pycompat.iswindows:

    # XXX: Not sure if there are newer versions of python where this would fail,

    # but apparently socket.SOL_IP used to be 0, and socket.IPPROTO_IP is 0, so

    # this would work with older versions of python.

    #

    # https://github.com/python/cpython/issues/101960

    _SOL_IP = socket.IPPROTO_IP

class DNSEntry:

    """A DNS entry"""

    def __init__(self, name, type, clazz):

        self.key = name.lower()

        self.name = name

        self.type = type

        self.clazz = clazz & _CLASS_MASK

        self.unique = (clazz & _CLASS_UNIQUE) != 0

    def __eq__(self, other):

        """Equality test on name, type, and class"""

        if isinstance(other, DNSEntry):

            return (

                self.name == other.name

                and self.type == other.type

                and self.clazz == other.clazz

            )

        return 0

    def __ne__(self, other):

        """Non-equality test"""

        return not self.__eq__(other)

    def getClazz(self, clazz):

        """Class accessor"""

        try:

            return _CLASSES[clazz]

        except KeyError:

            return b"?(%s)" % clazz

    def getType(self, type):

        """Type accessor"""

        try:

            return _TYPES[type]

        except KeyError:

            return b"?(%s)" % type

    def toString(self, hdr, other):

        """String representation with additional information"""

        result = b"%s[%s,%s" % (

            hdr,

            self.getType(self.type),

            self.getClazz(self.clazz),

        )

        if self.unique:

            result += b"-unique,"

        else:

            result += b","

        result += self.name

        if other is not None:

            result += b",%s]" % other

        else:

            result += b"]"

        return result

class DNSQuestion(DNSEntry):

    """A DNS question entry"""

    def __init__(self, name, type, clazz):

        if isinstance(name, str):

            name = name.encode('ascii')

        if not name.endswith(b".local."):

            raise NonLocalNameException(name)

        DNSEntry.__init__(self, name, type, clazz)

    def answeredBy(self, rec):

        """Returns true if the question is answered by the record"""

        return (

            self.clazz == rec.clazz

            and (self.type == rec.type or self.type == _TYPE_ANY)

            and self.name == rec.name

        )

    def __repr__(self):

        """String representation"""

        return DNSEntry.toString(self, b"question", None)

class DNSRecord(DNSEntry):

    """A DNS record - like a DNS entry, but has a TTL"""

    def __init__(self, name, type, clazz, ttl):

        DNSEntry.__init__(self, name, type, clazz)

        self.ttl = ttl

        self.created = currentTimeMillis()

    def __eq__(self, other):

        """Tests equality as per DNSRecord"""

        if isinstance(other, DNSRecord):

            return DNSEntry.__eq__(self, other)

        return 0

    def suppressedBy(self, msg):

        """Returns true if any answer in a message can suffice for the

        information held in this record."""

        for record in msg.answers:

            if self.suppressedByAnswer(record):

                return 1

        return 0

    def suppressedByAnswer(self, other):

        """Returns true if another record has same name, type and class,

        and if its TTL is at least half of this record's."""

        if self == other and other.ttl > (self.ttl / 2):

            return 1

        return 0

    def getExpirationTime(self, percent):

        """Returns the time at which this record will have expired

        by a certain percentage."""

        return self.created + (percent * self.ttl * 10)

    def getRemainingTTL(self, now):

        """Returns the remaining TTL in seconds."""

        return max(0, (self.getExpirationTime(100) - now) / 1000)

    def isExpired(self, now):

        """Returns true if this record has expired."""

        return self.getExpirationTime(100) <= now

    def isStale(self, now):

        """Returns true if this record is at least half way expired."""

        return self.getExpirationTime(50) <= now

    def resetTTL(self, other):

        """Sets this record's TTL and created time to that of

        another record."""

        self.created = other.created

        self.ttl = other.ttl

    def write(self, out):

        """Abstract method"""

        raise AbstractMethodException

    def toString(self, hdr, other):

        """String representation with additional information"""

        arg = b"%s/%s,%s" % (

            self.ttl,

            self.getRemainingTTL(currentTimeMillis()),

            other,

        )

        return DNSEntry.toString(self, b"record", arg)

class DNSAddress(DNSRecord):

    """A DNS address record"""

    def __init__(self, name, type, clazz, ttl, address):

        DNSRecord.__init__(self, name, type, clazz, ttl)

        self.address = address

    def write(self, out):

        """Used in constructing an outgoing packet"""

        out.writeString(self.address, len(self.address))

    def __eq__(self, other):

        """Tests equality on address"""

        if isinstance(other, DNSAddress):

            return self.address == other.address

        return 0

    def __repr__(self):

        """String representation"""

        try:

            return socket.inet_ntoa(self.address)

        except Exception:

            return self.address

class DNSHinfo(DNSRecord):

    """A DNS host information record"""

    def __init__(self, name, type, clazz, ttl, cpu, os):

        DNSRecord.__init__(self, name, type, clazz, ttl)

        self.cpu = cpu

        self.os = os

    def write(self, out):

        """Used in constructing an outgoing packet"""

        out.writeString(self.cpu, len(self.cpu))

        out.writeString(self.os, len(self.os))

    def __eq__(self, other):

        """Tests equality on cpu and os"""

        if isinstance(other, DNSHinfo):

            return self.cpu == other.cpu and self.os == other.os

        return 0

    def __repr__(self):

        """String representation"""

        return self.cpu + b" " + self.os

class DNSPointer(DNSRecord):

    """A DNS pointer record"""

    def __init__(self, name, type, clazz, ttl, alias):

        DNSRecord.__init__(self, name, type, clazz, ttl)

        self.alias = alias

    def write(self, out):

        """Used in constructing an outgoing packet"""

        out.writeName(self.alias)

    def __eq__(self, other):

        """Tests equality on alias"""

        if isinstance(other, DNSPointer):

            return self.alias == other.alias

        return 0

    def __repr__(self):

        """String representation"""

        return self.toString(b'', self.alias)

class DNSText(DNSRecord):

    """A DNS text record"""

    def __init__(self, name, type, clazz, ttl, text):

        DNSRecord.__init__(self, name, type, clazz, ttl)

        self.text = text

    def write(self, out):

        """Used in constructing an outgoing packet"""

        out.writeString(self.text, len(self.text))

    def __eq__(self, other):

        """Tests equality on text"""

        if isinstance(other, DNSText):

            return self.text == other.text

        return 0

    def __repr__(self):

        """String representation"""

        if len(self.text) > 10:

            return self.toString(b'', self.text[:7] + b"...")

        else:

            return self.toString(b'', self.text)

class DNSService(DNSRecord):

    """A DNS service record"""

    def __init__(self, name, type, clazz, ttl, priority, weight, port, server):