blob: 7319e5791f9d89813eb08c3c6f61714d75c84376 [file] [log] [blame]
#!/usr/bin/env python
# Copyright 2017 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""This script is called without any arguments to re-generate all of the *.pem
files in the script's parent directory.
"""
from pyasn1.codec.der import decoder, encoder
from pyasn1_modules import rfc2560, rfc2459
from pyasn1.type import univ, useful
import hashlib, datetime
import subprocess
import os
from OpenSSL import crypto
import base64
NEXT_SERIAL = 0
# 1/1/2017 00:00 GMT
CERT_DATE = datetime.datetime(2017, 1, 1, 0, 0)
# 1/1/2018 00:00 GMT
CERT_EXPIRE = CERT_DATE + datetime.timedelta(days=365)
# 2/1/2017 00:00 GMT
REVOKE_DATE = datetime.datetime(2017, 2, 1, 0, 0)
# 3/1/2017 00:00 GMT
THIS_DATE = datetime.datetime(2017, 3, 1, 0, 0)
# 3/2/2017 00:00 GMT
PRODUCED_DATE = datetime.datetime(2017, 3, 2, 0, 0)
# 6/1/2017 00:00 GMT
NEXT_DATE = datetime.datetime(2017, 6, 1, 0, 0)
sha1oid = univ.ObjectIdentifier('1.3.14.3.2.26')
sha1rsaoid = univ.ObjectIdentifier('1.2.840.113549.1.1.5')
sha256oid = univ.ObjectIdentifier('2.16.840.1.101.3.4.2.1')
sha256rsaoid = univ.ObjectIdentifier('1.2.840.113549.1.1.11')
def SigAlgOid(sig_alg):
if sig_alg == 'sha1':
return sha1rsaoid
return sha256rsaoid
def CreateCert(name, signer=None, ocsp=False):
global NEXT_SERIAL
pkey = crypto.PKey()
pkey.generate_key(crypto.TYPE_RSA, 1024)
cert = crypto.X509()
cert.set_version(2)
cert.get_subject().CN = name
cert.set_pubkey(pkey)
cert.set_serial_number(NEXT_SERIAL)
NEXT_SERIAL += 1
cert.set_notBefore(CERT_DATE.strftime('%Y%m%d%H%M%SZ'))
cert.set_notAfter(CERT_EXPIRE.strftime('%Y%m%d%H%M%SZ'))
if ocsp:
cert.add_extensions(
[crypto.X509Extension('extendedKeyUsage', False, 'OCSPSigning')])
if signer:
cert.set_issuer(signer[1].get_subject())
cert.sign(signer[2], 'sha1')
else:
cert.set_issuer(cert.get_subject())
cert.sign(pkey, 'sha1')
asn1cert = decoder.decode(
crypto.dump_certificate(crypto.FILETYPE_ASN1, cert),
asn1Spec=rfc2459.Certificate())[0]
if not signer:
signer = [asn1cert]
return (asn1cert, cert, pkey, signer[0])
def CreateExtension(oid='1.2.3.4', critical=False):
ext = rfc2459.Extension()
ext.setComponentByName('extnID', univ.ObjectIdentifier(oid))
ext.setComponentByName('extnValue', 'DEADBEEF')
if critical:
ext.setComponentByName('critical', univ.Boolean('True'))
else:
ext.setComponentByName('critical', univ.Boolean('False'))
return ext
ROOT_CA = CreateCert('Test CA', None)
CA = CreateCert('Test Intermediate CA', ROOT_CA)
CA_LINK = CreateCert('Test OCSP Signer', CA, True)
CA_BADLINK = CreateCert('Test False OCSP Signer', CA, False)
CERT = CreateCert('Test Cert', CA)
JUNK_CERT = CreateCert('Random Cert', None)
EXTENSION = CreateExtension()
def GetName(c):
rid = rfc2560.ResponderID()
subject = c[0].getComponentByName('tbsCertificate').getComponentByName(
'subject')
rn = rid.componentType.getTypeByPosition(0).clone()
for i in range(len(subject)):
rn.setComponentByPosition(i, subject.getComponentByPosition(i))
rid.setComponentByName('byName', rn)
return rid
def GetKeyHash(c):
rid = rfc2560.ResponderID()
spk = c[0].getComponentByName('tbsCertificate').getComponentByName(
'subjectPublicKeyInfo').getComponentByName('subjectPublicKey')
keyHash = hashlib.sha1(encoder.encode(spk)[4:]).digest()
rid.setComponentByName('byKey', keyHash)
return rid
def CreateSingleResponse(cert=CERT,
status=0,
next=None,
revoke_time=None,
reason=None,
extensions=[]):
sr = rfc2560.SingleResponse()
cid = sr.setComponentByName('certID').getComponentByName('certID')
issuer_tbs = cert[3].getComponentByName('tbsCertificate')
tbs = cert[0].getComponentByName('tbsCertificate')
name_hash = hashlib.sha1(
encoder.encode(issuer_tbs.getComponentByName('subject'))).digest()
key_hash = hashlib.sha1(
encoder.encode(
issuer_tbs.getComponentByName('subjectPublicKeyInfo')
.getComponentByName('subjectPublicKey'))[4:]).digest()
sn = tbs.getComponentByName('serialNumber')
ha = cid.setComponentByName('hashAlgorithm').getComponentByName(
'hashAlgorithm')
ha.setComponentByName('algorithm', sha1oid)
cid.setComponentByName('issuerNameHash', name_hash)
cid.setComponentByName('issuerKeyHash', key_hash)
cid.setComponentByName('serialNumber', sn)
cs = rfc2560.CertStatus()
if status == 0:
cs.setComponentByName('good')
elif status == 1:
ri = cs.componentType.getTypeByPosition(1).clone()
if revoke_time == None:
revoke_time = REVOKE_DATE
ri.setComponentByName('revocationTime',
useful.GeneralizedTime(
revoke_time.strftime('%Y%m%d%H%M%SZ')))
if reason:
ri.setComponentByName('revocationReason', reason)
cs.setComponentByName('revoked', ri)
else:
ui = cs.componentType.getTypeByPosition(2).clone()
cs.setComponentByName('unknown', ui)
sr.setComponentByName('certStatus', cs)
sr.setComponentByName('thisUpdate',
useful.GeneralizedTime(
THIS_DATE.strftime('%Y%m%d%H%M%SZ')))
if next:
sr.setComponentByName('nextUpdate', next.strftime('%Y%m%d%H%M%SZ'))
if extensions:
elist = sr.setComponentByName('singleExtensions').getComponentByName(
'singleExtensions')
for i in range(len(extensions)):
elist.setComponentByPosition(i, extensions[i])
return sr
def Create(signer=None,
response_status=0,
response_type='1.3.6.1.5.5.7.48.1.1',
signature=None,
version=1,
responder=None,
responses=None,
extensions=None,
certs=None,
sigAlg='sha1'):
ocsp = rfc2560.OCSPResponse()
ocsp.setComponentByName('responseStatus', response_status)
if response_status != 0:
return ocsp
tbs = rfc2560.ResponseData()
if version != 1:
tbs.setComponentByName('version', version)
if not signer:
signer = CA
if not responder:
responder = GetName(signer)
tbs.setComponentByName('responderID', responder)
tbs.setComponentByName('producedAt',
useful.GeneralizedTime(
PRODUCED_DATE.strftime('%Y%m%d%H%M%SZ')))
rlist = tbs.setComponentByName('responses').getComponentByName('responses')
if responses == None:
responses = [CreateSingleResponse(CERT, 0)]
if responses:
for i in range(len(responses)):
rlist.setComponentByPosition(i, responses[i])
if extensions:
elist = tbs.setComponentByName('responseExtensions').getComponentByName(
'responseExtensions')
for i in range(len(extensions)):
elist.setComponentByPosition(i, extensions[i])
sa = rfc2459.AlgorithmIdentifier()
sa.setComponentByName('algorithm', SigAlgOid(sigAlg))
# TODO(mattm): If pyasn1 gives an error
# "Component value is tag-incompatible: Null() vs Any()", try hacking
# pyasn1_modules/rfc2459.py's AlgorithmIdentifier to specify univ.Null as the
# type for 'parameters'. (Which is an ugly hack, but lets the script work.)
sa.setComponentByName('parameters', univ.Null())
basic = rfc2560.BasicOCSPResponse()
basic.setComponentByName('tbsResponseData', tbs)
basic.setComponentByName('signatureAlgorithm', sa)
if not signature:
signature = crypto.sign(signer[2], encoder.encode(tbs), sigAlg)
basic.setComponentByName('signature',
univ.BitString("'%s'H" % (signature.encode('hex'))))
if certs:
cs = basic.setComponentByName('certs').getComponentByName('certs')
for i in range(len(certs)):
cs.setComponentByPosition(i, certs[i][0])
rbytes = ocsp.componentType.getTypeByPosition(1)
rbytes.setComponentByName('responseType',
univ.ObjectIdentifier(response_type))
rbytes.setComponentByName('response', encoder.encode(basic))
ocsp.setComponentByName('responseBytes', rbytes)
return ocsp
def MakePemBlock(der, name):
b64 = base64.b64encode(der)
wrapped = '\n'.join(b64[pos:pos + 64] for pos in xrange(0, len(b64), 64))
return '-----BEGIN %s-----\n%s\n-----END %s-----' % (name, wrapped, name)
def WriteStringToFile(data, path):
with open(path, "w") as f:
f.write(data)
def ReadFileToString(path):
with open(path, 'r') as f:
return f.read()
def CreateOCSPRequestDer(issuer_cert_pem, cert_pem):
'''Uses OpenSSL to generate a basic OCSPRequest for |cert_pem|.'''
issuer_path = "tmp_issuer.pem"
cert_path = "tmp_cert.pem"
request_path = "tmp_request.der"
WriteStringToFile(issuer_cert_pem, issuer_path)
WriteStringToFile(cert_pem, cert_path)
p = subprocess.Popen(["openssl", "ocsp", "-no_nonce", "-issuer", issuer_path,
"-cert", cert_path, "-reqout", request_path],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
stdout_data, stderr_data = p.communicate()
os.remove(issuer_path)
os.remove(cert_path)
result = None
if p.returncode == 0:
result = ReadFileToString(request_path)
os.remove(request_path)
return result
def Store(fname, description, ca, data):
ca_cert_pem = crypto.dump_certificate(crypto.FILETYPE_PEM, ca[1])
cert_pem = crypto.dump_certificate(crypto.FILETYPE_PEM, CERT[1])
ocsp_request_der = CreateOCSPRequestDer(ca_cert_pem, cert_pem)
out = ('%s\n%s\n%s\n\n%s\n%s') % (
description,
MakePemBlock(encoder.encode(data), "OCSP RESPONSE"),
ca_cert_pem.replace('CERTIFICATE', 'CA CERTIFICATE'),
cert_pem,
MakePemBlock(ocsp_request_der, "OCSP REQUEST"))
open('%s.pem' % fname, 'w').write(out)
Store(
'no_response',
'No SingleResponses attached to the response',
CA,
Create(responses=[]))
Store(
'malformed_request',
'Has a status of MALFORMED_REQUEST',
CA,
Create(response_status=1))
Store(
'bad_status',
'Has an invalid status larger than the defined Status enumeration',
CA,
Create(response_status=17))
Store(
'bad_ocsp_type',
'Has an invalid OCSP OID',
CA,
Create(response_type='1.3.6.1.5.5.7.48.1.2'))
Store(
'bad_signature',
'Has an invalid signature',
CA,
Create(signature='\xde\xad\xbe\xef'))
Store('ocsp_sign_direct', 'Signed directly by the issuer', CA,
Create(signer=CA, certs=[]))
Store('ocsp_sign_indirect', 'Signed indirectly through an intermediate', CA,
Create(signer=CA_LINK, certs=[CA_LINK]))
Store('ocsp_sign_indirect_missing',
'Signed indirectly through a missing intermediate', CA,
Create(signer=CA_LINK, certs=[]))
Store('ocsp_sign_bad_indirect',
'Signed through an intermediate without the correct key usage', CA,
Create(signer=CA_BADLINK, certs=[CA_BADLINK]))
Store('ocsp_extra_certs', 'Includes extra certs', CA,
Create(signer=CA, certs=[CA, CA_LINK]))
Store('has_version', 'Includes a default version V1', CA, Create(version=1))
Store(
'responder_name',
'Uses byName to identify the signer',
CA,
Create(responder=GetName(CA)))
# TODO(eroman): pyasn1 module has a bug in rfc2560.ResponderID() that will use
# IMPLICIT rather than EXPLICIT tagging for byKey
# (https://github.com/etingof/pyasn1-modules/issues/8). If using an affected
# version of the library you will need to patch pyasn1_modules/rfc2560.py and
# replace "implicitTag" with "explicitTag" in ResponderID to generate this
# test data correctly.
Store(
'responder_id',
'Uses byKey to identify the signer',
CA,
Create(responder=GetKeyHash(CA)))
Store(
'has_extension',
'Includes an x509v3 extension',
CA,
Create(extensions=[EXTENSION]))
Store(
'good_response',
'Is a valid response for the cert',
CA,
Create(responses=[CreateSingleResponse(CERT, 0)]))
Store('good_response_sha256',
'Is a valid response for the cert with a SHA256 signature', CA,
Create(responses=[CreateSingleResponse(CERT, 0)], sigAlg='sha256'))
Store(
'good_response_next_update',
'Is a valid response for the cert until nextUpdate',
CA,
Create(responses=[CreateSingleResponse(CERT, 0, next=NEXT_DATE)]))
Store(
'revoke_response',
'Is a REVOKE response for the cert',
CA,
Create(responses=[CreateSingleResponse(CERT, 1)]))
Store(
'revoke_response_reason',
'Is a REVOKE response for the cert with a reason',
CA,
Create(responses=[
CreateSingleResponse(CERT, 1, revoke_time=REVOKE_DATE, reason=1)
]))
Store(
'unknown_response',
'Is an UNKNOWN response for the cert',
CA,
Create(responses=[CreateSingleResponse(CERT, 2)]))
Store(
'multiple_response',
'Has multiple responses for the cert',
CA,
Create(responses=[
CreateSingleResponse(CERT, 0),
CreateSingleResponse(CERT, 2)
]))
Store(
'other_response',
'Is a response for a different cert',
CA,
Create(responses=[
CreateSingleResponse(JUNK_CERT, 0),
CreateSingleResponse(JUNK_CERT, 1)
]))
Store(
'has_single_extension',
'Has an extension in the SingleResponse',
CA,
Create(responses=[
CreateSingleResponse(CERT, 0, extensions=[CreateExtension()])
]))
Store(
'has_critical_single_extension',
'Has a critical extension in the SingleResponse', CA,
Create(responses=[
CreateSingleResponse(
CERT, 0, extensions=[CreateExtension('1.2.3.4', critical=True)])
]))
Store(
'has_critical_response_extension',
'Has a critical extension in the ResponseData', CA,
Create(
responses=[CreateSingleResponse(CERT, 0)],
extensions=[CreateExtension('1.2.3.4', critical=True)]))
Store(
'has_critical_ct_extension',
'Has a critical CT extension in the SingleResponse', CA,
Create(responses=[
CreateSingleResponse(
CERT,
0,
extensions=[
CreateExtension('1.3.6.1.4.1.11129.2.4.5', critical=True)
])
]))
Store('missing_response', 'Missing a response for the cert', CA,
Create(response_status=0, responses=[]))