[thirdparty/pdns.git] / regression-tests.recursor-dnssec / test_AggressiveNSECCache.py

import dns
from recursortests import RecursorTest
import os
import requests
import subprocess
import time
import extendederrors

class AggressiveNSECCacheBase(RecursorTest):
    __test__ = False
    _wsPort = 8042
    _wsTimeout = 10
    _wsPassword = 'secretpassword'
    _apiKey = 'secretapikey'
    #_recursorStartupDelay = 4.0
    _config_template = """
    dnssec=validate
    aggressive-nsec-cache-size=10000
    aggressive-cache-max-nsec3-hash-cost=204
    nsec3-max-iterations=150
    webserver=yes
    webserver-port=%d
    webserver-address=127.0.0.1
    webserver-password=%s
    api-key=%s
    devonly-regression-test-mode
    extended-resolution-errors=yes
    """ % (_wsPort, _wsPassword, _apiKey)

    @classmethod
    def wipe(cls):
        confdir = os.path.join('configs', cls._confdir)
        # Only wipe examples, as wiping the root triggers root NS refreshes
        cls.wipeRecursorCache(confdir, "example$")

    def getMetric(self, name):
        headers = {'x-api-key': self._apiKey}
        url = 'http://127.0.0.1:' + str(self._wsPort) + '/api/v1/servers/localhost/statistics'
        r = requests.get(url, headers=headers, timeout=self._wsTimeout)
        self.assertTrue(r)
        self.assertEqual(r.status_code, 200)
        self.assertTrue(r.json())
        content = r.json()

        for entry in content:
            if entry['name'] == name:
                return int(entry['value'])

        self.assertTrue(False)

    def testNoEDE(self):
        # This isn't an aggresive cache check, but the strcuture is very similar to the others,
        # so letys place it here.
        # It test the issue that an intermediate EDE does not get reported with the final answer
        # https://github.com/PowerDNS/pdns/pull/12694
        self.wipe()

        res = self.sendQuery('host1.sub.secure.example.', 'TXT')
        self.assertRcodeEqual(res, dns.rcode.NOERROR)
        self.assertAnswerEmpty(res)
        self.assertAuthorityHasSOA(res)
        self.assertMessageIsAuthenticated(res)
        self.assertEqual(res.edns, 0)
        self.assertEqual(len(res.options), 0)

        res = self.sendQuery('host1.sub.secure.example.', 'A')
        self.assertRcodeEqual(res, dns.rcode.NOERROR)
        self.assertMessageIsAuthenticated(res)
        self.assertEqual(res.edns, 0)
        self.assertEqual(len(res.options), 0)

    def testNoData(self):
        self.wipe()

        # first we query a nonexistent type, to get the NSEC in our cache
        entries = self.getMetric('aggressive-nsec-cache-entries')
        res = self.sendQuery('host1.secure.example.', 'TXT')
        self.assertRcodeEqual(res, dns.rcode.NOERROR)
        self.assertAnswerEmpty(res)
        self.assertAuthorityHasSOA(res)
        self.assertMessageIsAuthenticated(res)
        self.assertGreater(self.getMetric('aggressive-nsec-cache-entries'), entries)

        # now we ask for a different type, we should generate the answer from the NSEC,
        # and no outgoing query should be made
        nbQueries = self.getMetric('all-outqueries')
        entries = self.getMetric('aggressive-nsec-cache-entries')
        res = self.sendQuery('host1.secure.example.', 'AAAA')
        self.assertRcodeEqual(res, dns.rcode.NOERROR)
        self.assertAnswerEmpty(res)
        self.assertAuthorityHasSOA(res)
        self.assertMessageIsAuthenticated(res)
        self.assertEqual(nbQueries, self.getMetric('all-outqueries'))
        self.assertEqual(self.getMetric('aggressive-nsec-cache-entries'), entries)
        self.assertEqual(res.edns, 0)
        self.assertEqual(len(res.options), 1)
        self.assertEqual(res.options[0].otype, 15)
        self.assertEqual(res.options[0], extendederrors.ExtendedErrorOption(29, b'Result synthesized from aggressive NSEC cache (RFC8198)'))

