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

#!/usr/bin/env python
import threading
import dns
from dnsdisttests import DNSDistTest, pickAvailablePort

class TestTrailingDataToBackend(DNSDistTest):

    # this test suite uses a different responder port
    # because, contrary to the other ones, its
    # responders allow trailing data and we don't want
    # to mix things up.
    _testServerPort = pickAvailablePort()
    _verboseMode = True
    _config_template = """
    newServer{address="127.0.0.1:%s"}

    function replaceTrailingData(dq)
        local success = dq:setTrailingData("ABC")
        if not success then
            return DNSAction.ServFail, ""
        end
        return DNSAction.None, ""
    end
    addAction("added.trailing.tests.powerdns.com.", LuaAction(replaceTrailingData))

    function fillBuffer(dq)
        local available = 4096 - dq.len
        if dq.tcp then
            available = 65535 - dq.len
        end
        local tail = string.rep("A", available)
        local success = dq:setTrailingData(tail)
        if not success then
            return DNSAction.ServFail, ""
        end
        return DNSAction.None, ""
    end
    addAction("max.trailing.tests.powerdns.com.", LuaAction(fillBuffer))

    function exceedBuffer(dq)
        local available = 4096 - dq.len
        if dq.tcp then
            available = 65535 - dq.len
        end
        local tail = string.rep("A", available + 1)
        local success = dq:setTrailingData(tail)
        if not success then
            return DNSAction.ServFail, ""
        end
        return DNSAction.None, ""
    end
    addAction("limited.trailing.tests.powerdns.com.", LuaAction(exceedBuffer))
    """
    @classmethod
    def startResponders(cls):
        print("Launching responders..")

        # Respond REFUSED to queries with trailing data.
        cls._UDPResponder = threading.Thread(name='UDP Responder', target=cls.UDPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue, dns.rcode.REFUSED])
        cls._UDPResponder.daemon = True
        cls._UDPResponder.start()

        # Respond REFUSED to queries with trailing data.
        cls._TCPResponder = threading.Thread(name='TCP Responder', target=cls.TCPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue, dns.rcode.REFUSED])
        cls._TCPResponder.daemon = True
        cls._TCPResponder.start()

    def testTrailingPassthrough(self):
        """
        Trailing data: Pass through

        """
        name = 'passthrough.trailing.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,
                                    '127.0.0.1')
        response.answer.append(rrset)
        expectedResponse = dns.message.make_response(query)
        expectedResponse.set_rcode(dns.rcode.REFUSED)

        raw = query.to_wire()
        raw = raw + b'A'* 20

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

    def testTrailingCapacity(self):
        """
        Trailing data: Fill buffer

        """
        name = 'max.trailing.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,
                                    '127.0.0.1')
        response.answer.append(rrset)
        expectedResponse = dns.message.make_response(query)
        expectedResponse.set_rcode(dns.rcode.REFUSED)

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

    def testTrailingLimited(self):
        """
        Trailing data: Reject buffer overflows

        """
        name = 'limited.trailing.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,
                                    '127.0.0.1')
        response.answer.append(rrset)
        expectedResponse = dns.message.make_response(query)
        expectedResponse.set_rcode(dns.rcode.SERVFAIL)

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (_, receivedResponse) = sender(query, response, useQueue=False)
            self.assertTrue(receivedResponse)
            self.assertEqual(receivedResponse, expectedResponse)

    def testTrailingAdded(self):
        """
        Trailing data: Add

        """
        name = 'added.trailing.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,
                                    '127.0.0.1')
        response.answer.append(rrset)
        expectedResponse = dns.message.make_response(query)
        expectedResponse.set_rcode(dns.rcode.REFUSED)

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

