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