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

#!/usr/bin/env python
import base64
import socket
import time
import dns
import dns.message
from dnsdisttests import DNSDistTest
import dnscrypt

class DNSCryptTest(DNSDistTest):
    """
    dnsdist is configured to accept DNSCrypt queries on 127.0.0.1:_dnsDistPortDNSCrypt.
    The provider's keys have been generated with:
    generateDNSCryptProviderKeys("DNSCryptProviderPublic.key", "DNSCryptProviderPrivate.key")
    Be careful to change the _providerFingerprint below if you want to regenerate the keys.
    """

    _dnsDistPort = 5340
    _dnsDistPortDNSCrypt = 8443

    _consoleKey = DNSDistTest.generateConsoleKey()
    _consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')

    _providerFingerprint = 'E1D7:2108:9A59:BF8D:F101:16FA:ED5E:EA6A:9F6C:C78F:7F91:AF6B:027E:62F4:69C3:B1AA'
    _providerName = "2.provider.name"
    _resolverCertificateSerial = 42

    # valid from 60s ago until 2h from now
    _resolverCertificateValidFrom = int(time.time() - 60)
    _resolverCertificateValidUntil = int(time.time() + 7200)

    _dnsdistStartupDelay = 10

    def doDNSCryptQuery(self, client, query, response, tcp):
        self._toResponderQueue.put(response)
        data = client.query(query.to_wire(), tcp=tcp)
        receivedResponse = dns.message.from_wire(data)
        receivedQuery = None
        if not self._fromResponderQueue.empty():
            receivedQuery = self._fromResponderQueue.get(query)

        self.assertTrue(receivedQuery)
        self.assertTrue(receivedResponse)
        receivedQuery.id = query.id
        self.assertEquals(query, receivedQuery)
        self.assertEquals(response, receivedResponse)


