Deprecate {csr, keys} dirs & automatically truncate lineages (#9537)

Based on my design [here](https://docs.google.com/document/d/1jGh_bZPnrhi96KzuIcyCJfnudl4m3pRPGkiK4fTo8e4/edit?usp=sharing). 

Fixes https://github.com/certbot/certbot/issues/4634 and https://github.com/certbot/certbot/issues/4635.

- [x] Deprecate `NamespaceConfig.csr_dir`,`NamespaceConfig.key_dir`, ~~`constants.CSR_DIR` and `constants.KEY_DIR`~~. (`constants` is `_internal` so we can just delete it eventually).
- [x] Update `certbot.crypto_util.generate_csr` and `.generate_key` to make `csr_dir` and `key_dir` optional, respectively.
- [x] Change `certbot._internal.client.Client.obtain_certificate` to no longer include `csr_dir` and `key_dir` to the `.generate_csr` and `.generate_key` calls, respectively.
- Automatically delete unwanted lineage items:
  - [x] In `certbot._internal.storage.RenewableCert`, add a function to truncate the lineage history according to the criteria (keep the current and the 5 prior certificates). 
      - [x] Add a test suite for `truncate` 
  - [x] In `certbot._internal.renewal.renew_cert`, call the lineage truncation function after the symlinks have been updated for the renewal.


* Stop writing new files to /csr and /keys

* storage: add lineage truncation

* remove unused code

* deprecate keys_dir and csr_dir

* update CHANGELOG

* just keep 5 prior certificates, dont be clever with expiry

* docs: remove reference to /archive and /keys

* filter {csr,key}_dir deprecations directly in tests

certbot-ci/certbot_integration_tests/certbot_tests/test_main.py

@@ -329,7 +329,6 @@ def test_renew_with_changed_private_key_complexity(context: IntegrationTestsCont
 
     context.certbot(['renew', '--rsa-key-size', '2048'])
 
-    assert_cert_count_for_lineage(context.config_dir, certname, 3)
     key3 = join(context.config_dir, 'archive', certname, 'privkey3.pem')
     assert_rsa_key(key3, 2048)
 
@@ -437,38 +436,37 @@ def test_reuse_key(context: IntegrationTestsContext) -> None:
 
     with open(join(context.config_dir, 'archive/{0}/privkey1.pem').format(certname), 'r') as file:
         privkey1 = file.read()
+    with open(join(context.config_dir, 'archive/{0}/cert1.pem').format(certname), 'r') as file:
+        cert1 = file.read()
     with open(join(context.config_dir, 'archive/{0}/privkey2.pem').format(certname), 'r') as file:
         privkey2 = file.read()
+    with open(join(context.config_dir, 'archive/{0}/cert2.pem').format(certname), 'r') as file:
+        cert2 = file.read()
     assert privkey1 == privkey2
 
     context.certbot(['--cert-name', certname, '--domains', certname, '--force-renewal'])
 
     with open(join(context.config_dir, 'archive/{0}/privkey3.pem').format(certname), 'r') as file:
         privkey3 = file.read()
+    with open(join(context.config_dir, 'archive/{0}/cert3.pem').format(certname), 'r') as file:
+        cert3 = file.read()
     assert privkey2 != privkey3
 
     context.certbot(['--cert-name', certname, '--domains', certname,
                      '--reuse-key','--force-renewal'])
-    context.certbot(['renew', '--cert-name', certname, '--no-reuse-key', '--force-renewal'])
-    context.certbot(['renew', '--cert-name', certname, '--force-renewal'])
-
     with open(join(context.config_dir, 'archive/{0}/privkey4.pem').format(certname), 'r') as file:
         privkey4 = file.read()
+    context.certbot(['renew', '--cert-name', certname, '--no-reuse-key', '--force-renewal'])
     with open(join(context.config_dir, 'archive/{0}/privkey5.pem').format(certname), 'r') as file:
         privkey5 = file.read()
+    context.certbot(['renew', '--cert-name', certname, '--force-renewal'])
     with open(join(context.config_dir, 'archive/{0}/privkey6.pem').format(certname), 'r') as file:
         privkey6 = file.read()
+
     assert privkey3 == privkey4
     assert privkey4 != privkey5
     assert privkey5 != privkey6
 
-    with open(join(context.config_dir, 'archive/{0}/cert1.pem').format(certname), 'r') as file:
-        cert1 = file.read()
-    with open(join(context.config_dir, 'archive/{0}/cert2.pem').format(certname), 'r') as file:
-        cert2 = file.read()
-    with open(join(context.config_dir, 'archive/{0}/cert3.pem').format(certname), 'r') as file:
-        cert3 = file.read()
-
     assert len({cert1, cert2, cert3}) == 3
 
 
@@ -615,7 +613,6 @@ def test_renew_with_ec_keys(context: IntegrationTestsContext) -> None:
     # to the lineage key type, Certbot should keep the lineage key type. The curve will still
     # change to the default value, in order to stay consistent with the behavior of certonly.
     context.certbot(['certonly', '--force-renewal', '-d', certname])
-    assert_cert_count_for_lineage(context.config_dir, certname, 3)
     key3 = join(context.config_dir, 'archive', certname, 'privkey3.pem')
     assert 200 < os.stat(key3).st_size < 250  # ec keys of 256 bits are ~225 bytes
     assert_elliptic_key(key3, SECP256R1)
@@ -629,14 +626,12 @@ def test_renew_with_ec_keys(context: IntegrationTestsContext) -> None:
 
     context.certbot(['certonly', '--force-renewal', '-d', certname,
                      '--key-type', 'rsa', '--cert-name', certname])
-    assert_cert_count_for_lineage(context.config_dir, certname, 4)
     key4 = join(context.config_dir, 'archive', certname, 'privkey4.pem')
     assert_rsa_key(key4)
 
     # We expect that the previous behavior of requiring both --cert-name and
     # --key-type to be set to not apply to the renew subcommand.
     context.certbot(['renew', '--force-renewal', '--key-type', 'ecdsa'])
-    assert_cert_count_for_lineage(context.config_dir, certname, 5)
     key5 = join(context.config_dir, 'archive', certname, 'privkey5.pem')
     assert 200 < os.stat(key5).st_size < 250  # ec keys of 256 bits are ~225 bytes
     assert_elliptic_key(key5, SECP256R1)

certbot-nginx/certbot_nginx/_internal/configurator.py

@@ -693,6 +693,7 @@ class NginxConfigurator(common.Configurator):
         le_key = crypto_util.generate_key(
             key_size=1024, key_dir=tmp_dir, keyname="key.pem",
             strict_permissions=self.config.strict_permissions)
+        assert le_key.file is not None
         key = OpenSSL.crypto.load_privatekey(
             OpenSSL.crypto.FILETYPE_PEM, le_key.pem)
         cert = acme_crypto_util.gen_ss_cert(key, domains=[socket.gethostname()])

certbot/CHANGELOG.md

@@ -10,7 +10,10 @@ Certbot adheres to [Semantic Versioning](https://semver.org/).
 
 ### Changed
 
-*
+* Certbot will no longer save previous CSRs and certificate private keys to `/etc/letsencrypt/csr` and `/etc/letsencrypt/keys`, respectively. These directories may be safely deleted.
+* Certbot will now only keep the current and 5 previous certificates in the `/etc/letsencrypt/archive` directory for each certificate lineage. Any prior certificates will be automatically deleted upon renewal. This number may be further lowered in future releases.
+  * As always, users should only reference the certificate files within `/etc/letsencrypt/live` and never use `/etc/letsencrypt/archive` directly. See [Where are my certificates?](https://eff-certbot.readthedocs.io/en/stable/using.html#where-are-my-certificates) in the Certbot User Guide.
+* `certbot.configuration.NamespaceConfig.key_dir` and `.csr_dir` are now deprecated.
 
 ### Fixed
 

certbot/certbot/_internal/client.py

@@ -416,13 +416,13 @@ class Client:
         else:
             key = key or crypto_util.generate_key(
                 key_size=key_size,
-                key_dir=self.config.key_dir,
+                key_dir=None,
                 key_type=self.config.key_type,
                 elliptic_curve=elliptic_curve,
                 strict_permissions=self.config.strict_permissions,
             )
-            csr = crypto_util.generate_csr(key, domains, self.config.csr_dir,
-                                           self.config.must_staple, self.config.strict_permissions)
+            csr = crypto_util.generate_csr(
+                key, domains, None, self.config.must_staple, self.config.strict_permissions)
 
         try:
             orderr = self._get_order_and_authorizations(csr.data, self.config.allow_subset_of_names)
@@ -433,7 +433,7 @@ class Client:
             if self.config.allow_subset_of_names:
                 successful_domains = self._successful_domains_from_error(error, domains)
                 if successful_domains != domains and len(successful_domains) != 0:
-                    return self._retry_obtain_certificate(key, csr, domains, successful_domains)
+                    return self._retry_obtain_certificate(domains, successful_domains)
             raise
         authzr = orderr.authorizations
         auth_domains = {a.body.identifier.value for a in authzr}
@@ -445,7 +445,7 @@ class Client:
         # domains contains a wildcard because the ACME spec forbids identifiers
         # in authzs from containing a wildcard character.
         if self.config.allow_subset_of_names and successful_domains != domains:
-            return self._retry_obtain_certificate(key, csr, domains, successful_domains)
+            return self._retry_obtain_certificate(domains, successful_domains)
         else:
             try:
                 cert, chain = self.obtain_certificate_from_csr(csr, orderr)
@@ -457,7 +457,7 @@ class Client:
                 if self.config.allow_subset_of_names:
                     successful_domains = self._successful_domains_from_error(error, domains)
                     if successful_domains != domains and len(successful_domains) != 0:
-                        return self._retry_obtain_certificate(key, csr, domains, successful_domains)
+                        return self._retry_obtain_certificate(domains, successful_domains)
                 raise
 
     def _get_order_and_authorizations(self, csr_pem: bytes,
@@ -540,16 +540,12 @@ class Client:
             return successful_domains
         return []
 
-    def _retry_obtain_certificate(self, key: util.Key,
-                                csr: util.CSR, domains: List[str], successful_domains: List[str]
+    def _retry_obtain_certificate(self, domains: List[str], successful_domains: List[str]
                                 ) -> Tuple[bytes, bytes, util.Key, util.CSR]:
         failed_domains = [d for d in domains if d not in successful_domains]
         domains_list = ", ".join(failed_domains)
         display_util.notify("Unable to obtain a certificate with every requested "
             f"domain. Retrying without: {domains_list}")
-        if not self.config.dry_run:
-            os.remove(key.file)
-            os.remove(csr.file)
         return self.obtain_certificate(successful_domains)
 
     def _choose_lineagename(self, domains: List[str], certname: Optional[str]) -> str:

certbot/certbot/_internal/renewal.py

@@ -392,6 +392,7 @@ def renew_cert(config: configuration.NamespaceConfig, domains: Optional[List[str
         # TODO: Check return value of save_successor
         lineage.save_successor(prior_version, new_cert, new_key.pem, new_chain, config)
         lineage.update_all_links_to(lineage.latest_common_version())
+        lineage.truncate()
 
     hooks.renew_hook(config, domains, lineage.live_dir)
 

certbot/certbot/_internal/storage.py

@@ -1,4 +1,5 @@
 """Renewable certificates storage."""
+# pylint: disable=too-many-lines
 import datetime
 import glob
 import logging
@@ -1243,3 +1244,32 @@ class RenewableCert(interfaces.RenewableCert):
         self.configuration = config_with_defaults(self.configfile)
 
         return target_version
+
+    def truncate(self, num_prior_certs_to_keep: int = 5) -> None:
+        """Delete unused historical certificate, chain and key items from the lineage.
+
+        A certificate version will be deleted if it is:
+          1. not the current target, and
+          2. not a previous version within num_prior_certs_to_keep.
+
+        :param num_prior_certs_to_keep: How many prior certificate versions to keep.
+
+        """
+        # Do not want to delete the current or the previous num_prior_certs_to_keep certs
+        current_version = self.latest_common_version()
+        versions_to_delete = set(self.available_versions("cert"))
+        versions_to_delete -= set(range(current_version,
+                                        current_version - 1 - num_prior_certs_to_keep, -1))
+        archive = self.archive_dir
+
+        # Delete the remaining lineage items kinds for those certificate versions.
+        for ver in versions_to_delete:
+            logger.debug("Deleting %s/cert%d.pem and related items during clean up",
+                         archive, ver)
+            for kind in ALL_FOUR:
+                item_path = os.path.join(archive, f"{kind}{ver}.pem")
+                try:
+                    if os.path.exists(item_path):
+                        os.unlink(item_path)
+                except OSError:
+                    logger.debug("Failed to clean up %s", item_path, exc_info=True)

certbot/certbot/configuration.py

@@ -5,6 +5,7 @@ from typing import Any
 from typing import List
 from typing import Optional
 from urllib import parse
+import warnings
 
 from certbot import errors
 from certbot import util
@@ -150,6 +151,8 @@ class NamespaceConfig:
     @property
     def csr_dir(self) -> str:
         """Directory where new Certificate Signing Requests (CSRs) are saved."""
+        warnings.warn("NamespaceConfig.csr_dir is deprecated and will be removed in an upcoming "
+                      "release of Certbot", DeprecationWarning)
         return os.path.join(self.namespace.config_dir, constants.CSR_DIR)
 
     @property
@@ -160,6 +163,8 @@ class NamespaceConfig:
     @property
     def key_dir(self) -> str:
         """Keys storage."""
+        warnings.warn("NamespaceConfig.key_dir is deprecated and will be removed in an upcoming "
+                      "release of Certbot", DeprecationWarning)
         return os.path.join(self.namespace.config_dir, constants.KEY_DIR)
 
     @property

certbot/certbot/crypto_util.py

@@ -51,7 +51,7 @@ logger = logging.getLogger(__name__)
 
 # High level functions
 
-def generate_key(key_size: int, key_dir: str, key_type: str = "rsa",
+def generate_key(key_size: int, key_dir: Optional[str], key_type: str = "rsa",
                  elliptic_curve: str = "secp256r1", keyname: str = "key-certbot.pem",
                  strict_permissions: bool = True) -> util.Key:
     """Initializes and saves a privkey.
@@ -62,7 +62,7 @@ def generate_key(key_size: int, key_dir: str, key_type: str = "rsa",
         already exists at the path.
 
     :param int key_size: key size in bits if key size is rsa.
-    :param str key_dir: Key save directory.
+    :param str key_dir: Optional key save directory.
     :param str key_type: Key Type [rsa, ecdsa]
     :param str elliptic_curve: Name of the elliptic curve if key type is ecdsa.
     :param str keyname: Filename of key
@@ -85,27 +85,29 @@ def generate_key(key_size: int, key_dir: str, key_type: str = "rsa",
         raise err
 
     # Save file
-    util.make_or_verify_dir(key_dir, 0o700, strict_permissions)
-    key_f, key_path = util.unique_file(
-        os.path.join(key_dir, keyname), 0o600, "wb")
-    with key_f:
-        key_f.write(key_pem)
-    if key_type == 'rsa':
-        logger.debug("Generating RSA key (%d bits): %s", key_size, key_path)
-    else:
-        logger.debug("Generating ECDSA key (%d bits): %s", key_size, key_path)
+    key_path = None
+    if key_dir:
+        util.make_or_verify_dir(key_dir, 0o700, strict_permissions)
+        key_f, key_path = util.unique_file(
+            os.path.join(key_dir, keyname), 0o600, "wb")
+        with key_f:
+            key_f.write(key_pem)
+        if key_type == 'rsa':
+            logger.debug("Generating RSA key (%d bits): %s", key_size, key_path)
+        else:
+            logger.debug("Generating ECDSA key (%d bits): %s", key_size, key_path)
 
     return util.Key(key_path, key_pem)
 
 
-def generate_csr(privkey: util.Key, names: Union[List[str], Set[str]], path: str,
+def generate_csr(privkey: util.Key, names: Union[List[str], Set[str]], path: Optional[str],
                  must_staple: bool = False, strict_permissions: bool = True) -> util.CSR:
     """Initialize a CSR with the given private key.
 
     :param privkey: Key to include in the CSR
     :type privkey: :class:`certbot.util.Key`
     :param set names: `str` names to include in the CSR
-    :param str path: Certificate save directory.
+    :param str path: Optional certificate save directory.
     :param bool must_staple: If true, include the TLS Feature extension "OCSP Must-Staple"
     :param bool strict_permissions: If true and path exists, an exception is raised if
         the directory doesn't have 0755 permissions or isn't owned by the current user.
@@ -117,13 +119,15 @@ def generate_csr(privkey: util.Key, names: Union[List[str], Set[str]], path: str
     csr_pem = acme_crypto_util.make_csr(
         privkey.pem, names, must_staple=must_staple)
 
-    # Save CSR
-    util.make_or_verify_dir(path, 0o755, strict_permissions)
-    csr_f, csr_filename = util.unique_file(
-        os.path.join(path, "csr-certbot.pem"), 0o644, "wb")
-    with csr_f:
-        csr_f.write(csr_pem)
-    logger.debug("Creating CSR: %s", csr_filename)
+    # Save CSR, if requested
+    csr_filename = None
+    if path:
+        util.make_or_verify_dir(path, 0o755, strict_permissions)
+        csr_f, csr_filename = util.unique_file(
+            os.path.join(path, "csr-certbot.pem"), 0o644, "wb")
+        with csr_f:
+            csr_f.write(csr_pem)
+        logger.debug("Creating CSR: %s", csr_filename)
 
     return util.CSR(csr_filename, csr_pem, "pem")
 
@@ -254,10 +258,10 @@ def make_key(bits: int = 1024, key_type: str = "rsa",
     return crypto.dump_privatekey(crypto.FILETYPE_PEM, key)
 
 
-def valid_privkey(privkey: str) -> bool:
+def valid_privkey(privkey: Union[str, bytes]) -> bool:
     """Is valid RSA private key?
 
-    :param str privkey: Private key file contents in PEM
+    :param privkey: Private key file contents in PEM
 
     :returns: Validity of private key.
     :rtype: bool

certbot/certbot/util.py

@@ -1,7 +1,6 @@
 """Utilities for all Certbot."""
 import argparse
 import atexit
-import collections
 import errno
 import logging
 import platform
@@ -14,6 +13,7 @@ from typing import Callable
 from typing import Dict
 from typing import IO
 from typing import List
+from typing import NamedTuple
 from typing import Optional
 from typing import Set
 from typing import Tuple
@@ -35,9 +35,18 @@ if _USE_DISTRO:
 logger = logging.getLogger(__name__)
 
 
-Key = collections.namedtuple("Key", "file pem")
-# Note: form is the type of data, "pem" or "der"
-CSR = collections.namedtuple("CSR", "file data form")
+class Key(NamedTuple):
+    """Container for an optional file path and contents for a PEM-formated private key."""
+    file: Optional[str]
+    pem: bytes
+
+
+class CSR(NamedTuple):
+    """Container for an optional file path and contents for a PEM or DER-formatted CSR."""
+    file: Optional[str]
+    data: bytes
+    # Note: form is the type of data, "pem" or "der"
+    form: str
 
 
 # ANSI SGR escape codes

certbot/docs/using.rst

@@ -879,10 +879,6 @@ private key file, you will also need to use ``chgrp`` and ``chmod
 0640`` to allow the server to read
 ``/etc/letsencrypt/live/$domain/privkey.pem``.
 
-.. note:: ``/etc/letsencrypt/archive`` and ``/etc/letsencrypt/keys``
-   contain all previous keys and certificates, while
-   ``/etc/letsencrypt/live`` symlinks to the latest versions.
-
 The following files are available:
 
 ``privkey.pem``

certbot/tests/client_test.py

@@ -373,19 +373,18 @@ class ClientTest(ClientTestCommon):
 
         mock_crypto_util.generate_key.assert_called_once_with(
             key_size=self.config.rsa_key_size,
-            key_dir=self.config.key_dir,
+            key_dir=None,
             key_type=self.config.key_type,
             elliptic_curve="secp256r1",
             strict_permissions=True,
         )
         mock_crypto_util.generate_csr.assert_called_once_with(
-            mock.sentinel.key, self.eg_domains, self.config.csr_dir, False, True)
+            mock.sentinel.key, self.eg_domains, None, False, True)
         mock_crypto_util.cert_and_chain_from_fullchain.assert_called_once_with(
             self.eg_order.fullchain_pem)
 
     @mock.patch("certbot._internal.client.crypto_util")
-    @mock.patch("certbot.compat.os.remove")
-    def test_obtain_certificate_partial_success(self, mock_remove, mock_crypto_util):
+    def test_obtain_certificate_partial_success(self, mock_crypto_util):
         csr = util.CSR(form="pem", file=mock.sentinel.csr_file, data=CSR_SAN)
         key = util.CSR(form="pem", file=mock.sentinel.key_file, data=CSR_SAN)
         mock_crypto_util.generate_csr.return_value = csr
@@ -398,12 +397,10 @@ class ClientTest(ClientTestCommon):
 
         self.assertEqual(mock_crypto_util.generate_key.call_count, 2)
         self.assertEqual(mock_crypto_util.generate_csr.call_count, 2)
-        self.assertEqual(mock_remove.call_count, 2)
         self.assertEqual(mock_crypto_util.cert_and_chain_from_fullchain.call_count, 1)
 
     @mock.patch("certbot._internal.client.crypto_util")
-    @mock.patch("certbot.compat.os.remove")
-    def test_obtain_certificate_finalize_order_partial_success(self, mock_remove, mock_crypto_util):
+    def test_obtain_certificate_finalize_order_partial_success(self, mock_crypto_util):
         from acme import messages
         csr = util.CSR(form="pem", file=mock.sentinel.csr_file, data=CSR_SAN)
         key = util.CSR(form="pem", file=mock.sentinel.key_file, data=CSR_SAN)
@@ -435,9 +432,8 @@ class ClientTest(ClientTestCommon):
         successful_domains = [d for d in self.eg_domains if d != 'example.com']
         self.assertEqual(mock_crypto_util.generate_key.call_count, 2)
         mock_crypto_util.generate_csr.assert_has_calls([
-            mock.call(key, self.eg_domains, self.config.csr_dir, self.config.must_staple, self.config.strict_permissions),
-            mock.call(key, successful_domains, self.config.csr_dir, self.config.must_staple, self.config.strict_permissions)])
-        self.assertEqual(mock_remove.call_count, 2)
+            mock.call(key, self.eg_domains, None, self.config.must_staple, self.config.strict_permissions),
+            mock.call(key, successful_domains, None, self.config.must_staple, self.config.strict_permissions)])
         self.assertEqual(mock_crypto_util.cert_and_chain_from_fullchain.call_count, 1)
 
     @mock.patch("certbot._internal.client.crypto_util")
@@ -496,8 +492,7 @@ class ClientTest(ClientTestCommon):
         self.assertEqual(mock_crypto_util.cert_and_chain_from_fullchain.call_count, 0)
 
     @mock.patch("certbot._internal.client.crypto_util")
-    @mock.patch("certbot.compat.os.remove")
-    def test_obtain_certificate_get_order_partial_success(self, mock_remove, mock_crypto_util):
+    def test_obtain_certificate_get_order_partial_success(self, mock_crypto_util):
         from acme import messages
         csr = util.CSR(form="pem", file=mock.sentinel.csr_file, data=CSR_SAN)
         key = util.CSR(form="pem", file=mock.sentinel.key_file, data=CSR_SAN)
@@ -529,9 +524,8 @@ class ClientTest(ClientTestCommon):
         successful_domains = [d for d in self.eg_domains if d != 'example.com']
         self.assertEqual(mock_crypto_util.generate_key.call_count, 2)
         mock_crypto_util.generate_csr.assert_has_calls([
-            mock.call(key, self.eg_domains, self.config.csr_dir, self.config.must_staple, self.config.strict_permissions),
-            mock.call(key, successful_domains, self.config.csr_dir, self.config.must_staple, self.config.strict_permissions)])
-        self.assertEqual(mock_remove.call_count, 2)
+            mock.call(key, self.eg_domains, None, self.config.must_staple, self.config.strict_permissions),
+            mock.call(key, successful_domains, None, self.config.must_staple, self.config.strict_permissions)])
         self.assertEqual(mock_crypto_util.cert_and_chain_from_fullchain.call_count, 1)
 
     @mock.patch("certbot._internal.client.crypto_util")

certbot/tests/configuration_test.py

@@ -1,6 +1,7 @@
 """Tests for certbot.configuration."""
 import unittest
 from unittest import mock
+import warnings
 
 from certbot import errors
 from certbot._internal import constants
@@ -55,15 +56,17 @@ class NamespaceConfigTest(test_util.ConfigTestCase):
         self.assertEqual(
             os.path.normpath(self.config.backup_dir),
             os.path.normpath(os.path.join(self.config.work_dir, 'backups')))
-        self.assertEqual(
-            os.path.normpath(self.config.csr_dir),
-            os.path.normpath(os.path.join(self.config.config_dir, 'csr')))
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore", DeprecationWarning)
+            self.assertEqual(
+                os.path.normpath(self.config.csr_dir),
+                os.path.normpath(os.path.join(self.config.config_dir, 'csr')))
+            self.assertEqual(
+                os.path.normpath(self.config.key_dir),
+                os.path.normpath(os.path.join(self.config.config_dir, 'keys')))
         self.assertEqual(
             os.path.normpath(self.config.in_progress_dir),
             os.path.normpath(os.path.join(self.config.work_dir, '../p')))
-        self.assertEqual(
-            os.path.normpath(self.config.key_dir),
-            os.path.normpath(os.path.join(self.config.config_dir, 'keys')))
         self.assertEqual(
             os.path.normpath(self.config.temp_checkpoint_dir),
             os.path.normpath(os.path.join(self.config.work_dir, 't')))
@@ -97,9 +100,11 @@ class NamespaceConfigTest(test_util.ConfigTestCase):
                          os.path.join(os.getcwd(), logs_base))
         self.assertTrue(os.path.isabs(config.accounts_dir))
         self.assertTrue(os.path.isabs(config.backup_dir))
-        self.assertTrue(os.path.isabs(config.csr_dir))
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore", DeprecationWarning)
+            self.assertTrue(os.path.isabs(config.csr_dir))
+            self.assertTrue(os.path.isabs(config.key_dir))
         self.assertTrue(os.path.isabs(config.in_progress_dir))
-        self.assertTrue(os.path.isabs(config.key_dir))
         self.assertTrue(os.path.isabs(config.temp_checkpoint_dir))
 
     @mock.patch('certbot.configuration.constants')

certbot/tests/storage_test.py

@@ -839,6 +839,23 @@ class RenewableCertTests(BaseRenewableCertTest):
         storage.RenewableCert(self.config_file.filename, self.config,
             update_symlinks=True)
 
+    def test_truncate(self):
+        # It should not do anything when there's less than 5 cert history
+        for kind in ALL_FOUR:
+            self._write_out_kind(kind, 1)
+        with mock.patch('certbot.compat.os.unlink') as mock_unlink:
+            self.test_rc.truncate()
+            mock_unlink.assert_not_called()
+
+        # It should truncate the excess when there's more than 5 cert history
+        for kind in ALL_FOUR:
+            for i in range(2, 8):
+                self._write_out_kind(kind, i)
+        with mock.patch('certbot.compat.os.unlink') as mock_unlink:
+            self.test_rc.truncate()
+            self.assertEqual(mock_unlink.call_count, 1 * len(ALL_FOUR))
+            self.assertIn("1.pem", mock_unlink.call_args_list[0][0][0])
+
 class DeleteFilesTest(BaseRenewableCertTest):
     """Tests for certbot._internal.storage.delete_files"""
     def setUp(self):