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

#!/usr/bin/env python
import base64
import dns
import os
import subprocess
import unittest
from dnsdisttests import DNSDistTest

class DNSDistOCSPStaplingTest(DNSDistTest):

    @classmethod
    def checkOCSPStaplingStatus(cls, addr, port, serverName, caFile):
        testcmd = ['openssl', 's_client', '-CAfile', caFile, '-connect', '%s:%d' % (addr, port), '-status', '-servername', serverName ]
        output = None
        try:
            process = subprocess.Popen(testcmd, stdout=subprocess.PIPE, stdin=subprocess.PIPE, stderr=subprocess.STDOUT, close_fds=True)
            output = process.communicate(input='')
        except subprocess.CalledProcessError as exc:
            raise AssertionError('openssl s_client failed (%d): %s' % (exc.returncode, exc.output))

        return output[0].decode()

    @classmethod
    def getOCSPSerial(cls, output):
        serialNumber = None
        for line in output.splitlines():
            line = line.strip()
            print(line)
            if line.startswith('Serial Number:'):
                (_, serialNumber) = line.split(':')
                break

        return serialNumber

@unittest.skipIf('SKIP_DOH_TESTS' in os.environ, 'DNS over HTTPS tests are disabled')
class TestOCSPStaplingDOH(DNSDistOCSPStaplingTest):

    _consoleKey = DNSDistTest.generateConsoleKey()
    _consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
    _serverKey = 'server.key'
    _serverCert = 'server.chain'
    _serverName = 'tls.tests.dnsdist.org'
    _ocspFile = 'server.ocsp'
    _caCert = 'ca.pem'
    _caKey = 'ca.key'
    _dohServerPort = 8443
    _config_template = """
    newServer{address="127.0.0.1:%s"}
    setKey("%s")
    controlSocket("127.0.0.1:%s")

    -- generate an OCSP response file for our certificate, valid one day
    generateOCSPResponse('%s', '%s', '%s', '%s', 1, 0)
    addDOHLocal("127.0.0.1:%s", "%s", "%s", { "/" }, { ocspResponses={"%s"}})
    """
    _config_params = ['_testServerPort', '_consoleKeyB64', '_consolePort', '_serverCert', '_caCert', '_caKey', '_ocspFile', '_dohServerPort', '_serverCert', '_serverKey', '_ocspFile']

    @classmethod
    def setUpClass(cls):

        # for some reason, @unittest.skipIf() is not applied to derived classes with some versions of Python
        if 'SKIP_DOH_TESTS' in os.environ:
            raise unittest.SkipTest('DNS over HTTPS tests are disabled')

        cls.startResponders()
        cls.startDNSDist()
        cls.setUpSockets()

        print("Launching tests..")

    def testOCSPStapling(self):
        """
        OCSP Stapling: DOH
        """
        output = self.checkOCSPStaplingStatus('127.0.0.1', self._dohServerPort, self._serverName, self._caCert)
        self.assertIn('OCSP Response Status: successful (0x0)', output)

        serialNumber = self.getOCSPSerial(output)
        self.assertTrue(serialNumber)

        self.generateNewCertificateAndKey()
        self.sendConsoleCommand("generateOCSPResponse('%s', '%s', '%s', '%s', 1, 0)" % (self._serverCert, self._caCert, self._caKey, self._ocspFile))
        self.sendConsoleCommand("reloadAllCertificates()")

        output = self.checkOCSPStaplingStatus('127.0.0.1', self._dohServerPort, self._serverName, self._caCert)
        self.assertIn('OCSP Response Status: successful (0x0)', output)
        serialNumber2 = self.getOCSPSerial(output)
        self.assertTrue(serialNumber2)
        self.assertNotEqual(serialNumber, serialNumber2)

