Debian: write out only one CA certificate per file

ca-certificates populates /etc/ssl/certs with symlinks to its input
files and then runs 'openssl rehash' to create the symlinks that libssl
uses to look up a CA certificate to see if it is trused.

'openssl rehash' ignores any files that contain more than one
certificate: <https://bugs.debian.org/cgi-bin/bugreport.cgi?bug=945274>.

With this change, we write out trusted CA certificates to
/usr/local/share/ca-certificates/ipa-ca, one certificate per file.

The logic that decides whether to reload the store is moved up into the
original `insert_ca_certs_into_systemwide_ca_store` and
`remove_ca_certs_from_systemwide_ca_store` methods. These methods now
also handle any exceptions that may be thrown while updating the store.

The functions that actually manipulate the store are factored out into
new `platform_{insert,remove}_ca_certs` methods, which implementations
must override.

These new methods also orchestrate the cleanup of deprecated files (such
as `/etc/pki/ca-trust/source/anchors/ipa-ca.crt`), rather than having
the cleanup code be included in the same method that creates
`/etc/pki/ca-trust/source/ipa.p11-kit`.

As well as creating `/usr/local/share/ca-certificates/ipa-ca`, Debian
systems will now also have
`/usr/local/share/ca-certificates/ipa.p11-kit` be created. Note that
`p11-kit` in Debian does not use this file.

Fixes: https://pagure.io/freeipa/issue/8106
Reviewed-By: Alexander Bokovoy <abokovoy@redhat.com>
Reviewed-By: Timo Aaltonen <tjaalton@debian.org>
Reviewed-By: Rob Crittenden <rcritten@redhat.com>

ipaplatform/base/paths.py

@@ -101,8 +101,12 @@ class BasePathNamespace:
     OPENLDAP_LDAP_CONF = "/etc/openldap/ldap.conf"
     PAM_LDAP_CONF = "/etc/pam_ldap.conf"
     PASSWD = "/etc/passwd"
+    # Trusted CA certificates used to be written out to this file. In newer
+    # versions of FreeIPA, it has been replaced by IPA_P11_KIT.
     SYSTEMWIDE_IPA_CA_CRT = "/etc/pki/ca-trust/source/anchors/ipa-ca.crt"
     IPA_P11_KIT = "/etc/pki/ca-trust/source/ipa.p11-kit"
+    CA_CERTIFICATES_BUNDLE_PEM = None
+    CA_CERTIFICATES_DIR = None
     NSS_DB_DIR = "/etc/pki/nssdb"
     PKI_TOMCAT = "/etc/pki/pki-tomcat"
     PKI_TOMCAT_ALIAS_DIR = "/etc/pki/pki-tomcat/alias"

ipaplatform/base/tasks.py

@@ -71,6 +71,23 @@ class BaseTaskNamespace:
         Returns True if the operation succeeded, False otherwise.
         """
 
+        try:
+            if self.platform_insert_ca_certs(ca_certs):
+                return self.reload_systemwide_ca_store()
+        except Exception:
+            logger.exception('Could not populate systemwide CA store')
+
+        return False
+
+    def platform_insert_ca_certs(self, ca_certs):
+        """
+        Platform implementations override this method to implement
+        population of the systemwide CA store.
+
+        Returns True if changes were made to the CA store, False otherwise.
+
+        Raises Exception if something went wrong.
+        """
         raise NotImplementedError()
 
     def remove_ca_certs_from_systemwide_ca_store(self):
@@ -81,6 +98,25 @@ class BaseTaskNamespace:
         Returns True if the operation succeeded, False otherwise.
         """
 
+        try:
+            if self.platform_remove_ca_certs():
+                return self.reload_systemwide_ca_store()
+        except Exception:
+            logger.exception(
+                'Could not remove certificates from systemwide CA store'
+            )
+
+        return False
+
+    def platform_remove_ca_certs(self):
+        """
+        Platform implementations override this method to implement
+        removal of certificates from the systemwide CA store.
+
+        Returns True if changes were made to the CA store, False otherwise.
+
+        Raises Exception if something went wrong.
+        """
         raise NotImplementedError()
 
     def get_svc_list_file(self):

ipaplatform/debian/paths.py

@@ -43,7 +43,13 @@ class DebianPathNamespace(BasePathNamespace):
     CHRONY_CONF = "/etc/chrony/chrony.conf"
     OPENLDAP_LDAP_CONF = "/etc/ldap/ldap.conf"
     ETC_DEBIAN_VERSION = "/etc/debian_version"