class TestDNSCrypt(DNSCryptTest):
    _config_template = """
    setKey("%s")
    controlSocket("127.0.0.1:%s")
    generateDNSCryptCertificate("DNSCryptProviderPrivate.key", "DNSCryptResolver.cert", "DNSCryptResolver.key", %d, %d, %d)
    addDNSCryptBind("127.0.0.1:%d", "%s", "DNSCryptResolver.cert", "DNSCryptResolver.key")
    newServer{address="127.0.0.1:%s"}
    """

    _config_params = ['_consoleKeyB64', '_consolePort', '_resolverCertificateSerial', '_resolverCertificateValidFrom', '_resolverCertificateValidUntil', '_dnsDistPortDNSCrypt', '_providerName', '_testServerPort']

    def testSimpleA(self):
        """
        DNSCrypt: encrypted A query
        """
        client = dnscrypt.DNSCryptClient(self._providerName, self._providerFingerprint, "127.0.0.1", 8443)
        name = 'a.dnscrypt.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.2.0.1')
        response.answer.append(rrset)

        self.doDNSCryptQuery(client, query, response, False)
        self.doDNSCryptQuery(client, query, response, True)

    def testResponseLargerThanPaddedQuery(self):
        """
        DNSCrypt: response larger than query

        Send a small encrypted query (don't forget to take
        the padding into account) and check that the response
        is truncated.
        """
        client = dnscrypt.DNSCryptClient(self._providerName, self._providerFingerprint, "127.0.0.1", 8443)
        name = 'smallquerylargeresponse.dnscrypt.tests.powerdns.com.'
        query = dns.message.make_query(name, 'TXT', 'IN', use_edns=True, payload=4096)
        response = dns.message.make_response(query)
        rrset = dns.rrset.from_text(name,
                                    3600,
                                    dns.rdataclass.IN,
                                    dns.rdatatype.TXT,
                                    'A'*255)
        response.answer.append(rrset)

        self._toResponderQueue.put(response)
        data = client.query(query.to_wire())
        receivedQuery = None
        if not self._fromResponderQueue.empty():
            receivedQuery = self._fromResponderQueue.get(query)

        receivedResponse = dns.message.from_wire(data)

        self.assertTrue(receivedQuery)
        receivedQuery.id = query.id
        self.assertEquals(query, receivedQuery)
        self.assertEquals(receivedResponse.question, response.question)
        self.assertTrue(receivedResponse.flags & ~dns.flags.TC)
        self.assertTrue(len(receivedResponse.answer) == 0)
        self.assertTrue(len(receivedResponse.authority) == 0)
        self.assertTrue(len(receivedResponse.additional) == 0)

    def testCertRotation(self):
        """
        DNSCrypt: certificate rotation
        """
        client = dnscrypt.DNSCryptClient(self._providerName, self._providerFingerprint, "127.0.0.1", 8443)
        client.refreshResolverCertificates()

        cert = client.getResolverCertificate()
        self.assertTrue(cert)
        self.assertEquals(cert.serial, self._resolverCertificateSerial)

        name = 'rotation.dnscrypt.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.2.0.1')
        response.answer.append(rrset)

        self.doDNSCryptQuery(client, query, response, False)
        self.doDNSCryptQuery(client, query, response, True)

        # generate a new certificate
        self.sendConsoleCommand("generateDNSCryptCertificate('DNSCryptProviderPrivate.key', 'DNSCryptResolver.cert.2', 'DNSCryptResolver.key.2', {!s}, {:.0f}, {:.0f})".format(self._resolverCertificateSerial + 1, self._resolverCertificateValidFrom, self._resolverCertificateValidUntil))
        # add that new certificate
        self.sendConsoleCommand("getDNSCryptBind(0):loadNewCertificate('DNSCryptResolver.cert.2', 'DNSCryptResolver.key.2')")

        oldSerial = self.sendConsoleCommand("getDNSCryptBind(0):getCertificate(0):getSerial()")
        self.assertEquals(int(oldSerial), self._resolverCertificateSerial)
        effectiveSerial = self.sendConsoleCommand("getDNSCryptBind(0):getCertificate(1):getSerial()")
        self.assertEquals(int(effectiveSerial), self._resolverCertificateSerial + 1)
        tsStart = self.sendConsoleCommand("getDNSCryptBind(0):getCertificate(1):getTSStart()")
        self.assertEquals(int(tsStart), self._resolverCertificateValidFrom)
        tsEnd = self.sendConsoleCommand("getDNSCryptBind(0):getCertificate(1):getTSEnd()")
        self.assertEquals(int(tsEnd), self._resolverCertificateValidUntil)

        # we should still be able to send queries with the previous certificate
        self.doDNSCryptQuery(client, query, response, False)
        self.doDNSCryptQuery(client, query, response, True)
        cert = client.getResolverCertificate()
        self.assertTrue(cert)
        self.assertEquals(cert.serial, self._resolverCertificateSerial)

        # but refreshing should get us the new one
        client.refreshResolverCertificates()
        cert = client.getResolverCertificate()
        self.assertTrue(cert)
        self.assertEquals(cert.serial, self._resolverCertificateSerial + 1)
        # we should still get the old ones
        certs = client.getAllResolverCertificates(True)
        self.assertEquals(len(certs), 2)
        self.assertEquals(certs[0].serial, self._resolverCertificateSerial)
        self.assertEquals(certs[1].serial, self._resolverCertificateSerial + 1)

        # generate a third certificate, this time in memory
        self.sendConsoleCommand("getDNSCryptBind(0):generateAndLoadInMemoryCertificate('DNSCryptProviderPrivate.key', {!s}, {:.0f}, {:.0f})".format(self._resolverCertificateSerial + 2, self._resolverCertificateValidFrom, self._resolverCertificateValidUntil))

        # we should still be able to send queries with the previous certificate
        self.doDNSCryptQuery(client, query, response, False)
        self.doDNSCryptQuery(client, query, response, True)
        cert = client.getResolverCertificate()
        self.assertTrue(cert)
        self.assertEquals(cert.serial, self._resolverCertificateSerial + 1)

        # but refreshing should get us the new one
        client.refreshResolverCertificates()
        cert = client.getResolverCertificate()
        self.assertTrue(cert)
        self.assertEquals(cert.serial, self._resolverCertificateSerial + 2)
        # we should still get the old ones
        certs = client.getAllResolverCertificates(True)
        self.assertEquals(len(certs), 3)
        self.assertEquals(certs[0].serial, self._resolverCertificateSerial)
        self.assertEquals(certs[1].serial, self._resolverCertificateSerial + 1)
        self.assertEquals(certs[2].serial, self._resolverCertificateSerial + 2)

        # generate a fourth certificate, still in memory
        self.sendConsoleCommand("getDNSCryptBind(0):generateAndLoadInMemoryCertificate('DNSCryptProviderPrivate.key', {!s}, {:.0f}, {:.0f})".format(self._resolverCertificateSerial + 3, self._resolverCertificateValidFrom, self._resolverCertificateValidUntil))

        # mark the old ones as inactive
        self.sendConsoleCommand("getDNSCryptBind(0):markInactive({!s})".format(self._resolverCertificateSerial))
        self.sendConsoleCommand("getDNSCryptBind(0):markInactive({!s})".format(self._resolverCertificateSerial + 1))
        self.sendConsoleCommand("getDNSCryptBind(0):markInactive({!s})".format(self._resolverCertificateSerial + 2))
        # we should still be able to send queries with the third one
        self.doDNSCryptQuery(client, query, response, False)
        self.doDNSCryptQuery(client, query, response, True)
        cert = client.getResolverCertificate()
        self.assertTrue(cert)
        self.assertEquals(cert.serial, self._resolverCertificateSerial + 2)
        # now remove them
        self.sendConsoleCommand("getDNSCryptBind(0):removeInactiveCertificate({!s})".format(self._resolverCertificateSerial))
        self.sendConsoleCommand("getDNSCryptBind(0):removeInactiveCertificate({!s})".format(self._resolverCertificateSerial + 1))
        self.sendConsoleCommand("getDNSCryptBind(0):removeInactiveCertificate({!s})".format(self._resolverCertificateSerial + 2))

        # we should not be able to send with the old ones anymore
        try:
            data = client.query(query.to_wire())
        except socket.timeout:
            data = None
        self.assertEquals(data, None)

        # refreshing should get us the fourth one
        client.refreshResolverCertificates()
        cert = client.getResolverCertificate()
        self.assertTrue(cert)
        self.assertEquals(cert.serial, self._resolverCertificateSerial + 3)
        # and only that one
        certs = client.getAllResolverCertificates(True)
        self.assertEquals(len(certs), 1)
        # and we should be able to query with it
        self.doDNSCryptQuery(client, query, response, False)
        self.doDNSCryptQuery(client, query, response, True)
        cert = client.getResolverCertificate()
        self.assertTrue(cert)
        self.assertEquals(cert.serial, self._resolverCertificateSerial + 3)

