regression-tests.dnsdist/test_OCSP.py

   1 #!/usr/bin/env python
   2 import base64
   3 import dns
   4 import os
   5 import subprocess
   6 import unittest
   7 from dnsdisttests import DNSDistTest, pickAvailablePort
   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:
  19             raise AssertionError('openssl s_client failed (%d): %s' % (exc.returncode, exc.output))
  20
  21         return output[0].decode()
  22
  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
  35     def getTLSProvider(self):
  36         return self.sendConsoleCommand("getBind(0):getEffectiveTLSProvider()").rstrip()
  37
  38     @classmethod
  39     def setUpClass(cls):
  40         cls.generateNewCertificateAndKey('server-ocsp')
  41         cls.startResponders()
  42         cls.startDNSDist()
  43         cls.setUpSockets()
  44
  45 @unittest.skipIf('SKIP_DOH_TESTS' in os.environ, 'DNS over HTTPS tests are disabled')
  46 class TestOCSPStaplingDOH(DNSDistOCSPStaplingTest):
  47
  48     _consoleKey = DNSDistTest.generateConsoleKey()
  49     _consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
  50     _serverKey = 'server-ocsp.key'
  51     _serverCert = 'server-ocsp.chain'
  52     _serverName = 'tls.tests.dnsdist.org'
  53     _ocspFile = 'server.ocsp'
  54     _caCert = 'ca.pem'
  55     _caKey = 'ca.key'
  56     _dohWithNGHTTP2ServerPort = pickAvailablePort()
  57     _dohWithH2OServerPort = pickAvailablePort()
  58     _config_template = """
  59     newServer{address="127.0.0.1:%d"}
  60     setKey("%s")
  61     controlSocket("127.0.0.1:%d")
  62
  63     -- generate an OCSP response file for our certificate, valid one day
  64     generateOCSPResponse('%s', '%s', '%s', '%s', 1, 0)
  65     addDOHLocal("127.0.0.1:%d", "%s", "%s", { "/" }, { ocspResponses={"%s"}, library='nghttp2'})
  66     addDOHLocal("127.0.0.1:%d", "%s", "%s", { "/" }, { ocspResponses={"%s"}, library='h2o'})
  67     """
  68     _config_params = ['_testServerPort', '_consoleKeyB64', '_consolePort', '_serverCert', '_caCert', '_caKey', '_ocspFile', '_dohWithNGHTTP2ServerPort', '_serverCert', '_serverKey', '_ocspFile', '_dohWithH2OServerPort', '_serverCert', '_serverKey', '_ocspFile']
  69
  70     @classmethod
  71     def setUpClass(cls):
  72
  73         # for some reason, @unittest.skipIf() is not applied to derived classes with some versions of Python
  74         if 'SKIP_DOH_TESTS' in os.environ:
  75             raise unittest.SkipTest('DNS over HTTPS tests are disabled')
  76
  77         cls.generateNewCertificateAndKey('server-ocsp')
  78         cls.startResponders()
  79         cls.startDNSDist()
  80         cls.setUpSockets()
  81
  82         print("Launching tests..")
  83
  84     def testOCSPStapling(self):
  85         """
  86         OCSP Stapling: DOH
  87         """
  88         for port in [self._dohWithNGHTTP2ServerPort, self._dohWithH2OServerPort]:
  89             output = self.checkOCSPStaplingStatus('127.0.0.1', port, self._serverName, self._caCert)
  90             self.assertIn('OCSP Response Status: successful (0x0)', output)
  91
  92             serialNumber = self.getOCSPSerial(output)
  93             self.assertTrue(serialNumber)
  94
  95             self.generateNewCertificateAndKey('server-ocsp')
  96             self.sendConsoleCommand("generateOCSPResponse('%s', '%s', '%s', '%s', 1, 0)" % (self._serverCert, self._caCert, self._caKey, self._ocspFile))
  97             self.sendConsoleCommand("reloadAllCertificates()")
  98
  99             output = self.checkOCSPStaplingStatus('127.0.0.1', port, self._serverName, self._caCert)
 100             self.assertIn('OCSP Response Status: successful (0x0)', output)
 101             serialNumber2 = self.getOCSPSerial(output)
 102             self.assertTrue(serialNumber2)
 103             self.assertNotEqual(serialNumber, serialNumber2)
 104
 105 class TestBrokenOCSPStaplingDoH(DNSDistOCSPStaplingTest):
 106
 107     _consoleKey = DNSDistTest.generateConsoleKey()
 108     _consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
 109     _serverKey = 'server-ocsp.key'
 110     _serverCert = 'server-ocsp.chain'
 111     _serverName = 'tls.tests.dnsdist.org'
 112     _caCert = 'ca.pem'
 113     # invalid OCSP file!
 114     _ocspFile = '/dev/null'
 115     _dohWithNGHTTP2ServerPort = pickAvailablePort()
 116     _dohWithH2OServerPort = pickAvailablePort()
 117     _config_template = """
 118     newServer{address="127.0.0.1:%s"}
 119     setKey("%s")
 120     controlSocket("127.0.0.1:%s")
 121
 122     addDOHLocal("127.0.0.1:%d", "%s", "%s", { "/" }, { ocspResponses={"%s"}, library='nghttp2'})
 123     addDOHLocal("127.0.0.1:%d", "%s", "%s", { "/" }, { ocspResponses={"%s"}, library='h2o'})
 124
 125     """
 126     _config_params = ['_testServerPort', '_consoleKeyB64', '_consolePort', '_dohWithNGHTTP2ServerPort', '_serverCert', '_serverKey', '_ocspFile', '_dohWithH2OServerPort', '_serverCert', '_serverKey', '_ocspFile']
 127
 128     def testBrokenOCSPStapling(self):
 129         """
 130         OCSP Stapling: Broken (DoH)
 131         """
 132         for port in [self._dohWithNGHTTP2ServerPort, self._dohWithH2OServerPort]:
 133             output = self.checkOCSPStaplingStatus('127.0.0.1', port, self._serverName, self._caCert)
 134             self.assertNotIn('OCSP Response Status: successful (0x0)', output)
 135
 136 class TestOCSPStaplingTLSGnuTLS(DNSDistOCSPStaplingTest):
 137
 138     _consoleKey = DNSDistTest.generateConsoleKey()
 139     _consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
 140     _serverKey = 'server-ocsp.key'
 141     _serverCert = 'server-ocsp.chain'
 142     _serverName = 'tls.tests.dnsdist.org'
 143     _ocspFile = 'server.ocsp'
 144     _caCert = 'ca.pem'
 145     _caKey = 'ca.key'
 146     _tlsServerPort = pickAvailablePort()
 147     _config_template = """
 148     newServer{address="127.0.0.1:%s"}
 149     setKey("%s")
 150     controlSocket("127.0.0.1:%s")
 151
 152     -- generate an OCSP response file for our certificate, valid one day
 153     generateOCSPResponse('%s', '%s', '%s', '%s', 1, 0)
 154     addTLSLocal("127.0.0.1:%s", "%s", "%s", { provider="gnutls", ocspResponses={"%s"}})
 155     """
 156     _config_params = ['_testServerPort', '_consoleKeyB64', '_consolePort', '_serverCert', '_caCert', '_caKey', '_ocspFile', '_tlsServerPort', '_serverCert', '_serverKey', '_ocspFile']
 157
 158     def testOCSPStapling(self):
 159         """
 160         OCSP Stapling: TLS (GnuTLS)
 161         """
 162         output = self.checkOCSPStaplingStatus('127.0.0.1', self._tlsServerPort, self._serverName, self._caCert)
 163         self.assertIn('OCSP Response Status: successful (0x0)', output)
 164         self.assertEqual(self.getTLSProvider(), "gnutls")
 165
 166         serialNumber = self.getOCSPSerial(output)
 167         self.assertTrue(serialNumber)
 168
 169         self.generateNewCertificateAndKey('server-ocsp')
 170         self.sendConsoleCommand("generateOCSPResponse('%s', '%s', '%s', '%s', 1, 0)" % (self._serverCert, self._caCert, self._caKey, self._ocspFile))
 171         self.sendConsoleCommand("reloadAllCertificates()")
 172
 173         output = self.checkOCSPStaplingStatus('127.0.0.1', self._tlsServerPort, self._serverName, self._caCert)
 174         self.assertIn('OCSP Response Status: successful (0x0)', output)
 175         serialNumber2 = self.getOCSPSerial(output)
 176         self.assertTrue(serialNumber2)
 177         self.assertNotEqual(serialNumber, serialNumber2)
 178
 179 class TestBrokenOCSPStaplingTLSGnuTLS(DNSDistOCSPStaplingTest):
 180
 181     _consoleKey = DNSDistTest.generateConsoleKey()
 182     _consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
 183     _serverKey = 'server-ocsp.key'
 184     _serverCert = 'server-ocsp.chain'
 185     _serverName = 'tls.tests.dnsdist.org'
 186     _caCert = 'ca.pem'
 187     # invalid OCSP file!
 188     _ocspFile = '/dev/null'
 189     _tlsServerPort = pickAvailablePort()
 190     _config_template = """
 191     newServer{address="127.0.0.1:%s"}
 192     setKey("%s")
 193     controlSocket("127.0.0.1:%s")
 194
 195     addTLSLocal("127.0.0.1:%s", "%s", "%s", { provider="gnutls", ocspResponses={"%s"}})
 196     """
 197     _config_params = ['_testServerPort', '_consoleKeyB64', '_consolePort', '_tlsServerPort', '_serverCert', '_serverKey', '_ocspFile']
 198
 199     def testBrokenOCSPStapling(self):
 200         """
 201         OCSP Stapling: Broken (GnuTLS)
 202         """
 203         output = self.checkOCSPStaplingStatus('127.0.0.1', self._tlsServerPort, self._serverName, self._caCert)
 204         self.assertNotIn('OCSP Response Status: successful (0x0)', output)
 205         self.assertEqual(self.getTLSProvider(), "gnutls")
 206
 207 class TestOCSPStaplingTLSOpenSSL(DNSDistOCSPStaplingTest):
 208
 209     _consoleKey = DNSDistTest.generateConsoleKey()
 210     _consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
 211     _serverKey = 'server-ocsp.key'
 212     _serverCert = 'server-ocsp.chain'
 213     _serverName = 'tls.tests.dnsdist.org'
 214     _ocspFile = 'server.ocsp'
 215     _caCert = 'ca.pem'
 216     _caKey = 'ca.key'
 217     _tlsServerPort = pickAvailablePort()
 218     _config_template = """
 219     newServer{address="127.0.0.1:%s"}
 220     setKey("%s")
 221     controlSocket("127.0.0.1:%s")
 222
 223     -- generate an OCSP response file for our certificate, valid one day
 224     generateOCSPResponse('%s', '%s', '%s', '%s', 1, 0)
 225     addTLSLocal("127.0.0.1:%s", "%s", "%s", { provider="openssl", ocspResponses={"%s"}})
 226     """
 227     _config_params = ['_testServerPort', '_consoleKeyB64', '_consolePort', '_serverCert', '_caCert', '_caKey', '_ocspFile', '_tlsServerPort', '_serverCert', '_serverKey', '_ocspFile']
 228
 229     def testOCSPStapling(self):
 230         """
 231         OCSP Stapling: TLS (OpenSSL)
 232         """
 233         output = self.checkOCSPStaplingStatus('127.0.0.1', self._tlsServerPort, self._serverName, self._caCert)
 234         self.assertIn('OCSP Response Status: successful (0x0)', output)
 235         self.assertEqual(self.getTLSProvider(), "openssl")
 236
 237         serialNumber = self.getOCSPSerial(output)
 238         self.assertTrue(serialNumber)
 239
 240         self.generateNewCertificateAndKey('server-ocsp')
 241         self.sendConsoleCommand("generateOCSPResponse('%s', '%s', '%s', '%s', 1, 0)" % (self._serverCert, self._caCert, self._caKey, self._ocspFile))
 242         self.sendConsoleCommand("reloadAllCertificates()")
 243
 244         output = self.checkOCSPStaplingStatus('127.0.0.1', self._tlsServerPort, self._serverName, self._caCert)
 245         self.assertIn('OCSP Response Status: successful (0x0)', output)
 246         serialNumber2 = self.getOCSPSerial(output)
 247         self.assertTrue(serialNumber2)
 248         self.assertNotEqual(serialNumber, serialNumber2)
 249
 250 class TestBrokenOCSPStaplingTLSOpenSSL(DNSDistOCSPStaplingTest):
 251
 252     _consoleKey = DNSDistTest.generateConsoleKey()
 253     _consoleKeyB64 = base64.b64encode(_consoleKey).decode('ascii')
 254     _serverKey = 'server-ocsp.key'
 255     _serverCert = 'server-ocsp.chain'
 256     _serverName = 'tls.tests.dnsdist.org'
 257     _caCert = 'ca.pem'
 258     # invalid OCSP file!
 259     _ocspFile = '/dev/null'
 260     _tlsServerPort = pickAvailablePort()
 261     _config_template = """
 262     newServer{address="127.0.0.1:%s"}
 263     setKey("%s")
 264     controlSocket("127.0.0.1:%s")
 265
 266     addTLSLocal("127.0.0.1:%s", "%s", "%s", { provider="openssl", ocspResponses={"%s"}})
 267     """
 268     _config_params = ['_testServerPort', '_consoleKeyB64', '_consolePort', '_tlsServerPort', '_serverCert', '_serverKey', '_ocspFile']
 269
 270     def testBrokenOCSPStapling(self):
 271         """
 272         OCSP Stapling: Broken (OpenSSL)
 273         """
 274         output = self.checkOCSPStaplingStatus('127.0.0.1', self._tlsServerPort, self._serverName, self._caCert)
 275         self.assertNotIn('OCSP Response Status: successful (0x0)', output)
 276         self.assertEqual(self.getTLSProvider(), "openssl")