dnsdist: Add regression for destination address harvesting with QUIC

(cherry picked from commit 88913b8327e1cc443b9e22224e961c0fa47bfdb7)

diff --git a/regression-tests.dnsdist/quictests.py b/regression-tests.dnsdist/quictests.py
index d47baa75e8f49c66b363fd25de10722b1c043f5d..b4ec180722869ed750c2d9f5c6a3e7533b06f409 100644 (file)
--- a/regression-tests.dnsdist/quictests.py
+++ b/regression-tests.dnsdist/quictests.py
@@ -170,6 +170,28 @@ class QUICWithCacheTests(object):
 
         self.assertEqual(total, 1)
 
+class QUICGetLocalAddressOnAnyBindTests(object):
+
+    def testGetLocalAddressOnAnyBind(self):
+        """
+        QUIC: Return CNAME containing the local address for an ANY bind
+        """
+        name = 'local-address-any.quic.tests.powerdns.com.'
+        query = dns.message.make_query(name, 'A', 'IN')
+        # dnsdist set RA = RD for spoofed responses
+        query.flags &= ~dns.flags.RD
+
+        response = dns.message.make_response(query)
+        rrset = dns.rrset.from_text(name,
+                                    60,
+                                    dns.rdataclass.IN,
+                                    dns.rdatatype.CNAME,
+                                    'address-was-127-0-0-1.local-address-any.advanced.tests.powerdns.com.')
+        response.answer.append(rrset)
+
+        (_, receivedResponse) = self.sendQUICQuery(query, response=None, useQueue=False)
+        self.assertEqual(receivedResponse, response)
+
 class QUICXFRTests(object):
 
     def testXFR(self):

diff --git a/regression-tests.dnsdist/test_DOH3.py b/regression-tests.dnsdist/test_DOH3.py
index 422096206a98982ffee10dcb79fb50895cbad5d9..9634c914c7d8d2acecd7523abb15ad0aed9b3943 100644 (file)
--- a/regression-tests.dnsdist/test_DOH3.py
+++ b/regression-tests.dnsdist/test_DOH3.py
@@ -4,7 +4,7 @@ import clientsubnetoption
 
 from dnsdisttests import DNSDistTest
 from dnsdisttests import pickAvailablePort
-from quictests import QUICTests, QUICWithCacheTests, QUICACLTests, QUICXFRTests
+from quictests import QUICTests, QUICWithCacheTests, QUICACLTests, QUICGetLocalAddressOnAnyBindTests, QUICXFRTests
 import doh3client
 
 class TestDOH3(QUICTests, DNSDistTest):
@@ -93,6 +93,36 @@ class TestDOH3Specifics(DNSDistTest):
         self.assertEqual(expectedQuery, receivedQuery)
         self.assertEqual(receivedResponse, response)
 
+class TestDOH3GetLocalAddressOnAnyBind(QUICGetLocalAddressOnAnyBindTests, DNSDistTest):
+    _serverKey = 'server.key'
+    _serverCert = 'server.chain'
+    _serverName = 'tls.tests.dnsdist.org'
+    _caCert = 'ca.pem'
+    _doqServerPort = pickAvailablePort()
+    _dohBaseURL = ("https://%s:%d/" % (_serverName, _doqServerPort))
+    _config_template = """
+    function answerBasedOnLocalAddress(dq)
+      local dest = tostring(dq.localaddr)
+      local i, j = string.find(dest, "[0-9.]+")
+      local addr = string.sub(dest, i, j)
+      local dashAddr = string.gsub(addr, "[.]", "-")
+      return DNSAction.Spoof, "address-was-"..dashAddr..".local-address-any.advanced.tests.powerdns.com."
+    end
+    addAction("local-address-any.quic.tests.powerdns.com.", LuaAction(answerBasedOnLocalAddress))
+    newServer{address="127.0.0.1:%s"}
+    addDOH3Local("0.0.0.0:%d", "%s", "%s")
+    addDOH3Local("[::]:%d", "%s", "%s")
+    """
+    _config_params = ['_testServerPort', '_doqServerPort','_serverCert', '_serverKey', '_doqServerPort','_serverCert', '_serverKey']
+    _acl = ['127.0.0.1/32', '::1/128']
+    _skipListeningOnCL = True
+
+    def getQUICConnection(self):
+        return self.getDOQConnection(self._doqServerPort, self._caCert)
+
+    def sendQUICQuery(self, query, response=None, useQueue=True, connection=None):
+        return self.sendDOH3Query(self._doqServerPort, self._dohBaseURL, query, response=response, caFile=self._caCert, useQueue=useQueue, serverName=self._serverName, connection=connection)
+
 class TestDOH3XFR(QUICXFRTests, DNSDistTest):
     _serverKey = 'server.key'
     _serverCert = 'server.chain'