class TestDNSCryptWithCache(DNSCryptTest):

    _config_params = ['_resolverCertificateSerial', '_resolverCertificateValidFrom', '_resolverCertificateValidUntil', '_dnsDistPortDNSCrypt', '_providerName', '_testServerPort']
    _config_template = """
    generateDNSCryptCertificate("DNSCryptProviderPrivate.key", "DNSCryptResolver.cert", "DNSCryptResolver.key", %d, %d, %d)
    addDNSCryptBind("127.0.0.1:%d", "%s", "DNSCryptResolver.cert", "DNSCryptResolver.key")
    pc = newPacketCache(5, {maxTTL=86400, minTTL=1})
    getPool(""):setCache(pc)
    newServer{address="127.0.0.1:%s"}
    """

    def testCachedSimpleA(self):
        """
        DNSCrypt: encrypted A query served from cache
        """
        misses = 0
        client = dnscrypt.DNSCryptClient(self._providerName, self._providerFingerprint, "127.0.0.1", 8443)
        name = 'cacheda.dnscrypt.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.2.0.1')
        response.answer.append(rrset)

        # first query to fill the cache
        self._toResponderQueue.put(response)
        data = client.query(query.to_wire())
        receivedResponse = dns.message.from_wire(data)
        receivedQuery = None
        if not self._fromResponderQueue.empty():
            receivedQuery = self._fromResponderQueue.get(query)

        self.assertTrue(receivedQuery)
        self.assertTrue(receivedResponse)
        receivedQuery.id = query.id
        self.assertEquals(query, receivedQuery)
        self.assertEquals(response, receivedResponse)
        misses += 1

        # second query should get a cached response
        data = client.query(query.to_wire())
        receivedResponse = dns.message.from_wire(data)
        receivedQuery = None
        if not self._fromResponderQueue.empty():
            receivedQuery = self._fromResponderQueue.get(query)

        self.assertEquals(receivedQuery, None)
        self.assertTrue(receivedResponse)
        self.assertEquals(response, receivedResponse)
        total = 0
        for key in self._responsesCounter:
            total += self._responsesCounter[key]
        self.assertEquals(total, misses)

