123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427 |
- #!/usr/bin/env python3
- # Copyright 2017 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.
- """Pretty-prints certificates as an openssl-annotated PEM file."""
- import argparse
- import base64
- import errno
- import hashlib
- import os
- import re
- import subprocess
- import sys
- import traceback
- def read_file_to_string(path):
- with open(path, 'rb') as f:
- return f.read()
- def read_certificates_data_from_server(hostname):
- """Uses openssl to fetch the PEM-encoded certificates for an SSL server."""
- p = subprocess.Popen(["openssl", "s_client", "-showcerts",
- "-servername", hostname,
- "-connect", hostname + ":443"],
- stdin=subprocess.PIPE,
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE)
- result = p.communicate()
- if p.returncode == 0:
- return result[0]
- sys.stderr.write("Failed getting certificates for %s:\n%s\n" % (
- hostname, result[1]))
- return b""
- def read_sources_from_commandline(sources):
- """Processes the command lines and returns an array of all the sources
- bytes."""
- sources_bytes = []
- if not sources:
- # If no command-line arguments were given to the program, read input from
- # stdin.
- sources_bytes.append(sys.stdin.buffer.read())
- else:
- for arg in sources:
- # If the argument identifies a file path, read it
- if os.path.exists(arg):
- sources_bytes.append(read_file_to_string(arg))
- else:
- # Otherwise treat it as a web server address.
- sources_bytes.append(read_certificates_data_from_server(arg))
- return sources_bytes
- def strip_indentation_whitespace(text):
- """Strips leading whitespace from each line."""
- stripped_lines = [line.lstrip() for line in text.split(b"\n")]
- return b"\n".join(stripped_lines)
- def strip_all_whitespace(text):
- pattern = re.compile(rb'\s+')
- return re.sub(pattern, b'', text).replace(rb'\n', b'\n')
- def extract_certificates_from_pem(pem_bytes):
- certificates_der = []
- regex = re.compile(
- rb'-----BEGIN (CERTIFICATE|PKCS7)-----(.*?)(-----END \1-----|$)',
- re.DOTALL)
- for match in regex.finditer(pem_bytes):
- if not match.group(3):
- sys.stderr.write(
- "\nUnterminated %s block, input is corrupt or truncated\n" %
- match.group(1))
- continue
- der = base64.b64decode(strip_all_whitespace(match.group(2)))
- if match.group(1) == b'CERTIFICATE':
- certificates_der.append(der)
- else:
- certificates_der.extend(extract_certificates_from_der_pkcs7(der))
- return certificates_der
- def extract_certificates_from_der_pkcs7(der_bytes):
- pkcs7_certs_pem = process_data_with_command(
- ['openssl','pkcs7','-print_certs', '-inform', 'DER'], der_bytes)
- # The output will be one or more PEM encoded certificates.
- # (Or CRLS, but those will be ignored.)
- if pkcs7_certs_pem:
- return extract_certificates_from_pem(pkcs7_certs_pem)
- return []
- def extract_certificates_from_der_ascii(input_text):
- certificates_der = []
- # Look for beginning and end of Certificate SEQUENCE. The indentation is
- # significant. (The SEQUENCE must be non-indented, and the rest of the DER
- # ASCII must be indented until the closing } which again is non-indented.)
- # The output of der2ascii meets this, but it is not a requirement of the DER
- # ASCII language.
- # TODO(mattm): consider alternate approach of doing ascii2der on entire
- # input, and handling the multiple concatenated DER certificates.
- regex = re.compile(r'^(SEQUENCE {.*?^})', re.DOTALL | re.MULTILINE)
- for match in regex.finditer(input_text):
- der_ascii_bytes = match.group(1)
- der_bytes = process_data_with_command(["ascii2der"], der_ascii_bytes)
- if der_bytes:
- certificates_der.append(der_bytes)
- return certificates_der
- def decode_netlog_hexdump(netlog_text):
- lines = netlog_text.splitlines()
- # Skip the text preceeding the actual hexdump.
- while lines and 'bytes =' not in lines[0]:
- del lines[0]
- if not lines:
- return None
- del lines[0]
- bytes = []
- hex_re = re.compile('\s*([0-9A-Fa-f ]{48})')
- for line in lines:
- m = hex_re.search(line)
- if not m:
- break
- hex_string = m.group(1)
- bytes.extend(chr(int(part, 16)) for part in hex_string.split())
- return ''.join(bytes)
- class ByteReader:
- """Iteratively consume data from a byte string.
- Automatically tracks and advances current position in the string as data is
- consumed, and will throw an exception if attempting to read past the end of
- the string.
- """
- def __init__(self, data):
- self.data = data
- self.pos = 0
- def consume_byte(self):
- i = ord(self.data[self.pos])
- self.pos += 1
- return i
- def consume_int16(self):
- return ((self.consume_byte() << 8) + self.consume_byte())
- def consume_int24(self):
- return ((self.consume_byte() << 16) + (self.consume_byte() << 8) +
- self.consume_byte())
- def consume_bytes(self, n):
- b = self.data[self.pos:self.pos+n]
- if len(b) != n:
- raise IndexError('requested:%d bytes actual:%d bytes'%(n, len(b)))
- self.pos += n
- return b
- def remaining_byte_count(self):
- return len(self.data) - self.pos
- def decode_tls10_certificate_message(reader):
- message_length = reader.consume_int24()
- if reader.remaining_byte_count() != message_length:
- raise RuntimeError(
- 'message_length(%d) != remaining_byte_count(%d)\n' % (
- message_length, reader.remaining_byte_count()))
- certificate_list_length = reader.consume_int24()
- if reader.remaining_byte_count() != certificate_list_length:
- raise RuntimeError(
- 'certificate_list_length(%d) != remaining_byte_count(%d)\n' % (
- certificate_list_length, reader.remaining_byte_count()))
- certificates_der = []
- while reader.remaining_byte_count():
- cert_len = reader.consume_int24()
- certificates_der.append(reader.consume_bytes(cert_len))
- return certificates_der
- def decode_tls13_certificate_message(reader):
- message_length = reader.consume_int24()
- if reader.remaining_byte_count() != message_length:
- raise RuntimeError(
- 'message_length(%d) != remaining_byte_count(%d)\n' % (
- message_length, reader.remaining_byte_count()))
- # Ignore certificate_request_context.
- certificate_request_context_length = reader.consume_byte()
- reader.consume_bytes(certificate_request_context_length)
- certificate_list_length = reader.consume_int24()
- if reader.remaining_byte_count() != certificate_list_length:
- raise RuntimeError(
- 'certificate_list_length(%d) != remaining_byte_count(%d)\n' % (
- certificate_list_length, reader.remaining_byte_count()))
- certificates_der = []
- while reader.remaining_byte_count():
- # Assume certificate_type is X.509.
- cert_len = reader.consume_int24()
- certificates_der.append(reader.consume_bytes(cert_len))
- # Ignore extensions.
- extension_len = reader.consume_int16()
- reader.consume_bytes(extension_len)
- return certificates_der
- def decode_tls_certificate_message(certificate_message):
- reader = ByteReader(certificate_message)
- if reader.consume_byte() != 11:
- sys.stderr.write('HandshakeType != 11. Not a Certificate Message.\n')
- return []
- # The TLS certificate message encoding changed in TLS 1.3. Rather than
- # require pasting in and parsing the whole handshake to discover the TLS
- # version, just try parsing the message with both the old and new encodings.
- # First try the old style certificate message:
- try:
- return decode_tls10_certificate_message(reader)
- except (IndexError, RuntimeError):
- tls10_traceback = traceback.format_exc()
- # Restart the ByteReader and consume the HandshakeType byte again.
- reader = ByteReader(certificate_message)
- reader.consume_byte()
- # Try the new style certificate message:
- try:
- return decode_tls13_certificate_message(reader)
- except (IndexError, RuntimeError):
- tls13_traceback = traceback.format_exc()
- # Neither attempt succeeded, just dump some error info:
- sys.stderr.write("Couldn't parse TLS certificate message\n")
- sys.stderr.write("TLS1.0 parse attempt:\n%s\n" % tls10_traceback)
- sys.stderr.write("TLS1.3 parse attempt:\n%s\n" % tls13_traceback)
- sys.stderr.write("\n")
- return []
- def extract_tls_certificate_message(netlog_text):
- raw_certificate_message = decode_netlog_hexdump(netlog_text)
- if not raw_certificate_message:
- return []
- return decode_tls_certificate_message(raw_certificate_message)
- def extract_certificates(source_bytes):
- if b"BEGIN CERTIFICATE" in source_bytes or b"BEGIN PKCS7" in source_bytes:
- return extract_certificates_from_pem(source_bytes)
- if b"SEQUENCE {" in source_bytes:
- return extract_certificates_from_der_ascii(source_bytes)
- if b"SSL_HANDSHAKE_MESSAGE_RECEIVED" in source_bytes:
- return extract_tls_certificate_message(source_bytes)
- # DER encoding of PKCS #7 signedData OID (1.2.840.113549.1.7.2)
- if b"\x06\x09\x2a\x86\x48\x86\xf7\x0d\x01\x07\x02" in source_bytes:
- return extract_certificates_from_der_pkcs7(source_bytes)
- # Otherwise assume it is the DER for a single certificate
- return [source_bytes]
- def process_data_with_command(command, data):
- try:
- p = subprocess.Popen(command,
- stdin=subprocess.PIPE,
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE)
- except OSError as e:
- if e.errno == errno.ENOENT:
- sys.stderr.write("Failed to execute %s\n" % command[0])
- return b""
- raise
- result = p.communicate(data)
- if p.returncode == 0:
- return result[0]
- # Otherwise failed.
- sys.stderr.write("Failed: %s: %s\n" % (" ".join(command), result[1]))
- return b""
- def openssl_text_pretty_printer(certificate_der, unused_certificate_number):
- return process_data_with_command(["openssl", "x509", "-text", "-inform",
- "DER", "-noout"], certificate_der)
- def pem_pretty_printer(certificate_der, unused_certificate_number):
- return process_data_with_command(["openssl", "x509", "-inform", "DER",
- "-outform", "PEM"], certificate_der)
- def der2ascii_pretty_printer(certificate_der, unused_certificate_number):
- return process_data_with_command(["der2ascii"], certificate_der)
- def header_pretty_printer(certificate_der, certificate_number):
- cert_hash = hashlib.sha256(certificate_der).hexdigest()
- s = """===========================================
- Certificate%d: %s
- ===========================================""" % (certificate_number, cert_hash)
- return s.encode("ascii")
- # This is actually just used as a magic value, since pretty_print_certificates
- # special-cases der output.
- def der_printer():
- raise RuntimeError
- def pretty_print_certificates(certificates_der, pretty_printers):
- # Need to special-case DER output to avoid adding any newlines, and to
- # only allow a single certificate to be output.
- if pretty_printers == [der_printer]:
- if len(certificates_der) > 1:
- sys.stderr.write("DER output only supports a single certificate, "
- "ignoring %d remaining certs\n" % (
- len(certificates_der) - 1))
- return certificates_der[0]
- result = b""
- for i in range(len(certificates_der)):
- certificate_der = certificates_der[i]
- pretty = []
- for pretty_printer in pretty_printers:
- pretty_printed = pretty_printer(certificate_der, i)
- if pretty_printed:
- pretty.append(pretty_printed)
- result += b"\n".join(pretty) + b"\n"
- return result
- def parse_outputs(outputs):
- pretty_printers = []
- output_map = {"der2ascii": der2ascii_pretty_printer,
- "openssl_text": openssl_text_pretty_printer,
- "pem": pem_pretty_printer,
- "header": header_pretty_printer,
- "der": der_printer}
- for output_name in outputs.split(','):
- if output_name not in output_map:
- sys.stderr.write("Invalid output type: %s\n" % output_name)
- return []
- pretty_printers.append(output_map[output_name])
- if der_printer in pretty_printers and len(pretty_printers) > 1:
- sys.stderr.write("Output type der must be used alone.\n")
- return []
- return pretty_printers
- def main():
- parser = argparse.ArgumentParser(
- description=__doc__, formatter_class=argparse.RawTextHelpFormatter)
- parser.add_argument('sources', metavar='SOURCE', nargs='*',
- help='''Each SOURCE can be one of:
- (1) A server name such as www.google.com.
- (2) A PEM [*] file containing one or more CERTIFICATE or PKCS7 blocks
- (3) A file containing one or more DER ASCII certificates
- (4) A text NetLog dump of a TLS certificate message
- (must include the SSL_HANDSHAKE_MESSAGE_RECEIVED line)
- (5) A binary file containing DER-encoded PKCS #7 signedData
- (6) A binary file containing DER-encoded certificate
- When multiple SOURCEs are listed, all certificates in them
- are concatenated. If no SOURCE is given then data will be
- read from stdin.
- [*] Parsing of PEM files is relaxed - leading indentation
- whitespace will be stripped (needed for copy-pasting data
- from NetLogs).''')
- parser.add_argument(
- '--output', dest='outputs', action='store',
- default="header,der2ascii,openssl_text,pem",
- help='output formats to use. Default: %(default)s')
- args = parser.parse_args()
- sources_bytes = read_sources_from_commandline(args.sources)
- pretty_printers = parse_outputs(args.outputs)
- if not pretty_printers:
- sys.stderr.write('No pretty printers selected.\n')
- sys.exit(1)
- certificates_der = []
- for source_bytes in sources_bytes:
- certificates_der.extend(extract_certificates(source_bytes))
- sys.stdout.buffer.write(
- pretty_print_certificates(certificates_der, pretty_printers))
- if __name__ == "__main__":
- main()
|