Implement delete command (#3913)

* organize cert_manager.py

* add delete files to cert manager and storage

* add tests

* add to main and cli

* Clean up all related files we can find, even if some are missing.

* error messages, debug logs, and remove RenewerConfiguration

* add logs for failure to remove

* remove renewer_config_file
This commit is contained in:
Erica Portnoy
2016-12-15 20:23:02 -08:00
committed by GitHub
parent 16361bfd06
commit 81fd0cd32c
14 changed files with 416 additions and 226 deletions
+112 -93
View File
@@ -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))
+11 -6
View File
@@ -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",
+1 -2
View File
@@ -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):
+8 -16
View File
@@ -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
-3
View File
@@ -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`)."""
-3
View File
@@ -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(
+8
View File
@@ -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
"""
+4 -21
View File
@@ -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 = []
+117 -23
View File
@@ -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
+45 -35
View File
@@ -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")
+3 -16
View File
@@ -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__':
+6 -2
View File
@@ -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)
+1 -2
View File
@@ -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
+100 -4
View File
@@ -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