class AggressiveNSECCacheNSEC(AggressiveNSECCacheBase):
    _confdir = 'AggressiveNSECCacheNSEC'
    __test__ = True

    # we can't use the same tests for NSEC and NSEC3 because the hashed NSEC3s
    # do not deny the same names than the non-hashed NSECs do
    def testNXD(self):
        self.wipe()

        # first we query a nonexistent name, to get the needed NSECs (name + widcard) in our cache
        entries = self.getMetric('aggressive-nsec-cache-entries')
        hits = self.getMetric('aggressive-nsec-cache-nsec-hits')
        res = self.sendQuery('host2.secure.example.', 'TXT')
        self.assertRcodeEqual(res, dns.rcode.NXDOMAIN)
        self.assertAnswerEmpty(res)
        self.assertAuthorityHasSOA(res)
        self.assertMessageIsAuthenticated(res)
        self.assertGreater(self.getMetric('aggressive-nsec-cache-entries'), entries)
        self.assertEqual(self.getMetric('aggressive-nsec-cache-nsec-hits'), hits)

        # now we ask for a different name that is covered by the NSEC,
        # we should generate the answer from the NSEC and no outgoing query should be made
        nbQueries = self.getMetric('all-outqueries')
        entries = self.getMetric('aggressive-nsec-cache-entries')
        hits = self.getMetric('aggressive-nsec-cache-nsec-hits')
        res = self.sendQuery('host3.secure.example.', 'AAAA')
        self.assertRcodeEqual(res, dns.rcode.NXDOMAIN)
        self.assertAnswerEmpty(res)
        self.assertAuthorityHasSOA(res)
        self.assertMessageIsAuthenticated(res)
        self.assertEqual(nbQueries, self.getMetric('all-outqueries'))
        self.assertEqual(self.getMetric('aggressive-nsec-cache-entries'), entries)
        self.assertGreater(self.getMetric('aggressive-nsec-cache-nsec-hits'), hits)
        self.assertEqual(res.edns, 0)
        self.assertEqual(len(res.options), 1)
        self.assertEqual(res.options[0].otype, 15)
        self.assertEqual(res.options[0], extendederrors.ExtendedErrorOption(29, b'Result synthesized from aggressive NSEC cache (RFC8198)'))

    def testWildcard(self):
        self.wipe()

        # first we query a nonexistent name, but for which a wildcard matches,
        # to get the NSEC in our cache
        res = self.sendQuery('test1.wildcard.secure.example.', 'A')
        expected = dns.rrset.from_text('test1.wildcard.secure.example.', 0, dns.rdataclass.IN, 'A', '{prefix}.10'.format(prefix=self._PREFIX))
        self.assertRcodeEqual(res, dns.rcode.NOERROR)
        self.assertMatchingRRSIGInAnswer(res, expected)
        self.assertMessageIsAuthenticated(res)

        # now we ask for a different name, we should generate the answer from the NSEC and the wildcard,
        # and no outgoing query should be made
        hits = self.getMetric('aggressive-nsec-cache-nsec-wc-hits')
        nbQueries = self.getMetric('all-outqueries')
        res = self.sendQuery('test2.wildcard.secure.example.', 'A')
        expected = dns.rrset.from_text('test2.wildcard.secure.example.', 0, dns.rdataclass.IN, 'A', '{prefix}.10'.format(prefix=self._PREFIX))
        self.assertRcodeEqual(res, dns.rcode.NOERROR)
        self.assertMatchingRRSIGInAnswer(res, expected)
        self.assertMessageIsAuthenticated(res)
        self.assertEqual(nbQueries, self.getMetric('all-outqueries'))
        self.assertGreater(self.getMetric('aggressive-nsec-cache-nsec-wc-hits'), hits)
        self.assertEqual(res.edns, 0)
        self.assertEqual(len(res.options), 1)
        self.assertEqual(res.options[0].otype, 15)
        self.assertEqual(res.options[0], extendederrors.ExtendedErrorOption(29, b'Result synthesized from aggressive NSEC cache (RFC8198)'))

        # now we ask for a type that does not exist at the wildcard
        hits = self.getMetric('aggressive-nsec-cache-nsec-hits')
        nbQueries = self.getMetric('all-outqueries')
        res = self.sendQuery('test1.wildcard.secure.example.', 'AAAA')
        self.assertRcodeEqual(res, dns.rcode.NOERROR)
        self.assertAnswerEmpty(res)
        self.assertAuthorityHasSOA(res)
        self.assertMessageIsAuthenticated(res)
        self.assertEqual(nbQueries, self.getMetric('all-outqueries'))
        self.assertGreater(self.getMetric('aggressive-nsec-cache-nsec-hits'), hits)
        self.assertEqual(res.edns, 0)
        self.assertEqual(len(res.options), 1)
        self.assertEqual(res.options[0].otype, 15)
        self.assertEqual(res.options[0], extendederrors.ExtendedErrorOption(29, b'Result synthesized from aggressive NSEC cache (RFC8198)'))

        # we can also ask a different type, for a different name that is covered
        # by the NSEC and matches the wildcard (but the type does not exist)
        hits = self.getMetric('aggressive-nsec-cache-nsec-wc-hits')
        nbQueries = self.getMetric('all-outqueries')
        res = self.sendQuery('test3.wildcard.secure.example.', 'TXT')
        self.assertRcodeEqual(res, dns.rcode.NOERROR)
        self.assertAnswerEmpty(res)
        self.assertAuthorityHasSOA(res)
        self.assertMessageIsAuthenticated(res)
        self.assertEqual(nbQueries, self.getMetric('all-outqueries'))
        self.assertGreater(self.getMetric('aggressive-nsec-cache-nsec-hits'), hits)
        self.assertEqual(res.edns, 0)
        self.assertEqual(len(res.options), 1)
        self.assertEqual(res.options[0].otype, 15)
        self.assertEqual(res.options[0], extendederrors.ExtendedErrorOption(29, b'Result synthesized from aggressive NSEC cache (RFC8198)'))

    def test_Bogus(self):
        self.wipe()

        # query a name in example to fill the aggressive negcache
        res = self.sendQuery('example.', 'A')
        self.assertRcodeEqual(res, dns.rcode.NOERROR)
        self.assertAnswerEmpty(res)
        self.assertEqual(1, self.getMetric('aggressive-nsec-cache-entries'))

        # query a name in a Bogus zone
        res = self.sendQuery('ted1.bogus.example.', 'A')
        self.assertRcodeEqual(res, dns.rcode.SERVFAIL)
        self.assertAnswerEmpty(res)

        # disable validation
        msg = dns.message.make_query('ted1.bogus.example.', 'A', want_dnssec=True)
        msg.flags |= dns.flags.CD

        res = self.sendUDPQuery(msg)
        self.assertRcodeEqual(res, dns.rcode.NXDOMAIN)
        self.assertAnswerEmpty(res)
        self.assertAuthorityHasSOA(res)

        # check that we _do not_ use the aggressive NSEC cache
        nbQueries = self.getMetric('all-outqueries')
        msg = dns.message.make_query('ted2.bogus.example.', 'A', want_dnssec=True)
        msg.flags |= dns.flags.CD

        res = self.sendUDPQuery(msg)
        self.assertRcodeEqual(res, dns.rcode.NXDOMAIN)
        self.assertAnswerEmpty(res)
        self.assertAuthorityHasSOA(res)
        self.assertGreater(self.getMetric('all-outqueries'), nbQueries)

        # Check that we stil have one aggressive cache entry
        self.assertEqual(1, self.getMetric('aggressive-nsec-cache-entries'))
        print(res.options)
        self.assertEqual(res.edns, 0)
        self.assertEqual(len(res.options), 1)
        self.assertEqual(res.options[0].otype, 15)
        self.assertEqual(res.options[0], extendederrors.ExtendedErrorOption(9, b''))

