[thirdparty/pdns.git] / regression-tests.dnsdist / test_DOH3.py

#!/usr/bin/env python
import dns
import clientsubnetoption

from dnsdisttests import DNSDistTest
from dnsdisttests import pickAvailablePort
from quictests import QUICTests, QUICWithCacheTests, QUICACLTests, QUICGetLocalAddressOnAnyBindTests
import doh3client

class TestDOH3(QUICTests, 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 = """
    newServer{address="127.0.0.1:%d"}

    addAction("drop.doq.tests.powerdns.com.", DropAction())
    addAction("refused.doq.tests.powerdns.com.", RCodeAction(DNSRCode.REFUSED))
    addAction("spoof.doq.tests.powerdns.com.", SpoofAction("1.2.3.4"))
    addAction("no-backend.doq.tests.powerdns.com.", PoolAction('this-pool-has-no-backend'))

    addDOH3Local("127.0.0.1:%d", "%s", "%s", {keyLogFile='/tmp/keys'})
    """
    _config_params = ['_testServerPort', '_doqServerPort','_serverCert', '_serverKey']
    _verboseMode = 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 TestDOH3ACL(QUICACLTests, 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 = """
    newServer{address="127.0.0.1:%d"}

    setACL("192.0.2.1/32")
    addDOH3Local("127.0.0.1:%d", "%s", "%s", {keyLogFile='/tmp/keys'})
    """
    _config_params = ['_testServerPort', '_doqServerPort','_serverCert', '_serverKey']
    _verboseMode = 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 TestDOH3Specifics(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 = """
    newServer{address="127.0.0.1:%d"}

    addDOH3Local("127.0.0.1:%d", "%s", "%s", {keyLogFile='/tmp/keys'})
    """
    _config_params = ['_testServerPort', '_doqServerPort','_serverCert', '_serverKey']
    _verboseMode = True

    def testDOH3Post(self):
        """
        QUIC: Simple POST query
        """
        name = 'simple.post.doq.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN', use_edns=False)
        query.id = 0
        expectedQuery = dns.message.make_query(name, 'A', 'IN', use_edns=True, payload=4096)
        expectedQuery.id = 0
        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    3600,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.A,
                                    '127.0.0.1')
        response.answer.append(rrset)
        (receivedQuery, receivedResponse) = self.sendDOH3Query(self._doqServerPort, self._dohBaseURL, query, response=response, caFile=self._caCert, serverName=self._serverName, post=True)
        self.assertTrue(receivedQuery)
        self.assertTrue(receivedResponse)
        receivedQuery.id = expectedQuery.id
        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)
Commit	Line	Data
4f0b10a9 CHB	1	#!/usr/bin/env python
	2	import dns
	3	import clientsubnetoption
	4
	5	from dnsdisttests import DNSDistTest
	6	from dnsdisttests import pickAvailablePort
88913b83	7	from quictests import QUICTests, QUICWithCacheTests, QUICACLTests, QUICGetLocalAddressOnAnyBindTests
4f0b10a9 CHB	8	import doh3client
4f0b10a9 CHB	9
ac70190e	10	class TestDOH3(QUICTests, DNSDistTest):
4f0b10a9 CHB	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
ac70190e	25	addDOH3Local("127.0.0.1:%d", "%s", "%s", {keyLogFile='/tmp/keys'})
4f0b10a9 CHB	26	"""
4f0b10a9 CHB	27	_config_params = ['_testServerPort', '_doqServerPort','_serverCert', '_serverKey']
2aaf9ecd CHB	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']
4f0b10a9 CHB	50	_verboseMode = True
4f0b10a9 CHB	51
ac70190e RG	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)
d0439b42 CHB	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)
88913b83 RG	95
	96	class TestDOH3GetLocalAddressOnAnyBind(QUICGetLocalAddressOnAnyBindTests, DNSDistTest):
	97	_serverKey = 'server.key'
	98	_serverCert = 'server.chain'
	99	_serverName = 'tls.tests.dnsdist.org'
	100	_caCert = 'ca.pem'
	101	_doqServerPort = pickAvailablePort()
	102	_dohBaseURL = ("https://%s:%d/" % (_serverName, _doqServerPort))
	103	_config_template = """
	104	function answerBasedOnLocalAddress(dq)
	105	local dest = tostring(dq.localaddr)
	106	local i, j = string.find(dest, "[0-9.]+")
	107	local addr = string.sub(dest, i, j)
	108	local dashAddr = string.gsub(addr, "[.]", "-")
	109	return DNSAction.Spoof, "address-was-"..dashAddr..".local-address-any.advanced.tests.powerdns.com."
	110	end
	111	addAction("local-address-any.quic.tests.powerdns.com.", LuaAction(answerBasedOnLocalAddress))
	112	newServer{address="127.0.0.1:%s"}
	113	addDOH3Local("0.0.0.0:%d", "%s", "%s")
	114	addDOH3Local("[::]:%d", "%s", "%s")
	115	"""
	116	_config_params = ['_testServerPort', '_doqServerPort','_serverCert', '_serverKey', '_doqServerPort','_serverCert', '_serverKey']
	117	_acl = ['127.0.0.1/32', '::1/128']
	118	_skipListeningOnCL = True
	119
	120	def getQUICConnection(self):
	121	return self.getDOQConnection(self._doqServerPort, self._caCert)
	122
	123	def sendQUICQuery(self, query, response=None, useQueue=True, connection=None):
	124	return self.sendDOH3Query(self._doqServerPort, self._dohBaseURL, query, response=response, caFile=self._caCert, useQueue=useQueue, serverName=self._serverName, connection=connection)