regression-tests.dnsdist/test_DOH3.py

   1 #!/usr/bin/env python
   2 import dns
   3 import clientsubnetoption
   4
   5 from dnsdisttests import DNSDistTest
   6 from dnsdisttests import pickAvailablePort
   7 from quictests import QUICTests, QUICWithCacheTests, QUICACLTests
   8 import doh3client
   9
  10 class TestDOH3(QUICTests, DNSDistTest):
  11     _serverKey = 'server.key'
  12     _serverCert = 'server.chain'
  13     _serverName = 'tls.tests.dnsdist.org'
  14     _caCert = 'ca.pem'
  15     _doqServerPort = pickAvailablePort()
  16     _dohBaseURL = ("https://%s:%d/" % (_serverName, _doqServerPort))
  17     _config_template = """
  18     newServer{address="127.0.0.1:%d"}
  19
  20     addAction("drop.doq.tests.powerdns.com.", DropAction())
  21     addAction("refused.doq.tests.powerdns.com.", RCodeAction(DNSRCode.REFUSED))
  22     addAction("spoof.doq.tests.powerdns.com.", SpoofAction("1.2.3.4"))
  23     addAction("no-backend.doq.tests.powerdns.com.", PoolAction('this-pool-has-no-backend'))
  24
  25     addDOH3Local("127.0.0.1:%d", "%s", "%s", {keyLogFile='/tmp/keys'})
  26     """
  27     _config_params = ['_testServerPort', '_doqServerPort','_serverCert', '_serverKey']
  28     _verboseMode = True
  29
  30     def getQUICConnection(self):
  31         return self.getDOQConnection(self._doqServerPort, self._caCert)
  32
  33     def sendQUICQuery(self, query, response=None, useQueue=True, connection=None):
  34         return self.sendDOH3Query(self._doqServerPort, self._dohBaseURL, query, response=response, caFile=self._caCert, useQueue=useQueue, serverName=self._serverName, connection=connection)
  35
  36 class TestDOH3ACL(QUICACLTests, DNSDistTest):
  37     _serverKey = 'server.key'
  38     _serverCert = 'server.chain'
  39     _serverName = 'tls.tests.dnsdist.org'
  40     _caCert = 'ca.pem'
  41     _doqServerPort = pickAvailablePort()
  42     _dohBaseURL = ("https://%s:%d/" % (_serverName, _doqServerPort))
  43     _config_template = """
  44     newServer{address="127.0.0.1:%d"}
  45
  46     setACL("192.0.2.1/32")
  47     addDOH3Local("127.0.0.1:%d", "%s", "%s", {keyLogFile='/tmp/keys'})
  48     """
  49     _config_params = ['_testServerPort', '_doqServerPort','_serverCert', '_serverKey']
  50     _verboseMode = True
  51
  52     def getQUICConnection(self):
  53         return self.getDOQConnection(self._doqServerPort, self._caCert)
  54
  55     def sendQUICQuery(self, query, response=None, useQueue=True, connection=None):
  56         return self.sendDOH3Query(self._doqServerPort, self._dohBaseURL, query, response=response, caFile=self._caCert, useQueue=useQueue, serverName=self._serverName, connection=connection)
  57
  58 class TestDOH3Specifics(DNSDistTest):
  59     _serverKey = 'server.key'
  60     _serverCert = 'server.chain'
  61     _serverName = 'tls.tests.dnsdist.org'
  62     _caCert = 'ca.pem'
  63     _doqServerPort = pickAvailablePort()
  64     _dohBaseURL = ("https://%s:%d/" % (_serverName, _doqServerPort))
  65     _config_template = """
  66     newServer{address="127.0.0.1:%d"}
  67
  68     addDOH3Local("127.0.0.1:%d", "%s", "%s", {keyLogFile='/tmp/keys'})
  69     """
  70     _config_params = ['_testServerPort', '_doqServerPort','_serverCert', '_serverKey']
  71     _verboseMode = True
  72
  73     def testDOH3Post(self):
  74         """
  75         QUIC: Simple POST query
  76         """
  77         name = 'simple.post.doq.tests.powerdns.com.'
  78         query = dns.message.make_query(name, 'A', 'IN', use_edns=False)
  79         query.id = 0
  80         expectedQuery = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096)
  81         expectedQuery.id = 0
  82         response = dns.message.make_response(query)
  83         rrset = dns.rrset.from_text(name,
  84                                     3600,
  85                                     dns.rdataclass.IN,
  86                                     dns.rdatatype.A,
  87                                     '127.0.0.1')
  88         response.answer.append(rrset)
  89         (receivedQuery, receivedResponse) = self.sendDOH3Query(self._doqServerPort, self._dohBaseURL, query, response=response, caFile=self._caCert, serverName=self._serverName, post=True)
  90         self.assertTrue(receivedQuery)
  91         self.assertTrue(receivedResponse)
  92         receivedQuery.id = expectedQuery.id
  93         self.assertEqual(expectedQuery, receivedQuery)
  94         self.assertEqual(receivedResponse, response)