-    IPA_P11_KIT = "/usr/local/share/ca-certificates/ipa-ca.crt"
+    # Old versions of freeipa wrote all trusted certificates to a single
+    # file, which is not supported by ca-certificates.
+    CA_CERTIFICATES_BUNDLE_PEM = "/usr/local/share/ca-certificates/ipa-ca.crt"
+    CA_CERTIFICATES_DIR = "/usr/local/share/ca-certificates/ipa-ca"
+    # Debian's p11-kit does not use ipa.p11-kit, so the file is provided
+    # for information only.
+    IPA_P11_KIT = "/usr/local/share/ca-certificates/ipa.p11-kit"
     ETC_SYSCONFIG_DIR = "/etc/default"
     SYSCONFIG_AUTOFS = "/etc/default/autofs"
     SYSCONFIG_DIRSRV = "/etc/default/dirsrv"

ipaplatform/debian/tasks.py

@@ -8,12 +8,21 @@ This module contains default Debian-specific implementations of system tasks.
 
 from __future__ import absolute_import
 
+import logging
+import os
+import shutil
+from pathlib import Path
+
 from ipaplatform.base.tasks import BaseTaskNamespace
 from ipaplatform.redhat.tasks import RedHatTaskNamespace
 from ipaplatform.paths import paths
 
 from ipapython import directivesetter
 from ipapython import ipautil
+from ipapython.dn import DN
+
+logger = logging.getLogger(__name__)
+
 
 class DebianTaskNamespace(RedHatTaskNamespace):
     @staticmethod
@@ -88,4 +97,113 @@ class DebianTaskNamespace(RedHatTaskNamespace):
     def restore_pkcs11_modules(self, fstore):
         pass
 
+    def platform_insert_ca_certs(self, ca_certs):
+        # ca-certificates does not use this file, so it doesn't matter if we
+        # fail to create it.
+        try:
+            self.write_p11kit_certs(paths.IPA_P11_KIT, ca_certs),
+        except Exception:
+            logger.exception("""\
+Could not create p11-kit anchor trust file. On Debian this file is not
+used by ca-certificates and is provided for information only.\
+""")
+
+        return any([
+            self.write_ca_certificates_dir(
+                paths.CA_CERTIFICATES_DIR, ca_certs
+            ),
+            self.remove_ca_certificates_bundle(
+                paths.CA_CERTIFICATES_BUNDLE_PEM
+            ),
+        ])
+
+    def write_ca_certificates_dir(self, directory, ca_certs):
+        # pylint: disable=ipa-forbidden-import
+        from ipalib import x509  # FixMe: break import cycle
+        # pylint: enable=ipa-forbidden-import
+
+        path = Path(directory)
+        try:
+            path.mkdir(mode=0o755, exist_ok=True)
+        except Exception:
+            logger.error("Could not create %s", path)
+            raise
+
+        for cert, nickname, trusted, _ext_key_usage in ca_certs:
+            if not trusted:
+                continue
+
+            # I'm not handling errors here because they have already
+            # been checked by the time we get here
+            subject = DN(cert.subject)
+            issuer = DN(cert.issuer)
+
+            # Construct the certificate filename using the Subject DN so that
+            # the user can see which CA a particular file is for, and include
+            # the serial number to disambiguate clashes where a subordinate CA
+            # had a new certificate issued.
+            #
+            # Strictly speaking, certificates are uniquely idenified by (Issuer
+            # DN, Serial Number). Do we care about the possibility of a clash
+            # where a subordinate CA had two certificates issued by different
+            # CAs who used the same serial number?)
+            filename = f'{subject.ldap_text()} {cert.serial_number}.crt'
+
+            # pylint: disable=old-division
+            cert_path = path / filename
+            # pylint: enable=old-division
+            try:
+                f = open(cert_path, 'w')
+            except Exception:
+                logger.error("Could not create %s", cert_path)
+                raise
+
+            with f:
+                try:
+                    os.fchmod(f.fileno(), 0o644)
+                except Exception:
+                    logger.error("Could not set mode of %s", cert_path)
+                    raise
+
+                try:
+                    f.write(f"""\
+This file was created by IPA. Do not edit.
+
+Description: {nickname}
+Subject: {subject.ldap_text()}
+Issuer: {issuer.ldap_text()}
+Serial Number (dec): {cert.serial_number}
+Serial Number (hex): {cert.serial_number:#x}
+
+""")
+                    pem = cert.public_bytes(x509.Encoding.PEM).decode('ascii')
+                    f.write(pem)
+                except Exception:
+                    logger.error("Could not write to %s", cert_path)
+                    raise
+
+        return True
+
+    def platform_remove_ca_certs(self):
+        return any([
+            self.remove_ca_certificates_dir(paths.CA_CERTIFICATES_DIR),
+            self.remove_ca_certificates_bundle(paths.IPA_P11_KIT),
+            self.remove_ca_certificates_bundle(
+                paths.CA_CERTIFICATES_BUNDLE_PEM
+            ),
+        ])
+
+    def remove_ca_certificates_dir(self, directory):
+        path = Path(paths.CA_CERTIFICATES_DIR)
+        if not path.exists():
+            return False
+
+        try:
+            shutil.rmtree(path)
+        except Exception:
+            logger.error("Could not remove %s", path)
+            raise
+
+        return True
+
 tasks = DebianTaskNamespace()