class TestTrailingDataToDnsdist(DNSDistTest):
    _verboseMode = True
    _config_template = """
    newServer{address="127.0.0.1:%s"}

    addAction(AndRule({QNameRule("dropped.trailing.tests.powerdns.com."), TrailingDataRule()}), DropAction())

    function removeTrailingData(dq)
        local success = dq:setTrailingData("")
        if not success then
            print("Trailing removal failed")
            return DNSAction.ServFail, ""
        end
        return DNSAction.None, ""
    end
    addAction("removed.trailing.tests.powerdns.com.", LuaAction(removeTrailingData))

    function reportTrailingData(dq)
        local tail = dq:getTrailingData()
        return DNSAction.Spoof, "-" .. tail .. ".echoed.trailing.tests.powerdns.com."
    end
    addAction("echoed.trailing.tests.powerdns.com.", LuaAction(reportTrailingData))

    function replaceTrailingData(dq)
        local success = dq:setTrailingData("ABC")
        if not success then
            return DNSAction.ServFail, ""
        end
        return DNSAction.None, ""
    end
    addAction("replaced.trailing.tests.powerdns.com.", LuaAction(replaceTrailingData))
    addAction("replaced.trailing.tests.powerdns.com.", LuaAction(reportTrailingData))

    function reportTrailingHex(dq)
        local tail = dq:getTrailingData()
        local hex = string.gsub(tail, ".", function(ch)
            return string.sub(string.format("\\x2502X", string.byte(ch)), -2)
        end)
        return DNSAction.Spoof, "-0x" .. hex .. ".echoed-hex.trailing.tests.powerdns.com."
    end
    addAction("echoed-hex.trailing.tests.powerdns.com.", LuaAction(reportTrailingHex))

    function replaceTrailingData_unsafe(dq)
        local success = dq:setTrailingData("\\xB0\\x00\\xDE\\xADB\\xF0\\x9F\\x91\\xBB\\xC3\\xBE")
        if not success then
            return DNSAction.ServFail, ""
        end
        return DNSAction.None, ""
    end
    addAction("replaced-unsafe.trailing.tests.powerdns.com.", LuaAction(replaceTrailingData_unsafe))
    addAction("replaced-unsafe.trailing.tests.powerdns.com.", LuaAction(reportTrailingHex))
    """

    def testTrailingDropped(self):
        """
        Trailing data: Drop query

        """
        name = 'dropped.trailing.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,
                                    '127.0.0.1')
        response.answer.append(rrset)

        raw = query.to_wire()
        raw = raw + b'A'* 20

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)

            # Verify that queries with no trailing data make it through.
            (receivedQuery, receivedResponse) = sender(query, response)
            self.assertTrue(receivedQuery)
            self.assertTrue(receivedResponse)
            receivedQuery.id = query.id
            self.assertEqual(query, receivedQuery)
            self.assertEqual(response, receivedResponse)

            # Verify that queries with trailing data don't make it through.
            (_, receivedResponse) = sender(raw, response, rawQuery=True, useQueue=False)
            self.assertEqual(receivedResponse, None)

    def testTrailingRemoved(self):
        """
        Trailing data: Remove

        """
        name = 'removed.trailing.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,
                                    '127.0.0.1')
        response.answer.append(rrset)

        raw = query.to_wire()
        raw = raw + b'A'* 20

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

    def testTrailingRead(self):
        """
        Trailing data: Echo

        """
        name = 'echoed.trailing.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)
        response.set_rcode(dns.rcode.SERVFAIL)
        expectedResponse = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    60,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.CNAME,
                                    '-TrailingData.echoed.trailing.tests.powerdns.com.')
        expectedResponse.answer.append(rrset)

        raw = query.to_wire()
        raw = raw + b'TrailingData'

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (_, receivedResponse) = sender(raw, response=None, rawQuery=True, useQueue=False)
            self.assertTrue(receivedResponse)
            expectedResponse.flags = receivedResponse.flags
            self.assertEqual(receivedResponse, expectedResponse)

    def testTrailingReplaced(self):
        """
        Trailing data: Replace

        """
        name = 'replaced.trailing.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)
        response.set_rcode(dns.rcode.SERVFAIL)
        expectedResponse = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    60,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.CNAME,
                                    '-ABC.echoed.trailing.tests.powerdns.com.')
        expectedResponse.answer.append(rrset)

        raw = query.to_wire()
        raw = raw + b'TrailingData'

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (_, receivedResponse) = sender(raw, response=None, rawQuery=True, useQueue=False)
            self.assertTrue(receivedResponse)
            expectedResponse.flags = receivedResponse.flags
            self.assertEqual(receivedResponse, expectedResponse)

    def testTrailingReadUnsafe(self):
        """
        Trailing data: Echo as hex

        """
        name = 'echoed-hex.trailing.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)
        response.set_rcode(dns.rcode.SERVFAIL)
        expectedResponse = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    60,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.CNAME,
                                    '-0x0000DEAD.echoed-hex.trailing.tests.powerdns.com.')
        expectedResponse.answer.append(rrset)

        raw = query.to_wire()
        raw = raw + b'\x00\x00\xDE\xAD'

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (_, receivedResponse) = sender(raw, response=None, rawQuery=True, useQueue=False)
            self.assertTrue(receivedResponse)
            expectedResponse.flags = receivedResponse.flags
            self.assertEqual(receivedResponse, expectedResponse)

    def testTrailingReplacedUnsafe(self):
        """
        Trailing data: Replace with null and/or non-ASCII bytes

        """
        name = 'replaced-unsafe.trailing.tests.powerdns.com.'
        query = dns.message.make_query(name, 'A', 'IN')
        response = dns.message.make_response(query)
        response.set_rcode(dns.rcode.SERVFAIL)
        expectedResponse = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    60,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.CNAME,
                                    '-0xB000DEAD42F09F91BBC3BE.echoed-hex.trailing.tests.powerdns.com.')
        expectedResponse.answer.append(rrset)

        raw = query.to_wire()
        raw = raw + b'TrailingData'

        for method in ("sendUDPQuery", "sendTCPQuery"):
            sender = getattr(self, method)
            (_, receivedResponse) = sender(raw, response=None, rawQuery=True, useQueue=False)
            self.assertTrue(receivedResponse)
            expectedResponse.flags = receivedResponse.flags
            self.assertEqual(receivedResponse, expectedResponse)
