regression-tests.dnsdist/test_ProxyProtocol.py

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