regression-tests.dnsdist/test_ProxyProtocol.py

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