Commit	Line	Data
55baa1f2 RG	1	#!/usr/bin/env python
	2	import threading
	3	import dns
630eb526	4	from dnsdisttests import DNSDistTest, pickAvailablePort
55baa1f2	5
4aa08b62	6	class TestTrailingDataToBackend(DNSDistTest):
55baa1f2 RG	7
	8	# this test suite uses a different responder port
	9	# because, contrary to the other ones, its
	10	# responders allow trailing data and we don't want
	11	# to mix things up.
630eb526	12	_testServerPort = pickAvailablePort()
51f07ad4	13	_verboseMode = True
55baa1f2 RG	14	_config_template = """
55baa1f2 RG	15	newServer{address="127.0.0.1:%s"}
93446f2d RG	16
93446f2d RG	17	function replaceTrailingData(dq)
7d243a5e	18	local success = dq:setTrailingData("ABC")
93446f2d RG	19	if not success then
	20	return DNSAction.ServFail, ""
	21	end
	22	return DNSAction.None, ""
	23	end
955de53b	24	addAction("added.trailing.tests.powerdns.com.", LuaAction(replaceTrailingData))
93446f2d RG	25
93446f2d RG	26	function fillBuffer(dq)
f1c712f7 RG	27	local available = 4096 - dq.len
	28	if dq.tcp then
	29	available = 65535 - dq.len
	30	end
7d243a5e	31	local tail = string.rep("A", available)
93446f2d RG	32	local success = dq:setTrailingData(tail)
	33	if not success then
	34	return DNSAction.ServFail, ""
	35	end
	36	return DNSAction.None, ""
	37	end
955de53b	38	addAction("max.trailing.tests.powerdns.com.", LuaAction(fillBuffer))
93446f2d RG	39
93446f2d RG	40	function exceedBuffer(dq)
f1c712f7 RG	41	local available = 4096 - dq.len
	42	if dq.tcp then
	43	available = 65535 - dq.len
	44	end
7d243a5e	45	local tail = string.rep("A", available + 1)
93446f2d RG	46	local success = dq:setTrailingData(tail)
	47	if not success then
	48	return DNSAction.ServFail, ""
	49	end
	50	return DNSAction.None, ""
	51	end
955de53b	52	addAction("limited.trailing.tests.powerdns.com.", LuaAction(exceedBuffer))
55baa1f2 RG	53	"""
	54	@classmethod
	55	def startResponders(cls):
	56	print("Launching responders..")
	57
