Add OCSPVerifyResult for tracking stapled OCSP responses cross-platform.

net/tools/testserver/testserver.py

 #!/usr/bin/env python
 # Copyright 2013 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 """This is a simple HTTP/FTP/TCP/UDP/BASIC_AUTH_PROXY/WEBSOCKET server used for
 testing Chrome.
 
 It supports several test URLs, as specified by the handlers in TestPageHandler.
 By default, it listens on an ephemeral port and sends the port number back to
@@ skipping 62 matching lines @@
 SERVER_HTTP = 0
 SERVER_FTP = 1
 SERVER_TCP_ECHO = 2
 SERVER_UDP_ECHO = 3
 SERVER_BASIC_AUTH_PROXY = 4
 SERVER_WEBSOCKET = 5
 
 # Default request queue size for WebSocketServer.
 _DEFAULT_REQUEST_QUEUE_SIZE = 128
 
+OCSP_STATES_NO_SINGLE_RESPONSE = {
+  minica.OCSP_STATE_INVALID_RESPONSE,
+  minica.OCSP_STATE_UNAUTHORIZED,
+  minica.OCSP_STATE_TRY_LATER,
+  minica.OCSP_STATE_INVALID_RESPONSE_DATA,
+}
+
 class WebSocketOptions:
   """Holds options for WebSocketServer."""
 
   def __init__(self, host, port, data_dir):
     self.request_queue_size = _DEFAULT_REQUEST_QUEUE_SIZE
     self.server_host = host
     self.port = port
     self.websock_handlers = data_dir
     self.scan_dir = None
     self.allow_handlers_outside_root_dir = False
@@ skipping 1796 matching lines @@
             raise testserver_base.OptionError(
                 'specified server cert file not found: ' +
                 self.options.cert_and_key_file + ' exiting...')
           pem_cert_and_key = file(self.options.cert_and_key_file, 'r').read()
         else:
           # generate a new certificate and run an OCSP server for it.
           self.__ocsp_server = OCSPServer((host, 0), OCSPHandler)
           print ('OCSP server started on %s:%d...' %
               (host, self.__ocsp_server.server_port))
 
-          ocsp_state = None
+          ocsp_states = list()
+          for ocsp_state_arg in self.options.ocsp.split(':'):
+            if ocsp_state_arg == 'ok':
+              ocsp_state = minica.OCSP_STATE_GOOD
+            elif ocsp_state_arg == 'revoked':
+              ocsp_state = minica.OCSP_STATE_REVOKED
+            elif ocsp_state_arg == 'invalid':
+              ocsp_state = minica.OCSP_STATE_INVALID_RESPONSE
+            elif ocsp_state_arg == 'unauthorized':
+              ocsp_state = minica.OCSP_STATE_UNAUTHORIZED
+            elif ocsp_state_arg == 'unknown':
+              ocsp_state = minica.OCSP_STATE_UNKNOWN
+            elif ocsp_state_arg == 'later':
+              ocsp_state = minica.OCSP_STATE_TRY_LATER
+            elif ocsp_state_arg == 'invalid_data':
+              ocsp_state = minica.OCSP_STATE_INVALID_RESPONSE_DATA
+            elif ocsp_state_arg == "mismatched_serial":
+              ocsp_state = minica.OCSP_STATE_MISMATCHED_SERIAL
+            else:
+              raise testserver_base.OptionError('unknown OCSP status: ' +
+                  ocsp_state_arg)
+            ocsp_states.append(ocsp_state)
 
