regression-tests.recursor-dnssec/test_RecDnstap.py

   1 import os
   2 import socket
   3 import struct
   4 import sys
   5 import threading
   6 import dns
   7 import dnstap_pb2
   8 from nose import SkipTest
   9 from recursortests import RecursorTest
  10
  11 FSTRM_CONTROL_ACCEPT = 0x01
  12 FSTRM_CONTROL_START = 0x02
  13 FSTRM_CONTROL_STOP = 0x03
  14 FSTRM_CONTROL_READY = 0x04
  15 FSTRM_CONTROL_FINISH = 0x05
  16
  17 # Python2/3 compatibility hacks
  18 try:
  19     from queue import Queue
  20 except ImportError:
  21     from Queue import Queue
  22
  23 try:
  24     range = xrange
  25 except NameError:
  26     pass
  27
  28
  29 def checkDnstapBase(testinstance, dnstap, protocol, initiator):
  30     testinstance.assertTrue(dnstap)
  31     testinstance.assertTrue(dnstap.HasField('identity'))
  32     #testinstance.assertEqual(dnstap.identity, b'a.server')
  33     testinstance.assertTrue(dnstap.HasField('version'))
  34     #testinstance.assertIn(b'dnsdist ', dnstap.version)
  35     testinstance.assertTrue(dnstap.HasField('type'))
  36     testinstance.assertEqual(dnstap.type, dnstap.MESSAGE)
  37     testinstance.assertTrue(dnstap.HasField('message'))
  38     testinstance.assertTrue(dnstap.message.HasField('socket_protocol'))
  39     testinstance.assertEqual(dnstap.message.socket_protocol, protocol)
  40     testinstance.assertTrue(dnstap.message.HasField('socket_family'))
  41     testinstance.assertEquals(dnstap.message.socket_family, dnstap_pb2.INET)
  42     #
  43     # We cannot check the query address and port since we only log outgoing queries via dnstap
  44     #
  45     #testinstance.assertTrue(dnstap.message.HasField('query_address'))
  46     #testinstance.assertEquals(socket.inet_ntop(socket.AF_INET, dnstap.message.query_address), initiator)
  47     testinstance.assertTrue(dnstap.message.HasField('response_address'))
  48     testinstance.assertEquals(socket.inet_ntop(socket.AF_INET, dnstap.message.response_address), initiator)
  49     testinstance.assertTrue(dnstap.message.HasField('response_port'))
  50     testinstance.assertEquals(dnstap.message.response_port, 53)
  51
  52
  53 def checkDnstapQuery(testinstance, dnstap, protocol, initiator='127.0.0.1'):
  54     testinstance.assertEquals(dnstap.message.type, dnstap_pb2.Message.RESOLVER_QUERY)
  55     checkDnstapBase(testinstance, dnstap, protocol, initiator)
  56
  57     testinstance.assertTrue(dnstap.message.HasField('query_time_sec'))
  58     testinstance.assertTrue(dnstap.message.HasField('query_time_nsec'))
  59
  60     testinstance.assertTrue(dnstap.message.HasField('query_message'))
  61     #
  62     # We cannot compare the incoming query with the outgoing one
  63     # The IDs and some other fields will be different
  64     #
  65     #wire_message = dns.message.from_wire(dnstap.message.query_message)
  66     #testinstance.assertEqual(wire_message, query)
  67
  68
  69 def checkDnstapExtra(testinstance, dnstap, expected):
  70     testinstance.assertTrue(dnstap.HasField('extra'))
  71     testinstance.assertEqual(dnstap.extra, expected)
  72
  73
  74 def checkDnstapNoExtra(testinstance, dnstap):
  75     testinstance.assertFalse(dnstap.HasField('extra'))
  76
  77
  78 def checkDnstapResponse(testinstance, dnstap, protocol, response, initiator='127.0.0.1'):
  79     testinstance.assertEquals(dnstap.message.type, dnstap_pb2.Message.RESOLVER_RESPONSE)
  80     checkDnstapBase(testinstance, dnstap, protocol, initiator)
  81
  82     testinstance.assertTrue(dnstap.message.HasField('query_time_sec'))
  83     testinstance.assertTrue(dnstap.message.HasField('query_time_nsec'))
  84
  85     testinstance.assertTrue(dnstap.message.HasField('response_time_sec'))
  86     testinstance.assertTrue(dnstap.message.HasField('response_time_nsec'))
  87
  88     testinstance.assertTrue(dnstap.message.response_time_sec > dnstap.message.query_time_sec or \
  89                             dnstap.message.response_time_nsec > dnstap.message.query_time_nsec)
  90
  91     testinstance.assertTrue(dnstap.message.HasField('response_message'))
  92     wire_message = dns.message.from_wire(dnstap.message.response_message)
  93     testinstance.assertEqual(wire_message, response)
  94
  95 def fstrm_get_control_frame_type(data):
  96     (t,) = struct.unpack("!L", data[0:4])
  97     return t
  98
  99
 100 def fstrm_make_control_frame_reply(cft):
 101     if cft == FSTRM_CONTROL_READY:
 102         # Reply with ACCEPT frame and content-type
 103         contenttype = b'protobuf:dnstap.Dnstap'
 104         frame = struct.pack('!LLL', FSTRM_CONTROL_ACCEPT, 1,
 105                             len(contenttype)) + contenttype
 106         buf = struct.pack("!LL", 0, len(frame)) + frame
 107         return buf
 108     elif cft == FSTRM_CONTROL_START:
 109         return None
 110     else:
 111         raise Exception('unhandled control frame ' + cft)
 112
 113
 114 def fstrm_read_and_dispatch_control_frame(conn):
 115     data = conn.recv(4)
 116     if not data:
 117         raise Exception('length of control frame payload could not be read')
 118     (datalen,) = struct.unpack("!L", data)
 119     data = conn.recv(datalen)
 120     cft = fstrm_get_control_frame_type(data)
 121     reply = fstrm_make_control_frame_reply(cft)
 122     if reply:
 123         conn.send(reply)
 124     return cft
 125
 126
 127 def fstrm_handle_bidir_connection(conn, on_data):
 128     data = None
 129     while True:
 130         data = conn.recv(4)
 131         if not data:
 132             break
 133         (datalen,) = struct.unpack("!L", data)
 134         if datalen == 0:
 135             # control frame length follows
 136             cft = fstrm_read_and_dispatch_control_frame(conn)
 137             if cft == FSTRM_CONTROL_STOP:
 138                 break
 139         else:
 140             # data frame
 141             data = conn.recv(datalen)
 142             if not data:
 143                 break
 144
 145             on_data(data)
 146
 147
 148
 149 class DNSTapServerParams(object):
 150     def __init__(self, path):
 151         self.queue = Queue()
 152         self.path = path
 153
 154
 155 DNSTapServerParameters = DNSTapServerParams("/tmp/dnstap.sock")
 156 DNSTapListeners = []
 157
 158 class TestRecursorDNSTap(RecursorTest):
 159     @classmethod
 160     def FrameStreamUnixListener(cls, conn, param):
 161         while True:
 162             try:
 163                 fstrm_handle_bidir_connection(conn, lambda data: \
 164                 param.queue.put(data, True, timeout=2.0))
 165             except socket.error as e:
 166                 if e.errno == 9:
 167                     break
 168                 sys.stderr.write("Unexpected socket error %s\n" % str(e))
 169                 sys.exit(1)
 170             except exception as e:
 171                 sys.stderr.write("Unexpected socket error %s\n" % str(e))
 172                 sys.exit(1)
 173         conn.close()
 174
 175     @classmethod
 176     def FrameStreamUnixListenerMain(cls, param):
 177         sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
 178         try:
 179             try:
 180                 os.remove(param.path)
 181             except:
 182                 pass
 183             sock.bind(param.path)
 184             sock.listen(100)
 185         except socket.error as e:
 186             sys.stderr.write("Error binding/listening in the framestream listener: %s\n" % str(e))
 187             sys.exit(1)
 188         DNSTapListeners.append(sock)
 189         while True:
 190             try:
 191                 (conn, addr) = sock.accept()
 192                 listener = threading.Thread(name='DNSTap Worker', target=cls.FrameStreamUnixListener, args=[conn, param])
 193                 listener.setDaemon(True)
 194                 listener.start()
 195             except socket.error as e:
 196                 if e.errno != 9:
 197                     sys.stderr.write("Socket error on accept: %s\n" % str(e))
 198                 else:
 199                     break
 200         sock.close()
 201
 202     @classmethod
 203     def setUpClass(cls):
 204         if os.environ.get("NODNSTAPTESTS") == "1":
 205             raise SkipTest("Not Yet Supported")
 206
 207         cls.setUpSockets()
 208
 209         cls.startResponders()
 210
 211         listener = threading.Thread(name='DNSTap Listener', target=cls.FrameStreamUnixListenerMain, args=[DNSTapServerParameters])
 212         listener.setDaemon(True)
 213         listener.start()
 214
 215         confdir = os.path.join('configs', cls._confdir)
 216         cls.createConfigDir(confdir)
 217
 218         cls.generateRecursorConfig(confdir)
 219         cls.startRecursor(confdir, cls._recursorPort)
 220
 221     def setUp(self):
 222         # Make sure the queue is empty, in case
 223         # a previous test failed
 224         while not DNSTapServerParameters.queue.empty():
 225             DNSTapServerParameters.queue.get(False)
 226
 227     @classmethod
 228     def generateRecursorConfig(cls, confdir):
 229         authzonepath = os.path.join(confdir, 'example.zone')
 230         with open(authzonepath, 'w') as authzone:
 231             authzone.write("""$ORIGIN example.
 232 @ 3600 IN SOA {soa}
 233 a 3600 IN A 192.0.2.42
 234 tagged 3600 IN A 192.0.2.84
 235 query-selected 3600 IN A 192.0.2.84
 236 answer-selected 3600 IN A 192.0.2.84
 237 types 3600 IN A 192.0.2.84
 238 types 3600 IN AAAA 2001:DB8::1
 239 types 3600 IN TXT "Lorem ipsum dolor sit amet"
 240 types 3600 IN MX 10 a.example.
 241 types 3600 IN SPF "v=spf1 -all"
 242 types 3600 IN SRV 10 20 443 a.example.
 243 cname 3600 IN CNAME a.example.
 244
 245 """.format(soa=cls._SOA))
 246         super(TestRecursorDNSTap, cls).generateRecursorConfig(confdir)
 247
 248     @classmethod
 249     def tearDownClass(cls):
 250         cls.tearDownRecursor()
 251         for listerner in DNSTapListeners:
 252             listerner.close()
 253
 254 class DNSTapDefaultTest(TestRecursorDNSTap):
 255     """
 256     This test makes sure that we correctly export outgoing queries over DNSTap.
 257     It must be improved and setup env so we can check for incoming responses, but makes sure for now
 258     that the recursor at least connects to the DNSTap server.
 259     """
 260
 261     _confdir = 'DNSTapDefault'
 262     _config_template = """
 263 auth-zones=example=configs/%s/example.zone""" % _confdir
 264     _lua_config_file = """
 265 dnstapFrameStreamServer({"%s"})
 266     """ % DNSTapServerParameters.path
 267
 268     def getFirstDnstap(self):
 269         try:
 270             data = DNSTapServerParameters.queue.get(True, timeout=2.0)
 271         except:
 272             data = False
 273         self.assertTrue(data)
 274         dnstap = dnstap_pb2.Dnstap()
 275         dnstap.ParseFromString(data)
 276         return dnstap
 277
 278     def testA(self):
 279         name = 'www.example.org.'
 280         query = dns.message.make_query(name, 'A', want_dnssec=True)
 281         query.flags |= dns.flags.RD
 282         res = self.sendUDPQuery(query)
 283         self.assertNotEquals(res, None)
 284
 285         # check the dnstap message corresponding to the UDP query
 286         dnstap = self.getFirstDnstap()
 287
 288         checkDnstapQuery(self, dnstap, dnstap_pb2.UDP, '127.0.0.8')
 289         # We don't expect a response
 290         checkDnstapNoExtra(self, dnstap)
 291
 292 class DNSTapLogNoQueriesTest(TestRecursorDNSTap):
 293     """
 294     This test makes sure that we correctly export outgoing queries over DNSTap.
 295     It must be improved and setup env so we can check for incoming responses, but makes sure for now
 296     that the recursor at least connects to the DNSTap server.
 297     """
 298
 299     _confdir = 'DNSTapLogNoQueries'
 300     _config_template = """
 301 auth-zones=example=configs/%s/example.zone""" % _confdir
 302     _lua_config_file = """
 303 dnstapFrameStreamServer({"%s"}, {logQueries=false})
 304     """ % (DNSTapServerParameters.path)
 305
 306     def testA(self):
 307         name = 'www.example.org.'
 308         query = dns.message.make_query(name, 'A', want_dnssec=True)
 309         query.flags |= dns.flags.RD
 310         res = self.sendUDPQuery(query)
 311         self.assertNotEquals(res, None)
 312
 313         # We don't expect anything
 314         self.assertTrue(DNSTapServerParameters.queue.empty())