class AggressiveNSECCacheNSEC3(AggressiveNSECCacheBase):
    _confdir = 'AggressiveNSECCacheNSEC3'
    __test__ = True

    @classmethod
    def secureZone(cls, confdir, zonename, key=None):
        zone = '.' if zonename == 'ROOT' else zonename
        if not key:
            pdnsutilCmd = [os.environ['PDNSUTIL'],
                           '--config-dir=%s' % confdir,
                           'secure-zone',
                           zone]
        else:
            keyfile = os.path.join(confdir, 'dnssec.key')
            with open(keyfile, 'w') as fdKeyfile:
                fdKeyfile.write(key)

            pdnsutilCmd = [os.environ['PDNSUTIL'],
                           '--config-dir=%s' % confdir,
                           'import-zone-key',
                           zone,
                           keyfile,
                           'active',
                           'ksk']

        print(' '.join(pdnsutilCmd))
        try:
            subprocess.check_output(pdnsutilCmd, stderr=subprocess.STDOUT)
        except subprocess.CalledProcessError as e:
            raise AssertionError('%s failed (%d): %s' % (pdnsutilCmd, e.returncode, e.output))

        params = "1 0 100 AABBCCDDEEFF112233"

        if zone == "optout.example":
            params = "1 1 100 AABBCCDDEEFF112233"

        pdnsutilCmd = [os.environ['PDNSUTIL'],
                       '--config-dir=%s' % confdir,
                       'set-nsec3',
                       zone,
                       params]

        print(' '.join(pdnsutilCmd))
        try:
            subprocess.check_output(pdnsutilCmd, stderr=subprocess.STDOUT)
        except subprocess.CalledProcessError as e:
            raise AssertionError('%s failed (%d): %s' % (pdnsutilCmd, e.returncode, e.output))

    def testNXD(self):
        self.wipe()

        # first we query a nonexistent name, to get the needed NSEC3s in our cache
        res = self.sendQuery('host2.secure.example.', 'TXT')
        self.assertRcodeEqual(res, dns.rcode.NXDOMAIN)
        self.assertAnswerEmpty(res)
        self.assertAuthorityHasSOA(res)
        self.assertMessageIsAuthenticated(res)

        # now we ask for a different name that is covered by the NSEC3s,
        # we should generate the answer from the NSEC3s and no outgoing query should be made
        nbQueries = self.getMetric('all-outqueries')
        res = self.sendQuery('host6.secure.example.', 'AAAA')
        self.assertRcodeEqual(res, dns.rcode.NXDOMAIN)
        self.assertAnswerEmpty(res)
        self.assertAuthorityHasSOA(res)
        self.assertMessageIsAuthenticated(res)
        self.assertEqual(nbQueries, self.getMetric('all-outqueries'))
        self.assertEqual(res.edns, 0)
        self.assertEqual(len(res.options), 1)
        self.assertEqual(res.options[0].otype, 15)
        self.assertEqual(res.options[0], extendederrors.ExtendedErrorOption(29, b'Result synthesized from aggressive NSEC cache (RFC8198)'))

    def testWildcard(self):
        self.wipe()

        # first let's get the SOA and wildcard NSEC in our cache by asking a name that matches the wildcard
        # but a type that does not exist
        res = self.sendQuery('test1.wildcard.secure.example.', 'AAAA')
        self.assertRcodeEqual(res, dns.rcode.NOERROR)
        self.assertAnswerEmpty(res)
        self.assertAuthorityHasSOA(res)
        self.assertMessageIsAuthenticated(res)

        # we query a nonexistent name, but for which a wildcard matches,
        # to get the NSEC3 in our cache
        res = self.sendQuery('test5.wildcard.secure.example.', 'A')
        expected = dns.rrset.from_text('test5.wildcard.secure.example.', 0, dns.rdataclass.IN, 'A', '{prefix}.10'.format(prefix=self._PREFIX))
        self.assertRcodeEqual(res, dns.rcode.NOERROR)
        self.assertMatchingRRSIGInAnswer(res, expected)
        self.assertMessageIsAuthenticated(res)

        # now we ask for a different name, we should generate the answer from the NSEC3s and the wildcard,
        # and no outgoing query should be made
        nbQueries = self.getMetric('all-outqueries')
        res = self.sendQuery('test6.wildcard.secure.example.', 'A')
        expected = dns.rrset.from_text('test6.wildcard.secure.example.', 0, dns.rdataclass.IN, 'A', '{prefix}.10'.format(prefix=self._PREFIX))
        self.assertRcodeEqual(res, dns.rcode.NOERROR)
        self.assertMatchingRRSIGInAnswer(res, expected)
        self.assertMessageIsAuthenticated(res)
        self.assertEqual(nbQueries, self.getMetric('all-outqueries'))
        self.assertEqual(res.edns, 0)
        self.assertEqual(len(res.options), 1)
        self.assertEqual(res.options[0].otype, 15)
        self.assertEqual(res.options[0], extendederrors.ExtendedErrorOption(29, b'Result synthesized from aggressive NSEC cache (RFC8198)'))

        # now we ask for a type that does not exist at the wildcard
        nbQueries = self.getMetric('all-outqueries')
        res = self.sendQuery('test5.wildcard.secure.example.', 'AAAA')
        self.assertRcodeEqual(res, dns.rcode.NOERROR)
        self.assertAnswerEmpty(res)
        self.assertAuthorityHasSOA(res)
        self.assertMessageIsAuthenticated(res)
        self.assertEqual(nbQueries, self.getMetric('all-outqueries'))
        self.assertEqual(res.edns, 0)
        self.assertEqual(len(res.options), 1)
        self.assertEqual(res.options[0].otype, 15)
        self.assertEqual(res.options[0], extendederrors.ExtendedErrorOption(29, b'Result synthesized from aggressive NSEC cache (RFC8198)'))

        # we can also ask a different type, for a different name that is covered
        # by the NSEC3s and matches the wildcard (but the type does not exist)
        nbQueries = self.getMetric('all-outqueries')
        res = self.sendQuery('test6.wildcard.secure.example.', 'TXT')
        self.assertRcodeEqual(res, dns.rcode.NOERROR)
        self.assertAnswerEmpty(res)
        self.assertAuthorityHasSOA(res)
        self.assertMessageIsAuthenticated(res)
        self.assertEqual(nbQueries, self.getMetric('all-outqueries'))
        self.assertEqual(res.edns, 0)
        self.assertEqual(len(res.options), 1)
        self.assertEqual(res.options[0].otype, 15)
        self.assertEqual(res.options[0], extendederrors.ExtendedErrorOption(29, b'Result synthesized from aggressive NSEC cache (RFC8198)'))

    def test_OptOut(self):
        self.wipe()

        # query a name in an opt-out zone
        res = self.sendQuery('ns2.optout.example.', 'A')
        self.assertRcodeEqual(res, dns.rcode.NXDOMAIN)
        self.assertAnswerEmpty(res)
        self.assertAuthorityHasSOA(res)

        # check that we _do not_ use the aggressive NSEC cache
        nbQueries = self.getMetric('all-outqueries')
        res = self.sendQuery('ns3.optout.example.', 'A')
        self.assertRcodeEqual(res, dns.rcode.NXDOMAIN)
        self.assertAnswerEmpty(res)
        self.assertAuthorityHasSOA(res)
        self.assertGreater(self.getMetric('all-outqueries'), nbQueries)
        self.assertEqual(res.edns, 0)
        self.assertEqual(len(res.options), 0)