93446f2d RG	58	# Respond REFUSED to queries with trailing data.
93446f2d RG	59	cls._UDPResponder = threading.Thread(name='UDP Responder', target=cls.UDPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue, dns.rcode.REFUSED])
630eb526	60	cls._UDPResponder.daemon = True
55baa1f2	61	cls._UDPResponder.start()
5df86a8a	62
93446f2d RG	63	# Respond REFUSED to queries with trailing data.
93446f2d RG	64	cls._TCPResponder = threading.Thread(name='TCP Responder', target=cls.TCPResponder, args=[cls._testServerPort, cls._toResponderQueue, cls._fromResponderQueue, dns.rcode.REFUSED])
630eb526	65	cls._TCPResponder.daemon = True
55baa1f2 RG	66	cls._TCPResponder.start()
55baa1f2 RG	67
4aa08b62	68	def testTrailingPassthrough(self):
55baa1f2	69	"""
4aa08b62	70	Trailing data: Pass through
55baa1f2 RG	71
55baa1f2 RG	72	"""
4aa08b62	73	name = 'passthrough.trailing.tests.powerdns.com.'
55baa1f2 RG	74	query = dns.message.make_query(name, 'A', 'IN')
	75	response = dns.message.make_response(query)
	76	rrset = dns.rrset.from_text(name,
	77	3600,
	78	dns.rdataclass.IN,
	79	dns.rdatatype.A,
	80	'127.0.0.1')
	81	response.answer.append(rrset)
4aa08b62	82	expectedResponse = dns.message.make_response(query)
93446f2d	83	expectedResponse.set_rcode(dns.rcode.REFUSED)
55baa1f2 RG	84
55baa1f2 RG	85	raw = query.to_wire()
b4f23783	86	raw = raw + b'A'* 20
55baa1f2	87
4aa08b62 RG	88	for method in ("sendUDPQuery", "sendTCPQuery"):
4aa08b62 RG	89	sender = getattr(self, method)
4aa08b62 RG	90	(receivedQuery, receivedResponse) = sender(raw, response, rawQuery=True)
	91	self.assertTrue(receivedQuery)
	92	self.assertTrue(receivedResponse)
	93	receivedQuery.id = query.id
4bfebc93 CH	94	self.assertEqual(receivedQuery, query)
4bfebc93 CH	95	self.assertEqual(receivedResponse, expectedResponse)
4aa08b62	96
93446f2d RG	97	def testTrailingCapacity(self):
	98	"""
	99	Trailing data: Fill buffer
	100
	101	"""
	102	name = 'max.trailing.tests.powerdns.com.'
	103	query = dns.message.make_query(name, 'A', 'IN')
	104	response = dns.message.make_response(query)
	105	rrset = dns.rrset.from_text(name,
	106	3600,
	107	dns.rdataclass.IN,
	108	dns.rdatatype.A,
	109	'127.0.0.1')
	110	response.answer.append(rrset)
	111	expectedResponse = dns.message.make_response(query)
	112	expectedResponse.set_rcode(dns.rcode.REFUSED)
	113
	114	for method in ("sendUDPQuery", "sendTCPQuery"):
	115	sender = getattr(self, method)
93446f2d RG	116	(receivedQuery, receivedResponse) = sender(query, response)
	117	self.assertTrue(receivedQuery)
	118	self.assertTrue(receivedResponse)
	119	receivedQuery.id = query.id
4bfebc93 CH	120	self.assertEqual(receivedQuery, query)
4bfebc93 CH	121	self.assertEqual(receivedResponse, expectedResponse)
93446f2d RG	122
	123	def testTrailingLimited(self):
	124	"""
	125	Trailing data: Reject buffer overflows
	126
	127	"""
	128	name = 'limited.trailing.tests.powerdns.com.'
	129	query = dns.message.make_query(name, 'A', 'IN')
	130	response = dns.message.make_response(query)
	131	rrset = dns.rrset.from_text(name,
	132	3600,
	133	dns.rdataclass.IN,
	134	dns.rdatatype.A,
	135	'127.0.0.1')
	136	response.answer.append(rrset)
	137	expectedResponse = dns.message.make_response(query)
	138	expectedResponse.set_rcode(dns.rcode.SERVFAIL)
	139
	140	for method in ("sendUDPQuery", "sendTCPQuery"):
	141	sender = getattr(self, method)