class TestOCSPStaplingTLSGnuTLS(DNSDistOCSPStaplingTest):

    _consoleKey = DNSDistTest.generateConsoleKey()
    _consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
    _serverKey = 'server.key'
    _serverCert = 'server.chain'
    _serverName = 'tls.tests.dnsdist.org'
    _ocspFile = 'server.ocsp'
    _caCert = 'ca.pem'
    _caKey = 'ca.key'
    _tlsServerPort = 8443
    _config_template = """
    newServer{address="127.0.0.1:%s"}
    setKey("%s")
    controlSocket("127.0.0.1:%s")

    -- generate an OCSP response file for our certificate, valid one day
    generateOCSPResponse('%s', '%s', '%s', '%s', 1, 0)
    addTLSLocal("127.0.0.1:%s", "%s", "%s", { provider="gnutls", ocspResponses={"%s"}})
    """
    _config_params = ['_testServerPort', '_consoleKeyB64', '_consolePort', '_serverCert', '_caCert', '_caKey', '_ocspFile', '_tlsServerPort', '_serverCert', '_serverKey', '_ocspFile']

    def testOCSPStapling(self):
        """
        OCSP Stapling: TLS (GnuTLS)
        """
        output = self.checkOCSPStaplingStatus('127.0.0.1', self._tlsServerPort, self._serverName, self._caCert)
        self.assertIn('OCSP Response Status: successful (0x0)', output)

        serialNumber = self.getOCSPSerial(output)
        self.assertTrue(serialNumber)

        self.generateNewCertificateAndKey()
        self.sendConsoleCommand("generateOCSPResponse('%s', '%s', '%s', '%s', 1, 0)" % (self._serverCert, self._caCert, self._caKey, self._ocspFile))
        self.sendConsoleCommand("reloadAllCertificates()")

        output = self.checkOCSPStaplingStatus('127.0.0.1', self._tlsServerPort, self._serverName, self._caCert)
        self.assertIn('OCSP Response Status: successful (0x0)', output)
        serialNumber2 = self.getOCSPSerial(output)
        self.assertTrue(serialNumber2)
        self.assertNotEqual(serialNumber, serialNumber2)

class TestOCSPStaplingTLSOpenSSL(DNSDistOCSPStaplingTest):

    _consoleKey = DNSDistTest.generateConsoleKey()
    _consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
    _serverKey = 'server.key'
    _serverCert = 'server.chain'
    _serverName = 'tls.tests.dnsdist.org'
    _ocspFile = 'server.ocsp'
    _caCert = 'ca.pem'
    _caKey = 'ca.key'
    _tlsServerPort = 8443
    _config_template = """
    newServer{address="127.0.0.1:%s"}
    setKey("%s")
    controlSocket("127.0.0.1:%s")

    -- generate an OCSP response file for our certificate, valid one day
    generateOCSPResponse('%s', '%s', '%s', '%s', 1, 0)
    addTLSLocal("127.0.0.1:%s", "%s", "%s", { provider="openssl", ocspResponses={"%s"}})
    """
    _config_params = ['_testServerPort', '_consoleKeyB64', '_consolePort', '_serverCert', '_caCert', '_caKey', '_ocspFile', '_tlsServerPort', '_serverCert', '_serverKey', '_ocspFile']

    def testOCSPStapling(self):
        """
        OCSP Stapling: TLS (OpenSSL)
        """
        output = self.checkOCSPStaplingStatus('127.0.0.1', self._tlsServerPort, self._serverName, self._caCert)
        self.assertIn('OCSP Response Status: successful (0x0)', output)

        serialNumber = self.getOCSPSerial(output)
        self.assertTrue(serialNumber)

        self.generateNewCertificateAndKey()
        self.sendConsoleCommand("generateOCSPResponse('%s', '%s', '%s', '%s', 1, 0)" % (self._serverCert, self._caCert, self._caKey, self._ocspFile))
        self.sendConsoleCommand("reloadAllCertificates()")

        output = self.checkOCSPStaplingStatus('127.0.0.1', self._tlsServerPort, self._serverName, self._caCert)
        self.assertIn('OCSP Response Status: successful (0x0)', output)
        serialNumber2 = self.getOCSPSerial(output)
        self.assertTrue(serialNumber2)
        self.assertNotEqual(serialNumber, serialNumber2)
Commit	Line	Data
8c15553e	1	#!/usr/bin/env python
1d896c34	2	import base64
8c15553e	3	import dns
13291274	4	import os
8c15553e	5	import subprocess
13291274	6	import unittest
8c15553e RG	7	from dnsdisttests import DNSDistTest
	8
	9	class DNSDistOCSPStaplingTest(DNSDistTest):
	10
	11	@classmethod
	12	def checkOCSPStaplingStatus(cls, addr, port, serverName, caFile):
	13	testcmd = ['openssl', 's_client', '-CAfile', caFile, '-connect', '%s:%d' % (addr, port), '-status', '-servername', serverName ]
	14	output = None
	15	try:
	16	process = subprocess.Popen(testcmd, stdout=subprocess.PIPE, stdin=subprocess.PIPE, stderr=subprocess.STDOUT, close_fds=True)
	17	output = process.communicate(input='')
	18	except subprocess.CalledProcessError as exc:
