regression-tests.dnsdist/test_DOQ.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 doqclient import quic_bogus_query
   8 from quictests import QUICTests, QUICWithCacheTests, QUICACLTests
   9 import doqclient
  10
  11 class TestDOQBogus(DNSDistTest):
  12     _serverKey = 'server.key'
  13     _serverCert = 'server.chain'
  14     _serverName = 'tls.tests.dnsdist.org'
  15     _caCert = 'ca.pem'
  16     _doqServerPort = pickAvailablePort()
  17     _config_template = """
  18     newServer{address="127.0.0.1:%d"}
  19
  20     addDOQLocal("127.0.0.1:%d", "%s", "%s")
  21     """
  22     _config_params = ['_testServerPort', '_doqServerPort','_serverCert', '_serverKey']
  23     _verboseMode = True
  24
  25     def testDOQBogus(self):
  26         """
  27         DOQ: Test a bogus query (wrong packed length)
  28         """
  29         name = 'bogus.doq.tests.powerdns.com.'
  30         query = dns.message.make_query(name, 'A', 'IN', use_edns=False)
  31         query.id = 0
  32         expectedQuery = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096)
  33         expectedQuery.id = 0
  34
  35         try:
  36             message = quic_bogus_query(query, '127.0.0.1', 2.0, self._doqServerPort, verify=self._caCert, server_hostname=self._serverName)
  37             self.assertFalse(True)
  38         except doqclient.StreamResetError as e :
  39             self.assertEqual(e.error, 2);
  40
  41 class TestDOQ(QUICTests, DNSDistTest):
  42     _serverKey = 'server.key'
  43     _serverCert = 'server.chain'
  44     _serverName = 'tls.tests.dnsdist.org'
  45     _caCert = 'ca.pem'
  46     _doqServerPort = pickAvailablePort()
  47     _config_template = """
  48     newServer{address="127.0.0.1:%d"}
  49
  50     addAction("drop.doq.tests.powerdns.com.", DropAction())
  51     addAction("refused.doq.tests.powerdns.com.", RCodeAction(DNSRCode.REFUSED))
  52     addAction("spoof.doq.tests.powerdns.com.", SpoofAction("1.2.3.4"))
  53     addAction("no-backend.doq.tests.powerdns.com.", PoolAction('this-pool-has-no-backend'))
  54
  55     addDOQLocal("127.0.0.1:%d", "%s", "%s")
  56     """
  57     _config_params = ['_testServerPort', '_doqServerPort','_serverCert', '_serverKey']
  58     _verboseMode = True
  59
  60     def getQUICConnection(self):
  61         return self.getDOQConnection(self._doqServerPort, self._caCert)
  62
  63     def sendQUICQuery(self, query, response=None, useQueue=True, connection=None):
  64         return self.sendDOQQuery(self._doqServerPort, query, response=response, caFile=self._caCert, useQueue=useQueue, serverName=self._serverName, connection=connection)
  65
  66 class TestDOQWithCache(QUICWithCacheTests, DNSDistTest):
  67     _serverKey = 'server.key'
  68     _serverCert = 'server.chain'
  69     _serverName = 'tls.tests.dnsdist.org'
  70     _caCert = 'ca.pem'
  71     _doqServerPort = pickAvailablePort()
  72     _config_template = """
  73     newServer{address="127.0.0.1:%d"}
  74
  75     addDOQLocal("127.0.0.1:%d", "%s", "%s")
  76
  77     pc = newPacketCache(100, {maxTTL=86400, minTTL=1})
  78     getPool(""):setCache(pc)
  79     """
  80     _config_params = ['_testServerPort', '_doqServerPort','_serverCert', '_serverKey']
  81     _verboseMode = True
  82
  83     def getQUICConnection(self):
  84         return self.getDOQConnection(self._doqServerPort, self._caCert)
  85
  86     def sendQUICQuery(self, query, response=None, useQueue=True, connection=None):
  87         return self.sendDOQQuery(self._doqServerPort, query, response=response, caFile=self._caCert, useQueue=useQueue, serverName=self._serverName, connection=connection)
  88
  89 class TestDOQWithACL(QUICACLTests, DNSDistTest):
  90     _serverKey = 'server.key'
  91     _serverCert = 'server.chain'
  92     _serverName = 'tls.tests.dnsdist.org'
  93     _caCert = 'ca.pem'
  94     _doqServerPort = pickAvailablePort()
  95     _config_template = """
  96     newServer{address="127.0.0.1:%d"}
  97
  98     setACL("192.0.2.1/32")
  99     addDOQLocal("127.0.0.1:%d", "%s", "%s")
 100     """
 101     _config_params = ['_testServerPort', '_doqServerPort','_serverCert', '_serverKey']
 102     _verboseMode = True
 103
 104     def getQUICConnection(self):
 105         return self.getDOQConnection(self._doqServerPort, self._caCert)
 106
 107     def sendQUICQuery(self, query, response=None, useQueue=True, connection=None):
 108         return self.sendDOQQuery(self._doqServerPort, query, response=response, caFile=self._caCert, useQueue=useQueue, serverName=self._serverName, connection=connection)