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

#!/usr/bin/env python
from datetime import datetime, timedelta
import time
import dns
from dnsdisttests import DNSDistTest

class TestResponseRuleNXDelayed(DNSDistTest):

    _config_template = """
    newServer{address="127.0.0.1:%s"}
    addResponseAction(RCodeRule(DNSRCode.NXDOMAIN), DelayResponseAction(1000))
    """

    def testNXDelayed(self):
        """
        Responses: Delayed on NXDomain

        Send an A query to "delayed.responses.tests.powerdns.com.",
        check that the response delay is longer than 1000 ms
        for a NXDomain response over UDP, shorter for a NoError one.
        """
        name = 'delayed.responses.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)

        # NX over UDP
        response.set_rcode(dns.rcode.NXDOMAIN)
        begin = datetime.now()
        (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
        end = datetime.now()
        receivedQuery.id = query.id
        self.assertEqual(query, receivedQuery)
        self.assertEqual(response, receivedResponse)
        self.assertTrue((end - begin) > timedelta(0, 1))

        # NoError over UDP
        response.set_rcode(dns.rcode.NOERROR)
        begin = datetime.now()
        (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
        end = datetime.now()
        receivedQuery.id = query.id
        self.assertEqual(query, receivedQuery)
        self.assertEqual(response, receivedResponse)
        self.assertTrue((end - begin) < timedelta(0, 1))

        # NX over TCP
        response.set_rcode(dns.rcode.NXDOMAIN)
        begin = datetime.now()
        (receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
        end = datetime.now()
        receivedQuery.id = query.id
        self.assertEqual(query, receivedQuery)
        self.assertEqual(response, receivedResponse)
        self.assertTrue((end - begin) < timedelta(0, 1))

class TestResponseRuleERCode(DNSDistTest):

    _extraStartupSleep = 1
    _config_template = """
    newServer{address="127.0.0.1:%s"}
    addResponseAction(ERCodeRule(DNSRCode.BADVERS), DelayResponseAction(1000))
    """

    def testBADVERSDelayed(self):
        """
        Responses: Delayed on BADVERS

        Send an A query to "delayed.responses.tests.powerdns.com.",
        check that the response delay is longer than 1000 ms
        for a BADVERS response over UDP, shorter for BADKEY and NoError.
        """
        name = 'delayed.responses.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)
        response.use_edns(edns=True)

        # BADVERS over UDP
        # BADVERS == 16, so rcode==0, ercode==1
        response.set_rcode(dns.rcode.BADVERS)
        begin = datetime.now()
        (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
        end = datetime.now()
        receivedQuery.id = query.id
        self.assertEqual(query, receivedQuery)
        self.assertEqual(response, receivedResponse)
        self.assertTrue((end - begin) > timedelta(0, 1))

        # BADKEY (17, an ERCode) over UDP
        response.set_rcode(17)
        begin = datetime.now()
        (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
        end = datetime.now()
        receivedQuery.id = query.id
        self.assertEqual(query, receivedQuery)
        self.assertEqual(response, receivedResponse)
        self.assertTrue((end - begin) < timedelta(0, 1))

        # NoError (non-ERcode, basic RCode bits match BADVERS) over UDP
        response.set_rcode(dns.rcode.NOERROR)
        begin = datetime.now()
        (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
        end = datetime.now()
        receivedQuery.id = query.id
        self.assertEqual(query, receivedQuery)
        self.assertEqual(response, receivedResponse)
        self.assertTrue((end - begin) < timedelta(0, 1))

class TestResponseRuleQNameDropped(DNSDistTest):

    _config_template = """
    newServer{address="127.0.0.1:%s"}
    addResponseAction("drop.responses.tests.powerdns.com.", DropResponseAction())
    """

    def testDropped(self):
        """
        Responses: Dropped on QName

        Send an A query to "drop.responses.tests.powerdns.com.",
        check that the response (not the query) is dropped.
        """
        name = 'drop.responses.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (receivedQuery, receivedResponse) = sender(query, response)
            receivedQuery.id = query.id
            self.assertEqual(query, receivedQuery)
            self.assertEqual(receivedResponse, None)

    def testNotDropped(self):
        """
        Responses: NOT Dropped on QName

        Send an A query to "dontdrop.responses.tests.powerdns.com.",
        check that the response is not dropped.
        """
        name = 'dontdrop.responses.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (receivedQuery, receivedResponse) = sender(query, response)
            receivedQuery.id = query.id
            self.assertEqual(query, receivedQuery)
            self.assertEqual(response, receivedResponse)

class TestResponseRuleQNameAllowed(DNSDistTest):

    _config_template = """
    newServer{address="127.0.0.1:%s"}
    addResponseAction("allow.responses.tests.powerdns.com.", AllowResponseAction())
    addResponseAction(AllRule(), DropResponseAction())
    """

    def testAllowed(self):
        """
        Responses: Allowed on QName

        Send an A query to "allow.responses.tests.powerdns.com.",
        check that the response is allowed.
        """
        name = 'allow.responses.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (receivedQuery, receivedResponse) = sender(query, response)
            receivedQuery.id = query.id
            self.assertEqual(query, receivedQuery)
            self.assertEqual(response, receivedResponse)

    def testNotAllowed(self):
        """
        Responses: Not allowed on QName

        Send an A query to "dontallow.responses.tests.powerdns.com.",
        check that the response is dropped.
        """
        name = 'dontallow.responses.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (receivedQuery, receivedResponse) = sender(query, response)
            receivedQuery.id = query.id
            self.assertEqual(query, receivedQuery)
            self.assertEqual(receivedResponse, None)

class TestResponseRuleEditTTL(DNSDistTest):

    _ttl = 5
    _config_params = ['_testServerPort', '_ttl']
    _config_template = """
    newServer{address="127.0.0.1:%s"}

    function editTTLCallback(section, class, type, ttl)
      return %d
    end

    function editTTLFunc(dr)
      dr:editTTLs(editTTLCallback)
      return DNSAction.None, ""
    end

    addResponseAction(AllRule(), LuaResponseAction(editTTLFunc))
    """

    def testTTLEdited(self):
        """
        Responses: Alter the TTLs
        """
        name = 'editttl.responses.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    3600,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.A,
                                    '192.0.2.1')
        response.answer.append(rrset)

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (receivedQuery, receivedResponse) = sender(query, response)
            receivedQuery.id = query.id
            self.assertEqual(query, receivedQuery)
            self.assertEqual(response, receivedResponse)
            self.assertNotEqual(response.answer[0].ttl, receivedResponse.answer[0].ttl)
            self.assertEqual(receivedResponse.answer[0].ttl, self._ttl)

class TestResponseRuleLimitTTL(DNSDistTest):

    _lowttl = 60
    _defaulttl = 3600
    _highttl = 18000
    _config_params = ['_lowttl', '_highttl', '_testServerPort']
    _config_template = """
    local ffi = require("ffi")
    local lowttl = %d
    local highttl = %d

    function luaFFISetMinTTL(dr)
      ffi.C.dnsdist_ffi_dnsresponse_set_min_ttl(dr, highttl)
      return DNSResponseAction.None, ""
    end
    function luaFFISetMaxTTL(dr)
      ffi.C.dnsdist_ffi_dnsresponse_set_max_ttl(dr, lowttl)
      return DNSResponseAction.None, ""
    end

    newServer{address="127.0.0.1:%s"}

    addResponseAction("min.responses.tests.powerdns.com.", SetMinTTLResponseAction(highttl))
    addResponseAction("max.responses.tests.powerdns.com.", SetMaxTTLResponseAction(lowttl))
    addResponseAction("ffi.min.limitttl.responses.tests.powerdns.com.", LuaFFIResponseAction(luaFFISetMinTTL))
    addResponseAction("ffi.max.limitttl.responses.tests.powerdns.com.", LuaFFIResponseAction(luaFFISetMaxTTL))
    """

    def testLimitTTL(self):
        """
        Responses: Alter the TTLs via Limiter
        """
        name = 'min.responses.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    3600,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.A,
                                    '192.0.2.1')
        response.answer.append(rrset)

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (receivedQuery, receivedResponse) = sender(query, response)
            receivedQuery.id = query.id
            self.assertEqual(query, receivedQuery)
            self.assertEqual(response, receivedResponse)
            self.assertNotEqual(response.answer[0].ttl, receivedResponse.answer[0].ttl)
            self.assertEqual(receivedResponse.answer[0].ttl, self._highttl)

        name = 'max.responses.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    3600,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.A,
                                    '192.0.2.1')
        response.answer.append(rrset)

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (receivedQuery, receivedResponse) = sender(query, response)
            receivedQuery.id = query.id
            self.assertEqual(query, receivedQuery)
            self.assertEqual(response, receivedResponse)
            self.assertNotEqual(response.answer[0].ttl, receivedResponse.answer[0].ttl)
            self.assertEqual(receivedResponse.answer[0].ttl, self._lowttl)

    def testLimitTTLFFI(self):
        """
        Responses: Alter the TTLs via Limiter
        """
        name = 'ffi.min.responses.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    3600,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.A,
                                    '192.0.2.1')
        response.answer.append(rrset)

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (receivedQuery, receivedResponse) = sender(query, response)
            receivedQuery.id = query.id
            self.assertEqual(query, receivedQuery)
            self.assertEqual(response, receivedResponse)
            self.assertNotEqual(response.answer[0].ttl, receivedResponse.answer[0].ttl)
            self.assertEqual(receivedResponse.answer[0].ttl, self._highttl)

        name = 'ffi.max.responses.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    3600,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.A,
                                    '192.0.2.1')
        response.answer.append(rrset)

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (receivedQuery, receivedResponse) = sender(query, response)
            receivedQuery.id = query.id
            self.assertEqual(query, receivedQuery)
            self.assertEqual(response, receivedResponse)
            self.assertNotEqual(response.answer[0].ttl, receivedResponse.answer[0].ttl)
            self.assertEqual(receivedResponse.answer[0].ttl, self._lowttl)

class TestSetReducedTTL(DNSDistTest):

    _percentage = 42
    _initialTTL = 100
    _config_params = ['_percentage', '_testServerPort']
    _config_template = """
    addResponseAction(AllRule(), SetReducedTTLResponseAction(%d))
    newServer{address="127.0.0.1:%s"}
    """

    def testLimitTTL(self):
        """
        Responses: Reduce TTL to 42%
        """
        name = 'reduced-ttl.responses.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    self._initialTTL,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.A,
                                    '192.0.2.1')
        response.answer.append(rrset)

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (receivedQuery, receivedResponse) = sender(query, response)
            receivedQuery.id = query.id
            self.assertEqual(query, receivedQuery)
            self.assertEqual(response, receivedResponse)
            self.assertNotEqual(response.answer[0].ttl, receivedResponse.answer[0].ttl)
            self.assertEqual(receivedResponse.answer[0].ttl, self._percentage)