class TestDNSCryptAutomaticRotation(DNSCryptTest):
    _config_template = """
    setKey("%s")
    controlSocket("127.0.0.1:%s")
    generateDNSCryptCertificate("DNSCryptProviderPrivate.key", "DNSCryptResolver.cert", "DNSCryptResolver.key", %d, %d, %d)
    addDNSCryptBind("127.0.0.1:%d", "%s", "DNSCryptResolver.cert", "DNSCryptResolver.key")
    newServer{address="127.0.0.1:%s"}

    local last = 0
    serial = %d
    function maintenance()
      local now = os.time()
      if ((now - last) > 2) then
        serial = serial + 1
        getDNSCryptBind(0):generateAndLoadInMemoryCertificate('DNSCryptProviderPrivate.key', serial, now - 60, now + 120)
        last = now
      end
    end
    """

    _config_params = ['_consoleKeyB64', '_consolePort', '_resolverCertificateSerial', '_resolverCertificateValidFrom', '_resolverCertificateValidUntil', '_dnsDistPortDNSCrypt', '_providerName', '_testServerPort', '_resolverCertificateSerial']

    def testCertRotation(self):
        """
        DNSCrypt: automatic certificate rotation
        """
        client = dnscrypt.DNSCryptClient(self._providerName, self._providerFingerprint, "127.0.0.1", 8443)

        client.refreshResolverCertificates()
        cert = client.getResolverCertificate()
        self.assertTrue(cert)
        firstSerial = cert.serial
        self.assertGreaterEqual(cert.serial, self._resolverCertificateSerial)

        time.sleep(3)

        client.refreshResolverCertificates()
        cert = client.getResolverCertificate()
        self.assertTrue(cert)
        secondSerial = cert.serial
        self.assertGreater(cert.serial, firstSerial)

        name = 'automatic-rotation.dnscrypt.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.2.0.1')
        response.answer.append(rrset)

        self.doDNSCryptQuery(client, query, response, False)
        self.doDNSCryptQuery(client, query, response, True)
Commit	Line	Data
b8db58a2	1	#!/usr/bin/env python
79500db5	2	import base64
43234e76	3	import socket
b8db58a2	4	import time
b1bec9f0 RG	5	import dns
b1bec9f0 RG	6	import dns.message
b8db58a2 RG	7	from dnsdisttests import DNSDistTest
	8	import dnscrypt
	9
79500db5	10	class DNSCryptTest(DNSDistTest):
b8db58a2 RG	11	"""
	12	dnsdist is configured to accept DNSCrypt queries on 127.0.0.1:_dnsDistPortDNSCrypt.
	13	The provider's keys have been generated with:
	14	generateDNSCryptProviderKeys("DNSCryptProviderPublic.key", "DNSCryptProviderPrivate.key")
	15	Be careful to change the _providerFingerprint below if you want to regenerate the keys.
	16	"""
	17
	18	_dnsDistPort = 5340
	19	_dnsDistPortDNSCrypt = 8443
