signing/bin/update_release_keys.py - third_party/chromite - Git at Google

 # Copyright 2019 The ChromiumOS Authors
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.

 """Import keys from legacy signer config."""

 import configparser
 import logging
 import os

 import yaml  # pylint: disable=import-error

 from chromite.lib import commandline
 from chromite.lib import osutils
 from chromite.signing.lib import keys


 METADATA_VERSION = 0
 PROD_DIR = "/cros"
 BASE_DIR = os.path.join(PROD_DIR, "v2-keys")
 AU_KEYDIR = "update-payload-key"


 class Error(Exception):
     """Base error class."""


 class KeyimportError(Error):
     """Key Import Errors."""


 class KeyringData:
     """A collection of keyset information."""

     def __init__(self, prod_dir, base_dir, options, signer_config):
         self.prod_dir = prod_dir
         self.base_dir = base_dir
         self.options = options
         self.signer_config = signer_config
         self.prod_keys = os.path.join(prod_dir, "keys")
         self.public = os.path.join(base_dir, "public")
         self.private = "../private"
         self.configs = "../config"
         self.contents_yaml = os.path.join(self.base_dir, "contents.yaml")
         self.keysets = self._ReadRepoYaml()

     def _ReadRepoYaml(self):
         """Read the directory of active keysets in the repo."""
         default = {"metadata-version": METADATA_VERSION}
         if os.path.exists(self.contents_yaml):
             with open(self.contents_yaml, encoding="utf-8") as fp:
                 ret = yaml.safe_load(fp)
             if not ret or "metadata-version" not in ret:
                 logging.error("%s: no metadata-version.", self.contents_yaml)
                 raise KeyimportError("metadata-version missing.")
             if ret["metadata-version"] != METADATA_VERSION:
                 logging.error(
                     "%s: metadata-version %s",
                     self.contents_yaml,
                     ret["metadata-version"],
                 )
                 raise KeyimportError("metadata-version bad.")
             return ret
         else:
             return default

     def WriteRepoYaml(self):
         """Write repo.yaml."""
         self._WriteConfig(self.contents_yaml, self.keysets)

     def _WriteConfig(self, filename, config):
         """Write the config

         Args:
             filename: path to file.
             config: keyset dictionary.
         """
         content = yaml.dump(config, default_flow_style=False)
         osutils.WriteFile(filename, content)

     def ImportKeyset(self, setname, directory):
         """Import a Keyset.

         Args:
             setname: Friendly setname (e.g., sarien-mp-v3)
             directory: Directory where keyset is stored. (e.g., SarienMPKeys-v3)

         Returns:
             True if all files were processed.
         """
         config_file = os.path.join(self.configs, "%s.yaml" % setname)
         if os.path.exists(config_file):
             if self.options.new_only:
                 logging.debug(
                     "Skipping existing keyset %s (%s).", setname, directory
                 )
                 return False
             if self.options.update:
                 osutils.SafeUnlink(config_file)
             else:
                 raise KeyimportError("%s.yaml already exists" % setname)

         logging.info("Processing keyset %s (%s):", setname, directory)

         set_config = {
             "name": setname,
             "metadata-version": METADATA_VERSION,
             "directory": directory,
         }

         if setname.startswith("test-keys-"):
             group = "test-keysets"
         elif "-premp" in setname:
             group = "premp-keysets"
         elif "-mp" in setname:
             group = "mp-keysets"
         # These two are special case.
         elif setname in ("update_signer", "oci_containers_signer"):
             group = "mp-keysets"
         else:
             # This will cause the schema validation to fail, notifying the user.
             logging.warning("Bad keyset setname %s", setname)
             group = "bad-keysets"
         self.keysets.setdefault(group, []).append(setname)
         self.keysets[group].sort()
         return self._WriteConfig(config_file, set_config)


 def ParseArgs(argv):
     """Parse the commandline arguments."""
     parser = commandline.ArgumentParser(description=__doc__)
     parser.add_argument("-P", "--prefix", type="path", default="/")
     parser.add_argument("-d", "--directory", default=BASE_DIR)
     parser.add_argument(
         "-D", "--discover", action="store_true", help="Discover keysets"
     )
     parser.add_argument("-s", "-p", "--production", default=PROD_DIR)
     parser.add_argument(
         "-N", "--new-only", action="store_true", help="Only process new keysets"
     )
     parser.add_argument("-u", "--update", action="store_true")
     parser.add_argument("keysets", nargs="*")

     options = parser.parse_args(argv)
     options.Freeze()
     return options


 def ParseSignerConfig(prod_path):
     """Return the parsed signer config.

     Args:
         prod_path: Path to production checkout.  Typically '/cros'.

     Returns:
         Parsed signer config.
     """
     config = configparser.ConfigParser()
     config_path = os.path.join(prod_path, "signer/configs/cros_common.config")
     if not os.path.exists(config_path):
         logging.warning("%s not found, using production config.", config_path)
         config_path = "/cros/signer/configs/cros_common.config"
     config.read(config_path)
     return config


 def DiscoverKeysets(keysets_dir):
     """Discover keysets.

     Args:
         keysets_dir: directory where the keysets live.  Typically /cros/keys.

     Returns:
         A sorted list of (setname: directory) tuples.
     """
     _, dirs, _ = next(os.walk(keysets_dir))
     ret = {}
     for src in sorted(dirs):
         path = os.path.join(keysets_dir, src)
         keyset = keys.Keyset(path)
         if keyset.name != "unknown":
             if keyset.name in ret:
                 logging.warning(
                     "Ignoring %s because name is duplicate of %s",
                     src,
                     ret[keyset.name],
                 )
             else:
                 logging.info("Discovered %s in %s", keyset.name, src)
                 ret[keyset.name] = src
     return sorted(ret.items())


 def main(argv):
     """Import the keys."""
     options = ParseArgs(argv)
     # Always prepend the prefix directory.
     production = os.path.join(options.prefix, options.production.lstrip("/"))
     directory = os.path.join(options.prefix, options.directory.lstrip("/"))
     config = ParseSignerConfig(production)

     keydata = KeyringData(production, directory, options, config)

     if osutils.SafeMakedirs(keydata.public):
         os.chdir(keydata.public)
         osutils.SafeMakedirs(keydata.private)
         osutils.SafeMakedirs(keydata.configs)
     else:
         os.chdir(keydata.public)

     if not options.discover and config.has_section("keysets"):
         config_keysets = config.items("keysets")
     else:
         logging.notice("Discovering keysets in %s" % options.production)
         config_keysets = DiscoverKeysets(
             os.path.join(options.production, "keys")
         )

     # Handle autoupdate keysets first.
     if not options.keysets:
         keysets = config_keysets
     else:
         # Allow '<setname>:<directory>' as well as '<setname>'.
         keysets = []
         keyset_dict = dict(config_keysets)
         for opt in options.keysets:
             if ":" in opt:
                 keysets.append(opt.split(":", 1))
             elif opt in keyset_dict:
                 keysets.append((opt, keyset_dict[opt]))

     # If the auto-update keyset is present, process it first.
     au_keysets = [x for x in keysets if x[1] == AU_KEYDIR]
     for keyset in au_keysets:
         keysets.remove(keyset)
     keysets = au_keysets + keysets
     logging.notice("Processing: %s", " ".join(x[0] for x in keysets))

     for setname, directory in keysets:
         # TODO(lamontjones): consider flattening the namespace by combining
         #  multiple directories/sets into a common directory. That would require
         #  adding version numbers to colliding files.
         keydata.ImportKeyset(setname, directory)

     keydata.WriteRepoYaml()
     logging.notice("Done")
	# Copyright 2019 The ChromiumOS Authors
	# Use of this source code is governed by a BSD-style license that can be
	# found in the LICENSE file.

	"""Import keys from legacy signer config."""

	import configparser
	import logging
	import os

	import yaml # pylint: disable=import-error

	from chromite.lib import commandline
	from chromite.lib import osutils
	from chromite.signing.lib import keys


	METADATA_VERSION = 0
	PROD_DIR = "/cros"
	BASE_DIR = os.path.join(PROD_DIR, "v2-keys")
	AU_KEYDIR = "update-payload-key"


	class Error(Exception):
	"""Base error class."""


	class KeyimportError(Error):
	"""Key Import Errors."""


	class KeyringData:
	"""A collection of keyset information."""

	def __init__(self, prod_dir, base_dir, options, signer_config):
	self.prod_dir = prod_dir
	self.base_dir = base_dir
	self.options = options
	self.signer_config = signer_config
	self.prod_keys = os.path.join(prod_dir, "keys")
	self.public = os.path.join(base_dir, "public")
	self.private = "../private"
	self.configs = "../config"
	self.contents_yaml = os.path.join(self.base_dir, "contents.yaml")
	self.keysets = self._ReadRepoYaml()

	def _ReadRepoYaml(self):
	"""Read the directory of active keysets in the repo."""
	default = {"metadata-version": METADATA_VERSION}
	if os.path.exists(self.contents_yaml):
	with open(self.contents_yaml, encoding="utf-8") as fp:
	ret = yaml.safe_load(fp)
	if not ret or "metadata-version" not in ret:
	logging.error("%s: no metadata-version.", self.contents_yaml)
	raise KeyimportError("metadata-version missing.")
	if ret["metadata-version"] != METADATA_VERSION:
	logging.error(
	"%s: metadata-version %s",
	self.contents_yaml,
	ret["metadata-version"],
	)
	raise KeyimportError("metadata-version bad.")
	return ret
	else:
	return default

	def WriteRepoYaml(self):
	"""Write repo.yaml."""
	self._WriteConfig(self.contents_yaml, self.keysets)

	def _WriteConfig(self, filename, config):
	"""Write the config

	Args:
	filename: path to file.
	config: keyset dictionary.
	"""
	content = yaml.dump(config, default_flow_style=False)
	osutils.WriteFile(filename, content)

	def ImportKeyset(self, setname, directory):
	"""Import a Keyset.

	Args:
	setname: Friendly setname (e.g., sarien-mp-v3)
	directory: Directory where keyset is stored. (e.g., SarienMPKeys-v3)

	Returns:
	True if all files were processed.
	"""
	config_file = os.path.join(self.configs, "%s.yaml" % setname)
	if os.path.exists(config_file):
	if self.options.new_only:
	logging.debug(
	"Skipping existing keyset %s (%s).", setname, directory
	)
	return False
	if self.options.update:
	osutils.SafeUnlink(config_file)
	else:
	raise KeyimportError("%s.yaml already exists" % setname)

	logging.info("Processing keyset %s (%s):", setname, directory)

	set_config = {
	"name": setname,
	"metadata-version": METADATA_VERSION,
	"directory": directory,
	}

	if setname.startswith("test-keys-"):
	group = "test-keysets"
	elif "-premp" in setname:
	group = "premp-keysets"
	elif "-mp" in setname:
	group = "mp-keysets"
	# These two are special case.
	elif setname in ("update_signer", "oci_containers_signer"):
	group = "mp-keysets"
	else:
	# This will cause the schema validation to fail, notifying the user.
	logging.warning("Bad keyset setname %s", setname)
	group = "bad-keysets"
	self.keysets.setdefault(group, []).append(setname)
	self.keysets[group].sort()
	return self._WriteConfig(config_file, set_config)


	def ParseArgs(argv):
	"""Parse the commandline arguments."""
	parser = commandline.ArgumentParser(description=__doc__)
	parser.add_argument("-P", "--prefix", type="path", default="/")
	parser.add_argument("-d", "--directory", default=BASE_DIR)
	parser.add_argument(
	"-D", "--discover", action="store_true", help="Discover keysets"
	)
	parser.add_argument("-s", "-p", "--production", default=PROD_DIR)
	parser.add_argument(
	"-N", "--new-only", action="store_true", help="Only process new keysets"
	)
	parser.add_argument("-u", "--update", action="store_true")
	parser.add_argument("keysets", nargs="*")

	options = parser.parse_args(argv)
	options.Freeze()
	return options


	def ParseSignerConfig(prod_path):
	"""Return the parsed signer config.

	Args:
	prod_path: Path to production checkout. Typically '/cros'.

	Returns:
	Parsed signer config.
	"""
	config = configparser.ConfigParser()
	config_path = os.path.join(prod_path, "signer/configs/cros_common.config")
	if not os.path.exists(config_path):
	logging.warning("%s not found, using production config.", config_path)
	config_path = "/cros/signer/configs/cros_common.config"
	config.read(config_path)
	return config


	def DiscoverKeysets(keysets_dir):
	"""Discover keysets.

	Args:
	keysets_dir: directory where the keysets live. Typically /cros/keys.

	Returns:
	A sorted list of (setname: directory) tuples.
	"""
	_, dirs, _ = next(os.walk(keysets_dir))
	ret = {}
	for src in sorted(dirs):
	path = os.path.join(keysets_dir, src)
	keyset = keys.Keyset(path)
	if keyset.name != "unknown":
	if keyset.name in ret:
	logging.warning(
	"Ignoring %s because name is duplicate of %s",
	src,
	ret[keyset.name],
	)
	else:
	logging.info("Discovered %s in %s", keyset.name, src)
	ret[keyset.name] = src
	return sorted(ret.items())


	def main(argv):
	"""Import the keys."""
	options = ParseArgs(argv)
	# Always prepend the prefix directory.
	production = os.path.join(options.prefix, options.production.lstrip("/"))
	directory = os.path.join(options.prefix, options.directory.lstrip("/"))
	config = ParseSignerConfig(production)

	keydata = KeyringData(production, directory, options, config)

	if osutils.SafeMakedirs(keydata.public):
	os.chdir(keydata.public)
	osutils.SafeMakedirs(keydata.private)
	osutils.SafeMakedirs(keydata.configs)
	else:
	os.chdir(keydata.public)

	if not options.discover and config.has_section("keysets"):
	config_keysets = config.items("keysets")
	else:
	logging.notice("Discovering keysets in %s" % options.production)
	config_keysets = DiscoverKeysets(
	os.path.join(options.production, "keys")
	)

	# Handle autoupdate keysets first.
	if not options.keysets:
	keysets = config_keysets
	else:
	# Allow '<setname>:<directory>' as well as '<setname>'.
	keysets = []
	keyset_dict = dict(config_keysets)
	for opt in options.keysets:
	if ":" in opt:
	keysets.append(opt.split(":", 1))
	elif opt in keyset_dict:
	keysets.append((opt, keyset_dict[opt]))

	# If the auto-update keyset is present, process it first.
	au_keysets = [x for x in keysets if x[1] == AU_KEYDIR]
	for keyset in au_keysets:
	keysets.remove(keyset)
	keysets = au_keysets + keysets
	logging.notice("Processing: %s", " ".join(x[0] for x in keysets))

	for setname, directory in keysets:
	# TODO(lamontjones): consider flattening the namespace by combining
	# multiple directories/sets into a common directory. That would require
	# adding version numbers to colliding files.
	keydata.ImportKeyset(setname, directory)

	keydata.WriteRepoYaml()
	logging.notice("Done")