class TestResponseLuaActionReturnSyntax(DNSDistTest):

    _config_template = """
    newServer{address="127.0.0.1:%s"}
    function customDelay(dr)
      return DNSResponseAction.Delay, "1000"
    end
    function customDrop(dr)
      return DNSResponseAction.Drop
    end
    addResponseAction("drop.responses.tests.powerdns.com.", LuaResponseAction(customDrop))
    addResponseAction(RCodeRule(DNSRCode.NXDOMAIN), LuaResponseAction(customDelay))
    """

    def testResponseActionDelayed(self):
        """
        Responses: Delayed via LuaResponseAction

        Send an A query to "delayed.responses.tests.powerdns.com.",
        check that the response delay is longer than 1000 ms
        for a NXDomain response over UDP, shorter for a NoError one.
        """
        name = 'delayed.responses.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)

        # NX over UDP
        response.set_rcode(dns.rcode.NXDOMAIN)
        begin = datetime.now()
        (receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
        end = datetime.now()
        receivedQuery.id = query.id
        self.assertEqual(query, receivedQuery)
        self.assertEqual(response, receivedResponse)
        self.assertTrue((end - begin) > timedelta(0, 1))

    def testDropped(self):
        """
        Responses: Dropped via user defined LuaResponseAction

        Send an A query to "drop.responses.tests.powerdns.com.",
        check that the response (not the query) is dropped.
        """
        name = 'drop.responses.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (receivedQuery, receivedResponse) = sender(query, response)
            receivedQuery.id = query.id
            self.assertEqual(query, receivedQuery)
            self.assertEqual(receivedResponse, None)

class TestResponseClearRecordsType(DNSDistTest):

    _config_params = ['_testServerPort']
    _config_template = """
    local ffi = require("ffi")

    function luafct(dr)
      ffi.C.dnsdist_ffi_dnsresponse_clear_records_type(dr, DNSQType.AAAA)
      return DNSResponseAction.HeaderModify, ""
    end

    newServer{address="127.0.0.1:%s"}

    addResponseAction("ffi.clear-records-type.responses.tests.powerdns.com.", LuaFFIResponseAction(luafct))
    addResponseAction("clear-records-type.responses.tests.powerdns.com.", ClearRecordTypesResponseAction(DNSQType.AAAA))
    """

    def testClearedFFI(self):
        """
        Responses: Removes records of a given type (FFI API)
        """
        name = 'ffi.clear-records-type.responses.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)
        expectedResponse = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    3600,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.A,
                                    '192.0.2.1')
        response.answer.append(rrset)
        expectedResponse.answer.append(rrset)
        rrset = dns.rrset.from_text(name,
                                    3660,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.AAAA,
                                    '2001:DB8::1', '2001:DB8::2')
        response.answer.append(rrset)
        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (receivedQuery, receivedResponse) = sender(query, response)
            receivedQuery.id = query.id
            self.assertEqual(query, receivedQuery)
            self.assertEqual(expectedResponse, receivedResponse)

    def testCleared(self):
        """
        Responses: Removes records of a given type
        """
        name = 'clear-records-type.responses.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)
        expectedResponse = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    3600,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.A,
                                    '192.0.2.1')
        response.answer.append(rrset)
        expectedResponse.answer.append(rrset)
        rrset = dns.rrset.from_text(name,
                                    3660,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.AAAA,
                                    '2001:DB8::1', '2001:DB8::2')
        response.answer.append(rrset)
        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (receivedQuery, receivedResponse) = sender(query, response)
            receivedQuery.id = query.id
            self.assertEqual(query, receivedQuery)
            self.assertEqual(expectedResponse, receivedResponse)
