regression-tests.recursor-dnssec/test_RecDnstap.py

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