- Fix #17: Add python module example from Jan Janak, that is a

plugin for the Unbound DNS resolver to resolve DNS records in
  multicast DNS [RFC 6762] via Avahi.


git-svn-id: file:///svn/unbound/trunk@5151 be551aaa-1e26-0410-a405-d3ace91eadb9

doc/Changelog

@@ -2,6 +2,9 @@
 	- Fix to use event_assign with libevent for thread-safety.
 	- verbose information about auth zone lookup process, also lookup
 	  start, timeout and fail.
+	- Fix #17: Add python module example from Jan Janak, that is a
+	  plugin for the Unbound DNS resolver to resolve DNS records in
+	  multicast DNS [RFC 6762] via Avahi.
 
 5 April 2019: Wouter
 	- Fix to reinit event structure for accepted TCP (and TLS) sockets.

pythonmod/examples/avahi-resolver.py

@@ -0,0 +1,567 @@
+#!/usr/bin/env python3
+#
+# A plugin for the Unbound DNS resolver to resolve DNS records in
+# multicast DNS [RFC 6762] via Avahi.
+#
+# Copyright (C) 2018-2019 Internet Real-Time Lab, Columbia University
+# http://www.cs.columbia.edu/irt/
+#
+# Written by Jan Janak <janakj@cs.columbia.edu>
+#
+# Permission is hereby granted, free of charge, to any person
+# obtaining a copy of this software and associated documentation files
+# (the "Software"), to deal in the Software without restriction,
+# including without limitation the rights to use, copy, modify, merge,
+# publish, distribute, sublicense, and/or sell copies of the Software,
+# and to permit persons to whom the Software is furnished to do so,
+# subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be
+# included in all copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
+# BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
+# ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+# CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+#
+#
+# Dependendies:
+#   Unbound with pythonmodule configured for Python 3
+#   dnspython [http://www.dnspython.org]
+#   pydbus [https://github.com/LEW21/pydbus]
+#
+# To enable Python 3 support, configure Unbound as follows:
+#   PYTHON_VERSION=3 ./configure --with-pythonmodule
+#
+# The plugin in meant to be used as a fallback resolver that resolves
+# records in multicast DNS if the upstream server cannot be reached or
+# provides no answer (NXDOMAIN).
+#
+# mDNS requests for negative records, i.e., records for which Avahi
+# returns no answer (NXDOMAIN), are expensive. Since there is no
+# single authoritative server in mDNS, such requests terminate only
+# via a timeout. The timeout is about a second (if MDNS_TIMEOUT is not
+# configured), or the value configured via MDNS_TIMEOUT. The
+# corresponding Unbound thread will be blocked for this amount of
+# time. For this reason, it is important to configure an appropriate
+# number of threads in unbound.conf and limit the RR types and names
+# that will be resolved via Avahi via the environment variables
+# described later.
+#
+# An example unbound.conf with the plugin enabled:
+#
+# | server:
+# |   module-config: "validator python iterator"
+# |   num-threads: 32
+# |   cache-max-negative-ttl: 60
+# |   cache-max-ttl: 60
+#
+#
+# The plugin can also be run interactively. Provide the name and
+# record type to be resolved as command line arguments and the
+# resolved record will be printed to standard output:
+#
+#   $ ./avahi-resolver.py voip-phx4.phxnet.org A
+#   voip-phx4.phxnet.org. 120 IN A 10.4.3.2
+#
+#
+# The behavior of the plugin can be controlled via the following
+# environment variables:
+#
+# DBUS_SYSTEM_BUS_ADDRESS
+#
+# The address of the system DBus bus, in the format expected by DBus,
+# e.g., unix:path=/run/avahi/system-bus.sock
+#
+#
+# DEBUG
+#
+# Set this environment variable to "yes", "true", "on", or "1" to
+# enable debugging. In debugging mode, the plugin will output a lot
+# more information about what it is doing either to the standard
+# output (when run interactively) or to Unbound via log_info and
+# log_error.
+#
+# By default debugging is disabled.
+#
+#
+# MDNS_TTL
+#
+# Avahi does not provide the TTL value for the records it returns.
+# This environment variable can be used to configure the TTL value for
+# such records.
+#
+# The default value is 120 seconds.
+#
+#
+# MDNS_TIMEOUT
+#
+# The maximum amount of time (in milliseconds) an Avahi request is
+# allowed to run. This value sets the time it takes to resolve
+# negative (non-existent) records in Avahi. If unset, the request
+# terminates when Avahi sends the "AllForNow" signal, telling the
+# client that more records are unlikely to arrive. This takes roughly
+# about one second. You may need to configure a longer value here on
+# slower networks, e.g., networks that relay mDNS packets such as
+# MANETs.
+#
+#
+# MDNS_GETONE
+#
+# If set to "true", "1", or "on", an Avahi request will terminate as
+# soon as at least one record has been found. If there are multiple
+# nodes in the mDNS network publishing the same record, only one (or
+# subset) will be returned.
+#
+# If set to "false", "0", or "off", the plugin will gather records for
+# MDNS_TIMEOUT and return all records found. This is only useful in
+# networks where multiple nodes are known to publish different records
+# under the same name and the client needs to be able to obtain them
+# all. When configured this way, all Avahi requests will always take
+# MDNS_TIMEOUT to complete!
+#
+# This option is set to true by default.
+#
+#
+# MDNS_REJECT_TYPES
+#
+# A comma-separated list of record types that will NOT be resolved in
+# mDNS via Avahi. Use this environment variable to prevent specific
+# record types from being resolved via Avahi. For example, if your
+# network does not support IPv6, you can put AAAA on this list.
+#
+# The default value is an empty list.
+#
+# Example: MDNS_REJECT_TYPES=aaaa,mx,soa
+#
+#
+# MDNS_ACCEPT_TYPES
+#
+# If set, a record type will be resolved via Avahi if and only if it
+# is present on this comma-separated list. In other words, this is a
+# whitelist.
+#
+# The default value is an empty list which means all record types will
+# be resolved via Avahi.
+#
+# Example: MDNS_ACCEPT_TYPES=a,ptr,txt,srv,aaaa,cname
+#
+#
+# MDNS_REJECT_NAMES
+#
+# If the name being resolved matches the regular expression in this
+# environment variable, the name will NOT be resolved via Avahi. In
+# other words, this environment variable provides a blacklist.
+#
+# The default value is empty--no names will be reject.
+#
+# Example: MDNS_REJECT_NAMES=(^|\.)example\.com\.$
+#
+#
+# MDNS_ACCEPT_NAMES
+#
+# If set to a regular expression, a name will be resolved via Avahi if
+# and only if it matches the regular expression. In other words, this
+# variable provides a whitelist.
+#
+# The default value is empty--all names will be resolved via Avahi.
+#
+# Example: MDNS_ACCEPT_NAMES=^.*\.example\.com\.$
+#
+
+import os
+import re
+import array
+import threading
+import traceback
+import dns.rdata
+import dns.rdatatype
+import dns.rdataclass
+from queue import Queue
+from gi.repository import GLib
+from pydbus import SystemBus
+
+
+IF_UNSPEC    = -1
+PROTO_UNSPEC = -1
+
+sysbus = None
+avahi = None
+trampoline = dict()
+thread_local = threading.local()
+dbus_thread = None
+dbus_loop = None
+
+
+def str2bool(v):
+    if v.lower() in ['false', 'no', '0', 'off', '']:
+        return False
+    return True
+
+
+def dbg(msg):
+    if DEBUG != False:
+        log_info('avahi-resolver: %s' % msg)
+
+
+#
+# Although pydbus has an internal facility for handling signals, we
+# cannot use that with Avahi. When responding from an internal cache,
+# Avahi sends the first signal very quickly, before pydbus has had a
+# chance to subscribe for the signal. This will result in lost signal
+# and missed data:
+#
+# https://github.com/LEW21/pydbus/issues/87
+#
+# As a workaround, we subscribe to all signals before creating a
+# record browser and do our own signal matching and dispatching via
+# the following function.
+#
+def signal_dispatcher(connection, sender, path, interface, name, args):
+    o = trampoline.get(path, None)
+    if o is None:
+        return
+
+    if   name == 'ItemNew':    o.itemNew(*args)
+    elif name == 'ItemRemove': o.itemRemove(*args)
+    elif name == 'AllForNow':  o.allForNow(*args)
+    elif name == 'Failure':    o.failure(*args)
+
+
+class RecordBrowser:
+    def __init__(self, callback, name, type_, timeout=None, getone=True):
+        self.callback = callback
+        self.records = []
+        self.error = None
+        self.getone = getone
+
+        self.timer = None if timeout is None else GLib.timeout_add(timeout, self.timedOut)
+
+        self.browser_path = avahi.RecordBrowserNew(IF_UNSPEC, PROTO_UNSPEC, name, dns.rdataclass.IN, type_, 0)
+        trampoline[self.browser_path] = self
+        self.browser = sysbus.get('.Avahi', self.browser_path)
+        self.dbg('Created RecordBrowser(name=%s, type=%s, getone=%s, timeout=%s)'
+                   % (name, dns.rdatatype.to_text(type_), getone, timeout))
+
+    def dbg(self, msg):
+        dbg('[%s] %s' % (self.browser_path, msg))
+
+    def _done(self):
+        del trampoline[self.browser_path]
+        self.dbg('Freeing')
+        self.browser.Free()
+
+        if self.timer is not None:
+            self.dbg('Removing timer')
+            GLib.source_remove(self.timer)
+
+        self.callback(self.records, self.error)
+
+    def itemNew(self, interface, protocol, name, class_, type_, rdata, flags):
+        self.dbg('Got signal ItemNew')
+        self.records.append((name, class_, type_, rdata))
+        if self.getone:
+            self._done()
+
+    def itemRemove(self, interface, protocol, name, class_, type_, rdata, flags):
+        self.dbg('Got signal ItemRemove')
+        self.records.remove((name, class_, type_, rdata))
+
+    def failure(self, error):
+        self.dbg('Got signal Failure')
+        self.error = Exception(error)
+        self._done()
+
+    def allForNow(self):
+        self.dbg('Got signal AllForNow')
+        if self.timer is None:
+            self._done()
+
+    def timedOut(self):
+        self.dbg('Timed out')
+        self._done()
+        return False
+
+
+#
+# This function runs the main event loop for DBus (GLib). This
+# function must be run in a dedicated worker thread.
+#
+def dbus_main():
+    global sysbus, avahi, dbus_loop
+
+    dbg('Connecting to system DBus')
+    sysbus = SystemBus()
+
+    dbg('Subscribing to .Avahi.RecordBrowser signals')
+    sysbus.con.signal_subscribe('org.freedesktop.Avahi',
+        'org.freedesktop.Avahi.RecordBrowser',
+        None, None, None, 0, signal_dispatcher)
+
+    avahi = sysbus.get('.Avahi', '/')
+
+    dbg("Connected to Avahi Daemon: %s (API %s) [%s]"
+             % (avahi.GetVersionString(), avahi.GetAPIVersion(), avahi.GetHostNameFqdn()))
+
+    dbg('Starting DBus main loop')
+    dbus_loop = GLib.MainLoop()
+    dbus_loop.run()
+
+
+#
+# This function must be run in the DBus worker thread. It creates a
+# new RecordBrowser instance and once it has finished doing it thing,
+# it will send the result back to the original thread via the queue.
+#
+def start_resolver(queue, *args, **kwargs):
+    try:
+        RecordBrowser(lambda *v: queue.put_nowait(v), *args, **kwargs)
+    except Exception as e:
+        queue.put_nowait((None, e))
+
+    return False
+
+
+#
+# To resolve a request, we setup a queue, post a task to the DBus
+# worker thread, and wait for the result (or error) to arrive over the
+# queue. If the worker thread reports an error, raise the error as an
+# exception.
+#
+def resolve(*args, **kwargs):
+    try:
+        queue = thread_local.queue
+    except AttributeError:
+        dbg('Creating new per-thread queue')
+        queue = Queue()
+        thread_local.queue = queue
+
+    GLib.idle_add(lambda: start_resolver(queue, *args, **kwargs))
+
+    records, error = queue.get()
+    queue.task_done()
+
+    if error is not None:
+        raise error
+
+    return records
+
+
+def parse_type_list(lst):
+    return list(map(dns.rdatatype.from_text, [v.strip() for v in lst.split(',') if len(v)]))
+
+
+def init(*args, **kwargs):
+    global dbus_thread, DEBUG
+    global MDNS_TTL, MDNS_GETONE, MDNS_TIMEOUT
+    global MDNS_REJECT_TYPES, MDNS_ACCEPT_TYPES
+    global MDNS_REJECT_NAMES, MDNS_ACCEPT_NAMES
+
+    DEBUG = str2bool(os.environ.get('DEBUG', str(False)))
+
+    MDNS_TTL = int(os.environ.get('MDNS_TTL', 120))
+    dbg("TTL for records from Avahi: %d" % MDNS_TTL)
+
+    MDNS_REJECT_TYPES = parse_type_list(os.environ.get('MDNS_REJECT_TYPES', ''))
+    if MDNS_REJECT_TYPES:
+        dbg('Types NOT resolved via Avahi: %s' % MDNS_REJECT_TYPES)
+
+    MDNS_ACCEPT_TYPES = parse_type_list(os.environ.get('MDNS_ACCEPT_TYPES', ''))
+    if MDNS_ACCEPT_TYPES:
+        dbg('ONLY resolving the following types via Avahi: %s' % MDNS_ACCEPT_TYPES)
+
+    v = os.environ.get('MDNS_REJECT_NAMES', None)
+    MDNS_REJECT_NAMES = re.compile(v, flags=re.I | re.S) if v is not None else None
+    if MDNS_REJECT_NAMES is not None:
+        dbg('Names NOT resolved via Avahi: %s' % MDNS_REJECT_NAMES.pattern)
+
+    v = os.environ.get('MDNS_ACCEPT_NAMES', None)
+    MDNS_ACCEPT_NAMES = re.compile(v, flags=re.I | re.S) if v is not None else None
+    if MDNS_ACCEPT_NAMES is not None:
+        dbg('ONLY resolving the following names via Avahi: %s' % MDNS_ACCEPT_NAMES.pattern)
+
+    v = os.environ.get('MDNS_TIMEOUT', None)
+    MDNS_TIMEOUT = int(v) if v is not None else None
+    if MDNS_TIMEOUT is not None:
+        dbg('Avahi request timeout: %s' % MDNS_TIMEOUT)
+
+    MDNS_GETONE = str2bool(os.environ.get('MDNS_GETONE', str(True)))
+    dbg('Terminate Avahi requests on first record: %s' % MDNS_GETONE)
+
+    dbus_thread = threading.Thread(target=dbus_main)
+    dbus_thread.daemon = True
+    dbus_thread.start()
+
+
+def deinit(*args, **kwargs):
+    dbus_loop.quit()
+    dbus_thread.join()
+    return True
+
+
+def inform_super(id, qstate, superqstate, qdata):
+    return True
+
+
+def get_rcode(msg):
+    if not msg:
+        return RCODE_SERVFAIL
+
+    return msg.rep.flags & 0xf
+
+
+def rr2text(rec, ttl):
+    name, class_, type_, rdata = rec
+    wire = array.array('B', rdata).tostring()
+    return '%s. %d %s %s %s' % (
+        name,
+        ttl,
+        dns.rdataclass.to_text(class_),
+        dns.rdatatype.to_text(type_),
+        dns.rdata.from_wire(class_, type_, wire, 0, len(wire), None))
+
+
+def operate(id, event, qstate, qdata):
+    qi = qstate.qinfo
+    name = qi.qname_str
+    type_ = qi.qtype
+    type_str = dns.rdatatype.to_text(type_)
+    class_ = qi.qclass
+    class_str = dns.rdataclass.to_text(class_)
+    rc = get_rcode(qstate.return_msg)
+
+    if event == MODULE_EVENT_NEW or event == MODULE_EVENT_PASS:
+        qstate.ext_state[id] = MODULE_WAIT_MODULE
+        return True
+
+    if event != MODULE_EVENT_MODDONE:
+        log_err("avahi-resolver: Unexpected event %d" % event)
+        qstate.ext_state[id] = MODULE_ERROR
+        return True
+
+    qstate.ext_state[id] = MODULE_FINISHED
+
+    # Only resolve via Avahi if we got NXDOMAIn from the upstream DNS
+    # server, or if we could not reach the upstream DNS server. If we
+    # got some records for the name from the upstream DNS server
+    # already, do not resolve the record in Avahi.
+    if rc != RCODE_NXDOMAIN and rc != RCODE_SERVFAIL:
+        return True
+
+    dbg("Got request for '%s %s %s'" % (name, class_str, type_str))
+
+    # Avahi only supports the IN class
+    if class_ != RR_CLASS_IN:
+        dbg('Rejected, Avahi only supports the IN class')
+        return True
+
+    # Avahi does not support meta queries (e.g., ANY)
+    if dns.rdatatype.is_metatype(type_):
+        dbg('Rejected, Avahi does not support the type %s' % type_str)
+        return True
+
+    # If we have a type blacklist and the requested type is on the
+    # list, reject it.
+    if MDNS_REJECT_TYPES and type_ in MDNS_REJECT_TYPES:
+        dbg('Rejected, type %s is on the blacklist' % type_str)
+        return True
+
+    # If we have a type whitelist and if the requested type is not on
+    # the list, reject it.
+    if MDNS_ACCEPT_TYPES and type_ not in MDNS_ACCEPT_TYPES:
+        dbg('Rejected, type %s is not on the whitelist' % type_str)
+        return True
+
+    # If we have a name blacklist and if the requested name matches
+    # the blacklist, reject it.
+    if MDNS_REJECT_NAMES is not None:
+        if MDNS_REJECT_NAMES.search(name):
+            dbg('Rejected, name %s is on the blacklist' % name)
+            return True
+
+    # If we have a name whitelist and if the requested name does not
+    # match the whitelist, reject it.
+    if MDNS_ACCEPT_NAMES is not None:
+        if not MDNS_ACCEPT_NAMES.search(name):
+            dbg('Rejected, name %s is not on the whitelist' % name)
+            return True
+
+    dbg("Resolving '%s %s %s' via Avahi" % (name, class_str, type_str))
+
+    recs = resolve(name, type_, getone=MDNS_GETONE, timeout=MDNS_TIMEOUT)
+
+    if not recs:
+        dbg('Result: Not found (NXDOMAIN)')
+        qstate.return_rcode = RCODE_NXDOMAIN
+        return True
+
+    m = DNSMessage(name, type_, class_, PKT_QR | PKT_RD | PKT_RA)
+    for r in recs:
+        s = rr2text(r, MDNS_TTL)
+        dbg('Result: %s' % s)
+        m.answer.append(s)
+
+    if not m.set_return_msg(qstate):
+        raise Exception("Error in set_return_msg")
+
+    if not storeQueryInCache(qstate, qstate.return_msg.qinfo, qstate.return_msg.rep, 0):
+        raise Exception("Error in storeQueryInCache")
+
+    qstate.return_msg.rep.security = 2
+    qstate.return_rcode = RCODE_NOERROR
+    return True
+
+
+#
+# It does not appear to be sufficient to check __name__ to determine
+# whether we are being run in interactive mode. As a workaround, try
+# to import module unboundmodule and if that fails, assume we're being
+# run in interactive mode.
+#
+try:
+    import unboundmodule
+    embedded = True
+except ImportError:
+    embedded = False
+
+if __name__ == '__main__' and not embedded:
+    import sys
+
+    def log_info(msg):
+        print(msg)
+
+    def log_err(msg):
+        print('ERROR: %s' % msg, file=sys.stderr)
+
+    if len(sys.argv) != 3:
+        print('Usage: %s <name> <rr_type>' % sys.argv[0])
+        sys.exit(2)
+
+    name = sys.argv[1]
+    type_str = sys.argv[2]
+
+    try:
+        type_ = dns.rdatatype.from_text(type_str)
+    except dns.rdatatype.UnknownRdatatype:
+        log_err('Unsupported DNS record type "%s"' % type_str)
+        sys.exit(2)
+
+    if dns.rdatatype.is_metatype(type_):
+        log_err('Meta record type "%s" cannot be resolved via Avahi' % type_str)
+        sys.exit(2)
+
+    init()
+    try:
+        recs = resolve(name, type_, getone=MDNS_GETONE, timeout=MDNS_TIMEOUT)
+        if not len(recs):
+            print('%s not found (NXDOMAIN)' % name)
+            sys.exit(1)
+
+        for r in recs:
+            print(rr2text(r, MDNS_TTL))
+    finally:
+        deinit()