Commit	Line	Data
788c3243 RG	1	#!/usr/bin/env python
	2	from datetime import datetime, timedelta
	3	import time
	4	import dns
	5	from dnsdisttests import DNSDistTest
	6
	7	class TestResponseRuleNXDelayed(DNSDistTest):
	8
	9	_config_template = """
	10	newServer{address="127.0.0.1:%s"}
d3ec24f9	11	addResponseAction(RCodeRule(DNSRCode.NXDOMAIN), DelayResponseAction(1000))
788c3243 RG	12	"""
	13
	14	def testNXDelayed(self):
	15	"""
	16	Responses: Delayed on NXDomain
	17
	18	Send an A query to "delayed.responses.tests.powerdns.com.",
	19	check that the response delay is longer than 1000 ms
	20	for a NXDomain response over UDP, shorter for a NoError one.
	21	"""
	22	name = 'delayed.responses.tests.powerdns.com.'
	23	query = dns.message.make_query(name, 'A', 'IN')
	24	response = dns.message.make_response(query)
	25
	26	# NX over UDP
	27	response.set_rcode(dns.rcode.NXDOMAIN)
	28	begin = datetime.now()
	29	(receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
	30	end = datetime.now()
	31	receivedQuery.id = query.id
4bfebc93 CH	32	self.assertEqual(query, receivedQuery)
4bfebc93 CH	33	self.assertEqual(response, receivedResponse)
788c3243 RG	34	self.assertTrue((end - begin) > timedelta(0, 1))
	35
	36	# NoError over UDP
	37	response.set_rcode(dns.rcode.NOERROR)
	38	begin = datetime.now()
	39	(receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
	40	end = datetime.now()
	41	receivedQuery.id = query.id
4bfebc93 CH	42	self.assertEqual(query, receivedQuery)
4bfebc93 CH	43	self.assertEqual(response, receivedResponse)
788c3243 RG	44	self.assertTrue((end - begin) < timedelta(0, 1))
	45
	46	# NX over TCP
	47	response.set_rcode(dns.rcode.NXDOMAIN)
	48	begin = datetime.now()
	49	(receivedQuery, receivedResponse) = self.sendTCPQuery(query, response)
	50	end = datetime.now()
	51	receivedQuery.id = query.id
4bfebc93 CH	52	self.assertEqual(query, receivedQuery)
4bfebc93 CH	53	self.assertEqual(response, receivedResponse)
788c3243 RG	54	self.assertTrue((end - begin) < timedelta(0, 1))
788c3243 RG	55
d83feb68 CH	56	class TestResponseRuleERCode(DNSDistTest):
d83feb68 CH	57
b38aaeb5	58	_extraStartupSleep = 1
d83feb68 CH	59	_config_template = """
d83feb68 CH	60	newServer{address="127.0.0.1:%s"}
d3ec24f9	61	addResponseAction(ERCodeRule(DNSRCode.BADVERS), DelayResponseAction(1000))
d83feb68 CH	62	"""
	63
	64	def testBADVERSDelayed(self):
	65	"""
	66	Responses: Delayed on BADVERS
	67
	68	Send an A query to "delayed.responses.tests.powerdns.com.",
	69	check that the response delay is longer than 1000 ms
	70	for a BADVERS response over UDP, shorter for BADKEY and NoError.
	71	"""
	72	name = 'delayed.responses.tests.powerdns.com.'
	73	query = dns.message.make_query(name, 'A', 'IN')
	74	response = dns.message.make_response(query)
	75	response.use_edns(edns=True)
	76
	77	# BADVERS over UDP
	78	# BADVERS == 16, so rcode==0, ercode==1
	79	response.set_rcode(dns.rcode.BADVERS)
	80	begin = datetime.now()
	81	(receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
	82	end = datetime.now()
	83	receivedQuery.id = query.id
4bfebc93 CH	84	self.assertEqual(query, receivedQuery)
4bfebc93 CH	85	self.assertEqual(response, receivedResponse)
d83feb68 CH	86	self.assertTrue((end - begin) > timedelta(0, 1))
	87
	88	# BADKEY (17, an ERCode) over UDP
	89	response.set_rcode(17)
	90	begin = datetime.now()
	91	(receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
	92	end = datetime.now()
	93	receivedQuery.id = query.id
4bfebc93 CH	94	self.assertEqual(query, receivedQuery)
4bfebc93 CH	95	self.assertEqual(response, receivedResponse)
d83feb68 CH	96	self.assertTrue((end - begin) < timedelta(0, 1))
	97
	98	# NoError (non-ERcode, basic RCode bits match BADVERS) over UDP
	99	response.set_rcode(dns.rcode.NOERROR)
	100	begin = datetime.now()
	101	(receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
	102	end = datetime.now()
	103	receivedQuery.id = query.id
4bfebc93 CH	104	self.assertEqual(query, receivedQuery)
4bfebc93 CH	105	self.assertEqual(response, receivedResponse)
d83feb68 CH	106	self.assertTrue((end - begin) < timedelta(0, 1))
d83feb68 CH	107
788c3243 RG	108	class TestResponseRuleQNameDropped(DNSDistTest):
	109
	110	_config_template = """
	111	newServer{address="127.0.0.1:%s"}
	112	addResponseAction("drop.responses.tests.powerdns.com.", DropResponseAction())
	113	"""
	114
	115	def testDropped(self):
	116	"""
	117	Responses: Dropped on QName
	118
	119	Send an A query to "drop.responses.tests.powerdns.com.",
	120	check that the response (not the query) is dropped.
	121	"""
	122	name = 'drop.responses.tests.powerdns.com.'
	123	query = dns.message.make_query(name, 'A', 'IN')
	124	response = dns.message.make_response(query)
	125
6ca2e796 RG	126	for method in ("sendUDPQuery", "sendTCPQuery"):
	127	sender = getattr(self, method)
	128	(receivedQuery, receivedResponse) = sender(query, response)
	129	receivedQuery.id = query.id
4bfebc93 CH	130	self.assertEqual(query, receivedQuery)
4bfebc93 CH	131	self.assertEqual(receivedResponse, None)
788c3243 RG	132
	133	def testNotDropped(self):
	134	"""
	135	Responses: NOT Dropped on QName
	136
	137	Send an A query to "dontdrop.responses.tests.powerdns.com.",
	138	check that the response is not dropped.
	139	"""
	140	name = 'dontdrop.responses.tests.powerdns.com.'
	141	query = dns.message.make_query(name, 'A', 'IN')
	142	response = dns.message.make_response(query)
	143
6ca2e796 RG	144	for method in ("sendUDPQuery", "sendTCPQuery"):
	145	sender = getattr(self, method)
	146	(receivedQuery, receivedResponse) = sender(query, response)
	147	receivedQuery.id = query.id
4bfebc93 CH	148	self.assertEqual(query, receivedQuery)
4bfebc93 CH	149	self.assertEqual(response, receivedResponse)
788c3243 RG	150
	151	class TestResponseRuleQNameAllowed(DNSDistTest):
	152
	153	_config_template = """
	154	newServer{address="127.0.0.1:%s"}
	155	addResponseAction("allow.responses.tests.powerdns.com.", AllowResponseAction())
	156	addResponseAction(AllRule(), DropResponseAction())
	157	"""
	158
	159	def testAllowed(self):
	160	"""
	161	Responses: Allowed on QName
	162
	163	Send an A query to "allow.responses.tests.powerdns.com.",
	164	check that the response is allowed.
	165	"""
	166	name = 'allow.responses.tests.powerdns.com.'
	167	query = dns.message.make_query(name, 'A', 'IN')
	168	response = dns.message.make_response(query)
	169
6ca2e796 RG	170	for method in ("sendUDPQuery", "sendTCPQuery"):
	171	sender = getattr(self, method)
	172	(receivedQuery, receivedResponse) = sender(query, response)
	173	receivedQuery.id = query.id
4bfebc93 CH	174	self.assertEqual(query, receivedQuery)
4bfebc93 CH	175	self.assertEqual(response, receivedResponse)
788c3243 RG	176
	177	def testNotAllowed(self):
	178	"""
	179	Responses: Not allowed on QName
	180
	181	Send an A query to "dontallow.responses.tests.powerdns.com.",
	182	check that the response is dropped.
	183	"""
	184	name = 'dontallow.responses.tests.powerdns.com.'
	185	query = dns.message.make_query(name, 'A', 'IN')
	186	response = dns.message.make_response(query)
	187
6ca2e796 RG	188	for method in ("sendUDPQuery", "sendTCPQuery"):
	189	sender = getattr(self, method)
	190	(receivedQuery, receivedResponse) = sender(query, response)
	191	receivedQuery.id = query.id
4bfebc93 CH	192	self.assertEqual(query, receivedQuery)
4bfebc93 CH	193	self.assertEqual(receivedResponse, None)
153d5065 RG	194
	195	class TestResponseRuleEditTTL(DNSDistTest):
	196
	197	_ttl = 5
	198	_config_params = ['_testServerPort', '_ttl']
	199	_config_template = """
	200	newServer{address="127.0.0.1:%s"}
	201
	202	function editTTLCallback(section, class, type, ttl)
	203	return %d
	204	end
	205
	206	function editTTLFunc(dr)
	207	dr:editTTLs(editTTLCallback)
	208	return DNSAction.None, ""
	209	end
	210
a2ff35e3	211	addResponseAction(AllRule(), LuaResponseAction(editTTLFunc))
153d5065 RG	212	"""
	213
	214	def testTTLEdited(self):
	215	"""
	216	Responses: Alter the TTLs
	217	"""
	218	name = 'editttl.responses.tests.powerdns.com.'
	219	query = dns.message.make_query(name, 'A', 'IN')
	220	response = dns.message.make_response(query)
	221	rrset = dns.rrset.from_text(name,
	222	3600,
	223	dns.rdataclass.IN,
	224	dns.rdatatype.A,
	225	'192.0.2.1')
	226	response.answer.append(rrset)
	227
6ca2e796 RG	228	for method in ("sendUDPQuery", "sendTCPQuery"):
	229	sender = getattr(self, method)
	230	(receivedQuery, receivedResponse) = sender(query, response)
	231	receivedQuery.id = query.id
4bfebc93 CH	232	self.assertEqual(query, receivedQuery)
	233	self.assertEqual(response, receivedResponse)
	234	self.assertNotEqual(response.answer[0].ttl, receivedResponse.answer[0].ttl)
	235	self.assertEqual(receivedResponse.answer[0].ttl, self._ttl)
63cdc73d	236
aa8886da CHB	237	class TestResponseRuleLimitTTL(DNSDistTest):
	238
	239	_lowttl = 60
	240	_defaulttl = 3600
	241	_highttl = 18000
	242	_config_params = ['_lowttl', '_highttl', '_testServerPort']
	243	_config_template = """
	244	local ffi = require("ffi")
	245	local lowttl = %d
	246	local highttl = %d
	247
	248	function luaFFISetMinTTL(dr)
	249	ffi.C.dnsdist_ffi_dnsresponse_set_min_ttl(dr, highttl)
07a6346d	250	return DNSResponseAction.None, ""
aa8886da CHB	251	end
	252	function luaFFISetMaxTTL(dr)
	253	ffi.C.dnsdist_ffi_dnsresponse_set_max_ttl(dr, lowttl)
07a6346d	254	return DNSResponseAction.None, ""
aa8886da CHB	255	end
	256
	257	newServer{address="127.0.0.1:%s"}
	258
	259	addResponseAction("min.responses.tests.powerdns.com.", SetMinTTLResponseAction(highttl))
	260	addResponseAction("max.responses.tests.powerdns.com.", SetMaxTTLResponseAction(lowttl))
bb5f0d6e CHB	261	addResponseAction("ffi.min.limitttl.responses.tests.powerdns.com.", LuaFFIResponseAction(luaFFISetMinTTL))
bb5f0d6e CHB	262	addResponseAction("ffi.max.limitttl.responses.tests.powerdns.com.", LuaFFIResponseAction(luaFFISetMaxTTL))
aa8886da CHB	263	"""
	264
	265	def testLimitTTL(self):
	266	"""
	267	Responses: Alter the TTLs via Limiter
	268	"""
	269	name = 'min.responses.tests.powerdns.com.'
	270	query = dns.message.make_query(name, 'A', 'IN')
	271	response = dns.message.make_response(query)
	272	rrset = dns.rrset.from_text(name,
	273	3600,
	274	dns.rdataclass.IN,
	275	dns.rdatatype.A,
	276	'192.0.2.1')
	277	response.answer.append(rrset)
	278
	279	for method in ("sendUDPQuery", "sendTCPQuery"):
	280	sender = getattr(self, method)
	281	(receivedQuery, receivedResponse) = sender(query, response)
	282	receivedQuery.id = query.id
	283	self.assertEqual(query, receivedQuery)
	284	self.assertEqual(response, receivedResponse)
	285	self.assertNotEqual(response.answer[0].ttl, receivedResponse.answer[0].ttl)
	286	self.assertEqual(receivedResponse.answer[0].ttl, self._highttl)
	287
	288	name = 'max.responses.tests.powerdns.com.'
	289	query = dns.message.make_query(name, 'A', 'IN')
	290	response = dns.message.make_response(query)
	291	rrset = dns.rrset.from_text(name,
	292	3600,
	293	dns.rdataclass.IN,
	294	dns.rdatatype.A,
	295	'192.0.2.1')
	296	response.answer.append(rrset)
	297
	298	for method in ("sendUDPQuery", "sendTCPQuery"):
	299	sender = getattr(self, method)
	300	(receivedQuery, receivedResponse) = sender(query, response)
	301	receivedQuery.id = query.id
	302	self.assertEqual(query, receivedQuery)
	303	self.assertEqual(response, receivedResponse)
	304	self.assertNotEqual(response.answer[0].ttl, receivedResponse.answer[0].ttl)
	305	self.assertEqual(receivedResponse.answer[0].ttl, self._lowttl)
	306
	307	def testLimitTTLFFI(self):
	308	"""
	309	Responses: Alter the TTLs via Limiter
	310	"""
	311	name = 'ffi.min.responses.tests.powerdns.com.'
	312	query = dns.message.make_query(name, 'A', 'IN')
	313	response = dns.message.make_response(query)
	314	rrset = dns.rrset.from_text(name,
	315	3600,
	316	dns.rdataclass.IN,
	317	dns.rdatatype.A,
	318	'192.0.2.1')
	319	response.answer.append(rrset)
	320
	321	for method in ("sendUDPQuery", "sendTCPQuery"):
	322	sender = getattr(self, method)
	323	(receivedQuery, receivedResponse) = sender(query, response)
	324	receivedQuery.id = query.id
	325	self.assertEqual(query, receivedQuery)
	326	self.assertEqual(response, receivedResponse)
327	self.assertNotEqual(response.answer[0].ttl, receivedResponse.answer[0].ttl)
328	self.assertEqual(receivedResponse.answer[0].ttl, self._highttl)
329
330	name = 'ffi.max.responses.tests.powerdns.com.'
331	query = dns.message.make_query(name, 'A', 'IN')
332	response = dns.message.make_response(query)
333	rrset = dns.rrset.from_text(name,
334	3600,
335	dns.rdataclass.IN,
336	dns.rdatatype.A,
337	'192.0.2.1')
338	response.answer.append(rrset)
339
340	for method in ("sendUDPQuery", "sendTCPQuery"):
341	sender = getattr(self, method)
342	(receivedQuery, receivedResponse) = sender(query, response)
343	receivedQuery.id = query.id
344	self.assertEqual(query, receivedQuery)
345	self.assertEqual(response, receivedResponse)
346	self.assertNotEqual(response.answer[0].ttl, receivedResponse.answer[0].ttl)
347	self.assertEqual(receivedResponse.answer[0].ttl, self._lowttl)
348
d55fe43a RG	349	class TestSetReducedTTL(DNSDistTest):
	350
	351	_percentage = 42
	352	_initialTTL = 100
	353	_config_params = ['_percentage', '_testServerPort']
	354	_config_template = """
	355	addResponseAction(AllRule(), SetReducedTTLResponseAction(%d))
	356	newServer{address="127.0.0.1:%s"}
	357	"""
	358
	359	def testLimitTTL(self):
	360	"""
	361	Responses: Reduce TTL to 42%
	362	"""
	363	name = 'reduced-ttl.responses.tests.powerdns.com.'
	364	query = dns.message.make_query(name, 'A', 'IN')
	365	response = dns.message.make_response(query)
	366	rrset = dns.rrset.from_text(name,
	367	self._initialTTL,
	368	dns.rdataclass.IN,
	369	dns.rdatatype.A,
	370	'192.0.2.1')
	371	response.answer.append(rrset)
	372
	373	for method in ("sendUDPQuery", "sendTCPQuery"):
	374	sender = getattr(self, method)
	375	(receivedQuery, receivedResponse) = sender(query, response)
	376	receivedQuery.id = query.id
	377	self.assertEqual(query, receivedQuery)
	378	self.assertEqual(response, receivedResponse)
	379	self.assertNotEqual(response.answer[0].ttl, receivedResponse.answer[0].ttl)
	380	self.assertEqual(receivedResponse.answer[0].ttl, self._percentage)
	381
63cdc73d CHB	382	class TestResponseLuaActionReturnSyntax(DNSDistTest):
	383
	384	_config_template = """
	385	newServer{address="127.0.0.1:%s"}
	386	function customDelay(dr)
	387	return DNSResponseAction.Delay, "1000"
	388	end
	389	function customDrop(dr)
	390	return DNSResponseAction.Drop
	391	end
	392	addResponseAction("drop.responses.tests.powerdns.com.", LuaResponseAction(customDrop))
d3ec24f9	393	addResponseAction(RCodeRule(DNSRCode.NXDOMAIN), LuaResponseAction(customDelay))
63cdc73d CHB	394	"""
	395
	396	def testResponseActionDelayed(self):
	397	"""
	398	Responses: Delayed via LuaResponseAction
	399
	400	Send an A query to "delayed.responses.tests.powerdns.com.",
	401	check that the response delay is longer than 1000 ms
	402	for a NXDomain response over UDP, shorter for a NoError one.
	403	"""
	404	name = 'delayed.responses.tests.powerdns.com.'
	405	query = dns.message.make_query(name, 'A', 'IN')
	406	response = dns.message.make_response(query)
	407
	408	# NX over UDP
	409	response.set_rcode(dns.rcode.NXDOMAIN)
	410	begin = datetime.now()
	411	(receivedQuery, receivedResponse) = self.sendUDPQuery(query, response)
	412	end = datetime.now()
	413	receivedQuery.id = query.id
4bfebc93 CH	414	self.assertEqual(query, receivedQuery)
4bfebc93 CH	415	self.assertEqual(response, receivedResponse)
63cdc73d CHB	416	self.assertTrue((end - begin) > timedelta(0, 1))
	417
	418	def testDropped(self):
	419	"""
	420	Responses: Dropped via user defined LuaResponseAction
	421
	422	Send an A query to "drop.responses.tests.powerdns.com.",
	423	check that the response (not the query) is dropped.
	424	"""
	425	name = 'drop.responses.tests.powerdns.com.'
	426	query = dns.message.make_query(name, 'A', 'IN')
	427	response = dns.message.make_response(query)
	428
6ca2e796 RG	429	for method in ("sendUDPQuery", "sendTCPQuery"):
	430	sender = getattr(self, method)
	431	(receivedQuery, receivedResponse) = sender(query, response)
	432	receivedQuery.id = query.id
4bfebc93 CH	433	self.assertEqual(query, receivedQuery)
4bfebc93 CH	434	self.assertEqual(receivedResponse, None)
eb8d953f	435
eb8d953f CHB	436	class TestResponseClearRecordsType(DNSDistTest):
	437
	438	_config_params = ['_testServerPort']
	439	_config_template = """
	440	local ffi = require("ffi")
	441
	442	function luafct(dr)
	443	ffi.C.dnsdist_ffi_dnsresponse_clear_records_type(dr, DNSQType.AAAA)
	444	return DNSResponseAction.HeaderModify, ""
	445	end
	446
	447	newServer{address="127.0.0.1:%s"}
	448
bb09e03d CHB	449	addResponseAction("ffi.clear-records-type.responses.tests.powerdns.com.", LuaFFIResponseAction(luafct))
bb09e03d CHB	450	addResponseAction("clear-records-type.responses.tests.powerdns.com.", ClearRecordTypesResponseAction(DNSQType.AAAA))
eb8d953f CHB	451	"""
	452
	453	def testClearedFFI(self):
	454	"""
	455	Responses: Removes records of a given type (FFI API)
	456	"""
	457	name = 'ffi.clear-records-type.responses.tests.powerdns.com.'
	458	query = dns.message.make_query(name, 'A', 'IN')
	459	response = dns.message.make_response(query)
bb09e03d CHB	460	expectedResponse = dns.message.make_response(query)
	461	rrset = dns.rrset.from_text(name,
	462	3600,
	463	dns.rdataclass.IN,
	464	dns.rdatatype.A,
	465	'192.0.2.1')
	466	response.answer.append(rrset)
	467	expectedResponse.answer.append(rrset)
	468	rrset = dns.rrset.from_text(name,
	469	3660,
	470	dns.rdataclass.IN,
	471	dns.rdatatype.AAAA,
	472	'2001:DB8::1', '2001:DB8::2')
	473	response.answer.append(rrset)
	474	for method in ("sendUDPQuery", "sendTCPQuery"):
	475	sender = getattr(self, method)
	476	(receivedQuery, receivedResponse) = sender(query, response)
	477	receivedQuery.id = query.id
	478	self.assertEqual(query, receivedQuery)
	479	self.assertEqual(expectedResponse, receivedResponse)
	480
	481	def testCleared(self):
	482	"""
	483	Responses: Removes records of a given type
	484	"""
	485	name = 'clear-records-type.responses.tests.powerdns.com.'
	486	query = dns.message.make_query(name, 'A', 'IN')
	487	response = dns.message.make_response(query)
eb8d953f CHB	488	expectedResponse = dns.message.make_response(query)
	489	rrset = dns.rrset.from_text(name,
	490	3600,
	491	dns.rdataclass.IN,
	492	dns.rdatatype.A,
	493	'192.0.2.1')
	494	response.answer.append(rrset)
	495	expectedResponse.answer.append(rrset)
	496	rrset = dns.rrset.from_text(name,
	497	3660,
	498	dns.rdataclass.IN,
	499	dns.rdatatype.AAAA,
	500	'2001:DB8::1', '2001:DB8::2')
	501	response.answer.append(rrset)
	502	for method in ("sendUDPQuery", "sendTCPQuery"):
	503	sender = getattr(self, method)
	504	(receivedQuery, receivedResponse) = sender(query, response)
	505	receivedQuery.id = query.id
	506	self.assertEqual(query, receivedQuery)
	507	self.assertEqual(expectedResponse, receivedResponse)