90186270	142	(_, receivedResponse) = sender(query, response, useQueue=False)
93446f2d	143	self.assertTrue(receivedResponse)
4bfebc93	144	self.assertEqual(receivedResponse, expectedResponse)
93446f2d RG	145
	146	def testTrailingAdded(self):
	147	"""
	148	Trailing data: Add
	149
	150	"""
	151	name = 'added.trailing.tests.powerdns.com.'
	152	query = dns.message.make_query(name, 'A', 'IN')
	153	response = dns.message.make_response(query)
	154	rrset = dns.rrset.from_text(name,
	155	3600,
	156	dns.rdataclass.IN,
	157	dns.rdatatype.A,
	158	'127.0.0.1')
	159	response.answer.append(rrset)
	160	expectedResponse = dns.message.make_response(query)
	161	expectedResponse.set_rcode(dns.rcode.REFUSED)
	162
	163	for method in ("sendUDPQuery", "sendTCPQuery"):
	164	sender = getattr(self, method)
93446f2d RG	165	(receivedQuery, receivedResponse) = sender(query, response)
	166	self.assertTrue(receivedQuery)
	167	self.assertTrue(receivedResponse)
	168	receivedQuery.id = query.id
4bfebc93 CH	169	self.assertEqual(receivedQuery, query)
4bfebc93 CH	170	self.assertEqual(receivedResponse, expectedResponse)
93446f2d	171
4aa08b62	172	class TestTrailingDataToDnsdist(DNSDistTest):
51f07ad4	173	_verboseMode = True
4aa08b62 RG	174	_config_template = """
4aa08b62 RG	175	newServer{address="127.0.0.1:%s"}
93446f2d RG	176
	177	addAction(AndRule({QNameRule("dropped.trailing.tests.powerdns.com."), TrailingDataRule()}), DropAction())
	178
	179	function removeTrailingData(dq)
7d243a5e	180	local success = dq:setTrailingData("")
93446f2d	181	if not success then
51f07ad4	182	print("Trailing removal failed")
93446f2d RG	183	return DNSAction.ServFail, ""
	184	end
	185	return DNSAction.None, ""
	186	end
955de53b	187	addAction("removed.trailing.tests.powerdns.com.", LuaAction(removeTrailingData))
93446f2d	188
157445b5	189	function reportTrailingData(dq)
7d243a5e RG	190	local tail = dq:getTrailingData()
7d243a5e RG	191	return DNSAction.Spoof, "-" .. tail .. ".echoed.trailing.tests.powerdns.com."
157445b5	192	end
955de53b	193	addAction("echoed.trailing.tests.powerdns.com.", LuaAction(reportTrailingData))
93446f2d RG	194
93446f2d RG	195	function replaceTrailingData(dq)
7d243a5e	196	local success = dq:setTrailingData("ABC")
93446f2d RG	197	if not success then
	198	return DNSAction.ServFail, ""
	199	end
	200	return DNSAction.None, ""
	201	end
955de53b PD	202	addAction("replaced.trailing.tests.powerdns.com.", LuaAction(replaceTrailingData))
955de53b PD	203	addAction("replaced.trailing.tests.powerdns.com.", LuaAction(reportTrailingData))
6b32cb3e RG	204
	205	function reportTrailingHex(dq)
	206	local tail = dq:getTrailingData()
	207	local hex = string.gsub(tail, ".", function(ch)
bf0ff88f	208	return string.sub(string.format("\\x2502X", string.byte(ch)), -2)
6b32cb3e RG	209	end)
	210	return DNSAction.Spoof, "-0x" .. hex .. ".echoed-hex.trailing.tests.powerdns.com."
	211	end
955de53b	212	addAction("echoed-hex.trailing.tests.powerdns.com.", LuaAction(reportTrailingHex))
6b32cb3e RG	213
6b32cb3e RG	214	function replaceTrailingData_unsafe(dq)
bf0ff88f	215	local success = dq:setTrailingData("\\xB0\\x00\\xDE\\xADB\\xF0\\x9F\\x91\\xBB\\xC3\\xBE")
6b32cb3e RG	216	if not success then
	217	return DNSAction.ServFail, ""
	218	end
	219	return DNSAction.None, ""
	220	end
955de53b PD	221	addAction("replaced-unsafe.trailing.tests.powerdns.com.", LuaAction(replaceTrailingData_unsafe))
955de53b PD	222	addAction("replaced-unsafe.trailing.tests.powerdns.com.", LuaAction(reportTrailingHex))
4aa08b62	223	"""
55baa1f2 RG	224
	225	def testTrailingDropped(self):
	226	"""
4aa08b62	227	Trailing data: Drop query
55baa1f2 RG	228
	229	"""
	230	name = 'dropped.trailing.tests.powerdns.com.'