Commit	Line	Data
dd54e182 RG	1	import dns
	2	from recursortests import RecursorTest
	3	import os
	4	import requests
	5	import subprocess
2ec80d48	6	import time
811bddf9	7	import extendederrors
dd54e182 RG	8
	9	class AggressiveNSECCacheBase(RecursorTest):
	10	__test__ = False
	11	_wsPort = 8042
40795092	12	_wsTimeout = 10
dd54e182 RG	13	_wsPassword = 'secretpassword'
dd54e182 RG	14	_apiKey = 'secretapikey'
7d3d2f4f	15	#_recursorStartupDelay = 4.0
dd54e182 RG	16	_config_template = """
	17	dnssec=validate
	18	aggressive-nsec-cache-size=10000
15e973d6	19	aggressive-cache-max-nsec3-hash-cost=204
b199e480	20	nsec3-max-iterations=150
dd54e182 RG	21	webserver=yes
	22	webserver-port=%d
	23	webserver-address=127.0.0.1
	24	webserver-password=%s
	25	api-key=%s
7d3d2f4f	26	devonly-regression-test-mode
811bddf9	27	extended-resolution-errors=yes
dd54e182 RG	28	""" % (_wsPort, _wsPassword, _apiKey)
	29
	30	@classmethod
2ec80d48	31	def wipe(cls):
dd54e182	32	confdir = os.path.join('configs', cls._confdir)
7d3d2f4f OM	33	# Only wipe examples, as wiping the root triggers root NS refreshes
7d3d2f4f OM	34	cls.wipeRecursorCache(confdir, "example$")
dd54e182 RG	35
	36	def getMetric(self, name):
	37	headers = {'x-api-key': self._apiKey}
	38	url = 'http://127.0.0.1:' + str(self._wsPort) + '/api/v1/servers/localhost/statistics'
	39	r = requests.get(url, headers=headers, timeout=self._wsTimeout)
	40	self.assertTrue(r)
