regression-tests.recursor-dnssec/test_RecDnstap.py

   1 import errno
   2 import os
   3 import socket
   4 import struct
   5 import sys
   6 import threading
   7 import dns
   8 import dnstap_pb2
   9 from unittest import SkipTest
  10 from recursortests import RecursorTest
  11
  12 FSTRM_CONTROL_ACCEPT = 0x01
  13 FSTRM_CONTROL_START = 0x02
  14 FSTRM_CONTROL_STOP = 0x03
  15 FSTRM_CONTROL_READY = 0x04
  16 FSTRM_CONTROL_FINISH = 0x05
  17
  18 # Python2/3 compatibility hacks
  19 try:
  20     from queue import Queue
  21 except ImportError:
  22     from Queue import Queue
  23
  24 try:
  25     range = xrange
  26 except NameError:
  27     pass
  28
  29
  30 def checkDnstapBase(testinstance, dnstap, protocol, initiator, responder, response_port=53):
  31     testinstance.assertTrue(dnstap)
  32     testinstance.assertTrue(dnstap.HasField('identity'))
  33     #testinstance.assertEqual(dnstap.identity, b'a.server')
  34     testinstance.assertTrue(dnstap.HasField('version'))
  35     #testinstance.assertIn(b'dnsdist ', dnstap.version)
  36     testinstance.assertTrue(dnstap.HasField('type'))
  37     testinstance.assertEqual(dnstap.type, dnstap.MESSAGE)
  38     testinstance.assertTrue(dnstap.HasField('message'))
  39     testinstance.assertTrue(dnstap.message.HasField('socket_protocol'))
  40     testinstance.assertEqual(dnstap.message.socket_protocol, protocol)
  41     testinstance.assertTrue(dnstap.message.HasField('socket_family'))
  42     testinstance.assertEqual(dnstap.message.socket_family, dnstap_pb2.INET)
  43     #
  44     # The query address and port are from the the recursor, we don't know the port
  45     #
  46     testinstance.assertTrue(dnstap.message.HasField('query_address'))
  47     testinstance.assertEqual(socket.inet_ntop(socket.AF_INET, dnstap.message.query_address), initiator)
  48     testinstance.assertTrue(dnstap.message.HasField('query_port'))
  49     testinstance.assertTrue(dnstap.message.HasField('response_address'))
  50     testinstance.assertEqual(socket.inet_ntop(socket.AF_INET, dnstap.message.response_address), responder)
  51     testinstance.assertTrue(dnstap.message.HasField('response_port'))
  52     testinstance.assertEqual(dnstap.message.response_port, response_port)
  53
  54
  55 def checkDnstapQuery(testinstance, dnstap, protocol, initiator, responder):
  56     testinstance.assertEqual(dnstap.message.type, dnstap_pb2.Message.RESOLVER_QUERY)
  57     checkDnstapBase(testinstance, dnstap, protocol, initiator, responder)
  58
  59     testinstance.assertTrue(dnstap.message.HasField('query_time_sec'))
  60     testinstance.assertTrue(dnstap.message.HasField('query_time_nsec'))
  61
  62     testinstance.assertTrue(dnstap.message.HasField('query_message'))
  63     #
  64     # We cannot compare the incoming query with the outgoing one
  65     # The IDs and some other fields will be different
  66     #
  67     #wire_message = dns.message.from_wire(dnstap.message.query_message)
  68     #testinstance.assertEqual(wire_message, query)
  69
  70 def checkDnstapNOD(testinstance, dnstap, protocol, initiator, responder, response_port, query_zone):
  71     testinstance.assertEqual(dnstap.message.type, dnstap_pb2.Message.CLIENT_QUERY)
  72     checkDnstapBase(testinstance, dnstap, protocol, initiator, responder, response_port)
  73
  74     testinstance.assertTrue(dnstap.message.HasField('query_time_sec'))
  75     testinstance.assertTrue(dnstap.message.HasField('query_time_nsec'))
  76
  77     testinstance.assertTrue(dnstap.message.HasField('query_zone'))
  78     testinstance.assertEqual(dns.name.from_wire(dnstap.message.query_zone, 0)[0].to_text(), query_zone)
  79
  80 def checkDnstapUDR(testinstance, dnstap, protocol, initiator, responder, response_port, query_zone):
  81     testinstance.assertEqual(dnstap.message.type, dnstap_pb2.Message.RESOLVER_RESPONSE)
  82     checkDnstapBase(testinstance, dnstap, protocol, initiator, responder, response_port)
  83
  84     testinstance.assertTrue(dnstap.message.HasField('query_time_sec'))
  85     testinstance.assertTrue(dnstap.message.HasField('query_time_nsec'))
  86
  87     testinstance.assertTrue(dnstap.message.HasField('query_zone'))
  88     testinstance.assertEqual(dns.name.from_wire(dnstap.message.query_zone, 0)[0].to_text(), query_zone)
  89
  90     testinstance.assertTrue(dnstap.message.HasField('response_message'))
  91     wire_message = dns.message.from_wire(dnstap.message.response_message)
  92
  93 def checkDnstapExtra(testinstance, dnstap, expected):
  94     testinstance.assertTrue(dnstap.HasField('extra'))
  95     testinstance.assertEqual(dnstap.extra, expected)
  96
  97
  98 def checkDnstapNoExtra(testinstance, dnstap):
  99     testinstance.assertFalse(dnstap.HasField('extra'))
 100
 101
 102 def checkDnstapResponse(testinstance, dnstap, protocol, response, initiator, responder):
 103     testinstance.assertEqual(dnstap.message.type, dnstap_pb2.Message.RESOLVER_RESPONSE)
 104     checkDnstapBase(testinstance, dnstap, protocol, initiator, responder)
 105
 106     testinstance.assertTrue(dnstap.message.HasField('query_time_sec'))
 107     testinstance.assertTrue(dnstap.message.HasField('query_time_nsec'))
 108
 109     testinstance.assertTrue(dnstap.message.HasField('response_time_sec'))
 110     testinstance.assertTrue(dnstap.message.HasField('response_time_nsec'))
 111
 112     testinstance.assertTrue(dnstap.message.response_time_sec > dnstap.message.query_time_sec or \
 113                             dnstap.message.response_time_nsec > dnstap.message.query_time_nsec)
 114
 115     testinstance.assertTrue(dnstap.message.HasField('response_message'))
 116     wire_message = dns.message.from_wire(dnstap.message.response_message)
 117     testinstance.assertEqual(wire_message, response)
 118
 119 def fstrm_get_control_frame_type(data):
 120     (t,) = struct.unpack("!L", data[0:4])
 121     return t
 122
 123
 124 def fstrm_make_control_frame_reply(cft):
 125     if cft == FSTRM_CONTROL_READY:
 126         # Reply with ACCEPT frame and content-type
 127         contenttype = b'protobuf:dnstap.Dnstap'
 128         frame = struct.pack('!LLL', FSTRM_CONTROL_ACCEPT, 1,
 129                             len(contenttype)) + contenttype
 130         buf = struct.pack("!LL", 0, len(frame)) + frame
 131         return buf
 132     elif cft == FSTRM_CONTROL_START:
 133         return None
 134     else:
 135         raise Exception('unhandled control frame ' + cft)
 136
 137
 138 def fstrm_read_and_dispatch_control_frame(conn):
 139     data = conn.recv(4)
 140     if not data:
 141         raise Exception('length of control frame payload could not be read')
 142     (datalen,) = struct.unpack("!L", data)
 143     data = conn.recv(datalen)
 144     cft = fstrm_get_control_frame_type(data)
 145     reply = fstrm_make_control_frame_reply(cft)
 146     if reply:
 147         conn.send(reply)
 148     return cft
 149
 150
 151 def fstrm_handle_bidir_connection(conn, on_data):
 152     data = None
 153     while True:
 154         data = conn.recv(4)
 155         if not data:
 156             break
 157         (datalen,) = struct.unpack("!L", data)
 158         if datalen == 0:
 159             # control frame length follows
 160             cft = fstrm_read_and_dispatch_control_frame(conn)
 161             if cft == FSTRM_CONTROL_STOP:
 162                 break
 163         else:
 164             # data frame
 165             data = conn.recv(datalen)
 166             if not data:
 167                 break
 168
 169             on_data(data)
 170
 171
 172
 173 class DNSTapServerParams(object):
 174     def __init__(self, path):
 175         self.queue = Queue()
 176         self.path = path
 177
 178
 179 DNSTapServerParameters = DNSTapServerParams("/tmp/dnstap.sock")
 180 DNSTapListeners = []
 181
 182 class TestRecursorDNSTap(RecursorTest):
 183     @classmethod
 184     def FrameStreamUnixListener(cls, conn, param):
 185         while True:
 186             try:
 187                 fstrm_handle_bidir_connection(conn, lambda data: \
 188                 param.queue.put(data, True, timeout=2.0))
 189             except socket.error as e:
 190                 if e.errno in (errno.EBADF, errno.EPIPE):
 191                     break
 192                 sys.stderr.write("Unexpected socket error %s\n" % str(e))
 193                 sys.exit(1)
 194             except exception as e:
 195                 sys.stderr.write("Unexpected socket error %s\n" % str(e))
 196                 sys.exit(1)
 197         conn.close()
 198
 199     @classmethod
 200     def FrameStreamUnixListenerMain(cls, param):
 201         sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
 202         try:
 203             try:
 204                 os.remove(param.path)
 205             except:
 206                 pass
 207             sock.bind(param.path)
 208             sock.listen(100)
 209         except socket.error as e:
 210             sys.stderr.write("Error binding/listening in the framestream listener: %s\n" % str(e))
 211             sys.exit(1)
 212         DNSTapListeners.append(sock)
 213         while True:
 214             try:
 215                 (conn, addr) = sock.accept()
 216                 listener = threading.Thread(name='DNSTap Worker', target=cls.FrameStreamUnixListener, args=[conn, param])
 217                 listener.setDaemon(True)
 218                 listener.start()
 219             except socket.error as e:
 220                 if e.errno != errno.EBADF:
 221                     sys.stderr.write("Socket error on accept: %s\n" % str(e))
 222                 else:
 223                     break
 224         sock.close()
 225
 226     @classmethod
 227     def setUpClass(cls):
 228         if os.environ.get("NODNSTAPTESTS") == "1":
 229             raise SkipTest("Not Yet Supported")
 230
 231         cls.setUpSockets()
 232
 233         cls.startResponders()
 234
 235         listener = threading.Thread(name='DNSTap Listener', target=cls.FrameStreamUnixListenerMain, args=[DNSTapServerParameters])
 236         listener.setDaemon(True)
 237         listener.start()
 238
 239         confdir = os.path.join('configs', cls._confdir)
 240         cls.createConfigDir(confdir)
 241
 242         cls.generateRecursorConfig(confdir)
 243         cls.startRecursor(confdir, cls._recursorPort)
 244
 245     def setUp(self):
 246         # Make sure the queue is empty, in case
 247         # a previous test failed
 248         while not DNSTapServerParameters.queue.empty():
 249             DNSTapServerParameters.queue.get(False)
 250
 251     @classmethod
 252     def generateRecursorConfig(cls, confdir):
 253         authzonepath = os.path.join(confdir, 'example.zone')
 254         with open(authzonepath, 'w') as authzone:
 255             authzone.write("""$ORIGIN example.
 256 @ 3600 IN SOA {soa}
 257 a 3600 IN A 192.0.2.42
 258 tagged 3600 IN A 192.0.2.84
 259 query-selected 3600 IN A 192.0.2.84
 260 answer-selected 3600 IN A 192.0.2.84
 261 types 3600 IN A 192.0.2.84
 262 types 3600 IN AAAA 2001:DB8::1
 263 types 3600 IN TXT "Lorem ipsum dolor sit amet"
 264 types 3600 IN MX 10 a.example.
 265 types 3600 IN SPF "v=spf1 -all"
 266 types 3600 IN SRV 10 20 443 a.example.
 267 cname 3600 IN CNAME a.example.
 268
 269 """.format(soa=cls._SOA))
 270         super(TestRecursorDNSTap, cls).generateRecursorConfig(confdir)
 271
 272     @classmethod
 273     def tearDownClass(cls):
 274         cls.tearDownRecursor()
 275         for listerner in DNSTapListeners:
 276             listerner.close()
 277
 278 class DNSTapDefaultTest(TestRecursorDNSTap):
 279     """
 280     This test makes sure that we correctly export outgoing queries over DNSTap.
 281     It must be improved and setup env so we can check for incoming responses, but makes sure for now
 282     that the recursor at least connects to the DNSTap server.
 283     """
 284
 285     _confdir = 'DNSTapDefault'
 286     _config_template = """
 287 auth-zones=example=configs/%s/example.zone""" % _confdir
 288     _lua_config_file = """
 289 dnstapFrameStreamServer({"%s"})
 290     """ % DNSTapServerParameters.path
 291
 292     def getFirstDnstap(self):
 293         try:
 294             data = DNSTapServerParameters.queue.get(True, timeout=2.0)
 295         except:
 296             data = False
 297         self.assertTrue(data)
 298         dnstap = dnstap_pb2.Dnstap()
 299         dnstap.ParseFromString(data)
 300         return dnstap
 301
 302     def testA(self):
 303         name = 'www.example.org.'
 304         query = dns.message.make_query(name, 'A', want_dnssec=True)
 305         query.flags |= dns.flags.RD
 306         res = self.sendUDPQuery(query)
 307         self.assertNotEqual(res, None)
 308
 309         # check the dnstap message corresponding to the UDP query
 310         dnstap = self.getFirstDnstap()
 311
 312         checkDnstapQuery(self, dnstap, dnstap_pb2.UDP, '127.0.0.1', '127.0.0.8')
 313         # We don't expect a response
 314         checkDnstapNoExtra(self, dnstap)
 315
 316 class DNSTapLogNoQueriesTest(TestRecursorDNSTap):
 317
 318     _confdir = 'DNSTapLogNoQueries'
 319     _config_template = """
 320 auth-zones=example=configs/%s/example.zone""" % _confdir
 321     _lua_config_file = """
 322 dnstapFrameStreamServer({"%s"}, {logQueries=false})
 323     """ % (DNSTapServerParameters.path)
 324
 325     def testA(self):
 326         name = 'www.example.org.'
 327         query = dns.message.make_query(name, 'A', want_dnssec=True)
 328         query.flags |= dns.flags.RD
 329         res = self.sendUDPQuery(query)
 330         self.assertNotEqual(res, None)
 331
 332         # We don't expect anything
 333         self.assertTrue(DNSTapServerParameters.queue.empty())
 334
 335 class DNSTapLogNODTest(TestRecursorDNSTap):
 336     """
 337     This test makes sure that we correctly export outgoing queries over DNSTap.
 338     It must be improved and setup env so we can check for incoming responses, but makes sure for now
 339     that the recursor at least connects to the DNSTap server.
 340     """
 341
 342     _confdir = 'DNSTapLogNODQueries'
 343     _config_template = """
 344 new-domain-tracking=yes
 345 new-domain-history-dir=configs/%s/nod
 346 unique-response-tracking=yes
 347 unique-response-history-dir=configs/%s/udr
 348 auth-zones=example=configs/%s/example.zone""" % (_confdir, _confdir, _confdir)
 349     _lua_config_file = """
 350 dnstapNODFrameStreamServer({"%s"})
 351     """ % (DNSTapServerParameters.path)
 352
 353     @classmethod
 354     def generateRecursorConfig(cls, confdir):
 355         for directory in ["nod", "udr"]:
 356             path = os.path.join('configs', cls._confdir, directory)
 357             cls.createConfigDir(path)
 358         super(DNSTapLogNODTest, cls).generateRecursorConfig(confdir)
 359
 360     def getFirstDnstap(self):
 361         try:
 362             data = DNSTapServerParameters.queue.get(True, timeout=2.0)
 363         except:
 364             data = False
 365         self.assertTrue(data)
 366         dnstap = dnstap_pb2.Dnstap()
 367         dnstap.ParseFromString(data)
 368         return dnstap
 369
 370     def testA(self):
 371         name = 'www.example.org.'
 372         query = dns.message.make_query(name, 'A', want_dnssec=True)
 373         query.flags |= dns.flags.RD
 374         res = self.sendUDPQuery(query)
 375         self.assertNotEqual(res, None)
 376
 377         # check the dnstap message corresponding to the UDP query
 378         dnstap = self.getFirstDnstap()
 379
 380         checkDnstapNOD(self, dnstap, dnstap_pb2.UDP, '127.0.0.1', '127.0.0.1', 5300, name)
 381         # We don't expect a response
 382         checkDnstapNoExtra(self, dnstap)
 383
 384 class DNSTapLogUDRTest(TestRecursorDNSTap):
 385
 386     _confdir = 'DNSTapLogUDRResponses'
 387     _config_template = """
 388 new-domain-tracking=yes
 389 new-domain-history-dir=configs/%s/nod
 390 unique-response-tracking=yes
 391 unique-response-history-dir=configs/%s/udr
 392 auth-zones=example=configs/%s/example.zone""" % (_confdir, _confdir, _confdir)
 393     _lua_config_file = """
 394 dnstapNODFrameStreamServer({"%s"}, {logNODs=false, logUDRs=true})
 395     """ % (DNSTapServerParameters.path)
 396
 397     @classmethod
 398     def generateRecursorConfig(cls, confdir):
 399         for directory in ["nod", "udr"]:
 400             path = os.path.join('configs', cls._confdir, directory)
 401             cls.createConfigDir(path)
 402         super(DNSTapLogUDRTest, cls).generateRecursorConfig(confdir)
 403
 404     def getFirstDnstap(self):
 405         try:
 406             data = DNSTapServerParameters.queue.get(True, timeout=2.0)
 407         except:
 408             data = False
 409         self.assertTrue(data)
 410         dnstap = dnstap_pb2.Dnstap()
 411         dnstap.ParseFromString(data)
 412         return dnstap
 413
 414     def testA(self):
 415         name = 'types.example.'
 416         query = dns.message.make_query(name, 'A', want_dnssec=True)
 417         query.flags |= dns.flags.RD
 418         res = self.sendUDPQuery(query)
 419         self.assertNotEqual(res, None)
 420
 421         # check the dnstap message corresponding to the UDP query
 422         dnstap = self.getFirstDnstap()
 423
 424         checkDnstapUDR(self, dnstap, dnstap_pb2.UDP, '127.0.0.1', '127.0.0.1', 5300, name)
 425         # We don't expect a rpasesponse
 426         checkDnstapNoExtra(self, dnstap)
 427
 428 class DNSTapLogNODUDRTest(TestRecursorDNSTap):
 429
 430     _confdir = 'DNSTapLogNODUDRs'
 431     _config_template = """
 432 new-domain-tracking=yes
 433 new-domain-history-dir=configs/%s/nod
 434 unique-response-tracking=yes
 435 unique-response-history-dir=configs/%s/udr
 436 auth-zones=example=configs/%s/example.zone""" % (_confdir, _confdir, _confdir)
 437     _lua_config_file = """
 438 dnstapNODFrameStreamServer({"%s"}, {logNODs=true, logUDRs=true})
 439     """ % (DNSTapServerParameters.path)
 440
 441     @classmethod
 442     def generateRecursorConfig(cls, confdir):
 443         for directory in ["nod", "udr"]:
 444             path = os.path.join('configs', cls._confdir, directory)
 445             cls.createConfigDir(path)
 446         super(DNSTapLogNODUDRTest, cls).generateRecursorConfig(confdir)
 447
 448     def getFirstDnstap(self):
 449         try:
 450             data = DNSTapServerParameters.queue.get(True, timeout=2.0)
 451         except:
 452             data = False
 453         self.assertTrue(data)
 454         dnstap = dnstap_pb2.Dnstap()
 455         dnstap.ParseFromString(data)
 456         return dnstap
 457
 458     def testA(self):
 459         name = 'types.example.'
 460         query = dns.message.make_query(name, 'A', want_dnssec=True)
 461         query.flags |= dns.flags.RD
 462         res = self.sendUDPQuery(query)
 463         self.assertNotEqual(res, None)
 464
 465         dnstap = self.getFirstDnstap()
 466         checkDnstapUDR(self, dnstap, dnstap_pb2.UDP, '127.0.0.1', '127.0.0.1', 5300, name)
 467
 468         dnstap = self.getFirstDnstap()
 469         checkDnstapNOD(self, dnstap, dnstap_pb2.UDP, '127.0.0.1', '127.0.0.1', 5300, name)
 470
 471         checkDnstapNoExtra(self, dnstap)