build/pgo/genpgocert.py

   1 #!/usr/bin/env python
   2 # This Source Code Form is subject to the terms of the Mozilla Public
   3 # License, v. 2.0. If a copy of the MPL was not distributed with this
   4 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
   5
   6 # This script exists to generate the Certificate Authority and server
   7 # certificates used for SSL testing in Mochitest. The already generated
   8 # certs are located at $topsrcdir/build/pgo/certs/ .
   9
  10 import os
  11 import random
  12 import re
  13 import shutil
  14 import subprocess
  15 import sys
  16
  17 import mozinfo
  18 from mozbuild.base import BinaryNotFoundException, MozbuildObject
  19 from mozfile import NamedTemporaryFile, TemporaryDirectory
  20 from mozprofile.permissions import ServerLocations
  21
  22 dbFiles = [
  23     re.compile(r"^cert[0-9]+\.db$"),
  24     re.compile(r"^key[0-9]+\.db$"),
  25     re.compile(r"^secmod\.db$"),
  26 ]
  27
  28
  29 def unlinkDbFiles(path):
  30     for root, dirs, files in os.walk(path):
  31         for name in files:
  32             for dbFile in dbFiles:
  33                 if dbFile.match(name) and os.path.exists(os.path.join(root, name)):
  34                     os.unlink(os.path.join(root, name))
  35
  36
  37 def dbFilesExist(path):
  38     for root, dirs, files in os.walk(path):
  39         for name in files:
  40             for dbFile in dbFiles:
  41                 if dbFile.match(name) and os.path.exists(os.path.join(root, name)):
  42                     return True
  43     return False
  44
  45
  46 def runUtil(util, args, inputdata=None, outputstream=None):
  47     env = os.environ.copy()
  48     if mozinfo.os == "linux":
  49         pathvar = "LD_LIBRARY_PATH"
  50         app_path = os.path.dirname(util)
  51         if pathvar in env:
  52             env[pathvar] = "%s%s%s" % (app_path, os.pathsep, env[pathvar])
  53         else:
  54             env[pathvar] = app_path
  55     proc = subprocess.Popen(
  56         [util] + args,
  57         env=env,
  58         stdin=subprocess.PIPE if inputdata else None,
  59         stdout=outputstream,
  60         universal_newlines=True,
  61     )
  62     proc.communicate(inputdata)
  63     return proc.returncode
  64
  65
  66 def createRandomFile(randomFile):
  67     for count in xrange(0, 2048):
  68         randomFile.write(chr(random.randint(0, 255)))
  69
  70
  71 def writeCertspecForServerLocations(fd):
  72     locations = ServerLocations(
  73         os.path.join(build.topsrcdir, "build", "pgo", "server-locations.txt")
  74     )
  75     SAN = []
  76     for loc in [
  77         i for i in iter(locations) if i.scheme == "https" and "nocert" not in i.options
  78     ]:
  79         customCertOption = False
  80         customCertRE = re.compile(r"^cert=(?:\w+)")
  81         for _ in [i for i in loc.options if customCertRE.match(i)]:
  82             customCertOption = True
  83             break
  84
  85         if "ipV4Address" in loc.options:
  86             loc.host = "ip4:" + loc.host
  87
  88         if not customCertOption:
  89             SAN.append(loc.host)
  90
  91     fd.write(
  92         "issuer:printableString/CN=Temporary Certificate Authority/O=Mozilla Testing/OU=Profile Guided Optimization\n"  # NOQA: E501
  93     )
  94     fd.write("subject:{}\n".format(SAN[0]))
  95     fd.write("extension:subjectAlternativeName:{}\n".format(",".join(SAN)))
  96
  97
  98 def constructCertDatabase(build, srcDir):
  99     try:
 100         certutil = build.get_binary_path(what="certutil")
 101         pk12util = build.get_binary_path(what="pk12util")
 102     except BinaryNotFoundException as e:
 103         print("{}\n\n{}\n".format(e, e.help()))
 104         return 1
 105     openssl = shutil.which("openssl")
 106     pycert = os.path.join(build.topsrcdir, "security", "manager", "tools", "pycert.py")
 107     pykey = os.path.join(build.topsrcdir, "security", "manager", "tools", "pykey.py")
 108
 109     with NamedTemporaryFile(mode="wt+") as pwfile, TemporaryDirectory() as pemfolder:
 110         pwfile.write("\n")
 111         pwfile.flush()
 112
 113         if dbFilesExist(srcDir):
 114             # Make sure all DB files from src are really deleted
 115             unlinkDbFiles(srcDir)
 116
 117         # Copy  all .certspec and .keyspec files to a temporary directory
 118         for root, dirs, files in os.walk(srcDir):
 119             for spec in [
 120                 i for i in files if i.endswith(".certspec") or i.endswith(".keyspec")
 121             ]:
 122                 shutil.copyfile(os.path.join(root, spec), os.path.join(pemfolder, spec))
 123
 124         # Write a certspec for the "server-locations.txt" file to that temporary directory
 125         pgoserver_certspec = os.path.join(pemfolder, "pgoserver.certspec")
 126         if os.path.exists(pgoserver_certspec):
 127             raise Exception(
 128                 "{} already exists, which isn't allowed".format(pgoserver_certspec)
 129             )
 130         with open(pgoserver_certspec, "w") as fd:
 131             writeCertspecForServerLocations(fd)
 132
 133         # Generate certs for all certspecs
 134         for root, dirs, files in os.walk(pemfolder):
 135             for certspec in [i for i in files if i.endswith(".certspec")]:
 136                 name = certspec.split(".certspec")[0]
 137                 pem = os.path.join(pemfolder, "{}.cert.pem".format(name))
 138
 139                 print("Generating public certificate {} (pem={})".format(name, pem))
 140
 141                 with open(os.path.join(root, certspec), "r") as certspec_file:
 142                     certspec_data = certspec_file.read()
 143                     with open(pem, "w") as pem_file:
 144                         status = runUtil(
 145                             pycert, [], inputdata=certspec_data, outputstream=pem_file
 146                         )
 147                         if status:
 148                             return status
 149
 150                 status = runUtil(
 151                     certutil,
 152                     [
 153                         "-A",
 154                         "-n",
 155                         name,
 156                         "-t",
 157                         "P,,",
 158                         "-i",
 159                         pem,
 160                         "-d",
 161                         srcDir,
 162                         "-f",
 163                         pwfile.name,
 164                     ],
 165                 )
 166                 if status:
 167                     return status
 168
 169             for keyspec in [i for i in files if i.endswith(".keyspec")]:
 170                 parts = keyspec.split(".")
 171                 name = parts[0]
 172                 key_type = parts[1]
 173                 if key_type not in ["ca", "client", "server"]:
 174                     raise Exception(
 175                         "{}: keyspec filenames must be of the form XXX.client.keyspec "
 176                         "or XXX.ca.keyspec (key_type={})".format(keyspec, key_type)
 177                     )
 178                 key_pem = os.path.join(pemfolder, "{}.key.pem".format(name))
 179
 180                 print("Generating private key {} (pem={})".format(name, key_pem))
 181
 182                 with open(os.path.join(root, keyspec), "r") as keyspec_file:
 183                     keyspec_data = keyspec_file.read()
 184                     with open(key_pem, "w") as pem_file:
 185                         status = runUtil(
 186                             pykey, [], inputdata=keyspec_data, outputstream=pem_file
 187                         )
 188                         if status:
 189                             return status
 190
 191                 cert_pem = os.path.join(pemfolder, "{}.cert.pem".format(name))
 192                 if not os.path.exists(cert_pem):
 193                     raise Exception(
 194                         "There has to be a corresponding certificate named {} for "
 195                         "the keyspec {}".format(cert_pem, keyspec)
 196                     )
 197
 198                 p12 = os.path.join(pemfolder, "{}.key.p12".format(name))
 199                 print(
 200                     "Converting private key {} to PKCS12 (p12={})".format(key_pem, p12)
 201                 )
 202                 status = runUtil(
 203                     openssl,
 204                     [
 205                         "pkcs12",
 206                         "-export",
 207                         "-inkey",
 208                         key_pem,
 209                         "-in",
 210                         cert_pem,
 211                         "-name",
 212                         name,
 213                         "-out",
 214                         p12,
 215                         "-passout",
 216                         "file:" + pwfile.name,
 217                     ],
 218                 )
 219                 if status:
 220                     return status
 221
 222                 print("Importing private key {} to database".format(key_pem))
 223                 status = runUtil(
 224                     pk12util,
 225                     ["-i", p12, "-d", srcDir, "-w", pwfile.name, "-k", pwfile.name],
 226                 )
 227                 if status:
 228                     return status
 229
 230                 if key_type == "ca":
 231                     shutil.copyfile(
 232                         cert_pem, os.path.join(srcDir, "{}.ca".format(name))
 233                     )
 234                 elif key_type == "client":
 235                     shutil.copyfile(p12, os.path.join(srcDir, "{}.client".format(name)))
 236                 elif key_type == "server":
 237                     pass  # Nothing to do for server keys
 238                 else:
 239                     raise Exception(
 240                         "State error: Unknown keyspec key_type: {}".format(key_type)
 241                     )
 242
 243     return 0
 244
 245
 246 build = MozbuildObject.from_environment()
 247 certdir = os.path.join(build.topsrcdir, "build", "pgo", "certs")
 248 certificateStatus = constructCertDatabase(build, certdir)
 249 if certificateStatus:
 250     print("TEST-UNEXPECTED-FAIL | SSL Server Certificate generation")
 251 sys.exit(certificateStatus)