4bfebc93	41	self.assertEqual(r.status_code, 200)
dd54e182 RG	42	self.assertTrue(r.json())
	43	content = r.json()
	44
	45	for entry in content:
	46	if entry['name'] == name:
	47	return int(entry['value'])
	48
	49	self.assertTrue(False)
	50
03f779fd OM	51	def testNoEDE(self):
	52	# This isn't an aggresive cache check, but the strcuture is very similar to the others,
	53	# so letys place it here.
	54	# It test the issue that an intermediate EDE does not get reported with the final answer
	55	# https://github.com/PowerDNS/pdns/pull/12694
	56	self.wipe()
	57
	58	res = self.sendQuery('host1.sub.secure.example.', 'TXT')
	59	self.assertRcodeEqual(res, dns.rcode.NOERROR)
	60	self.assertAnswerEmpty(res)
	61	self.assertAuthorityHasSOA(res)
	62	self.assertMessageIsAuthenticated(res)
	63	self.assertEqual(res.edns, 0)
	64	self.assertEqual(len(res.options), 0)
	65
	66	res = self.sendQuery('host1.sub.secure.example.', 'A')
	67	self.assertRcodeEqual(res, dns.rcode.NOERROR)
	68	self.assertMessageIsAuthenticated(res)
	69	self.assertEqual(res.edns, 0)
	70	self.assertEqual(len(res.options), 0)
	71
dd54e182	72	def testNoData(self):
2ec80d48	73	self.wipe()
7d3d2f4f	74
e746a2f6	75	# first we query a nonexistent type, to get the NSEC in our cache
dd54e182 RG	76	entries = self.getMetric('aggressive-nsec-cache-entries')
	77	res = self.sendQuery('host1.secure.example.', 'TXT')
	78	self.assertRcodeEqual(res, dns.rcode.NOERROR)
	79	self.assertAnswerEmpty(res)
	80	self.assertAuthorityHasSOA(res)
	81	self.assertMessageIsAuthenticated(res)
	82	self.assertGreater(self.getMetric('aggressive-nsec-cache-entries'), entries)
	83
	84	# now we ask for a different type, we should generate the answer from the NSEC,
	85	# and no outgoing query should be made
	86	nbQueries = self.getMetric('all-outqueries')
	87	entries = self.getMetric('aggressive-nsec-cache-entries')
	88	res = self.sendQuery('host1.secure.example.', 'AAAA')
	89	self.assertRcodeEqual(res, dns.rcode.NOERROR)
	90	self.assertAnswerEmpty(res)
	91	self.assertAuthorityHasSOA(res)
	92	self.assertMessageIsAuthenticated(res)
4bfebc93 CH	93	self.assertEqual(nbQueries, self.getMetric('all-outqueries'))
4bfebc93 CH	94	self.assertEqual(self.getMetric('aggressive-nsec-cache-entries'), entries)
811bddf9 OM	95	self.assertEqual(res.edns, 0)
	96	self.assertEqual(len(res.options), 1)
	97	self.assertEqual(res.options[0].otype, 15)
7f8f0893	98	self.assertEqual(res.options[0], extendederrors.ExtendedErrorOption(29, b'Result synthesized from aggressive NSEC cache (RFC8198)'))
dd54e182 RG	99
	100	class AggressiveNSECCacheNSEC(AggressiveNSECCacheBase):
	101	_confdir = 'AggressiveNSECCacheNSEC'
	102	__test__ = True
	103
	104	# we can't use the same tests for NSEC and NSEC3 because the hashed NSEC3s
	105	# do not deny the same names than the non-hashed NSECs do
	106	def testNXD(self):
2ec80d48	107	self.wipe()
7d3d2f4f	108
e746a2f6	109	# first we query a nonexistent name, to get the needed NSECs (name + widcard) in our cache
dd54e182 RG	110	entries = self.getMetric('aggressive-nsec-cache-entries')
	111	hits = self.getMetric('aggressive-nsec-cache-nsec-hits')
	112	res = self.sendQuery('host2.secure.example.', 'TXT')
	113	self.assertRcodeEqual(res, dns.rcode.NXDOMAIN)
	114	self.assertAnswerEmpty(res)
	115	self.assertAuthorityHasSOA(res)
	116	self.assertMessageIsAuthenticated(res)
	117	self.assertGreater(self.getMetric('aggressive-nsec-cache-entries'), entries)
4bfebc93	118	self.assertEqual(self.getMetric('aggressive-nsec-cache-nsec-hits'), hits)
dd54e182 RG	119
	120	# now we ask for a different name that is covered by the NSEC,
	121	# we should generate the answer from the NSEC and no outgoing query should be made
	122	nbQueries = self.getMetric('all-outqueries')
	123	entries = self.getMetric('aggressive-nsec-cache-entries')
	124	hits = self.getMetric('aggressive-nsec-cache-nsec-hits')
	125	res = self.sendQuery('host3.secure.example.', 'AAAA')
	126	self.assertRcodeEqual(res, dns.rcode.NXDOMAIN)
	127	self.assertAnswerEmpty(res)
	128	self.assertAuthorityHasSOA(res)
	129	self.assertMessageIsAuthenticated(res)