4aa08b62 RG	231	query = dns.message.make_query(name, 'A', 'IN')
	232	response = dns.message.make_response(query)
	233	rrset = dns.rrset.from_text(name,
	234	3600,
	235	dns.rdataclass.IN,
	236	dns.rdatatype.A,
	237	'127.0.0.1')
	238	response.answer.append(rrset)
55baa1f2 RG	239
55baa1f2 RG	240	raw = query.to_wire()
b4f23783	241	raw = raw + b'A'* 20
55baa1f2	242
4aa08b62 RG	243	for method in ("sendUDPQuery", "sendTCPQuery"):
	244	sender = getattr(self, method)
	245
	246	# Verify that queries with no trailing data make it through.
4aa08b62 RG	247	(receivedQuery, receivedResponse) = sender(query, response)
	248	self.assertTrue(receivedQuery)
	249	self.assertTrue(receivedResponse)
	250	receivedQuery.id = query.id
4bfebc93 CH	251	self.assertEqual(query, receivedQuery)
4bfebc93 CH	252	self.assertEqual(response, receivedResponse)
4aa08b62 RG	253
4aa08b62 RG	254	# Verify that queries with trailing data don't make it through.
90186270	255	(_, receivedResponse) = sender(raw, response, rawQuery=True, useQueue=False)
4bfebc93	256	self.assertEqual(receivedResponse, None)
157445b5	257
93446f2d RG	258	def testTrailingRemoved(self):
	259	"""
	260	Trailing data: Remove
	261
	262	"""
	263	name = 'removed.trailing.tests.powerdns.com.'
	264	query = dns.message.make_query(name, 'A', 'IN')
	265	response = dns.message.make_response(query)
	266	rrset = dns.rrset.from_text(name,
	267	3600,
	268	dns.rdataclass.IN,
	269	dns.rdatatype.A,
	270	'127.0.0.1')
	271	response.answer.append(rrset)
	272
	273	raw = query.to_wire()
	274	raw = raw + b'A'* 20
	275
	276	for method in ("sendUDPQuery", "sendTCPQuery"):
	277	sender = getattr(self, method)
93446f2d RG	278	(receivedQuery, receivedResponse) = sender(raw, response, rawQuery=True)
	279	self.assertTrue(receivedQuery)
	280	self.assertTrue(receivedResponse)
	281	receivedQuery.id = query.id
4bfebc93 CH	282	self.assertEqual(receivedQuery, query)
4bfebc93 CH	283	self.assertEqual(receivedResponse, response)
93446f2d	284
157445b5 RG	285	def testTrailingRead(self):
157445b5 RG	286	"""
6b32cb3e	287	Trailing data: Echo
157445b5 RG	288
	289	"""
	290	name = 'echoed.trailing.tests.powerdns.com.'
	291	query = dns.message.make_query(name, 'A', 'IN')
	292	response = dns.message.make_response(query)
	293	response.set_rcode(dns.rcode.SERVFAIL)
	294	expectedResponse = dns.message.make_response(query)
	295	rrset = dns.rrset.from_text(name,
	296	60,
	297	dns.rdataclass.IN,
	298	dns.rdatatype.CNAME,
93446f2d RG	299	'-TrailingData.echoed.trailing.tests.powerdns.com.')
	300	expectedResponse.answer.append(rrset)
	301
	302	raw = query.to_wire()
	303	raw = raw + b'TrailingData'
	304
	305	for method in ("sendUDPQuery", "sendTCPQuery"):
	306	sender = getattr(self, method)
