regression-tests.dnsdist/test_Protobuf.py

   1 #!/usr/bin/env python
   2 import base64
   3 import threading
   4 import socket
   5 import struct
   6 import sys
   7 import time
   8 from dnsdisttests import DNSDistTest, pickAvailablePort, Queue
   9 from proxyprotocol import ProxyProtocol
  10
  11 import dns
  12 import dnsmessage_pb2
  13 import extendederrors
  14
  15 class DNSDistProtobufTest(DNSDistTest):
  16     _protobufServerPort = pickAvailablePort()
  17     _protobufQueue = Queue()
  18     _protobufServerID = 'dnsdist-server-1'
  19     _protobufCounter = 0
  20
  21     @classmethod
  22     def ProtobufListener(cls, port):
  23         cls._backgroundThreads[threading.get_native_id()] = True
  24         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  25         sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
  26         try:
  27             sock.bind(("127.0.0.1", port))
  28         except socket.error as e:
  29             print("Error binding in the protbuf listener: %s" % str(e))
  30             sys.exit(1)
  31
  32         sock.listen(100)
  33         sock.settimeout(1.0)
  34         while True:
  35             try:
  36                 (conn, _) = sock.accept()
  37             except socket.timeout:
  38                 if cls._backgroundThreads.get(threading.get_native_id(), False) == False:
  39                     del cls._backgroundThreads[threading.get_native_id()]
  40                     break
  41                 else:
  42                     continue
  43
  44             data = None
  45             while True:
  46                 data = conn.recv(2)
  47                 if not data:
  48                     break
  49                 (datalen,) = struct.unpack("!H", data)
  50                 data = conn.recv(datalen)
  51                 if not data:
  52                     break
  53
  54                 cls._protobufQueue.put(data, True, timeout=2.0)
  55
  56             conn.close()
  57         sock.close()
  58
  59     @classmethod
  60     def startResponders(cls):
  61         cls._UDPResponder = threading.Thread(name='UDP Responder', target=cls.UDPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue])
  62         cls._UDPResponder.daemon = True
  63         cls._UDPResponder.start()
  64
  65         cls._TCPResponder = threading.Thread(name='TCP Responder', target=cls.TCPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue])
  66         cls._TCPResponder.daemon = True
  67         cls._TCPResponder.start()
  68
  69         cls._protobufListener = threading.Thread(name='Protobuf Listener', target=cls.ProtobufListener, args=[cls._protobufServerPort])
  70         cls._protobufListener.daemon = True
  71         cls._protobufListener.start()
  72
  73     def getFirstProtobufMessage(self):
  74         self.assertFalse(self._protobufQueue.empty())
  75         data = self._protobufQueue.get(False)
  76         self.assertTrue(data)
  77         msg = dnsmessage_pb2.PBDNSMessage()
  78         msg.ParseFromString(data)
  79         return msg
  80
  81     def checkProtobufBase(self, msg, protocol, query, initiator, normalQueryResponse=True, v6=False):
  82         self.assertTrue(msg)
  83         self.assertTrue(msg.HasField('timeSec'))
  84         self.assertTrue(msg.HasField('socketFamily'))
  85         if v6:
  86             self.assertEqual(msg.socketFamily, dnsmessage_pb2.PBDNSMessage.INET6)
  87         else:
  88             self.assertEqual(msg.socketFamily, dnsmessage_pb2.PBDNSMessage.INET)
  89         self.assertTrue(msg.HasField('from'))
  90         fromvalue = getattr(msg, 'from')
  91         if v6:
  92             self.assertEqual(socket.inet_ntop(socket.AF_INET6, fromvalue), initiator)
  93         else:
  94             self.assertEqual(socket.inet_ntop(socket.AF_INET, fromvalue), initiator)
  95         self.assertTrue(msg.HasField('socketProtocol'))
  96         self.assertEqual(msg.socketProtocol, protocol)
  97         self.assertTrue(msg.HasField('messageId'))
  98         self.assertTrue(msg.HasField('id'))
  99         self.assertEqual(msg.id, query.id)
 100         self.assertTrue(msg.HasField('inBytes'))
 101         self.assertTrue(msg.HasField('serverIdentity'))
 102         self.assertEqual(msg.serverIdentity, self._protobufServerID.encode('utf-8'))
 103
 104         if normalQueryResponse and (protocol == dnsmessage_pb2.PBDNSMessage.UDP or protocol == dnsmessage_pb2.PBDNSMessage.TCP):
 105           # compare inBytes with length of query/response
 106           self.assertEqual(msg.inBytes, len(query.to_wire()))
 107         # dnsdist doesn't set the existing EDNS Subnet for now,
 108         # although it might be set from Lua
 109         # self.assertTrue(msg.HasField('originalRequestorSubnet'))
 110         # self.assertEqual(len(msg.originalRequestorSubnet), 4)
 111         # self.assertEqual(socket.inet_ntop(socket.AF_INET, msg.originalRequestorSubnet), '127.0.0.1')
 112
 113     def checkProtobufQuery(self, msg, protocol, query, qclass, qtype, qname, initiator='127.0.0.1', v6=False):
 114         self.assertEqual(msg.type, dnsmessage_pb2.PBDNSMessage.DNSQueryType)
 115         self.checkProtobufBase(msg, protocol, query, initiator, v6=v6)
 116         # dnsdist doesn't fill the responder field for responses
 117         # because it doesn't keep the information around.
 118         self.assertTrue(msg.HasField('to'))
 119         if not v6:
 120             self.assertEqual(socket.inet_ntop(socket.AF_INET, msg.to), '127.0.0.1')
 121         self.assertTrue(msg.HasField('question'))
 122         self.assertTrue(msg.question.HasField('qClass'))
 123         self.assertEqual(msg.question.qClass, qclass)
 124         self.assertTrue(msg.question.HasField('qType'))
 125         self.assertEqual(msg.question.qClass, qtype)
 126         self.assertTrue(msg.question.HasField('qName'))
 127         self.assertEqual(msg.question.qName, qname)
 128
 129     def checkProtobufTags(self, tags, expectedTags):
 130         # only differences will be in new list
 131         listx = set(tags) ^ set(expectedTags)
 132         # exclusive or of lists should be empty
 133         self.assertEqual(len(listx), 0, "Protobuf tags don't match")
 134
 135     def checkProtobufQueryConvertedToResponse(self, msg, protocol, response, initiator='127.0.0.0'):
 136         self.assertEqual(msg.type, dnsmessage_pb2.PBDNSMessage.DNSResponseType)
 137         # skip comparing inBytes (size of the query) with the length of the generated response
 138         self.checkProtobufBase(msg, protocol, response, initiator, False)
 139         self.assertTrue(msg.HasField('response'))
 140         self.assertTrue(msg.response.HasField('queryTimeSec'))
 141
 142     def checkProtobufResponse(self, msg, protocol, response, initiator='127.0.0.1', v6=False):
 143         self.assertEqual(msg.type, dnsmessage_pb2.PBDNSMessage.DNSResponseType)
 144         self.checkProtobufBase(msg, protocol, response, initiator, v6=v6)
 145         self.assertTrue(msg.HasField('response'))
 146         self.assertTrue(msg.response.HasField('queryTimeSec'))
 147
 148     def checkProtobufResponseRecord(self, record, rclass, rtype, rname, rttl):
 149         self.assertTrue(record.HasField('class'))
 150         self.assertEqual(getattr(record, 'class'), rclass)
 151         self.assertTrue(record.HasField('type'))
 152         self.assertEqual(record.type, rtype)
 153         self.assertTrue(record.HasField('name'))
 154         self.assertEqual(record.name, rname)
 155         self.assertTrue(record.HasField('ttl'))
 156         self.assertEqual(record.ttl, rttl)
 157         self.assertTrue(record.HasField('rdata'))
 158
 159 class TestProtobuf(DNSDistProtobufTest):
 160     _config_params = ['_testServerPort', '_protobufServerPort', '_protobufServerID', '_protobufServerID']
 161     _config_template = """
 162     luasmn = newSuffixMatchNode()
 163     luasmn:add(newDNSName('lua.protobuf.tests.powerdns.com.'))
 164
 165     function alterProtobufResponse(dq, protobuf)
 166       if luasmn:check(dq.qname) then
 167         requestor = newCA(tostring(dq.remoteaddr))              -- called by testLuaProtobuf()
 168         if requestor:isIPv4() then
 169           requestor:truncate(24)
 170         else
 171           requestor:truncate(56)
 172         end
 173         protobuf:setRequestor(requestor)
 174
 175         local tableTags = {}
 176         table.insert(tableTags, "TestLabel1,TestData1")
 177         table.insert(tableTags, "TestLabel2,TestData2")
 178
 179         protobuf:setTagArray(tableTags)
 180
 181         protobuf:setTag('TestLabel3,TestData3')
 182
 183         protobuf:setTag("Response,456")
 184
 185       else
 186
 187         local tableTags = {}                                    -- called by testProtobuf()
 188         table.insert(tableTags, "TestLabel1,TestData1")
 189         table.insert(tableTags, "TestLabel2,TestData2")
 190         protobuf:setTagArray(tableTags)
 191
 192         protobuf:setTag('TestLabel3,TestData3')
 193
 194         protobuf:setTag("Response,456")
 195
 196       end
 197     end
 198
 199     function alterProtobufQuery(dq, protobuf)
 200
 201       if luasmn:check(dq.qname) then
 202         requestor = newCA(tostring(dq.remoteaddr))              -- called by testLuaProtobuf()
 203         if requestor:isIPv4() then
 204           requestor:truncate(24)
 205         else
 206           requestor:truncate(56)
 207         end
 208         protobuf:setRequestor(requestor)
 209
 210         local tableTags = {}
 211         tableTags = dq:getTagArray()                            -- get table from DNSQuery
 212
 213         local tablePB = {}
 214           for k, v in pairs( tableTags) do
 215           table.insert(tablePB, k .. "," .. v)
 216         end
 217
 218         protobuf:setTagArray(tablePB)                           -- store table in protobuf
 219         protobuf:setTag("Query,123")                            -- add another tag entry in protobuf
 220
 221         protobuf:setResponseCode(DNSRCode.NXDOMAIN)             -- set protobuf response code to be NXDOMAIN
 222
 223         local strReqName = tostring(dq.qname)                   -- get request dns name
 224
 225         protobuf:setProtobufResponseType()                      -- set protobuf to look like a response and not a query, with 0 default time
 226
 227         blobData = '\127' .. '\000' .. '\000' .. '\002'         -- 127.0.0.2, note: lua 5.1 can only embed decimal not hex
 228
 229         protobuf:addResponseRR(strReqName, 1, 1, 123, blobData) -- add a RR to the protobuf
 230
 231         protobuf:setBytes(65)                                   -- set the size of the query to confirm in checkProtobufBase
 232
 233       else
 234
 235         local tableTags = {}                                    -- called by testProtobuf()
 236         table.insert(tableTags, "TestLabel1,TestData1")
 237         table.insert(tableTags, "TestLabel2,TestData2")
 238
 239         protobuf:setTagArray(tableTags)
 240         protobuf:setTag('TestLabel3,TestData3')
 241         protobuf:setTag("Query,123")
 242
 243       end
 244     end
 245
 246     function alterLuaFirst(dq)                                  -- called when dnsdist receives new request
 247       local tt = {}
 248       tt["TestLabel1"] = "TestData1"
 249       tt["TestLabel2"] = "TestData2"
 250
 251       dq:setTagArray(tt)
 252
 253       dq:setTag("TestLabel3","TestData3")
 254       return DNSAction.None, ""                         -- continue to the next rule
 255     end
 256
 257     newServer{address="127.0.0.1:%s", useClientSubnet=true}
 258     rl = newRemoteLogger('127.0.0.1:%s')
 259
 260     addAction(AllRule(), LuaAction(alterLuaFirst))                                                      -- Add tags to DNSQuery first
 261
 262     addAction(AllRule(), RemoteLogAction(rl, alterProtobufQuery, {serverID='%s'}))                              -- Send protobuf message before lookup
 263
 264     addResponseAction(AllRule(), RemoteLogResponseAction(rl, alterProtobufResponse, true, {serverID='%s'}))     -- Send protobuf message after lookup
 265
 266     """
 267
 268     def testProtobuf(self):
 269         """
 270         Protobuf: Send data to a protobuf server
 271         """
 272         name = 'query.protobuf.tests.powerdns.com.'
 273
 274         target = 'target.protobuf.tests.powerdns.com.'
 275         query = dns.message.make_query(name, 'A', 'IN')
 276         response = dns.message.make_response(query)
 277
 278         rrset = dns.rrset.from_text(name,
 279                                     3600,
 280                                     dns.rdataclass.IN,
 281                                     dns.rdatatype.CNAME,
 282                                     target)
 283         response.answer.append(rrset)
 284
 285         rrset = dns.rrset.from_text(target,
 286                                     3600,
 287                                     dns.rdataclass.IN,
 288                                     dns.rdatatype.A,
 289                                     '127.0.0.1')
 290         response.answer.append(rrset)
 291
 292         (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
 293         self.assertTrue(receivedQuery)
 294         self.assertTrue(receivedResponse)
 295         receivedQuery.id = query.id
 296         self.assertEqual(query, receivedQuery)
 297         self.assertEqual(response, receivedResponse)
 298
 299         if self._protobufQueue.empty():
 300             # let the protobuf messages the time to get there
 301             time.sleep(1)
 302
 303         # check the protobuf message corresponding to the UDP query
 304         msg = self.getFirstProtobufMessage()
 305
 306         self.checkProtobufQuery(msg, dnsmessage_pb2.PBDNSMessage.UDP, query, dns.rdataclass.IN, dns.rdatatype.A, name)
 307         self.checkProtobufTags(msg.response.tags, [u"TestLabel1,TestData1", u"TestLabel2,TestData2", u"TestLabel3,TestData3", u"Query,123"])
 308
 309         # check the protobuf message corresponding to the UDP response
 310         msg = self.getFirstProtobufMessage()
 311         self.checkProtobufResponse(msg, dnsmessage_pb2.PBDNSMessage.UDP, response)
 312         self.checkProtobufTags(msg.response.tags, [ u"TestLabel1,TestData1", u"TestLabel2,TestData2", u"TestLabel3,TestData3", u"Response,456"])
 313         self.assertEqual(len(msg.response.rrs), 2)
 314         rr = msg.response.rrs[0]
 315         self.checkProtobufResponseRecord(rr, dns.rdataclass.IN, dns.rdatatype.CNAME, name, 3600)
 316         self.assertEqual(rr.rdata.decode('utf-8'), target)
 317         rr = msg.response.rrs[1]
 318         self.checkProtobufResponseRecord(rr, dns.rdataclass.IN, dns.rdatatype.A, target, 3600)
 319         self.assertEqual(socket.inet_ntop(socket.AF_INET, rr.rdata), '127.0.0.1')
 320
 321         (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
 322         self.assertTrue(receivedQuery)
 323         self.assertTrue(receivedResponse)
 324         receivedQuery.id = query.id
 325         self.assertEqual(query, receivedQuery)
 326         self.assertEqual(response, receivedResponse)
 327
 328         if self._protobufQueue.empty():
 329             # let the protobuf messages the time to get there
 330             time.sleep(1)
 331
 332         # check the protobuf message corresponding to the TCP query
 333         msg = self.getFirstProtobufMessage()
 334
 335         self.checkProtobufQuery(msg, dnsmessage_pb2.PBDNSMessage.TCP, query, dns.rdataclass.IN, dns.rdatatype.A, name)
 336         self.checkProtobufTags(msg.response.tags, [u"TestLabel1,TestData1", u"TestLabel2,TestData2", u"TestLabel3,TestData3", u"Query,123"])
 337
 338         # check the protobuf message corresponding to the TCP response
 339         msg = self.getFirstProtobufMessage()
 340         self.checkProtobufResponse(msg, dnsmessage_pb2.PBDNSMessage.TCP, response)
 341         self.checkProtobufTags(msg.response.tags, [ u"TestLabel1,TestData1", u"TestLabel2,TestData2", u"TestLabel3,TestData3", u"Response,456"])
 342         self.assertEqual(len(msg.response.rrs), 2)
 343         rr = msg.response.rrs[0]
 344         self.checkProtobufResponseRecord(rr, dns.rdataclass.IN, dns.rdatatype.CNAME, name, 3600)
 345         self.assertEqual(rr.rdata.decode('utf-8'), target)
 346         rr = msg.response.rrs[1]
 347         self.checkProtobufResponseRecord(rr, dns.rdataclass.IN, dns.rdatatype.A, target, 3600)
 348         self.assertEqual(socket.inet_ntop(socket.AF_INET, rr.rdata), '127.0.0.1')
 349
 350     def testLuaProtobuf(self):
 351
 352         """
 353         Protobuf: Check that the Lua callback rewrote the initiator
 354         """
 355         name = 'lua.protobuf.tests.powerdns.com.'
 356         query = dns.message.make_query(name, 'A', 'IN')
 357         response = dns.message.make_response(query)
 358         rrset = dns.rrset.from_text(name,
 359                                     3600,
 360                                     dns.rdataclass.IN,
 361                                     dns.rdatatype.A,
 362                                     '127.0.0.1')
 363         response.answer.append(rrset)
 364
 365
 366         (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
 367
 368         self.assertTrue(receivedQuery)
 369         self.assertTrue(receivedResponse)
 370         receivedQuery.id = query.id
 371         self.assertEqual(query, receivedQuery)
 372         self.assertEqual(response, receivedResponse)
 373
 374         if self._protobufQueue.empty():
 375             # let the protobuf messages the time to get there
 376             time.sleep(1)
 377
 378         # check the protobuf message corresponding to the UDP query
 379         msg = self.getFirstProtobufMessage()
 380
 381         self.checkProtobufQueryConvertedToResponse(msg, dnsmessage_pb2.PBDNSMessage.UDP, response, '127.0.0.0')
 382         self.checkProtobufTags(msg.response.tags, [ u"TestLabel1,TestData1", u"TestLabel2,TestData2", u"TestLabel3,TestData3", u"Query,123"])
 383
 384         # check the protobuf message corresponding to the UDP response
 385         msg = self.getFirstProtobufMessage()
 386         self.checkProtobufResponse(msg, dnsmessage_pb2.PBDNSMessage.UDP, response, '127.0.0.0')
 387         self.checkProtobufTags(msg.response.tags, [ u"TestLabel1,TestData1", u"TestLabel2,TestData2", u"TestLabel3,TestData3", u"Response,456"])
 388         self.assertEqual(len(msg.response.rrs), 1)
 389         for rr in msg.response.rrs:
 390             self.checkProtobufResponseRecord(rr, dns.rdataclass.IN, dns.rdatatype.A, name, 3600)
 391             self.assertEqual(socket.inet_ntop(socket.AF_INET, rr.rdata), '127.0.0.1')
 392
 393         (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
 394         self.assertTrue(receivedQuery)
 395         self.assertTrue(receivedResponse)
 396         receivedQuery.id = query.id
 397         self.assertEqual(query, receivedQuery)
 398         self.assertEqual(response, receivedResponse)
 399
 400         if self._protobufQueue.empty():
 401             # let the protobuf messages the time to get there
 402             time.sleep(1)
 403
 404         # check the protobuf message corresponding to the TCP query
 405         msg = self.getFirstProtobufMessage()
 406         self.checkProtobufQueryConvertedToResponse(msg, dnsmessage_pb2.PBDNSMessage.TCP, response, '127.0.0.0')
 407         self.checkProtobufTags(msg.response.tags, [ u"TestLabel1,TestData1", u"TestLabel2,TestData2", u"TestLabel3,TestData3", u"Query,123"])
 408
 409         # check the protobuf message corresponding to the TCP response
 410         msg = self.getFirstProtobufMessage()
 411         self.checkProtobufResponse(msg, dnsmessage_pb2.PBDNSMessage.TCP, response, '127.0.0.0')
 412         self.checkProtobufTags(msg.response.tags, [ u"TestLabel1,TestData1", u"TestLabel2,TestData2", u"TestLabel3,TestData3", u"Response,456"])
 413         self.assertEqual(len(msg.response.rrs), 1)
 414         for rr in msg.response.rrs:
 415             self.checkProtobufResponseRecord(rr, dns.rdataclass.IN, dns.rdatatype.A, name, 3600)
 416             self.assertEqual(socket.inet_ntop(socket.AF_INET, rr.rdata), '127.0.0.1')
 417
 418 class TestProtobufMetaTags(DNSDistProtobufTest):
 419     _config_params = ['_testServerPort', '_protobufServerPort']
 420     _config_template = """
 421     newServer{address="127.0.0.1:%s"}
 422     rl = newRemoteLogger('127.0.0.1:%d')
 423
 424     addAction(AllRule(), SetTagAction('my-tag-key', 'my-tag-value'))
 425     addAction(AllRule(), SetTagAction('my-empty-key', ''))
 426     addAction(AllRule(), RemoteLogAction(rl, nil, {serverID='dnsdist-server-1', exportTags='*'}, {b64='b64-content', ['my-tag-export-name']='tag:my-tag-key'}))
 427     addResponseAction(AllRule(), SetTagResponseAction('my-tag-key2', 'my-tag-value2'))
 428     addResponseAction(AllRule(), RemoteLogResponseAction(rl, nil, false, {serverID='dnsdist-server-1', exportTags='my-empty-key,my-tag-key2'}, {['my-tag-export-name']='tags'}))
 429     """
 430
 431     def testProtobufMeta(self):
 432         """
 433         Protobuf: Meta values
 434         """
 435         name = 'meta.protobuf.tests.powerdns.com.'
 436         query = dns.message.make_query(name, 'A', 'IN')
 437         response = dns.message.make_response(query)
 438         rrset = dns.rrset.from_text(name,
 439                                     3600,
 440                                     dns.rdataclass.IN,
 441                                     dns.rdatatype.A,
 442                                     '127.0.0.1')
 443         response.answer.append(rrset)
 444
 445         (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
 446         self.assertTrue(receivedQuery)
 447         self.assertTrue(receivedResponse)
 448         receivedQuery.id = query.id
 449         self.assertEqual(query, receivedQuery)
 450         self.assertEqual(response, receivedResponse)
 451
 452         if self._protobufQueue.empty():
 453             # let the protobuf messages the time to get there
 454             time.sleep(1)
 455
 456         # check the protobuf message corresponding to the UDP query
 457         msg = self.getFirstProtobufMessage()
 458
 459         self.checkProtobufQuery(msg, dnsmessage_pb2.PBDNSMessage.UDP, query, dns.rdataclass.IN, dns.rdatatype.A, name)
 460         # regular tags
 461         self.assertEqual(len(msg.response.tags), 2)
 462         self.assertIn('my-tag-key:my-tag-value', msg.response.tags)
 463         self.assertIn('my-empty-key', msg.response.tags)
 464         # meta tags
 465         self.assertEqual(len(msg.meta), 2)
 466         tags = {}
 467         for entry in msg.meta:
 468             tags[entry.key] = entry.value.stringVal
 469
 470         self.assertIn('b64', tags)
 471         self.assertIn('my-tag-export-name', tags)
 472
 473         b64EncodedQuery = base64.b64encode(query.to_wire()).decode('ascii')
 474         self.assertEqual(tags['b64'], [b64EncodedQuery])
 475         self.assertEqual(tags['my-tag-export-name'], ['my-tag-value'])
 476
 477         # check the protobuf message corresponding to the UDP response
 478         msg = self.getFirstProtobufMessage()
 479         self.checkProtobufResponse(msg, dnsmessage_pb2.PBDNSMessage.UDP, response)
 480         # regular tags
 481         self.assertEqual(len(msg.response.tags), 2)
 482         self.assertIn('my-tag-key2:my-tag-value2', msg.response.tags)
 483         self.assertIn('my-empty-key', msg.response.tags)
 484         # meta tags
 485         self.assertEqual(len(msg.meta), 1)
 486         self.assertEqual(msg.meta[0].key, 'my-tag-export-name')
 487         self.assertEqual(len(msg.meta[0].value.stringVal), 3)
 488         self.assertIn('my-tag-key:my-tag-value', msg.meta[0].value.stringVal)
 489         self.assertIn('my-tag-key2:my-tag-value2', msg.meta[0].value.stringVal)
 490         # no ':' when the value is empty
 491         self.assertIn('my-empty-key', msg.meta[0].value.stringVal)
 492
 493 class TestProtobufExtendedDNSErrorTags(DNSDistProtobufTest):
 494     _config_params = ['_testServerPort', '_protobufServerPort']
 495     _config_template = """
 496     newServer{address="127.0.0.1:%s"}
 497     rl = newRemoteLogger('127.0.0.1:%d')
 498
 499     addAction(AllRule(), RemoteLogAction(rl, nil, {serverID='dnsdist-server-1'}))
 500     addResponseAction(AllRule(), RemoteLogResponseAction(rl, nil, false, {serverID='dnsdist-server-1', exportExtendedErrorsToMeta='extended-error'}))
 501     """
 502
 503     def testProtobufExtendedError(self):
 504         """
 505         Protobuf: Extended Error
 506         """
 507         name = 'extended-error.protobuf.tests.powerdns.com.'
 508         query = dns.message.make_query(name, 'A', 'IN')
 509         response = dns.message.make_response(query)
 510         rrset = dns.rrset.from_text(name,
 511                                     3600,
 512                                     dns.rdataclass.IN,
 513                                     dns.rdatatype.A,
 514                                     '127.0.0.1')
 515         response.answer.append(rrset)
 516         ede = extendederrors.ExtendedErrorOption(15, b'Blocked by RPZ!')
 517         response.use_edns(edns=True, payload=4096, options=[ede])
 518
 519         (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
 520         self.assertTrue(receivedQuery)
 521         self.assertTrue(receivedResponse)
 522         receivedQuery.id = query.id
 523         self.assertEqual(query, receivedQuery)
 524         self.assertEqual(response, receivedResponse)
 525
 526         if self._protobufQueue.empty():
 527             # let the protobuf messages the time to get there
 528             time.sleep(1)
 529
 530         # check the protobuf message corresponding to the UDP query
 531         msg = self.getFirstProtobufMessage()
 532
 533         self.checkProtobufQuery(msg, dnsmessage_pb2.PBDNSMessage.UDP, query, dns.rdataclass.IN, dns.rdatatype.A, name)
 534
 535         # meta tags
 536         self.assertEqual(len(msg.meta), 0)
 537
 538         # check the protobuf message corresponding to the UDP response
 539         msg = self.getFirstProtobufMessage()
 540         self.checkProtobufResponse(msg, dnsmessage_pb2.PBDNSMessage.UDP, response)
 541
 542         # meta tags
 543         self.assertEqual(len(msg.meta), 1)
 544
 545         self.assertEqual(msg.meta[0].key, 'extended-error')
 546         self.assertEqual(len(msg.meta[0].value.intVal), 1)
 547         self.assertEqual(len(msg.meta[0].value.stringVal), 1)
 548         self.assertIn(15, msg.meta[0].value.intVal)
 549         self.assertIn('Blocked by RPZ!', msg.meta[0].value.stringVal)
 550
 551 class TestProtobufMetaDOH(DNSDistProtobufTest):
 552
 553     _serverKey = 'server.key'
 554     _serverCert = 'server.chain'
 555     _serverName = 'tls.tests.dnsdist.org'
 556     _caCert = 'ca.pem'
 557     _tlsServerPort = pickAvailablePort()
 558     _dohWithNGHTTP2ServerPort = pickAvailablePort()
 559     _dohWithNGHTTP2BaseURL = ("https://%s:%d/dns-query" % (_serverName, _dohWithNGHTTP2ServerPort))
 560     _dohWithH2OServerPort = pickAvailablePort()
 561     _dohWithH2OBaseURL = ("https://%s:%d/dns-query" % (_serverName, _dohWithH2OServerPort))
 562     _config_template = """
 563     newServer{address="127.0.0.1:%d"}
 564     rl = newRemoteLogger('127.0.0.1:%d')
 565
 566     addTLSLocal("127.0.0.1:%s", "%s", "%s", { provider="openssl" })
 567     addDOHLocal("127.0.0.1:%d", "%s", "%s", { '/dns-query' }, { keepIncomingHeaders=true, library='nghttp2' })
 568     addDOHLocal("127.0.0.1:%d", "%s", "%s", { '/dns-query' }, { keepIncomingHeaders=true, library='h2o' })
 569
 570     local mytags = {path='doh-path', host='doh-host', ['query-string']='doh-query-string', scheme='doh-scheme', agent='doh-header:user-agent'}
 571     addAction(AllRule(), RemoteLogAction(rl, nil, {serverID='dnsdist-server-1'}, mytags))
 572     addResponseAction(AllRule(), RemoteLogResponseAction(rl, nil, false, {serverID='dnsdist-server-1'}, mytags))
 573     """
 574     _config_params = ['_testServerPort', '_protobufServerPort', '_tlsServerPort', '_serverCert', '_serverKey', '_dohWithNGHTTP2ServerPort', '_serverCert', '_serverKey', '_dohWithH2OServerPort', '_serverCert', '_serverKey']
 575
 576     def testProtobufMetaDoH(self):
 577         """
 578         Protobuf: Meta values - DoH
 579         """
 580         name = 'meta-doh.protobuf.tests.powerdns.com.'
 581         query = dns.message.make_query(name, 'A', 'IN')
 582         response = dns.message.make_response(query)
 583         rrset = dns.rrset.from_text(name,
 584                                     3600,
 585                                     dns.rdataclass.IN,
 586                                     dns.rdatatype.A,
 587                                     '127.0.0.1')
 588         response.answer.append(rrset)
 589
 590         for method in ("sendUDPQuery", "sendTCPQuery", "sendDOTQueryWrapper", "sendDOHWithNGHTTP2QueryWrapper", "sendDOHWithH2OQueryWrapper"):
 591             sender = getattr(self, method)
 592             (receivedQuery, receivedResponse) = sender(query, response)
 593
 594             self.assertTrue(receivedQuery)
 595             self.assertTrue(receivedResponse)
 596             receivedQuery.id = query.id
 597             self.assertEqual(query, receivedQuery)
 598             self.assertEqual(response, receivedResponse)
 599
 600             if self._protobufQueue.empty():
 601                 # let the protobuf messages the time to get there
 602                 time.sleep(1)
 603
 604             # check the protobuf message corresponding to the query
 605             msg = self.getFirstProtobufMessage()
 606
 607             if method == "sendUDPQuery":
 608                 pbMessageType = dnsmessage_pb2.PBDNSMessage.UDP
 609             elif method == "sendTCPQuery":
 610                 pbMessageType = dnsmessage_pb2.PBDNSMessage.TCP
 611             elif method == "sendDOTQueryWrapper":
 612                 pbMessageType = dnsmessage_pb2.PBDNSMessage.DOT
 613             elif method == "sendDOHWithNGHTTP2QueryWrapper" or method == "sendDOHWithH2OQueryWrapper":
 614                 pbMessageType = dnsmessage_pb2.PBDNSMessage.DOH
 615
 616             self.checkProtobufQuery(msg, pbMessageType, query, dns.rdataclass.IN, dns.rdatatype.A, name)
 617             self.assertEqual(len(msg.meta), 5)
 618             tags = {}
 619             for entry in msg.meta:
 620                 self.assertEqual(len(entry.value.stringVal), 1)
 621                 tags[entry.key] = entry.value.stringVal[0]
 622
 623             self.assertIn('agent', tags)
 624             if method == "sendDOHWithNGHTTP2QueryWrapper" or method == "sendDOHWithH2OQueryWrapper":
 625                 self.assertIn('PycURL', tags['agent'])
 626                 self.assertIn('host', tags)
 627                 if method == "sendDOHWithNGHTTP2QueryWrapper":
 628                     self.assertEqual(tags['host'], self._serverName + ':' + str(self._dohWithNGHTTP2ServerPort))
 629                 elif method == "sendDOHWithH2OQueryWrapper":
 630                     self.assertEqual(tags['host'], self._serverName + ':' + str(self._dohWithH2OServerPort))
 631                 self.assertIn('path', tags)
 632                 self.assertEqual(tags['path'], '/dns-query')
 633                 self.assertIn('query-string', tags)
 634                 self.assertIn('?dns=', tags['query-string'])
 635                 self.assertIn('scheme', tags)
 636                 self.assertEqual(tags['scheme'], 'https')
 637
 638             # check the protobuf message corresponding to the response
 639             msg = self.getFirstProtobufMessage()
 640             self.checkProtobufResponse(msg, pbMessageType, response)
 641             self.assertEqual(len(msg.meta), 5)
 642             tags = {}
 643             for entry in msg.meta:
 644                 self.assertEqual(len(entry.value.stringVal), 1)
 645                 tags[entry.key] = entry.value.stringVal[0]
 646
 647             self.assertIn('agent', tags)
 648             if method == "sendDOHWithNGHTTP2QueryWrapper" or method == "sendDOHWithH2OQueryWrapper":
 649                 self.assertIn('PycURL', tags['agent'])
 650                 self.assertIn('host', tags)
 651                 if method == "sendDOHWithNGHTTP2QueryWrapper":
 652                     self.assertEqual(tags['host'], self._serverName + ':' + str(self._dohWithNGHTTP2ServerPort))
 653                 elif method == "sendDOHWithH2OQueryWrapper":
 654                     self.assertEqual(tags['host'], self._serverName + ':' + str(self._dohWithH2OServerPort))
 655                 self.assertIn('path', tags)
 656                 self.assertEqual(tags['path'], '/dns-query')
 657                 self.assertIn('query-string', tags)
 658                 self.assertIn('?dns=', tags['query-string'])
 659                 self.assertIn('scheme', tags)
 660                 self.assertEqual(tags['scheme'], 'https')
 661
 662 class TestProtobufMetaProxy(DNSDistProtobufTest):
 663
 664     _config_params = ['_testServerPort', '_protobufServerPort']
 665     _config_template = """
 666     setProxyProtocolACL( { "127.0.0.1/32" } )
 667
 668     newServer{address="127.0.0.1:%d"}
 669     rl = newRemoteLogger('127.0.0.1:%d')
 670
 671     local mytags = {pp='proxy-protocol-values', pp42='proxy-protocol-value:42'}
 672     addAction(AllRule(), RemoteLogAction(rl, nil, {serverID='dnsdist-server-1'}, mytags))
 673
 674     -- proxy protocol values are NOT passed to the response
 675     """
 676
 677     def testProtobufMetaProxy(self):
 678         """
 679         Protobuf: Meta values - Proxy
 680         """
 681         name = 'meta-proxy.protobuf.tests.powerdns.com.'
 682         query = dns.message.make_query(name, 'A', 'IN')
 683         response = dns.message.make_response(query)
 684         rrset = dns.rrset.from_text(name,
 685                                     3600,
 686                                     dns.rdataclass.IN,
 687                                     dns.rdatatype.A,
 688                                     '127.0.0.1')
 689         response.answer.append(rrset)
 690
 691         destAddr = "2001:db8::9"
 692         destPort = 9999
 693         srcAddr = "2001:db8::8"
 694         srcPort = 8888
 695         udpPayload = ProxyProtocol.getPayload(False, False, True, srcAddr, destAddr, srcPort, destPort, [ [ 2, b'foo'], [ 42, b'proxy'] ])
 696         (receivedQuery, receivedResponse) = self.sendUDPQuery(udpPayload + query.to_wire(), response, rawQuery=True)
 697
 698         self.assertTrue(receivedQuery)
 699         self.assertTrue(receivedResponse)
 700         receivedQuery.id = query.id
 701         self.assertEqual(query, receivedQuery)
 702         self.assertEqual(response, receivedResponse)
 703
 704         if self._protobufQueue.empty():
 705             # let the protobuf messages the time to get there
 706             time.sleep(1)
 707
 708         # check the protobuf message corresponding to the UDP query
 709         msg = self.getFirstProtobufMessage()
 710
 711         self.checkProtobufQuery(msg, dnsmessage_pb2.PBDNSMessage.UDP, query, dns.rdataclass.IN, dns.rdatatype.A, name, initiator='2001:db8::8', v6=True)
 712         self.assertEqual(len(msg.meta), 2)
 713         tags = {}
 714         for entry in msg.meta:
 715             tags[entry.key] = entry.value.stringVal
 716
 717         self.assertIn('pp42', tags)
 718         self.assertEqual(tags['pp42'], ['proxy'])
 719         self.assertIn('pp', tags)
 720         self.assertEqual(len(tags['pp']), 2)
 721         self.assertIn('2:foo', tags['pp'])
 722         self.assertIn('42:proxy', tags['pp'])
 723
 724 class TestProtobufIPCipher(DNSDistProtobufTest):
 725     _config_params = ['_testServerPort', '_protobufServerPort', '_protobufServerID', '_protobufServerID']
 726     _config_template = """
 727     newServer{address="127.0.0.1:%s", useClientSubnet=true}
 728     key = makeIPCipherKey("some 16-byte key")
 729     rl = newRemoteLogger('127.0.0.1:%s')
 730     addAction(AllRule(), RemoteLogAction(rl, nil, {serverID='%s', ipEncryptKey=key})) -- Send protobuf message before lookup
 731     addResponseAction(AllRule(), RemoteLogResponseAction(rl, nil, true, {serverID='%s', ipEncryptKey=key})) -- Send protobuf message after lookup
 732
 733     """
 734
 735     def testProtobuf(self):
 736         """
 737         Protobuf: Send data to a protobuf server, with pseudonymization
 738         """
 739         name = 'query.protobuf-ipcipher.tests.powerdns.com.'
 740
 741         target = 'target.protobuf-ipcipher.tests.powerdns.com.'
 742         query = dns.message.make_query(name, 'A', 'IN')
 743         response = dns.message.make_response(query)
 744
 745         rrset = dns.rrset.from_text(name,
 746                                     3600,
 747                                     dns.rdataclass.IN,
 748                                     dns.rdatatype.CNAME,
 749                                     target)
 750         response.answer.append(rrset)
 751
 752         rrset = dns.rrset.from_text(target,
 753                                     3600,
 754                                     dns.rdataclass.IN,
 755                                     dns.rdatatype.A,
 756                                     '127.0.0.1')
 757         response.answer.append(rrset)
 758
 759         (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
 760         self.assertTrue(receivedQuery)
 761         self.assertTrue(receivedResponse)
 762         receivedQuery.id = query.id
 763         self.assertEqual(query, receivedQuery)
 764         self.assertEqual(response, receivedResponse)
 765
 766         if self._protobufQueue.empty():
 767             # let the protobuf messages the time to get there
 768             time.sleep(1)
 769
 770         # check the protobuf message corresponding to the UDP query
 771         msg = self.getFirstProtobufMessage()
 772
 773         # 108.41.239.98 is 127.0.0.1 pseudonymized with ipcipher and the current key
 774         self.checkProtobufQuery(msg, dnsmessage_pb2.PBDNSMessage.UDP, query, dns.rdataclass.IN, dns.rdatatype.A, name, '108.41.239.98')
 775
 776         # check the protobuf message corresponding to the UDP response
 777         msg = self.getFirstProtobufMessage()
 778         self.checkProtobufResponse(msg, dnsmessage_pb2.PBDNSMessage.UDP, response, '108.41.239.98')
 779
 780         self.assertEqual(len(msg.response.rrs), 2)
 781         rr = msg.response.rrs[0]
 782         self.checkProtobufResponseRecord(rr, dns.rdataclass.IN, dns.rdatatype.CNAME, name, 3600)
 783         self.assertEqual(rr.rdata.decode('ascii'), target)
 784         rr = msg.response.rrs[1]
 785         self.checkProtobufResponseRecord(rr, dns.rdataclass.IN, dns.rdatatype.A, target, 3600)
 786         self.assertEqual(socket.inet_ntop(socket.AF_INET, rr.rdata), '127.0.0.1')
 787
 788         (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
 789         self.assertTrue(receivedQuery)
 790         self.assertTrue(receivedResponse)
 791         receivedQuery.id = query.id
 792         self.assertEqual(query, receivedQuery)
 793         self.assertEqual(response, receivedResponse)
 794
 795         if self._protobufQueue.empty():
 796             # let the protobuf messages the time to get there
 797             time.sleep(1)
 798
 799         # check the protobuf message corresponding to the TCP query
 800         msg = self.getFirstProtobufMessage()
 801         # 108.41.239.98 is 127.0.0.1 pseudonymized with ipcipher and the current key
 802         self.checkProtobufQuery(msg, dnsmessage_pb2.PBDNSMessage.TCP, query, dns.rdataclass.IN, dns.rdatatype.A, name, '108.41.239.98')
 803
 804         # check the protobuf message corresponding to the TCP response
 805         msg = self.getFirstProtobufMessage()
 806         self.checkProtobufResponse(msg, dnsmessage_pb2.PBDNSMessage.TCP, response, '108.41.239.98')
 807         self.assertEqual(len(msg.response.rrs), 2)
 808         rr = msg.response.rrs[0]
 809         self.checkProtobufResponseRecord(rr, dns.rdataclass.IN, dns.rdatatype.CNAME, name, 3600)
 810         self.assertEqual(rr.rdata.decode('ascii'), target)
 811         rr = msg.response.rrs[1]
 812         self.checkProtobufResponseRecord(rr, dns.rdataclass.IN, dns.rdatatype.A, target, 3600)
 813         self.assertEqual(socket.inet_ntop(socket.AF_INET, rr.rdata), '127.0.0.1')