4bfebc93 CH	130	self.assertEqual(nbQueries, self.getMetric('all-outqueries'))
4bfebc93 CH	131	self.assertEqual(self.getMetric('aggressive-nsec-cache-entries'), entries)
dd54e182	132	self.assertGreater(self.getMetric('aggressive-nsec-cache-nsec-hits'), hits)
811bddf9 OM	133	self.assertEqual(res.edns, 0)
	134	self.assertEqual(len(res.options), 1)
	135	self.assertEqual(res.options[0].otype, 15)
7f8f0893	136	self.assertEqual(res.options[0], extendederrors.ExtendedErrorOption(29, b'Result synthesized from aggressive NSEC cache (RFC8198)'))
dd54e182 RG	137
dd54e182 RG	138	def testWildcard(self):
2ec80d48	139	self.wipe()
7d3d2f4f	140
e746a2f6	141	# first we query a nonexistent name, but for which a wildcard matches,
dd54e182 RG	142	# to get the NSEC in our cache
	143	res = self.sendQuery('test1.wildcard.secure.example.', 'A')
	144	expected = dns.rrset.from_text('test1.wildcard.secure.example.', 0, dns.rdataclass.IN, 'A', '{prefix}.10'.format(prefix=self._PREFIX))
	145	self.assertRcodeEqual(res, dns.rcode.NOERROR)
	146	self.assertMatchingRRSIGInAnswer(res, expected)
	147	self.assertMessageIsAuthenticated(res)
	148
	149	# now we ask for a different name, we should generate the answer from the NSEC and the wildcard,
	150	# and no outgoing query should be made
	151	hits = self.getMetric('aggressive-nsec-cache-nsec-wc-hits')
	152	nbQueries = self.getMetric('all-outqueries')
	153	res = self.sendQuery('test2.wildcard.secure.example.', 'A')
	154	expected = dns.rrset.from_text('test2.wildcard.secure.example.', 0, dns.rdataclass.IN, 'A', '{prefix}.10'.format(prefix=self._PREFIX))
	155	self.assertRcodeEqual(res, dns.rcode.NOERROR)
	156	self.assertMatchingRRSIGInAnswer(res, expected)
	157	self.assertMessageIsAuthenticated(res)
4bfebc93	158	self.assertEqual(nbQueries, self.getMetric('all-outqueries'))
dd54e182	159	self.assertGreater(self.getMetric('aggressive-nsec-cache-nsec-wc-hits'), hits)
811bddf9 OM	160	self.assertEqual(res.edns, 0)
	161	self.assertEqual(len(res.options), 1)
	162	self.assertEqual(res.options[0].otype, 15)
7f8f0893	163	self.assertEqual(res.options[0], extendederrors.ExtendedErrorOption(29, b'Result synthesized from aggressive NSEC cache (RFC8198)'))
dd54e182 RG	164
	165	# now we ask for a type that does not exist at the wildcard
	166	hits = self.getMetric('aggressive-nsec-cache-nsec-hits')
	167	nbQueries = self.getMetric('all-outqueries')
	168	res = self.sendQuery('test1.wildcard.secure.example.', 'AAAA')
	169	self.assertRcodeEqual(res, dns.rcode.NOERROR)
	170	self.assertAnswerEmpty(res)
	171	self.assertAuthorityHasSOA(res)
	172	self.assertMessageIsAuthenticated(res)
4bfebc93	173	self.assertEqual(nbQueries, self.getMetric('all-outqueries'))
dd54e182	174	self.assertGreater(self.getMetric('aggressive-nsec-cache-nsec-hits'), hits)
811bddf9 OM	175	self.assertEqual(res.edns, 0)
	176	self.assertEqual(len(res.options), 1)
	177	self.assertEqual(res.options[0].otype, 15)
7f8f0893	178	self.assertEqual(res.options[0], extendederrors.ExtendedErrorOption(29, b'Result synthesized from aggressive NSEC cache (RFC8198)'))
dd54e182 RG	179
	180	# we can also ask a different type, for a different name that is covered
	181	# by the NSEC and matches the wildcard (but the type does not exist)
	182	hits = self.getMetric('aggressive-nsec-cache-nsec-wc-hits')
	183	nbQueries = self.getMetric('all-outqueries')
	184	res = self.sendQuery('test3.wildcard.secure.example.', 'TXT')
	185	self.assertRcodeEqual(res, dns.rcode.NOERROR)
	186	self.assertAnswerEmpty(res)
	187	self.assertAuthorityHasSOA(res)
	188	self.assertMessageIsAuthenticated(res)
4bfebc93	189	self.assertEqual(nbQueries, self.getMetric('all-outqueries'))
dd54e182	190	self.assertGreater(self.getMetric('aggressive-nsec-cache-nsec-hits'), hits)
811bddf9 OM	191	self.assertEqual(res.edns, 0)
	192	self.assertEqual(len(res.options), 1)
	193	self.assertEqual(res.options[0].otype, 15)
7f8f0893	194	self.assertEqual(res.options[0], extendederrors.ExtendedErrorOption(29, b'Result synthesized from aggressive NSEC cache (RFC8198)'))
dd54e182 RG	195
dd54e182 RG	196	def test_Bogus(self):
2ec80d48 OM	197	self.wipe()
	198
	199	# query a name in example to fill the aggressive negcache
	200	res = self.sendQuery('example.', 'A')
	201	self.assertRcodeEqual(res, dns.rcode.NOERROR)
	202	self.assertAnswerEmpty(res)
	203	self.assertEqual(1, self.getMetric('aggressive-nsec-cache-entries'))
	204
