Check user-provided certificates for expiration

src/ipahealthcheck/ipa/certs.py

@@ -453,6 +453,87 @@ def check(self):
                 yield Result(self, constants.SUCCESS, key=id)
 
 
+@registry
+class IPAUserProvidedExpirationCheck(IPAPlugin):
+    """
+    If we detect user-provided certificates then check to see if
+    they are expiring soon.
+
+    This only applies to the HTTP, DS and KRB certificates.
+    """
+    def validate_cert(self, id, cert, certfile=None, nssdb=None,
+                      nickname=None):
+        now = datetime.now(tz=timezone.utc)
+        notafter = cert.not_valid_after_utc
+
+        if certfile:
+            location = certfile
+        if nssdb:
+            location = "{}:{}".format(nssdb, nickname)
+
+        if now > notafter:
+            logger.debug("expired")
+            yield Result(self, constants.ERROR,
+                         key=id,
+                         location=location,
+                         expiration_date=generalized_time(notafter),
+                         msg='Request id {key} expired on '
+                             '{expiration_date}')
+            return
+
+        delta = notafter - now
+        diff = int(delta.total_seconds() / DAY)
+        if diff < int(self.config.cert_expiration_days):
+            logger.debug("expiring soon")
+            yield Result(self, constants.WARNING,
+                         key=id,
+                         location=location,
+                         expiration_date=generalized_time(notafter),
+                         days=diff,
+                         msg='Request id {key} expires in {days} '
+                             'days. You need to manually renew this '
+                             'certificate.')
+            return
+
+        yield Result(self, constants.SUCCESS, location=location, key=id)
+
+    @duration
+    def check(self):
+        for id, certfile in (
+            ("HTTP", paths.HTTPD_CERT_FILE),
+            ("KDC", paths.KDC_CERT)
+        ):
+            try:
+                cert = x509.load_certificate_from_file(certfile)
+            except Exception as e:
+                yield Result(self, constants.ERROR,
+                             key=id,
+                             certfile=certfile,
+                             error=str(e),
+                             msg='Request id {key}: Unable to open cert '
+                                 'file \'{certfile}\': {error}')
+                continue
+            issued = is_ipa_issued_cert(api, cert)
+            if issued is None:
+                logger.debug('Unable to determine if \'%s\' was issued by IPA '
+                             'because no LDAP connection, assuming yes.')
+                continue
+            elif issued is True:
+                logging.debug("Issued by IPA, skipping")
+                continue
+            yield from self.validate_cert(id, cert, certfile=certfile)
+
+        ds_db_dirname = dsinstance.config_dirname(self.serverid)
+        ds_db = certs.CertDB(api.env.realm, nssdir=ds_db_dirname)
+        nickname = self.ds.get_server_cert_nickname(self.serverid)
+        cert = ds_db.get_cert_from_db(nickname)
+        if is_ipa_issued_cert(api, cert):
+            logging.debug("Issued by IPA, skipping")
+            return
+        yield from self.validate_cert("LDAP", cert, nssdb=ds_db_dirname,
+                                      nickname=nickname)
+
+
 @registry
 class IPACertTracking(IPAPlugin):
     """Compare the certificates tracked by certmonger to those that

tests/common.py

@@ -0,0 +1,71 @@
+#
+# Copyright (C) 2025 FreeIPA Contributors see COPYING for license
+#
+from datetime import datetime, timedelta, timezone
+
+
+class mock_Cert:
+    """Fake up a certificate.
+
+    The contents are the NSS nickname of the certificate.
+    """
+
+    def __init__(
+        self,
+        text,
+        issuer="CN=Someone",
+        not_after=datetime.now(tz=timezone.utc),
+    ):
+        self.text = text
+        self._issuer = issuer
+        self._not_valid_after_utc = not_after
+
+    def public_bytes(self, encoding):
+        return self.text.encode("utf-8")
+
+    @property
+    def issuer(self):
+        return self._issuer
+
+    @property
+    def not_valid_after_utc(self):
+        return self._not_valid_after_utc
+
+
+class mock_CertDB:
+    def __init__(self, trust, expiration_days=0):
+        """A dict of nickname + NSSdb trust flags"""
+        self.trust = trust
+        self._expiration_days = expiration_days
+
+    def list_certs(self):
+        return [
+            (nickname, self.trust[nickname]) for nickname in self.trust
+        ]
+
+    def get_cert_from_db(self, nickname):
+        """Return the nickname. This will match the value of get_directive"""
+        notafter = datetime.now(tz=timezone.utc) + timedelta(
+            days=self._expiration_days
+        )
+        return mock_Cert(nickname, not_after=notafter)
+
+
+class mock_NSSDatabase:
+    def __init__(self, nssdir, token=None, pwd_file=None, trust=None):
+        self.trust = trust
+        self.token = token
+
+    def list_certs(self):
+        return [
+            (nickname, self.trust[nickname]) for nickname in self.trust
+        ]
+
+
+def my_unparse_trust_flags(trust_flags):
+    return trust_flags
+
+
+class DsInstance:
+    def get_server_cert_nickname(self, serverid):
+        return "Server-Cert"

tests/test_ipa_nssdb.py

@@ -9,28 +9,7 @@
 from ipahealthcheck.ipa.certs import IPACertNSSTrust
 from ipaplatform.paths import paths
 from unittest.mock import Mock, patch
-
-
-class mock_CertDB:
-    def __init__(self, trust):
-        """A dict of nickname + NSSdb trust flags"""
-        self.trust = trust
-
-    def list_certs(self):
-        return [(nickname, self.trust[nickname]) for nickname in self.trust]
-
-
-class mock_NSSDatabase:
-    def __init__(self, nssdir, token=None, pwd_file=None, trust=None):
-        self.trust = trust
-        self.token = token
-
-    def list_certs(self):
-        return [(nickname, self.trust[nickname]) for nickname in self.trust]
-
-
-def my_unparse_trust_flags(trust_flags):
-    return trust_flags
+from common import mock_NSSDatabase, my_unparse_trust_flags
 
 
 # These tests make some assumptions about the order in which the

tests/test_ipa_userprovided_expiration.py

@@ -0,0 +1,110 @@
+#
+# Copyright (C) 2025 FreeIPA Contributors see COPYING for license
+#
+
+from util import capture_results
+from base import BaseTest
+from common import DsInstance
+from ipahealthcheck.core import config, constants
+from ipahealthcheck.ipa.plugin import registry
+from ipahealthcheck.ipa.certs import IPAUserProvidedExpirationCheck
+from unittest.mock import Mock, patch
+from ipapython.dn import DN
+from common import mock_CertDB
+
+from datetime import datetime, timedelta, timezone
+
+CERT_EXPIRATION_DAYS = 30
+
+
+class IPACertificate:
+    def __init__(self, not_valid_after, serial_number=1):
+        self.subject = 'CN=RA AGENT'
+        self.issuer = 'CN=ISSUER'
+        self.serial_number = serial_number
+        self.not_valid_after_utc = not_valid_after
+
+
+class TestIPACertificateFile(BaseTest):
+    patches = {
+        'ipaserver.install.dsinstance.DsInstance':
+        Mock(return_value=DsInstance()),
+        'ipalib.install.certstore.get_ca_subject':
+        Mock(return_value=DN("CN=EXTERNAL")),
+        'ipaserver.install.certs.is_ipa_issued_cert':
+        Mock(return_value=False),
+    }
+
+    @patch('ipalib.x509.load_certificate_from_file')
+    @patch('ipaserver.install.certs.CertDB')
+    def test_certfile_expiration(self, mock_certdb, mock_load_cert):
+        cert = IPACertificate(not_valid_after=datetime.now(tz=timezone.utc) +
+                              timedelta(days=CERT_EXPIRATION_DAYS))
+        mock_load_cert.return_value = cert
+        mock_certdb.return_value = mock_CertDB({
+            'Server-Cert cert-pki-ca': 'u,u,u',
+        }, expiration_days=CERT_EXPIRATION_DAYS)
+
+        framework = object()
+        registry.initialize(framework, config.Config)
+        f = IPAUserProvidedExpirationCheck(registry)
+
+        f.config.cert_expiration_days = '28'
+        self.results = capture_results(f)
+
+        assert len(self.results) == 3
+
+        for result in self.results.results:
+            assert result.result == constants.SUCCESS
+            assert result.source == 'ipahealthcheck.ipa.certs'
+            assert result.check == 'IPAUserProvidedExpirationCheck'
+
+    @patch('ipalib.x509.load_certificate_from_file')
+    @patch('ipaserver.install.certs.CertDB')
+    def test_certfile_expiration_warning(self, mock_certdb, mock_load_cert):
+        cert = IPACertificate(not_valid_after=datetime.now(tz=timezone.utc) +
+                              timedelta(days=7))
+        mock_load_cert.return_value = cert
+        mock_certdb.return_value = mock_CertDB({
+            'Server-Cert cert-pki-ca': 'u,u,u',
+        }, expiration_days=7)
+
+        framework = object()
+        registry.initialize(framework, config.Config)
+        f = IPAUserProvidedExpirationCheck(registry)
+
+        f.config.cert_expiration_days = str(CERT_EXPIRATION_DAYS)
+        self.results = capture_results(f)
+
+        assert len(self.results) == 3
+
+        for result in self.results.results:
+            assert result.result == constants.WARNING
+            assert result.source == 'ipahealthcheck.ipa.certs'
+            assert result.check == 'IPAUserProvidedExpirationCheck'
+            assert result.kw.get('days') == 6
+
+    @patch('ipalib.x509.load_certificate_from_file')
+    @patch('ipaserver.install.certs.CertDB')
+    def test_certfile_expiration_expired(self, mock_certdb, mock_load_cert):
+        cert = IPACertificate(not_valid_after=datetime.now(tz=timezone.utc) +
+                              timedelta(days=-100))
+        mock_load_cert.return_value = cert
+        mock_certdb.return_value = mock_CertDB({
+            'Server-Cert cert-pki-ca': 'u,u,u',
+        }, expiration_days=-100)
+
+        framework = object()
+        registry.initialize(framework, config.Config)
+        f = IPAUserProvidedExpirationCheck(registry)
+
+        f.config.cert_expiration_days = str(CERT_EXPIRATION_DAYS)
+        self.results = capture_results(f)
+
+        assert len(self.results) == 3
+
+        for result in self.results.results:
+            assert result.result == constants.ERROR
+            assert result.source == 'ipahealthcheck.ipa.certs'
+            assert result.check == 'IPAUserProvidedExpirationCheck'
+            assert 'expiration_date' in result.kw