Overhaul/rewrite of certificate handling as follows: (#1962)

Change "ca-certs" references to "ca_certs".

New certificates are written to individual files, with an incrementing
number as part of their filename, rather than all being placed in a
single file. This resolves issues caused when certificate files
containing more than a single certificate are placed in /etc/ssl/certs
(by utilities such as "update-ca-certificates" run by ca_certs).

Alpine / Debian / Ubuntu:

The current behaviour, whilst it works, is incorrect with regard to
the design of the underlying OS utilities for managing certificates.
For "remove_defaults" the system-installed certificate files should not
be actually deleted (otherwise it becomes problematic if someone wishes
to later re-enable one or more of them), rather they should be
deactivated and these OSes already provide the means to do so - this MR
modifies the certificate entries in the /etc/ca-certificates.conf file
by prefixing them with "!" - when the update-ca-certificate utility is
then run it will *not* place such delimited certificates into either the
/etc/ssl/certs/ directory (via symlinks) nor add them to the
(re)generated certificates bundle file.

Additionally it is incorrect for added certificates to be placed in the
/usr/share/ca-certificates directory - this location is intended for
standard/"official" certificates, the /usr/local/share/ca-certificates
directory is intended for "local" or "site-specific" certificates and so
this PR adds them there instead - for certs in
/usr/local/share/ca-certificates the update-ca-certificates utility will
automatically use them, there is *no* need to add their filenames to the
/etc/ca-certificates.conf file.

LP: #1931174

cloudinit/config/cc_ca_certs.py

@@ -8,45 +8,47 @@
 from logging import Logger
 from textwrap import dedent
 
+from cloudinit import log as logging
 from cloudinit import subp, util
 from cloudinit.cloud import Cloud
 from cloudinit.config import Config
 from cloudinit.config.schema import MetaSchema, get_meta_doc
 from cloudinit.settings import PER_INSTANCE
 
+LOG = logging.getLogger(__name__)
+
 DEFAULT_CONFIG = {
-    "ca_cert_path": "/usr/share/ca-certificates/",
-    "ca_cert_filename": "cloud-init-ca-certs.crt",
+    "ca_cert_path": None,
+    "ca_cert_local_path": "/usr/local/share/ca-certificates/",
+    "ca_cert_filename": "cloud-init-ca-cert-{cert_index}.crt",
     "ca_cert_config": "/etc/ca-certificates.conf",
-    "ca_cert_system_path": "/etc/ssl/certs/",
     "ca_cert_update_cmd": ["update-ca-certificates"],
 }
 DISTRO_OVERRIDES = {
     "rhel": {
-        "ca_cert_path": "/usr/share/pki/ca-trust-source/",
-        "ca_cert_filename": "anchors/cloud-init-ca-certs.crt",
+        "ca_cert_path": "/etc/pki/ca-trust/",
+        "ca_cert_local_path": "/usr/share/pki/ca-trust-source/",
+        "ca_cert_filename": "anchors/cloud-init-ca-cert-{cert_index}.crt",
         "ca_cert_config": None,
-        "ca_cert_system_path": "/etc/pki/ca-trust/",
         "ca_cert_update_cmd": ["update-ca-trust"],
-    }
+    },
 }
 
 MODULE_DESCRIPTION = """\
-This module adds CA certificates to ``/etc/ca-certificates.conf`` and updates
-the ssl cert cache using ``update-ca-certificates``. The default certificates
-can be removed from the system with the configuration option
-``remove_defaults``.
+This module adds CA certificates to the system's CA store and updates any
+related files using the appropriate OS-specific utility. The default CA
+certificates can be disabled/deleted from use by the system with the
+configuration option ``remove_defaults``.
 
 .. note::
     certificates must be specified using valid yaml. in order to specify a
     multiline certificate, the yaml multiline list syntax must be used
 
 .. note::
-    For Alpine Linux the "remove_defaults" functionality works if the
-    ca-certificates package is installed but not if the
-    ca-certificates-bundle package is installed.
+    Alpine Linux requires the ca-certificates package to be installed in
+    order to provide the ``update-ca-certificates`` command.
 """