79500db5 RG	20
79500db5 RG	21	_consoleKey = DNSDistTest.generateConsoleKey()
b4f23783	22	_consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
b8db58a2	23
b8db58a2 RG	24	_providerFingerprint = 'E1D7:2108:9A59:BF8D:F101:16FA:ED5E:EA6A:9F6C:C78F:7F91:AF6B:027E:62F4:69C3:B1AA'
b8db58a2 RG	25	_providerName = "2.provider.name"
bd64cc44	26	_resolverCertificateSerial = 42
79500db5	27
bd64cc44	28	# valid from 60s ago until 2h from now
13a325f5 RG	29	_resolverCertificateValidFrom = int(time.time() - 60)
13a325f5 RG	30	_resolverCertificateValidUntil = int(time.time() + 7200)
79500db5	31
617dfe22	32	_dnsdistStartupDelay = 10
b8db58a2	33
79500db5 RG	34	def doDNSCryptQuery(self, client, query, response, tcp):
	35	self._toResponderQueue.put(response)
	36	data = client.query(query.to_wire(), tcp=tcp)
	37	receivedResponse = dns.message.from_wire(data)
	38	receivedQuery = None
	39	if not self._fromResponderQueue.empty():
	40	receivedQuery = self._fromResponderQueue.get(query)
	41
	42	self.assertTrue(receivedQuery)
	43	self.assertTrue(receivedResponse)
	44	receivedQuery.id = query.id
	45	self.assertEquals(query, receivedQuery)
	46	self.assertEquals(response, receivedResponse)
	47
	48
	49	class TestDNSCrypt(DNSCryptTest):
	50	_config_template = """
	51	setKey("%s")
	52	controlSocket("127.0.0.1:%s")
	53	generateDNSCryptCertificate("DNSCryptProviderPrivate.key", "DNSCryptResolver.cert", "DNSCryptResolver.key", %d, %d, %d)
	54	addDNSCryptBind("127.0.0.1:%d", "%s", "DNSCryptResolver.cert", "DNSCryptResolver.key")
	55	newServer{address="127.0.0.1:%s"}
	56	"""
	57
	58	_config_params = ['_consoleKeyB64', '_consolePort', '_resolverCertificateSerial', '_resolverCertificateValidFrom', '_resolverCertificateValidUntil', '_dnsDistPortDNSCrypt', '_providerName', '_testServerPort']
	59
b8db58a2 RG	60	def testSimpleA(self):
b8db58a2 RG	61	"""
617dfe22	62	DNSCrypt: encrypted A query
b8db58a2 RG	63	"""
	64	client = dnscrypt.DNSCryptClient(self._providerName, self._providerFingerprint, "127.0.0.1", 8443)
	65	name = 'a.dnscrypt.tests.powerdns.com.'
	66	query = dns.message.make_query(name, 'A', 'IN')
	67	response = dns.message.make_response(query)
	68	rrset = dns.rrset.from_text(name,
	69	3600,
	70	dns.rdataclass.IN,
	71	dns.rdatatype.A,
fcffc585	72	'192.2.0.1')
b8db58a2 RG	73	response.answer.append(rrset)
b8db58a2 RG	74
79500db5 RG	75	self.doDNSCryptQuery(client, query, response, False)
79500db5 RG	76	self.doDNSCryptQuery(client, query, response, True)
fcffc585	77
bd64cc44 RG	78	def testResponseLargerThanPaddedQuery(self):
bd64cc44 RG	79	"""
617dfe22 RG	80	DNSCrypt: response larger than query
617dfe22 RG	81
bd64cc44 RG	82	Send a small encrypted query (don't forget to take
	83	the padding into account) and check that the response
	84	is truncated.
	85	"""
	86	client = dnscrypt.DNSCryptClient(self._providerName, self._providerFingerprint, "127.0.0.1", 8443)
	87	name = 'smallquerylargeresponse.dnscrypt.tests.powerdns.com.'
	88	query = dns.message.make_query(name, 'TXT', 'IN', use_edns=True, payload=4096)
	89	response = dns.message.make_response(query)
	90	rrset = dns.rrset.from_text(name,
	91	3600,
	92	dns.rdataclass.IN,
	93	dns.rdatatype.TXT,
	94	'A'*255)
	95	response.answer.append(rrset)
	96
	97	self._toResponderQueue.put(response)
	98	data = client.query(query.to_wire())
	99	receivedQuery = None
	100	if not self._fromResponderQueue.empty():
	101	receivedQuery = self._fromResponderQueue.get(query)
	102
	103	receivedResponse = dns.message.from_wire(data)
	104
	105	self.assertTrue(receivedQuery)
	106	receivedQuery.id = query.id
	107	self.assertEquals(query, receivedQuery)
	108	self.assertEquals(receivedResponse.question, response.question)
	109	self.assertTrue(receivedResponse.flags & ~dns.flags.TC)
	110	self.assertTrue(len(receivedResponse.answer) == 0)
	111	self.assertTrue(len(receivedResponse.authority) == 0)
	112	self.assertTrue(len(receivedResponse.additional) == 0)
	113