1d896c34	19	raise AssertionError('openssl s_client failed (%d): %s' % (exc.returncode, exc.output))
8c15553e RG	20
	21	return output[0].decode()
	22
1d896c34 RG	23	@classmethod
	24	def getOCSPSerial(cls, output):
	25	serialNumber = None
	26	for line in output.splitlines():
	27	line = line.strip()
	28	print(line)
	29	if line.startswith('Serial Number:'):
	30	(_, serialNumber) = line.split(':')
	31	break
	32
	33	return serialNumber
	34
13291274	35	@unittest.skipIf('SKIP_DOH_TESTS' in os.environ, 'DNS over HTTPS tests are disabled')
8c15553e RG	36	class TestOCSPStaplingDOH(DNSDistOCSPStaplingTest):
8c15553e RG	37
1d896c34 RG	38	_consoleKey = DNSDistTest.generateConsoleKey()
1d896c34 RG	39	_consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
8c15553e RG	40	_serverKey = 'server.key'
	41	_serverCert = 'server.chain'
	42	_serverName = 'tls.tests.dnsdist.org'
	43	_ocspFile = 'server.ocsp'
	44	_caCert = 'ca.pem'
	45	_caKey = 'ca.key'
	46	_dohServerPort = 8443
	47	_config_template = """
	48	newServer{address="127.0.0.1:%s"}
1d896c34 RG	49	setKey("%s")
1d896c34 RG	50	controlSocket("127.0.0.1:%s")
8c15553e RG	51
	52	-- generate an OCSP response file for our certificate, valid one day
	53	generateOCSPResponse('%s', '%s', '%s', '%s', 1, 0)
	54	addDOHLocal("127.0.0.1:%s", "%s", "%s", { "/" }, { ocspResponses={"%s"}})
	55	"""
1d896c34	56	_config_params = ['_testServerPort', '_consoleKeyB64', '_consolePort', '_serverCert', '_caCert', '_caKey', '_ocspFile', '_dohServerPort', '_serverCert', '_serverKey', '_ocspFile']
8c15553e	57
13291274 RG	58	@classmethod
	59	def setUpClass(cls):
	60
	61	# for some reason, @unittest.skipIf() is not applied to derived classes with some versions of Python
	62	if 'SKIP_DOH_TESTS' in os.environ:
	63	raise unittest.SkipTest('DNS over HTTPS tests are disabled')
	64
	65	cls.startResponders()
	66	cls.startDNSDist()
	67	cls.setUpSockets()
	68
	69	print("Launching tests..")
	70
8c15553e RG	71	def testOCSPStapling(self):
	72	"""
	73	OCSP Stapling: DOH
	74	"""
	75	output = self.checkOCSPStaplingStatus('127.0.0.1', self._dohServerPort, self._serverName, self._caCert)
	76	self.assertIn('OCSP Response Status: successful (0x0)', output)
	77
1d896c34 RG	78	serialNumber = self.getOCSPSerial(output)
	79	self.assertTrue(serialNumber)
	80
	81	self.generateNewCertificateAndKey()
	82	self.sendConsoleCommand("generateOCSPResponse('%s', '%s', '%s', '%s', 1, 0)" % (self._serverCert, self._caCert, self._caKey, self._ocspFile))
	83	self.sendConsoleCommand("reloadAllCertificates()")
	84
	85	output = self.checkOCSPStaplingStatus('127.0.0.1', self._dohServerPort, self._serverName, self._caCert)
	86	self.assertIn('OCSP Response Status: successful (0x0)', output)
	87	serialNumber2 = self.getOCSPSerial(output)
	88	self.assertTrue(serialNumber2)
4bfebc93	89	self.assertNotEqual(serialNumber, serialNumber2)
1d896c34	90
8c15553e RG	91	class TestOCSPStaplingTLSGnuTLS(DNSDistOCSPStaplingTest):
8c15553e RG	92
1d896c34 RG	93	_consoleKey = DNSDistTest.generateConsoleKey()
1d896c34 RG	94	_consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
8c15553e RG	95	_serverKey = 'server.key'
	96	_serverCert = 'server.chain'
	97	_serverName = 'tls.tests.dnsdist.org'
	98	_ocspFile = 'server.ocsp'
	99	_caCert = 'ca.pem'
	100	_caKey = 'ca.key'
	101	_tlsServerPort = 8443
	102	_config_template = """
	103	newServer{address="127.0.0.1:%s"}
1d896c34 RG	104	setKey("%s")
1d896c34 RG	105	controlSocket("127.0.0.1:%s")
8c15553e RG	106
	107	-- generate an OCSP response file for our certificate, valid one day
	108	generateOCSPResponse('%s', '%s', '%s', '%s', 1, 0)
	109	addTLSLocal("127.0.0.1:%s", "%s", "%s", { provider="gnutls", ocspResponses={"%s"}})
	110	"""
