Don't require certificates to have unique ipaCertSubject

In the wild a public CA issued a new subordinate CA certificate
with an identical subject to another, with a new private key.
This was uninstallable using ipa-cacert-manage because it would
fail with "subject public key info mismatch" during verification
because a different certificate with the same subject but
different public key was installed.

I'm not sure of the reasoning to prevent this situation but I
see it as giving users flexibility. This may be hurtful to them
but they can always remove any affected certs.

This is backwards compatible with older releases from the client
perspective. Older servers will choke on the duplicates and
won't be able to manage these.

A new serial number option is added for displaying the list of
certificates and for use when deleting one with a duplicate subject.

ipa-cacert-manage delete on systems without this patch will
successfully remove ALL of the requested certificates. There is no
way to distinguish. At least it won't break anything and the
deleted certificates can be re-added.

Fixes: https://pagure.io/freeipa/issue/9652

Signed-off-by: Rob Crittenden <rcritten@redhat.com>
Reviewed-By: Alexander Bokovoy <abokovoy@redhat.com>
Reviewed-By: Florence Blanc-Renaud <frenaud@redhat.com>

install/restart_scripts/renew_ca_cert.in

@@ -168,7 +168,7 @@ def _main():
                     ca_certs = []
 
                 realm_nickname = get_ca_nickname(api.env.realm)
-                for ca_cert, ca_nick, ca_flags in ca_certs:
+                for ca_cert, ca_nick, ca_flags, _serial in ca_certs:
                     try:
                         if ca_nick == realm_nickname:
                             ca_nick = 'caSigningCert cert-pki-ca'
@@ -180,7 +180,7 @@ def _main():
 
                 # Pass Dogtag's self-tests
                 for ca_nick in db.find_root_cert(nickname)[-2:-1]:
-                    ca_flags = dict(cc[1:] for cc in ca_certs)[ca_nick]
+                    ca_flags = dict(cc[1:3] for cc in ca_certs)[ca_nick]
                     usages = ca_flags.usages or set()
                     ca_flags_modified = TrustFlags(ca_flags.has_key,
                         True, True,

install/tools/man/ipa-cacert-manage.1

@@ -57,6 +57,14 @@ Important: this does not replace IPA CA but adds the provided certificate as a k
 Please do not forget to run ipa-certupdate on the master, all the replicas and all the clients after this command in order to update IPA certificates databases.
 .sp
 The supported formats for the certificate files are DER, PEM and PKCS#7 format.
+.sp
+CA certificates with the same subject but different private keys maybe installed simultaneously with the following restrictions from NSS:
+.IP \[bu]
+The certificates cannot have different NSS trust flags.
+.IP \[bu]
+The nickname is not configurable between different certificates of the same subject. It will always be the same (even if you try).
+.sp
+Additionally CA certificates with the same subject should include the Authority Key Identifier extension in order to identify the public key of the certificate issuer (CA) that signed the certificate (it may be itself). Similarly it should have a Subject Key Identifier extension. This is used to create the trust chain not through subjects but by using the SKID and AKID which is what allows duplicate certificate subjects to be resolved correctly. Without an AKID multiple certificates of the same subject will not resolve as expected.
 .RE
 .TP
 \fBdelete\fR
@@ -153,6 +161,9 @@ p \- not trusted
 .TP
 \fB\-f\fR, \fB\-\-force\fR
 Force a CA certificate to be removed even if chain validation fails.
+.TP
+\fB\-s\fR \fISERIAL_NUMBER\fR, \fB\-\-serial\fR=\fISERIAL_NUMBER\fR
+Serial number of the certificate to delete (decimal). This is needed to determine which certificate to remove if there are multiple certificates stored with the same name.
 .SH "EXIT STATUS"
 0 if the command was successful
 

install/updates/10-uniqueness.update

@@ -15,23 +15,6 @@ default:nsslapd-pluginId: NSUniqueAttr
 default:nsslapd-pluginVersion: 1.1.0
 default:nsslapd-pluginVendor: Fedora Project
 
-dn: cn=certificate store subject uniqueness,cn=plugins,cn=config
-default:objectClass: top
-default:objectClass: nsSlapdPlugin
-default:objectClass: extensibleObject
-default:cn: certificate store subject uniqueness
-default:nsslapd-pluginDescription: Enforce unique attribute values
-default:nsslapd-pluginPath: libattr-unique-plugin
-default:nsslapd-pluginInitfunc: NSUniqueAttr_Init
-default:nsslapd-pluginType: preoperation
-default:nsslapd-pluginEnabled: on
-default:uniqueness-attribute-name: ipaCertSubject
-default:uniqueness-subtrees: cn=certificates,cn=ipa,cn=etc,$SUFFIX
-default:nsslapd-plugin-depends-on-type: database
-default:nsslapd-pluginId: NSUniqueAttr
-default:nsslapd-pluginVersion: 1.1.0
-default:nsslapd-pluginVendor: Fedora Project
-
 dn: cn=certificate store issuer/serial uniqueness,cn=plugins,cn=config
 default:objectClass: top
 default:objectClass: nsSlapdPlugin
@@ -128,3 +111,7 @@ default:nsslapd-plugin-depends-on-type: database
 default:nsslapd-pluginId: NSUniqueAttr
 default:nsslapd-pluginVersion: 1.1.0
 default:nsslapd-pluginVendor: Fedora Project
+
+# A unique ipaCertSubject is no longer required
+dn: cn=certificate store subject uniqueness,cn=plugins,cn=config
+deleteentry: cn=certificate store subject uniqueness,cn=plugins,cn=config

ipaclient/install/client.py

@@ -3200,15 +3200,15 @@ def _install(options, tdict):
         ca_certs = certstore.make_compat_ca_certs(ca_certs, cli_realm,
                                                   ca_subject)
     ca_certs_trust = [(c, n, certstore.key_policy_to_trust_flags(t, True, u))
-                      for (c, n, t, u) in ca_certs]
+                      for (c, n, t, u, s) in ca_certs]
 
     x509.write_certificate_list(
-        [c for c, n, t, u in ca_certs if t is not False],
+        [c for c, n, t, u, s in ca_certs if t is not False],
         paths.KDC_CA_BUNDLE_PEM,
         mode=0o644
     )
     x509.write_certificate_list(
-        [c for c, n, t, u in ca_certs if t is not False],
+        [c for c, n, t, u, s in ca_certs if t is not False],
         paths.CA_BUNDLE_PEM,
         mode=0o644
     )

