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
   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 TestDOH3Specifics(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     addDOH3Local("127.0.0.1:%d", "%s", "%s", {keyLogFile='/tmp/keys'})
  47     """
  48     _config_params = ['_testServerPort', '_doqServerPort','_serverCert', '_serverKey']
  49     _verboseMode = True
  50
  51     def testDOH3Post(self):
  52         """
  53         QUIC: Simple POST query
  54         """
  55         name = 'simple.post.doq.tests.powerdns.com.'
  56         query = dns.message.make_query(name, 'A', 'IN', use_edns=False)
  57         query.id = 0
  58         expectedQuery = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096)
  59         expectedQuery.id = 0
  60         response = dns.message.make_response(query)
  61         rrset = dns.rrset.from_text(name,
  62                                     3600,
  63                                     dns.rdataclass.IN,
  64                                     dns.rdatatype.A,
  65                                     '127.0.0.1')
  66         response.answer.append(rrset)
  67         (receivedQuery, receivedResponse) = self.sendDOH3Query(self._doqServerPort, self._dohBaseURL, query, response=response, caFile=self._caCert, serverName=self._serverName, post=True)
  68         self.assertTrue(receivedQuery)
  69         self.assertTrue(receivedResponse)
  70         receivedQuery.id = expectedQuery.id
  71         self.assertEqual(expectedQuery, receivedQuery)
  72         self.assertEqual(receivedResponse, response)