update-certdata

#!/usr/bin/python3
# vim:set et sw=4:
#
# update-certdata - download certdata.txt and apply to system trust

# Prints extraneous debuging info and sest DESTDIR to './dest' by default
debug = False

import base64
import datetime
import os
import re
import shutil
import ssl
import subprocess
import sys
import tempfile
import textwrap
import time
import urllib.request, urllib.parse, urllib.error

# Get DESTDIR value from environment if set
DESTDIR = os.getenv('DESTDIR')
if DESTDIR is None:
  DESTDIR = ''
  if debug == True:
    DESTDIR = './dest'

# Do not depend on path, provide them in the config file if not correct
MD5SUM  = "/usr/bin/md5sum"
OPENSSL = "/usr/bin/openssl"
PRINTF  = "/usr/bin/printf" # Yeah, I'm cheating here, it works
TRUST   = "/usr/bin/trust"

# Destination configuration
PKIDIR     = DESTDIR + "/etc/pki"
SSLDIR     = DESTDIR + "/etc/ssl"
ANCHORDIR  = PKIDIR + "/anchors"
ANCHORLIST = PKIDIR + "/anchors.md5sums"
BUNDLEDIR  = PKIDIR + "/tls/certs"
SABUNDLE   = BUNDLEDIR + "/ca-bundle.crt"
SMBUNDLE   = BUNDLEDIR + "/email-ca-bundle.crt"
CSBUNDLE   = BUNDLEDIR + "/objsign-ca-bundle.crt"
CERTDIR    = SSLDIR + "/certs"
JAVADIR    = PKIDIR + "/tls/java"
KEYSTORE   = JAVADIR + "/cacerts"
NSSDB      = PKIDIR + "/nssdb"
LOCALDIR   = SSLDIR + "/local"
LIBEXECDIR = "/usr/libexec/ca-tools"
CERTDATA   = SSLDIR + "/certdata.txt"
MOZCA      = LIBEXECDIR + "/mozilla-ca-root.pem"
SRCHOST = "hg.mozilla.org"
SRCURL = "https://" + SRCHOST + "/projects/nss/raw-file/tip/lib/ckfw/builtins/certdata.txt"
SRCDAT = SRCURL.replace("raw-file", "log")
PROXY = ''

# Create a temporary working directory
TEMPDIR = tempfile.mkdtemp(prefix="tmp.certdata.")
if debug == True:
  print("Tempdir is %s" % (TEMPDIR))

# Create default ssl context with MOZZILA_CA_ROOT as the only CA
context = ssl.create_default_context()
context.load_verify_locations(cafile=MOZCA)

def get_cert_hash(filepath):
  gethash = [OPENSSL, "x509", "-hash", "-in", filepath, "-noout"]
  certhash = subprocess.check_output(gethash).decode('utf-8').strip()
  return certhash;

def get_trust_value(trustlist):
  # Certs in CCADB will never be distrusted, but left for completeness
  if trust == 'Distrusted':
    trust_ext_oid = "1.3.6.1.4.1.3319.6.10.1"
    trust_bool = "false"
    trust_ext_value = "0.%06%0a%2b%06%01%04%01%99w%06%0a%01%04 0%1e%06%08%2b%06%01%05%05%07%03%04%06%08%2b%06%01%05%05%07%03%01%06%08%2b%06%01%05%05%07%03%03"
  else:
    trust_ext_oid = "2.5.29.37"
    trust_bool = "true"
    if trust == 'satrust':
      trust_ext_value = "0%16%06%03U%1d%25%01%01%ff%04%0c0%0a%06%08%2b%06%01%05%05%07%03%01"
    if trust == 'satrust,smtrust':
      trust_ext_value = "0 %06%03U%1d%25%01%01%ff%04%160%14%06%08%2b%06%01%05%05%07%03%04%06%08%2b%06%01%05%05%07%03%01"
    if trust == 'satrust,cstrust':
      trust_ext_value = "0 %06%03U%1d%25%01%01%ff%04%160%14%06%08%2b%06%01%05%05%07%03%01%06%08%2b%06%01%05%05%07%03%03"
    if trust == 'satrust,smtrust,cstrust':
      trust_ext_value = "0%2a%06%03U%1d%25%01%01%ff%04 0%1e%06%08%2b%06%01%05%05%07%03%04%06%08%2b%06%01%05%05%07%03%01%06%08%2b%06%01%05%05%07%03%03"
    if trust == 'smtrust':
      trust_ext_value = "0%16%06%03U%1d%25%01%01%ff%04%0c0%0a%06%08%2b%06%01%05%05%07%03%04"
    if trust == 'smtrust,cstrust':
      trust_ext_value = "0 %06%03U%1d%25%01%01%ff%04%160%14%06%08%2b%06%01%05%05%07%03%04%06%08%2b%06%01%05%05%07%03%03"
    if trust == 'cstrust':
      trust_ext_value = "0%16%06%03U%1d%25%01%01%ff%04%0c0%0a%06%08%2b%06%01%05%05%07%03%03"
    if trust == '':
      trust_ext_value = "0%18%06%03U%1d%25%01%01%ff%04%0e0%0c%06%0a%2b%06%01%04%01%99w%06%0a%10"
  return trust_ext_oid,trust_ext_value,trust_bool;

