regression-tests.dnsdist/test_ProxyProtocol.py

   1 #!/usr/bin/env python
   2
   3 import dns
   4 import socket
   5 import struct
   6 import sys
   7 import threading
   8
   9 from dnsdisttests import DNSDistTest
  10 from proxyprotocol import ProxyProtocol
  11
  12 # Python2/3 compatibility hacks
  13 try:
  14   from queue import Queue
  15 except ImportError:
  16   from Queue import Queue
  17
  18 def ProxyProtocolUDPResponder(port, fromQueue, toQueue):
  19     sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
  20     sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
  21     try:
  22         sock.bind(("127.0.0.1", port))
  23     except socket.error as e:
  24         print("Error binding in the Proxy Protocol UDP responder: %s" % str(e))
  25         sys.exit(1)
  26
  27     while True:
  28         data, addr = sock.recvfrom(4096)
  29
  30         proxy = ProxyProtocol()
  31         if len(data) < proxy.HEADER_SIZE:
  32             continue
  33
  34         if not proxy.parseHeader(data):
  35             continue
  36
  37         if proxy.local:
  38             # likely a healthcheck
  39             data = data[proxy.HEADER_SIZE:]
  40             request = dns.message.from_wire(data)
  41             response = dns.message.make_response(request)
  42             wire = response.to_wire()
  43             sock.settimeout(2.0)
  44             sock.sendto(wire, addr)
  45             sock.settimeout(None)
  46
  47             continue
  48
  49         payload = data[:(proxy.HEADER_SIZE + proxy.contentLen)]
  50         dnsData = data[(proxy.HEADER_SIZE + proxy.contentLen):]
  51         toQueue.put([payload, dnsData], True, 2.0)
  52         # computing the correct ID for the response
  53         request = dns.message.from_wire(dnsData)
  54         response = fromQueue.get(True, 2.0)
  55         response.id = request.id
  56
  57         sock.settimeout(2.0)
  58         sock.sendto(response.to_wire(), addr)
  59         sock.settimeout(None)
  60
  61     sock.close()
  62
  63 def ProxyProtocolTCPResponder(port, fromQueue, toQueue):
  64     # be aware that this responder will not accept a new connection
  65     # until the last one has been closed. This is done on purpose to
  66     # to check for connection reuse, making sure that a lot of connections
  67     # are not opened in parallel.
  68     sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  69     sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
  70     sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
  71     try:
  72         sock.bind(("127.0.0.1", port))
  73     except socket.error as e:
  74         print("Error binding in the TCP responder: %s" % str(e))
  75         sys.exit(1)
  76
  77     sock.listen(100)
  78     while True:
  79         (conn, _) = sock.accept()
  80         conn.settimeout(5.0)
  81         # try to read the entire Proxy Protocol header
  82         proxy = ProxyProtocol()
  83         header = conn.recv(proxy.HEADER_SIZE)
  84         if not header:
  85             conn.close()
  86             continue
  87
  88         if not proxy.parseHeader(header):
  89             conn.close()
  90             continue
  91
  92         proxyContent = conn.recv(proxy.contentLen)
  93         if not proxyContent:
  94             conn.close()
  95             continue
  96
  97         payload = header + proxyContent
  98         while True:
  99           try:
 100             data = conn.recv(2)
 101           except socket.timeout:
 102             data = None
 103
 104           if not data:
 105             conn.close()
 106             break
 107
 108           (datalen,) = struct.unpack("!H", data)
 109           data = conn.recv(datalen)
 110
 111           toQueue.put([payload, data], True, 2.0)
 112
 113           response = fromQueue.get(True, 2.0)
 114           if not response:
 115             conn.close()
 116             break
 117
 118           # computing the correct ID for the response
 119           request = dns.message.from_wire(data)
 120           response.id = request.id
 121
 122           wire = response.to_wire()
 123           conn.send(struct.pack("!H", len(wire)))
 124           conn.send(wire)
 125
 126         conn.close()
 127
 128     sock.close()
 129
 130 toProxyQueue = Queue()
 131 fromProxyQueue = Queue()
 132 proxyResponderPort = 5470
 133
 134 udpResponder = threading.Thread(name='UDP Proxy Protocol Responder', target=ProxyProtocolUDPResponder, args=[proxyResponderPort, toProxyQueue, fromProxyQueue])
 135 udpResponder.setDaemon(True)
 136 udpResponder.start()
 137 tcpResponder = threading.Thread(name='TCP Proxy Protocol Responder', target=ProxyProtocolTCPResponder, args=[proxyResponderPort, toProxyQueue, fromProxyQueue])
 138 tcpResponder.setDaemon(True)
 139 tcpResponder.start()
 140
 141 class ProxyProtocolTest(DNSDistTest):
 142     _proxyResponderPort = proxyResponderPort
 143     _config_params = ['_proxyResponderPort']
 144
 145     def checkMessageProxyProtocol(self, receivedProxyPayload, source, destination, isTCP, values=[], v6=False, sourcePort=None, destinationPort=None):
 146       proxy = ProxyProtocol()
 147       self.assertTrue(proxy.parseHeader(receivedProxyPayload))
 148       self.assertEqual(proxy.version, 0x02)
 149       self.assertEqual(proxy.command, 0x01)
 150       if v6:
 151         self.assertEqual(proxy.family, 0x02)
 152       else:
 153         self.assertEqual(proxy.family, 0x01)
 154       if not isTCP:
 155         self.assertEqual(proxy.protocol, 0x02)
 156       else:
 157         self.assertEqual(proxy.protocol, 0x01)
 158       self.assertGreater(proxy.contentLen, 0)
 159
 160       self.assertTrue(proxy.parseAddressesAndPorts(receivedProxyPayload))
 161       self.assertEqual(proxy.source, source)
 162       self.assertEqual(proxy.destination, destination)
 163       if sourcePort:
 164         self.assertEqual(proxy.sourcePort, sourcePort)
 165       if destinationPort:
 166         self.assertEqual(proxy.destinationPort, destinationPort)
 167       else:
 168         self.assertEqual(proxy.destinationPort, self._dnsDistPort)
 169
 170       self.assertTrue(proxy.parseAdditionalValues(receivedProxyPayload))
 171       proxy.values.sort()
 172       values.sort()
 173       self.assertEqual(proxy.values, values)
 174
 175 class TestProxyProtocol(ProxyProtocolTest):
 176     """
 177     dnsdist is configured to prepend a Proxy Protocol header to the query
 178     """
 179
 180     _config_template = """
 181     newServer{address="127.0.0.1:%d", useProxyProtocol=true}
 182
 183     function addValues(dq)
 184       local values = { [0]="foo", [42]="bar" }
 185       dq:setProxyProtocolValues(values)
 186       return DNSAction.None
 187     end
 188
 189     addAction("values-lua.proxy.tests.powerdns.com.", LuaAction(addValues))
 190     addAction("values-action.proxy.tests.powerdns.com.", SetProxyProtocolValuesAction({ ["1"]="dnsdist", ["255"]="proxy-protocol"}))
 191     """
 192     _config_params = ['_proxyResponderPort']
 193
 194     def testProxyUDP(self):
 195         """
 196         Proxy Protocol: no value (UDP)
 197         """
 198         name = 'simple-udp.proxy.tests.powerdns.com.'
 199         query = dns.message.make_query(name, 'A', 'IN')
 200         response = dns.message.make_response(query)
 201
 202         toProxyQueue.put(response, True, 2.0)
 203
 204         data = query.to_wire()
 205         self._sock.send(data)
 206         receivedResponse = None
 207         try:
 208             self._sock.settimeout(2.0)
 209             data = self._sock.recv(4096)
 210         except socket.timeout:
 211             print('timeout')
 212             data = None
 213         if data:
 214             receivedResponse = dns.message.from_wire(data)
 215
 216         (receivedProxyPayload, receivedDNSData) = fromProxyQueue.get(True, 2.0)
 217         self.assertTrue(receivedProxyPayload)
 218         self.assertTrue(receivedDNSData)
 219         self.assertTrue(receivedResponse)
 220
 221         receivedQuery = dns.message.from_wire(receivedDNSData)
 222         receivedQuery.id = query.id
 223         receivedResponse.id = response.id
 224         self.assertEqual(receivedQuery, query)
 225         self.assertEqual(receivedResponse, response)
 226         self.checkMessageProxyProtocol(receivedProxyPayload, '127.0.0.1', '127.0.0.1', False)
 227
 228     def testProxyTCP(self):
 229       """
 230         Proxy Protocol: no value (TCP)
 231       """
 232       name = 'simple-tcp.proxy.tests.powerdns.com.'
 233       query = dns.message.make_query(name, 'A', 'IN')
 234       response = dns.message.make_response(query)
 235
 236       toProxyQueue.put(response, True, 2.0)
 237
 238       conn = self.openTCPConnection(2.0)
 239       data = query.to_wire()
 240       self.sendTCPQueryOverConnection(conn, data, rawQuery=True)
 241       receivedResponse = None
 242       try:
 243         receivedResponse = self.recvTCPResponseOverConnection(conn)
 244       except socket.timeout:
 245             print('timeout')
 246
 247       (receivedProxyPayload, receivedDNSData) = fromProxyQueue.get(True, 2.0)
 248       self.assertTrue(receivedProxyPayload)
 249       self.assertTrue(receivedDNSData)
 250       self.assertTrue(receivedResponse)
 251
 252       receivedQuery = dns.message.from_wire(receivedDNSData)
 253       receivedQuery.id = query.id
 254       receivedResponse.id = response.id
 255       self.assertEqual(receivedQuery, query)
 256       self.assertEqual(receivedResponse, response)
 257       self.checkMessageProxyProtocol(receivedProxyPayload, '127.0.0.1', '127.0.0.1', True)
 258
 259     def testProxyUDPWithValuesFromLua(self):
 260         """
 261         Proxy Protocol: values from Lua (UDP)
 262         """
 263         name = 'values-lua.proxy.tests.powerdns.com.'
 264         query = dns.message.make_query(name, 'A', 'IN')
 265         response = dns.message.make_response(query)
 266
 267         toProxyQueue.put(response, True, 2.0)
 268
 269         data = query.to_wire()
 270         self._sock.send(data)
 271         receivedResponse = None
 272         try:
 273             self._sock.settimeout(2.0)
 274             data = self._sock.recv(4096)
 275         except socket.timeout:
 276             print('timeout')
 277             data = None
 278         if data:
 279             receivedResponse = dns.message.from_wire(data)
 280
 281         (receivedProxyPayload, receivedDNSData) = fromProxyQueue.get(True, 2.0)
 282         self.assertTrue(receivedProxyPayload)
 283         self.assertTrue(receivedDNSData)
 284         self.assertTrue(receivedResponse)
 285
 286         receivedQuery = dns.message.from_wire(receivedDNSData)
 287         receivedQuery.id = query.id
 288         receivedResponse.id = response.id
 289         self.assertEqual(receivedQuery, query)
 290         self.assertEqual(receivedResponse, response)
 291         self.checkMessageProxyProtocol(receivedProxyPayload, '127.0.0.1', '127.0.0.1', False, [ [0, b'foo'] , [ 42, b'bar'] ])
 292
 293     def testProxyTCPWithValuesFromLua(self):
 294       """
 295         Proxy Protocol: values from Lua (TCP)
 296       """
 297       name = 'values-lua.proxy.tests.powerdns.com.'
 298       query = dns.message.make_query(name, 'A', 'IN')
 299       response = dns.message.make_response(query)
 300
 301       toProxyQueue.put(response, True, 2.0)
 302
 303       conn = self.openTCPConnection(2.0)
 304       data = query.to_wire()
 305       self.sendTCPQueryOverConnection(conn, data, rawQuery=True)
 306       receivedResponse = None
 307       try:
 308         receivedResponse = self.recvTCPResponseOverConnection(conn)
 309       except socket.timeout:
 310             print('timeout')
 311
 312       (receivedProxyPayload, receivedDNSData) = fromProxyQueue.get(True, 2.0)
 313       self.assertTrue(receivedProxyPayload)
 314       self.assertTrue(receivedDNSData)
 315       self.assertTrue(receivedResponse)
 316
 317       receivedQuery = dns.message.from_wire(receivedDNSData)
 318       receivedQuery.id = query.id
 319       receivedResponse.id = response.id
 320       self.assertEqual(receivedQuery, query)
 321       self.assertEqual(receivedResponse, response)
 322       self.checkMessageProxyProtocol(receivedProxyPayload, '127.0.0.1', '127.0.0.1', True, [ [0, b'foo'] , [ 42, b'bar'] ])
 323
 324     def testProxyUDPWithValuesFromAction(self):
 325         """
 326         Proxy Protocol: values from Action (UDP)
 327         """
 328         name = 'values-action.proxy.tests.powerdns.com.'
 329         query = dns.message.make_query(name, 'A', 'IN')
 330         response = dns.message.make_response(query)
 331
 332         toProxyQueue.put(response, True, 2.0)
 333
 334         data = query.to_wire()
 335         self._sock.send(data)
 336         receivedResponse = None
 337         try:
 338             self._sock.settimeout(2.0)
 339             data = self._sock.recv(4096)
 340         except socket.timeout:
 341             print('timeout')
 342             data = None
 343         if data:
 344             receivedResponse = dns.message.from_wire(data)
 345
 346         (receivedProxyPayload, receivedDNSData) = fromProxyQueue.get(True, 2.0)
 347         self.assertTrue(receivedProxyPayload)
 348         self.assertTrue(receivedDNSData)
 349         self.assertTrue(receivedResponse)
 350
 351         receivedQuery = dns.message.from_wire(receivedDNSData)
 352         receivedQuery.id = query.id
 353         receivedResponse.id = response.id
 354         self.assertEqual(receivedQuery, query)
 355         self.assertEqual(receivedResponse, response)
 356         self.checkMessageProxyProtocol(receivedProxyPayload, '127.0.0.1', '127.0.0.1', False, [ [1, b'dnsdist'] , [ 255, b'proxy-protocol'] ])
 357
 358     def testProxyTCPWithValuesFromAction(self):
 359       """
 360         Proxy Protocol: values from Action (TCP)
 361       """
 362       name = 'values-action.proxy.tests.powerdns.com.'
 363       query = dns.message.make_query(name, 'A', 'IN')
 364       response = dns.message.make_response(query)
 365
 366       toProxyQueue.put(response, True, 2.0)
 367
 368       conn = self.openTCPConnection(2.0)
 369       data = query.to_wire()
 370       self.sendTCPQueryOverConnection(conn, data, rawQuery=True)
 371       receivedResponse = None
 372       try:
 373         receivedResponse = self.recvTCPResponseOverConnection(conn)
 374       except socket.timeout:
 375             print('timeout')
 376
 377       (receivedProxyPayload, receivedDNSData) = fromProxyQueue.get(True, 2.0)
 378       self.assertTrue(receivedProxyPayload)
 379       self.assertTrue(receivedDNSData)
 380       self.assertTrue(receivedResponse)
 381
 382       receivedQuery = dns.message.from_wire(receivedDNSData)
 383       receivedQuery.id = query.id
 384       receivedResponse.id = response.id
 385       self.assertEqual(receivedQuery, query)
 386       self.assertEqual(receivedResponse, response)
 387       self.checkMessageProxyProtocol(receivedProxyPayload, '127.0.0.1', '127.0.0.1', True, [ [1, b'dnsdist'] , [ 255, b'proxy-protocol'] ])
 388
 389     def testProxyTCPSeveralQueriesOnSameConnection(self):
 390       """
 391         Proxy Protocol: Several queries on the same TCP connection
 392       """
 393       name = 'several-queries-same-conn.proxy.tests.powerdns.com.'
 394       query = dns.message.make_query(name, 'A', 'IN')
 395       response = dns.message.make_response(query)
 396
 397       conn = self.openTCPConnection(2.0)
 398       data = query.to_wire()
 399
 400       for idx in range(10):
 401         toProxyQueue.put(response, True, 2.0)
 402         self.sendTCPQueryOverConnection(conn, data, rawQuery=True)
 403         receivedResponse = None
 404         try:
 405           receivedResponse = self.recvTCPResponseOverConnection(conn)
 406         except socket.timeout:
 407           print('timeout')
 408
 409         (receivedProxyPayload, receivedDNSData) = fromProxyQueue.get(True, 2.0)
 410         self.assertTrue(receivedProxyPayload)
 411         self.assertTrue(receivedDNSData)
 412         self.assertTrue(receivedResponse)
 413
 414         receivedQuery = dns.message.from_wire(receivedDNSData)
 415         receivedQuery.id = query.id
 416         receivedResponse.id = response.id
 417         self.assertEqual(receivedQuery, query)
 418         self.assertEqual(receivedResponse, response)
 419         self.checkMessageProxyProtocol(receivedProxyPayload, '127.0.0.1', '127.0.0.1', True, [])
 420
 421 class TestProxyProtocolIncoming(ProxyProtocolTest):
 422     """
 423     dnsdist is configured to prepend a Proxy Protocol header to the query and expect one on incoming queries
 424     """
 425
 426     _config_template = """
 427     setProxyProtocolACL( { "127.0.0.1/32" } )
 428     newServer{address="127.0.0.1:%d", useProxyProtocol=true}
 429
 430     function addValues(dq)
 431       dq:addProxyProtocolValue(0, 'foo')
 432       dq:addProxyProtocolValue(42, 'bar')
 433       return DNSAction.None
 434     end
 435
 436     -- refuse queries with no TLV value type 2
 437     addAction(NotRule(ProxyProtocolValueRule(2)), RCodeAction(DNSRCode.REFUSED))
 438     -- or with a TLV value type 3 different from "proxy"
 439     addAction(NotRule(ProxyProtocolValueRule(3, "proxy")), RCodeAction(DNSRCode.REFUSED))
 440
 441     function answerBasedOnForwardedDest(dq)
 442       local port = dq.localaddr:getPort()
 443       local dest = dq.localaddr:toString()
 444       return DNSAction.Spoof, "address-was-"..dest.."-port-was-"..port..".proxy-protocol-incoming.tests.powerdns.com."
 445     end
 446     addAction("get-forwarded-dest.proxy-protocol-incoming.tests.powerdns.com.", LuaAction(answerBasedOnForwardedDest))
 447
 448     function answerBasedOnForwardedSrc(dq)
 449       local port = dq.remoteaddr:getPort()
 450       local src = dq.remoteaddr:toString()
 451       return DNSAction.Spoof, "address-was-"..src.."-port-was-"..port..".proxy-protocol-incoming.tests.powerdns.com."
 452     end
 453     addAction("get-forwarded-src.proxy-protocol-incoming.tests.powerdns.com.", LuaAction(answerBasedOnForwardedSrc))
 454
 455     -- add these values for all queries
 456     addAction("proxy-protocol-incoming.tests.powerdns.com.", LuaAction(addValues))
 457     addAction("proxy-protocol-incoming.tests.powerdns.com.", SetAdditionalProxyProtocolValueAction(1, "dnsdist"))
 458     addAction("proxy-protocol-incoming.tests.powerdns.com.", SetAdditionalProxyProtocolValueAction(255, "proxy-protocol"))
 459
 460     -- override all existing values
 461     addAction("override.proxy-protocol-incoming.tests.powerdns.com.", SetProxyProtocolValuesAction({["50"]="overridden"}))
 462     """
 463     _config_params = ['_proxyResponderPort']
 464     _verboseMode = True
 465
 466     def testNoHeader(self):
 467         """
 468         Incoming Proxy Protocol: no header
 469         """
 470         # no proxy protocol header while one is expected, should be dropped
 471         name = 'no-header.incoming-proxy-protocol.tests.powerdns.com.'
 472         query = dns.message.make_query(name, 'A', 'IN')
 473
 474         for method in ("sendUDPQuery", "sendTCPQuery"):
 475           sender = getattr(self, method)
 476           (_, receivedResponse) = sender(query, response=None)
 477           self.assertEqual(receivedResponse, None)
 478
 479     def testIncomingProxyDest(self):
 480         """
 481         Incoming Proxy Protocol: values from Lua
 482         """
 483         name = 'get-forwarded-dest.proxy-protocol-incoming.tests.powerdns.com.'
 484         query = dns.message.make_query(name, 'A', 'IN')
 485         # dnsdist set RA = RD for spoofed responses
 486         query.flags &= ~dns.flags.RD
 487
 488         destAddr = "2001:db8::9"
 489         destPort = 9999
 490         srcAddr = "2001:db8::8"
 491         srcPort = 8888
 492         response = dns.message.make_response(query)
 493         rrset = dns.rrset.from_text(name,
 494                                     60,
 495                                     dns.rdataclass.IN,
 496                                     dns.rdatatype.CNAME,
 497                                     "address-was-{}-port-was-{}.proxy-protocol-incoming.tests.powerdns.com.".format(destAddr, destPort, self._dnsDistPort))
 498         response.answer.append(rrset)
 499
 500         udpPayload = ProxyProtocol.getPayload(False, False, True, srcAddr, destAddr, srcPort, destPort, [ [ 2, b'foo'], [ 3, b'proxy'] ])
 501         (_, receivedResponse) = self.sendUDPQuery(udpPayload + query.to_wire(), response=None, useQueue=False, rawQuery=True)
 502         self.assertEqual(receivedResponse, response)
 503
 504         tcpPayload = ProxyProtocol.getPayload(False, True, True, srcAddr, destAddr, srcPort, destPort, [ [ 2, b'foo'], [ 3, b'proxy'] ])
 505         wire = query.to_wire()
 506
 507         receivedResponse = None
 508         try:
 509           conn = self.openTCPConnection(2.0)
 510           conn.send(tcpPayload)
 511           conn.send(struct.pack("!H", len(wire)))
 512           conn.send(wire)
 513           receivedResponse = self.recvTCPResponseOverConnection(conn)
 514         except socket.timeout:
 515           print('timeout')
 516         self.assertEqual(receivedResponse, response)
 517
 518     def testProxyUDPWithValuesFromLua(self):
 519         """
 520         Incoming Proxy Protocol: values from Lua (UDP)
 521         """
 522         name = 'values-lua.proxy-protocol-incoming.tests.powerdns.com.'
 523         query = dns.message.make_query(name, 'A', 'IN')
 524         response = dns.message.make_response(query)
 525
 526         destAddr = "2001:db8::9"
 527         destPort = 9999
 528         srcAddr = "2001:db8::8"
 529         srcPort = 8888
 530         response = dns.message.make_response(query)
 531
 532         udpPayload = ProxyProtocol.getPayload(False, False, True, srcAddr, destAddr, srcPort, destPort, [ [ 2, b'foo'], [ 3, b'proxy'] ])
 533         toProxyQueue.put(response, True, 2.0)
 534         (_, receivedResponse) = self.sendUDPQuery(udpPayload + query.to_wire(), response=None, useQueue=False, rawQuery=True)
 535
 536         (receivedProxyPayload, receivedDNSData) = fromProxyQueue.get(True, 2.0)
 537         self.assertTrue(receivedProxyPayload)
 538         self.assertTrue(receivedDNSData)
 539         self.assertTrue(receivedResponse)
 540
 541         receivedQuery = dns.message.from_wire(receivedDNSData)
 542         receivedQuery.id = query.id
 543         receivedResponse.id = response.id
 544         self.assertEqual(receivedQuery, query)
 545         self.assertEqual(receivedResponse, response)
 546         self.checkMessageProxyProtocol(receivedProxyPayload, srcAddr, destAddr, False, [ [0, b'foo'], [1, b'dnsdist'], [ 2, b'foo'], [3, b'proxy'], [ 42, b'bar'], [255, b'proxy-protocol'] ], True, srcPort, destPort)
 547
 548     def testProxyTCPWithValuesFromLua(self):
 549         """
 550         Incoming Proxy Protocol: values from Lua (TCP)
 551         """
 552         name = 'values-lua.proxy-protocol-incoming.tests.powerdns.com.'
 553         query = dns.message.make_query(name, 'A', 'IN')
 554         response = dns.message.make_response(query)
 555
 556         destAddr = "2001:db8::9"
 557         destPort = 9999
 558         srcAddr = "2001:db8::8"
 559         srcPort = 8888
 560         response = dns.message.make_response(query)
 561
 562         tcpPayload = ProxyProtocol.getPayload(False, True, True, srcAddr, destAddr, srcPort, destPort, [ [ 2, b'foo'], [ 3, b'proxy'] ])
 563
 564         toProxyQueue.put(response, True, 2.0)
 565
 566         wire = query.to_wire()
 567
 568         receivedResponse = None
 569         try:
 570           conn = self.openTCPConnection(2.0)
 571           conn.send(tcpPayload)
 572           conn.send(struct.pack("!H", len(wire)))
 573           conn.send(wire)
 574           receivedResponse = self.recvTCPResponseOverConnection(conn)
 575         except socket.timeout:
 576           print('timeout')
 577         self.assertEqual(receivedResponse, response)
 578
 579         (receivedProxyPayload, receivedDNSData) = fromProxyQueue.get(True, 2.0)
 580         self.assertTrue(receivedProxyPayload)
 581         self.assertTrue(receivedDNSData)
 582         self.assertTrue(receivedResponse)
 583
 584         receivedQuery = dns.message.from_wire(receivedDNSData)
 585         receivedQuery.id = query.id
 586         receivedResponse.id = response.id
 587         self.assertEqual(receivedQuery, query)
 588         self.assertEqual(receivedResponse, response)
 589         self.checkMessageProxyProtocol(receivedProxyPayload, srcAddr, destAddr, True, [ [0, b'foo'], [1, b'dnsdist'], [ 2, b'foo'], [3, b'proxy'], [ 42, b'bar'], [255, b'proxy-protocol'] ], True, srcPort, destPort)
 590
 591     def testProxyUDPWithValueOverride(self):
 592         """
 593         Incoming Proxy Protocol: override existing value (UDP)
 594         """
 595         name = 'override.proxy-protocol-incoming.tests.powerdns.com.'
 596         query = dns.message.make_query(name, 'A', 'IN')
 597         response = dns.message.make_response(query)
 598
 599         destAddr = "2001:db8::9"
 600         destPort = 9999
 601         srcAddr = "2001:db8::8"
 602         srcPort = 8888
 603         response = dns.message.make_response(query)
 604
 605         udpPayload = ProxyProtocol.getPayload(False, False, True, srcAddr, destAddr, srcPort, destPort, [ [2, b'foo'], [3, b'proxy'], [ 50, b'initial-value']])
 606         toProxyQueue.put(response, True, 2.0)
 607         (_, receivedResponse) = self.sendUDPQuery(udpPayload + query.to_wire(), response=None, useQueue=False, rawQuery=True)
 608
 609         (receivedProxyPayload, receivedDNSData) = fromProxyQueue.get(True, 2.0)
 610         self.assertTrue(receivedProxyPayload)
 611         self.assertTrue(receivedDNSData)
 612         self.assertTrue(receivedResponse)
 613
 614         receivedQuery = dns.message.from_wire(receivedDNSData)
 615         receivedQuery.id = query.id
 616         receivedResponse.id = response.id
 617         self.assertEqual(receivedQuery, query)
 618         self.assertEqual(receivedResponse, response)
 619         self.checkMessageProxyProtocol(receivedProxyPayload, srcAddr, destAddr, False, [ [50, b'overridden'] ], True, srcPort, destPort)
 620
 621     def testProxyTCPSeveralQueriesOverConnection(self):
 622         """
 623         Incoming Proxy Protocol: Several queries over the same connection (TCP)
 624         """
 625         name = 'several-queries.proxy-protocol-incoming.tests.powerdns.com.'
 626         query = dns.message.make_query(name, 'A', 'IN')
 627         response = dns.message.make_response(query)
 628
 629         destAddr = "2001:db8::9"
 630         destPort = 9999
 631         srcAddr = "2001:db8::8"
 632         srcPort = 8888
 633         response = dns.message.make_response(query)
 634
 635         tcpPayload = ProxyProtocol.getPayload(False, True, True, srcAddr, destAddr, srcPort, destPort, [ [ 2, b'foo'], [ 3, b'proxy'] ])
 636
 637         toProxyQueue.put(response, True, 2.0)
 638
 639         wire = query.to_wire()
 640
 641         receivedResponse = None
 642         conn = self.openTCPConnection(2.0)
 643         try:
 644           conn.send(tcpPayload)
 645           conn.send(struct.pack("!H", len(wire)))
 646           conn.send(wire)
 647           receivedResponse = self.recvTCPResponseOverConnection(conn)
 648         except socket.timeout:
 649           print('timeout')
 650         self.assertEqual(receivedResponse, response)
 651
 652         (receivedProxyPayload, receivedDNSData) = fromProxyQueue.get(True, 2.0)
 653         self.assertTrue(receivedProxyPayload)
 654         self.assertTrue(receivedDNSData)
 655         self.assertTrue(receivedResponse)
 656
 657         receivedQuery = dns.message.from_wire(receivedDNSData)
 658         receivedQuery.id = query.id
 659         receivedResponse.id = response.id
 660         self.assertEqual(receivedQuery, query)
 661         self.assertEqual(receivedResponse, response)
 662         self.checkMessageProxyProtocol(receivedProxyPayload, srcAddr, destAddr, True, [ [0, b'foo'], [1, b'dnsdist'], [ 2, b'foo'], [3, b'proxy'], [ 42, b'bar'], [255, b'proxy-protocol'] ], True, srcPort, destPort)
 663
 664         for idx in range(5):
 665           receivedResponse = None
 666           toProxyQueue.put(response, True, 2.0)
 667           try:
 668             conn.send(struct.pack("!H", len(wire)))
 669             conn.send(wire)
 670             receivedResponse = self.recvTCPResponseOverConnection(conn)
 671           except socket.timeout:
 672             print('timeout')
 673
 674           self.assertEqual(receivedResponse, response)
 675
 676           (receivedProxyPayload, receivedDNSData) = fromProxyQueue.get(True, 2.0)
 677           self.assertTrue(receivedProxyPayload)
 678           self.assertTrue(receivedDNSData)
 679           self.assertTrue(receivedResponse)
 680
 681           receivedQuery = dns.message.from_wire(receivedDNSData)
 682           receivedQuery.id = query.id
 683           receivedResponse.id = response.id
 684           self.assertEqual(receivedQuery, query)
 685           self.assertEqual(receivedResponse, response)
 686           self.checkMessageProxyProtocol(receivedProxyPayload, srcAddr, destAddr, True, [ [0, b'foo'], [1, b'dnsdist'], [ 2, b'foo'], [3, b'proxy'], [ 42, b'bar'], [255, b'proxy-protocol'] ], True, srcPort, destPort)
 687
 688 class TestProxyProtocolNotExpected(DNSDistTest):
 689     """
 690     dnsdist is configured to expect a Proxy Protocol header on incoming queries but not from 127.0.0.1
 691     """
 692
 693     _config_template = """
 694     setProxyProtocolACL( { "192.0.2.1/32" } )
 695     newServer{address="127.0.0.1:%d"}
 696     """
 697     # NORMAL responder, does not expect a proxy protocol payload!
 698     _config_params = ['_testServerPort']
 699     _verboseMode = True
 700
 701     def testNoHeader(self):
 702         """
 703         Unexpected Proxy Protocol: no header
 704         """
 705         # no proxy protocol header and none is expected from this source, should be passed on
 706         name = 'no-header.unexpected-proxy-protocol.tests.powerdns.com.'
 707         query = dns.message.make_query(name, 'A', 'IN')
 708         response = dns.message.make_response(query)
 709         rrset = dns.rrset.from_text(name,
 710                                     60,
 711                                     dns.rdataclass.IN,
 712                                     dns.rdatatype.A,
 713                                     '127.0.0.1')
 714
 715         response.answer.append(rrset)
 716
 717         for method in ("sendUDPQuery", "sendTCPQuery"):
 718           sender = getattr(self, method)
 719           (receivedQuery, receivedResponse) = sender(query, response)
 720           receivedQuery.id = query.id
 721           self.assertEqual(query, receivedQuery)
 722           self.assertEqual(response, receivedResponse)
 723
 724     def testIncomingProxyDest(self):
 725         """
 726         Unexpected Proxy Protocol: should be dropped
 727         """
 728         name = 'with-proxy-payload.unexpected-protocol-incoming.tests.powerdns.com.'
 729         query = dns.message.make_query(name, 'A', 'IN')
 730
 731         # Make sure that the proxy payload does NOT turn into a legal qname
 732         destAddr = "ff:db8::ffff"
 733         destPort = 65535
 734         srcAddr = "ff:db8::ffff"
 735         srcPort = 65535
 736
 737         udpPayload = ProxyProtocol.getPayload(False, False, True, srcAddr, destAddr, srcPort, destPort, [ [ 2, b'foo'], [ 3, b'proxy'] ])
 738         (_, receivedResponse) = self.sendUDPQuery(udpPayload + query.to_wire(), response=None, useQueue=False, rawQuery=True)
 739         self.assertEqual(receivedResponse, None)
 740
 741         tcpPayload = ProxyProtocol.getPayload(False, True, True, srcAddr, destAddr, srcPort, destPort, [ [ 2, b'foo'], [ 3, b'proxy'] ])
 742         wire = query.to_wire()
 743
 744         receivedResponse = None
 745         try:
 746           conn = self.openTCPConnection(2.0)
 747           conn.send(tcpPayload)
 748           conn.send(struct.pack("!H", len(wire)))
 749           conn.send(wire)
 750           receivedResponse = self.recvTCPResponseOverConnection(conn)
 751         except socket.timeout:
 752           print('timeout')
 753         self.assertEqual(receivedResponse, None)