diff --git a/.gitignore b/.gitignore index dc84959d..264bddcf 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,12 @@ build/ +*.pyc +*.pyo +Thumbs.db +.DS_Store +.project +.pydevproject +.settings +.idea +.vslick +.cache diff --git a/README.rst b/README.rst index 3e812c80..e04f1470 100644 --- a/README.rst +++ b/README.rst @@ -110,6 +110,13 @@ Here's an example: If you want to customize that you need to specify ``interfaces`` argument when constructing ``Zeroconf`` object (see the code for details). +If you don't know the name of the service you need to browse for, try: + +.. code-block:: python + + from zeroconf import ZeroconfServiceTypes + print('\n'.join(ZeroconfServiceTypes.find())) + See examples directory for more. Changelog diff --git a/setup.cfg b/setup.cfg index 51017e12..24b129b7 100644 --- a/setup.cfg +++ b/setup.cfg @@ -5,3 +5,4 @@ universal = 1 show-source = 1 import-order-style=google application-import-names=zeroconf +max-line-length=110 diff --git a/test_zeroconf.py b/test_zeroconf.py index 6ecf6728..b42244bc 100644 --- a/test_zeroconf.py +++ b/test_zeroconf.py @@ -7,21 +7,22 @@ import logging import socket import struct +import time import unittest from threading import Event -from mock import Mock from six import indexbytes from six.moves import xrange import zeroconf as r from zeroconf import ( + DNSHinfo, DNSText, - Listener, ServiceBrowser, ServiceInfo, ServiceStateChange, Zeroconf, + ZeroconfServiceTypes, ) log = logging.getLogger('zeroconf') @@ -66,6 +67,19 @@ def test_match_question(self): self.assertEqual(len(generated.questions), len(parsed.questions)) self.assertEqual(question, parsed.questions[0]) + def test_dns_hinfo(self): + generated = r.DNSOutgoing(0) + generated.add_additional_answer( + DNSHinfo('irrelevant', r._TYPE_HINFO, 0, 0, 'cpu', 'os')) + parsed = r.DNSIncoming(generated.packet()) + self.assertEqual(parsed.answers[0].cpu, u'cpu') + self.assertEqual(parsed.answers[0].os, u'os') + + generated = r.DNSOutgoing(0) + generated.add_additional_answer( + DNSHinfo('irrelevant', r._TYPE_HINFO, 0, 0, 'cpu', 'x' * 257)) + self.assertRaises(r.NamePartTooLongException, generated.packet) + class PacketForm(unittest.TestCase): @@ -151,6 +165,111 @@ def test_launch_and_close(self): rv.close() +class Exceptions(unittest.TestCase): + + def test_bad_service_info_name(self): + browser = Zeroconf() + self.assertRaises(r.BadTypeInNameException, + browser.get_service_info, "type", "type_not") + browser.close() + + +class ServiceTypesQuery(unittest.TestCase): + + def test_integration_with_listener(self): + + type_ = "_test_service_type._tcp.local." + name = "xxxyyy" + registration_name = "%s.%s" % (name, type_) + + zeroconf_registrar = Zeroconf(interfaces=['127.0.0.1']) + desc = {'path': '/~paulsm/'} + info = ServiceInfo( + type_, registration_name, + socket.inet_aton("10.0.1.2"), 80, 0, 0, + desc, "ash-2.local.") + zeroconf_registrar.register_service(info) + + try: + service_types = ZeroconfServiceTypes.find(timeout=0.5) + assert type_ in service_types + service_types = ZeroconfServiceTypes.find( + zc=zeroconf_registrar, timeout=0.5) + assert type_ in service_types + + finally: + zeroconf_registrar.close() + + +class ListenerTest(unittest.TestCase): + + def test_integration_with_listener_class(self): + + service_added = Event() + service_removed = Event() + + type_ = "_http._tcp.local." + name = "xxxyyy" + registration_name = "%s.%s" % (name, type_) + + class MyListener(object): + def add_service(self, zeroconf, type, name): + zeroconf.get_service_info(type, name) + service_added.set() + + def remove_service(self, zeroconf, type, name): + service_removed.set() + + zeroconf_browser = Zeroconf() + zeroconf_browser.add_service_listener(type_, MyListener()) + + properties = dict( + prop_none=None, + prop_string=b'a_prop', + prop_float=1.0, + prop_blank=b'a blanked string', + prop_true=1, + prop_false=0, + ) + + zeroconf_registrar = Zeroconf() + desc = {'path': '/~paulsm/'} + desc.update(properties) + info = ServiceInfo( + type_, registration_name, + socket.inet_aton("10.0.1.2"), 80, 0, 0, + desc, "ash-2.local.") + zeroconf_registrar.register_service(info) + + try: + service_added.wait(1) + assert service_added.is_set() + + # short pause to allow multicast timers to expire + time.sleep(2) + + # clear the answer cache to force query + for record in zeroconf_browser.cache.entries(): + zeroconf_browser.cache.remove(record) + + # get service info without answer cache + info = zeroconf_browser.get_service_info(type_, registration_name) + + assert info.properties[b'prop_none'] is False + assert info.properties[b'prop_string'] == properties['prop_string'] + assert info.properties[b'prop_float'] is False + assert info.properties[b'prop_blank'] == properties['prop_blank'] + assert info.properties[b'prop_true'] is True + assert info.properties[b'prop_false'] is False + + zeroconf_registrar.unregister_service(info) + service_removed.wait(1) + assert service_removed.is_set() + finally: + zeroconf_registrar.close() + zeroconf_browser.close() + + def test_integration(): service_added = Event() service_removed = Event() @@ -179,26 +298,14 @@ def on_service_state_change(zeroconf, service_type, state_change, name): try: service_added.wait(1) assert service_added.is_set() - zeroconf_registrar.unregister_service(info) - service_removed.wait(1) - assert service_removed.is_set() + # Don't remove service, allow close() to cleanup + finally: zeroconf_registrar.close() browser.cancel() zeroconf_browser.close() -def test_listener_handles_closed_socket_situation_gracefully(): - error = socket.error(socket.EBADF) - error.errno = socket.EBADF - - zeroconf = Mock() - zeroconf.socket.recvfrom.side_effect = error - - listener = Listener(zeroconf) - listener.handle_read(zeroconf.socket) - - def test_dnstext_repr_works(): # There was an issue on Python 3 that prevented DNSText's repr # from working when the text was longer than 10 bytes diff --git a/zeroconf.py b/zeroconf.py index 8c00d2b4..6451ef77 100644 --- a/zeroconf.py +++ b/zeroconf.py @@ -1,4 +1,5 @@ -from __future__ import absolute_import, division, print_function, unicode_literals +from __future__ import ( + absolute_import, division, print_function, unicode_literals) """ Multicast DNS Service Discovery for Python, v0.14-wmcbrine Copyright 2003 Paul Scott-Murphy, 2014 William McBrine @@ -64,10 +65,6 @@ def emit(self, record): if log.level == logging.NOTSET: log.setLevel(logging.WARN) -# hook for threads - -_GLOBAL_DONE = False - # Some timing constants _UNREGISTER_TIME = 125 @@ -197,10 +194,10 @@ class DNSEntry(object): """A DNS entry""" - def __init__(self, name, type, class_): + def __init__(self, name, type_, class_): self.key = name.lower() self.name = name - self.type = type + self.type = type_ self.class_ = class_ & _CLASS_MASK self.unique = (class_ & _CLASS_UNIQUE) != 0 @@ -215,11 +212,13 @@ def __ne__(self, other): """Non-equality test""" return not self.__eq__(other) - def get_class_(self, class_): + @staticmethod + def get_class_(class_): """Class accessor""" return _CLASSES.get(class_, "?(%s)" % class_) - def get_type(self, t): + @staticmethod + def get_type(t): """Type accessor""" return _TYPES.get(t, "?(%s)" % t) @@ -233,7 +232,7 @@ def to_string(self, hdr, other): result += "," result += self.name if other is not None: - result += ",%s]" % (other) + result += ",%s]" % other else: result += "]" return result @@ -243,10 +242,10 @@ class DNSQuestion(DNSEntry): """A DNS question entry""" - def __init__(self, name, type, class_): + def __init__(self, name, type_, class_): # if not name.endswith(".local."): # raise NonLocalNameException - DNSEntry.__init__(self, name, type, class_) + DNSEntry.__init__(self, name, type_, class_) def answered_by(self, rec): """Returns true if the question is answered by the record""" @@ -263,8 +262,8 @@ class DNSRecord(DNSEntry): """A DNS record - like a DNS entry, but has a TTL""" - def __init__(self, name, type, class_, ttl): - DNSEntry.__init__(self, name, type, class_) + def __init__(self, name, type_, class_, ttl): + DNSEntry.__init__(self, name, type_, class_) self.ttl = ttl self.created = current_time_millis() @@ -292,7 +291,7 @@ def get_expiration_time(self, percent): def get_remaining_ttl(self, now): """Returns the remaining TTL in seconds.""" - return max(0, (self.get_expiration_time(100) - now) / 1000) + return max(0, (self.get_expiration_time(100) - now) / 1000.0) def is_expired(self, now): """Returns true if this record has expired.""" @@ -313,9 +312,9 @@ def write(self, out): raise AbstractMethodException def to_string(self, other): - """String representation with addtional information""" - arg = "%s/%s,%s" % (self.ttl, - self.get_remaining_ttl(current_time_millis()), other) + """String representation with additional information""" + arg = "%s/%s,%s" % ( + self.ttl, self.get_remaining_ttl(current_time_millis()), other) return DNSEntry.to_string(self, "record", arg) @@ -323,8 +322,8 @@ class DNSAddress(DNSRecord): """A DNS address record""" - def __init__(self, name, type, class_, ttl, address): - DNSRecord.__init__(self, name, type, class_, ttl) + def __init__(self, name, type_, class_, ttl, address): + DNSRecord.__init__(self, name, type_, class_, ttl) self.address = address def write(self, out): @@ -348,15 +347,21 @@ class DNSHinfo(DNSRecord): """A DNS host information record""" - def __init__(self, name, type, class_, ttl, cpu, os): - DNSRecord.__init__(self, name, type, class_, ttl) - self.cpu = cpu - self.os = os + def __init__(self, name, type_, class_, ttl, cpu, os): + DNSRecord.__init__(self, name, type_, class_, ttl) + try: + self.cpu = cpu.decode('utf-8') + except AttributeError: + self.cpu = cpu + try: + self.os = os.decode('utf-8') + except AttributeError: + self.os = os def write(self, out): """Used in constructing an outgoing packet""" - out.write_string(self.cpu) - out.write_string(self.oso) + out.write_character_string(self.cpu.encode('utf-8')) + out.write_character_string(self.os.encode('utf-8')) def __eq__(self, other): """Tests equality on cpu and os""" @@ -372,8 +377,8 @@ class DNSPointer(DNSRecord): """A DNS pointer record""" - def __init__(self, name, type, class_, ttl, alias): - DNSRecord.__init__(self, name, type, class_, ttl) + def __init__(self, name, type_, class_, ttl, alias): + DNSRecord.__init__(self, name, type_, class_, ttl) self.alias = alias def write(self, out): @@ -418,8 +423,9 @@ class DNSService(DNSRecord): """A DNS service record""" - def __init__(self, name, type, class_, ttl, priority, weight, port, server): - DNSRecord.__init__(self, name, type, class_, ttl) + def __init__(self, name, type_, class_, ttl, + priority, weight, port, server): + DNSRecord.__init__(self, name, type_, class_, ttl) self.priority = priority self.weight = weight self.port = port @@ -455,6 +461,8 @@ def __init__(self, data): self.data = data self.questions = [] self.answers = [] + self.id = 0 + self.flags = 0 self.num_questions = 0 self.num_answers = 0 self.num_authorities = 0 @@ -464,24 +472,25 @@ def __init__(self, data): self.read_questions() self.read_others() - def unpack(self, format): - length = struct.calcsize(format) - info = struct.unpack(format, self.data[self.offset:self.offset + length]) + def unpack(self, format_): + length = struct.calcsize(format_) + info = struct.unpack( + format_, self.data[self.offset:self.offset + length]) self.offset += length return info def read_header(self): """Reads header portion of packet""" (self.id, self.flags, self.num_questions, self.num_answers, - self.num_quthorities, self.num_additionals) = self.unpack(b'!6H') + self.num_authorities, self.num_additionals) = self.unpack(b'!6H') def read_questions(self): """Reads questions section of packet""" for i in xrange(self.num_questions): name = self.read_name() - type, class_ = self.unpack(b'!HH') + type_, class_ = self.unpack(b'!HH') - question = DNSQuestion(name, type, class_) + question = DNSQuestion(name, type_, class_) self.questions.append(question) def read_int(self): @@ -510,24 +519,30 @@ def read_others(self): n = self.num_answers + self.num_authorities + self.num_additionals for i in xrange(n): domain = self.read_name() - type, class_, ttl, length = self.unpack(b'!HHiH') + type_, class_, ttl, length = self.unpack(b'!HHiH') rec = None - if type == _TYPE_A: - rec = DNSAddress(domain, type, class_, ttl, self.read_string(4)) - elif type == _TYPE_CNAME or type == _TYPE_PTR: - rec = DNSPointer(domain, type, class_, ttl, self.read_name()) - elif type == _TYPE_TXT: - rec = DNSText(domain, type, class_, ttl, self.read_string(length)) - elif type == _TYPE_SRV: - rec = DNSService(domain, type, class_, ttl, - self.read_unsigned_short(), self.read_unsigned_short(), - self.read_unsigned_short(), self.read_name()) - elif type == _TYPE_HINFO: - rec = DNSHinfo(domain, type, class_, ttl, - self.read_character_string(), self.read_character_string()) - elif type == _TYPE_AAAA: - rec = DNSAddress(domain, type, class_, ttl, self.read_string(16)) + if type_ == _TYPE_A: + rec = DNSAddress( + domain, type_, class_, ttl, self.read_string(4)) + elif type_ == _TYPE_CNAME or type_ == _TYPE_PTR: + rec = DNSPointer( + domain, type_, class_, ttl, self.read_name()) + elif type_ == _TYPE_TXT: + rec = DNSText( + domain, type_, class_, ttl, self.read_string(length)) + elif type_ == _TYPE_SRV: + rec = DNSService( + domain, type_, class_, ttl, + self.read_unsigned_short(), self.read_unsigned_short(), + self.read_unsigned_short(), self.read_name()) + elif type_ == _TYPE_HINFO: + rec = DNSHinfo( + domain, type_, class_, ttl, + self.read_character_string(), self.read_character_string()) + elif type_ == _TYPE_AAAA: + rec = DNSAddress( + domain, type_, class_, ttl, self.read_string(16)) else: # Try to ignore types we don't know about # Skip the payload for the resource record so the next @@ -553,7 +568,7 @@ def read_name(self): """Reads a domain name from the packet""" result = '' off = self.offset - next = -1 + next_ = -1 first = off while True: @@ -566,8 +581,8 @@ def read_name(self): result = ''.join((result, self.read_utf(off, length) + '.')) off += length elif t == 0xC0: - if next < 0: - next = off + 1 + if next_ < 0: + next_ = off + 1 off = ((length & 0x3F) << 8) | indexbytes(self.data, off) if off >= first: # TODO raise more specific exception @@ -577,8 +592,8 @@ def read_name(self): # TODO raise more specific exception raise Exception("Bad domain name at %s" % (off,)) - if next >= 0: - self.offset = next + if next_ >= 0: + self.offset = next_ else: self.offset = off @@ -626,9 +641,9 @@ def add_additional_answer(self, record): """Adds an additional answer""" self.additionals.append(record) - def pack(self, format, value): - self.data.append(struct.pack(format, value)) - self.size += struct.calcsize(format) + def pack(self, format_, value): + self.data.append(struct.pack(format_, value)) + self.size += struct.calcsize(format_) def write_byte(self, value): """Writes a single byte to the packet""" @@ -662,6 +677,14 @@ def write_utf(self, s): self.write_byte(length) self.write_string(utfstr) + def write_character_string(self, value): + assert isinstance(value, bytes) + length = len(value) + if length > 256: + raise NamePartTooLongException + self.write_byte(length) + self.write_string(value) + def write_name(self, name): """Writes a domain name to the packet""" @@ -768,14 +791,16 @@ def get(self, entry): matching entry.""" try: list_ = self.cache[entry.key] - return list_[list_.index(entry)] + for cached_entry in list_: + if entry.__eq__(cached_entry): + return cached_entry except (KeyError, ValueError): return None - def get_by_details(self, name, type, class_): + def get_by_details(self, name, type_, class_): """Gets an entry by details. Will return None if there is no matching entry.""" - entry = DNSEntry(name, type, class_) + entry = DNSEntry(name, type_, class_) return self.get(entry) def entries_with_name(self, name): @@ -790,7 +815,7 @@ def entries(self): if not self.cache: return [] else: - # copy the cache before running the reduce, to avoid size change during iteration + # avoid size change during iteration by copying the cache values = list(self.cache.values()) return reduce(lambda a, b: a + b, values) @@ -809,7 +834,7 @@ class Engine(threading.Thread): """ def __init__(self, zc): - threading.Thread.__init__(self) + threading.Thread.__init__(self, name='zeroconf-Engine') self.daemon = True self.zc = zc self.readers = {} # maps socket to reader @@ -818,43 +843,37 @@ def __init__(self, zc): self.start() def run(self): - while not _GLOBAL_DONE: - rs = self.get_readers() - if len(rs) == 0: - # No sockets to manage, but we wait for the timeout - # or addition of a socket - # - with self.condition: + while not self.zc.done: + with self.condition: + rs = self.readers.keys() + if len(rs) == 0: + # No sockets to manage, but we wait for the timeout + # or addition of a socket self.condition.wait(self.timeout) - else: + + if len(rs) != 0: try: rr, wr, er = select.select(rs, [], [], self.timeout) - for socket_ in rr: - try: - self.readers[socket_].handle_read(socket_) - except Exception as e: # TODO stop catching all Exceptions - log.exception('Unknown error, possibly benign: %r', e) - except Exception as e: # TODO stop catching all Exceptions - log.exception('Unknown error, possibly benign: %r', e) - - def get_readers(self): - result = [] - with self.condition: - result = self.readers.keys() - return result - - def add_reader(self, reader, socket): - with self.condition: - self.readers[socket] = reader - self.condition.notify() - - def del_reader(self, socket): + if not self.zc.done: + for socket_ in rr: + reader = self.readers.get(socket_) + if reader: + reader.handle_read(socket_) + + except socket.error as e: + # If the socket was closed by another thread, during + # shutdown, ignore it and exit + if e.errno != socket.EBADF or not self.zc.done: + raise + + def add_reader(self, reader, socket_): with self.condition: - del self.readers[socket] + self.readers[socket_] = reader self.condition.notify() - def notify(self): + def del_reader(self, socket_): with self.condition: + del self.readers[socket_] self.condition.notify() @@ -865,24 +884,15 @@ class Listener(object): to cache information as it arrives. It requires registration with an Engine object in order to have - the read() method called when a socket is availble for reading.""" + the read() method called when a socket is available for reading.""" def __init__(self, zc): self.zc = zc + self.data = None def handle_read(self, socket_): - try: - data, (addr, port) = socket_.recvfrom(_MAX_MSG_ABSOLUTE) - except socket.error as e: - # If the socket was closed by another thread -- which happens - # regularly on shutdown -- an EBADF exception is thrown here. - # Ignore it. - if e.errno == socket.EBADF: - return - else: - raise e - else: - log.debug('Received %r from %r:%r', data, addr, port) + data, (addr, port) = socket_.recvfrom(_MAX_MSG_ABSOLUTE) + log.debug('Received %r from %r:%r', data, addr, port) self.data = data msg = DNSIncoming(data) @@ -907,7 +917,7 @@ class Reaper(threading.Thread): have expired.""" def __init__(self, zc): - threading.Thread.__init__(self) + threading.Thread.__init__(self, name='zeroconf-Reaper') self.daemon = True self.zc = zc self.start() @@ -915,7 +925,7 @@ def __init__(self, zc): def run(self): while True: self.zc.wait(10 * 1000) - if _GLOBAL_DONE: + if self.zc.done: return now = current_time_millis() for record in self.zc.cache.entries(): @@ -962,7 +972,8 @@ class ServiceBrowser(threading.Thread): def __init__(self, zc, type_, handlers=None, listener=None): """Creates a browser for a specific type""" assert handlers or listener, 'You need to specify at least one handler' - threading.Thread.__init__(self) + threading.Thread.__init__(self, + name='zeroconf-ServiceBrowser' + type_) self.daemon = True self.zc = zc self.type = type_ @@ -1040,14 +1051,15 @@ def enqueue_callback(state_change, name): def cancel(self): self.done = True - self.zc.notify_all() + self.zc.remove_listener(self) + self.join() def run(self): while True: now = current_time_millis() if len(self._handlers_to_call) == 0 and self.next_time > now: self.zc.wait(self.next_time - now) - if _GLOBAL_DONE or self.done: + if self.zc.done or self.done: return now = current_time_millis() @@ -1061,7 +1073,7 @@ def run(self): self.next_time = now + self.delay self.delay = min(20 * 1000, self.delay * 2) - if len(self._handlers_to_call) > 0: + if len(self._handlers_to_call) > 0 and not self.zc.done: handler = self._handlers_to_call.pop(0) handler(self.zc) @@ -1070,7 +1082,7 @@ class ServiceInfo(object): """Service information""" - def __init__(self, type, name, address=None, port=None, weight=0, + def __init__(self, type_, name, address=None, port=None, weight=0, priority=0, properties=None, server=None): """Create a service description. @@ -1084,9 +1096,9 @@ def __init__(self, type, name, address=None, port=None, weight=0, bytes for the text field) server: fully qualified name for service host (defaults to name)""" - if not name.endswith(type): + if not name.endswith(type_): raise BadTypeInNameException - self.type = type + self.type = type_ self.name = name self.address = address self.port = port @@ -1096,6 +1108,7 @@ def __init__(self, type, name, address=None, port=None, weight=0, self.server = server else: self.server = name + self._properties = {} self._set_properties(properties) @property @@ -1106,7 +1119,7 @@ def _set_properties(self, properties): """Sets properties and text of this info from a dictionary""" if isinstance(properties, dict): self._properties = properties - list = [] + list_ = [] result = b'' for key, value in iteritems(properties): if isinstance(key, text_type): @@ -1125,8 +1138,8 @@ def _set_properties(self, properties): suffix = b'false' else: suffix = b'' - list.append(b'='.join((key, suffix))) - for item in list: + list_.append(b'='.join((key, suffix))) + for item in list_: result = b''.join((result, int2byte(len(item)), item)) self.text = result else: @@ -1185,8 +1198,9 @@ def update_record(self, zc, now, record): self.weight = record.weight self.priority = record.priority # self.address = None - self.update_record(zc, now, - zc.cache.get_by_details(self.server, _TYPE_A, _CLASS_IN)) + self.update_record( + zc, now, zc.cache.get_by_details( + self.server, _TYPE_A, _CLASS_IN)) elif record.type == _TYPE_TXT: if record.name == self.name: self._set_text(record.text) @@ -1197,41 +1211,58 @@ def request(self, zc, timeout): """ now = current_time_millis() delay = _LISTENER_TIME - next = now + delay + next_ = now + delay last = now + timeout - result = False + + record_types_for_check_cache = [ + (_TYPE_SRV, _CLASS_IN), + (_TYPE_TXT, _CLASS_IN), + ] + if self.server is not None: + record_types_for_check_cache.append((_TYPE_A, _CLASS_IN)) + for record_type in record_types_for_check_cache: + cached = zc.cache.get_by_details(self.name, *record_type) + if cached: + self.update_record(zc, now, cached) + + if None not in (self.server, self.address, self.text): + return True + try: zc.add_listener(self, DNSQuestion(self.name, _TYPE_ANY, _CLASS_IN)) - while (self.server is None or self.address is None or - self.text is None): + while None in (self.server, self.address, self.text): if last <= now: return False - if next <= now: + if next_ <= now: out = DNSOutgoing(_FLAGS_QR_QUERY) - out.add_question(DNSQuestion(self.name, _TYPE_SRV, - _CLASS_IN)) - out.add_answer_at_time(zc.cache.get_by_details(self.name, - _TYPE_SRV, _CLASS_IN), now) - out.add_question(DNSQuestion(self.name, _TYPE_TXT, - _CLASS_IN)) - out.add_answer_at_time(zc.cache.get_by_details(self.name, - _TYPE_TXT, _CLASS_IN), now) + out.add_question( + DNSQuestion(self.name, _TYPE_SRV, _CLASS_IN)) + out.add_answer_at_time( + zc.cache.get_by_details( + self.name, _TYPE_SRV, _CLASS_IN), now) + + out.add_question( + DNSQuestion(self.name, _TYPE_TXT, _CLASS_IN)) + out.add_answer_at_time( + zc.cache.get_by_details( + self.name, _TYPE_TXT, _CLASS_IN), now) + if self.server is not None: - out.add_question(DNSQuestion(self.server, - _TYPE_A, _CLASS_IN)) - out.add_answer_at_time(zc.cache.get_by_details(self.server, - _TYPE_A, _CLASS_IN), now) + out.add_question( + DNSQuestion(self.server, _TYPE_A, _CLASS_IN)) + out.add_answer_at_time( + zc.cache.get_by_details( + self.server, _TYPE_A, _CLASS_IN), now) zc.send(out) - next = now + delay - delay = delay * 2 + next_ = now + delay + delay *= 2 - zc.wait(min(next, last) - now) + zc.wait(min(next_, last) - now) now = current_time_millis() - result = True finally: zc.remove_listener(self) - return result + return True def __eq__(self, other): """Tests equality of service name""" @@ -1257,6 +1288,46 @@ def __repr__(self): ) +class ZeroconfServiceTypes(object): + """ + Return all of the advertised services on any local networks + """ + def __init__(self): + self.found_services = set() + + def add_service(self, zc, type_, name): + self.found_services.add(name) + + def remove_service(self, zc, type_, name): + pass + + @classmethod + def find(cls, zc=None, timeout=5): + """ + Return all of the advertised services on any local networks. + + :param zc: Zeroconf() instance. Pass in if already have an + instance running or if non-default interfaces are needed + :param timeout: seconds to wait for any responses + :return: tuple of service type strings + """ + local_zc = zc or Zeroconf() + listener = cls() + browser = ServiceBrowser( + local_zc, '_services._dns-sd._udp.local.', listener=listener) + + # wait for responses + time.sleep(timeout) + + # close down anything we opened + if zc is None: + local_zc.close() + else: + browser.cancel() + + return tuple(sorted(listener.found_services)) + + @enum.unique class InterfaceChoice(enum.Enum): Default = 1 @@ -1307,7 +1378,8 @@ def new_socket(): else: try: s.setsockopt(socket.SOL_SOCKET, reuseport, 1) - except (OSError, socket.error) as err: # OSError on python 3, socket.error on python 2 + except (OSError, socket.error) as err: + # OSError on python 3, socket.error on python 2 if not err.errno == errno.ENOPROTOOPT: raise @@ -1343,8 +1415,8 @@ def __init__( :type interfaces: :class:`InterfaceChoice` or sequence of ip addresses """ - global _GLOBAL_DONE - _GLOBAL_DONE = False + # hook for threads + self._GLOBAL_DONE = False self._listen_socket = new_socket() interfaces = normalize_interface_choice(interfaces, socket.AF_INET) @@ -1365,8 +1437,8 @@ def __init__( ) elif get_errno(e) == errno.EADDRNOTAVAIL: log.info( - 'Address not available when adding %s to multicast group, ' - 'it is expected to happen on some systems', i, + 'Address not available when adding %s to multicast ' + 'group, it is expected to happen on some systems', i, ) continue else: @@ -1379,7 +1451,7 @@ def __init__( self._respond_sockets.append(respond_socket) self.listeners = [] - self.browsers = [] + self.browsers = {} self.services = {} self.servicetypes = {} @@ -1392,39 +1464,49 @@ def __init__( self.engine.add_reader(self.listener, self._listen_socket) self.reaper = Reaper(self) + self.debug = None + + @property + def done(self): + return self._GLOBAL_DONE + def wait(self, timeout): """Calling thread waits for a given number of milliseconds or until notified.""" with self.condition: - self.condition.wait(timeout / 1000) + self.condition.wait(timeout / 1000.0) def notify_all(self): """Notifies all waiting threads""" with self.condition: self.condition.notify_all() - def get_service_info(self, type, name, timeout=3000): + def get_service_info(self, type_, name, timeout=3000): """Returns network's service information for a particular name and type, or None if no service matches by the timeout, which defaults to 3 seconds.""" - info = ServiceInfo(type, name) + info = ServiceInfo(type_, name) if info.request(self, timeout): return info return None - def add_service_listener(self, type, listener): + def add_service_listener(self, type_, listener): """Adds a listener for a particular service type. This object will then have its update_record method called when information arrives for that type.""" self.remove_service_listener(listener) - self.browsers.append(ServiceBrowser(self, type, listener)) + self.browsers[listener] = ServiceBrowser(self, type_, listener) def remove_service_listener(self, listener): """Removes a listener from the set that is currently listening.""" - for browser in self.browsers: - if browser.listener == listener: - browser.cancel() - del browser + if listener in self.browsers: + self.browsers[listener].cancel() + del self.browsers[listener] + + def remove_all_service_listeners(self): + """Removes a listener from the set that is currently listening.""" + for listener in [k for k in self.browsers]: + self.remove_service_listener(listener) def register_service(self, info, ttl=_DNS_TTL): """Registers service information to the network with a default TTL @@ -1446,16 +1528,19 @@ def register_service(self, info, ttl=_DNS_TTL): now = current_time_millis() continue out = DNSOutgoing(_FLAGS_QR_RESPONSE | _FLAGS_AA) - out.add_answer_at_time(DNSPointer(info.type, _TYPE_PTR, - _CLASS_IN, ttl, info.name), 0) - out.add_answer_at_time(DNSService(info.name, _TYPE_SRV, - _CLASS_IN, ttl, info.priority, info.weight, info.port, - info.server), 0) - out.add_answer_at_time(DNSText(info.name, _TYPE_TXT, _CLASS_IN, - ttl, info.text), 0) + out.add_answer_at_time( + DNSPointer(info.type, _TYPE_PTR, _CLASS_IN, ttl, info.name), 0) + out.add_answer_at_time( + DNSService(info.name, _TYPE_SRV, _CLASS_IN, + ttl, info.priority, info.weight, info.port, + info.server), 0) + + out.add_answer_at_time( + DNSText(info.name, _TYPE_TXT, _CLASS_IN, ttl, info.text), 0) if info.address: - out.add_answer_at_time(DNSAddress(info.server, _TYPE_A, - _CLASS_IN, ttl, info.address), 0) + out.add_answer_at_time( + DNSAddress(info.server, _TYPE_A, _CLASS_IN, + ttl, info.address), 0) self.send(out) i += 1 next_time += _REGISTER_TIME @@ -1479,16 +1564,18 @@ def unregister_service(self, info): now = current_time_millis() continue out = DNSOutgoing(_FLAGS_QR_RESPONSE | _FLAGS_AA) - out.add_answer_at_time(DNSPointer(info.type, _TYPE_PTR, - _CLASS_IN, 0, info.name), 0) - out.add_answer_at_time(DNSService(info.name, _TYPE_SRV, - _CLASS_IN, 0, info.priority, info.weight, info.port, - info.name), 0) - out.add_answer_at_time(DNSText(info.name, _TYPE_TXT, _CLASS_IN, - 0, info.text), 0) + out.add_answer_at_time( + DNSPointer(info.type, _TYPE_PTR, _CLASS_IN, 0, info.name), 0) + out.add_answer_at_time( + DNSService(info.name, _TYPE_SRV, _CLASS_IN, 0, + info.priority, info.weight, info.port, info.name), 0) + out.add_answer_at_time( + DNSText(info.name, _TYPE_TXT, _CLASS_IN, 0, info.text), 0) + if info.address: - out.add_answer_at_time(DNSAddress(info.server, _TYPE_A, - _CLASS_IN, 0, info.address), 0) + out.add_answer_at_time( + DNSAddress(info.server, _TYPE_A, _CLASS_IN, 0, + info.address), 0) self.send(out) i += 1 next_time += _UNREGISTER_TIME @@ -1506,16 +1593,17 @@ def unregister_all_services(self): continue out = DNSOutgoing(_FLAGS_QR_RESPONSE | _FLAGS_AA) for info in self.services.values(): - out.add_answer_at_time(DNSPointer(info.type, _TYPE_PTR, - _CLASS_IN, 0, info.name), 0) - out.add_answer_at_time(DNSService(info.name, _TYPE_SRV, - _CLASS_IN, 0, info.priority, info.weight, - info.port, info.server), 0) - out.add_answer_at_time(DNSText(info.name, _TYPE_TXT, - _CLASS_IN, 0, info.text), 0) + out.add_answer_at_time(DNSPointer( + info.type, _TYPE_PTR, _CLASS_IN, 0, info.name), 0) + out.add_answer_at_time(DNSService( + info.name, _TYPE_SRV, _CLASS_IN, 0, + info.priority, info.weight, info.port, info.server), 0) + out.add_answer_at_time(DNSText( + info.name, _TYPE_TXT, _CLASS_IN, 0, info.text), 0) if info.address: - out.add_answer_at_time(DNSAddress(info.server, - _TYPE_A, _CLASS_IN, 0, info.address), 0) + out.add_answer_at_time(DNSAddress( + info.server, _TYPE_A, _CLASS_IN, 0, + info.address), 0) self.send(out) i += 1 next_time += _UNREGISTER_TIME @@ -1532,8 +1620,8 @@ def check_service(self, info): not record.is_expired(now) and record.alias == info.name): if info.name.find('.') < 0: - info.name = '%s.[%s:%s].%s' % (info.name, - info.address, info.port, info.type) + info.name = '%s.[%s:%s].%s' % ( + info.name, info.address, info.port, info.type) self.check_service(info) return @@ -1545,8 +1633,8 @@ def check_service(self, info): out = DNSOutgoing(_FLAGS_QR_QUERY | _FLAGS_AA) self.debug = out out.add_question(DNSQuestion(info.type, _TYPE_PTR, _CLASS_IN)) - out.add_authorative_answer(DNSPointer(info.type, _TYPE_PTR, - _CLASS_IN, _DNS_TTL, info.name)) + out.add_authorative_answer(DNSPointer( + info.type, _TYPE_PTR, _CLASS_IN, _DNS_TTL, info.name)) self.send(out) i += 1 next_time += _CHECK_TIME @@ -1591,10 +1679,10 @@ def handle_response(self, msg): entry = self.cache.get(record) if entry is not None: entry.reset_ttl(record) - record = entry else: self.cache.add(record) + for record in msg.answers: self.update_record(now, record) def handle_query(self, msg, addr, port): @@ -1615,16 +1703,16 @@ def handle_query(self, msg, addr, port): for stype in self.servicetypes.keys(): if out is None: out = DNSOutgoing(_FLAGS_QR_RESPONSE | _FLAGS_AA) - out.add_answer(msg, - DNSPointer("_services._dns-sd._udp.local.", - _TYPE_PTR, _CLASS_IN, _DNS_TTL, stype)) + out.add_answer(msg, DNSPointer( + "_services._dns-sd._udp.local.", _TYPE_PTR, + _CLASS_IN, _DNS_TTL, stype)) for service in self.services.values(): if question.name == service.type: if out is None: out = DNSOutgoing(_FLAGS_QR_RESPONSE | _FLAGS_AA) - out.add_answer(msg, - DNSPointer(service.type, _TYPE_PTR, - _CLASS_IN, _DNS_TTL, service.name)) + out.add_answer(msg, DNSPointer( + service.type, _TYPE_PTR, + _CLASS_IN, _DNS_TTL, service.name)) else: try: if out is None: @@ -1634,27 +1722,28 @@ def handle_query(self, msg, addr, port): if question.type in (_TYPE_A, _TYPE_ANY): for service in self.services.values(): if service.server == question.name.lower(): - out.add_answer(msg, DNSAddress(question.name, - _TYPE_A, _CLASS_IN | _CLASS_UNIQUE, - _DNS_TTL, service.address)) + out.add_answer(msg, DNSAddress( + question.name, _TYPE_A, + _CLASS_IN | _CLASS_UNIQUE, + _DNS_TTL, service.address)) service = self.services.get(question.name.lower(), None) if not service: continue if question.type in (_TYPE_SRV, _TYPE_ANY): - out.add_answer(msg, DNSService(question.name, - _TYPE_SRV, _CLASS_IN | _CLASS_UNIQUE, - _DNS_TTL, service.priority, service.weight, - service.port, service.server)) + out.add_answer(msg, DNSService( + question.name, _TYPE_SRV, _CLASS_IN | _CLASS_UNIQUE, + _DNS_TTL, service.priority, service.weight, + service.port, service.server)) if question.type in (_TYPE_TXT, _TYPE_ANY): - out.add_answer(msg, DNSText(question.name, - _TYPE_TXT, _CLASS_IN | _CLASS_UNIQUE, - _DNS_TTL, service.text)) + out.add_answer(msg, DNSText( + question.name, _TYPE_TXT, _CLASS_IN | _CLASS_UNIQUE, + _DNS_TTL, service.text)) if question.type == _TYPE_SRV: - out.add_additional_answer(DNSAddress(service.server, - _TYPE_A, _CLASS_IN | _CLASS_UNIQUE, - _DNS_TTL, service.address)) + out.add_additional_answer(DNSAddress( + service.server, _TYPE_A, _CLASS_IN | _CLASS_UNIQUE, + _DNS_TTL, service.address)) except Exception as e: # TODO stop catching all Exceptions log.exception('Unknown error, possibly benign: %r', e) @@ -1667,6 +1756,8 @@ def send(self, out, addr=_MDNS_ADDR, port=_MDNS_PORT): packet = out.packet() log.debug('Sending %r as %r...', out, packet) for s in self._respond_sockets: + if self._GLOBAL_DONE: + return bytes_sent = s.sendto(packet, 0, (addr, port)) if bytes_sent != len(packet): raise Error( @@ -1676,11 +1767,19 @@ def send(self, out, addr=_MDNS_ADDR, port=_MDNS_PORT): def close(self): """Ends the background threads, and prevent this instance from servicing further queries.""" - global _GLOBAL_DONE - if not _GLOBAL_DONE: - _GLOBAL_DONE = True - self.notify_all() - self.engine.notify() + if not self._GLOBAL_DONE: + self._GLOBAL_DONE = True + # remove service listeners + self.remove_all_service_listeners() self.unregister_all_services() - for s in [self._listen_socket] + self._respond_sockets: + + # shutdown recv socket and thread + self.engine.del_reader(self._listen_socket) + self._listen_socket.close() + self.engine.join() + + # shutdown the rest + self.notify_all() + self.reaper.join() + for s in self._respond_sockets: s.close()