print_certificates.py 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427
  1. #!/usr/bin/env python3
  2. # Copyright 2017 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. """Pretty-prints certificates as an openssl-annotated PEM file."""
  6. import argparse
  7. import base64
  8. import errno
  9. import hashlib
  10. import os
  11. import re
  12. import subprocess
  13. import sys
  14. import traceback
  15. def read_file_to_string(path):
  16. with open(path, 'rb') as f:
  17. return f.read()
  18. def read_certificates_data_from_server(hostname):
  19. """Uses openssl to fetch the PEM-encoded certificates for an SSL server."""
  20. p = subprocess.Popen(["openssl", "s_client", "-showcerts",
  21. "-servername", hostname,
  22. "-connect", hostname + ":443"],
  23. stdin=subprocess.PIPE,
  24. stdout=subprocess.PIPE,
  25. stderr=subprocess.PIPE)
  26. result = p.communicate()
  27. if p.returncode == 0:
  28. return result[0]
  29. sys.stderr.write("Failed getting certificates for %s:\n%s\n" % (
  30. hostname, result[1]))
  31. return b""
  32. def read_sources_from_commandline(sources):
  33. """Processes the command lines and returns an array of all the sources
  34. bytes."""
  35. sources_bytes = []
  36. if not sources:
  37. # If no command-line arguments were given to the program, read input from
  38. # stdin.
  39. sources_bytes.append(sys.stdin.buffer.read())
  40. else:
  41. for arg in sources:
  42. # If the argument identifies a file path, read it
  43. if os.path.exists(arg):
  44. sources_bytes.append(read_file_to_string(arg))
  45. else:
  46. # Otherwise treat it as a web server address.
  47. sources_bytes.append(read_certificates_data_from_server(arg))
  48. return sources_bytes
  49. def strip_indentation_whitespace(text):
  50. """Strips leading whitespace from each line."""
  51. stripped_lines = [line.lstrip() for line in text.split(b"\n")]
  52. return b"\n".join(stripped_lines)
  53. def strip_all_whitespace(text):
  54. pattern = re.compile(rb'\s+')
  55. return re.sub(pattern, b'', text).replace(rb'\n', b'\n')
  56. def extract_certificates_from_pem(pem_bytes):
  57. certificates_der = []
  58. regex = re.compile(
  59. rb'-----BEGIN (CERTIFICATE|PKCS7)-----(.*?)(-----END \1-----|$)',
  60. re.DOTALL)
  61. for match in regex.finditer(pem_bytes):
  62. if not match.group(3):
  63. sys.stderr.write(
  64. "\nUnterminated %s block, input is corrupt or truncated\n" %
  65. match.group(1))
  66. continue
  67. der = base64.b64decode(strip_all_whitespace(match.group(2)))
  68. if match.group(1) == b'CERTIFICATE':
  69. certificates_der.append(der)
  70. else:
  71. certificates_der.extend(extract_certificates_from_der_pkcs7(der))
  72. return certificates_der
  73. def extract_certificates_from_der_pkcs7(der_bytes):
  74. pkcs7_certs_pem = process_data_with_command(
  75. ['openssl','pkcs7','-print_certs', '-inform', 'DER'], der_bytes)
  76. # The output will be one or more PEM encoded certificates.
  77. # (Or CRLS, but those will be ignored.)
  78. if pkcs7_certs_pem:
  79. return extract_certificates_from_pem(pkcs7_certs_pem)
  80. return []
  81. def extract_certificates_from_der_ascii(input_text):
  82. certificates_der = []
  83. # Look for beginning and end of Certificate SEQUENCE. The indentation is
  84. # significant. (The SEQUENCE must be non-indented, and the rest of the DER
  85. # ASCII must be indented until the closing } which again is non-indented.)
  86. # The output of der2ascii meets this, but it is not a requirement of the DER
  87. # ASCII language.
  88. # TODO(mattm): consider alternate approach of doing ascii2der on entire
  89. # input, and handling the multiple concatenated DER certificates.
  90. regex = re.compile(r'^(SEQUENCE {.*?^})', re.DOTALL | re.MULTILINE)
  91. for match in regex.finditer(input_text):
  92. der_ascii_bytes = match.group(1)
  93. der_bytes = process_data_with_command(["ascii2der"], der_ascii_bytes)
  94. if der_bytes:
  95. certificates_der.append(der_bytes)
  96. return certificates_der
  97. def decode_netlog_hexdump(netlog_text):
  98. lines = netlog_text.splitlines()
  99. # Skip the text preceeding the actual hexdump.
  100. while lines and 'bytes =' not in lines[0]:
  101. del lines[0]
  102. if not lines:
  103. return None
  104. del lines[0]
  105. bytes = []
  106. hex_re = re.compile('\s*([0-9A-Fa-f ]{48})')
  107. for line in lines:
  108. m = hex_re.search(line)
  109. if not m:
  110. break
  111. hex_string = m.group(1)
  112. bytes.extend(chr(int(part, 16)) for part in hex_string.split())
  113. return ''.join(bytes)
  114. class ByteReader:
  115. """Iteratively consume data from a byte string.
  116. Automatically tracks and advances current position in the string as data is
  117. consumed, and will throw an exception if attempting to read past the end of
  118. the string.
  119. """
  120. def __init__(self, data):
  121. self.data = data
  122. self.pos = 0
  123. def consume_byte(self):
  124. i = ord(self.data[self.pos])
  125. self.pos += 1
  126. return i
  127. def consume_int16(self):
  128. return ((self.consume_byte() << 8) + self.consume_byte())
  129. def consume_int24(self):
  130. return ((self.consume_byte() << 16) + (self.consume_byte() << 8) +
  131. self.consume_byte())
  132. def consume_bytes(self, n):
  133. b = self.data[self.pos:self.pos+n]
  134. if len(b) != n:
  135. raise IndexError('requested:%d bytes actual:%d bytes'%(n, len(b)))
  136. self.pos += n
  137. return b
  138. def remaining_byte_count(self):
  139. return len(self.data) - self.pos
  140. def decode_tls10_certificate_message(reader):
  141. message_length = reader.consume_int24()
  142. if reader.remaining_byte_count() != message_length:
  143. raise RuntimeError(
  144. 'message_length(%d) != remaining_byte_count(%d)\n' % (
  145. message_length, reader.remaining_byte_count()))
  146. certificate_list_length = reader.consume_int24()
  147. if reader.remaining_byte_count() != certificate_list_length:
  148. raise RuntimeError(
  149. 'certificate_list_length(%d) != remaining_byte_count(%d)\n' % (
  150. certificate_list_length, reader.remaining_byte_count()))
  151. certificates_der = []
  152. while reader.remaining_byte_count():
  153. cert_len = reader.consume_int24()
  154. certificates_der.append(reader.consume_bytes(cert_len))
  155. return certificates_der
  156. def decode_tls13_certificate_message(reader):
  157. message_length = reader.consume_int24()
  158. if reader.remaining_byte_count() != message_length:
  159. raise RuntimeError(
  160. 'message_length(%d) != remaining_byte_count(%d)\n' % (
  161. message_length, reader.remaining_byte_count()))
  162. # Ignore certificate_request_context.
  163. certificate_request_context_length = reader.consume_byte()
  164. reader.consume_bytes(certificate_request_context_length)
  165. certificate_list_length = reader.consume_int24()
  166. if reader.remaining_byte_count() != certificate_list_length:
  167. raise RuntimeError(
  168. 'certificate_list_length(%d) != remaining_byte_count(%d)\n' % (
  169. certificate_list_length, reader.remaining_byte_count()))
  170. certificates_der = []
  171. while reader.remaining_byte_count():
  172. # Assume certificate_type is X.509.
  173. cert_len = reader.consume_int24()
  174. certificates_der.append(reader.consume_bytes(cert_len))
  175. # Ignore extensions.
  176. extension_len = reader.consume_int16()
  177. reader.consume_bytes(extension_len)
  178. return certificates_der
  179. def decode_tls_certificate_message(certificate_message):
  180. reader = ByteReader(certificate_message)
  181. if reader.consume_byte() != 11:
  182. sys.stderr.write('HandshakeType != 11. Not a Certificate Message.\n')
  183. return []
  184. # The TLS certificate message encoding changed in TLS 1.3. Rather than
  185. # require pasting in and parsing the whole handshake to discover the TLS
  186. # version, just try parsing the message with both the old and new encodings.
  187. # First try the old style certificate message:
  188. try:
  189. return decode_tls10_certificate_message(reader)
  190. except (IndexError, RuntimeError):
  191. tls10_traceback = traceback.format_exc()
  192. # Restart the ByteReader and consume the HandshakeType byte again.
  193. reader = ByteReader(certificate_message)
  194. reader.consume_byte()
  195. # Try the new style certificate message:
  196. try:
  197. return decode_tls13_certificate_message(reader)
  198. except (IndexError, RuntimeError):
  199. tls13_traceback = traceback.format_exc()
  200. # Neither attempt succeeded, just dump some error info:
  201. sys.stderr.write("Couldn't parse TLS certificate message\n")
  202. sys.stderr.write("TLS1.0 parse attempt:\n%s\n" % tls10_traceback)
  203. sys.stderr.write("TLS1.3 parse attempt:\n%s\n" % tls13_traceback)
  204. sys.stderr.write("\n")
  205. return []
  206. def extract_tls_certificate_message(netlog_text):
  207. raw_certificate_message = decode_netlog_hexdump(netlog_text)
  208. if not raw_certificate_message:
  209. return []
  210. return decode_tls_certificate_message(raw_certificate_message)
  211. def extract_certificates(source_bytes):
  212. if b"BEGIN CERTIFICATE" in source_bytes or b"BEGIN PKCS7" in source_bytes:
  213. return extract_certificates_from_pem(source_bytes)
  214. if b"SEQUENCE {" in source_bytes:
  215. return extract_certificates_from_der_ascii(source_bytes)
  216. if b"SSL_HANDSHAKE_MESSAGE_RECEIVED" in source_bytes:
  217. return extract_tls_certificate_message(source_bytes)
  218. # DER encoding of PKCS #7 signedData OID (1.2.840.113549.1.7.2)
  219. if b"\x06\x09\x2a\x86\x48\x86\xf7\x0d\x01\x07\x02" in source_bytes:
  220. return extract_certificates_from_der_pkcs7(source_bytes)
  221. # Otherwise assume it is the DER for a single certificate
  222. return [source_bytes]
  223. def process_data_with_command(command, data):
  224. try:
  225. p = subprocess.Popen(command,
  226. stdin=subprocess.PIPE,
  227. stdout=subprocess.PIPE,
  228. stderr=subprocess.PIPE)
  229. except OSError as e:
  230. if e.errno == errno.ENOENT:
  231. sys.stderr.write("Failed to execute %s\n" % command[0])
  232. return b""
  233. raise
  234. result = p.communicate(data)
  235. if p.returncode == 0:
  236. return result[0]
  237. # Otherwise failed.
  238. sys.stderr.write("Failed: %s: %s\n" % (" ".join(command), result[1]))
  239. return b""
  240. def openssl_text_pretty_printer(certificate_der, unused_certificate_number):
  241. return process_data_with_command(["openssl", "x509", "-text", "-inform",
  242. "DER", "-noout"], certificate_der)
  243. def pem_pretty_printer(certificate_der, unused_certificate_number):
  244. return process_data_with_command(["openssl", "x509", "-inform", "DER",
  245. "-outform", "PEM"], certificate_der)
  246. def der2ascii_pretty_printer(certificate_der, unused_certificate_number):
  247. return process_data_with_command(["der2ascii"], certificate_der)
  248. def header_pretty_printer(certificate_der, certificate_number):
  249. cert_hash = hashlib.sha256(certificate_der).hexdigest()
  250. s = """===========================================
  251. Certificate%d: %s
  252. ===========================================""" % (certificate_number, cert_hash)
  253. return s.encode("ascii")
  254. # This is actually just used as a magic value, since pretty_print_certificates
  255. # special-cases der output.
  256. def der_printer():
  257. raise RuntimeError
  258. def pretty_print_certificates(certificates_der, pretty_printers):
  259. # Need to special-case DER output to avoid adding any newlines, and to
  260. # only allow a single certificate to be output.
  261. if pretty_printers == [der_printer]:
  262. if len(certificates_der) > 1:
  263. sys.stderr.write("DER output only supports a single certificate, "
  264. "ignoring %d remaining certs\n" % (
  265. len(certificates_der) - 1))
  266. return certificates_der[0]
  267. result = b""
  268. for i in range(len(certificates_der)):
  269. certificate_der = certificates_der[i]
  270. pretty = []
  271. for pretty_printer in pretty_printers:
  272. pretty_printed = pretty_printer(certificate_der, i)
  273. if pretty_printed:
  274. pretty.append(pretty_printed)
  275. result += b"\n".join(pretty) + b"\n"
  276. return result
  277. def parse_outputs(outputs):
  278. pretty_printers = []
  279. output_map = {"der2ascii": der2ascii_pretty_printer,
  280. "openssl_text": openssl_text_pretty_printer,
  281. "pem": pem_pretty_printer,
  282. "header": header_pretty_printer,
  283. "der": der_printer}
  284. for output_name in outputs.split(','):
  285. if output_name not in output_map:
  286. sys.stderr.write("Invalid output type: %s\n" % output_name)
  287. return []
  288. pretty_printers.append(output_map[output_name])
  289. if der_printer in pretty_printers and len(pretty_printers) > 1:
  290. sys.stderr.write("Output type der must be used alone.\n")
  291. return []
  292. return pretty_printers
  293. def main():
  294. parser = argparse.ArgumentParser(
  295. description=__doc__, formatter_class=argparse.RawTextHelpFormatter)
  296. parser.add_argument('sources', metavar='SOURCE', nargs='*',
  297. help='''Each SOURCE can be one of:
  298. (1) A server name such as www.google.com.
  299. (2) A PEM [*] file containing one or more CERTIFICATE or PKCS7 blocks
  300. (3) A file containing one or more DER ASCII certificates
  301. (4) A text NetLog dump of a TLS certificate message
  302. (must include the SSL_HANDSHAKE_MESSAGE_RECEIVED line)
  303. (5) A binary file containing DER-encoded PKCS #7 signedData
  304. (6) A binary file containing DER-encoded certificate
  305. When multiple SOURCEs are listed, all certificates in them
  306. are concatenated. If no SOURCE is given then data will be
  307. read from stdin.
  308. [*] Parsing of PEM files is relaxed - leading indentation
  309. whitespace will be stripped (needed for copy-pasting data
  310. from NetLogs).''')
  311. parser.add_argument(
  312. '--output', dest='outputs', action='store',
  313. default="header,der2ascii,openssl_text,pem",
  314. help='output formats to use. Default: %(default)s')
  315. args = parser.parse_args()
  316. sources_bytes = read_sources_from_commandline(args.sources)
  317. pretty_printers = parse_outputs(args.outputs)
  318. if not pretty_printers:
  319. sys.stderr.write('No pretty printers selected.\n')
  320. sys.exit(1)
  321. certificates_der = []
  322. for source_bytes in sources_bytes:
  323. certificates_der.extend(extract_certificates(source_bytes))
  324. sys.stdout.buffer.write(
  325. pretty_print_certificates(certificates_der, pretty_printers))
  326. if __name__ == "__main__":
  327. main()