1d896c34	111	_config_params = ['_testServerPort', '_consoleKeyB64', '_consolePort', '_serverCert', '_caCert', '_caKey', '_ocspFile', '_tlsServerPort', '_serverCert', '_serverKey', '_ocspFile']
8c15553e RG	112
	113	def testOCSPStapling(self):
	114	"""
	115	OCSP Stapling: TLS (GnuTLS)
	116	"""
	117	output = self.checkOCSPStaplingStatus('127.0.0.1', self._tlsServerPort, self._serverName, self._caCert)
	118	self.assertIn('OCSP Response Status: successful (0x0)', output)
	119
1d896c34 RG	120	serialNumber = self.getOCSPSerial(output)
	121	self.assertTrue(serialNumber)
	122
	123	self.generateNewCertificateAndKey()
	124	self.sendConsoleCommand("generateOCSPResponse('%s', '%s', '%s', '%s', 1, 0)" % (self._serverCert, self._caCert, self._caKey, self._ocspFile))
	125	self.sendConsoleCommand("reloadAllCertificates()")
	126
	127	output = self.checkOCSPStaplingStatus('127.0.0.1', self._tlsServerPort, self._serverName, self._caCert)
	128	self.assertIn('OCSP Response Status: successful (0x0)', output)
	129	serialNumber2 = self.getOCSPSerial(output)
	130	self.assertTrue(serialNumber2)
4bfebc93	131	self.assertNotEqual(serialNumber, serialNumber2)
1d896c34	132
8c15553e RG	133	class TestOCSPStaplingTLSOpenSSL(DNSDistOCSPStaplingTest):
8c15553e RG	134
1d896c34 RG	135	_consoleKey = DNSDistTest.generateConsoleKey()
1d896c34 RG	136	_consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
8c15553e RG	137	_serverKey = 'server.key'
	138	_serverCert = 'server.chain'
	139	_serverName = 'tls.tests.dnsdist.org'
	140	_ocspFile = 'server.ocsp'
	141	_caCert = 'ca.pem'
	142	_caKey = 'ca.key'
	143	_tlsServerPort = 8443
	144	_config_template = """
	145	newServer{address="127.0.0.1:%s"}
1d896c34 RG	146	setKey("%s")
1d896c34 RG	147	controlSocket("127.0.0.1:%s")
8c15553e RG	148
	149	-- generate an OCSP response file for our certificate, valid one day
	150	generateOCSPResponse('%s', '%s', '%s', '%s', 1, 0)
	151	addTLSLocal("127.0.0.1:%s", "%s", "%s", { provider="openssl", ocspResponses={"%s"}})
	152	"""
1d896c34	153	_config_params = ['_testServerPort', '_consoleKeyB64', '_consolePort', '_serverCert', '_caCert', '_caKey', '_ocspFile', '_tlsServerPort', '_serverCert', '_serverKey', '_ocspFile']
8c15553e RG	154
	155	def testOCSPStapling(self):
	156	"""
	157	OCSP Stapling: TLS (OpenSSL)
	158	"""
	159	output = self.checkOCSPStaplingStatus('127.0.0.1', self._tlsServerPort, self._serverName, self._caCert)
	160	self.assertIn('OCSP Response Status: successful (0x0)', output)
1d896c34 RG	161
	162	serialNumber = self.getOCSPSerial(output)
	163	self.assertTrue(serialNumber)
	164
	165	self.generateNewCertificateAndKey()
	166	self.sendConsoleCommand("generateOCSPResponse('%s', '%s', '%s', '%s', 1, 0)" % (self._serverCert, self._caCert, self._caKey, self._ocspFile))
	167	self.sendConsoleCommand("reloadAllCertificates()")
	168
	169	output = self.checkOCSPStaplingStatus('127.0.0.1', self._tlsServerPort, self._serverName, self._caCert)
	170	self.assertIn('OCSP Response Status: successful (0x0)', output)
	171	serialNumber2 = self.getOCSPSerial(output)
	172	self.assertTrue(serialNumber2)
4bfebc93	173	self.assertNotEqual(serialNumber, serialNumber2)