net/data/ssl/scripts/crlsetutil.py

   1 #!/usr/bin/env python
   2 # Copyright (c) 2014 The Chromium Authors. All rights reserved.
   3 # Use of this source code is governed by a BSD-style license that can be
   4 # found in the LICENSE file.
   5
   6 """
   7 This utility takes a JSON input that describes a CRLSet and produces a
   8 CRLSet from it.
   9
  10 The input is taken on stdin and is a dict with the following keys:
  11   - BlockedBySPKI: An array of strings, where each string is a filename
  12       containing a PEM certificate, from which an SPKI will be extracted.
  13   - BlockedByHash: A dict of string to an array of ints, where the string is
  14       a filename containing a PEM format certificate, and the ints are the
  15       serial numbers. The listed serial numbers will be blocked when issued by
  16       the given certificate.
  17
  18 For example:
  19
  20 {
  21   "BlockedBySPKI": ["/tmp/blocked-certificate"],
  22   "BlockedByHash": {
  23     "/tmp/intermediate-certificate": [1, 2, 3]
  24   }
  25 }
  26 """
  27
  28 import hashlib
  29 import json
  30 import optparse
  31 import struct
  32 import sys
  33
  34
  35 def _pem_cert_to_binary(pem_filename):
  36   """Decodes the first PEM-encoded certificate in a given file into binary
  37
  38   Args:
  39     pem_filename: A filename that contains a PEM-encoded certificate. It may
  40         contain additional data (keys, textual representation) which will be
  41         ignored
  42
  43   Returns:
  44     A byte array containing the decoded certificate data
  45   """
  46   base64 = ""
  47   started = False
  48
  49   with open(pem_filename, 'r') as pem_file:
  50     for line in pem_file:
  51       if not started:
  52         if line.startswith('-----BEGIN CERTIFICATE'):
  53           started = True
  54       else:
  55         if line.startswith('-----END CERTIFICATE'):
  56           break
  57         base64 += line[:-1].strip()
  58
  59   return base64.decode('base64')
  60
  61
  62 def _parse_asn1_element(der_bytes):
  63   """Parses a DER-encoded tag/Length/Value into its component parts
  64
  65   Args:
  66     der_bytes: A DER-encoded ASN.1 data type
  67
  68   Returns:
  69     A tuple of the ASN.1 tag value, the length of the ASN.1 header that was
  70     read, the sequence of bytes for the value, and then any data from der_bytes
  71     that was not part of the tag/Length/Value.
  72   """
  73   tag = ord(der_bytes[0])
  74   length = ord(der_bytes[1])
  75   header_length = 2
  76
  77   if length & 0x80:
  78     num_length_bytes = length & 0x7f
  79     length = 0
  80     for i in xrange(2, 2 + num_length_bytes):
  81       length <<= 8
  82       length += ord(der_bytes[i])
  83     header_length = 2 + num_length_bytes
  84
  85   contents = der_bytes[:header_length + length]
  86   rest = der_bytes[header_length + length:]
  87
  88   return (tag, header_length, contents, rest)
  89
  90
  91 class ASN1Iterator(object):
  92   """Iterator that parses and iterates through a ASN.1 DER structure"""
  93
  94   def __init__(self, contents):
  95     self._tag = 0
  96     self._header_length = 0
  97     self._rest = None
  98     self._contents = contents
  99     self.step_into()
 100
 101   def step_into(self):
 102     """Begins processing the inner contents of the next ASN.1 element"""
 103     (self._tag, self._header_length, self._contents, self._rest) = (
 104         _parse_asn1_element(self._contents[self._header_length:]))
 105
 106   def step_over(self):
 107     """Skips/ignores the next ASN.1 element"""
 108     (self._tag, self._header_length, self._contents, self._rest) = (
 109         _parse_asn1_element(self._rest))
 110
 111   def tag(self):
 112     """Returns the ASN.1 tag of the current element"""
 113     return self._tag
 114
 115   def contents(self):
 116     """Returns the raw data of the current element"""
 117     return self._contents
 118
 119
 120 def _der_cert_to_spki(der_bytes):
 121   """Returns the subjectPublicKeyInfo of a DER-encoded certificate
 122
 123   Args:
 124     der_bytes: A DER-encoded certificate (RFC 5280)
 125
 126   Returns:
 127     A byte array containing the subjectPublicKeyInfo
 128   """
 129   iterator = ASN1Iterator(der_bytes)
 130   iterator.step_into()  # enter certificate structure
 131   iterator.step_into()  # enter TBSCertificate
 132   iterator.step_over()  # over version
 133   iterator.step_over()  # over serial
 134   iterator.step_over()  # over signature algorithm
 135   iterator.step_over()  # over issuer name
 136   iterator.step_over()  # over validity
 137   iterator.step_over()  # over subject name
 138   return iterator.contents()
 139
 140
 141 def pem_cert_file_to_spki_hash(pem_filename):
 142   """Gets the SHA-256 hash of the subjectPublicKeyInfo of a cert in a file
 143
 144   Args:
 145     pem_filename: A file containing a PEM-encoded certificate.
 146
 147   Returns:
 148     The SHA-256 hash of the first certificate in the file, as a byte sequence
 149   """
 150   return hashlib.sha256(
 151     _der_cert_to_spki(_pem_cert_to_binary(pem_filename))).digest()
 152
 153
 154 def main():
 155   parser = optparse.OptionParser(description=sys.modules[__name__].__doc__)
 156   parser.add_option('-o', '--output',
 157                     help='Specifies the output file. The default is stdout.')
 158   options, _ = parser.parse_args()
 159   outfile = sys.stdout
 160   if options.output and options.output != '-':
 161     outfile = open(options.output, 'wb')
 162
 163   config = json.load(sys.stdin)
 164   blocked_spkis = [
 165       pem_cert_file_to_spki_hash(pem_file).encode('base64').strip()
 166       for pem_file in config.get('BlockedBySPKI', [])]
 167   parents = {
 168     pem_cert_file_to_spki_hash(pem_file): serials
 169     for pem_file, serials in config.get('BlockedByHash', {}).iteritems()
 170   }
 171   header_json = {
 172       'Version': 0,
 173       'ContentType': 'CRLSet',
 174       'Sequence': 0,
 175       'DeltaFrom': 0,
 176       'NumParents': len(parents),
 177       'BlockedSPKIs': blocked_spkis,
 178   }
 179   header = json.dumps(header_json)
 180   outfile.write(struct.pack('<H', len(header)))
 181   outfile.write(header)
 182   for spki, serials in sorted(parents.iteritems()):
 183     outfile.write(spki)
 184     outfile.write(struct.pack('<I', len(serials)))
 185     for serial in serials:
 186       raw_serial = []
 187       if not serial:
 188         raw_serial = ['\x00']
 189       else:
 190         while serial:
 191           raw_serial.insert(0, chr(serial & 0xff))
 192           serial >>= 8
 193
 194     outfile.write(struct.pack('<B', len(raw_serial)))
 195     outfile.write(''.join(raw_serial))
 196   return 0
 197
 198
 199 if __name__ == '__main__':
 200   sys.exit(main())