def get_p11_label(srcpem):
  ostcmd = [OPENSSL, "x509", "-in", srcpem, "--noout", "-text"]
  openssl_text = subprocess.check_output(ostcmd).decode('utf-8').strip()
  # Note: When using CCADB, there is no CKA_LABEL available
  # label can be taken from the Subject line using CN, OU, or O (in that order)
  for line in openssl_text.split("\n"):
    if "Subject:" in line:
      subject_line = line.strip().replace(" = ", "=")
  if "CN" in subject_line:
    labelrex = re.search(r"CN=.*", subject_line)
    label = labelrex.group(0).replace("CN=", "")
  elif "OU" in subject_line:
    labelrex = re.search(r"OU=.*", subject_line)
    label = re.sub(",.*$", "", labelrex.group(0).replace("OU=", ""))
  elif "O" in subject_line:
    labelrex = re.search(r"O=.*", subject_line)
    label = re.sub(",.*$", "", labelrex.group(0).replace("O=", ""))
  return label;

def write_p11_file(srcpem,label,trust_oid,trust_value,trust_bool,trust,svr_dt_val,eml_dt_val):
  os.makedirs(TEMPDIR + "/anchors", mode = 0o755, exist_ok = True)
  ostcmd = [OPENSSL, "x509", "-in", srcpem, "--noout", "-text"]
  openssl_text = subprocess.check_output(ostcmd).decode('utf-8').strip()
  oskcmd = [OPENSSL, "x509", "-in", srcpem, "-noout", "-pubkey"]
  openssl_key = subprocess.check_output(oskcmd).decode('utf-8').strip()
  keyhash = get_cert_hash(srcpem)
  filepath = ANCHORDIR + "/" + keyhash + ".p11-kit"
  with open(srcpem, 'r') as file:
    srcpem_text = file.read()
  print("\nWriting anchor: %s\n  Label: %s\n  Keyhash: %s\n  Trust: %s\n" % (filepath, label, keyhash, trust.replace("satrust", "Server Auth").replace("smtrust", "E-mail").replace("cstrust", "Code Signing").replace(",", ", ")))
  writep11 = open(filepath, 'w')
  writep11.write("[p11-kit-object-v1]\n")
  writep11.write("label: \"" + label + "\"\n")
  writep11.write("class: x-certificate-extension\n")
  writep11.write("object-id: " + trust_oid + "\n")
  writep11.write("value: \"" + trust_value + "\"\n")
  writep11.write("modifiable: false\n")
  writep11.write(openssl_key + "\n\n")
  writep11.write("[p11-kit-object-v1]\n")
  writep11.write("label: \"" + label + "\"\n")
  if trust_bool == "true":
    writep11.write("trusted: true\n")
  else:
    writep11.write("x-distrusted: true\n")
  # Note: nss-mozilla-ca-policy is always true when using certdata.txt
  writep11.write("nss-mozilla-ca-policy: true\n")
  writep11.write("modifiable: false\n")
  writep11.write("nss-server-distrust-after: \"" + svr_dt_val + "\"\n")
  writep11.write("nss-email-distrust-after: \"" + eml_dt_val + "\"\n")
  writep11.write(srcpem_text)
  writep11.write(re.sub(r"\n", "\n#", (re.sub(r"^", '#', openssl_text))) + "\n")
  return 0;

