regression-tests.dnsdist/test_ProxyProtocol.py

   1 #!/usr/bin/env python
   2
   3 import copy
   4 import dns
   5 import selectors
   6 import socket
   7 import struct
   8 import sys
   9 import threading
  10 import time
  11
  12 from dnsdisttests import DNSDistTest, pickAvailablePort
  13 from proxyprotocol import ProxyProtocol
  14 from dnsdistdohtests import DNSDistDOHTest
  15
  16 # Python2/3 compatibility hacks
  17 try:
  18   from queue import Queue
  19 except ImportError:
  20   from Queue import Queue
  21
  22 def ProxyProtocolUDPResponder(port, fromQueue, toQueue):
  23     sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
  24     sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
  25     try:
  26         sock.bind(("127.0.0.1", port))
  27     except socket.error as e:
  28         print("Error binding in the Proxy Protocol UDP responder: %s" % str(e))
  29         sys.exit(1)
  30
  31     while True:
  32         data, addr = sock.recvfrom(4096)
  33
  34         proxy = ProxyProtocol()
  35         if len(data) < proxy.HEADER_SIZE:
  36             continue
  37
  38         if not proxy.parseHeader(data):
  39             continue
  40
  41         if proxy.local:
  42             # likely a healthcheck
  43             data = data[proxy.HEADER_SIZE:]
  44             request = dns.message.from_wire(data)
  45             response = dns.message.make_response(request)
  46             wire = response.to_wire()
  47             sock.settimeout(2.0)
  48             sock.sendto(wire, addr)
  49             sock.settimeout(None)
  50
  51             continue
  52
  53         payload = data[:(proxy.HEADER_SIZE + proxy.contentLen)]
  54         dnsData = data[(proxy.HEADER_SIZE + proxy.contentLen):]
  55         toQueue.put([payload, dnsData], True, 2.0)
  56         # computing the correct ID for the response
  57         request = dns.message.from_wire(dnsData)
  58         response = fromQueue.get(True, 2.0)
  59         response.id = request.id
  60
  61         sock.settimeout(2.0)
  62         sock.sendto(response.to_wire(), addr)
  63         sock.settimeout(None)
  64
  65     sock.close()
  66
  67 def ProxyProtocolTCPResponder(port, fromQueue, toQueue):
  68     # be aware that this responder will not accept a new connection
  69     # until the last one has been closed. This is done on purpose to
  70     # to check for connection reuse, making sure that a lot of connections
  71     # are not opened in parallel.
  72     sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  73     sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
  74     sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
  75     try:
  76         sock.bind(("127.0.0.1", port))
  77     except socket.error as e:
  78         print("Error binding in the TCP responder: %s" % str(e))
  79         sys.exit(1)
  80
  81     sock.listen(100)
  82     while True:
  83         (conn, _) = sock.accept()
  84         conn.settimeout(5.0)
  85         # try to read the entire Proxy Protocol header
  86         proxy = ProxyProtocol()
  87         header = conn.recv(proxy.HEADER_SIZE)
  88         if not header:
  89             conn.close()
  90             continue
  91
  92         if not proxy.parseHeader(header):
  93             conn.close()
  94             continue
  95
  96         proxyContent = conn.recv(proxy.contentLen)
  97         if not proxyContent:
  98             conn.close()
  99             continue
 100
 101         payload = header + proxyContent
 102         while True:
 103           try:
 104             data = conn.recv(2)
 105           except socket.timeout:
 106             data = None
 107
 108           if not data:
 109             conn.close()
 110             break
 111
 112           (datalen,) = struct.unpack("!H", data)
 113           data = conn.recv(datalen)
 114
 115           toQueue.put([payload, data], True, 2.0)
 116
 117           response = copy.deepcopy(fromQueue.get(True, 2.0))
 118           if not response:
 119             conn.close()
 120             break
 121
 122           # computing the correct ID for the response
 123           request = dns.message.from_wire(data)
 124           response.id = request.id
 125
 126           wire = response.to_wire()
 127           conn.send(struct.pack("!H", len(wire)))
 128           conn.send(wire)
 129
 130         conn.close()
 131
 132     sock.close()
 133
 134 toProxyQueue = Queue()
 135 fromProxyQueue = Queue()
 136 proxyResponderPort = pickAvailablePort()
 137
 138 udpResponder = threading.Thread(name='UDP Proxy Protocol Responder', target=ProxyProtocolUDPResponder, args=[proxyResponderPort, toProxyQueue, fromProxyQueue])
 139 udpResponder.daemon = True
 140 udpResponder.start()
 141 tcpResponder = threading.Thread(name='TCP Proxy Protocol Responder', target=ProxyProtocolTCPResponder, args=[proxyResponderPort, toProxyQueue, fromProxyQueue])
 142 tcpResponder.daemon = True
 143 tcpResponder.start()
 144
 145 backgroundThreads = {}
 146
 147 def MockTCPReverseProxyAddingProxyProtocol(listeningPort, forwardingPort):
 148     # this responder accepts TCP connections on the listening port,
 149     # and relay the raw content to a second TCP connection to the
 150     # forwarding port, after adding a Proxy Protocol v2 payload
 151     # containing the initial source IP and port, destination IP
 152     # and port.
 153     backgroundThreads[threading.get_native_id()] = True
 154     sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 155     sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
 156     sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
 157     try:
 158         sock.bind(("127.0.0.1", listeningPort))
 159     except socket.error as e:
 160         print("Error binding in the Mock TCP reverse proxy: %s" % str(e))
 161         sys.exit(1)
 162     sock.settimeout(0.5)
 163     sock.listen(100)
 164     while True:
 165         try:
 166             (incoming, _) = sock.accept()
 167         except socket.timeout:
 168             if backgroundThreads.get(threading.get_native_id(), False) == False:
 169                 del backgroundThreads[threading.get_native_id()]
 170                 break
 171             else:
 172               continue
 173
 174         incoming.settimeout(5.0)
 175         payload = ProxyProtocol.getPayload(False, True, False, '127.0.0.1', '127.0.0.1', incoming.getpeername()[1], listeningPort, [ [ 2, b'foo'], [ 3, b'proxy'] ])
 176
 177         outgoing = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 178         outgoing.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
 179         outgoing.settimeout(2.0)
 180         outgoing.connect(('127.0.0.1', forwardingPort))
 181
 182         outgoing.send(payload)
 183
 184         sel = selectors.DefaultSelector()
 185         def readFromClient(conn):
 186             data = conn.recv(512)
 187             if not data or len(data) == 0:
 188               return False
 189             outgoing.send(data)
 190             return True
 191
 192         def readFromBackend(conn):
 193             data = conn.recv(512)
 194             if not data or len(data) == 0:
 195               return False
 196             incoming.send(data)
 197             return True
 198
 199         sel.register(incoming, selectors.EVENT_READ, readFromClient)
 200         sel.register(outgoing, selectors.EVENT_READ, readFromBackend)
 201         done = False
 202         while not done:
 203           try:
 204             events = sel.select()
 205             for key, mask in events:
 206               if not (key.data)(key.fileobj):
 207                 done = True
 208                 break
 209           except socket.timeout:
 210             break
 211           except:
 212             break
 213
 214         incoming.close()
 215         outgoing.close()
 216
 217     sock.close()
 218
 219 class ProxyProtocolTest(DNSDistTest):
 220     _proxyResponderPort = proxyResponderPort
 221     _config_params = ['_proxyResponderPort']
 222
 223 class TestProxyProtocol(ProxyProtocolTest):
 224     """
 225     dnsdist is configured to prepend a Proxy Protocol header to the query
 226     """
 227
 228     _config_template = """
 229     newServer{address="127.0.0.1:%d", useProxyProtocol=true}
 230
 231     function addValues(dq)
 232       local values = { [0]="foo", [42]="bar" }
 233       dq:setProxyProtocolValues(values)
 234       return DNSAction.None
 235     end
 236
 237     addAction("values-lua.proxy.tests.powerdns.com.", LuaAction(addValues))
 238     addAction("values-action.proxy.tests.powerdns.com.", SetProxyProtocolValuesAction({ ["1"]="dnsdist", ["255"]="proxy-protocol"}))
 239     """
 240     _config_params = ['_proxyResponderPort']
 241     _verboseMode = True
 242
 243     def testProxyUDP(self):
 244         """
 245         Proxy Protocol: no value (UDP)
 246         """
 247         name = 'simple-udp.proxy.tests.powerdns.com.'
 248         query = dns.message.make_query(name, 'A', 'IN')
 249         response = dns.message.make_response(query)
 250
 251         toProxyQueue.put(response, True, 2.0)
 252
 253         data = query.to_wire()
 254         self._sock.send(data)
 255         receivedResponse = None
 256         try:
 257             self._sock.settimeout(2.0)
 258             data = self._sock.recv(4096)
 259         except socket.timeout:
 260             print('timeout')
 261             data = None
 262         if data:
 263             receivedResponse = dns.message.from_wire(data)
 264
 265         (receivedProxyPayload, receivedDNSData) = fromProxyQueue.get(True, 2.0)
 266         self.assertTrue(receivedProxyPayload)
 267         self.assertTrue(receivedDNSData)
 268         self.assertTrue(receivedResponse)
 269
 270         receivedQuery = dns.message.from_wire(receivedDNSData)
 271         receivedQuery.id = query.id
 272         receivedResponse.id = response.id
 273         self.assertEqual(receivedQuery, query)
 274         self.assertEqual(receivedResponse, response)
 275         self.checkMessageProxyProtocol(receivedProxyPayload, '127.0.0.1', '127.0.0.1', False)
 276
 277     def testProxyTCP(self):
 278       """
 279         Proxy Protocol: no value (TCP)
 280       """
 281       name = 'simple-tcp.proxy.tests.powerdns.com.'
 282       query = dns.message.make_query(name, 'A', 'IN')
 283       response = dns.message.make_response(query)
 284
 285       toProxyQueue.put(response, True, 2.0)
 286
 287       conn = self.openTCPConnection(2.0)
 288       data = query.to_wire()
 289       self.sendTCPQueryOverConnection(conn, data, rawQuery=True)
 290       receivedResponse = None
 291       try:
 292         receivedResponse = self.recvTCPResponseOverConnection(conn)
 293       except socket.timeout:
 294             print('timeout')
 295
 296       (receivedProxyPayload, receivedDNSData) = fromProxyQueue.get(True, 2.0)
 297       self.assertTrue(receivedProxyPayload)
 298       self.assertTrue(receivedDNSData)
 299       self.assertTrue(receivedResponse)
 300
 301       receivedQuery = dns.message.from_wire(receivedDNSData)
 302       receivedQuery.id = query.id
 303       receivedResponse.id = response.id
 304       self.assertEqual(receivedQuery, query)
 305       self.assertEqual(receivedResponse, response)
 306       self.checkMessageProxyProtocol(receivedProxyPayload, '127.0.0.1', '127.0.0.1', True)
 307
 308     def testProxyUDPWithValuesFromLua(self):
 309         """
 310         Proxy Protocol: values from Lua (UDP)
 311         """
 312         name = 'values-lua.proxy.tests.powerdns.com.'
 313         query = dns.message.make_query(name, 'A', 'IN')
 314         response = dns.message.make_response(query)
 315
 316         toProxyQueue.put(response, True, 2.0)
 317
 318         data = query.to_wire()
 319         self._sock.send(data)
 320         receivedResponse = None
 321         try:
 322             self._sock.settimeout(2.0)
 323             data = self._sock.recv(4096)
 324         except socket.timeout:
 325             print('timeout')
 326             data = None
 327         if data:
 328             receivedResponse = dns.message.from_wire(data)
 329
 330         (receivedProxyPayload, receivedDNSData) = fromProxyQueue.get(True, 2.0)
 331         self.assertTrue(receivedProxyPayload)
 332         self.assertTrue(receivedDNSData)
 333         self.assertTrue(receivedResponse)
 334
 335         receivedQuery = dns.message.from_wire(receivedDNSData)
 336         receivedQuery.id = query.id
 337         receivedResponse.id = response.id
 338         self.assertEqual(receivedQuery, query)
 339         self.assertEqual(receivedResponse, response)
 340         self.checkMessageProxyProtocol(receivedProxyPayload, '127.0.0.1', '127.0.0.1', False, [ [0, b'foo'] , [ 42, b'bar'] ])
 341
 342     def testProxyTCPWithValuesFromLua(self):
 343       """
 344         Proxy Protocol: values from Lua (TCP)
 345       """
 346       name = 'values-lua.proxy.tests.powerdns.com.'
 347       query = dns.message.make_query(name, 'A', 'IN')
 348       response = dns.message.make_response(query)
 349
 350       toProxyQueue.put(response, True, 2.0)
 351
 352       conn = self.openTCPConnection(2.0)
 353       data = query.to_wire()
 354       self.sendTCPQueryOverConnection(conn, data, rawQuery=True)
 355       receivedResponse = None
 356       try:
 357         receivedResponse = self.recvTCPResponseOverConnection(conn)
 358       except socket.timeout:
 359             print('timeout')
 360
 361       (receivedProxyPayload, receivedDNSData) = fromProxyQueue.get(True, 2.0)
 362       self.assertTrue(receivedProxyPayload)
 363       self.assertTrue(receivedDNSData)
 364       self.assertTrue(receivedResponse)
 365
 366       receivedQuery = dns.message.from_wire(receivedDNSData)
 367       receivedQuery.id = query.id
 368       receivedResponse.id = response.id
 369       self.assertEqual(receivedQuery, query)
 370       self.assertEqual(receivedResponse, response)
 371       self.checkMessageProxyProtocol(receivedProxyPayload, '127.0.0.1', '127.0.0.1', True, [ [0, b'foo'] , [ 42, b'bar'] ])
 372
 373     def testProxyUDPWithValuesFromAction(self):
 374         """
 375         Proxy Protocol: values from Action (UDP)
 376         """
 377         name = 'values-action.proxy.tests.powerdns.com.'
 378         query = dns.message.make_query(name, 'A', 'IN')
 379         response = dns.message.make_response(query)
 380
 381         toProxyQueue.put(response, True, 2.0)
 382
 383         data = query.to_wire()
 384         self._sock.send(data)
 385         receivedResponse = None
 386         try:
 387             self._sock.settimeout(2.0)
 388             data = self._sock.recv(4096)
 389         except socket.timeout:
 390             print('timeout')
 391             data = None
 392         if data:
 393             receivedResponse = dns.message.from_wire(data)
 394
 395         (receivedProxyPayload, receivedDNSData) = fromProxyQueue.get(True, 2.0)
 396         self.assertTrue(receivedProxyPayload)
 397         self.assertTrue(receivedDNSData)
 398         self.assertTrue(receivedResponse)
 399
 400         receivedQuery = dns.message.from_wire(receivedDNSData)
 401         receivedQuery.id = query.id
 402         receivedResponse.id = response.id
 403         self.assertEqual(receivedQuery, query)
 404         self.assertEqual(receivedResponse, response)
 405         self.checkMessageProxyProtocol(receivedProxyPayload, '127.0.0.1', '127.0.0.1', False, [ [1, b'dnsdist'] , [ 255, b'proxy-protocol'] ])
 406
 407     def testProxyTCPWithValuesFromAction(self):
 408       """
 409         Proxy Protocol: values from Action (TCP)
 410       """
 411       name = 'values-action.proxy.tests.powerdns.com.'
 412       query = dns.message.make_query(name, 'A', 'IN')
 413       response = dns.message.make_response(query)
 414
 415       toProxyQueue.put(response, True, 2.0)
 416
 417       conn = self.openTCPConnection(2.0)
 418       data = query.to_wire()
 419       self.sendTCPQueryOverConnection(conn, data, rawQuery=True)
 420       receivedResponse = None
 421       try:
 422         receivedResponse = self.recvTCPResponseOverConnection(conn)
 423       except socket.timeout:
 424             print('timeout')
 425
 426       (receivedProxyPayload, receivedDNSData) = fromProxyQueue.get(True, 2.0)
 427       self.assertTrue(receivedProxyPayload)
 428       self.assertTrue(receivedDNSData)
 429       self.assertTrue(receivedResponse)
 430
 431       receivedQuery = dns.message.from_wire(receivedDNSData)
 432       receivedQuery.id = query.id
 433       receivedResponse.id = response.id
 434       self.assertEqual(receivedQuery, query)
 435       self.assertEqual(receivedResponse, response)
 436       self.checkMessageProxyProtocol(receivedProxyPayload, '127.0.0.1', '127.0.0.1', True, [ [1, b'dnsdist'] , [ 255, b'proxy-protocol'] ])
 437
 438     def testProxyTCPSeveralQueriesOnSameConnection(self):
 439       """
 440         Proxy Protocol: Several queries on the same TCP connection
 441       """
 442       name = 'several-queries-same-conn.proxy.tests.powerdns.com.'
 443       query = dns.message.make_query(name, 'A', 'IN')
 444       response = dns.message.make_response(query)
 445
 446       conn = self.openTCPConnection(2.0)
 447       data = query.to_wire()
 448
 449       for idx in range(10):
 450         toProxyQueue.put(response, True, 2.0)
 451         self.sendTCPQueryOverConnection(conn, data, rawQuery=True)
 452         receivedResponse = None
 453         try:
 454           receivedResponse = self.recvTCPResponseOverConnection(conn)
 455         except socket.timeout:
 456           print('timeout')
 457
 458         (receivedProxyPayload, receivedDNSData) = fromProxyQueue.get(True, 2.0)
 459         self.assertTrue(receivedProxyPayload)
 460         self.assertTrue(receivedDNSData)
 461         self.assertTrue(receivedResponse)
 462
 463         receivedQuery = dns.message.from_wire(receivedDNSData)
 464         receivedQuery.id = query.id
 465         receivedResponse.id = response.id
 466         self.assertEqual(receivedQuery, query)
 467         self.assertEqual(receivedResponse, response)
 468         self.checkMessageProxyProtocol(receivedProxyPayload, '127.0.0.1', '127.0.0.1', True, [])
 469
 470 class TestProxyProtocolIncoming(ProxyProtocolTest):
 471     """
 472     dnsdist is configured to prepend a Proxy Protocol header to the query and expect one on incoming queries
 473     """
 474
 475     _config_template = """
 476     addDOHLocal("127.0.0.1:%s", "%s", "%s", {"/"}, {library='nghttp2', proxyProtocolOutsideTLS=true})
 477     setProxyProtocolACL( { "127.0.0.1/32" } )
 478     newServer{address="127.0.0.1:%d", useProxyProtocol=true}
 479
 480     function addValues(dq)
 481       dq:addProxyProtocolValue(0, 'foo')
 482       dq:addProxyProtocolValue(42, 'bar')
 483       return DNSAction.None
 484     end
 485
 486     -- refuse queries with no TLV value type 2
 487     addAction(NotRule(ProxyProtocolValueRule(2)), RCodeAction(DNSRCode.REFUSED))
 488     -- or with a TLV value type 3 different from "proxy"
 489     addAction(NotRule(ProxyProtocolValueRule(3, "proxy")), RCodeAction(DNSRCode.REFUSED))
 490
 491     function answerBasedOnForwardedDest(dq)
 492       local port = dq.localaddr:getPort()
 493       local dest = dq.localaddr:toString()
 494       return DNSAction.Spoof, "address-was-"..dest.."-port-was-"..port..".proxy-protocol-incoming.tests.powerdns.com."
 495     end
 496     addAction("get-forwarded-dest.proxy-protocol-incoming.tests.powerdns.com.", LuaAction(answerBasedOnForwardedDest))
 497
 498     function answerBasedOnForwardedSrc(dq)
 499       local port = dq.remoteaddr:getPort()
 500       local src = dq.remoteaddr:toString()
 501       return DNSAction.Spoof, "address-was-"..src.."-port-was-"..port..".proxy-protocol-incoming.tests.powerdns.com."
 502     end
 503     addAction("get-forwarded-src.proxy-protocol-incoming.tests.powerdns.com.", LuaAction(answerBasedOnForwardedSrc))
 504
 505     -- add these values for all queries
 506     addAction("proxy-protocol-incoming.tests.powerdns.com.", LuaAction(addValues))
 507     addAction("proxy-protocol-incoming.tests.powerdns.com.", SetAdditionalProxyProtocolValueAction(1, "dnsdist"))
 508     addAction("proxy-protocol-incoming.tests.powerdns.com.", SetAdditionalProxyProtocolValueAction(255, "proxy-protocol"))
 509
 510     -- override all existing values
 511     addAction("override.proxy-protocol-incoming.tests.powerdns.com.", SetProxyProtocolValuesAction({["50"]="overridden"}))
 512     """
 513     _serverKey = 'server.key'
 514     _serverCert = 'server.chain'
 515     _serverName = 'tls.tests.dnsdist.org'
 516     _caCert = 'ca.pem'
 517     _dohServerPort = 8443
 518     _dohBaseURL = ("https://%s:%d/" % (_serverName, _dohServerPort))
 519     _config_params = ['_dohServerPort', '_serverCert', '_serverKey', '_proxyResponderPort']
 520
 521     def testNoHeader(self):
 522         """
 523         Incoming Proxy Protocol: no header
 524         """
 525         # no proxy protocol header while one is expected, should be dropped
 526         name = 'no-header.incoming-proxy-protocol.tests.powerdns.com.'
 527         query = dns.message.make_query(name, 'A', 'IN')
 528
 529         for method in ("sendUDPQuery", "sendTCPQuery", "sendDOHQueryWrapper"):
 530           sender = getattr(self, method)
 531           try:
 532             (_, receivedResponse) = sender(query, response=None)
 533           except Exception:
 534             receivedResponse = None
 535           self.assertEqual(receivedResponse, None)
 536
 537     def testIncomingProxyDest(self):
 538         """
 539         Incoming Proxy Protocol: values from Lua
 540         """
 541         name = 'get-forwarded-dest.proxy-protocol-incoming.tests.powerdns.com.'
 542         query = dns.message.make_query(name, 'A', 'IN')
 543         # dnsdist set RA = RD for spoofed responses
 544         query.flags &= ~dns.flags.RD
 545
 546         destAddr = "2001:db8::9"
 547         destPort = 9999
 548         srcAddr = "2001:db8::8"
 549         srcPort = 8888
 550         response = dns.message.make_response(query)
 551         rrset = dns.rrset.from_text(name,
 552                                     60,
 553                                     dns.rdataclass.IN,
 554                                     dns.rdatatype.CNAME,
 555                                     "address-was-{}-port-was-{}.proxy-protocol-incoming.tests.powerdns.com.".format(destAddr, destPort, self._dnsDistPort))
 556         response.answer.append(rrset)
 557
 558         udpPayload = ProxyProtocol.getPayload(False, False, True, srcAddr, destAddr, srcPort, destPort, [ [ 2, b'foo'], [ 3, b'proxy'] ])
 559         (_, receivedResponse) = self.sendUDPQuery(udpPayload + query.to_wire(), response=None, useQueue=False, rawQuery=True)
 560         self.assertEqual(receivedResponse, response)
 561
 562         tcpPayload = ProxyProtocol.getPayload(False, True, True, srcAddr, destAddr, srcPort, destPort, [ [ 2, b'foo'], [ 3, b'proxy'] ])
 563         wire = query.to_wire()
 564
 565         receivedResponse = None
 566         try:
 567           conn = self.openTCPConnection(2.0)
 568           conn.send(tcpPayload)
 569           conn.send(struct.pack("!H", len(wire)))
 570           conn.send(wire)
 571           receivedResponse = self.recvTCPResponseOverConnection(conn)
 572         except socket.timeout:
 573           print('timeout')
 574         self.assertEqual(receivedResponse, response)
 575
 576     def testProxyUDPWithValuesFromLua(self):
 577         """
 578         Incoming Proxy Protocol: values from Lua (UDP)
 579         """
 580         name = 'values-lua.proxy-protocol-incoming.tests.powerdns.com.'
 581         query = dns.message.make_query(name, 'A', 'IN')
 582         response = dns.message.make_response(query)
 583
 584         destAddr = "2001:db8::9"
 585         destPort = 9999
 586         srcAddr = "2001:db8::8"
 587         srcPort = 8888
 588         response = dns.message.make_response(query)
 589
 590         udpPayload = ProxyProtocol.getPayload(False, False, True, srcAddr, destAddr, srcPort, destPort, [ [ 2, b'foo'], [ 3, b'proxy'] ])
 591         toProxyQueue.put(response, True, 2.0)
 592         (_, receivedResponse) = self.sendUDPQuery(udpPayload + query.to_wire(), response=None, useQueue=False, rawQuery=True)
 593
 594         (receivedProxyPayload, receivedDNSData) = fromProxyQueue.get(True, 2.0)
 595         self.assertTrue(receivedProxyPayload)
 596         self.assertTrue(receivedDNSData)
 597         self.assertTrue(receivedResponse)
 598
 599         receivedQuery = dns.message.from_wire(receivedDNSData)
 600         receivedQuery.id = query.id
 601         receivedResponse.id = response.id
 602         self.assertEqual(receivedQuery, query)
 603         self.assertEqual(receivedResponse, response)
 604         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)
 605
 606     def testProxyTCPWithValuesFromLua(self):
 607         """
 608         Incoming Proxy Protocol: values from Lua (TCP)
 609         """
 610         name = 'values-lua.proxy-protocol-incoming.tests.powerdns.com.'
 611         query = dns.message.make_query(name, 'A', 'IN')
 612         response = dns.message.make_response(query)
 613
 614         destAddr = "2001:db8::9"
 615         destPort = 9999
 616         srcAddr = "2001:db8::8"
 617         srcPort = 8888
 618         response = dns.message.make_response(query)
 619
 620         tcpPayload = ProxyProtocol.getPayload(False, True, True, srcAddr, destAddr, srcPort, destPort, [ [ 2, b'foo'], [ 3, b'proxy'] ])
 621
 622         toProxyQueue.put(response, True, 2.0)
 623
 624         wire = query.to_wire()
 625
 626         receivedResponse = None
 627         try:
 628           conn = self.openTCPConnection(2.0)
 629           conn.send(tcpPayload)
 630           conn.send(struct.pack("!H", len(wire)))
 631           conn.send(wire)
 632           receivedResponse = self.recvTCPResponseOverConnection(conn)
 633         except socket.timeout:
 634           print('timeout')
 635         self.assertEqual(receivedResponse, response)
 636
 637         (receivedProxyPayload, receivedDNSData) = fromProxyQueue.get(True, 2.0)
 638         self.assertTrue(receivedProxyPayload)
 639         self.assertTrue(receivedDNSData)
 640         self.assertTrue(receivedResponse)
 641
 642         receivedQuery = dns.message.from_wire(receivedDNSData)
 643         receivedQuery.id = query.id
 644         self.assertEqual(receivedQuery, query)
 645         self.assertEqual(receivedResponse, response)
 646         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)
 647
 648     def testProxyUDPWithValueOverride(self):
 649         """
 650         Incoming Proxy Protocol: override existing value (UDP)
 651         """
 652         name = 'override.proxy-protocol-incoming.tests.powerdns.com.'
 653         query = dns.message.make_query(name, 'A', 'IN')
 654         response = dns.message.make_response(query)
 655
 656         destAddr = "2001:db8::9"
 657         destPort = 9999
 658         srcAddr = "2001:db8::8"
 659         srcPort = 8888
 660         response = dns.message.make_response(query)
 661
 662         udpPayload = ProxyProtocol.getPayload(False, False, True, srcAddr, destAddr, srcPort, destPort, [ [2, b'foo'], [3, b'proxy'], [ 50, b'initial-value']])
 663         toProxyQueue.put(response, True, 2.0)
 664         (_, receivedResponse) = self.sendUDPQuery(udpPayload + query.to_wire(), response=None, useQueue=False, rawQuery=True)
 665
 666         (receivedProxyPayload, receivedDNSData) = fromProxyQueue.get(True, 2.0)
 667         self.assertTrue(receivedProxyPayload)
 668         self.assertTrue(receivedDNSData)
 669         self.assertTrue(receivedResponse)
 670
 671         receivedQuery = dns.message.from_wire(receivedDNSData)
 672         receivedQuery.id = query.id
 673         receivedResponse.id = response.id
 674         self.assertEqual(receivedQuery, query)
 675         self.assertEqual(receivedResponse, response)
 676         self.checkMessageProxyProtocol(receivedProxyPayload, srcAddr, destAddr, False, [ [50, b'overridden'] ], True, srcPort, destPort)
 677
 678     def testProxyTCPSeveralQueriesOverConnection(self):
 679         """
 680         Incoming Proxy Protocol: Several queries over the same connection (TCP)
 681         """
 682         name = 'several-queries.proxy-protocol-incoming.tests.powerdns.com.'
 683         query = dns.message.make_query(name, 'A', 'IN')
 684         response = dns.message.make_response(query)
 685
 686         destAddr = "2001:db8::9"
 687         destPort = 9999
 688         srcAddr = "2001:db8::8"
 689         srcPort = 8888
 690
 691         tcpPayload = ProxyProtocol.getPayload(False, True, True, srcAddr, destAddr, srcPort, destPort, [ [ 2, b'foo'], [ 3, b'proxy'] ])
 692
 693         toProxyQueue.put(response, True, 2.0)
 694
 695         wire = query.to_wire()
 696
 697         receivedResponse = None
 698         conn = self.openTCPConnection(2.0)
 699         try:
 700           conn.send(tcpPayload)
 701           conn.send(struct.pack("!H", len(wire)))
 702           conn.send(wire)
 703           receivedResponse = self.recvTCPResponseOverConnection(conn)
 704         except socket.timeout:
 705           print('timeout')
 706         self.assertEqual(receivedResponse, response)
 707
 708         (receivedProxyPayload, receivedDNSData) = fromProxyQueue.get(True, 2.0)
 709         self.assertTrue(receivedProxyPayload)
 710         self.assertTrue(receivedDNSData)
 711         self.assertTrue(receivedResponse)
 712
 713         receivedQuery = dns.message.from_wire(receivedDNSData)
 714         receivedQuery.id = query.id
 715         receivedResponse.id = response.id
 716         self.assertEqual(receivedQuery, query)
 717         self.assertEqual(receivedResponse, response)
 718         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)
 719
 720         for idx in range(5):
 721           receivedResponse = None
 722           toProxyQueue.put(response, True, 2.0)
 723           try:
 724             conn.send(struct.pack("!H", len(wire)))
 725             conn.send(wire)
 726             receivedResponse = self.recvTCPResponseOverConnection(conn)
 727           except socket.timeout:
 728             print('timeout')
 729
 730           self.assertEqual(receivedResponse, response)
 731
 732           (receivedProxyPayload, receivedDNSData) = fromProxyQueue.get(True, 2.0)
 733           self.assertTrue(receivedProxyPayload)
 734           self.assertTrue(receivedDNSData)
 735           self.assertTrue(receivedResponse)
 736
 737           receivedQuery = dns.message.from_wire(receivedDNSData)
 738           receivedQuery.id = query.id
 739           self.assertEqual(receivedQuery, query)
 740           self.assertEqual(receivedResponse, response)
 741           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)
 742
 743     def testProxyDoHSeveralQueriesOverConnection(self):
 744         """
 745         Incoming Proxy Protocol: Several queries over the same connection (DoH)
 746         """
 747         name = 'several-queries.proxy-protocol-incoming.tests.powerdns.com.'
 748         query = dns.message.make_query(name, 'A', 'IN')
 749         response = dns.message.make_response(query)
 750
 751         toProxyQueue.put(response, True, 2.0)
 752
 753         wire = query.to_wire()
 754
 755         reverseProxyPort = 13053
 756         reverseProxy = threading.Thread(name='Mock Proxy Protocol Reverse Proxy', target=MockTCPReverseProxyAddingProxyProtocol, args=[reverseProxyPort, self._dohServerPort])
 757         reverseProxy.start()
 758
 759         receivedResponse = None
 760         conn = self.openDOHConnection(reverseProxyPort, self._caCert, timeout=2.0)
 761
 762         reverseProxyBaseURL = ("https://%s:%d/" % (self._serverName, reverseProxyPort))
 763         (receivedQuery, receivedResponse) = self.sendDOHQuery(reverseProxyPort, self._serverName, reverseProxyBaseURL, query, response=response, caFile=self._caCert, useQueue=True, conn=conn)
 764         (receivedProxyPayload, receivedDNSData) = fromProxyQueue.get(True, 2.0)
 765         self.assertTrue(receivedProxyPayload)
 766         self.assertTrue(receivedDNSData)
 767         self.assertTrue(receivedResponse)
 768
 769         receivedQuery = dns.message.from_wire(receivedDNSData)
 770         receivedQuery.id = query.id
 771         receivedResponse.id = response.id
 772         self.assertEqual(receivedQuery, query)
 773         self.assertEqual(receivedResponse, response)
 774         self.checkMessageProxyProtocol(receivedProxyPayload, '127.0.0.1', '127.0.0.1', True, [ [0, b'foo'], [1, b'dnsdist'], [ 2, b'foo'], [3, b'proxy'], [ 42, b'bar'], [255, b'proxy-protocol'] ], v6=False, sourcePort=None, destinationPort=reverseProxyPort)
 775
 776         for idx in range(5):
 777           receivedResponse = None
 778           toProxyQueue.put(response, True, 2.0)
 779           (receivedQuery, receivedResponse) = self.sendDOHQuery(reverseProxyPort, self._serverName, reverseProxyBaseURL, query, response=response, caFile=self._caCert, useQueue=True, conn=conn)
 780           (receivedProxyPayload, receivedDNSData) = fromProxyQueue.get(True, 2.0)
 781           self.assertTrue(receivedProxyPayload)
 782           self.assertTrue(receivedDNSData)
 783           self.assertTrue(receivedResponse)
 784
 785           receivedQuery = dns.message.from_wire(receivedDNSData)
 786           receivedQuery.id = query.id
 787           receivedResponse.id = response.id
 788           self.assertEqual(receivedQuery, query)
 789           print(receivedResponse)
 790           print(response)
 791           self.assertEqual(receivedResponse, response)
 792           self.checkMessageProxyProtocol(receivedProxyPayload, '127.0.0.1', '127.0.0.1', True, [ [0, b'foo'], [1, b'dnsdist'], [ 2, b'foo'], [3, b'proxy'], [ 42, b'bar'], [255, b'proxy-protocol'] ], v6=False, sourcePort=None, destinationPort=reverseProxyPort)
 793
 794     @classmethod
 795     def tearDownClass(cls):
 796         cls._sock.close()
 797         for backgroundThread in cls._backgroundThreads:
 798             cls._backgroundThreads[backgroundThread] = False
 799         for backgroundThread in backgroundThreads:
 800             backgroundThreads[backgroundThread] = False
 801         cls.killProcess(cls._dnsdist)
 802
 803 class TestProxyProtocolNotExpected(DNSDistTest):
 804     """
 805     dnsdist is configured to expect a Proxy Protocol header on incoming queries but not from 127.0.0.1
 806     """
 807
 808     _config_template = """
 809     setProxyProtocolACL( { "192.0.2.1/32" } )
 810     newServer{address="127.0.0.1:%d"}
 811     """
 812     # NORMAL responder, does not expect a proxy protocol payload!
 813     _config_params = ['_testServerPort']
 814     _verboseMode = True
 815
 816     def testNoHeader(self):
 817         """
 818         Unexpected Proxy Protocol: no header
 819         """
 820         # no proxy protocol header and none is expected from this source, should be passed on
 821         name = 'no-header.unexpected-proxy-protocol.tests.powerdns.com.'
 822         query = dns.message.make_query(name, 'A', 'IN')
 823         response = dns.message.make_response(query)
 824         rrset = dns.rrset.from_text(name,
 825                                     60,
 826                                     dns.rdataclass.IN,
 827                                     dns.rdatatype.A,
 828                                     '127.0.0.1')
 829
 830         response.answer.append(rrset)
 831
 832         for method in ("sendUDPQuery", "sendTCPQuery"):
 833           sender = getattr(self, method)
 834           (receivedQuery, receivedResponse) = sender(query, response)
 835           receivedQuery.id = query.id
 836           self.assertEqual(query, receivedQuery)
 837           self.assertEqual(response, receivedResponse)
 838
 839     def testIncomingProxyDest(self):
 840         """
 841         Unexpected Proxy Protocol: should be dropped
 842         """
 843         name = 'with-proxy-payload.unexpected-protocol-incoming.tests.powerdns.com.'
 844         query = dns.message.make_query(name, 'A', 'IN')
 845
 846         # Make sure that the proxy payload does NOT turn into a legal qname
 847         destAddr = "ff:db8::ffff"
 848         destPort = 65535
 849         srcAddr = "ff:db8::ffff"
 850         srcPort = 65535
 851
 852         udpPayload = ProxyProtocol.getPayload(False, False, True, srcAddr, destAddr, srcPort, destPort, [ [ 2, b'foo'], [ 3, b'proxy'] ])
 853         (_, receivedResponse) = self.sendUDPQuery(udpPayload + query.to_wire(), response=None, useQueue=False, rawQuery=True)
 854         self.assertEqual(receivedResponse, None)
 855
 856         tcpPayload = ProxyProtocol.getPayload(False, True, True, srcAddr, destAddr, srcPort, destPort, [ [ 2, b'foo'], [ 3, b'proxy'] ])
 857         wire = query.to_wire()
 858
 859         receivedResponse = None
 860         try:
 861           conn = self.openTCPConnection(2.0)
 862           conn.send(tcpPayload)
 863           conn.send(struct.pack("!H", len(wire)))
 864           conn.send(wire)
 865           receivedResponse = self.recvTCPResponseOverConnection(conn)
 866         except socket.timeout:
 867           print('timeout')
 868         self.assertEqual(receivedResponse, None)
 869
 870 class TestDOHWithOutgoingProxyProtocol(DNSDistDOHTest):
 871
 872     _serverKey = 'server.key'
 873     _serverCert = 'server.chain'
 874     _serverName = 'tls.tests.dnsdist.org'
 875     _caCert = 'ca.pem'
 876     _dohWithNGHTTP2ServerPort = pickAvailablePort()
 877     _dohWithNGHTTP2BaseURL = ("https://%s:%d/dns-query" % (_serverName, _dohWithNGHTTP2ServerPort))
 878     _dohWithH2OServerPort = pickAvailablePort()
 879     _dohWithH2OBaseURL = ("https://%s:%d/dns-query" % (_serverName, _dohWithH2OServerPort))
 880     _proxyResponderPort = proxyResponderPort
 881     _config_template = """
 882     newServer{address="127.0.0.1:%s", useProxyProtocol=true}
 883     addDOHLocal("127.0.0.1:%d", "%s", "%s", { '/dns-query' }, { trustForwardedForHeader=true, library='nghttp2' })
 884     addDOHLocal("127.0.0.1:%d", "%s", "%s", { '/dns-query' }, { trustForwardedForHeader=true, library='h2o' })
 885     setACL( { "::1/128", "127.0.0.0/8" } )
 886     """
 887     _config_params = ['_proxyResponderPort', '_dohWithNGHTTP2ServerPort', '_serverCert', '_serverKey', '_dohWithH2OServerPort', '_serverCert', '_serverKey']
 888     _verboseMode = True
 889
 890     def testTruncation(self):
 891         """
 892         DOH: Truncation over UDP
 893         """
 894         # the query is first forwarded over UDP, leading to a TC=1 answer from the
 895         # backend, then over TCP
 896         name = 'truncated-udp.doh.proxy-protocol.tests.powerdns.com.'
 897         query = dns.message.make_query(name, 'A', 'IN')
 898         query.id = 42
 899         expectedQuery = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096)
 900         expectedQuery.id = 42
 901         response = dns.message.make_response(query)
 902         rrset = dns.rrset.from_text(name,
 903                                     3600,
 904                                     dns.rdataclass.IN,
 905                                     dns.rdatatype.A,
 906                                     '127.0.0.1')
 907         response.answer.append(rrset)
 908
 909         for (port,url) in [(self._dohWithNGHTTP2ServerPort, self._dohWithNGHTTP2BaseURL), (self._dohWithH2OServerPort, self._dohWithH2OBaseURL)]:
 910           # first response is a TC=1
 911           tcResponse = dns.message.make_response(query)
 912           tcResponse.flags |= dns.flags.TC
 913           toProxyQueue.put(tcResponse, True, 2.0)
 914
 915           ((receivedProxyPayload, receivedDNSData), receivedResponse) = self.sendDOHQuery(port, self._serverName, url, query, caFile=self._caCert, response=response, fromQueue=fromProxyQueue, toQueue=toProxyQueue)
 916           # first query, received by the responder over UDP
 917           self.assertTrue(receivedProxyPayload)
 918           self.assertTrue(receivedDNSData)
 919           receivedQuery = dns.message.from_wire(receivedDNSData)
 920           self.assertTrue(receivedQuery)
 921           receivedQuery.id = expectedQuery.id
 922           self.assertEqual(expectedQuery, receivedQuery)
 923           self.checkQueryEDNSWithoutECS(expectedQuery, receivedQuery)
 924           self.checkMessageProxyProtocol(receivedProxyPayload, '127.0.0.1', '127.0.0.1', True, destinationPort=port)
 925
 926           # check the response
 927           self.assertTrue(receivedResponse)
 928           self.assertEqual(response, receivedResponse)
 929
 930           # check the second query, received by the responder over TCP
 931           (receivedProxyPayload, receivedDNSData) = fromProxyQueue.get(True, 2.0)
 932           self.assertTrue(receivedDNSData)
 933           receivedQuery = dns.message.from_wire(receivedDNSData)
 934           self.assertTrue(receivedQuery)
 935           receivedQuery.id = expectedQuery.id
 936           self.assertEqual(expectedQuery, receivedQuery)
 937           self.checkQueryEDNSWithoutECS(expectedQuery, receivedQuery)
 938           self.checkMessageProxyProtocol(receivedProxyPayload, '127.0.0.1', '127.0.0.1', True, destinationPort=port)
 939
 940           # make sure we consumed everything
 941           self.assertTrue(toProxyQueue.empty())
 942           self.assertTrue(fromProxyQueue.empty())
 943
 944     def testAddressFamilyMismatch(self):
 945         """
 946         DOH with IPv6 X-Forwarded-For to an IPv4 endpoint
 947         """
 948         name = 'x-forwarded-for-af-mismatch.doh.outgoing-proxy-protocol.tests.powerdns.com.'
 949         query = dns.message.make_query(name, 'A', 'IN', use_edns=False)
 950         query.id = 0
 951         expectedQuery = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096)
 952         expectedQuery.id = 0
 953         response = dns.message.make_response(query)
 954         rrset = dns.rrset.from_text(name,
 955                                     3600,
 956                                     dns.rdataclass.IN,
 957                                     dns.rdatatype.A,
 958                                     '127.0.0.1')
 959         response.answer.append(rrset)
 960
 961         for (port,url) in [(self._dohWithNGHTTP2ServerPort, self._dohWithNGHTTP2BaseURL), (self._dohWithH2OServerPort, self._dohWithH2OBaseURL)]:
 962           # the query should be dropped
 963           (receivedQuery, receivedResponse) = self.sendDOHQuery(port, self._serverName, url, query, caFile=self._caCert, customHeaders=['x-forwarded-for: [::1]:8080'], useQueue=False)
 964           self.assertFalse(receivedQuery)
 965           self.assertFalse(receivedResponse)
 966
 967           # make sure the timeout is detected, if any
 968           time.sleep(4)
 969
 970           # this one should not
 971           ((receivedProxyPayload, receivedDNSData), receivedResponse) = self.sendDOHQuery(port, self._serverName, url, query, caFile=self._caCert, customHeaders=['x-forwarded-for: 127.0.0.42:8080'], response=response, fromQueue=fromProxyQueue, toQueue=toProxyQueue)
 972           self.assertTrue(receivedProxyPayload)
 973           self.assertTrue(receivedDNSData)
 974           receivedQuery = dns.message.from_wire(receivedDNSData)
 975           self.assertTrue(receivedQuery)
 976           receivedQuery.id = expectedQuery.id
 977           self.assertEqual(expectedQuery, receivedQuery)
 978           self.checkQueryEDNSWithoutECS(expectedQuery, receivedQuery)
 979           self.checkMessageProxyProtocol(receivedProxyPayload, '127.0.0.42', '127.0.0.1', True, destinationPort=port)
 980           # check the response
 981           self.assertTrue(receivedResponse)
 982           receivedResponse.id = response.id
 983           self.assertEqual(response, receivedResponse)