certdata2bundle.py — retrieves Mozilla/NS root trust store, re-encodes DER/base64 as PEM bundle sans metadata

certdata2bundle.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# vim:set et sw=4:
#
# certdata2bundle.py
# retrieves CA certificates from the Mozilla/NSS root trust store in base64/DER
# format and re-encodes them as a concatenated PEM bundle sans metadata
# writes all CA certificates to /etc/ssl/cacerts.pem
#
# Copyright (C) 2019 Kevin M. Gallagher <[email protected]>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.

# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.

# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.

# import base64
import os.path
import argparse
import sys
import hashlib
import certifi
import filecmp
import tempfile

# import textwrap
# import io
from pick import pick
import re
import urllib3
from cryptography import x509
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives import hashes
from cryptography.x509.oid import NameOID

ca_bundle_output = "/etc/ssl/cacerts.crt"
base64_certs = []
x509_certs = []


def certdata():
    certdata_url = "https://hg.mozilla.org/mozilla-central/raw-file/tip/security/nss/lib/ckfw/builtins/certdata.txt"
    http = urllib3.PoolManager(cert_reqs="CERT_REQUIRED", ca_certs=certifi.where())
    try:
        r = http.request(
            "GET",
            certdata_url,
            timeout=30,
            retries=urllib3.Retry(5, raise_on_redirect=False),
        )
    except Exception as e:
        print(e)
        sys.exit(1)
    if r.status == 200:
        return r.data.decode("utf-8")
    else:
        return None


def extract_certs(data):
    certs = re.findall(
        r"CKA_VALUE MULTILINE_OCTAL.*?END", data, flags=re.DOTALL | re.MULTILINE
    )
    return list(filter(None, certs))


def parse_cmd_args():
    description = """
# Remotely retrieves CA certificates from the Mozilla/NSS root trust store, using TLS
# in base64/DER format and re-encodes them as a PEM sans metadata
# Default mode (--bundle) is to write all CA certificates to /etc/ssl/cacerts.pem
# However (--individual) will write CA certificates to separate files in /etc/ssl/certs/ but only
# if the content has changed, in which case you'll be prompted, or if the CA isn't already found in the system trust store.
  """
    parser = argparse.ArgumentParser(
        description=description, formatter_class=argparse.RawTextHelpFormatter
    )
    group = parser.add_mutually_exclusive_group()
    group.add_argument(
        "--bundle",
        required=False,
        action="store_true",
        default=True,
        help="Write a single concatenated PEM bundle of all CA certs",
    )
    group.add_argument(
        "--individual",
        required=False,
        action="store_true",
        default=False,
        help="Write CA certificates to individual *.PEM files in /etc/ssl/certs",
    )
    return parser.parse_args()


def main():
    args = parse_cmd_args()
    base64_certs = extract_certs(certdata())
    print("Found " + str(len(base64_certs)) + " trusted certificates.")

    for cert in base64_certs:
        ca_dict = {"name": None, "pubkey": None, "fingerprint": None, "serial": None}
        cert = cert.lstrip("CKA_VALUE MULTILINE_OCTAL")
        cert = cert.rstrip("END")
        cert = cert.replace("\n", "")
        cert = cert.encode("utf-8").decode("unicode_escape").encode("latin-1")
        try:
            x509_cert = x509.load_der_x509_certificate(cert, default_backend())
        except Exception as e:
            print(e)
            continue
        try:
            ca_dict["name"] = str(
                x509_cert.subject.get_attributes_for_oid(NameOID.COMMON_NAME)[0].value
            )
        except IndexError as e:
            ca_dict["name"] = ""
            pass
        ca_dict["pubkey"] = x509_cert.public_bytes(serialization.Encoding.PEM)
        ca_dict["fingerprint"] = "".join(
            "{:02x}".format(x) for x in x509_cert.fingerprint(hashes.SHA256())
        )
        ca_dict["serial"] = str(x509_cert.serial_number)
        x509_certs.append(ca_dict)
        if args.individual:
            ca_filename = "/etc/ssl/certs/" + ca_dict["name"].replace(" ", "_") + ".pem"
            if os.path.isfile(ca_filename):
                with tempfile.NamedTemporaryFile(delete=False) as ca_tmpfile:
                    ca_tmpfile.write(ca_dict["pubkey"])
                    ca_tmpfile_name = ca_tmpfile.name
                    ca_tmpfile.flush()
                    ca_tmpfile.close()
                if not filecmp.cmp(ca_filename, ca_tmpfile_name):
                    prompt = (
                        ca_filename
                        + " already exists, but the content has changed. Do you want to overwrite it?"
                    )
                    options = ["Yes", "No"]
                    selected = pick(
                        options, prompt, multi_select=False, min_selection_count=1
                    )
                    if selected[1] == 0:
                        f = open(ca_filename, "wb")
                        f.write(ca_dict["pubkey"])
                        f.close()
                else:
                    f = open(ca_filename, "wb")
                    f.write(ca_dict["pubkey"])
                    f.close()
        print(
            "Processed "
            + ca_dict["name"]
            + ", SHA256 fingerprint: "
            + ca_dict["fingerprint"]
            + ", serial: "
            + ca_dict["serial"]
            + "."
        )

    if args.bundle:
        f = open(ca_bundle_output, "wb")

        for cert in x509_certs:
            f.write(cert["pubkey"])
        f.close()
        print(
            "Wrote "
            + str(len(x509_certs))
            + " PEM CA certificates to "
            + ca_bundle_output
            + "."
        )
        f = open(ca_bundle_output, "rb")
        f_bytes = f.read()
        ca_bundle_hash = hashlib.sha256(f_bytes).hexdigest()
        f.close()
        print(
            "For integrity, "
            + ca_bundle_output
            + " has SHA256 hash of "
            + ca_bundle_hash
        )
        sys.exit(0)


if __name__ == "__main__":
    main()