diff --git a/certbot/cert_manager.py b/certbot/cert_manager.py index 0f6f4c730..d8752554a 100644 --- a/certbot/cert_manager.py +++ b/certbot/cert_manager.py @@ -6,10 +6,8 @@ import pytz import traceback import zope.component -from certbot import configuration from certbot import errors from certbot import interfaces -from certbot import renewal from certbot import storage from certbot import util @@ -17,6 +15,10 @@ from certbot.display import util as display_util logger = logging.getLogger(__name__) +################### +# Commands +################### + def update_live_symlinks(config): """Update the certificate file family symlinks to use archive_dir. @@ -26,36 +28,22 @@ def update_live_symlinks(config): .. note:: This assumes that the installation is using a Reverter object. :param config: Configuration. - :type config: :class:`certbot.interfaces.IConfig` + :type config: :class:`certbot.configuration.NamespaceConfig` """ - renewer_config = configuration.RenewerConfiguration(config) - for renewal_file in renewal.renewal_conf_files(renewer_config): - storage.RenewableCert(renewal_file, - configuration.RenewerConfiguration(renewer_config), - update_symlinks=True) + for renewal_file in storage.renewal_conf_files(config): + storage.RenewableCert(renewal_file, config, update_symlinks=True) def rename_lineage(config): """Rename the specified lineage to the new name. :param config: Configuration. - :type config: :class:`certbot.interfaces.IConfig` + :type config: :class:`certbot.configuration.NamespaceConfig` """ disp = zope.component.getUtility(interfaces.IDisplay) - renewer_config = configuration.RenewerConfiguration(config) - certname = config.certname - if not certname: - filenames = renewal.renewal_conf_files(renewer_config) - choices = [storage.lineagename_for_filename(name) for name in filenames] - if not choices: - raise errors.Error("No existing certificates found.") - code, index = disp.menu("Which certificate would you like to rename?", - choices, ok_label="Select", flag="--cert-name") - if code != display_util.OK or not index in range(0, len(choices)): - raise errors.Error("User ended interaction.") - certname = choices[index] + certname = _get_certname(config, "rename") new_certname = config.new_certname if not new_certname: @@ -68,10 +56,110 @@ def rename_lineage(config): if not lineage: raise errors.ConfigurationError("No existing certificate with name " "{0} found.".format(certname)) - storage.rename_renewal_config(certname, new_certname, renewer_config) + storage.rename_renewal_config(certname, new_certname, config) disp.notification("Successfully renamed {0} to {1}." .format(certname, new_certname), pause=False) +def certificates(config): + """Display information about certs configured with Certbot + + :param config: Configuration. + :type config: :class:`certbot.configuration.NamespaceConfig` + """ + parsed_certs = [] + parse_failures = [] + for renewal_file in storage.renewal_conf_files(config): + try: + renewal_candidate = storage.RenewableCert(renewal_file, config) + parsed_certs.append(renewal_candidate) + except Exception as e: # pylint: disable=broad-except + logger.warning("Renewal configuration file %s produced an " + "unexpected error: %s. Skipping.", renewal_file, e) + logger.debug("Traceback was:\n%s", traceback.format_exc()) + parse_failures.append(renewal_file) + + # Describe all the certs + _describe_certs(parsed_certs, parse_failures) + +def delete(config): + """Delete Certbot files associated with a certificate lineage.""" + certname = _get_certname(config, "delete") + storage.delete_files(config, certname) + disp = zope.component.getUtility(interfaces.IDisplay) + disp.notification("Deleted all files relating to certificate {0}." + .format(certname), pause=False) + +################### +# Public Helpers +################### + +def lineage_for_certname(config, certname): + """Find a lineage object with name certname.""" + def update_cert_for_name_match(candidate_lineage, rv): + """Return cert if it has name certname, else return rv + """ + matching_lineage_name_cert = rv + if candidate_lineage.lineagename == certname: + matching_lineage_name_cert = candidate_lineage + return matching_lineage_name_cert + return _search_lineages(config, update_cert_for_name_match, None) + +def domains_for_certname(config, certname): + """Find the domains in the cert with name certname.""" + def update_domains_for_name_match(candidate_lineage, rv): + """Return domains if certname matches, else return rv + """ + matching_domains = rv + if candidate_lineage.lineagename == certname: + matching_domains = candidate_lineage.names() + return matching_domains + return _search_lineages(config, update_domains_for_name_match, None) + +def find_duplicative_certs(config, domains): + """Find existing certs that duplicate the request.""" + def update_certs_for_domain_matches(candidate_lineage, rv): + """Return cert as identical_names_cert if it matches, + or subset_names_cert if it matches as subset + """ + # TODO: Handle these differently depending on whether they are + # expired or still valid? + identical_names_cert, subset_names_cert = rv + candidate_names = set(candidate_lineage.names()) + if candidate_names == set(domains): + identical_names_cert = candidate_lineage + elif candidate_names.issubset(set(domains)): + # This logic finds and returns the largest subset-names cert + # in the case where there are several available. + if subset_names_cert is None: + subset_names_cert = candidate_lineage + elif len(candidate_names) > len(subset_names_cert.names()): + subset_names_cert = candidate_lineage + return (identical_names_cert, subset_names_cert) + + return _search_lineages(config, update_certs_for_domain_matches, (None, None)) + + +################### +# Private Helpers +################### + +def _get_certname(config, verb): + """Get certname from flag, interactively, or error out. + """ + certname = config.certname + if not certname: + disp = zope.component.getUtility(interfaces.IDisplay) + filenames = storage.renewal_conf_files(config) + choices = [storage.lineagename_for_filename(name) for name in filenames] + if not choices: + raise errors.Error("No existing certificates found.") + code, index = disp.menu("Which certificate would you like to {0}?".format(verb), + choices, ok_label="Select", flag="--cert-name") + if code != display_util.OK or not index in range(0, len(choices)): + raise errors.Error("User ended interaction.") + certname = choices[index] + return certname + def _report_lines(msgs): """Format a results report for a category of single-line renewal outcomes""" return " " + "\n ".join(str(msg) for msg in msgs) @@ -126,42 +214,18 @@ def _describe_certs(parsed_certs, parse_failures): disp = zope.component.getUtility(interfaces.IDisplay) disp.notification("\n".join(out), pause=False, wrap=False) -def certificates(config): - """Display information about certs configured with Certbot - - :param config: Configuration. - :type config: :class:`certbot.interfaces.IConfig` - """ - renewer_config = configuration.RenewerConfiguration(config) - parsed_certs = [] - parse_failures = [] - for renewal_file in renewal.renewal_conf_files(renewer_config): - try: - renewal_candidate = storage.RenewableCert(renewal_file, - configuration.RenewerConfiguration(config)) - parsed_certs.append(renewal_candidate) - except Exception as e: # pylint: disable=broad-except - logger.warning("Renewal configuration file %s produced an " - "unexpected error: %s. Skipping.", renewal_file, e) - logger.debug("Traceback was:\n%s", traceback.format_exc()) - parse_failures.append(renewal_file) - - # Describe all the certs - _describe_certs(parsed_certs, parse_failures) - -def _search_lineages(config, func, initial_rv): +def _search_lineages(cli_config, func, initial_rv): """Iterate func over unbroken lineages, allowing custom return conditions. Allows flexible customization of return values, including multiple return values and complex checks. """ - cli_config = configuration.RenewerConfiguration(config) configs_dir = cli_config.renewal_configs_dir # Verify the directory is there util.make_or_verify_dir(configs_dir, mode=0o755, uid=os.geteuid()) rv = initial_rv - for renewal_file in renewal.renewal_conf_files(cli_config): + for renewal_file in storage.renewal_conf_files(cli_config): try: candidate_lineage = storage.RenewableCert(renewal_file, cli_config) except (errors.CertStorageError, IOError): @@ -170,48 +234,3 @@ def _search_lineages(config, func, initial_rv): continue rv = func(candidate_lineage, rv) return rv - -def lineage_for_certname(config, certname): - """Find a lineage object with name certname.""" - def update_cert_for_name_match(candidate_lineage, rv): - """Return cert if it has name certname, else return rv - """ - matching_lineage_name_cert = rv - if candidate_lineage.lineagename == certname: - matching_lineage_name_cert = candidate_lineage - return matching_lineage_name_cert - return _search_lineages(config, update_cert_for_name_match, None) - -def domains_for_certname(config, certname): - """Find the domains in the cert with name certname.""" - def update_domains_for_name_match(candidate_lineage, rv): - """Return domains if certname matches, else return rv - """ - matching_domains = rv - if candidate_lineage.lineagename == certname: - matching_domains = candidate_lineage.names() - return matching_domains - return _search_lineages(config, update_domains_for_name_match, None) - -def find_duplicative_certs(config, domains): - """Find existing certs that duplicate the request.""" - def update_certs_for_domain_matches(candidate_lineage, rv): - """Return cert as identical_names_cert if it matches, - or subset_names_cert if it matches as subset - """ - # TODO: Handle these differently depending on whether they are - # expired or still valid? - identical_names_cert, subset_names_cert = rv - candidate_names = set(candidate_lineage.names()) - if candidate_names == set(domains): - identical_names_cert = candidate_lineage - elif candidate_names.issubset(set(domains)): - # This logic finds and returns the largest subset-names cert - # in the case where there are several available. - if subset_names_cert is None: - subset_names_cert = candidate_lineage - elif len(candidate_names) > len(subset_names_cert.names()): - subset_names_cert = candidate_lineage - return (identical_names_cert, subset_names_cert) - - return _search_lineages(config, update_certs_for_domain_matches, (None, None)) diff --git a/certbot/cli.py b/certbot/cli.py index ef6257304..89c171829 100644 --- a/certbot/cli.py +++ b/certbot/cli.py @@ -82,6 +82,7 @@ manage certificates: certificates Display information about certs you have from Certbot revoke Revoke a certificate (supply --cert-path) rename Rename a certificate + delete Delete a certificate manage your account with Let's Encrypt: register Create a Let's Encrypt ACME account @@ -352,13 +353,17 @@ VERB_HELP = [ "short": "List all certificates managed by Certbot", "opts": "List all certificates managed by Certbot" }), + ("delete", { + "short": "Clean up all files related to a certificate", + "opts": "Options for deleting a certificate" + }), ("revoke", { "short": "Revoke a certificate specified with --cert-path", "opts": "Options for revocation of certs" }), ("rename", { "short": "Change a certificate's name (for management purposes)", - "opts": "Options changing certificate names" + "opts": "Options for changing certificate names" }), ("register", { "short": "Register for account with Let's Encrypt / other ACME server", @@ -410,7 +415,8 @@ class HelpfulArgumentParser(object): "register": main.register, "renew": main.renew, "revoke": main.revoke, "rollback": main.rollback, "everything": main.run, "update_symlinks": main.update_symlinks, - "certificates": main.certificates, "rename": main.rename} + "certificates": main.certificates, "rename": main.rename, + "delete": main.delete} # List of topics for which additional help can be provided HELP_TOPICS = ["all", "security", "paths", "automation", "testing"] + list(self.VERBS) @@ -763,7 +769,7 @@ def _add_all_groups(helpful): helpful.add_group("paths", description="Arguments changing execution paths & servers") helpful.add_group("manage", description="Various subcommands and flags are available for managing your certificates:", - verbs=["certificates", "renew", "revoke", "rename"]) + verbs=["certificates", "delete", "renew", "revoke", "rename"]) # VERBS for verb, docs in VERB_HELP: @@ -810,13 +816,12 @@ def prepare_and_parse_args(plugins, args, detect_defaults=False): # pylint: dis "multiple -d flags or enter a comma separated list of domains " "as a parameter. (default: Ask)") helpful.add( - [None, "run", "certonly", "manage"], + [None, "run", "certonly", "manage", "rename", "delete"], "--cert-name", dest="certname", metavar="CERTNAME", default=None, help="Certificate name to apply. Only one certificate name can be used " "per Certbot run. To see certificate names, run 'certbot certificates'. " - "If there is no existing certificate with this name and " - "domains are requested, create a new certificate with this name.") + "When creating a new certificate, specifies the new certificate's name.") helpful.add( ["rename", "manage"], "--updated-cert-name", dest="new_certname", diff --git a/certbot/client.py b/certbot/client.py index 6ff14bc56..cfd2b8487 100644 --- a/certbot/client.py +++ b/certbot/client.py @@ -15,7 +15,6 @@ import certbot from certbot import account from certbot import auth_handler -from certbot import configuration from certbot import constants from certbot import crypto_util from certbot import errors @@ -307,7 +306,7 @@ class Client(object): new_name, OpenSSL.crypto.dump_certificate( OpenSSL.crypto.FILETYPE_PEM, certr.body.wrapped), key.pem, crypto_util.dump_pyopenssl_chain(chain), - configuration.RenewerConfiguration(self.config.namespace)) + self.config) def save_certificate(self, certr, chain_cert, cert_path, chain_path, fullchain_path): diff --git a/certbot/configuration.py b/certbot/configuration.py index 1d4243272..d25378922 100644 --- a/certbot/configuration.py +++ b/certbot/configuration.py @@ -25,9 +25,16 @@ class NamespaceConfig(object): - `csr_dir` - `in_progress_dir` - `key_dir` - - `renewer_config_file` - `temp_checkpoint_dir` + And the following paths are dynamically resolved using + :attr:`~certbot.interfaces.IConfig.config_dir` and relative + paths defined in :py:mod:`certbot.constants`: + + - `default_archive_dir` + - `live_dir` + - `renewal_configs_dir` + :ivar namespace: Namespace typically produced by :meth:`argparse.ArgumentParser.parse_args`. :type namespace: :class:`argparse.Namespace` @@ -85,16 +92,6 @@ class NamespaceConfig(object): new_ns = copy.deepcopy(self.namespace) return type(self)(new_ns) - -class RenewerConfiguration(object): - """Configuration wrapper for renewer.""" - - def __init__(self, namespace): - self.namespace = namespace - - def __getattr__(self, name): - return getattr(self.namespace, name) - @property def default_archive_dir(self): # pylint: disable=missing-docstring return os.path.join(self.namespace.config_dir, constants.ARCHIVE_DIR) @@ -108,11 +105,6 @@ class RenewerConfiguration(object): return os.path.join( self.namespace.config_dir, constants.RENEWAL_CONFIGS_DIR) - @property - def renewer_config_file(self): # pylint: disable=missing-docstring - return os.path.join( - self.namespace.config_dir, constants.RENEWER_CONFIG_FILENAME) - def check_config_sanity(config): """Validate command line options and display error message if diff --git a/certbot/constants.py b/certbot/constants.py index 117301380..2e74cfd86 100644 --- a/certbot/constants.py +++ b/certbot/constants.py @@ -92,6 +92,3 @@ TEMP_CHECKPOINT_DIR = "temp_checkpoint" RENEWAL_CONFIGS_DIR = "renewal" """Renewal configs directory, relative to `IConfig.config_dir`.""" - -RENEWER_CONFIG_FILENAME = "renewer.conf" -"""Renewer config file name (relative to `IConfig.config_dir`).""" diff --git a/certbot/interfaces.py b/certbot/interfaces.py index 6388bf936..f2b3faf21 100644 --- a/certbot/interfaces.py +++ b/certbot/interfaces.py @@ -223,9 +223,6 @@ class IConfig(zope.interface.Interface): temp_checkpoint_dir = zope.interface.Attribute( "Temporary checkpoint directory.") - renewer_config_file = zope.interface.Attribute( - "Location of renewal configuration file.") - no_verify_ssl = zope.interface.Attribute( "Disable verification of the ACME server's certificate.") tls_sni_01_port = zope.interface.Attribute( diff --git a/certbot/main.py b/certbot/main.py index 24f38172a..6a1653193 100644 --- a/certbot/main.py +++ b/certbot/main.py @@ -511,6 +511,14 @@ def rename(config, unused_plugins): """ cert_manager.rename_lineage(config) +def delete(config, unused_plugins): + """Delete a certificate + + Use the information in the config file to delete an existing + lineage. + """ + cert_manager.delete(config) + def certificates(config, unused_plugins): """Display information about certs configured with Certbot """ diff --git a/certbot/renewal.py b/certbot/renewal.py index a057a63a9..c9eca9a3a 100644 --- a/certbot/renewal.py +++ b/certbot/renewal.py @@ -1,7 +1,6 @@ """Functionality for autorenewal and associated juggling of configurations""" from __future__ import print_function import copy -import glob import logging import os import traceback @@ -11,7 +10,6 @@ import zope.component import OpenSSL -from certbot import configuration from certbot import cli from certbot import crypto_util @@ -34,17 +32,6 @@ STR_CONFIG_ITEMS = ["config_dir", "logs_dir", "work_dir", "user_agent", INT_CONFIG_ITEMS = ["rsa_key_size", "tls_sni_01_port", "http01_port"] -def renewal_conf_files(config): - """Return /path/to/*.conf in the renewal conf directory""" - return glob.glob(os.path.join(config.renewal_configs_dir, "*.conf")) - -def renewal_file_for_certname(config, certname): - """Return /path/to/certname.conf in the renewal conf directory""" - path = os.path.join(config.renewal_configs_dir, "{0}.conf".format(certname)) - if not os.path.exists(path): - raise errors.CertStorageError("No certificate found with name {0}.".format(certname)) - return path - def _reconstitute(config, full_path): """Try to instantiate a RenewableCert, updating config with relevant items. @@ -63,8 +50,7 @@ def _reconstitute(config, full_path): """ try: - renewal_candidate = storage.RenewableCert( - full_path, configuration.RenewerConfiguration(config)) + renewal_candidate = storage.RenewableCert(full_path, config) except (errors.CertStorageError, IOError) as exc: logger.warning(exc) logger.warning("Renewal configuration file %s is broken. Skipping.", full_path) @@ -246,9 +232,8 @@ def renew_cert(config, le_client, lineage): new_cert = OpenSSL.crypto.dump_certificate( OpenSSL.crypto.FILETYPE_PEM, new_certr.body.wrapped) new_chain = crypto_util.dump_pyopenssl_chain(new_chain) - renewal_conf = configuration.RenewerConfiguration(config.namespace) # TODO: Check return value of save_successor - lineage.save_successor(prior_version, new_cert, new_key.pem, new_chain, renewal_conf) + lineage.save_successor(prior_version, new_cert, new_key.pem, new_chain, config) lineage.update_all_links_to(lineage.latest_common_version()) hooks.renew_hook(config, lineage.names(), lineage.live_dir) @@ -317,12 +302,10 @@ def handle_renewal_request(config): "command. The renew verb may provide other options " "for selecting certificates to renew in the future.") - renewer_config = configuration.RenewerConfiguration(config) - if config.certname: - conf_files = [renewal_file_for_certname(renewer_config, config.certname)] + conf_files = [storage.renewal_file_for_certname(config, config.certname)] else: - conf_files = renewal_conf_files(renewer_config) + conf_files = storage.renewal_conf_files(config) renew_successes = [] renew_failures = [] diff --git a/certbot/storage.py b/certbot/storage.py index 671922bee..e4fc21a85 100644 --- a/certbot/storage.py +++ b/certbot/storage.py @@ -1,5 +1,6 @@ """Renewable certificates storage.""" import datetime +import glob import logging import os import re @@ -7,6 +8,7 @@ import re import configobj import parsedatetime import pytz +import shutil import six import certbot @@ -20,9 +22,22 @@ from certbot import util logger = logging.getLogger(__name__) ALL_FOUR = ("cert", "privkey", "chain", "fullchain") +README = "README" CURRENT_VERSION = util.get_strict_version(certbot.__version__) +def renewal_conf_files(config): + """Return /path/to/*.conf in the renewal conf directory""" + return glob.glob(os.path.join(config.renewal_configs_dir, "*.conf")) + +def renewal_file_for_certname(config, certname): + """Return /path/to/certname.conf in the renewal conf directory""" + path = os.path.join(config.renewal_configs_dir, "{0}.conf".format(certname)) + if not os.path.exists(path): + raise errors.CertStorageError("No certificate found with name {0} (expected " + "{1}).".format(certname, path)) + return path + def config_with_defaults(config=None): """Merge supplied config, if provided, on top of builtin defaults.""" defaults_copy = configobj.ConfigObj(constants.RENEWER_DEFAULTS) @@ -99,13 +114,11 @@ def write_renewal_config(o_filename, n_filename, archive_dir, target, relevant_d def rename_renewal_config(prev_name, new_name, cli_config): """Renames cli_config.certname's config to cli_config.new_certname. - :param .RenewerConfiguration cli_config: parsed command line + :param .NamespaceConfig cli_config: parsed command line arguments """ - prev_filename = os.path.join( - cli_config.renewal_configs_dir, prev_name) + ".conf" - new_filename = os.path.join( - cli_config.renewal_configs_dir, new_name) + ".conf" + prev_filename = renewal_filename_for_lineagename(cli_config, prev_name) + new_filename = renewal_filename_for_lineagename(cli_config, new_name) if os.path.exists(new_filename): raise errors.ConfigurationError("The new certificate name " "is already in use.") @@ -122,15 +135,14 @@ def update_configuration(lineagename, archive_dir, target, cli_config): :param str lineagename: Name of the lineage being modified :param str archive_dir: Absolute path to the archive directory :param dict target: Maps ALL_FOUR to their symlink paths - :param .RenewerConfiguration cli_config: parsed command line + :param .NamespaceConfig cli_config: parsed command line arguments :returns: Configuration object for the updated config file :rtype: configobj.ConfigObj """ - config_filename = os.path.join( - cli_config.renewal_configs_dir, lineagename) + ".conf" + config_filename = renewal_filename_for_lineagename(cli_config, lineagename) temp_filename = config_filename + ".new" # If an existing tempfile exists, delete it @@ -198,10 +210,98 @@ def lineagename_for_filename(config_filename): "renewal config file name must end in .conf") return os.path.basename(config_filename[:-len(".conf")]) +def renewal_filename_for_lineagename(config, lineagename): + """Returns the lineagename for a configuration filename. + """ + return os.path.join(config.renewal_configs_dir, lineagename) + ".conf" + def _relpath_from_file(archive_dir, from_file): """Path to a directory from a file""" return os.path.relpath(archive_dir, os.path.dirname(from_file)) +def _full_archive_path(config_obj, cli_config, lineagename): + """Returns the full archive path for a lineagename + + Uses cli_config to determine archive path if not available from config_obj. + + :param configobj.ConfigObj config_obj: Renewal conf file contents (can be None) + :param configuration.NamespaceConfig cli_config: Main config file + :param str lineagename: Certificate name + """ + if config_obj and "archive_dir" in config_obj: + return config_obj["archive_dir"] + else: + return os.path.join(cli_config.default_archive_dir, lineagename) + +def _full_live_path(cli_config, lineagename): + """Returns the full default live path for a lineagename""" + return os.path.join(cli_config.live_dir, lineagename) + +def delete_files(config, certname): + """Delete all files related to the certificate. + + If some files are not found, ignore them and continue. + """ + renewal_filename = renewal_file_for_certname(config, certname) + # file exists + full_default_archive_dir = _full_archive_path(None, config, certname) + full_default_live_dir = _full_live_path(config, certname) + try: + renewal_config = configobj.ConfigObj(renewal_filename) + except configobj.ConfigObjError: + # config is corrupted + logger.warning("Could not parse %s. You may wish to manually " + "delete the contents of %s and %s.", renewal_filename, + full_default_live_dir, full_default_archive_dir) + raise errors.CertStorageError( + "error parsing {0}".format(renewal_filename)) + finally: + # we couldn't read it, but let's at least delete it + # if this was going to fail, it already would have. + os.remove(renewal_filename) + logger.debug("Removed %s", renewal_filename) + + # cert files and (hopefully) live directory + # it's not guaranteed that the files are in our default storage + # structure. so, first delete the cert files. + directory_names = set() + for kind in ALL_FOUR: + link = renewal_config.get(kind) + try: + os.remove(link) + logger.debug("Removed %s", link) + except OSError: + logger.debug("Unable to delete %s", link) + directory = os.path.dirname(link) + directory_names.add(directory) + + # if all four were in the same directory, and the only thing left + # is the README file (or nothing), delete that directory. + # this will be wrong in very few but some cases. + if len(directory_names) == 1: + # delete the README file + directory = directory_names.pop() + readme_path = os.path.join(directory, README) + try: + os.remove(readme_path) + logger.debug("Removed %s", readme_path) + except OSError: + logger.debug("Unable to delete %s", readme_path) + # if it's now empty, delete the directory + try: + os.rmdir(directory) # only removes empty directories + logger.debug("Removed %s", directory) + except OSError: + logger.debug("Unable to remove %s; may not be empty.", directory) + + # archive directory + try: + archive_path = _full_archive_path(renewal_config, config, certname) + shutil.rmtree(archive_path) + logger.debug("Removed %s", archive_path) + except OSError: + logger.debug("Unable to remove %s", archive_path) + class RenewableCert(object): # pylint: disable=too-many-instance-attributes,too-many-public-methods @@ -244,7 +344,7 @@ class RenewableCert(object): :param str config_filename: the path to the renewal config file that defines this lineage. - :param .RenewerConfiguration: parsed command line arguments + :param .NamespaceConfig: parsed command line arguments :raises .CertStorageError: if the configuration file's name didn't end in ".conf", or the file is missing or broken. @@ -303,11 +403,8 @@ class RenewableCert(object): @property def archive_dir(self): """Returns the default or specified archive directory""" - if "archive_dir" in self.configuration: - return self.configuration["archive_dir"] - else: - return os.path.join( - self.cli_config.default_archive_dir, self.lineagename) + return _full_archive_path(self.configuration, + self.cli_config, self.lineagename) def relative_archive_dir(self, from_file): """Returns the default or specified archive directory as a relative path @@ -827,7 +924,7 @@ class RenewableCert(object): :param str cert: the initial certificate version in PEM format :param str privkey: the private key in PEM format :param str chain: the certificate chain in PEM format - :param .RenewerConfiguration cli_config: parsed command line + :param .NamespaceConfig cli_config: parsed command line arguments :returns: the newly-created RenewalCert object @@ -843,16 +940,13 @@ class RenewableCert(object): logger.debug("Creating directory %s.", i) config_file, config_filename = util.unique_lineage_name( cli_config.renewal_configs_dir, lineagename) - if not config_filename.endswith(".conf"): - raise errors.CertStorageError( - "renewal config file name must end in .conf") # Determine where on disk everything will go # lineagename will now potentially be modified based on which # renewal configuration file could actually be created - lineagename = os.path.basename(config_filename)[:-len(".conf")] - archive = os.path.join(cli_config.default_archive_dir, lineagename) - live_dir = os.path.join(cli_config.live_dir, lineagename) + lineagename = lineagename_for_filename(config_filename) + archive = _full_archive_path(None, cli_config, lineagename) + live_dir = _full_live_path(cli_config, lineagename) if os.path.exists(archive): raise errors.CertStorageError( "archive directory exists for " + lineagename) @@ -887,7 +981,7 @@ class RenewableCert(object): f.write(cert + chain) # Write a README file to the live directory - readme_path = os.path.join(live_dir, "README") + readme_path = os.path.join(live_dir, README) with open(readme_path, "w") as f: logger.debug("Writing README to %s.", readme_path) f.write("This directory contains your keys and certificates.\n\n" @@ -928,7 +1022,7 @@ class RenewableCert(object): :param str new_privkey: the new private key, in PEM format, or ``None``, if the private key has not changed :param str new_chain: the new chain, in PEM format - :param .RenewerConfiguration cli_config: parsed command line + :param .NamespaceConfig cli_config: parsed command line arguments :returns: the new version number that was created diff --git a/certbot/tests/cert_manager_test.py b/certbot/tests/cert_manager_test.py index e04e25da8..ae673fba1 100644 --- a/certbot/tests/cert_manager_test.py +++ b/certbot/tests/cert_manager_test.py @@ -25,12 +25,12 @@ class BaseCertManagerTest(unittest.TestCase): os.makedirs(os.path.join(self.tempdir, "renewal")) - self.cli_config = mock.MagicMock( + self.cli_config = configuration.NamespaceConfig(mock.MagicMock( config_dir=self.tempdir, work_dir=self.tempdir, logs_dir=self.tempdir, quiet=False, - ) + )) self.domains = { "example.org": None, @@ -47,7 +47,7 @@ class BaseCertManagerTest(unittest.TestCase): junk.close() def _set_up_config(self, domain, custom_archive): - # TODO: maybe provide RenewerConfiguration.make_dirs? + # TODO: maybe provide NamespaceConfig.make_dirs? # TODO: main() should create those dirs, c.f. #902 os.makedirs(os.path.join(self.tempdir, "live", domain)) config = configobj.ConfigObj() @@ -111,6 +111,21 @@ class UpdateLiveSymlinksTest(BaseCertManagerTest): os.chdir(prev_dir) +class DeleteTest(storage_test.BaseRenewableCertTest): + """Tests for certbot.cert_manager.delete + """ + @mock.patch('zope.component.getUtility') + @mock.patch('certbot.cert_manager.lineage_for_certname') + @mock.patch('certbot.storage.delete_files') + def test_delete(self, mock_delete_files, mock_lineage_for_certname, unused_get_utility): + """Test delete""" + mock_lineage_for_certname.return_value = self.test_rc + self.cli_config.certname = "example.org" + from certbot import cert_manager + cert_manager.delete(self.cli_config) + self.assertTrue(mock_delete_files.called) + + class CertificatesTest(BaseCertManagerTest): """Tests for certbot.cert_manager.certificates """ @@ -151,12 +166,12 @@ class CertificatesTest(BaseCertManagerTest): def test_certificates_no_files(self, mock_utility, mock_logger): tempdir = tempfile.mkdtemp() - cli_config = mock.MagicMock( + cli_config = configuration.NamespaceConfig(mock.MagicMock( config_dir=tempdir, work_dir=tempdir, logs_dir=tempdir, quiet=False, - ) + )) os.makedirs(os.path.join(tempdir, "renewal")) self._certificates(cli_config) @@ -202,90 +217,85 @@ class CertificatesTest(BaseCertManagerTest): self.assertTrue('INVALID: TEST CERT' in out) -class SearchLineagesTest(unittest.TestCase): +class SearchLineagesTest(BaseCertManagerTest): """Tests for certbot.cert_manager._search_lineages.""" - @mock.patch('certbot.configuration.RenewerConfiguration') @mock.patch('certbot.util.make_or_verify_dir') - @mock.patch('certbot.renewal.renewal_conf_files') + @mock.patch('certbot.storage.renewal_conf_files') @mock.patch('certbot.storage.RenewableCert') def test_cert_storage_error(self, mock_renewable_cert, mock_renewal_conf_files, - mock_make_or_verify_dir, mock_renewer_config): + mock_make_or_verify_dir): mock_renewal_conf_files.return_value = ["badfile"] mock_renewable_cert.side_effect = errors.CertStorageError from certbot import cert_manager # pylint: disable=protected-access - self.assertEqual(cert_manager._search_lineages(None, lambda x: x, "check"), "check") + self.assertEqual(cert_manager._search_lineages(self.cli_config, lambda x: x, "check"), + "check") self.assertTrue(mock_make_or_verify_dir.called) - self.assertTrue(mock_renewer_config) -class LineageForCertnameTest(unittest.TestCase): +class LineageForCertnameTest(BaseCertManagerTest): """Tests for certbot.cert_manager.lineage_for_certname""" - @mock.patch('certbot.configuration.RenewerConfiguration') @mock.patch('certbot.util.make_or_verify_dir') - @mock.patch('certbot.renewal.renewal_conf_files') + @mock.patch('certbot.storage.renewal_conf_files') @mock.patch('certbot.storage.RenewableCert') def test_found_match(self, mock_renewable_cert, mock_renewal_conf_files, - mock_make_or_verify_dir, mock_renewer_config): + mock_make_or_verify_dir): mock_renewal_conf_files.return_value = ["somefile.conf"] mock_match = mock.Mock(lineagename="example.com") mock_renewable_cert.return_value = mock_match from certbot import cert_manager - self.assertEqual(cert_manager.lineage_for_certname(None, "example.com"), mock_match) + self.assertEqual(cert_manager.lineage_for_certname(self.cli_config, "example.com"), + mock_match) self.assertTrue(mock_make_or_verify_dir.called) - self.assertTrue(mock_renewer_config) - @mock.patch('certbot.configuration.RenewerConfiguration') @mock.patch('certbot.util.make_or_verify_dir') - @mock.patch('certbot.renewal.renewal_conf_files') + @mock.patch('certbot.storage.renewal_conf_files') @mock.patch('certbot.storage.RenewableCert') def test_no_match(self, mock_renewable_cert, mock_renewal_conf_files, - mock_make_or_verify_dir, mock_renewer_config): + mock_make_or_verify_dir): mock_renewal_conf_files.return_value = ["somefile.conf"] mock_match = mock.Mock(lineagename="other.com") mock_renewable_cert.return_value = mock_match from certbot import cert_manager - self.assertEqual(cert_manager.lineage_for_certname(None, "example.com"), None) + self.assertEqual(cert_manager.lineage_for_certname(self.cli_config, "example.com"), + None) self.assertTrue(mock_make_or_verify_dir.called) - self.assertTrue(mock_renewer_config) -class DomainsForCertnameTest(unittest.TestCase): +class DomainsForCertnameTest(BaseCertManagerTest): """Tests for certbot.cert_manager.domains_for_certname""" - @mock.patch('certbot.configuration.RenewerConfiguration') @mock.patch('certbot.util.make_or_verify_dir') - @mock.patch('certbot.renewal.renewal_conf_files') + @mock.patch('certbot.storage.renewal_conf_files') @mock.patch('certbot.storage.RenewableCert') def test_found_match(self, mock_renewable_cert, mock_renewal_conf_files, - mock_make_or_verify_dir, mock_renewer_config): + mock_make_or_verify_dir): mock_renewal_conf_files.return_value = ["somefile.conf"] mock_match = mock.Mock(lineagename="example.com") domains = ["example.com", "example.org"] mock_match.names.return_value = domains mock_renewable_cert.return_value = mock_match from certbot import cert_manager - self.assertEqual(cert_manager.domains_for_certname(None, "example.com"), domains) + self.assertEqual(cert_manager.domains_for_certname(self.cli_config, "example.com"), + domains) self.assertTrue(mock_make_or_verify_dir.called) - self.assertTrue(mock_renewer_config) - @mock.patch('certbot.configuration.RenewerConfiguration') @mock.patch('certbot.util.make_or_verify_dir') - @mock.patch('certbot.renewal.renewal_conf_files') + @mock.patch('certbot.storage.renewal_conf_files') @mock.patch('certbot.storage.RenewableCert') def test_no_match(self, mock_renewable_cert, mock_renewal_conf_files, - mock_make_or_verify_dir, mock_renewer_config): + mock_make_or_verify_dir): mock_renewal_conf_files.return_value = ["somefile.conf"] mock_match = mock.Mock(lineagename="example.com") domains = ["example.com", "example.org"] mock_match.names.return_value = domains mock_renewable_cert.return_value = mock_match from certbot import cert_manager - self.assertEqual(cert_manager.domains_for_certname(None, "other.com"), None) + self.assertEqual(cert_manager.domains_for_certname(self.cli_config, "other.com"), + None) self.assertTrue(mock_make_or_verify_dir.called) - self.assertTrue(mock_renewer_config) class RenameLineageTest(BaseCertManagerTest): @@ -293,7 +303,7 @@ class RenameLineageTest(BaseCertManagerTest): def setUp(self): super(RenameLineageTest, self).setUp() - self.mock_config = configuration.RenewerConfiguration( + self.mock_config = configuration.NamespaceConfig( namespace=mock.MagicMock( config_dir=self.tempdir, work_dir=self.tempdir, @@ -307,7 +317,7 @@ class RenameLineageTest(BaseCertManagerTest): from certbot import cert_manager return cert_manager.rename_lineage(*args, **kwargs) - @mock.patch('certbot.renewal.renewal_conf_files') + @mock.patch('certbot.storage.renewal_conf_files') @mock.patch('certbot.main.zope.component.getUtility') def test_no_certname(self, mock_get_utility, mock_renewal_conf_files): mock_config = mock.Mock(certname=None, new_certname="two") diff --git a/certbot/tests/configuration_test.py b/certbot/tests/configuration_test.py index 5e59d0b86..183d6a95c 100644 --- a/certbot/tests/configuration_test.py +++ b/certbot/tests/configuration_test.py @@ -88,31 +88,19 @@ class NamespaceConfigTest(unittest.TestCase): self.assertTrue(os.path.isabs(config.key_dir)) self.assertTrue(os.path.isabs(config.temp_checkpoint_dir)) - -class RenewerConfigurationTest(unittest.TestCase): - """Test for certbot.configuration.RenewerConfiguration.""" - - def setUp(self): - self.namespace = mock.MagicMock(config_dir='/tmp/config') - from certbot.configuration import RenewerConfiguration - self.config = RenewerConfiguration(self.namespace) - @mock.patch('certbot.configuration.constants') - def test_dynamic_dirs(self, constants): + def test_renewal_dynamic_dirs(self, constants): constants.ARCHIVE_DIR = 'a' constants.LIVE_DIR = 'l' constants.RENEWAL_CONFIGS_DIR = 'renewal_configs' - constants.RENEWER_CONFIG_FILENAME = 'r.conf' self.assertEqual(self.config.default_archive_dir, '/tmp/config/a') self.assertEqual(self.config.live_dir, '/tmp/config/l') self.assertEqual( self.config.renewal_configs_dir, '/tmp/config/renewal_configs') - self.assertEqual(self.config.renewer_config_file, '/tmp/config/r.conf') - def test_absolute_paths(self): + def test_renewal_absolute_paths(self): from certbot.configuration import NamespaceConfig - from certbot.configuration import RenewerConfiguration config_base = "foo" work_base = "bar" @@ -125,12 +113,11 @@ class RenewerConfigurationTest(unittest.TestCase): mock_namespace.config_dir = config_base mock_namespace.work_dir = work_base mock_namespace.logs_dir = logs_base - config = RenewerConfiguration(NamespaceConfig(mock_namespace)) + config = NamespaceConfig(mock_namespace) self.assertTrue(os.path.isabs(config.default_archive_dir)) self.assertTrue(os.path.isabs(config.live_dir)) self.assertTrue(os.path.isabs(config.renewal_configs_dir)) - self.assertTrue(os.path.isabs(config.renewer_config_file)) if __name__ == '__main__': diff --git a/certbot/tests/main_test.py b/certbot/tests/main_test.py index a0d6cc418..3665f09bb 100644 --- a/certbot/tests/main_test.py +++ b/certbot/tests/main_test.py @@ -567,6 +567,11 @@ class MainTest(unittest.TestCase): # pylint: disable=too-many-public-methods self._call_no_clientmock(['certificates']) self.assertEqual(1, mock_cert_manager.call_count) + @mock.patch('certbot.cert_manager.delete') + def test_delete(self, mock_cert_manager): + self._call_no_clientmock(['delete']) + self.assertEqual(1, mock_cert_manager.call_count) + def test_plugins(self): flags = ['--init', '--prepare', '--authenticators', '--installers'] for args in itertools.chain( @@ -856,8 +861,7 @@ class MainTest(unittest.TestCase): # pylint: disable=too-many-public-methods rc_path = test_util.make_lineage(self, 'sample-renewal-ancient.conf') args = mock.MagicMock(account=None, email=None, webroot_path=None) config = configuration.NamespaceConfig(args) - lineage = storage.RenewableCert(rc_path, - configuration.RenewerConfiguration(config)) + lineage = storage.RenewableCert(rc_path, config) renewalparams = lineage.configuration["renewalparams"] # pylint: disable=protected-access renewal._restore_webroot_config(config, renewalparams) diff --git a/certbot/tests/renewal_test.py b/certbot/tests/renewal_test.py index 207b70041..8155595c2 100644 --- a/certbot/tests/renewal_test.py +++ b/certbot/tests/renewal_test.py @@ -21,8 +21,7 @@ class RenewalTest(unittest.TestCase): rc_path = util.make_lineage(self, 'sample-renewal-ancient.conf') args = mock.MagicMock(account=None, email=None, webroot_path=None) config = configuration.NamespaceConfig(args) - lineage = storage.RenewableCert( - rc_path, configuration.RenewerConfiguration(config)) + lineage = storage.RenewableCert(rc_path, config) renewalparams = lineage.configuration["renewalparams"] # pylint: disable=protected-access from certbot import renewal diff --git a/certbot/tests/storage_test.py b/certbot/tests/storage_test.py index ebe7d2243..a1fda6535 100644 --- a/certbot/tests/storage_test.py +++ b/certbot/tests/storage_test.py @@ -49,7 +49,7 @@ class BaseRenewableCertTest(unittest.TestCase): from certbot import storage self.tempdir = tempfile.mkdtemp() - self.cli_config = configuration.RenewerConfiguration( + self.cli_config = configuration.NamespaceConfig( namespace=mock.MagicMock( config_dir=self.tempdir, work_dir=self.tempdir, @@ -57,16 +57,22 @@ class BaseRenewableCertTest(unittest.TestCase): ) ) - # TODO: maybe provide RenewerConfiguration.make_dirs? + # TODO: maybe provide NamespaceConfig.make_dirs? # TODO: main() should create those dirs, c.f. #902 os.makedirs(os.path.join(self.tempdir, "live", "example.org")) - os.makedirs(os.path.join(self.tempdir, "archive", "example.org")) + archive_path = os.path.join(self.tempdir, "archive", "example.org") + os.makedirs(archive_path) os.makedirs(os.path.join(self.tempdir, "renewal")) config = configobj.ConfigObj() for kind in ALL_FOUR: - config[kind] = os.path.join(self.tempdir, "live", "example.org", + kind_path = os.path.join(self.tempdir, "live", "example.org", kind + ".pem") + config[kind] = kind_path + with open(os.path.join(self.tempdir, "live", "example.org", + "README"), 'a'): + pass + config["archive"] = archive_path config.filename = os.path.join(self.tempdir, "renewal", "example.org.conf") config.write() @@ -770,5 +776,95 @@ class RenewableCertTests(BaseRenewableCertTest): storage.RenewableCert(self.config.filename, self.cli_config, update_symlinks=True) +class DeleteFilesTest(BaseRenewableCertTest): + """Tests for certbot.storage.delete_files""" + def setUp(self): + super(DeleteFilesTest, self).setUp() + for kind in ALL_FOUR: + kind_path = os.path.join(self.tempdir, "live", "example.org", + kind + ".pem") + with open(kind_path, 'a'): + pass + self.config.write() + self.assertTrue(os.path.exists(os.path.join( + self.cli_config.renewal_configs_dir, "example.org.conf"))) + self.assertTrue(os.path.exists(os.path.join( + self.cli_config.live_dir, "example.org"))) + self.assertTrue(os.path.exists(os.path.join( + self.tempdir, "archive", "example.org"))) + + def _call(self): + from certbot import storage + with mock.patch("certbot.storage.logger"): + storage.delete_files(self.cli_config, "example.org") + + def test_delete_all_files(self): + self._call() + + self.assertFalse(os.path.exists(os.path.join( + self.cli_config.renewal_configs_dir, "example.org.conf"))) + self.assertFalse(os.path.exists(os.path.join( + self.cli_config.live_dir, "example.org"))) + self.assertFalse(os.path.exists(os.path.join( + self.tempdir, "archive", "example.org"))) + + def test_bad_renewal_config(self): + with open(self.config.filename, 'a') as config_file: + config_file.write("asdfasfasdfasdf") + + self.assertRaises(errors.CertStorageError, self._call) + self.assertTrue(os.path.exists(os.path.join( + self.cli_config.live_dir, "example.org"))) + self.assertFalse(os.path.exists(os.path.join( + self.cli_config.renewal_configs_dir, "example.org.conf"))) + + def test_no_renewal_config(self): + os.remove(self.config.filename) + self.assertRaises(errors.CertStorageError, self._call) + self.assertTrue(os.path.exists(os.path.join( + self.cli_config.live_dir, "example.org"))) + self.assertFalse(os.path.exists(self.config.filename)) + + def test_no_cert_file(self): + os.remove(os.path.join( + self.cli_config.live_dir, "example.org", "cert.pem")) + self._call() + self.assertFalse(os.path.exists(self.config.filename)) + self.assertFalse(os.path.exists(os.path.join( + self.cli_config.live_dir, "example.org"))) + self.assertFalse(os.path.exists(os.path.join( + self.tempdir, "archive", "example.org"))) + + def test_no_readme_file(self): + os.remove(os.path.join( + self.cli_config.live_dir, "example.org", "README")) + self._call() + self.assertFalse(os.path.exists(self.config.filename)) + self.assertFalse(os.path.exists(os.path.join( + self.cli_config.live_dir, "example.org"))) + self.assertFalse(os.path.exists(os.path.join( + self.tempdir, "archive", "example.org"))) + + def test_livedir_not_empty(self): + with open(os.path.join( + self.cli_config.live_dir, "example.org", "other_file"), 'a'): + pass + self._call() + self.assertFalse(os.path.exists(self.config.filename)) + self.assertTrue(os.path.exists(os.path.join( + self.cli_config.live_dir, "example.org"))) + self.assertFalse(os.path.exists(os.path.join( + self.tempdir, "archive", "example.org"))) + + def test_no_archive(self): + archive_dir = os.path.join(self.tempdir, "archive", "example.org") + os.rmdir(archive_dir) + self._call() + self.assertFalse(os.path.exists(self.config.filename)) + self.assertFalse(os.path.exists(os.path.join( + self.cli_config.live_dir, "example.org"))) + self.assertFalse(os.path.exists(archive_dir)) + + if __name__ == "__main__": unittest.main() # pragma: no cover