79500db5 RG	114	def testCertRotation(self):
	115	"""
	116	DNSCrypt: certificate rotation
	117	"""
	118	client = dnscrypt.DNSCryptClient(self._providerName, self._providerFingerprint, "127.0.0.1", 8443)
	119	client.refreshResolverCertificates()
	120
	121	cert = client.getResolverCertificate()
	122	self.assertTrue(cert)
	123	self.assertEquals(cert.serial, self._resolverCertificateSerial)
	124
	125	name = 'rotation.dnscrypt.tests.powerdns.com.'
	126	query = dns.message.make_query(name, 'A', 'IN')
	127	response = dns.message.make_response(query)
	128	rrset = dns.rrset.from_text(name,
	129	3600,
	130	dns.rdataclass.IN,
	131	dns.rdatatype.A,
	132	'192.2.0.1')
	133	response.answer.append(rrset)
	134
	135	self.doDNSCryptQuery(client, query, response, False)
	136	self.doDNSCryptQuery(client, query, response, True)
	137
	138	# generate a new certificate
	139	self.sendConsoleCommand("generateDNSCryptCertificate('DNSCryptProviderPrivate.key', 'DNSCryptResolver.cert.2', 'DNSCryptResolver.key.2', {!s}, {:.0f}, {:.0f})".format(self._resolverCertificateSerial + 1, self._resolverCertificateValidFrom, self._resolverCertificateValidUntil))
43234e76	140	# add that new certificate
79500db5 RG	141	self.sendConsoleCommand("getDNSCryptBind(0):loadNewCertificate('DNSCryptResolver.cert.2', 'DNSCryptResolver.key.2')")
79500db5 RG	142
c2baf928	143	oldSerial = self.sendConsoleCommand("getDNSCryptBind(0):getCertificate(0):getSerial()")
13a325f5	144	self.assertEquals(int(oldSerial), self._resolverCertificateSerial)
c2baf928	145	effectiveSerial = self.sendConsoleCommand("getDNSCryptBind(0):getCertificate(1):getSerial()")
13a325f5	146	self.assertEquals(int(effectiveSerial), self._resolverCertificateSerial + 1)
c2baf928	147	tsStart = self.sendConsoleCommand("getDNSCryptBind(0):getCertificate(1):getTSStart()")
13a325f5	148	self.assertEquals(int(tsStart), self._resolverCertificateValidFrom)
c2baf928	149	tsEnd = self.sendConsoleCommand("getDNSCryptBind(0):getCertificate(1):getTSEnd()")
13a325f5 RG	150	self.assertEquals(int(tsEnd), self._resolverCertificateValidUntil)
13a325f5 RG	151
79500db5 RG	152	# we should still be able to send queries with the previous certificate
	153	self.doDNSCryptQuery(client, query, response, False)
	154	self.doDNSCryptQuery(client, query, response, True)
	155	cert = client.getResolverCertificate()
	156	self.assertTrue(cert)
	157	self.assertEquals(cert.serial, self._resolverCertificateSerial)
	158
	159	# but refreshing should get us the new one
	160	client.refreshResolverCertificates()
	161	cert = client.getResolverCertificate()
	162	self.assertTrue(cert)
	163	self.assertEquals(cert.serial, self._resolverCertificateSerial + 1)
43234e76 RG	164	# we should still get the old ones
	165	certs = client.getAllResolverCertificates(True)
	166	self.assertEquals(len(certs), 2)
	167	self.assertEquals(certs[0].serial, self._resolverCertificateSerial)
	168	self.assertEquals(certs[1].serial, self._resolverCertificateSerial + 1)
79500db5 RG	169
	170	# generate a third certificate, this time in memory
	171	self.sendConsoleCommand("getDNSCryptBind(0):generateAndLoadInMemoryCertificate('DNSCryptProviderPrivate.key', {!s}, {:.0f}, {:.0f})".format(self._resolverCertificateSerial + 2, self._resolverCertificateValidFrom, self._resolverCertificateValidUntil))
	172
	173	# we should still be able to send queries with the previous certificate
	174	self.doDNSCryptQuery(client, query, response, False)
	175	self.doDNSCryptQuery(client, query, response, True)
	176	cert = client.getResolverCertificate()
	177	self.assertTrue(cert)
	178	self.assertEquals(cert.serial, self._resolverCertificateSerial + 1)
	179
	180	# but refreshing should get us the new one
	181	client.refreshResolverCertificates()
	182	cert = client.getResolverCertificate()
	183	self.assertTrue(cert)
	184	self.assertEquals(cert.serial, self._resolverCertificateSerial + 2)