-distros = ["alpine", "debian", "ubuntu", "rhel"]
+distros = ["alpine", "debian", "rhel", "ubuntu"]
 
 meta: MetaSchema = {
     "id": "cc_ca_certs",
@@ -79,11 +81,11 @@ def _distro_ca_certs_configs(distro_name):
     """Return a distro-specific ca_certs config dictionary
 
     @param distro_name: String providing the distro class name.
-    @returns: Dict of distro configurations for ca-cert.
+    @returns: Dict of distro configurations for ca_cert.
     """
     cfg = DISTRO_OVERRIDES.get(distro_name, DEFAULT_CONFIG)
     cfg["ca_cert_full_path"] = os.path.join(
-        cfg["ca_cert_path"], cfg["ca_cert_filename"]
+        cfg["ca_cert_local_path"], cfg["ca_cert_filename"]
     )
     return cfg
 
@@ -100,124 +102,145 @@ def update_ca_certs(distro_cfg):
 def add_ca_certs(distro_cfg, certs):
     """
     Adds certificates to the system. To actually apply the new certificates
-    you must also call L{update_ca_certs}.
+    you must also call the appropriate distro-specific utility such as
+    L{update_ca_certs}.
 
     @param distro_cfg: A hash providing _distro_ca_certs_configs function.
     @param certs: A list of certificate strings.
     """
     if not certs:
         return
-    # First ensure they are strings...
-    cert_file_contents = "\n".join([str(c) for c in certs])
-    util.write_file(
-        distro_cfg["ca_cert_full_path"], cert_file_contents, mode=0o644
-    )
-    update_cert_config(distro_cfg)
+    # Write each certificate to a separate file.
+    for cert_index, c in enumerate(certs, 1):
+        # First ensure they are strings...
+        cert_file_contents = str(c)
+        cert_file_name = distro_cfg["ca_cert_full_path"].format(
+            cert_index=cert_index
+        )
+        util.write_file(cert_file_name, cert_file_contents, mode=0o644)
 
 
-def update_cert_config(distro_cfg):
+def disable_default_ca_certs(distro_name, distro_cfg):
     """
-    Update Certificate config file to add the file path managed cloud-init
+    Disables all default trusted CA certificates. For Alpine, Debian and
+    Ubuntu to actually apply the changes you must also call
+    L{update_ca_certs}.
+
+    @param distro_name: String providing the distro class name.
+    @param distro_cfg: A hash providing _distro_ca_certs_configs function.
+    """
+    if distro_name == "rhel":
+        remove_default_ca_certs(distro_cfg)
+    elif distro_name in ["alpine", "debian", "ubuntu"]:
+        disable_system_ca_certs(distro_cfg)
+
+        if distro_name in ["debian", "ubuntu"]:
+            debconf_sel = (
+                "ca-certificates ca-certificates/trust_new_crts " + "select no"
+            )
+            subp.subp(("debconf-set-selections", "-"), debconf_sel)
+
+
+def disable_system_ca_certs(distro_cfg):
+    """
+    For every entry in the CA_CERT_CONFIG file prefix the entry with a "!"
+    in order to disable it.
 
     @param distro_cfg: A hash providing _distro_ca_certs_configs function.
     """
     if distro_cfg["ca_cert_config"] is None:
         return
-    if os.stat(distro_cfg["ca_cert_config"]).st_size == 0:
-        # If the CA_CERT_CONFIG file is empty (i.e. all existing
-        # CA certs have been deleted) then simply output a single
-        # line with the cloud-init cert filename.
-        out = "%s\n" % distro_cfg["ca_cert_filename"]
-    else:
-        # Append cert filename to CA_CERT_CONFIG file.
-        # We have to strip the content because blank lines in the file
-        # causes subsequent entries to be ignored. (LP: #1077020)
+    header_comment = (
+        "# Modified by cloud-init to deselect certs due to user-data"
+    )
+    added_header = False
+    if os.stat(distro_cfg["ca_cert_config"]).st_size != 0:
         orig = util.load_file(distro_cfg["ca_cert_config"])
-        cr_cont = "\n".join(
-            [
-                line
-                for line in orig.splitlines()
-                if line != distro_cfg["ca_cert_filename"]
-            ]
-        )
-        out = "%s\n%s\n" % (cr_cont.rstrip(), distro_cfg["ca_cert_filename"])
-    util.write_file(distro_cfg["ca_cert_config"], out, omode="wb")
+        out_lines = []
+        for line in orig.splitlines():
+            if line == header_comment:
+                added_header = True
+                out_lines.append(line)
+            elif line == "" or line[0] in ("#", "!"):
+                out_lines.append(line)
+            else:
+                if not added_header:
+                    out_lines.append(header_comment)
+                    added_header = True
+                out_lines.append("!" + line)
+    util.write_file(
+        distro_cfg["ca_cert_config"], "\n".join(out_lines) + "\n", omode="wb"
+    )
 
 
-def remove_default_ca_certs(distro_name, distro_cfg):
+def remove_default_ca_certs(distro_cfg):
     """
-    Removes all default trusted CA certificates from the system. To actually
-    apply the change you must also call L{update_ca_certs}.
+    Removes all default trusted CA certificates from the system.
 
-    @param distro_name: String providing the distro class name.
     @param distro_cfg: A hash providing _distro_ca_certs_configs function.
     """
-    util.delete_dir_contents(distro_cfg["ca_cert_path"])
-    util.delete_dir_contents(distro_cfg["ca_cert_system_path"])
-    util.write_file(distro_cfg["ca_cert_config"], "", mode=0o644)
+    if distro_cfg["ca_cert_path"] is None:
+        return
 
-    if distro_name in ["debian", "ubuntu"]:
-        debconf_sel = (
-            "ca-certificates ca-certificates/trust_new_crts " + "select no"
-        )
-        subp.subp(("debconf-set-selections", "-"), debconf_sel)
+    LOG.debug("Deleting system CA certificates")
+    util.delete_dir_contents(distro_cfg["ca_cert_path"])
+    util.delete_dir_contents(distro_cfg["ca_cert_local_path"])
 
 
 def handle(
     name: str, cfg: Config, cloud: Cloud, log: Logger, args: list
 ) -> None:
     """
-    Call to handle ca-cert sections in cloud-config file.
+    Call to handle ca_cert sections in cloud-config file.
 
-    @param name: The module name "ca-cert" from cloud.cfg
+    @param name: The module name "ca_cert" from cloud.cfg
     @param cfg: A nested dict containing the entire cloud config contents.
     @param cloud: The L{CloudInit} object in use.
     @param log: Pre-initialized Python logger object to use for logging.
     @param args: Any module arguments from cloud.cfg
     """
     if "ca-certs" in cfg:
-        log.warning(
+        LOG.warning(
             "DEPRECATION: key 'ca-certs' is now deprecated. Use 'ca_certs'"
             " instead."
         )
     elif "ca_certs" not in cfg:
-        log.debug(
+        LOG.debug(
             "Skipping module named %s, no 'ca_certs' key in configuration",
             name,
         )
         return
 
     if "ca-certs" in cfg and "ca_certs" in cfg:
-        log.warning(
+        LOG.warning(
             "Found both ca-certs (deprecated) and ca_certs config keys."
             " Ignoring ca-certs."
         )
     ca_cert_cfg = cfg.get("ca_certs", cfg.get("ca-certs"))
     distro_cfg = _distro_ca_certs_configs(cloud.distro.name)
 
-    # If there is a remove_defaults option set to true, remove the system
+    # If there is a remove_defaults option set to true, disable the system
     # default trusted CA certs first.
     if "remove-defaults" in ca_cert_cfg:
-        log.warning(
+        LOG.warning(
             "DEPRECATION: key 'ca-certs.remove-defaults' is now deprecated."
             " Use 'ca_certs.remove_defaults' instead."
         )
-        if ca_cert_cfg.get("remove-defaults", False):
-            log.debug("Removing default certificates")
-            remove_default_ca_certs(cloud.distro.name, distro_cfg)
-    elif ca_cert_cfg.get("remove_defaults", False):
-        log.debug("Removing default certificates")
-        remove_default_ca_certs(cloud.distro.name, distro_cfg)
+    if ca_cert_cfg.get(
+        "remove_defaults", ca_cert_cfg.get("remove-defaults", False)
+    ):
+        LOG.debug("Disabling/removing default certificates")
+        disable_default_ca_certs(cloud.distro.name, distro_cfg)
 
     # If we are given any new trusted CA certs to add, add them.
     if "trusted" in ca_cert_cfg:
         trusted_certs = util.get_cfg_option_list(ca_cert_cfg, "trusted")
         if trusted_certs:
-            log.debug("Adding %d certificates" % len(trusted_certs))
+            LOG.debug("Adding %d certificates", len(trusted_certs))
             add_ca_certs(distro_cfg, trusted_certs)
 
     # Update the system with the new cert configuration.
-    log.debug("Updating certificates")
+    LOG.debug("Updating certificates")
     update_ca_certs(distro_cfg)
 
 

doc/examples/cloud-config-ca-certs.txt

@@ -8,11 +8,12 @@
 # It should be passed as user-data when starting the instance.
 
 ca_certs:
-  # If present and set to True, the 'remove_defaults' parameter will remove
-  # all the default trusted CA certificates that are normally shipped with
-  # Ubuntu.
-  # This is mainly for paranoid admins - most users will not need this
-  # functionality.
+  # If present and set to True, the 'remove_defaults' parameter will either
+  # disable all the trusted CA certifications normally shipped with
+  # Alpine, Debian or Ubuntu. On RedHat, this action will delete those
+  # certificates.
+  # This is mainly for very security-sensitive use cases - most users will not
+  # need this functionality.
   remove_defaults: true
 
   # If present, the 'trusted' parameter should contain a certificate (or list

tests/integration_tests/modules/test_ca_certs.py

@@ -76,10 +76,10 @@ def test_certs_updated(self, class_client: IntegrationInstance):
                 unlinked_files.append(filename)
 
         assert ["ca-certificates.crt"] == unlinked_files
-        assert "cloud-init-ca-certs.pem" == links["a535c1f3.0"]
+        assert "cloud-init-ca-cert-1.pem" == links["a535c1f3.0"]
         assert (
-            "/usr/share/ca-certificates/cloud-init-ca-certs.crt"
-            == links["cloud-init-ca-certs.pem"]
+            "/usr/local/share/ca-certificates/cloud-init-ca-cert-1.crt"
+            == links["cloud-init-ca-cert-1.pem"]
         )
 
     def test_cert_installed(self, class_client: IntegrationInstance):