-          if self.options.ocsp == 'ok':
-            ocsp_state = minica.OCSP_STATE_GOOD
-          elif self.options.ocsp == 'revoked':
-            ocsp_state = minica.OCSP_STATE_REVOKED
-          elif self.options.ocsp == 'invalid':
-            ocsp_state = minica.OCSP_STATE_INVALID
-          elif self.options.ocsp == 'unauthorized':
-            ocsp_state = minica.OCSP_STATE_UNAUTHORIZED
-          elif self.options.ocsp == 'unknown':
-            ocsp_state = minica.OCSP_STATE_UNKNOWN
+          if len(ocsp_states) > 1:
+            if set(ocsp_states) & OCSP_STATES_NO_SINGLE_RESPONSE:
+              raise testserver_base.OptionError('Multiple OCSP responses '
+                  'incompatible with states ' + str(ocsp_states))
+
+          ocsp_dates = list()
+          for ocsp_date_arg in self.options.ocsp_date.split(':'):
+            if ocsp_date_arg == 'valid':
+              ocsp_date = minica.OCSP_DATE_VALID
+            elif ocsp_date_arg == 'old':
+              ocsp_date = minica.OCSP_DATE_OLD
+            elif ocsp_date_arg == 'early':
+              ocsp_date = minica.OCSP_DATE_EARLY
+            elif ocsp_date_arg == 'long':
+              ocsp_date = minica.OCSP_DATE_LONG
+            elif ocsp_date_arg == 'before_cert':
+              ocsp_date = minica.OCSP_DATE_AFTER_CERT
+            elif ocsp_date_arg == 'after_cert':
+              ocsp_date = minica.OCSP_DATE_AFTER_CERT
+            else:
+              raise testserver_base.OptionError('unknown OCSP date: ' +
+                  ocsp_date_arg)
+            ocsp_dates.append(ocsp_date)
+
+          if len(ocsp_states) != len(ocsp_dates):
+            raise testserver_base.OptionError('mismatched ocsp and ocsp-date '
+                'count')
+
+          ocsp_produced = None
+          if self.options.ocsp_produced == 'valid':
+            ocsp_produced = minica.OCSP_PRODUCED_VALID
+          elif self.options.ocsp_produced == 'before':
+            ocsp_produced = minica.OCSP_PRODUCED_BEFORE_CERT
+          elif self.options.ocsp_produced == 'after':
+            ocsp_produced = minica.OCSP_PRODUCED_AFTER_CERT
           else:
-            raise testserver_base.OptionError('unknown OCSP status: ' +
-                self.options.ocsp_status)
+            raise testserver_base.OptionError('unknown OCSP produced: ' +
+                self.options.ocsp_produced)
 
           (pem_cert_and_key, ocsp_der) = minica.GenerateCertKeyAndOCSP(
               subject = "127.0.0.1",
               ocsp_url = ("http://%s:%d/ocsp" %
                   (host, self.__ocsp_server.server_port)),
-              ocsp_state = ocsp_state,
+              ocsp_states = ocsp_states,
+              ocsp_dates = ocsp_dates,
+              ocsp_produced = ocsp_produced,
               serial = self.options.cert_serial)
 
           if self.options.ocsp_server_unavailable:
             # SEQUENCE containing ENUMERATED with value 3 (tryLater).
             self.__ocsp_server.ocsp_response = '30030a0103'.decode('hex')
           else:
             self.__ocsp_server.ocsp_response = ocsp_der
 
         for ca_cert in self.options.ssl_client_ca:
           if not os.path.isfile(ca_cert):
@@ skipping 151 matching lines @@
                                   'should be used.')
     self.option_parser.add_option('--cert-and-key-file',
                                   dest='cert_and_key_file', help='specify the '
                                   'path to the file containing the certificate '
                                   'and private key for the server in PEM '
                                   'format')
     self.option_parser.add_option('--ocsp', dest='ocsp', default='ok',
                                   help='The type of OCSP response generated '
                                   'for the automatically generated '
                                   'certificate. One of [ok,revoked,invalid]')
+    self.option_parser.add_option('--ocsp-date', dest='ocsp_date',
+                                  default='valid', help='The validity of the '
+                                  'range between thisUpdate and nextUpdate')
+    self.option_parser.add_option('--ocsp-produced', dest='ocsp_produced',
+                                  default='valid', help='producedAt relative '
+                                  'to certificate expiry')
     self.option_parser.add_option('--cert-serial', dest='cert_serial',
                                   default=0, type=int,
                                   help='If non-zero then the generated '
                                   'certificate will have this serial number')
     self.option_parser.add_option('--tls-intolerant', dest='tls_intolerant',
                                   default='0', type='int',
                                   help='If nonzero, certain TLS connections '
                                   'will be aborted in order to test version '
                                   'fallback. 1 means all TLS versions will be '
                                   'aborted. 2 means TLS 1.1 or higher will be '
@@ skipping 97 matching lines @@
                                   'an anonymous user.')
     self.option_parser.add_option('--disable-channel-id', action='store_true')
     self.option_parser.add_option('--disable-extended-master-secret',
                                   action='store_true')
     self.option_parser.add_option('--token-binding-params', action='append',
                                   default=[], type='int')
 
 
 if __name__ == '__main__':
   sys.exit(ServerRunner().main())