43234e76 RG	185	# we should still get the old ones
	186	certs = client.getAllResolverCertificates(True)
	187	self.assertEquals(len(certs), 3)
	188	self.assertEquals(certs[0].serial, self._resolverCertificateSerial)
	189	self.assertEquals(certs[1].serial, self._resolverCertificateSerial + 1)
	190	self.assertEquals(certs[2].serial, self._resolverCertificateSerial + 2)
	191
	192	# generate a fourth certificate, still in memory
	193	self.sendConsoleCommand("getDNSCryptBind(0):generateAndLoadInMemoryCertificate('DNSCryptProviderPrivate.key', {!s}, {:.0f}, {:.0f})".format(self._resolverCertificateSerial + 3, self._resolverCertificateValidFrom, self._resolverCertificateValidUntil))
	194
	195	# mark the old ones as inactive
	196	self.sendConsoleCommand("getDNSCryptBind(0):markInactive({!s})".format(self._resolverCertificateSerial))
	197	self.sendConsoleCommand("getDNSCryptBind(0):markInactive({!s})".format(self._resolverCertificateSerial + 1))
	198	self.sendConsoleCommand("getDNSCryptBind(0):markInactive({!s})".format(self._resolverCertificateSerial + 2))
	199	# we should still be able to send queries with the third one
	200	self.doDNSCryptQuery(client, query, response, False)
	201	self.doDNSCryptQuery(client, query, response, True)
	202	cert = client.getResolverCertificate()
	203	self.assertTrue(cert)
	204	self.assertEquals(cert.serial, self._resolverCertificateSerial + 2)
	205	# now remove them
	206	self.sendConsoleCommand("getDNSCryptBind(0):removeInactiveCertificate({!s})".format(self._resolverCertificateSerial))
	207	self.sendConsoleCommand("getDNSCryptBind(0):removeInactiveCertificate({!s})".format(self._resolverCertificateSerial + 1))
	208	self.sendConsoleCommand("getDNSCryptBind(0):removeInactiveCertificate({!s})".format(self._resolverCertificateSerial + 2))
	209
	210	# we should not be able to send with the old ones anymore
	211	try:
	212	data = client.query(query.to_wire())
	213	except socket.timeout:
	214	data = None
	215	self.assertEquals(data, None)
	216
	217	# refreshing should get us the fourth one
	218	client.refreshResolverCertificates()
	219	cert = client.getResolverCertificate()
	220	self.assertTrue(cert)
	221	self.assertEquals(cert.serial, self._resolverCertificateSerial + 3)
	222	# and only that one
	223	certs = client.getAllResolverCertificates(True)
	224	self.assertEquals(len(certs), 1)
	225	# and we should be able to query with it
	226	self.doDNSCryptQuery(client, query, response, False)
	227	self.doDNSCryptQuery(client, query, response, True)
	228	cert = client.getResolverCertificate()
	229	self.assertTrue(cert)
	230	self.assertEquals(cert.serial, self._resolverCertificateSerial + 3)
79500db5 RG	231
	232	class TestDNSCryptWithCache(DNSCryptTest):
	233
fcffc585 RG	234	_config_params = ['_resolverCertificateSerial', '_resolverCertificateValidFrom', '_resolverCertificateValidUntil', '_dnsDistPortDNSCrypt', '_providerName', '_testServerPort']
	235	_config_template = """
	236	generateDNSCryptCertificate("DNSCryptProviderPrivate.key", "DNSCryptResolver.cert", "DNSCryptResolver.key", %d, %d, %d)
	237	addDNSCryptBind("127.0.0.1:%d", "%s", "DNSCryptResolver.cert", "DNSCryptResolver.key")
7d294573	238	pc = newPacketCache(5, {maxTTL=86400, minTTL=1})
fcffc585 RG	239	getPool(""):setCache(pc)
	240	newServer{address="127.0.0.1:%s"}
	241	"""
	242
	243	def testCachedSimpleA(self):
	244	"""
	245	DNSCrypt: encrypted A query served from cache
	246	"""