ipaclient/install/ipa_certupdate.py

@@ -276,7 +276,7 @@ def update_db(path, certs):
     for name, flags in db.list_certs():
         if flags.ca:
             db.delete_cert(name)
-    for cert, nickname, trusted, eku in certs:
+    for cert, nickname, trusted, eku, _serial in certs:
         trust_flags = certstore.key_policy_to_trust_flags(trusted, True, eku)
         try:
             db.add_cert(cert, nickname, trust_flags)

ipalib/install/certstore.py

@@ -179,10 +179,9 @@ def update_ca_cert(ldap, base_dn, cert, trusted=None, ext_key_usage=None,
         # We are adding a new cert, validate it
         if entry.single_value['ipaCertSubject'].lower() != subject.lower():
             raise ValueError("subject name mismatch")
-        if entry.single_value['ipaPublicKey'] != public_key:
-            raise ValueError("subject public key info mismatch")
         entry['ipaCertIssuerSerial'].append(issuer_serial)
         entry['cACertificate;binary'].append(cert)
+        entry['ipaPublicKey'].append(public_key)
 
     # Update key trust
     if trusted is not None:
@@ -224,6 +223,38 @@ def update_ca_cert(ldap, base_dn, cert, trusted=None, ext_key_usage=None,
     clean_old_config(ldap, base_dn, dn, config_ipa, config_compat)
 
 
+def delete_ca_cert(ldap, base_dn, cert):
+    """
+    Remove a CA certificate in the certificate store.
+    """
+    subject, issuer_serial, _public_key = _parse_cert(cert)
+
+    filter = ldap.make_filter({'ipaCertSubject': subject})
+    result, _truncated = ldap.find_entries(
+        base_dn=DN(('cn', 'certificates'), ('cn', 'ipa'), ('cn', 'etc'),
+                   base_dn),
+        filter=filter,
+        attrs_list=['cn', 'ipaCertSubject', 'ipaCertIssuerSerial',
+                    'ipaPublicKey', 'ipaKeyTrust', 'ipaKeyExtUsage',
+                    'ipaConfigString', 'cACertificate;binary'])
+    entry = result[0]
+
+    for old_cert in entry['cACertificate;binary']:
+        # Check if we are adding a new cert
+        if old_cert == cert:
+            break
+    else:
+        raise ValueError("certificate not found")
+
+    entry['ipaCertIssuerSerial'].remove(issuer_serial)
+    entry['cACertificate;binary'].remove(cert)
+
+    if len(entry['ipaCertIssuerSerial']) == 0:
+        ldap.delete_entry(entry.dn)
+    else:
+        ldap.update_entry(entry)
+
+
 def put_ca_cert(ldap, base_dn, cert, nickname, trusted=None,
                 ext_key_usage=None, config_ipa=False, config_compat=False):
     """
@@ -309,11 +340,14 @@ def get_ca_certs(ldap, base_dn, compat_realm, compat_ipa_ca,
 
             for cert in entry.get('cACertificate;binary', []):
                 try:
-                    _parse_cert(cert)
+                    _subject, issuer_serial, _pkinfo = _parse_cert(cert)
                 except ValueError:
                     certs = []
                     break
-                certs.append((cert, nickname, trusted, ext_key_usage))
+                serial_number = issuer_serial.split(';')[1]
+                certs.append(
+                    (cert, nickname, trusted, ext_key_usage, serial_number)
+                )
     except errors.NotFound:
         try:
             ldap.get_entry(container_dn, [''])
@@ -381,9 +415,9 @@ def get_ca_certs_nss(ldap, base_dn, compat_realm, compat_ipa_ca,
 
     certs = get_ca_certs(ldap, base_dn, compat_realm, compat_ipa_ca,
                          filter_subject=filter_subject)
-    for cert, nickname, trusted, ext_key_usage in certs:
+    for cert, nickname, trusted, ext_key_usage, _serial_number in certs:
         trust_flags = key_policy_to_trust_flags(trusted, True, ext_key_usage)
-        nss_certs.append((cert, nickname, trust_flags))
+        nss_certs.append((cert, nickname, trust_flags, _serial_number))
 
     return nss_certs
 

ipaplatform/debian/tasks.py

@@ -126,7 +126,7 @@ used by ca-certificates and is provided for information only.\
             logger.error("Could not create %s", path)
             raise
 
-        for cert, nickname, trusted, _ext_key_usage in ca_certs:
+        for cert, nickname, trusted, _ext_key_usage, _serial in ca_certs:
             if not trusted:
                 continue
 

ipaplatform/redhat/tasks.py

@@ -329,7 +329,7 @@ class RedHatTaskNamespace(BaseTaskNamespace):
                 raise
 
             has_eku = set()
-            for cert, nickname, trusted, _ext_key_usage in ca_certs:
+            for cert, nickname, trusted, _ext_key_usage, _serial in ca_certs:
                 try:
                     subject = cert.subject_bytes
                     issuer = cert.issuer_bytes

ipapython/certdb.py

@@ -633,7 +633,7 @@ class NSSDatabase:
                 pkcs12_password_file.close()
 
     def import_files(self, files, import_keys=False, key_password=None,
-                     key_nickname=None):
+                     key_nickname=None, trust_flags=EMPTY_TRUST_FLAGS):
         """
         Import certificates and a single private key from multiple files
 
@@ -809,7 +809,7 @@ class NSSDatabase:
 
         for cert in extracted_certs:
             nickname = str(DN(cert.subject))
-            self.add_cert(cert, nickname, EMPTY_TRUST_FLAGS)
+            self.add_cert(cert, nickname, trust_flags)
 
         if extracted_key:
             with tempfile.NamedTemporaryFile() as in_file, \
@@ -867,6 +867,27 @@ class NSSDatabase:
         cert, _start = find_cert_from_txt(result.output, start=0)
         return cert
 
+    def get_all_certs(self, nickname):
+        """
+        :param nickname: nickname of the certificate in the NSS database
+        :returns: list of bytes of all certificates for the nickname
+        """
+        args = ['-L', '-n', nickname, '-a']
+        try:
+            result = self.run_certutil(args, capture_output=True)
+        except ipautil.CalledProcessError:
+            raise RuntimeError("Failed to get %s" % nickname)
+        certs = []
+
+        st = 0
+        while True:
+            try:
+                cert, st = find_cert_from_txt(result.output, start=st)
+            except RuntimeError:
+                break
+            certs.append(cert)
+        return certs
+
     def has_nickname(self, nickname):
         try:
             self.get_cert(nickname)
@@ -990,53 +1011,58 @@ class NSSDatabase:
             raise ValueError('invalid for server %s' % hostname)
 
     def verify_ca_cert_validity(self, nickname, minpathlen=None):
-        cert = self.get_cert(nickname)
-        self._verify_cert_validity(cert)
+        def verify_ca_cert(cert, nickname, minpathlen):
+            self._verify_cert_validity(cert)
 
-        if not cert.subject:
-            raise ValueError("has empty subject")
+            if not cert.subject:
+                raise ValueError("has empty subject")
 
-        try:
-            bc = cert.extensions.get_extension_for_class(
+            try:
+                bc = cert.extensions.get_extension_for_class(
                     cryptography.x509.BasicConstraints)
-        except cryptography.x509.ExtensionNotFound:
-            raise ValueError("missing basic constraints")
+            except cryptography.x509.ExtensionNotFound:
+                raise ValueError("missing basic constraints")
 
-        if not bc.value.ca:
-            raise ValueError("not a CA certificate")
-        if minpathlen is not None:
-            # path_length is None means no limitation
-            pl = bc.value.path_length
-            if pl is not None and pl < minpathlen:
-                raise ValueError(
-                    "basic contraint pathlen {}, must be at least {}".format(
-                        pl, minpathlen
+            if not bc.value.ca:
+                raise ValueError("not a CA certificate")
+            if minpathlen is not None:
+                # path_length is None means no limitation
+                pl = bc.value.path_length
+                if pl is not None and pl < minpathlen:
+                    raise ValueError(
+                        "basic contraint pathlen {}, "
+                        "must be at least {}".format(
+                            pl, minpathlen
+                        )
                     )
-                )
 
-        try:
-            ski = cert.extensions.get_extension_for_class(
+            try:
+                ski = cert.extensions.get_extension_for_class(
                     cryptography.x509.SubjectKeyIdentifier)
-        except cryptography.x509.ExtensionNotFound:
-            raise ValueError("missing subject key identifier extension")
-        else:
-            if len(ski.value.digest) == 0:
-                raise ValueError("subject key identifier must not be empty")
+            except cryptography.x509.ExtensionNotFound:
+                raise ValueError("missing subject key identifier extension")
+            else:
+                if len(ski.value.digest) == 0:
+                    raise ValueError("subject key identifier must not be empty")
 
-        try:
-            self.run_certutil(
-                [
-                    '-V',       # check validity of cert and attrs
-                    '-n', nickname,
-                    '-u', 'L',  # usage; 'L' means "SSL CA"
-                    '-e',       # check signature(s); this checks
-                                # key sizes, sig algorithm, etc.
-                ],
-                capture_output=True)
-        except ipautil.CalledProcessError as e:
-            # certutil output in case of error is
-            # 'certutil: certificate is invalid: <ERROR_STRING>\n'
-            raise ValueError(e.output)
+            try:
+                self.run_certutil(
+                    [
+                        '-V',       # check validity of cert and attrs
+                        '-n', nickname,
+                        '-u', 'L',  # usage; 'L' means "SSL CA"
+                        '-e',       # check signature(s); this checks
+                                    # key sizes, sig algorithm, etc.
+                    ],
+                    capture_output=True)
+            except ipautil.CalledProcessError as e:
+                # certutil output in case of error is
+                # 'certutil: certificate is invalid: <ERROR_STRING>\n'
+                raise ValueError(e.output)
+
+        certlist = self.get_all_certs(nickname)
+        for cert in certlist:
+            verify_ca_cert(cert, nickname, minpathlen)
 
     def verify_kdc_cert_validity(self, nickname, realm):
         nicknames = self.get_trust_chain(nickname)

ipaserver/install/certs.py

@@ -371,7 +371,7 @@ class CertDB:
             except RuntimeError:
                 break
 
-    def get_cert_from_db(self, nickname):
+    def get_cert_from_db(self, nickname, all=False):
         """
         Retrieve a certificate from the current NSS database for nickname.
         """
@@ -384,7 +384,10 @@ class CertDB:
             if token:
                 args.extend(['-h', token])
             result = self.run_certutil(args, capture_output=True)
-            return x509.load_pem_x509_certificate(result.raw_output)
+            if all:
+                return x509.load_certificate_list(result.raw_output)
+            else:
+                return x509.load_pem_x509_certificate(result.raw_output)
         except ipautil.CalledProcessError:
             return None
 

ipaserver/install/installutils.py

@@ -898,7 +898,8 @@ def load_pkcs12(cert_files, key_password, key_nickname, ca_cert_files,
 
         if ca_cert_files:
             try:
-                nssdb.import_files(ca_cert_files)
+                nssdb.import_files(ca_cert_files,
+                                   trust_flags=EXTERNAL_CA_TRUST_FLAGS)
             except RuntimeError as e:
                 raise ScriptError(str(e))
 

ipaserver/install/ipa_cacert_manage.py

@@ -101,6 +101,9 @@ class CACertManage(admintool.AdminTool):
         delete_group.add_option(
             "-f", "--force", action='store_true',
             help="Force removing the CA even if chain validation fails")
+        delete_group.add_option(
+            "-s", "--serial",
+            help="Serial number of the certificate to delete (decimal)")
         parser.add_option_group(delete_group)
 
     def validate_options(self):
@@ -413,6 +416,11 @@ class CACertManage(admintool.AdminTool):
                     "Nickname can only be used if only a single "
                     "certificate is loaded")
 
+            for nickname, trust_flags in imported:
+                if trust_flags.has_key:
+                    continue
+                tmpdb.trust_root_cert(nickname, EXTERNAL_CA_TRUST_FLAGS)
+
             # If a nickname was provided re-import the cert
             if options.nickname:
                 (nickname, trust_flags) = imported[0]
@@ -421,7 +429,7 @@ class CACertManage(admintool.AdminTool):
                 tmpdb.add_cert(cert, options.nickname, EXTERNAL_CA_TRUST_FLAGS)
                 imported = tmpdb.list_certs()
 
-            for ca_cert, ca_nickname, ca_trust_flags in ca_certs:
+            for ca_cert, ca_nickname, ca_trust_flags, _serial in ca_certs:
                 tmpdb.add_cert(ca_cert, ca_nickname, ca_trust_flags)
 
             for nickname, trust_flags in imported:
@@ -461,10 +469,11 @@ class CACertManage(admintool.AdminTool):
 
             for nickname, _trust_flags in imported:
                 try:
-                    cert = tmpdb.get_cert(nickname)
-                    certstore.put_ca_cert_nss(
-                        api.Backend.ldap2, api.env.basedn, cert, nickname,
-                        trust_flags)
+                    certlist = tmpdb.get_all_certs(nickname)
+                    for cert in certlist:
+                        certstore.put_ca_cert_nss(
+                            api.Backend.ldap2, api.env.basedn, cert, nickname,
+                            trust_flags)
                 except ValueError as e:
                     raise admintool.ScriptError(
                         "Failed to install the certificate: %s" % e)
@@ -476,8 +485,8 @@ class CACertManage(admintool.AdminTool):
                                               api.env.basedn,
                                               api.env.realm,
                                               False)
-        for _ca_cert, ca_nickname, _ca_trust_flags in ca_certs:
-            print(ca_nickname)
+        for _ca_cert, ca_nickname, _ca_trust_flags, serial in ca_certs:
+            print(f"{ca_nickname}  {serial}")
 
     def _delete_by_nickname(self, nicknames, options):
         conn = api.Backend.ldap2
@@ -489,9 +498,25 @@ class CACertManage(admintool.AdminTool):
 
         ipa_ca_nickname = get_ca_nickname(api.env.realm)
 
+        # Count the number of times the nickname appears in case we
+        # have a duplicate. If a serial number is provided we can skip
+        # this.
+        cert_count = 0
+        if not options.serial:
+            for nickname in nicknames:
+                for _ca_cert, ca_nickname, _ca_trust_flags, _serial in ca_certs:
+                    if ca_nickname == nickname:
+                        cert_count += 1
+            if cert_count > 1:
+                raise admintool.ScriptError(
+                    'Multiple matching certificates found (%d). Use the '
+                    '--serial option to specify which one to remove.' %
+                    cert_count
+                )
+
         for nickname in nicknames:
             found = False
-            for _ca_cert, ca_nickname, _ca_trust_flags in ca_certs:
+            for _ca_cert, ca_nickname, _ca_trust_flags, _serial in ca_certs:
                 if ca_nickname == nickname:
                     if ca_nickname == ipa_ca_nickname:
                         raise admintool.ScriptError(
@@ -508,13 +533,17 @@ class CACertManage(admintool.AdminTool):
 
         with certs.NSSDatabase() as tmpdb:
             tmpdb.create_db()
-            for ca_cert, ca_nickname, ca_trust_flags in ca_certs:
+            for ca_cert, ca_nickname, ca_trust_flags, serial in ca_certs:
+                if nickname == ca_nickname:
+                    if options.serial and options.serial == serial:
+                        continue
                 tmpdb.add_cert(ca_cert, ca_nickname, ca_trust_flags)
             loaded = tmpdb.list_certs()
             logger.debug("loaded raw certs '%s'", loaded)
 
-            for nickname in nicknames:
-                tmpdb.delete_cert(nickname)
+            if not options.serial:
+                for nickname in nicknames:
+                    tmpdb.delete_cert(nickname)
 
             for ca_nickname, _trust_flags in loaded:
                 if ca_nickname in nicknames:
@@ -526,8 +555,8 @@ class CACertManage(admintool.AdminTool):
                 try:
                     tmpdb.verify_ca_cert_validity(ca_nickname)
                 except ValueError as e:
-                    msg = "Verifying \'%s\' failed. Removing part of the " \
-                          "chain? %s" % (nickname, e)
+                    msg = "Verifying removal of \'%s\' failed. Removing " \
+                          "part of the chain? %s" % (nickname, e)
                     if options.force:
                         print(msg)
                         continue
@@ -535,15 +564,20 @@ class CACertManage(admintool.AdminTool):
                 else:
                     logger.debug("Verified %s", ca_nickname)
 
-        for _ca_cert, ca_nickname, _ca_trust_flags in ca_certs:
+        for ca_cert, ca_nickname, _ca_trust_flags, serial in ca_certs:
             if ca_nickname in nicknames:
-                container_dn = DN(('cn', 'certificates'), ('cn', 'ipa'),
-                                  ('cn', 'etc'), api.env.basedn)
-                dn = DN(('cn', nickname), container_dn)
+                if options.serial and options.serial != serial:
+                    continue
                 logger.debug("Deleting %s", ca_nickname)
-                conn.delete_entry(dn)
+                certstore.delete_ca_cert(conn, api.env.basedn, ca_cert)
+
                 return
 
+        raise admintool.ScriptError(
+            "Certificate with name %s and serial number %s not found"
+            % (ca_nickname, options.serial)
+        )
+
     def delete(self):
         nickname = self.args[1]
         self._delete_by_nickname([nickname], self.options)
@@ -556,17 +590,17 @@ class CACertManage(admintool.AdminTool):
                                               False)
 
         now = datetime.datetime.now(tz=datetime.timezone.utc)
-        for ca_cert, ca_nickname, _ca_trust_flags in ca_certs:
+        for ca_cert, ca_nickname, _ca_trust_flags, _serial in ca_certs:
             if ca_cert.not_valid_after_utc < now:
                 expired_certs.append(ca_nickname)
 
-
+        del_options = self.options
+        del_options.force = True
         if expired_certs:
-            self._delete_by_nickname(expired_certs, self.options)
-
             print("Expired certificates deleted:")
-            for nickname in expired_certs:
-                print(nickname)
+            for ca_cert in expired_certs:
+                self._delete_by_nickname([ca_cert], del_options)
+                print(ca_cert)
             print("Run ipa-certupdate on enrolled machines to apply changes.")
         else:
             print("No certificates were deleted")

ipaserver/install/ipa_server_certinstall.py

@@ -276,8 +276,9 @@ class ServerCertInstall(admintool.AdminTool):
             # import all the CA certs from nssdb into the temp db
             for nickname, flags in nssdb.list_certs():
                 if not flags.has_key:
-                    cert = nssdb.get_cert_from_db(nickname)
-                    tempnssdb.add_cert(cert, nickname, flags)
+                    certs = nssdb.get_cert_from_db(nickname, all=True)
+                    for cert in certs:
+                        tempnssdb.add_cert(cert, nickname, flags)
 
             # now get the server certs from tempnssdb and check their validity
             try:

ipaserver/install/krbinstance.py

@@ -536,7 +536,7 @@ class KrbInstance(service.Service):
                                           self.api.env.basedn,
                                           self.api.env.realm,
                                           False)
-        ca_certs = [c for c, _n, t, _u in ca_certs if t is not False]
+        ca_certs = [c for c, _n, t, _u, _s in ca_certs if t is not False]
         x509.write_certificate_list(ca_certs, paths.CACERT_PEM, mode=0o644)
 
     def issue_selfsigned_pkinit_certs(self):

ipaserver/install/service.py

@@ -541,7 +541,7 @@ class Service:
             pass
         else:
             with open(cafile, 'wb') as fd:
-                for cert, _unused1, _unused2, _unused3 in ca_certs:
+                for cert, _unused1, _unused2, _unused3, _unused4 in ca_certs:
                     fd.write(cert.public_bytes(x509.Encoding.PEM))
 
     def export_ca_certs_nssdb(self, db, ca_is_configured, conn=None):
@@ -561,7 +561,7 @@ class Service:
         except errors.NotFound:
             pass
         else:
-            for cert, nickname, trust_flags in ca_certs:
+            for cert, nickname, trust_flags, _serial in ca_certs:
                 db.add_cert(cert, nickname, trust_flags)
 
     def is_configured(self):

ipatests/test_integration/test_commands.py

@@ -123,6 +123,82 @@ letsencryptauthorityr3 = (
 )
 le_r3_nick = "CN=R3,O=Let's Encrypt,C=US"
 
+# Certificates for reproducing duplicate ipaCertSubject values.
+# The trick to creating the second intermediate is for the validity
+# period to be different. In this case the second CA certificate
+# was issued 3 years+1day after the original.
+originalsubjectchain = (
+    b'-----BEGIN CERTIFICATE-----\n'
+    b'MIIDcjCCAlqgAwIBAgICEAAwDQYJKoZIhvcNAQELBQAwRDEeMBwGA1UECgwVQ2Vy\n'
+    b'dGlmaWNhdGUgU2hhY2sgTHRkMSIwIAYDVQQDDBlDZXJ0aWZpY2F0ZSBTaGFjayBS\n'
+    b'b290IENBMB4XDTIxMDgwNzE4MDQyNloXDTQxMDgwMTE4MDQyNlowTDEeMBwGA1UE\n'
+    b'CgwVQ2VydGlmaWNhdGUgU2hhY2sgTHRkMSowKAYDVQQDDCFDZXJ0aWZpY2F0ZSBT\n'
+    b'aGFjayBJbnRlcm1lZGlhdGUgQ0EwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEK\n'
+    b'AoIBAQC2RNo7atuVWC/6tDCGforNFvvSFdUwqHxltFmg61i2hmdHAjTaYI1ZJdgB\n'
+    b'y7ApGc8RYc7tfaNrUNA8Chd/9Cu4eW2KuTnAozxytXQneNXloK2xb9iLIhETa1FC\n'
+    b'Hw5BbrmJSWjiVYQsM6bzeiFsKJs4qnP1T9iFHuqmggTtCTPajoYhn6ZKfK3pmB8P\n'
+    b'6XRcp5O9vUhNHJWdpuUjOL32fsBEpV0vKWlsemqDhJrhzj3+YCKt6xrSdpK64HUW\n'
+    b'Kf3YM/K4G6vU5M8DgSFex6T1u2vCsQYJ4Mv8LVCho8awTZoBsimy1tiM0V7GmmBE\n'
+    b'0Uck/U0381NBpNYdv7eyF682SbihAgMBAAGjZjBkMB0GA1UdDgQWBBTtHQCp1dBF\n'
+    b'ypsegtWcXhXDdopIgDAfBgNVHSMEGDAWgBRJuz/14J1ZXqvpOuikJJ62NtuiGTAS\n'
+    b'BgNVHRMBAf8ECDAGAQH/AgEBMA4GA1UdDwEB/wQEAwIBhjANBgkqhkiG9w0BAQsF\n'
+    b'AAOCAQEAkCBm6u+k/x4QoqqwOJvy8sjq7bUCh73qNPAFlqVSSB8UdCyu21EaXCj8\n'
+    b'dbZa3GNRGk6JACTEUVQ1SD8SkC1E1/IWuEzYOKOP6FmTFbC4V5zU9LAnGFJapS6Q\n'
+    b'CGwU2F44oflBbfOodFznqKPPuENX0gmm4ddvoT915WUOvVLKLuVujkU/ffGKAc8U\n'
+    b'RxRIJ3W2Ybjs9ANg7JqB3Ny8i5QAGHzjRVwU+IgTrJCYPS2DrRYtN3glKBTlyKyR\n'
+    b'xMy0PVKwVo/ItDO3fZ0fsAiIO+4pI51A0lFge5Bg/DzsotZxcWhdTelWjYI9JNca\n'
+    b'y2GPzV1wlxK+ui1uLCWEvKbPtaCfeQ==\n'
+    b'-----END CERTIFICATE-----\n'
+    b'-----BEGIN CERTIFICATE-----\n'
+    b'MIIDeTCCAmGgAwIBAgIUUbo+eGRT5jiS2eIoEzRhXaUx4gwwDQYJKoZIhvcNAQEL\n'
+    b'BQAwRDEeMBwGA1UECgwVQ2VydGlmaWNhdGUgU2hhY2sgTHRkMSIwIAYDVQQDDBlD\n'
+    b'ZXJ0aWZpY2F0ZSBTaGFjayBSb290IENBMB4XDTIxMDgwNzE4MDQyNloXDTQxMDgw\n'
+    b'MjE4MDQyNlowRDEeMBwGA1UECgwVQ2VydGlmaWNhdGUgU2hhY2sgTHRkMSIwIAYD\n'
+    b'VQQDDBlDZXJ0aWZpY2F0ZSBTaGFjayBSb290IENBMIIBIjANBgkqhkiG9w0BAQEF\n'
+    b'AAOCAQ8AMIIBCgKCAQEArh41PPmI6rg7nz3cRqsbCqGgD3+vAD4DNs/Cnp+vhM//\n'
+    b'7Di8FuMoyyLDpD+RdT/Vkvh2Xhp+OcjYSFLX8xeFRy0blfzel2Tq7PiD83BwewsG\n'
+    b'BOarlhkbQGxlGxkr4Fi6z0kNNAfbE2ZzBIs4XSppm7xl4YJyLQD0FkzdrU+zrZuK\n'
+    b'3ELQzk3UWfSSrnbYABY2LBgkny5m7y/kJOMyqn+/T1CUthXD3OpGtyQm2kuEooDZ\n'
+    b'xP1eq30gS8oGYAw2nR/8vJPuyeZaMxM4eNLuc35uq8/6pI+xNEpzGt7xAk1ul/xc\n'
+    b'ewOY2kjh4KJCNK/nCjALzxqhNRHhnH8bA6xtOcgdBwIDAQABo2MwYTAdBgNVHQ4E\n'
+    b'FgQUSbs/9eCdWV6r6TropCSetjbbohkwHwYDVR0jBBgwFoAUSbs/9eCdWV6r6Tro\n'
+    b'pCSetjbbohkwDwYDVR0TAQH/BAUwAwEB/zAOBgNVHQ8BAf8EBAMCAYYwDQYJKoZI\n'
+    b'hvcNAQELBQADggEBAC35stv/1WZhWblRTZP3XHhH0usHRGTUY7zNSrgS5sb3ERsf\n'
+    b'hgbmFbomra5jKaBqffToOZKLEo+n3tfIPokus35NUQn7ox/6qPp0rJEK8dfLx9jA\n'
+    b'0VTqREbgaAf5xLaX874++OTiM1sPVYG3Egsb1A/YCtDek8mZkKk21g+DZlFMOSDl\n'
+    b'Hw+c3gZUnv6bIT8P09z+9yca2Lvg/dpj2ln3PbOykXzwuGSoNxjUt2OSdCbwyN+f\n'
+    b'hO4NFtDvx74Ggi5bcTrz0ZKO6g8SQotii7cSKAdpIWDpXl8cfsK3SRbkCsg+Fg1S\n'
+    b'kMJEFyDEkKu8Qe6zwKXIAoeKULLO6ADgFVH9CmM=\n'
+    b'-----END CERTIFICATE-----\n'
+)
+interm_nick = "CN=Certificate Shack Intermediate CA,O=Certificate Shack Ltd"
+intermediate_serial = "4096"
+
+duplicatesubject = (
+    b'-----BEGIN CERTIFICATE-----\n'
+    b'MIIDcjCCAlqgAwIBAgICEAEwDQYJKoZIhvcNAQELBQAwRDEeMBwGA1UECgwVQ2Vy\n'
+    b'dGlmaWNhdGUgU2hhY2sgTHRkMSIwIAYDVQQDDBlDZXJ0aWZpY2F0ZSBTaGFjayBS\n'
+    b'b290IENBMB4XDTI0MDgwODE4MDQyNloXDTQ0MDgwMjE4MDQyNlowTDEeMBwGA1UE\n'
+    b'CgwVQ2VydGlmaWNhdGUgU2hhY2sgTHRkMSowKAYDVQQDDCFDZXJ0aWZpY2F0ZSBT\n'
+    b'aGFjayBJbnRlcm1lZGlhdGUgQ0EwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEK\n'
+    b'AoIBAQCzUmUBEO/w1wslS8H304/qfsbeIJX0C5Tm8K2H9JXoauFFej1GZoHqeE+x\n'
+    b'YQvSMuMFcKks3ps9+9yVKuBPtMwbmXsqwlQXORU8DuKhtRzKIOj7nEGw6AQIsfkG\n'
+    b'Q4DjD1ytXliyM7vVfxYD+P1CFDK4NR+K1JLdi3WkYOdCelOQMwNspN/ebiqvwonl\n'
+    b'2asQ6+a13Y0ln1AdrLBvqtR5Z+Gq5+tiC5tA+LKea0e3neQGKjfp/BNPJ+ooNHPR\n'
+    b'86iKDjBKAabvfrHLG2t6oo9+N4xRBGtPYQh9LOQPZ4OedciCo1s2zs+F+4/6co6T\n'
+    b'DsbQt7NJKQ3BJKosvZBhC62lc4evAgMBAAGjZjBkMB0GA1UdDgQWBBTvALT5i2gq\n'
+    b'8yq2Uh8lZGgMoKVClzAfBgNVHSMEGDAWgBRJuz/14J1ZXqvpOuikJJ62NtuiGTAS\n'
+    b'BgNVHRMBAf8ECDAGAQH/AgEBMA4GA1UdDwEB/wQEAwIBhjANBgkqhkiG9w0BAQsF\n'
+    b'AAOCAQEAVjx1aGNK08/Nhf0JYMxMb9Dqg5m7LNOVBs1jurPtwS3uN+84997GRqIQ\n'
+    b'i+gp/tQVF2YT/RAmt+X0aDLFiSkBcOk87zoFRkR7PZrhhtPo6pSVMN7ngD4/dmp9\n'
+    b'ESbiI8+iF5ZxqI7c3o2N/LtZpi+hWSCJ/xwbOl05jpNQ6ddl+UzDpJ0oNsyndiJA\n'
+    b'yciaCvluK027J4xNym166lqwm6CqiOkm8R/G6NJrEH2Xs5XBCyfeH9V0pkXDbrUe\n'
+    b'Ldqc9ys7l7/MGZi6Qg2nA7J8ErCkrI6eZOocJktSF6SRfXd1NqiqCiNZZQjD6XKZ\n'
+    b'4fMKTKPX6Q2k10iriAIn4RgVjzM05A==\n'
+    b'-----END CERTIFICATE-----\n'
+)
+duplicate_serial = "4097"
+
 
 class TestIPACommand(IntegrationTest):
     """
@@ -827,6 +903,12 @@ class TestIPACommand(IntegrationTest):
                 paths.IPA_CACERT_MANAGE,
                 'install',
                 filename])
+        # remove the subject of good_pkcs7 we just added to avoid
+        # future failures.
+        self.master.run_command([
+            paths.IPA_CACERT_MANAGE,
+            'delete',
+            'CN=Certificate Authority,O=EXAMPLE.COM'])
 
         for contents in (badcert,):
             self.master.put_file_contents(filename, contents)
@@ -1160,7 +1242,7 @@ class TestIPACommand(IntegrationTest):
             raiseonerr=False
         )
         assert result.returncode != 0
-        assert "Verifying \'%s\' failed. Removing part of the " \
+        assert "Verifying removal of \'%s\' failed. Removing part of the " \
                "chain? certutil: certificate is invalid: Peer's " \
                "Certificate issuer is not recognized." \
                % isrgrootx1_nick in result.stderr_text
@@ -1648,6 +1730,68 @@ class TestIPACommand(IntegrationTest):
             self.replicas[0], '/usr/sbin/ipa-replica-install'
         )
 
+    def test_ipa_cacert_manage_duplicate_certsubject(self):
+        """Test for ipa-cacert-manage install with duplicated
+           certificate subjects. This relies on the behavior
+           of NSS to show the certificates separately rather than
+           lumping the duplicates together. This requires different
+           validity periods, say 3 years + 1 day.
+        """
+
+        certfile = os.path.join(self.master.config.test_dir, 'chain.pem')
+        self.master.put_file_contents(certfile, originalsubjectchain)
+        result = self.master.run_command(
+            [paths.IPA_CACERT_MANAGE, 'install', certfile])
+
+        certs = self.master.run_command(
+            [paths.IPA_CACERT_MANAGE, 'list'], raiseonerr=False
+        ).stdout_text
+
+        assert f"{interm_nick}  {intermediate_serial}" in certs
+
+        certfile = os.path.join(self.master.config.test_dir, 'interm.pem')
+        self.master.put_file_contents(certfile, duplicatesubject)
+        result = self.master.run_command(
+            [paths.IPA_CACERT_MANAGE, 'install', certfile])
+
+        certs = self.master.run_command(
+            [paths.IPA_CACERT_MANAGE, 'list'], raiseonerr=False
+        ).stdout_text
+
+        # If the duplicate subject certificates are not sufficiently
+        # different in validity period, or prior to the this fix,
+        # the test will fail because only one of the duplicately named
+        # subject certificates will be visible: the second one (4097).
+        assert f"{interm_nick}  {intermediate_serial}" in certs
+        assert f"{interm_nick}  {duplicate_serial}" in certs
+
+        # Make sure we can install the new certs systemwide
+        # No assertions needed, it will work or it won't
+        self.master.run_command(["ipa-certupdate"])
+
+        # delete one of the duplicate subjects, no serial number
+        result = self.master.run_command(
+            ['ipa-cacert-manage', 'delete', interm_nick],
+            raiseonerr=False
+        )
+        assert result.returncode == 1
+        assert 'Multiple matching certificates' in result.stderr_text
+
+        # delete one of the duplicate subjects by the serial number
+        result = self.master.run_command(
+            ['ipa-cacert-manage', 'delete', interm_nick,
+             '--serial', intermediate_serial,],
+            raiseonerr=False
+        )
+        assert result.returncode == 0
+
+        certs = self.master.run_command(
+            [paths.IPA_CACERT_MANAGE, 'list'], raiseonerr=False
+        ).stdout_text
+
+        assert f"{interm_nick}  {intermediate_serial}" not in certs
+        assert f"{interm_nick}  {duplicate_serial}" in certs
+
 
 class TestIPACommandWithoutReplica(IntegrationTest):
     """
@@ -1883,7 +2027,10 @@ class TestIPACommandWithoutReplica(IntegrationTest):
         assert re.search(new_err_msg, dirsrv_error_log)
 
     def test_ipa_cacert_manage_prune(self):
-        """Test for ipa-cacert-manage prune"""
+        """Test for ipa-cacert-manage prune
+
+           This twiddles with time so should be run last in the class.
+        """
 
         certfile = os.path.join(self.master.config.test_dir, 'cert.pem')
         self.master.put_file_contents(certfile, isrgrootx1)