minica.py - chromium/src/net/tools/testserver - Git at Google

 # Copyright (c) 2012 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.

 import asn1
 import datetime
 import hashlib
 import itertools
 import os
 import time

 GENERALIZED_TIME_FORMAT = "%Y%m%d%H%M%SZ"

 OCSP_STATE_GOOD = 1
 OCSP_STATE_REVOKED = 2
 OCSP_STATE_INVALID_RESPONSE = 3
 OCSP_STATE_UNAUTHORIZED = 4
 OCSP_STATE_UNKNOWN = 5
 OCSP_STATE_TRY_LATER = 6
 OCSP_STATE_INVALID_RESPONSE_DATA = 7
 OCSP_STATE_MISMATCHED_SERIAL = 8

 OCSP_DATE_VALID = 1
 OCSP_DATE_OLD = 2
 OCSP_DATE_EARLY = 3
 OCSP_DATE_LONG = 4

 OCSP_PRODUCED_VALID = 1
 OCSP_PRODUCED_BEFORE_CERT = 2
 OCSP_PRODUCED_AFTER_CERT = 3

 # This file implements very minimal certificate and OCSP generation. It's
 # designed to test revocation checking.

 def RandomNumber(length_in_bytes):
   '''RandomNumber returns a random number of length 8*|length_in_bytes| bits'''
   rand = os.urandom(length_in_bytes)
   n = 0
   for x in rand:
     n <<= 8
     n |= ord(x)
   return n


 def ModExp(n, e, p):
   '''ModExp returns n^e mod p'''
   r = 1
   while e != 0:
     if e & 1:
       r = (r*n) % p
     e >>= 1
     n = (n*n) % p
   return r

 # PKCS1v15_SHA256_PREFIX is the ASN.1 prefix for a SHA256 signature.
 PKCS1v15_SHA256_PREFIX = '3031300d060960864801650304020105000420'.decode('hex')

 class RSA(object):
   def __init__(self, modulus, e, d):
     self.m = modulus
     self.e = e
     self.d = d

     self.modlen = 0
     m = modulus
     while m != 0:
       self.modlen += 1
       m >>= 8

   def Sign(self, message):
     digest = hashlib.sha256(message).digest()
     prefix = PKCS1v15_SHA256_PREFIX

     em = ['\xff'] * (self.modlen - 1 - len(prefix) - len(digest))
     em[0] = '\x00'
     em[1] = '\x01'
     em += "\x00" + prefix + digest

     n = 0
     for x in em:
       n <<= 8
       n |= ord(x)

     s = ModExp(n, self.d, self.m)
     out = []
     while s != 0:
       out.append(s & 0xff)
       s >>= 8
     out.reverse()
     return '\x00' * (self.modlen - len(out)) + asn1.ToBytes(out)

   def ToDER(self):
     return asn1.ToDER(asn1.SEQUENCE([self.m, self.e]))


 def Name(cn = None, c = None, o = None):
   names = asn1.SEQUENCE([])

   if cn is not None:
     names.children.append(
       asn1.SET([
         asn1.SEQUENCE([
           COMMON_NAME, cn,
         ])
       ])
     )

   if c is not None:
     names.children.append(
       asn1.SET([
         asn1.SEQUENCE([
           COUNTRY, c,
         ])
       ])
     )

   if o is not None:
     names.children.append(
       asn1.SET([
         asn1.SEQUENCE([
           ORGANIZATION, o,
         ])
       ])
     )

   return names


 # The private key and root certificate name are hard coded here:

 # This is the private key
 KEY = RSA(0x00a71998f2930bfe73d031a87f133d2f378eeeeed52a77e44d0fc9ff6f07ff32cbf3da999de4ed65832afcb0807f98787506539d258a0ce3c2c77967653099a9034a9b115a876c39a8c4e4ed4acd0c64095946fb39eeeb47a0704dbb018acf48c3a1c4b895fc409fb4a340a986b1afc45519ab9eca47c30185c771c64aa5ecf07d,
           3,
           0x6f6665f70cb2a9a28acbc5aa0cd374cfb49f49e371a542de0a86aa4a0554cc87f7e71113edf399021ca875aaffbafaf8aee268c3b15ded2c84fb9a4375bbc6011d841e57833bc6f998d25daf6fa7f166b233e3e54a4bae7a5aaaba21431324967d5ff3e1d4f413827994262115ca54396e7068d0afa7af787a5782bc7040e6d3)

 # And the same thing in PEM format
 KEY_PEM = '''-----BEGIN RSA PRIVATE KEY-----
 MIICXAIBAAKBgQCnGZjykwv+c9AxqH8TPS83ju7u1Sp35E0Pyf9vB/8yy/PamZ3k
 7WWDKvywgH+YeHUGU50ligzjwsd5Z2UwmakDSpsRWodsOajE5O1KzQxkCVlG+znu
 60egcE27AYrPSMOhxLiV/ECftKNAqYaxr8RVGaueykfDAYXHccZKpezwfQIBAwKB
 gG9mZfcMsqmiisvFqgzTdM+0n0njcaVC3gqGqkoFVMyH9+cRE+3zmQIcqHWq/7r6
 +K7iaMOxXe0shPuaQ3W7xgEdhB5XgzvG+ZjSXa9vp/FmsjPj5UpLrnpaqrohQxMk
 ln1f8+HU9BOCeZQmIRXKVDlucGjQr6eveHpXgrxwQObTAkEA2wBAfuduw5G0/VfN
 Wx66D5fbPccfYFqLM5LuTimLmNqzK2gIKXckB2sm44gJZ6wVlumaB1CSNug2LNYx
 3cAjUwJBAMNUo1hbI8ugqqwI9kpxv9+2Heea4BlnXbS6tYF8pvkHMoliuxNbXmmB
 u4zNB5iZ6V0ZZ4nvtUNo2cGr/h/Lcu8CQQCSACr/RPSCYSNTj948vya1D+d+hL+V
 kbIiYfQ0G7Jl5yIc8AVw+hgE8hntBVuacrkPRmaviwwkms7IjsvpKsI3AkEAgjhs
 5ZIX3RXHHVtO3EvVP86+mmdAEO+TzdHOVlMZ+1ohsOx8t5I+8QEnszNaZbvw6Lua
 W/UjgkXmgR1UFTJMnwJBAKErmAw21/g3SST0a4wlyaGT/MbXL8Ouwnb5IOKQVe55
 CZdeVeSh6cJ4hAcQKfr2s1JaZTJFIBPGKAif5HqpydA=
 -----END RSA PRIVATE KEY-----
 '''

 # Root certificate CN
 ISSUER_CN = "Testing CA"

 # All certificates are issued under this policy OID, in the Google arc:
 CERT_POLICY_OID = asn1.OID([1, 3, 6, 1, 4, 1, 11129, 2, 4, 1])

 # These result in the following root certificate:
 # -----BEGIN CERTIFICATE-----
 # MIIB0TCCATqgAwIBAgIBATANBgkqhkiG9w0BAQUFADAVMRMwEQYDVQQDEwpUZXN0aW5nIENBMB4X
 # DTEwMDEwMTA2MDAwMFoXDTMyMTIwMTA2MDAwMFowFTETMBEGA1UEAxMKVGVzdGluZyBDQTCBnTAN
 # BgkqhkiG9w0BAQEFAAOBiwAwgYcCgYEApxmY8pML/nPQMah/Ez0vN47u7tUqd+RND8n/bwf/Msvz
 # 2pmd5O1lgyr8sIB/mHh1BlOdJYoM48LHeWdlMJmpA0qbEVqHbDmoxOTtSs0MZAlZRvs57utHoHBN
 # uwGKz0jDocS4lfxAn7SjQKmGsa/EVRmrnspHwwGFx3HGSqXs8H0CAQOjMzAxMBIGA1UdEwEB/wQI
 # MAYBAf8CAQAwGwYDVR0gAQEABBEwDzANBgsrBgEEAdZ5AgHODzANBgkqhkiG9w0BAQUFAAOBgQA/
 # STb40A6D+93jMfLGQzXc997IsaJZdoPt7tYa8PqGJBL62EiTj+erd/H5pDZx/2/bcpOG4m9J56yg
 # wOohbllw2TM+oeEd8syzV6X+1SIPnGI56JRrm3UXcHYx1Rq5loM9WKAiz/WmIWmskljsEQ7+542p
 # q0pkHjs8nuXovSkUYA==
 # -----END CERTIFICATE-----

 # If you update any of the above, you can generate a new root with the
 # following line:
 #   print DERToPEM(MakeCertificate(ISSUER_CN, ISSUER_CN, 1, KEY, KEY, None))


 # Various OIDs

 AIA_OCSP = asn1.OID([1, 3, 6, 1, 5, 5, 7, 48, 1])
 AUTHORITY_INFORMATION_ACCESS = asn1.OID([1, 3, 6, 1, 5, 5, 7, 1, 1])
 BASIC_CONSTRAINTS = asn1.OID([2, 5, 29, 19])
 CERT_POLICIES = asn1.OID([2, 5, 29, 32])
 COMMON_NAME = asn1.OID([2, 5, 4, 3])
 COUNTRY = asn1.OID([2, 5, 4, 6])
 HASH_SHA1 = asn1.OID([1, 3, 14, 3, 2, 26])
 OCSP_TYPE_BASIC = asn1.OID([1, 3, 6, 1, 5, 5, 7, 48, 1, 1])
 ORGANIZATION = asn1.OID([2, 5, 4, 10])
 PUBLIC_KEY_RSA = asn1.OID([1, 2, 840, 113549, 1, 1, 1])
 SHA256_WITH_RSA_ENCRYPTION = asn1.OID([1, 2, 840, 113549, 1, 1, 11])


 def MakeCertificate(
     issuer_cn, subject_cn, serial, pubkey, privkey, ocsp_url = None):
   '''MakeCertificate returns a DER encoded certificate, signed by privkey.'''
   extensions = asn1.SEQUENCE([])

   # Default subject name fields
   c = "XX"
   o = "Testing Org"

   if issuer_cn == subject_cn:
     # Root certificate.
     c = None
     o = None
     extensions.children.append(
       asn1.SEQUENCE([
         basic_constraints,
         True,
         asn1.OCTETSTRING(asn1.ToDER(asn1.SEQUENCE([
           True, # IsCA
           0, # Path len
         ]))),
       ]))

   if ocsp_url is not None:
     extensions.children.append(
       asn1.SEQUENCE([
         AUTHORITY_INFORMATION_ACCESS,
         False,
         asn1.OCTETSTRING(asn1.ToDER(asn1.SEQUENCE([
           asn1.SEQUENCE([
             AIA_OCSP,
             asn1.Raw(asn1.TagAndLength(0x86, len(ocsp_url)) + ocsp_url),
           ]),
         ]))),
       ]))

   extensions.children.append(
     asn1.SEQUENCE([
       CERT_POLICIES,
       False,
       asn1.OCTETSTRING(asn1.ToDER(asn1.SEQUENCE([
         asn1.SEQUENCE([ # PolicyInformation
           CERT_POLICY_OID,
         ]),
       ]))),
     ])
   )

   tbsCert = asn1.ToDER(asn1.SEQUENCE([
       asn1.Explicit(0, 2), # Version
       serial,
       asn1.SEQUENCE([SHA256_WITH_RSA_ENCRYPTION, None]), # SignatureAlgorithm
       Name(cn = issuer_cn), # Issuer
       asn1.SEQUENCE([ # Validity
         asn1.UTCTime("100101060000Z"), # NotBefore
         asn1.UTCTime("321201060000Z"), # NotAfter
       ]),
       Name(cn = subject_cn, c = c, o = o), # Subject
       asn1.SEQUENCE([ # SubjectPublicKeyInfo
         asn1.SEQUENCE([ # Algorithm
           PUBLIC_KEY_RSA,
           None,
         ]),
         asn1.BitString(asn1.ToDER(pubkey)),
       ]),
       asn1.Explicit(3, extensions),
     ]))

   return asn1.ToDER(asn1.SEQUENCE([
     asn1.Raw(tbsCert),
     asn1.SEQUENCE([
       SHA256_WITH_RSA_ENCRYPTION,
       None,
     ]),
     asn1.BitString(privkey.Sign(tbsCert)),
   ]))

 def MakeOCSPSingleResponse(
     issuer_name_hash, issuer_key_hash, serial, ocsp_state, ocsp_date):
   cert_status = None
   if ocsp_state == OCSP_STATE_REVOKED:
     cert_status = asn1.Explicit(1, asn1.GeneralizedTime("20100101060000Z"))
   elif ocsp_state == OCSP_STATE_UNKNOWN:
     cert_status = asn1.Raw(asn1.TagAndLength(0x80 | 2, 0))
   elif ocsp_state == OCSP_STATE_GOOD:
     cert_status = asn1.Raw(asn1.TagAndLength(0x80 | 0, 0))
   elif ocsp_state == OCSP_STATE_MISMATCHED_SERIAL:
     cert_status = asn1.Raw(asn1.TagAndLength(0x80 | 0, 0))
     serial -= 1
   else:
     raise ValueError('Bad OCSP state: ' + str(ocsp_state))

   now = datetime.datetime.fromtimestamp(time.mktime(time.gmtime()))
   if ocsp_date == OCSP_DATE_VALID:
     thisUpdate = now - datetime.timedelta(days=1)
     nextUpdate = thisUpdate + datetime.timedelta(weeks=1)
   elif ocsp_date == OCSP_DATE_OLD:
     thisUpdate = now - datetime.timedelta(hours=1, weeks=1)
     nextUpdate = thisUpdate + datetime.timedelta(weeks=1)
   elif ocsp_date == OCSP_DATE_EARLY:
     thisUpdate = now + datetime.timedelta(hours=1)
     nextUpdate = thisUpdate + datetime.timedelta(weeks=1)
   elif ocsp_date == OCSP_DATE_LONG:
     thisUpdate = now - datetime.timedelta(days=365)
     nextUpdate = thisUpdate + datetime.timedelta(hours=1, days=365)
   elif ocsp_date == OCSP_DATE_BEFORE_CERT:
     thisUpdate = now - datetime.timedelta(days=1)
     nextUpdate = thisUpdate + datetime.timedelta(weeks=1)
   elif ocsp_date == OCSP_DATE_AFTER_CERT:
     thisUpdate = now - datetime.timedelta(days=1)
     nextUpdate = thisUpdate + datetime.timedelta(weeks=1)
   else:
     raise ValueError('Bad OCSP date: ' + str(ocsp_date))

   return asn1.SEQUENCE([ # SingleResponse
     asn1.SEQUENCE([ # CertID
       asn1.SEQUENCE([ # hashAlgorithm
         HASH_SHA1,
         None,
       ]),
       issuer_name_hash,
       issuer_key_hash,
       serial,
     ]),
     cert_status,
     asn1.GeneralizedTime( # thisUpdate
       thisUpdate.strftime(GENERALIZED_TIME_FORMAT)
     ),
     asn1.Explicit( # nextUpdate
       0,
       asn1.GeneralizedTime(nextUpdate.strftime(GENERALIZED_TIME_FORMAT))
     ),
   ])

 def MakeOCSPResponse(
     issuer_cn, issuer_key, serial, ocsp_states, ocsp_dates, ocsp_produced):
   # https://tools.ietf.org/html/rfc2560
   issuer_name_hash = asn1.OCTETSTRING(
       hashlib.sha1(asn1.ToDER(Name(cn = issuer_cn))).digest())

   issuer_key_hash = asn1.OCTETSTRING(
       hashlib.sha1(asn1.ToDER(issuer_key)).digest())

   now = datetime.datetime.fromtimestamp(time.mktime(time.gmtime()))
   if ocsp_produced == OCSP_PRODUCED_VALID:
     producedAt = now - datetime.timedelta(days=1)
   elif ocsp_produced == OCSP_PRODUCED_BEFORE_CERT:
     producedAt = datetime.datetime.strptime(
         "19100101050000Z", GENERALIZED_TIME_FORMAT)
   elif ocsp_produced == OCSP_PRODUCED_AFTER_CERT:
     producedAt = datetime.datetime.strptime(
         "20321201070000Z", GENERALIZED_TIME_FORMAT)
   else:
     raise ValueError('Bad OCSP produced: ' + str(ocsp_produced))

   single_responses = [
       MakeOCSPSingleResponse(issuer_name_hash, issuer_key_hash, serial,
           ocsp_state, ocsp_date)
       for ocsp_state, ocsp_date in itertools.izip(ocsp_states, ocsp_dates)
   ]

   basic_resp_data_der = asn1.ToDER(asn1.SEQUENCE([
     asn1.Explicit(2, issuer_key_hash),
     asn1.GeneralizedTime(producedAt.strftime(GENERALIZED_TIME_FORMAT)),
     asn1.SEQUENCE(single_responses),
   ]))

   basic_resp = asn1.SEQUENCE([
     asn1.Raw(basic_resp_data_der),
     asn1.SEQUENCE([
       SHA256_WITH_RSA_ENCRYPTION,
       None,
     ]),
     asn1.BitString(issuer_key.Sign(basic_resp_data_der)),
   ])

   resp = asn1.SEQUENCE([
     asn1.ENUMERATED(0),
     asn1.Explicit(0, asn1.SEQUENCE([
       OCSP_TYPE_BASIC,
       asn1.OCTETSTRING(asn1.ToDER(basic_resp)),
     ]))
   ])

   return asn1.ToDER(resp)


 def DERToPEM(der):
   pem = '-----BEGIN CERTIFICATE-----\n'
   pem += der.encode('base64')
   pem += '-----END CERTIFICATE-----\n'
   return pem

 # unauthorizedDER is an OCSPResponse with a status of 6:
 # SEQUENCE { ENUM(6) }
 unauthorizedDER = '30030a0106'.decode('hex')

 def GenerateCertKeyAndOCSP(subject = "127.0.0.1",
                            ocsp_url = "http://127.0.0.1",
                            ocsp_states = None,
                            ocsp_dates = None,
                            ocsp_produced = OCSP_PRODUCED_VALID,
                            serial = 0):
   '''GenerateCertKeyAndOCSP returns a (cert_and_key_pem, ocsp_der) where:
        * cert_and_key_pem contains a certificate and private key in PEM format
          with the given subject common name and OCSP URL.
        * ocsp_der contains a DER encoded OCSP response or None if ocsp_url is
          None'''

   if ocsp_states is None:
     ocsp_states = [OCSP_STATE_GOOD]
   if ocsp_dates is None:
     ocsp_dates = [OCSP_DATE_VALID]

   if serial == 0:
     serial = RandomNumber(16)
   cert_der = MakeCertificate(ISSUER_CN, bytes(subject), serial, KEY, KEY,
                              bytes(ocsp_url))
   cert_pem = DERToPEM(cert_der)

   ocsp_der = None
   if ocsp_url is not None:
     if ocsp_states[0] == OCSP_STATE_UNAUTHORIZED:
       ocsp_der = unauthorizedDER
     elif ocsp_states[0] == OCSP_STATE_INVALID_RESPONSE:
       ocsp_der = '3'
     elif ocsp_states[0] == OCSP_STATE_TRY_LATER:
       resp = asn1.SEQUENCE([
         asn1.ENUMERATED(3),
       ])
       ocsp_der = asn1.ToDER(resp)
     elif ocsp_states[0] == OCSP_STATE_INVALID_RESPONSE_DATA:
       invalid_data = asn1.ToDER(asn1.OCTETSTRING('not ocsp data'))
       basic_resp = asn1.SEQUENCE([
         asn1.Raw(invalid_data),
         asn1.SEQUENCE([
           SHA256_WITH_RSA_ENCRYPTION,
           None,
         ]),
         asn1.BitString(KEY.Sign(invalid_data)),
       ])
       resp = asn1.SEQUENCE([
         asn1.ENUMERATED(0),
         asn1.Explicit(0, asn1.SEQUENCE([
           OCSP_TYPE_BASIC,
           asn1.OCTETSTRING(asn1.ToDER(basic_resp)),
         ])),
       ])
       ocsp_der = asn1.ToDER(resp)
     else:
       ocsp_der = MakeOCSPResponse(
           ISSUER_CN, KEY, serial, ocsp_states, ocsp_dates, ocsp_produced)

   return (cert_pem + KEY_PEM, ocsp_der)
	# Copyright (c) 2012 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.

	import asn1
	import datetime
	import hashlib
	import itertools
	import os
	import time

	GENERALIZED_TIME_FORMAT = "%Y%m%d%H%M%SZ"

	OCSP_STATE_GOOD = 1
	OCSP_STATE_REVOKED = 2
	OCSP_STATE_INVALID_RESPONSE = 3
	OCSP_STATE_UNAUTHORIZED = 4
	OCSP_STATE_UNKNOWN = 5
	OCSP_STATE_TRY_LATER = 6
	OCSP_STATE_INVALID_RESPONSE_DATA = 7
	OCSP_STATE_MISMATCHED_SERIAL = 8

	OCSP_DATE_VALID = 1
	OCSP_DATE_OLD = 2
	OCSP_DATE_EARLY = 3
	OCSP_DATE_LONG = 4

	OCSP_PRODUCED_VALID = 1
	OCSP_PRODUCED_BEFORE_CERT = 2
	OCSP_PRODUCED_AFTER_CERT = 3

	# This file implements very minimal certificate and OCSP generation. It's
	# designed to test revocation checking.

	def RandomNumber(length_in_bytes):
	'''RandomNumber returns a random number of length 8*\|length_in_bytes\| bits'''
	rand = os.urandom(length_in_bytes)
	n = 0
	for x in rand:
	n <<= 8
	n \|= ord(x)
	return n


	def ModExp(n, e, p):
	'''ModExp returns n^e mod p'''
	r = 1
	while e != 0:
	if e & 1:
	r = (r*n) % p
	e >>= 1
	n = (n*n) % p
	return r

	# PKCS1v15_SHA256_PREFIX is the ASN.1 prefix for a SHA256 signature.
	PKCS1v15_SHA256_PREFIX = '3031300d060960864801650304020105000420'.decode('hex')

	class RSA(object):
	def __init__(self, modulus, e, d):
	self.m = modulus
	self.e = e
	self.d = d

	self.modlen = 0
	m = modulus
	while m != 0:
	self.modlen += 1
	m >>= 8

	def Sign(self, message):
	digest = hashlib.sha256(message).digest()
	prefix = PKCS1v15_SHA256_PREFIX

	em = ['\xff'] * (self.modlen - 1 - len(prefix) - len(digest))
	em[0] = '\x00'
	em[1] = '\x01'
	em += "\x00" + prefix + digest

	n = 0
	for x in em:
	n <<= 8
	n \|= ord(x)

	s = ModExp(n, self.d, self.m)
	out = []
	while s != 0:
	out.append(s & 0xff)
	s >>= 8
	out.reverse()
	return '\x00' * (self.modlen - len(out)) + asn1.ToBytes(out)

	def ToDER(self):
	return asn1.ToDER(asn1.SEQUENCE([self.m, self.e]))


	def Name(cn = None, c = None, o = None):
	names = asn1.SEQUENCE([])

	if cn is not None:
	names.children.append(
	asn1.SET([
	asn1.SEQUENCE([
	COMMON_NAME, cn,
	])
	])
	)

	if c is not None:
	names.children.append(
	asn1.SET([
	asn1.SEQUENCE([
	COUNTRY, c,
	])
	])
	)

	if o is not None:
	names.children.append(
	asn1.SET([
	asn1.SEQUENCE([
	ORGANIZATION, o,
	])
	])
	)

	return names


	# The private key and root certificate name are hard coded here:

	# This is the private key
	KEY = RSA(0x00a71998f2930bfe73d031a87f133d2f378eeeeed52a77e44d0fc9ff6f07ff32cbf3da999de4ed65832afcb0807f98787506539d258a0ce3c2c77967653099a9034a9b115a876c39a8c4e4ed4acd0c64095946fb39eeeb47a0704dbb018acf48c3a1c4b895fc409fb4a340a986b1afc45519ab9eca47c30185c771c64aa5ecf07d,
	3,
	0x6f6665f70cb2a9a28acbc5aa0cd374cfb49f49e371a542de0a86aa4a0554cc87f7e71113edf399021ca875aaffbafaf8aee268c3b15ded2c84fb9a4375bbc6011d841e57833bc6f998d25daf6fa7f166b233e3e54a4bae7a5aaaba21431324967d5ff3e1d4f413827994262115ca54396e7068d0afa7af787a5782bc7040e6d3)

	# And the same thing in PEM format
	KEY_PEM = '''-----BEGIN RSA PRIVATE KEY-----
	MIICXAIBAAKBgQCnGZjykwv+c9AxqH8TPS83ju7u1Sp35E0Pyf9vB/8yy/PamZ3k
	7WWDKvywgH+YeHUGU50ligzjwsd5Z2UwmakDSpsRWodsOajE5O1KzQxkCVlG+znu
	60egcE27AYrPSMOhxLiV/ECftKNAqYaxr8RVGaueykfDAYXHccZKpezwfQIBAwKB
	gG9mZfcMsqmiisvFqgzTdM+0n0njcaVC3gqGqkoFVMyH9+cRE+3zmQIcqHWq/7r6
	+K7iaMOxXe0shPuaQ3W7xgEdhB5XgzvG+ZjSXa9vp/FmsjPj5UpLrnpaqrohQxMk
	ln1f8+HU9BOCeZQmIRXKVDlucGjQr6eveHpXgrxwQObTAkEA2wBAfuduw5G0/VfN
	Wx66D5fbPccfYFqLM5LuTimLmNqzK2gIKXckB2sm44gJZ6wVlumaB1CSNug2LNYx
	3cAjUwJBAMNUo1hbI8ugqqwI9kpxv9+2Heea4BlnXbS6tYF8pvkHMoliuxNbXmmB
	u4zNB5iZ6V0ZZ4nvtUNo2cGr/h/Lcu8CQQCSACr/RPSCYSNTj948vya1D+d+hL+V
	kbIiYfQ0G7Jl5yIc8AVw+hgE8hntBVuacrkPRmaviwwkms7IjsvpKsI3AkEAgjhs
	5ZIX3RXHHVtO3EvVP86+mmdAEO+TzdHOVlMZ+1ohsOx8t5I+8QEnszNaZbvw6Lua
	W/UjgkXmgR1UFTJMnwJBAKErmAw21/g3SST0a4wlyaGT/MbXL8Ouwnb5IOKQVe55
	CZdeVeSh6cJ4hAcQKfr2s1JaZTJFIBPGKAif5HqpydA=
	-----END RSA PRIVATE KEY-----
	'''

	# Root certificate CN
	ISSUER_CN = "Testing CA"

	# All certificates are issued under this policy OID, in the Google arc:
	CERT_POLICY_OID = asn1.OID([1, 3, 6, 1, 4, 1, 11129, 2, 4, 1])

	# These result in the following root certificate:
	# -----BEGIN CERTIFICATE-----
	# MIIB0TCCATqgAwIBAgIBATANBgkqhkiG9w0BAQUFADAVMRMwEQYDVQQDEwpUZXN0aW5nIENBMB4X
	# DTEwMDEwMTA2MDAwMFoXDTMyMTIwMTA2MDAwMFowFTETMBEGA1UEAxMKVGVzdGluZyBDQTCBnTAN
	# BgkqhkiG9w0BAQEFAAOBiwAwgYcCgYEApxmY8pML/nPQMah/Ez0vN47u7tUqd+RND8n/bwf/Msvz
	# 2pmd5O1lgyr8sIB/mHh1BlOdJYoM48LHeWdlMJmpA0qbEVqHbDmoxOTtSs0MZAlZRvs57utHoHBN
	# uwGKz0jDocS4lfxAn7SjQKmGsa/EVRmrnspHwwGFx3HGSqXs8H0CAQOjMzAxMBIGA1UdEwEB/wQI
	# MAYBAf8CAQAwGwYDVR0gAQEABBEwDzANBgsrBgEEAdZ5AgHODzANBgkqhkiG9w0BAQUFAAOBgQA/
	# STb40A6D+93jMfLGQzXc997IsaJZdoPt7tYa8PqGJBL62EiTj+erd/H5pDZx/2/bcpOG4m9J56yg
	# wOohbllw2TM+oeEd8syzV6X+1SIPnGI56JRrm3UXcHYx1Rq5loM9WKAiz/WmIWmskljsEQ7+542p
	# q0pkHjs8nuXovSkUYA==
	# -----END CERTIFICATE-----

	# If you update any of the above, you can generate a new root with the
	# following line:
	# print DERToPEM(MakeCertificate(ISSUER_CN, ISSUER_CN, 1, KEY, KEY, None))


	# Various OIDs

	AIA_OCSP = asn1.OID([1, 3, 6, 1, 5, 5, 7, 48, 1])
	AUTHORITY_INFORMATION_ACCESS = asn1.OID([1, 3, 6, 1, 5, 5, 7, 1, 1])
	BASIC_CONSTRAINTS = asn1.OID([2, 5, 29, 19])
	CERT_POLICIES = asn1.OID([2, 5, 29, 32])
	COMMON_NAME = asn1.OID([2, 5, 4, 3])
	COUNTRY = asn1.OID([2, 5, 4, 6])
	HASH_SHA1 = asn1.OID([1, 3, 14, 3, 2, 26])
	OCSP_TYPE_BASIC = asn1.OID([1, 3, 6, 1, 5, 5, 7, 48, 1, 1])
	ORGANIZATION = asn1.OID([2, 5, 4, 10])
	PUBLIC_KEY_RSA = asn1.OID([1, 2, 840, 113549, 1, 1, 1])
	SHA256_WITH_RSA_ENCRYPTION = asn1.OID([1, 2, 840, 113549, 1, 1, 11])


	def MakeCertificate(
	issuer_cn, subject_cn, serial, pubkey, privkey, ocsp_url = None):
	'''MakeCertificate returns a DER encoded certificate, signed by privkey.'''
	extensions = asn1.SEQUENCE([])

	# Default subject name fields
	c = "XX"
	o = "Testing Org"

	if issuer_cn == subject_cn:
	# Root certificate.
	c = None
	o = None
	extensions.children.append(
	asn1.SEQUENCE([
	basic_constraints,
	True,
	asn1.OCTETSTRING(asn1.ToDER(asn1.SEQUENCE([
	True, # IsCA
	0, # Path len
	]))),
	]))

	if ocsp_url is not None:
	extensions.children.append(
	asn1.SEQUENCE([
	AUTHORITY_INFORMATION_ACCESS,
	False,
	asn1.OCTETSTRING(asn1.ToDER(asn1.SEQUENCE([
	asn1.SEQUENCE([
	AIA_OCSP,
	asn1.Raw(asn1.TagAndLength(0x86, len(ocsp_url)) + ocsp_url),
	]),
	]))),
	]))

	extensions.children.append(
	asn1.SEQUENCE([
	CERT_POLICIES,
	False,
	asn1.OCTETSTRING(asn1.ToDER(asn1.SEQUENCE([
	asn1.SEQUENCE([ # PolicyInformation
	CERT_POLICY_OID,
	]),
	]))),
	])
	)

	tbsCert = asn1.ToDER(asn1.SEQUENCE([
	asn1.Explicit(0, 2), # Version
	serial,
	asn1.SEQUENCE([SHA256_WITH_RSA_ENCRYPTION, None]), # SignatureAlgorithm
	Name(cn = issuer_cn), # Issuer
	asn1.SEQUENCE([ # Validity
	asn1.UTCTime("100101060000Z"), # NotBefore
	asn1.UTCTime("321201060000Z"), # NotAfter
	]),
	Name(cn = subject_cn, c = c, o = o), # Subject
	asn1.SEQUENCE([ # SubjectPublicKeyInfo
	asn1.SEQUENCE([ # Algorithm
	PUBLIC_KEY_RSA,
	None,
	]),
	asn1.BitString(asn1.ToDER(pubkey)),
	]),
	asn1.Explicit(3, extensions),
	]))

	return asn1.ToDER(asn1.SEQUENCE([
	asn1.Raw(tbsCert),
	asn1.SEQUENCE([
	SHA256_WITH_RSA_ENCRYPTION,
	None,
	]),
	asn1.BitString(privkey.Sign(tbsCert)),
	]))

	def MakeOCSPSingleResponse(
	issuer_name_hash, issuer_key_hash, serial, ocsp_state, ocsp_date):
	cert_status = None
	if ocsp_state == OCSP_STATE_REVOKED:
	cert_status = asn1.Explicit(1, asn1.GeneralizedTime("20100101060000Z"))
	elif ocsp_state == OCSP_STATE_UNKNOWN:
	cert_status = asn1.Raw(asn1.TagAndLength(0x80 \| 2, 0))
	elif ocsp_state == OCSP_STATE_GOOD:
	cert_status = asn1.Raw(asn1.TagAndLength(0x80 \| 0, 0))
	elif ocsp_state == OCSP_STATE_MISMATCHED_SERIAL:
	cert_status = asn1.Raw(asn1.TagAndLength(0x80 \| 0, 0))
	serial -= 1
	else:
	raise ValueError('Bad OCSP state: ' + str(ocsp_state))

	now = datetime.datetime.fromtimestamp(time.mktime(time.gmtime()))
	if ocsp_date == OCSP_DATE_VALID:
	thisUpdate = now - datetime.timedelta(days=1)
	nextUpdate = thisUpdate + datetime.timedelta(weeks=1)
	elif ocsp_date == OCSP_DATE_OLD:
	thisUpdate = now - datetime.timedelta(hours=1, weeks=1)
	nextUpdate = thisUpdate + datetime.timedelta(weeks=1)
	elif ocsp_date == OCSP_DATE_EARLY:
	thisUpdate = now + datetime.timedelta(hours=1)
	nextUpdate = thisUpdate + datetime.timedelta(weeks=1)
	elif ocsp_date == OCSP_DATE_LONG:
	thisUpdate = now - datetime.timedelta(days=365)
	nextUpdate = thisUpdate + datetime.timedelta(hours=1, days=365)
	elif ocsp_date == OCSP_DATE_BEFORE_CERT:
	thisUpdate = now - datetime.timedelta(days=1)
	nextUpdate = thisUpdate + datetime.timedelta(weeks=1)
	elif ocsp_date == OCSP_DATE_AFTER_CERT:
	thisUpdate = now - datetime.timedelta(days=1)
	nextUpdate = thisUpdate + datetime.timedelta(weeks=1)
	else:
	raise ValueError('Bad OCSP date: ' + str(ocsp_date))

	return asn1.SEQUENCE([ # SingleResponse
	asn1.SEQUENCE([ # CertID
	asn1.SEQUENCE([ # hashAlgorithm
	HASH_SHA1,
	None,
	]),
	issuer_name_hash,
	issuer_key_hash,
	serial,
	]),
	cert_status,
	asn1.GeneralizedTime( # thisUpdate
	thisUpdate.strftime(GENERALIZED_TIME_FORMAT)
	),
	asn1.Explicit( # nextUpdate
	0,
	asn1.GeneralizedTime(nextUpdate.strftime(GENERALIZED_TIME_FORMAT))
	),
	])

	def MakeOCSPResponse(
	issuer_cn, issuer_key, serial, ocsp_states, ocsp_dates, ocsp_produced):
	# https://tools.ietf.org/html/rfc2560
	issuer_name_hash = asn1.OCTETSTRING(
	hashlib.sha1(asn1.ToDER(Name(cn = issuer_cn))).digest())

	issuer_key_hash = asn1.OCTETSTRING(
	hashlib.sha1(asn1.ToDER(issuer_key)).digest())

	now = datetime.datetime.fromtimestamp(time.mktime(time.gmtime()))
	if ocsp_produced == OCSP_PRODUCED_VALID:
	producedAt = now - datetime.timedelta(days=1)
	elif ocsp_produced == OCSP_PRODUCED_BEFORE_CERT:
	producedAt = datetime.datetime.strptime(
	"19100101050000Z", GENERALIZED_TIME_FORMAT)
	elif ocsp_produced == OCSP_PRODUCED_AFTER_CERT:
	producedAt = datetime.datetime.strptime(
	"20321201070000Z", GENERALIZED_TIME_FORMAT)
	else:
	raise ValueError('Bad OCSP produced: ' + str(ocsp_produced))

	single_responses = [
	MakeOCSPSingleResponse(issuer_name_hash, issuer_key_hash, serial,
	ocsp_state, ocsp_date)
	for ocsp_state, ocsp_date in itertools.izip(ocsp_states, ocsp_dates)
	]

	basic_resp_data_der = asn1.ToDER(asn1.SEQUENCE([
	asn1.Explicit(2, issuer_key_hash),
	asn1.GeneralizedTime(producedAt.strftime(GENERALIZED_TIME_FORMAT)),
	asn1.SEQUENCE(single_responses),
	]))

	basic_resp = asn1.SEQUENCE([
	asn1.Raw(basic_resp_data_der),
	asn1.SEQUENCE([
	SHA256_WITH_RSA_ENCRYPTION,
	None,
	]),
	asn1.BitString(issuer_key.Sign(basic_resp_data_der)),
	])

	resp = asn1.SEQUENCE([
	asn1.ENUMERATED(0),
	asn1.Explicit(0, asn1.SEQUENCE([
	OCSP_TYPE_BASIC,
	asn1.OCTETSTRING(asn1.ToDER(basic_resp)),
	]))
	])

	return asn1.ToDER(resp)


	def DERToPEM(der):
	pem = '-----BEGIN CERTIFICATE-----\n'
	pem += der.encode('base64')
	pem += '-----END CERTIFICATE-----\n'
	return pem

	# unauthorizedDER is an OCSPResponse with a status of 6:
	# SEQUENCE { ENUM(6) }
	unauthorizedDER = '30030a0106'.decode('hex')

	def GenerateCertKeyAndOCSP(subject = "127.0.0.1",
	ocsp_url = "http://127.0.0.1",
	ocsp_states = None,
	ocsp_dates = None,
	ocsp_produced = OCSP_PRODUCED_VALID,
	serial = 0):
	'''GenerateCertKeyAndOCSP returns a (cert_and_key_pem, ocsp_der) where:
	* cert_and_key_pem contains a certificate and private key in PEM format
	with the given subject common name and OCSP URL.
	* ocsp_der contains a DER encoded OCSP response or None if ocsp_url is
	None'''

	if ocsp_states is None:
	ocsp_states = [OCSP_STATE_GOOD]
	if ocsp_dates is None:
	ocsp_dates = [OCSP_DATE_VALID]

	if serial == 0:
	serial = RandomNumber(16)
	cert_der = MakeCertificate(ISSUER_CN, bytes(subject), serial, KEY, KEY,
	bytes(ocsp_url))
	cert_pem = DERToPEM(cert_der)

	ocsp_der = None
	if ocsp_url is not None:
	if ocsp_states[0] == OCSP_STATE_UNAUTHORIZED:
	ocsp_der = unauthorizedDER
	elif ocsp_states[0] == OCSP_STATE_INVALID_RESPONSE:
	ocsp_der = '3'
	elif ocsp_states[0] == OCSP_STATE_TRY_LATER:
	resp = asn1.SEQUENCE([
	asn1.ENUMERATED(3),
	])
	ocsp_der = asn1.ToDER(resp)
	elif ocsp_states[0] == OCSP_STATE_INVALID_RESPONSE_DATA:
	invalid_data = asn1.ToDER(asn1.OCTETSTRING('not ocsp data'))
	basic_resp = asn1.SEQUENCE([
	asn1.Raw(invalid_data),
	asn1.SEQUENCE([
	SHA256_WITH_RSA_ENCRYPTION,
	None,
	]),
	asn1.BitString(KEY.Sign(invalid_data)),
	])
	resp = asn1.SEQUENCE([
	asn1.ENUMERATED(0),
	asn1.Explicit(0, asn1.SEQUENCE([
	OCSP_TYPE_BASIC,
	asn1.OCTETSTRING(asn1.ToDER(basic_resp)),
	])),
	])
	ocsp_der = asn1.ToDER(resp)
	else:
	ocsp_der = MakeOCSPResponse(
	ISSUER_CN, KEY, serial, ocsp_states, ocsp_dates, ocsp_produced)

	return (cert_pem + KEY_PEM, ocsp_der)