e91084ce	247	misses = 0
fcffc585 RG	248	client = dnscrypt.DNSCryptClient(self._providerName, self._providerFingerprint, "127.0.0.1", 8443)
	249	name = 'cacheda.dnscrypt.tests.powerdns.com.'
	250	query = dns.message.make_query(name, 'A', 'IN')
	251	response = dns.message.make_response(query)
	252	rrset = dns.rrset.from_text(name,
	253	3600,
	254	dns.rdataclass.IN,
	255	dns.rdatatype.A,
	256	'192.2.0.1')
	257	response.answer.append(rrset)
	258
	259	# first query to fill the cache
	260	self._toResponderQueue.put(response)
	261	data = client.query(query.to_wire())
	262	receivedResponse = dns.message.from_wire(data)
	263	receivedQuery = None
	264	if not self._fromResponderQueue.empty():
	265	receivedQuery = self._fromResponderQueue.get(query)
	266
	267	self.assertTrue(receivedQuery)
	268	self.assertTrue(receivedResponse)
	269	receivedQuery.id = query.id
	270	self.assertEquals(query, receivedQuery)
	271	self.assertEquals(response, receivedResponse)
e91084ce	272	misses += 1
fcffc585 RG	273
	274	# second query should get a cached response
	275	data = client.query(query.to_wire())
	276	receivedResponse = dns.message.from_wire(data)
	277	receivedQuery = None
	278	if not self._fromResponderQueue.empty():
	279	receivedQuery = self._fromResponderQueue.get(query)
	280
	281	self.assertEquals(receivedQuery, None)
	282	self.assertTrue(receivedResponse)
	283	self.assertEquals(response, receivedResponse)
e91084ce RG	284	total = 0
	285	for key in self._responsesCounter:
	286	total += self._responsesCounter[key]
	287	self.assertEquals(total, misses)
79500db5 RG	288
	289	class TestDNSCryptAutomaticRotation(DNSCryptTest):
	290	_config_template = """
	291	setKey("%s")
	292	controlSocket("127.0.0.1:%s")
	293	generateDNSCryptCertificate("DNSCryptProviderPrivate.key", "DNSCryptResolver.cert", "DNSCryptResolver.key", %d, %d, %d)
	294	addDNSCryptBind("127.0.0.1:%d", "%s", "DNSCryptResolver.cert", "DNSCryptResolver.key")
	295	newServer{address="127.0.0.1:%s"}
	296
	297	local last = 0
	298	serial = %d
	299	function maintenance()
	300	local now = os.time()
	301	if ((now - last) > 2) then
	302	serial = serial + 1
	303	getDNSCryptBind(0):generateAndLoadInMemoryCertificate('DNSCryptProviderPrivate.key', serial, now - 60, now + 120)
	304	last = now
	305	end
	306	end
	307	"""
	308
	309	_config_params = ['_consoleKeyB64', '_consolePort', '_resolverCertificateSerial', '_resolverCertificateValidFrom', '_resolverCertificateValidUntil', '_dnsDistPortDNSCrypt', '_providerName', '_testServerPort', '_resolverCertificateSerial']
	310
	311	def testCertRotation(self):
	312	"""
	313	DNSCrypt: automatic certificate rotation
	314	"""
	315	client = dnscrypt.DNSCryptClient(self._providerName, self._providerFingerprint, "127.0.0.1", 8443)
	316
	317	client.refreshResolverCertificates()
	318	cert = client.getResolverCertificate()
	319	self.assertTrue(cert)
	320	firstSerial = cert.serial
	321	self.assertGreaterEqual(cert.serial, self._resolverCertificateSerial)
	322
	323	time.sleep(3)
	324
	325	client.refreshResolverCertificates()
	326	cert = client.getResolverCertificate()
	327	self.assertTrue(cert)
	328	secondSerial = cert.serial
	329	self.assertGreater(cert.serial, firstSerial)
	330
	331	name = 'automatic-rotation.dnscrypt.tests.powerdns.com.'
	332	query = dns.message.make_query(name, 'A', 'IN')
	333	response = dns.message.make_response(query)
	334	rrset = dns.rrset.from_text(name,
	335	3600,
	336	dns.rdataclass.IN,
	337	dns.rdatatype.A,
	338	'192.2.0.1')
	339	response.answer.append(rrset)
	340
	341	self.doDNSCryptQuery(client, query, response, False)
	342	self.doDNSCryptQuery(client, query, response, True)