ipaplatform/redhat/tasks.py

@@ -28,6 +28,7 @@ from __future__ import print_function, absolute_import
 import ctypes
 import logging
 import os
+from pathlib import Path
 import socket
 import traceback
 import errno
@@ -296,127 +297,120 @@ class RedHatTaskNamespace(BaseTaskNamespace):
             logger.info("Systemwide CA database updated.")
             return True
 
-    def insert_ca_certs_into_systemwide_ca_store(self, ca_certs):
+    def platform_insert_ca_certs(self, ca_certs):
+        return any([
+            self.write_p11kit_certs(paths.IPA_P11_KIT, ca_certs),
+            self.remove_ca_certificates_bundle(
+                paths.SYSTEMWIDE_IPA_CA_CRT
+            ),
+        ])
+
+    def write_p11kit_certs(self, filename, ca_certs):
         # pylint: disable=ipa-forbidden-import
         from ipalib import x509  # FixMe: break import cycle
         from ipalib.errors import CertificateError
         # pylint: enable=ipa-forbidden-import
 
-        new_cacert_path = paths.SYSTEMWIDE_IPA_CA_CRT
+        path = Path(filename)
-
-        if os.path.exists(new_cacert_path):
-            try:
-                os.remove(new_cacert_path)
-            except OSError as e:
-                logger.error(
-                    "Could not remove %s: %s", new_cacert_path, e)
-                return False
-
-        new_cacert_path = paths.IPA_P11_KIT
-
         try:
-            f = open(new_cacert_path, 'w')
+            f = open(path, 'w')
-            os.fchmod(f.fileno(), 0o644)
+        except IOError:
-        except IOError as e:
+            logger.error("Failed to open %s", path)
-            logger.info("Failed to open %s: %s", new_cacert_path, e)
+            raise
-            return False
 
-        f.write("# This file was created by IPA. Do not edit.\n"
+        with f:
-                "\n")
+            f.write("# This file was created by IPA. Do not edit.\n"
+                    "\n")
 
-        has_eku = set()
-        for cert, nickname, trusted, _ext_key_usage in ca_certs:
             try:
-                subject = cert.subject_bytes
+                os.fchmod(f.fileno(), 0o644)
-                issuer = cert.issuer_bytes
+            except IOError:
-                serial_number = cert.serial_number_bytes
+                logger.error("Failed to set mode of %s", path)
-                public_key_info = cert.public_key_info_bytes
+                raise
-            except (PyAsn1Error, ValueError, CertificateError) as e:
-                logger.warning(
-                    "Failed to decode certificate \"%s\": %s", nickname, e)
-                continue
 
-            label = urllib.parse.quote(nickname)
+            has_eku = set()
-            subject = urllib.parse.quote(subject)
+            for cert, nickname, trusted, _ext_key_usage in ca_certs:
-            issuer = urllib.parse.quote(issuer)
-            serial_number = urllib.parse.quote(serial_number)
-            public_key_info = urllib.parse.quote(public_key_info)
-
-            obj = ("[p11-kit-object-v1]\n"
-                   "class: certificate\n"
-                   "certificate-type: x-509\n"
-                   "certificate-category: authority\n"
-                   "label: \"%(label)s\"\n"
-                   "subject: \"%(subject)s\"\n"
-                   "issuer: \"%(issuer)s\"\n"
-                   "serial-number: \"%(serial_number)s\"\n"
-                   "x-public-key-info: \"%(public_key_info)s\"\n" %
-                   dict(label=label,
-                        subject=subject,
-                        issuer=issuer,
-                        serial_number=serial_number,
-                        public_key_info=public_key_info))
-            if trusted is True:
-                obj += "trusted: true\n"
-            elif trusted is False:
-                obj += "x-distrusted: true\n"
-            obj += "{pem}\n\n".format(
-                pem=cert.public_bytes(x509.Encoding.PEM).decode('ascii'))
-            f.write(obj)
-
-            if (cert.extended_key_usage is not None and
-                    public_key_info not in has_eku):
                 try:
-                    ext_key_usage = cert.extended_key_usage_bytes
+                    subject = cert.subject_bytes
-                except PyAsn1Error as e:
+                    issuer = cert.issuer_bytes
-                    logger.warning(
+                    serial_number = cert.serial_number_bytes
-                        "Failed to encode extended key usage for \"%s\": %s",
+                    public_key_info = cert.public_key_info_bytes
-                        nickname, e)
+                except (PyAsn1Error, ValueError, CertificateError):
-                    continue
+                    logger.error(
-                value = urllib.parse.quote(ext_key_usage)
+                        "Failed to decode certificate \"%s\"", nickname)
+                    raise
+
+                label = urllib.parse.quote(nickname)
+                subject = urllib.parse.quote(subject)
+                issuer = urllib.parse.quote(issuer)
+                serial_number = urllib.parse.quote(serial_number)
+                public_key_info = urllib.parse.quote(public_key_info)
+
                 obj = ("[p11-kit-object-v1]\n"
-                       "class: x-certificate-extension\n"
+                       "class: certificate\n"
-                       "label: \"ExtendedKeyUsage for %(label)s\"\n"
+                       "certificate-type: x-509\n"
-                       "x-public-key-info: \"%(public_key_info)s\"\n"
+                       "certificate-category: authority\n"
-                       "object-id: 2.5.29.37\n"
+                       "label: \"%(label)s\"\n"
-                       "value: \"%(value)s\"\n\n" %
+                       "subject: \"%(subject)s\"\n"
+                       "issuer: \"%(issuer)s\"\n"
+                       "serial-number: \"%(serial_number)s\"\n"
+                       "x-public-key-info: \"%(public_key_info)s\"\n" %
                        dict(label=label,
-                            public_key_info=public_key_info,
+                            subject=subject,
-                            value=value))
+                            issuer=issuer,
+                            serial_number=serial_number,
+                            public_key_info=public_key_info))
+                if trusted is True:
+                    obj += "trusted: true\n"
+                elif trusted is False:
+                    obj += "x-distrusted: true\n"
+                obj += "{pem}\n\n".format(
+                    pem=cert.public_bytes(x509.Encoding.PEM).decode('ascii'))
+
                 f.write(obj)
-                has_eku.add(public_key_info)
 
-        f.close()
+                if (cert.extended_key_usage is not None and
-
+                        public_key_info not in has_eku):
-        # Add the CA to the systemwide CA trust database
+                    try:
-        if not self.reload_systemwide_ca_store():
+                        ext_key_usage = cert.extended_key_usage_bytes
-            return False
+                    except PyAsn1Error:
+                        logger.error(
+                            "Failed to encode extended key usage for \"%s\"",
+                            nickname)
+                        raise
+                    value = urllib.parse.quote(ext_key_usage)
+                    obj = ("[p11-kit-object-v1]\n"
+                           "class: x-certificate-extension\n"
+                           "label: \"ExtendedKeyUsage for %(label)s\"\n"
+                           "x-public-key-info: \"%(public_key_info)s\"\n"
+                           "object-id: 2.5.29.37\n"
+                           "value: \"%(value)s\"\n\n" %
+                           dict(label=label,
+                                public_key_info=public_key_info,
+                                value=value))
+                    f.write(obj)
+                    has_eku.add(public_key_info)
 
         return True
 
-    def remove_ca_certs_from_systemwide_ca_store(self):
+    def platform_remove_ca_certs(self):
-        result = True
+        return any([
-        update = False
+            self.remove_ca_certificates_bundle(paths.IPA_P11_KIT),
+            self.remove_ca_certificates_bundle(paths.SYSTEMWIDE_IPA_CA_CRT),
+        ])
 
-        # Remove CA cert from systemwide store
+    def remove_ca_certificates_bundle(self, filename):
-        for new_cacert_path in (paths.IPA_P11_KIT,
+        path = Path(filename)
-                                paths.SYSTEMWIDE_IPA_CA_CRT):
+        if not path.is_file():
-            if not os.path.exists(new_cacert_path):
+            return False
-                continue
-            try:
-                os.remove(new_cacert_path)
-            except OSError as e:
-                logger.error(
-                    "Could not remove %s: %s", new_cacert_path, e)
-                result = False
-            else:
-                update = True
 
-        if update:
+        try:
-            if not self.reload_systemwide_ca_store():
+            path.unlink()
-                return False
+        except Exception:
+            logger.error("Could not remove %s", path)
+            raise
 
-        return result
+        return True
 
     def backup_hostname(self, fstore, statestore):
         filepath = paths.ETC_HOSTNAME