#!/usr/bin/env python3
# Copyright 2014 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""
This utility takes a JSON input that describes a CRLSet and produces a
CRLSet from it.
The input is taken on stdin and is a dict with the following keys:
- BlockedBySPKI: An array of strings, where each string is a filename
containing a PEM certificate, from which an SPKI will be extracted.
- BlockedByHash: A dict of string to an array of strings. The dict key is
a filename containing a PEM certificate, representing the issuer cert,
while the array of strings contain the filenames of PEM format
certificates whose serials are blocked.
- LimitedSubjects: A dict of string to an array of strings, where the key is
a filename containing a PEM format certificate, and the strings are the
filenames of PEM format certificates. Certificates that share a Subject
with the key will be restricted to the set of SPKIs extracted from the
files in the values.
- Sequence: An optional integer sequence number to use for the CRLSet. If
not present, defaults to 1.
For example:
{
"BlockedBySPKI": ["/tmp/blocked-certificate"],
"BlockedByHash": {
"/tmp/intermediate-certificate": [1, 2, 3]
},
"LimitedSubjects": {
"/tmp/limited-certificate": [
"/tmp/limited-certificate",
"/tmp/limited-certificate2"
]
},
"Sequence": 23
}
"""
import base64
import collections
import hashlib
import json
import optparse
import six
import struct
import sys
def _pem_cert_to_binary(pem_filename):
"""Decodes the first PEM-encoded certificate in a given file into binary
Args:
pem_filename: A filename that contains a PEM-encoded certificate. It may
contain additional data (keys, textual representation) which will be
ignored
Returns:
A byte array containing the decoded certificate data
"""
pem_data = ""
started = False
with open(pem_filename, 'r') as pem_file:
for line in pem_file:
if not started:
if line.startswith('-----BEGIN CERTIFICATE'):
started = True
else:
if line.startswith('-----END CERTIFICATE'):
break
pem_data += line[:-1].strip()
return base64.b64decode(pem_data)
def _parse_asn1_element(der_bytes):
"""Parses a DER-encoded tag/Length/Value into its component parts
Args:
der_bytes: A DER-encoded ASN.1 data type
Returns:
A tuple of the ASN.1 tag value, the length of the ASN.1 header that was
read, the sequence of bytes for the value, and then any data from der_bytes
that was not part of the tag/Length/Value.
"""
tag = six.indexbytes(der_bytes, 0)
length = six.indexbytes(der_bytes, 1)
header_length = 2
if length & 0x80:
num_length_bytes = length & 0x7f
length = 0
for i in range(2, 2 + num_length_bytes):
length <<= 8
length += six.indexbytes(der_bytes, i)
header_length = 2 + num_length_bytes
contents = der_bytes[:header_length + length]
rest = der_bytes[header_length + length:]
return (tag, header_length, contents, rest)
class ASN1Iterator(object):
"""Iterator that parses and iterates through a ASN.1 DER structure"""
def __init__(self, contents):
self._tag = 0
self._header_length = 0
self._rest = None
self._contents = contents
self.step_into()
def step_into(self):
"""Begins processing the inner contents of the next ASN.1 element"""
(self._tag, self._header_length, self._contents, self._rest) = (
_parse_asn1_element(self._contents[self._header_length:]))
def step_over(self):
"""Skips/ignores the next ASN.1 element"""
(self._tag, self._header_length, self._contents, self._rest) = (
_parse_asn1_element(self._rest))
def tag(self):
"""Returns the ASN.1 tag of the current element"""
return self._tag
def contents(self):
"""Returns the raw data of the current element"""
return self._contents
def encoded_value(self):
"""Returns the encoded value of the current element (i.e. without header)"""
return self._contents[self._header_length:]
def _der_cert_to_spki(der_bytes):
"""Returns the subjectPublicKeyInfo of a DER-encoded certificate
Args:
der_bytes: A DER-encoded certificate (RFC 5280)
Returns:
A byte array containing the subjectPublicKeyInfo
"""
iterator = ASN1Iterator(der_bytes)
iterator.step_into() # enter certificate structure
iterator.step_into() # enter TBSCertificate
iterator.step_over() # over version
iterator.step_over() # over serial
iterator.step_over() # over signature algorithm
iterator.step_over() # over issuer name
iterator.step_over() # over validity
iterator.step_over() # over subject name
return iterator.contents()
def der_cert_to_spki_hash(der_cert):
"""Gets the SHA-256 hash of the subjectPublicKeyInfo of a DER encoded cert
Args:
der_cert: A string containing the DER-encoded certificate
Returns:
The SHA-256 hash of the certificate, as a byte sequence
"""
return hashlib.sha256(_der_cert_to_spki(der_cert)).digest()
def pem_cert_file_to_spki_hash(pem_filename):
"""Gets the SHA-256 hash of the subjectPublicKeyInfo of a cert in a file
Args:
pem_filename: A file containing a PEM-encoded certificate.
Returns:
The SHA-256 hash of the first certificate in the file, as a byte sequence
"""
return der_cert_to_spki_hash(_pem_cert_to_binary(pem_filename))
def der_cert_to_subject_hash(der_bytes):
"""Returns SHA256(subject) of a DER-encoded certificate
Args:
der_bytes: A DER-encoded certificate (RFC 5280)
Returns:
The SHA-256 hash of the certificate's subject.
"""
iterator = ASN1Iterator(der_bytes)
iterator.step_into() # enter certificate structure
iterator.step_into() # enter TBSCertificate
iterator.step_over() # over version
iterator.step_over() # over serial
iterator.step_over() # over signature algorithm
iterator.step_over() # over issuer name
iterator.step_over() # over validity
return hashlib.sha256(iterator.contents()).digest()
def pem_cert_file_to_subject_hash(pem_filename):
"""Gets the SHA-256 hash of the subject of a cert in a file
Args:
pem_filename: A file containing a PEM-encoded certificate.
Returns:
The SHA-256 hash of the subject of the first certificate in the file, as a
byte sequence
"""
return der_cert_to_subject_hash(_pem_cert_to_binary(pem_filename))
def der_cert_to_serial(der_bytes):
"""Gets the serial of a DER-encoded certificate, omitting leading 0x00
Args:
der_bytes: A DER-encoded certificates (RFC 5280)
Returns:
The encoded serial number value (omitting tag and length), and omitting
any leading 0x00 used to indicate it is a positive INTEGER.
"""
iterator = ASN1Iterator(der_bytes)
iterator.step_into() # enter certificate structure
iterator.step_into() # enter TBSCertificate
iterator.step_over() # over version
raw_serial = iterator.encoded_value()
if six.indexbytes(raw_serial, 0) == 0x00 and len(raw_serial) > 1:
raw_serial = raw_serial[1:]
return raw_serial
def pem_cert_file_to_serial(pem_filename):
"""Gets the DER-encoded serial of a cert in a file, omitting leading 0x00
Args:
pem_filename: A file containing a PEM-encoded certificate.
Returns:
The DER-encoded serial as a byte sequence
"""
return der_cert_to_serial(_pem_cert_to_binary(pem_filename))
def main():
parser = optparse.OptionParser(description=sys.modules[__name__].__doc__)
parser.add_option('-o', '--output',
help='Specifies the output file. The default is stdout.')
options, _ = parser.parse_args()
outfile = sys.stdout
if options.output and options.output != '-':
outfile = open(options.output, 'wb')
config = json.load(sys.stdin)
blocked_spkis = [
base64.b64encode(pem_cert_file_to_spki_hash(pem_file)).decode('ascii')
for pem_file in config.get('BlockedBySPKI', [])
]
parents = {
pem_cert_file_to_spki_hash(pem_file): [
pem_cert_file_to_serial(issued_cert_file)
for issued_cert_file in issued_certs
]
for pem_file, issued_certs in config.get('BlockedByHash', {}).items()
}
limited_subjects = {
base64.b64encode(pem_cert_file_to_subject_hash(pem_file)).decode('ascii'):
[
base64.b64encode(pem_cert_file_to_spki_hash(filename)).decode('ascii')
for filename in allowed_pems
]
for pem_file, allowed_pems in config.get('LimitedSubjects', {}).items()
}
known_interception_spkis = [
base64.b64encode(pem_cert_file_to_spki_hash(pem_file)).decode('ascii')
for pem_file in config.get('KnownInterceptionSPKIs', [])
]
blocked_interception_spkis = [
base64.b64encode(pem_cert_file_to_spki_hash(pem_file)).decode('ascii')
for pem_file in config.get('BlockedInterceptionSPKIs', [])
]
header_json = {
'Version': 0,
'ContentType': 'CRLSet',
'Sequence': int(config.get("Sequence", 1)),
'NumParents': len(parents),
'BlockedSPKIs': blocked_spkis,
'LimitedSubjects': limited_subjects,
'KnownInterceptionSPKIs': known_interception_spkis,
'BlockedInterceptionSPKIs': blocked_interception_spkis
}
header = json.dumps(header_json)
outfile.write(struct.pack('<H', len(header)))
outfile.write(header.encode('utf-8'))
for spki, serials in sorted(parents.items()):
outfile.write(spki)
outfile.write(struct.pack('<I', len(serials)))
for serial in serials:
raw_serial = []
if not serial:
raw_serial = b'\x00'
else:
raw_serial = serial
outfile.write(struct.pack('<B', len(raw_serial)))
outfile.write(raw_serial)
return 0
if __name__ == '__main__':
sys.exit(main())
Codebase Browser
chromium