dd54e182	205	# query a name in a Bogus zone
dd54e182 RG	206	res = self.sendQuery('ted1.bogus.example.', 'A')
	207	self.assertRcodeEqual(res, dns.rcode.SERVFAIL)
	208	self.assertAnswerEmpty(res)
	209
	210	# disable validation
	211	msg = dns.message.make_query('ted1.bogus.example.', 'A', want_dnssec=True)
	212	msg.flags \|= dns.flags.CD
	213
	214	res = self.sendUDPQuery(msg)
	215	self.assertRcodeEqual(res, dns.rcode.NXDOMAIN)
	216	self.assertAnswerEmpty(res)
	217	self.assertAuthorityHasSOA(res)
	218
	219	# check that we _do not_ use the aggressive NSEC cache
	220	nbQueries = self.getMetric('all-outqueries')
	221	msg = dns.message.make_query('ted2.bogus.example.', 'A', want_dnssec=True)
	222	msg.flags \|= dns.flags.CD
	223
	224	res = self.sendUDPQuery(msg)
	225	self.assertRcodeEqual(res, dns.rcode.NXDOMAIN)
	226	self.assertAnswerEmpty(res)
	227	self.assertAuthorityHasSOA(res)
	228	self.assertGreater(self.getMetric('all-outqueries'), nbQueries)
2ec80d48 OM	229
	230	# Check that we stil have one aggressive cache entry
	231	self.assertEqual(1, self.getMetric('aggressive-nsec-cache-entries'))
811bddf9 OM	232	print(res.options)
	233	self.assertEqual(res.edns, 0)
	234	self.assertEqual(len(res.options), 1)
	235	self.assertEqual(res.options[0].otype, 15)
	236	self.assertEqual(res.options[0], extendederrors.ExtendedErrorOption(9, b''))
dd54e182 RG	237
	238	class AggressiveNSECCacheNSEC3(AggressiveNSECCacheBase):
	239	_confdir = 'AggressiveNSECCacheNSEC3'
	240	__test__ = True
	241
	242	@classmethod
	243	def secureZone(cls, confdir, zonename, key=None):
	244	zone = '.' if zonename == 'ROOT' else zonename
	245	if not key:
	246	pdnsutilCmd = [os.environ['PDNSUTIL'],
	247	'--config-dir=%s' % confdir,
	248	'secure-zone',
	249	zone]
	250	else:
	251	keyfile = os.path.join(confdir, 'dnssec.key')
	252	with open(keyfile, 'w') as fdKeyfile:
	253	fdKeyfile.write(key)
	254
	255	pdnsutilCmd = [os.environ['PDNSUTIL'],
	256	'--config-dir=%s' % confdir,
	257	'import-zone-key',
	258	zone,
	259	keyfile,
	260	'active',
	261	'ksk']
	262
	263	print(' '.join(pdnsutilCmd))
	264	try:
	265	subprocess.check_output(pdnsutilCmd, stderr=subprocess.STDOUT)
	266	except subprocess.CalledProcessError as e:
	267	raise AssertionError('%s failed (%d): %s' % (pdnsutilCmd, e.returncode, e.output))
	268
	269	params = "1 0 100 AABBCCDDEEFF112233"
	270
	271	if zone == "optout.example":
	272	params = "1 1 100 AABBCCDDEEFF112233"
	273
	274	pdnsutilCmd = [os.environ['PDNSUTIL'],
	275	'--config-dir=%s' % confdir,
	276	'set-nsec3',
	277	zone,
	278	params]
	279
	280	print(' '.join(pdnsutilCmd))
	281	try:
	282	subprocess.check_output(pdnsutilCmd, stderr=subprocess.STDOUT)
	283	except subprocess.CalledProcessError as e:
	284	raise AssertionError('%s failed (%d): %s' % (pdnsutilCmd, e.returncode, e.output))
	285
	286	def testNXD(self):
7d3d2f4f	287	self.wipe()
dd54e182	288
e746a2f6	289	# first we query a nonexistent name, to get the needed NSEC3s in our cache
dd54e182 RG	290	res = self.sendQuery('host2.secure.example.', 'TXT')
	291	self.assertRcodeEqual(res, dns.rcode.NXDOMAIN)
	292	self.assertAnswerEmpty(res)
	293	self.assertAuthorityHasSOA(res)
	294	self.assertMessageIsAuthenticated(res)
	295
	296	# now we ask for a different name that is covered by the NSEC3s,
	297	# we should generate the answer from the NSEC3s and no outgoing query should be made
	298	nbQueries = self.getMetric('all-outqueries')
	299	res = self.sendQuery('host6.secure.example.', 'AAAA')
	300	self.assertRcodeEqual(res, dns.rcode.NXDOMAIN)
	301	self.assertAnswerEmpty(res)
	302	self.assertAuthorityHasSOA(res)
	303	self.assertMessageIsAuthenticated(res)
4bfebc93	304	self.assertEqual(nbQueries, self.getMetric('all-outqueries'))
811bddf9 OM	305	self.assertEqual(res.edns, 0)
	306	self.assertEqual(len(res.options), 1)
	307	self.assertEqual(res.options[0].otype, 15)