def write_openssl_store():
  # Create OpenSSL Trusted store
  print("Extracting OpenSSL certificates to:\n%s..." % (CERTDIR), end='')
  create_OTS = [TRUST, "extract", "--filter=ca-anchors", "--format=openssl-directory", "--overwrite", "--comment", CERTDIR]
  subprocess.run(create_OTS, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
  print("Done!")

def write_gnutls_bundle():
  # Create GNUTLS bundle file
  print("Extracting GNUTLS server auth certificates to:\n%s..." % (SABUNDLE), end='')
  create_GBF = [TRUST, "extract", "--filter=ca-anchors", "--format=pem-bundle", "--purpose", "server-auth", "--overwrite", "--comment", SABUNDLE]
  subprocess.run(create_GBF, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
  print("Done!")

def write_gnutls_smime():
  # Create GNUTLS S-Mime bundle
  print("Extracting GNUTLS S-Mime certificates to:\n%s..." % (SMBUNDLE), end='')
  create_GSM = [TRUST, "extract", "--filter=ca-anchors", "--format=pem-bundle", "--purpose", "email", "--overwrite", "--comment", SMBUNDLE]
  subprocess.run(create_GSM, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
  print("Done!")

def write_gnutls_cs():
  # Create GNUTLS Code Signing bundle
  print("Extracting GNUTLS code signing certificates to:\n%s..." % (CSBUNDLE), end='')
  create_GCS = [TRUST, "extract", "--filter=ca-anchors", "--format=pem-bundle", "--purpose", "code-signing", "--overwrite", "--comment", CSBUNDLE]
  subprocess.run(create_GCS, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
  print("Done!")

def write_java_cacerts():
  # Create Java certificate store
  print("Extracting Java cacerts (JKS) to:\n%s..." % (KEYSTORE), end='')
  create_JKS = [TRUST, "extract", "--filter=ca-anchors", "--format=java-cacerts", "--purpose", "server-auth", "--overwrite", "--comment", KEYSTORE]
  subprocess.run(create_JKS, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
  print("Done!")

# Get hg revision of existing certdata.txt
if os.path.isfile(CERTDATA):
  # Get the currnet revision
  search = "# Revision:"
  with open(CERTDATA, 'r') as cdfile:
    lines = cdfile.readlines()
    for line in lines:
      if line.find(search) != -1:
          oldrev = line.replace('# Revision:', '')
          if debug == True:
            print("oldrev = %s" % (oldrev))
          break
else:
  oldrev = ''

# Get the revision directly from hg
certdatalogdl = urllib.request.urlopen(SRCDAT, data=None, context=context)
certdatalogfn = TEMPDIR + "/certdata.txt.log"
certdatalogbytes = certdatalogdl.read()
certdatalogtext = certdatalogbytes.decode('utf-8')
savelogfile = open(certdatalogfn, 'w')
savelogfile.write(certdatalogtext)
savelogfile.close()

search = "<br/>created <i>"
with open(certdatalogfn, 'r') as newlog:
  lines = newlog.readlines()
  for line in lines:
    if line.find(search) != -1:
      newrev = re.sub("<br/>created <i>.*", '', line)
      if debug == True:
        print("newrev = %s" % (newrev))
      break

# Make sure destination directories exist
os.makedirs(ANCHORDIR, mode = 0o755, exist_ok = True)
os.makedirs(BUNDLEDIR, mode = 0o755, exist_ok = True)
os.makedirs(CERTDIR, mode = 0o755, exist_ok = True)
os.makedirs(JAVADIR, mode = 0o755, exist_ok = True)
os.makedirs(LOCALDIR, mode = 0o755, exist_ok = True)

# If the match, just use the existing file
if oldrev == newrev:
  print("Existing file %s is up to date. Skipping download." % (CERTDATA))
else:
  print("Downloading updated certdata.txt...")
  certdatadl = urllib.request.urlopen(SRCURL, data=None, context=context)
  certdatafn = TEMPDIR + "/certdata.txt"
  certdatabytes = certdatadl.read()
  certdatatext = certdatabytes.decode('utf-8')
  savefile = open(certdatafn, 'w')
  savefile.write("# Revision:" + newrev + "\n" + certdatatext)
  savefile.close()
  if oldrev != '':
    os.rename(CERTDATA, CERTDATA + ".old")
  os.rename(TEMPDIR + "/certdata.txt", CERTDATA)

# Before we populate it, we need to clear out ANCHORDIR
for p11file in os.listdir(ANCHORDIR):
  if debug == True:
    print("Checking file: %s" % (os.path.join(ANCHORDIR, p11file)))
  if p11file.endswith(".p11-kit"):
    os.remove(os.path.join(ANCHORDIR, p11file))

os.makedirs(TEMPDIR + "/certs", mode = 0o755, exist_ok = True)
with open(CERTDATA) as certdata_file:
  pattern = r'^# Certificate.*?^CKA_TRUST_STEP.*?$'
  parts = re.findall(pattern, certdata_file.read(), flags = re.DOTALL | re.MULTILINE)
  for part in parts:
    pattern = r'^CKA_VALUE MULTILINE_OCTAL.*?^END$'
    octval = re.search(pattern, part, flags = re.DOTALL | re.MULTILINE)
    octvalstr = octval.group(0).replace("END", '').replace("CKA_VALUE MULTILINE_OCTAL\n", '').replace("\n", '')
    certtemp = tempfile.NamedTemporaryFile(delete=False, dir=(TEMPDIR + "/certs"))
    der_dump_cmd = [PRINTF, octvalstr]
    der_data = subprocess.run(der_dump_cmd, stdout=subprocess.PIPE).stdout
    der_data_b64 = base64.b64encode(der_data)
    pem_data = textwrap.wrap(der_data_b64.decode(), 64)
    writetemp = open(certtemp.name, 'w')
    writetemp.write("-----BEGIN CERTIFICATE-----\n")
    writetemp.write("\n".join(pem_data))
    writetemp.write("\n-----END CERTIFICATE-----\n")
    writetemp.close()
    chash = get_cert_hash(certtemp.name)

    svr_distrust = "false"
    pattern = r'^CKA_NSS_SERVER_DISTRUST_AFTER.*?$'
    svr_dt_line = re.search(pattern, part, flags = re.DOTALL | re.MULTILINE)
    if not svr_dt_line is None:
      svr_dt_line = svr_dt_line.group(0).split(" ")
      if svr_dt_line[1] == "CK_BBOOL":
        if svr_dt_line[2] == "CK_FALSE":
          svr_dt_val = "%00"
      elif svr_dt_line[1] == "MULTILINE_OCTAL":
        pattern = r'^CKA_NSS_SERVER_DISTRUST_AFTER.*?^END$'
        svr_dt_mlv = re.search(pattern, part, flags = re.DOTALL | re.MULTILINE)
        svr_dt_mlv_str = svr_dt_mlv.group(0).replace("END", '').replace("CKA_NSS_SERVER_DISTRUST_AFTER MULTILINE_OCTAL\n", '').replace("\n", '')
        b64_sdump_cmd = [PRINTF, svr_dt_mlv_str]
        svr_dt_mlv_data = subprocess.run(b64_sdump_cmd, stdout=subprocess.PIPE).stdout
        svr_dt_val = str(svr_dt_mlv_data).replace("'", "").replace('b', '')
        # FIXME - Remove after p11-kit brakage is fixed
        # Set Distrusted if past the nss-server-distrust-after date
        if svr_dt_val != "%00":
          svr_dt_date  = svr_dt_val[:6]
          today_date = datetime.datetime.today().strftime('%y%m%d')
          if today_date > svr_dt_date:
            expdate = "20" + svr_dt_val[:2] + "-" + svr_dt_val[2:4] + "-" + svr_dt_val[4:6]
            print("The following cetificate is dustrsted for server auth after %s:" % (expdate))
            svr_distrust = "true"
      else:
        # We have no idea what this is so move on to the next file
        print("Error processing crtificate with hash: %s" % (chash))
        continue
    else:
      svr_dt_val = "%00"

    eml_distrust = "false"
    pattern = r'^CKA_NSS_EMAIL_DISTRUST_AFTER.*?$'
    eml_dt_line = re.search(pattern, part, flags = re.DOTALL | re.MULTILINE)
    if not eml_dt_line is None:
      eml_dt_line = eml_dt_line.group(0).split(" ")
      if eml_dt_line[1] == "CK_BBOOL":
        if eml_dt_line[2] == "CK_FALSE":
          eml_dt_val = "%00"
      elif eml_dt_line[1] == "MULTILINE_OCTAL":
        pattern = r'^CKA_NSS_EMAIL_DISTRUST_AFTER.*?^END$'
        eml_dt_mlv = re.search(pattern, part, flags = re.DOTALL | re.MULTILINE)
        eml_dt_mlv_str = eml_dt_mlv.group(0).replace("END", '').replace("CKA_NSS_EMAIL_DISTRUST_AFTER MULTILINE_OCTAL\n", '').replace("\n", '')
        b64_edump_cmd = [PRINTF, eml_dt_mlv_str]
        eml_dt_mlv_data = subprocess.run(b64_edump_cmd, stdout=subprocess.PIPE).stdout
        eml_dt_val = str(eml_dt_mlv_data).replace("'", "").replace('b', '')
        # FIXME - Remove after p11-kit breakage is fixed
        # Set Distrusted if past the nss-email-distrust-after date
        if eml_dt_val != "%00":
          eml_dt_date  = svr_dt_val[:6]
          today_date = datetime.datetime.today().strftime('%y%m%d')
          if today_date > eml_dt_date:
            expdate = "20" + eml_dt_val[:2] + "-" + eml_dt_val[2:4] + "-" + eml_dt_val[4:6]
            print("The following cetificate is dustrsted for email auth after %s:" % (expdate))
            eml_distrust = "true"
      else:
        # We have no idea what this is os move on to the next file
        print("Error processing crtificate with hash: %s" % (chash))
    else:
      eml_dt_val = "%00"

    # Determine trust
    trust = ""
    pattern = r'^CKA_TRUST_SERVER_AUTH.*?$'
    svr_trust_line = re.search(pattern, part, flags = re.DOTALL | re.MULTILINE).group(0).split(" ")
    # FIXME - Best we can do for now
    if svr_distrust == "true":
      trust = "Distrusted"
    elif svr_trust_line[1] == "CK_TRUST":
      if svr_trust_line[2] == "CKT_NSS_TRUSTED_DELEGATOR":
        trust = "satrust"
      elif svr_trust_line[2] == "CKT_NSS_MUST_VERIFY_TRUST":
        trust = ""
      elif svr_trust_line[2] == "CKT_NSS_NOT_TRUSTED":
        trust = "Distrusted"
      else:
        # We have no idea what this is os move on to the next file
        print("Error processing crtificate with hash: %s" % (chash))
        continue
    else:
      # We have no idea what this is os move on to the next file
      print("Error processing crtificate with hash: %s" % (chash))
      continue
    
    pattern = r'^CKA_TRUST_EMAIL_PROTECTION.*?$'
    eml_trust_line = re.search(pattern, part, flags = re.DOTALL | re.MULTILINE).group(0).split(" ")
    # FIXME - Best we can do for now
    if eml_distrust == "true":
      trust = "Distrusted"
    elif eml_trust_line[1] == "CK_TRUST":
      if eml_trust_line[2] == "CKT_NSS_TRUSTED_DELEGATOR":
        if trust == "":
          trust = "smtrust"
        else:
          trust += ",smtrust"
      elif eml_trust_line[2] == "CKT_NSS_MUST_VERIFY_TRUST":
        # Don't do anything
        xtrust = "" # may want to do more here later, so left it in with an effective noop
      elif eml_trust_line[2] == "CKT_NSS_NOT_TRUSTED":
        trust = "Distrusted"
      else:
        # We have no idea what this is os move on to the next file
        print("Error processing crtificate with hash: %s" % (chash))
        continue
    else:
      # We have no idea what this is os move on to the next file
      print("Error processing crtificate with hash: %s" % (chash))
      continue

    pattern = r'^CKA_TRUST_CODE_SIGNING.*?$'
    cs_trust_line = re.search(pattern, part, flags = re.DOTALL | re.MULTILINE).group(0).split(" ")
    if cs_trust_line[1] == "CK_TRUST":
      if cs_trust_line[2] == "CKT_NSS_TRUSTED_DELEGATOR":
        if trust == "":
          trust = "cstrust"
        else:
          trust += ",cstrust"
      elif cs_trust_line[2] == "CKT_NSS_MUST_VERIFY_TRUST":
        # Don't do anything
        xtrust = "" # may want to do more here later, so left it in with an effective noop
      elif cs_trust_line[2] == "CKT_NSS_NOT_TRUSTED":
        trust = "Distrusted"
      else:
        # We have no idea what this is os move on to the next file
        print("Error processing crtificate with hash: %s" % (chash))
        continue
    else:
      # We have no idea what this is os move on to the next file
      print("Error processing crtificate with hash: %s" % (chash))
      continue

    # Label
    pattern = r'^CKA_LABEL.*?$'
    label_line = re.search(pattern, part, flags = re.DOTALL | re.MULTILINE).group(0).split('"')
    label = label_line[1]
    if label == "":
      print("GETTING LABEL FROM PEM\n")
      label = get_p11_label(srcpem)

    trust_oid,trust_value,trust_bool = get_trust_value(trust)
    write_p11_file(certtemp.name,label,trust_oid,trust_value,trust_bool,trust,svr_dt_val,eml_dt_val)

# Process any local pem files
for pem in os.scandir(LOCALDIR):
  # provide a warning for files that end specifically with .pem
  if pem.name.endswith(".pem"):
    print("\n\033[93mWarning\033[0m: PEM format files are no longer supported in %s!" % (LOCALDIR))
    print("Skipping %s!" % (pem.name))
    print("Please convert the file to P11-Kit format using the \033[94madd-local-trust\033[0m utility.\n")
  if pem.name.endswith(".p11-kit"):
    print("\nCopying local trust file %s to the trust store." % (pem.name))
    shutil.copyfile(pem, ANCHORDIR + os.path.basename(pem))
  elif not pem.name.endswith(".pem"):
    print("\n\033[91mError\033[0m: Only P11-Kit format files are supported!")
    print("Skipping %s!" % (pem.name))

# Create an md5sum list of certificates - this is used to make sure
# that any certificates adding using the native trust utility also get
# added to the local directory
md5text = ''
for file in os.scandir(ANCHORDIR):
  if file.name.endswith(".p11-kit"):
    md5cmd = [MD5SUM, os.path.join(ANCHORDIR, file.name)]
    md5text += subprocess.check_output(md5cmd).decode('utf-8').strip() + "\n"
anchorlist_write = open(ANCHORLIST, 'w')
anchorlist_write.write(md5text.replace(DESTDIR, ''))
anchorlist_write.close()

# Only create the bundles if on a real system, else the generated files
# will use the real trust directory, not our newly created one
if DESTDIR == '':
  # FIXME: Move to functions for now and eliminate p11-kit requirement.
  write_openssl_store()
  write_gnutls_bundle()
  write_gnutls_smime()
  write_gnutls_cs()
  write_java_cacerts()

else:
  print("\n\033[93mWarning\033[0m: Not updating 3rd party bundles when DESTDIR is not empty.\n")

if debug == False:
  # Clean up the temp directory
  shutil.rmtree(TEMPDIR)