90186270	307	(_, receivedResponse) = sender(raw, response=None, rawQuery=True, useQueue=False)
93446f2d RG	308	self.assertTrue(receivedResponse)
93446f2d RG	309	expectedResponse.flags = receivedResponse.flags
4bfebc93	310	self.assertEqual(receivedResponse, expectedResponse)
93446f2d RG	311
	312	def testTrailingReplaced(self):
	313	"""
	314	Trailing data: Replace
	315
	316	"""
	317	name = 'replaced.trailing.tests.powerdns.com.'
	318	query = dns.message.make_query(name, 'A', 'IN')
	319	response = dns.message.make_response(query)
	320	response.set_rcode(dns.rcode.SERVFAIL)
	321	expectedResponse = dns.message.make_response(query)
	322	rrset = dns.rrset.from_text(name,
	323	60,
	324	dns.rdataclass.IN,
	325	dns.rdatatype.CNAME,
	326	'-ABC.echoed.trailing.tests.powerdns.com.')
157445b5 RG	327	expectedResponse.answer.append(rrset)
	328
	329	raw = query.to_wire()
	330	raw = raw + b'TrailingData'
	331
	332	for method in ("sendUDPQuery", "sendTCPQuery"):
	333	sender = getattr(self, method)
90186270	334	(_, receivedResponse) = sender(raw, response=None, rawQuery=True, useQueue=False)
157445b5 RG	335	self.assertTrue(receivedResponse)
157445b5 RG	336	expectedResponse.flags = receivedResponse.flags
4bfebc93	337	self.assertEqual(receivedResponse, expectedResponse)
6b32cb3e RG	338
	339	def testTrailingReadUnsafe(self):
	340	"""
	341	Trailing data: Echo as hex
	342
	343	"""
	344	name = 'echoed-hex.trailing.tests.powerdns.com.'
	345	query = dns.message.make_query(name, 'A', 'IN')
	346	response = dns.message.make_response(query)
	347	response.set_rcode(dns.rcode.SERVFAIL)
	348	expectedResponse = dns.message.make_response(query)
	349	rrset = dns.rrset.from_text(name,
	350	60,
	351	dns.rdataclass.IN,
	352	dns.rdatatype.CNAME,
	353	'-0x0000DEAD.echoed-hex.trailing.tests.powerdns.com.')
	354	expectedResponse.answer.append(rrset)
	355
	356	raw = query.to_wire()
	357	raw = raw + b'\x00\x00\xDE\xAD'
	358
	359	for method in ("sendUDPQuery", "sendTCPQuery"):
	360	sender = getattr(self, method)
90186270	361	(_, receivedResponse) = sender(raw, response=None, rawQuery=True, useQueue=False)
6b32cb3e RG	362	self.assertTrue(receivedResponse)
6b32cb3e RG	363	expectedResponse.flags = receivedResponse.flags
4bfebc93	364	self.assertEqual(receivedResponse, expectedResponse)
6b32cb3e RG	365
	366	def testTrailingReplacedUnsafe(self):
	367	"""
	368	Trailing data: Replace with null and/or non-ASCII bytes
	369
	370	"""
	371	name = 'replaced-unsafe.trailing.tests.powerdns.com.'
	372	query = dns.message.make_query(name, 'A', 'IN')
	373	response = dns.message.make_response(query)
	374	response.set_rcode(dns.rcode.SERVFAIL)
	375	expectedResponse = dns.message.make_response(query)
	376	rrset = dns.rrset.from_text(name,
	377	60,
	378	dns.rdataclass.IN,
	379	dns.rdatatype.CNAME,
bf0ff88f	380	'-0xB000DEAD42F09F91BBC3BE.echoed-hex.trailing.tests.powerdns.com.')
6b32cb3e RG	381	expectedResponse.answer.append(rrset)
	382
	383	raw = query.to_wire()
	384	raw = raw + b'TrailingData'
	385
	386	for method in ("sendUDPQuery", "sendTCPQuery"):
	387	sender = getattr(self, method)
90186270	388	(_, receivedResponse) = sender(raw, response=None, rawQuery=True, useQueue=False)
6b32cb3e RG	389	self.assertTrue(receivedResponse)
6b32cb3e RG	390	expectedResponse.flags = receivedResponse.flags
4bfebc93	391	self.assertEqual(receivedResponse, expectedResponse)