diff --git a/regression-tests.dnsdist/test_DOQ.py b/regression-tests.dnsdist/test_DOQ.py
index 033bdc7e8240ad6a74cade4f0706271bc9d77c70..5a817747e04c2fef38cf41da730dda93b3ed8b6f 100644 (file)
--- a/regression-tests.dnsdist/test_DOQ.py
+++ b/regression-tests.dnsdist/test_DOQ.py
@@ -6,7 +6,7 @@ import clientsubnetoption
 from dnsdisttests import DNSDistTest
 from dnsdisttests import pickAvailablePort
 from doqclient import quic_bogus_query
-from quictests import QUICTests, QUICWithCacheTests, QUICACLTests, QUICXFRTests
+from quictests import QUICTests, QUICWithCacheTests, QUICACLTests, QUICGetLocalAddressOnAnyBindTests, QUICXFRTests
 import doqclient
 from doqclient import quic_query
 
@@ -162,3 +162,32 @@ class TestDOQCertificateReloading(DNSDistTest):
         (_, secondSerial) = quic_query(query, '127.0.0.1', 0.5, self._doqServerPort, verify=self._caCert, server_hostname=self._serverName)
         # check that the serial is different
         self.assertNotEqual(serial, secondSerial)
+
+class TestDOQGetLocalAddressOnAnyBind(QUICGetLocalAddressOnAnyBindTests, DNSDistTest):
+    _serverKey = 'server.key'
+    _serverCert = 'server.chain'
+    _serverName = 'tls.tests.dnsdist.org'
+    _caCert = 'ca.pem'
+    _doqServerPort = pickAvailablePort()
+    _config_template = """
+    function answerBasedOnLocalAddress(dq)
+      local dest = tostring(dq.localaddr)
+      local i, j = string.find(dest, "[0-9.]+")
+      local addr = string.sub(dest, i, j)
+      local dashAddr = string.gsub(addr, "[.]", "-")
+      return DNSAction.Spoof, "address-was-"..dashAddr..".local-address-any.advanced.tests.powerdns.com."
+    end
+    addAction("local-address-any.quic.tests.powerdns.com.", LuaAction(answerBasedOnLocalAddress))
+    newServer{address="127.0.0.1:%s"}
+    addDOQLocal("0.0.0.0:%d", "%s", "%s")
+    addDOQLocal("[::]:%d", "%s", "%s")
+    """
+    _config_params = ['_testServerPort', '_doqServerPort','_serverCert', '_serverKey', '_doqServerPort','_serverCert', '_serverKey']
+    _acl = ['127.0.0.1/32', '::1/128']
+    _skipListeningOnCL = True
+
+    def getQUICConnection(self):
+        return self.getDOQConnection(self._doqServerPort, self._caCert)
+
+    def sendQUICQuery(self, query, response=None, useQueue=True, connection=None):
+        return self.sendDOQQuery(self._doqServerPort, query, response=response, caFile=self._caCert, useQueue=useQueue, serverName=self._serverName, connection=connection)