7f8f0893	308	self.assertEqual(res.options[0], extendederrors.ExtendedErrorOption(29, b'Result synthesized from aggressive NSEC cache (RFC8198)'))
dd54e182 RG	309
dd54e182 RG	310	def testWildcard(self):
7d3d2f4f	311	self.wipe()
dd54e182	312
8a64a821 RG	313	# first let's get the SOA and wildcard NSEC in our cache by asking a name that matches the wildcard
	314	# but a type that does not exist
	315	res = self.sendQuery('test1.wildcard.secure.example.', 'AAAA')
	316	self.assertRcodeEqual(res, dns.rcode.NOERROR)
	317	self.assertAnswerEmpty(res)
	318	self.assertAuthorityHasSOA(res)
	319	self.assertMessageIsAuthenticated(res)
	320
e746a2f6	321	# we query a nonexistent name, but for which a wildcard matches,
dd54e182 RG	322	# to get the NSEC3 in our cache
	323	res = self.sendQuery('test5.wildcard.secure.example.', 'A')
	324	expected = dns.rrset.from_text('test5.wildcard.secure.example.', 0, dns.rdataclass.IN, 'A', '{prefix}.10'.format(prefix=self._PREFIX))
	325	self.assertRcodeEqual(res, dns.rcode.NOERROR)
	326	self.assertMatchingRRSIGInAnswer(res, expected)
	327	self.assertMessageIsAuthenticated(res)
	328
	329	# now we ask for a different name, we should generate the answer from the NSEC3s and the wildcard,
	330	# and no outgoing query should be made
	331	nbQueries = self.getMetric('all-outqueries')
	332	res = self.sendQuery('test6.wildcard.secure.example.', 'A')
	333	expected = dns.rrset.from_text('test6.wildcard.secure.example.', 0, dns.rdataclass.IN, 'A', '{prefix}.10'.format(prefix=self._PREFIX))
	334	self.assertRcodeEqual(res, dns.rcode.NOERROR)
	335	self.assertMatchingRRSIGInAnswer(res, expected)
	336	self.assertMessageIsAuthenticated(res)
4bfebc93	337	self.assertEqual(nbQueries, self.getMetric('all-outqueries'))
811bddf9 OM	338	self.assertEqual(res.edns, 0)
	339	self.assertEqual(len(res.options), 1)
	340	self.assertEqual(res.options[0].otype, 15)
7f8f0893	341	self.assertEqual(res.options[0], extendederrors.ExtendedErrorOption(29, b'Result synthesized from aggressive NSEC cache (RFC8198)'))
dd54e182 RG	342
	343	# now we ask for a type that does not exist at the wildcard
	344	nbQueries = self.getMetric('all-outqueries')
	345	res = self.sendQuery('test5.wildcard.secure.example.', 'AAAA')
	346	self.assertRcodeEqual(res, dns.rcode.NOERROR)
	347	self.assertAnswerEmpty(res)
	348	self.assertAuthorityHasSOA(res)
	349	self.assertMessageIsAuthenticated(res)
4bfebc93	350	self.assertEqual(nbQueries, self.getMetric('all-outqueries'))
811bddf9 OM	351	self.assertEqual(res.edns, 0)
	352	self.assertEqual(len(res.options), 1)
	353	self.assertEqual(res.options[0].otype, 15)
7f8f0893	354	self.assertEqual(res.options[0], extendederrors.ExtendedErrorOption(29, b'Result synthesized from aggressive NSEC cache (RFC8198)'))
dd54e182 RG	355
	356	# we can also ask a different type, for a different name that is covered
	357	# by the NSEC3s and matches the wildcard (but the type does not exist)
	358	nbQueries = self.getMetric('all-outqueries')
	359	res = self.sendQuery('test6.wildcard.secure.example.', 'TXT')
	360	self.assertRcodeEqual(res, dns.rcode.NOERROR)
	361	self.assertAnswerEmpty(res)
	362	self.assertAuthorityHasSOA(res)
	363	self.assertMessageIsAuthenticated(res)
4bfebc93	364	self.assertEqual(nbQueries, self.getMetric('all-outqueries'))
811bddf9 OM	365	self.assertEqual(res.edns, 0)
	366	self.assertEqual(len(res.options), 1)
	367	self.assertEqual(res.options[0].otype, 15)
7f8f0893	368	self.assertEqual(res.options[0], extendederrors.ExtendedErrorOption(29, b'Result synthesized from aggressive NSEC cache (RFC8198)'))
dd54e182 RG	369
dd54e182 RG	370	def test_OptOut(self):
7d3d2f4f OM	371	self.wipe()
7d3d2f4f OM	372
dd54e182 RG	373	# query a name in an opt-out zone
	374	res = self.sendQuery('ns2.optout.example.', 'A')
	375	self.assertRcodeEqual(res, dns.rcode.NXDOMAIN)
	376	self.assertAnswerEmpty(res)
	377	self.assertAuthorityHasSOA(res)
	378
	379	# check that we _do not_ use the aggressive NSEC cache
	380	nbQueries = self.getMetric('all-outqueries')
	381	res = self.sendQuery('ns3.optout.example.', 'A')
	382	self.assertRcodeEqual(res, dns.rcode.NXDOMAIN)
	383	self.assertAnswerEmpty(res)
	384	self.assertAuthorityHasSOA(res)
	385	self.assertGreater(self.getMetric('all-outqueries'), nbQueries)
811bddf9 OM	386	self.assertEqual(res.edns, 0)
811bddf9 OM	387	self.assertEqual(len(res.options), 0)