upstream/certbot
1
0
Fork 0
mirror of https://github.com/certbot/certbot.git synced 2026-06-08 16:22:18 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Merge remote-tracking branch 'upstream/master' into 2478

Browse source
This commit is contained in:
Joona Hoikkala 2016-03-25 16:35:39 +02:00
commit 92a20d31cc
18 changed files with 940 additions and 785 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
README.rst
View file

@ -18,16 +18,17 @@ The Let's Encrypt Client is a fully-featured, extensible client for the Let's
Encrypt CA (or any other CA that speaks the `ACME
<https://github.com/ietf-wg-acme/acme/blob/master/draft-ietf-acme-acme.md>`_
protocol) that can automate the tasks of obtaining certificates and
configuring webservers to use them.
configuring webservers to use them. This client runs on Unix-based operating
systems.
Installation
------------
If ``letsencrypt`` is packaged for your OS, you can install it from there, and
run it by typing ``letsencrypt``. Because not all operating systems have
packages yet, we provide a temporary solution via the ``letsencrypt-auto``
wrapper script, which obtains some dependencies from your OS and puts others
in a python virtual environment::
If ``letsencrypt`` is packaged for your Unix OS, you can install it from
there, and run it by typing ``letsencrypt``. Because not all operating
systems have packages yet, we provide a temporary solution via the
``letsencrypt-auto`` wrapper script, which obtains some dependencies
from your OS and puts others in a python virtual environment::
user@webserver:~$ git clone https://github.com/letsencrypt/letsencrypt
user@webserver:~$ cd letsencrypt

472
letsencrypt/cli.py
View file

@ -1,8 +1,6 @@
"""Let's Encrypt command line argument & config processing."""
# pylint: disable=too-many-lines
from __future__ import print_function
import argparse
import copy
import glob
import json
import logging
@ -14,44 +12,30 @@ import traceback
import configargparse
import OpenSSL
import six
import zope.component
import zope.interface.exceptions
import zope.interface.verify
import letsencrypt
from letsencrypt import configuration
from letsencrypt import constants
from letsencrypt import crypto_util
from letsencrypt import errors
from letsencrypt import interfaces
from letsencrypt import le_util
from letsencrypt import storage
from letsencrypt.display import ops as display_ops
from letsencrypt.plugins import disco as plugins_disco
import letsencrypt.plugins.selection as plugin_selection
logger = logging.getLogger(__name__)
# Global, to save us from a lot of argument passing within the scope of this module
_parser = None
# These are the items which get pulled out of a renewal configuration
# file's renewalparams and actually used in the client configuration
# during the renewal process. We have to record their types here because
# the renewal configuration process loses this information.
STR_CONFIG_ITEMS = ["config_dir", "logs_dir", "work_dir", "user_agent",
"server", "account", "authenticator", "installer",
"standalone_supported_challenges"]
INT_CONFIG_ITEMS = ["rsa_key_size", "tls_sni_01_port", "http01_port"]
helpful_parser = None
# For help strings, figure out how the user ran us.
# When invoked from letsencrypt-auto, sys.argv[0] is something like:
# "/home/user/.local/share/letsencrypt/bin/letsencrypt"
# Note that this won't work if the user set VENV_PATH or XDG_DATA_HOME before running
# letsencrypt-auto (and sudo stops us from seeing if they did), so it should only be used
# for purposes where inability to detect letsencrypt-auto fails safely
# Note that this won't work if the user set VENV_PATH or XDG_DATA_HOME before
# running letsencrypt-auto (and sudo stops us from seeing if they did), so it
# should only be used for purposes where inability to detect letsencrypt-auto
# fails safely
fragment = os.path.join(".local", "share", "letsencrypt")
cli_command = "letsencrypt-auto" if fragment in sys.argv[0] else "letsencrypt"
@ -115,179 +99,23 @@ def usage_strings(plugins):
return USAGE % (apache_doc, nginx_doc), SHORT_USAGE
def should_renew(config, lineage):
"Return true if any of the circumstances for automatic renewal apply."
if config.renew_by_default:
logger.info("Auto-renewal forced with --force-renewal...")
return True
if lineage.should_autorenew(interactive=True):
logger.info("Cert is due for renewal, auto-renewing...")
return True
if config.dry_run:
logger.info("Cert not due for renewal, but simulating renewal for dry run")
return True
logger.info("Cert not yet due for renewal")
return False
def diagnose_configurator_problem(cfg_type, requested, plugins):
"""
Raise the most helpful error message about a plugin being unavailable
:param str cfg_type: either "installer" or "authenticator"
:param str requested: the plugin that was requested
:param .PluginsRegistry plugins: available plugins
:raises error.PluginSelectionError: if there was a problem
"""
if requested:
if requested not in plugins:
msg = "The requested {0} plugin does not appear to be installed".format(requested)
else:
msg = ("The {0} plugin is not working; there may be problems with "
"your existing configuration.\nThe error was: {1!r}"
.format(requested, plugins[requested].problem))
elif cfg_type == "installer":
if os.path.exists("/etc/debian_version"):
# Debian... installers are at least possible
msg = ('No installers seem to be present and working on your system; '
'fix that or try running letsencrypt with the "certonly" command')
else:
# XXX update this logic as we make progress on #788 and nginx support
msg = ('No installers are available on your OS yet; try running '
'"letsencrypt-auto certonly" to get a cert you can install manually')
else:
msg = "{0} could not be determined or is not installed".format(cfg_type)
raise errors.PluginSelectionError(msg)
def set_configurator(previously, now):
"""
Setting configurators multiple ways is okay, as long as they all agree
:param str previously: previously identified request for the installer/authenticator
:param str requested: the request currently being processed
"""
if not now:
# we're not actually setting anything
return previously
if previously:
if previously != now:
msg = "Too many flags setting configurators/installers/authenticators {0} -> {1}"
raise errors.PluginSelectionError(msg.format(repr(previously), repr(now)))
return now
def cli_plugin_requests(config):
"""
Figure out which plugins the user requested with CLI and config options
:returns: (requested authenticator string or None, requested installer string or None)
:rtype: tuple
"""
req_inst = req_auth = config.configurator
req_inst = set_configurator(req_inst, config.installer)
req_auth = set_configurator(req_auth, config.authenticator)
if config.nginx:
req_inst = set_configurator(req_inst, "nginx")
req_auth = set_configurator(req_auth, "nginx")
if config.apache:
req_inst = set_configurator(req_inst, "apache")
req_auth = set_configurator(req_auth, "apache")
if config.standalone:
req_auth = set_configurator(req_auth, "standalone")
if config.webroot:
req_auth = set_configurator(req_auth, "webroot")
if config.manual:
req_auth = set_configurator(req_auth, "manual")
logger.debug("Requested authenticator %s and installer %s", req_auth, req_inst)
return req_auth, req_inst
noninstaller_plugins = ["webroot", "manual", "standalone"]
def choose_configurator_plugins(config, plugins, verb):
"""
Figure out which configurator we're going to use, modifies
config.authenticator and config.istaller strings to reflect that choice if
necessary.
:raises errors.PluginSelectionError if there was a problem
:returns: (an `IAuthenticator` or None, an `IInstaller` or None)
:rtype: tuple
"""
req_auth, req_inst = cli_plugin_requests(config)
# Which plugins do we need?
if verb == "run":
need_inst = need_auth = True
if req_auth in noninstaller_plugins and not req_inst:
msg = ('With the {0} plugin, you probably want to use the "certonly" command, eg:{1}'
'{1} {2} certonly --{0}{1}{1}'
'(Alternatively, add a --installer flag. See https://eff.org/letsencrypt-plugins'
'{1} and "--help plugins" for more information.)'.format(
req_auth, os.linesep, cli_command))
raise errors.MissingCommandlineFlag(msg)
else:
need_inst = need_auth = False
if verb == "certonly":
need_auth = True
if verb == "install":
need_inst = True
if config.authenticator:
logger.warn("Specifying an authenticator doesn't make sense in install mode")
# Try to meet the user's request and/or ask them to pick plugins
authenticator = installer = None
if verb == "run" and req_auth == req_inst:
# Unless the user has explicitly asked for different auth/install,
# only consider offering a single choice
authenticator = installer = display_ops.pick_configurator(config, req_inst, plugins)
else:
if need_inst or req_inst:
installer = display_ops.pick_installer(config, req_inst, plugins)
if need_auth:
authenticator = display_ops.pick_authenticator(config, req_auth, plugins)
logger.debug("Selected authenticator %s and installer %s", authenticator, installer)
# Report on any failures
if need_inst and not installer:
diagnose_configurator_problem("installer", req_inst, plugins)
if need_auth and not authenticator:
diagnose_configurator_problem("authenticator", req_auth, plugins)
record_chosen_plugins(config, plugins, authenticator, installer)
return installer, authenticator
def record_chosen_plugins(config, plugins, auth, inst):
"Update the config entries to reflect the plugins we actually selected."
cn = config.namespace
cn.authenticator = plugins.find_init(auth).name if auth else "none"
cn.installer = plugins.find_init(inst).name if inst else "none"
def _set_by_cli(var):
def set_by_cli(var):
"""
Return True if a particular config variable has been set by the user
(CLI or config file) including if the user explicitly set it to the
default. Returns False if the variable was assigned a default value.
"""
detector = _set_by_cli.detector
detector = set_by_cli.detector
if detector is None:
# Setup on first run: `detector` is a weird version of config in which
# the default value of every attribute is wrangled to be boolean-false
plugins = plugins_disco.PluginsRegistry.find_all()
# reconstructed_args == sys.argv[1:], or whatever was passed to main()
reconstructed_args = _parser.args + [_parser.verb]
detector = _set_by_cli.detector = prepare_and_parse_args(
reconstructed_args = helpful_parser.args + [helpful_parser.verb]
detector = set_by_cli.detector = prepare_and_parse_args(
plugins, reconstructed_args, detect_defaults=True)
# propagate plugin requests: eg --standalone modifies config.authenticator
auth, inst = cli_plugin_requests(detector)
auth, inst = plugin_selection.cli_plugin_requests(detector)
detector.authenticator = auth if auth else ""
detector.installer = inst if inst else ""
logger.debug("Default Detector is %r", detector)
@ -312,264 +140,16 @@ def _set_by_cli(var):
else:
return False
# static housekeeping var
_set_by_cli.detector = None
def _restore_required_config_elements(config, renewalparams):
"""Sets non-plugin specific values in config from renewalparams
:param configuration.NamespaceConfig config: configuration for the
current lineage
:param configobj.Section renewalparams: parameters from the renewal
configuration file that defines this lineage
"""
# string-valued items to add if they're present
for config_item in STR_CONFIG_ITEMS:
if config_item in renewalparams and not _set_by_cli(config_item):
value = renewalparams[config_item]
# Unfortunately, we've lost type information from ConfigObj,
# so we don't know if the original was NoneType or str!
if value == "None":
value = None
setattr(config.namespace, config_item, value)
# int-valued items to add if they're present
for config_item in INT_CONFIG_ITEMS:
if config_item in renewalparams and not _set_by_cli(config_item):
config_value = renewalparams[config_item]
# the default value for http01_port was None during private beta
if config_item == "http01_port" and config_value == "None":
logger.info("updating legacy http01_port value")
int_value = flag_default("http01_port")
else:
try:
int_value = int(config_value)
except ValueError:
raise errors.Error(
"Expected a numeric value for {0}".format(config_item))
setattr(config.namespace, config_item, int_value)
set_by_cli.detector = None
def _restore_plugin_configs(config, renewalparams):
"""Sets plugin specific values in config from renewalparams
:param configuration.NamespaceConfig config: configuration for the
current lineage
:param configobj.Section renewalparams: Parameters from the renewal
configuration file that defines this lineage
"""
# Now use parser to get plugin-prefixed items with correct types
# XXX: the current approach of extracting only prefixed items
# related to the actually-used installer and authenticator
# works as long as plugins don't need to read plugin-specific
# variables set by someone else (e.g., assuming Apache
# configurator doesn't need to read webroot_ variables).
# Note: if a parameter that used to be defined in the parser is no
# longer defined, stored copies of that parameter will be
# deserialized as strings by this logic even if they were
# originally meant to be some other type.
if renewalparams["authenticator"] == "webroot":
_restore_webroot_config(config, renewalparams)
plugin_prefixes = []
else:
plugin_prefixes = [renewalparams["authenticator"]]
if renewalparams.get("installer", None) is not None:
plugin_prefixes.append(renewalparams["installer"])
for plugin_prefix in set(plugin_prefixes):
for config_item, config_value in six.iteritems(renewalparams):
if config_item.startswith(plugin_prefix + "_") and not _set_by_cli(config_item):
# Values None, True, and False need to be treated specially,
# As they don't get parsed correctly based on type
if config_value in ("None", "True", "False"):
# bool("False") == True
# pylint: disable=eval-used
setattr(config.namespace, config_item, eval(config_value))
continue
for action in _parser.parser._actions: # pylint: disable=protected-access
if action.type is not None and action.dest == config_item:
setattr(config.namespace, config_item,
action.type(config_value))
break
else:
setattr(config.namespace, config_item, str(config_value))
def _restore_webroot_config(config, renewalparams):
"""
webroot_map is, uniquely, a dict, and the general-purpose configuration
restoring logic is not able to correctly parse it from the serialized
form.
"""
if "webroot_map" in renewalparams:
# if the user does anything that would create a new webroot map on the
# CLI, don't use the old one
if not (_set_by_cli("webroot_map") or _set_by_cli("webroot_path")):
setattr(config.namespace, "webroot_map", renewalparams["webroot_map"])
elif "webroot_path" in renewalparams:
logger.info("Ancient renewal conf file without webroot-map, restoring webroot-path")
wp = renewalparams["webroot_path"]
if isinstance(wp, str): # prior to 0.1.0, webroot_path was a string
wp = [wp]
setattr(config.namespace, "webroot_path", wp)
def _reconstitute(config, full_path):
"""Try to instantiate a RenewableCert, updating config with relevant items.
This is specifically for use in renewal and enforces several checks
and policies to ensure that we can try to proceed with the renwal
request. The config argument is modified by including relevant options
read from the renewal configuration file.
:param configuration.NamespaceConfig config: configuration for the
current lineage
:param str full_path: Absolute path to the configuration file that
defines this lineage
:returns: the RenewableCert object or None if a fatal error occurred
:rtype: `storage.RenewableCert` or NoneType
"""
try:
renewal_candidate = storage.RenewableCert(
full_path, configuration.RenewerConfiguration(config))
except (errors.CertStorageError, IOError):
logger.warning("Renewal configuration file %s is broken. Skipping.", full_path)
logger.debug("Traceback was:\n%s", traceback.format_exc())
return None
if "renewalparams" not in renewal_candidate.configuration:
logger.warning("Renewal configuration file %s lacks "
"renewalparams. Skipping.", full_path)
return None
renewalparams = renewal_candidate.configuration["renewalparams"]
if "authenticator" not in renewalparams:
logger.warning("Renewal configuration file %s does not specify "
"an authenticator. Skipping.", full_path)
return None
# Now restore specific values along with their data types, if
# those elements are present.
try:
_restore_required_config_elements(config, renewalparams)
_restore_plugin_configs(config, renewalparams)
except (ValueError, errors.Error) as error:
logger.warning(
"An error occured while parsing %s. The error was %s. "
"Skipping the file.", full_path, error.message)
logger.debug("Traceback was:\n%s", traceback.format_exc())
return None
try:
for d in renewal_candidate.names():
process_domain(config, d)
except errors.ConfigurationError as error:
logger.warning("Renewal configuration file %s references a cert "
"that contains an invalid domain name. The problem "
"was: %s. Skipping.", full_path, error)
return None
return renewal_candidate
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 _renew_describe_results(config, renew_successes, renew_failures,
renew_skipped, parse_failures):
status = lambda x, msg: " " + "\n ".join(i + " (" + msg +")" for i in x)
if config.dry_run:
print("** DRY RUN: simulating 'letsencrypt renew' close to cert expiry")
print("** (The test certificates below have not been saved.)")
print()
if renew_skipped:
print("The following certs are not due for renewal yet:")
print(status(renew_skipped, "skipped"))
if not renew_successes and not renew_failures:
print("No renewals were attempted.")
elif renew_successes and not renew_failures:
print("Congratulations, all renewals succeeded. The following certs "
"have been renewed:")
print(status(renew_successes, "success"))
elif renew_failures and not renew_successes:
print("All renewal attempts failed. The following certs could not be "
"renewed:")
print(status(renew_failures, "failure"))
elif renew_failures and renew_successes:
print("The following certs were successfully renewed:")
print(status(renew_successes, "success"))
print("\nThe following certs could not be renewed:")
print(status(renew_failures, "failure"))
if parse_failures:
print("\nAdditionally, the following renewal configuration files "
"were invalid: ")
print(status(parse_failures, "parsefail"))
if config.dry_run:
print("** DRY RUN: simulating 'letsencrypt renew' close to cert expiry")
print("** (The test certificates above have not been saved.)")
def renew(config, unused_plugins):
"""Renew previously-obtained certificates."""
if config.domains != []:
raise errors.Error("Currently, the renew verb is only capable of "
"renewing all installed certificates that are due "
"to be renewed; individual domains cannot be "
"specified with this action. If you would like to "
"renew specific certificates, use the certonly "
"command. The renew verb may provide other options "
"for selecting certificates to renew in the future.")
renewer_config = configuration.RenewerConfiguration(config)
renew_successes = []
renew_failures = []
renew_skipped = []
parse_failures = []
for renewal_file in _renewal_conf_files(renewer_config):
print("Processing " + renewal_file)
lineage_config = copy.deepcopy(config)
# Note that this modifies config (to add back the configuration
# elements from within the renewal configuration file).
try:
renewal_candidate = _reconstitute(lineage_config, renewal_file)
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)
continue
try:
if renewal_candidate is None:
parse_failures.append(renewal_file)
else:
# XXX: ensure that each call here replaces the previous one
zope.component.provideUtility(lineage_config)
if should_renew(lineage_config, renewal_candidate):
plugins = plugins_disco.PluginsRegistry.find_all()
from letsencrypt import main
main.obtain_cert(lineage_config, plugins, renewal_candidate)
renew_successes.append(renewal_candidate.fullchain)
else:
renew_skipped.append(renewal_candidate.fullchain)
except Exception as e: # pylint: disable=broad-except
# obtain_cert (presumably) encountered an unanticipated problem.
logger.warning("Attempting to renew cert from %s produced an "
"unexpected error: %s. Skipping.", renewal_file, e)
logger.debug("Traceback was:\n%s", traceback.format_exc())
renew_failures.append(renewal_candidate.fullchain)
# Describe all the results
_renew_describe_results(config, renew_successes, renew_failures,
renew_skipped, parse_failures)
if renew_failures or parse_failures:
raise errors.Error("{0} renew failure(s), {1} parse failure(s)".format(
len(renew_failures), len(parse_failures)))
else:
logger.debug("no renewal failures")
def argparse_type(variable):
"Return our argparse type function for a config variable (default: str)"
# pylint: disable=protected-access
for action in helpful_parser.parser._actions:
if action.type is not None and action.dest == variable:
return action.type
return str
def read_file(filename, mode="rb"):
"""Returns the given file's contents.
@ -592,11 +172,15 @@ def read_file(filename, mode="rb"):
def flag_default(name):
"""Default value for CLI flag."""
# XXX: this is an internal housekeeping notion of defaults before
# argparse has been set up; it is not accurate for all flags. Call it
# with caution. Plugin defaults are missing, and some things are using
# defaults defined in this file, not in constants.py :(
return constants.CLI_DEFAULTS[name]
def config_help(name, hidden=False):
"""Help message for `.IConfig` attribute."""
"""Extract the help message for an `.IConfig` attribute."""
if hidden:
return argparse.SUPPRESS
else:
@ -633,7 +217,7 @@ class HelpfulArgumentParser(object):
self.VERBS = {"auth": main.obtain_cert, "certonly": main.obtain_cert,
"config_changes": main.config_changes, "run": main.run,
"install": main.install, "plugins": main.plugins_cmd,
"renew": renew, "revoke": main.revoke,
"renew": main.renew, "revoke": main.revoke,
"rollback": main.rollback, "everything": main.run}
# List of topics for which additional help can be provided
@ -844,7 +428,7 @@ class HelpfulArgumentParser(object):
def modify_arg_for_default_detection(self, *args, **kwargs):
"""
Adding an arg, but ensure that it has a default that evaluates to false,
so that _set_by_cli can tell if it was set. Only called if detect_defaults==True.
so that set_by_cli can tell if it was set. Only called if detect_defaults==True.
:param list *args: the names of this argument flag
:param dict **kwargs: various argparse settings for this argument
@ -1119,8 +703,8 @@ def prepare_and_parse_args(plugins, args, detect_defaults=False):
_plugins_parsing(helpful, plugins)
if not detect_defaults:
global _parser # pylint: disable=global-statement
_parser = helpful
global helpful_parser # pylint: disable=global-statement
helpful_parser = helpful
return helpful.parse_args()

3
letsencrypt/client.py
View file

@ -27,6 +27,7 @@ from letsencrypt import storage
from letsencrypt.display import ops as display_ops
from letsencrypt.display import enhancements
from letsencrypt.plugins import selection as plugin_selection
logger = logging.getLogger(__name__)
@ -534,7 +535,7 @@ def rollback(default_installer, checkpoints, config, plugins):
"""
# Misconfigurations are only a slight problems... allow the user to rollback
installer = display_ops.pick_installer(
installer = plugin_selection.pick_installer(
config, default_installer, plugins, question="Which installer "
"should be used for rollback?")

136
letsencrypt/display/ops.py
View file

@ -9,130 +9,10 @@ from letsencrypt import interfaces
from letsencrypt import le_util
from letsencrypt.display import util as display_util
logger = logging.getLogger(__name__)
# Define a helper function to avoid verbose code
util = zope.component.getUtility
def choose_plugin(prepared, question):
"""Allow the user to choose their plugin.
:param list prepared: List of `~.PluginEntryPoint`.
:param str question: Question to be presented to the user.
:returns: Plugin entry point chosen by the user.
:rtype: `~.PluginEntryPoint`
"""
opts = [plugin_ep.description_with_name +
(" [Misconfigured]" if plugin_ep.misconfigured else "")
for plugin_ep in prepared]
while True:
disp = util(interfaces.IDisplay)
code, index = disp.menu(question, opts, help_label="More Info")
if code == display_util.OK:
plugin_ep = prepared[index]
if plugin_ep.misconfigured:
util(interfaces.IDisplay).notification(
"The selected plugin encountered an error while parsing "
"your server configuration and cannot be used. The error "
"was:\n\n{0}".format(plugin_ep.prepare()),
height=display_util.HEIGHT, pause=False)
else:
return plugin_ep
elif code == display_util.HELP:
if prepared[index].misconfigured:
msg = "Reported Error: %s" % prepared[index].prepare()
else:
msg = prepared[index].init().more_info()
util(interfaces.IDisplay).notification(
msg, height=display_util.HEIGHT)
else:
return None
def pick_plugin(config, default, plugins, question, ifaces):
"""Pick plugin.
:param letsencrypt.interfaces.IConfig: Configuration
:param str default: Plugin name supplied by user or ``None``.
:param letsencrypt.plugins.disco.PluginsRegistry plugins:
All plugins registered as entry points.
:param str question: Question to be presented to the user in case
multiple candidates are found.
:param list ifaces: Interfaces that plugins must provide.
:returns: Initialized plugin.
:rtype: IPlugin
"""
if default is not None:
# throw more UX-friendly error if default not in plugins
filtered = plugins.filter(lambda p_ep: p_ep.name == default)
else:
if config.noninteractive_mode:
# it's really bad to auto-select the single available plugin in
# non-interactive mode, because an update could later add a second
# available plugin
raise errors.MissingCommandlineFlag(
"Missing command line flags. For non-interactive execution, "
"you will need to specify a plugin on the command line. Run "
"with '--help plugins' to see a list of options, and see "
"https://eff.org/letsencrypt-plugins for more detail on what "
"the plugins do and how to use them.")
filtered = plugins.visible().ifaces(ifaces)
filtered.init(config)
verified = filtered.verify(ifaces)
verified.prepare()
prepared = verified.available()
if len(prepared) > 1:
logger.debug("Multiple candidate plugins: %s", prepared)
plugin_ep = choose_plugin(prepared.values(), question)
if plugin_ep is None:
return None
else:
return plugin_ep.init()
elif len(prepared) == 1:
plugin_ep = prepared.values()[0]
logger.debug("Single candidate plugin: %s", plugin_ep)
if plugin_ep.misconfigured:
return None
return plugin_ep.init()
else:
logger.debug("No candidate plugin")
return None
def pick_authenticator(
config, default, plugins, question="How would you "
"like to authenticate with the Let's Encrypt CA?"):
"""Pick authentication plugin."""
return pick_plugin(
config, default, plugins, question, (interfaces.IAuthenticator,))
def pick_installer(config, default, plugins,
question="How would you like to install certificates?"):
"""Pick installer plugin."""
return pick_plugin(
config, default, plugins, question, (interfaces.IInstaller,))
def pick_configurator(
config, default, plugins,
question="How would you like to authenticate and install "
"certificates?"):
"""Pick configurator plugin."""
return pick_plugin(
config, default, plugins, question,
(interfaces.IAuthenticator, interfaces.IInstaller))
z_util = zope.component.getUtility
def get_email(more=False, invalid=False):
@ -182,7 +62,7 @@ def choose_account(accounts):
# Note this will get more complicated once we start recording authorizations
labels = [acc.slug for acc in accounts]
code, index = util(interfaces.IDisplay).menu(
code, index = z_util(interfaces.IDisplay).menu(
"Please choose an account", labels)
if code == display_util.OK:
return accounts[index]
@ -208,7 +88,7 @@ def choose_names(installer):
names = get_valid_domains(domains)
if not names:
manual = util(interfaces.IDisplay).yesno(
manual = z_util(interfaces.IDisplay).yesno(
"No names were found in your configuration files.{0}You should "
"specify ServerNames in your config files in order to allow for "
"accurate installation of your certificate.{0}"
@ -256,7 +136,7 @@ def _filter_names(names):
:rtype: tuple
"""
code, names = util(interfaces.IDisplay).checklist(
code, names = z_util(interfaces.IDisplay).checklist(
"Which names would you like to activate HTTPS for?",
tags=names, cli_flag="--domains")
return code, [str(s) for s in names]
@ -265,7 +145,7 @@ def _filter_names(names):
def _choose_names_manually():
"""Manually input names for those without an installer."""
code, input_ = util(interfaces.IDisplay).input(
code, input_ = z_util(interfaces.IDisplay).input(
"Please enter in your domain name(s) (comma and/or space separated) ",
cli_flag="--domains")
@ -300,7 +180,7 @@ def _choose_names_manually():
if retry_message:
# We had error in input
retry = util(interfaces.IDisplay).yesno(retry_message)
retry = z_util(interfaces.IDisplay).yesno(retry_message)
if retry:
return _choose_names_manually()
else:
@ -316,7 +196,7 @@ def success_installation(domains):
:param list domains: domain names which were enabled
"""
util(interfaces.IDisplay).notification(
z_util(interfaces.IDisplay).notification(
"Congratulations! You have successfully enabled {0}{1}{1}"
"You should test your configuration at:{1}{2}".format(
_gen_https_names(domains),
@ -335,7 +215,7 @@ def success_renewal(domains, action):
:param str action: can be "reinstall" or "renew"
"""
util(interfaces.IDisplay).notification(
z_util(interfaces.IDisplay).notification(
"Your existing certificate has been successfully {3}ed, and the "
"new certificate has been installed.{1}{1}"
"The new certificate covers the following domains: {0}{1}{1}"

20
letsencrypt/main.py
View file

@ -27,11 +27,12 @@ from letsencrypt import interfaces
from letsencrypt import le_util
from letsencrypt import log
from letsencrypt import reporter
from letsencrypt import renewal
from letsencrypt import storage
from letsencrypt.cli import choose_configurator_plugins, _renewal_conf_files, should_renew
from letsencrypt.display import util as display_util, ops as display_ops
from letsencrypt.plugins import disco as plugins_disco
from letsencrypt.plugins import selection as plug_sel
logger = logging.getLogger(__name__)
@ -186,7 +187,7 @@ def _handle_identical_cert_request(config, cert):
:rtype: tuple
"""
if should_renew(config, cert):
if renewal.should_renew(config, cert):
return "renew", cert
if config.reinstall:
# Set with --reinstall, force an identical certificate to be
@ -263,7 +264,7 @@ def _find_duplicative_certs(config, domains):
# Verify the directory is there
le_util.make_or_verify_dir(configs_dir, mode=0o755, uid=os.geteuid())
for renewal_file in _renewal_conf_files(cli_config):
for renewal_file in renewal.renewal_conf_files(cli_config):
try:
candidate_lineage = storage.RenewableCert(renewal_file, cli_config)
except (errors.CertStorageError, IOError):
@ -406,7 +407,7 @@ def install(config, plugins):
# this function ...
try:
installer, _ = choose_configurator_plugins(config, plugins, "install")
installer, _ = plug_sel.choose_configurator_plugins(config, plugins, "install")
except errors.PluginSelectionError as e:
return e.message
@ -462,7 +463,7 @@ def config_changes(config, unused_plugins):
def revoke(config, unused_plugins): # TODO: coop with renewal config
"""Revoke a previously obtained certificate."""
# For user-agent construction
config.namespace.installer = config.namespace.authenticator = "none"
config.namespace.installer = config.namespace.authenticator = "None"
if config.key_path is not None: # revocation by cert key
logger.debug("Revoking %s using cert key %s",
config.cert_path[0], config.key_path[0])
@ -481,7 +482,7 @@ def run(config, plugins): # pylint: disable=too-many-branches,too-many-locals
# TODO: Make run as close to auth + install as possible
# Possible difficulties: config.csr was hacked into auth
try:
installer, authenticator = choose_configurator_plugins(config, plugins, "run")
installer, authenticator = plug_sel.choose_configurator_plugins(config, plugins, "run")
except errors.PluginSelectionError as e:
return e.message
@ -511,7 +512,7 @@ def obtain_cert(config, plugins, lineage=None):
# pylint: disable=too-many-locals
try:
# installers are used in auth mode to determine domain names
installer, authenticator = choose_configurator_plugins(config, plugins, "certonly")
installer, authenticator = plug_sel.choose_configurator_plugins(config, plugins, "certonly")
except errors.PluginSelectionError as e:
logger.info("Could not choose appropriate plugin: %s", e)
raise
@ -552,6 +553,11 @@ def obtain_cert(config, plugins, lineage=None):
config.installer, "server; fullchain is", lineage.fullchain)
_suggest_donation_if_appropriate(config, action)
def renew(config, unused_plugins):
"""Renew previously-obtained certificates."""
renewal.renew_all_lineages(config)
def setup_log_file_handler(config, logfile, fmt):
"""Setup file debug logging."""

2
letsencrypt/plugins/common.py
View file

@ -52,7 +52,7 @@ class Plugin(object):
NOTE: if you add argpase arguments such that users setting them can
create a config entry that python's bool() would consider false (ie,
the use might set the variable to "", [], 0, etc), please ensure that
cli._set_by_cli() works for your variable.
cli.set_by_cli() works for your variable.
"""

1
letsencrypt/plugins/disco.py
View file

@ -4,6 +4,7 @@ import logging
import pkg_resources
import zope.interface
import zope.interface.verify
from letsencrypt import constants
from letsencrypt import errors

273
letsencrypt/plugins/selection.py Normal file
View file

@ -0,0 +1,273 @@
"""Decide which plugins to use for authentication & installation"""
from __future__ import print_function
import os
import logging
import zope.component
from letsencrypt import errors
from letsencrypt import interfaces
from letsencrypt.display import util as display_util
logger = logging.getLogger(__name__)
z_util = zope.component.getUtility
def pick_configurator(
config, default, plugins,
question="How would you like to authenticate and install "
"certificates?"):
"""Pick configurator plugin."""
return pick_plugin(
config, default, plugins, question,
(interfaces.IAuthenticator, interfaces.IInstaller))
def pick_installer(config, default, plugins,
question="How would you like to install certificates?"):
"""Pick installer plugin."""
return pick_plugin(
config, default, plugins, question, (interfaces.IInstaller,))
def pick_authenticator(
config, default, plugins, question="How would you "
"like to authenticate with the Let's Encrypt CA?"):
"""Pick authentication plugin."""
return pick_plugin(
config, default, plugins, question, (interfaces.IAuthenticator,))
def pick_plugin(config, default, plugins, question, ifaces):
"""Pick plugin.
:param letsencrypt.interfaces.IConfig: Configuration
:param str default: Plugin name supplied by user or ``None``.
:param letsencrypt.plugins.disco.PluginsRegistry plugins:
All plugins registered as entry points.
:param str question: Question to be presented to the user in case
multiple candidates are found.
:param list ifaces: Interfaces that plugins must provide.
:returns: Initialized plugin.
:rtype: IPlugin
"""
if default is not None:
# throw more UX-friendly error if default not in plugins
filtered = plugins.filter(lambda p_ep: p_ep.name == default)
else:
if config.noninteractive_mode:
# it's really bad to auto-select the single available plugin in
# non-interactive mode, because an update could later add a second
# available plugin
raise errors.MissingCommandlineFlag(
"Missing command line flags. For non-interactive execution, "
"you will need to specify a plugin on the command line. Run "
"with '--help plugins' to see a list of options, and see "
"https://eff.org/letsencrypt-plugins for more detail on what "
"the plugins do and how to use them.")
filtered = plugins.visible().ifaces(ifaces)
filtered.init(config)
verified = filtered.verify(ifaces)
verified.prepare()
prepared = verified.available()
if len(prepared) > 1:
logger.debug("Multiple candidate plugins: %s", prepared)
plugin_ep = choose_plugin(prepared.values(), question)
if plugin_ep is None:
return None
else:
return plugin_ep.init()
elif len(prepared) == 1:
plugin_ep = prepared.values()[0]
logger.debug("Single candidate plugin: %s", plugin_ep)
if plugin_ep.misconfigured:
return None
return plugin_ep.init()
else:
logger.debug("No candidate plugin")
return None
def choose_plugin(prepared, question):
"""Allow the user to choose their plugin.
:param list prepared: List of `~.PluginEntryPoint`.
:param str question: Question to be presented to the user.
:returns: Plugin entry point chosen by the user.
:rtype: `~.PluginEntryPoint`
"""
opts = [plugin_ep.description_with_name +
(" [Misconfigured]" if plugin_ep.misconfigured else "")
for plugin_ep in prepared]
while True:
disp = z_util(interfaces.IDisplay)
code, index = disp.menu(question, opts, help_label="More Info")
if code == display_util.OK:
plugin_ep = prepared[index]
if plugin_ep.misconfigured:
z_util(interfaces.IDisplay).notification(
"The selected plugin encountered an error while parsing "
"your server configuration and cannot be used. The error "
"was:\n\n{0}".format(plugin_ep.prepare()),
height=display_util.HEIGHT, pause=False)
else:
return plugin_ep
elif code == display_util.HELP:
if prepared[index].misconfigured:
msg = "Reported Error: %s" % prepared[index].prepare()
else:
msg = prepared[index].init().more_info()
z_util(interfaces.IDisplay).notification(
msg, height=display_util.HEIGHT)
else:
return None
noninstaller_plugins = ["webroot", "manual", "standalone"]
def record_chosen_plugins(config, plugins, auth, inst):
"Update the config entries to reflect the plugins we actually selected."
cn = config.namespace
cn.authenticator = plugins.find_init(auth).name if auth else "None"
cn.installer = plugins.find_init(inst).name if inst else "None"
def choose_configurator_plugins(config, plugins, verb):
"""
Figure out which configurator we're going to use, modifies
config.authenticator and config.installer strings to reflect that choice if
necessary.
:raises errors.PluginSelectionError if there was a problem
:returns: (an `IAuthenticator` or None, an `IInstaller` or None)
:rtype: tuple
"""
req_auth, req_inst = cli_plugin_requests(config)
# Which plugins do we need?
if verb == "run":
need_inst = need_auth = True
from letsencrypt.cli import cli_command
if req_auth in noninstaller_plugins and not req_inst:
msg = ('With the {0} plugin, you probably want to use the "certonly" command, eg:{1}'
'{1} {2} certonly --{0}{1}{1}'
'(Alternatively, add a --installer flag. See https://eff.org/letsencrypt-plugins'
'{1} and "--help plugins" for more information.)'.format(
req_auth, os.linesep, cli_command))
raise errors.MissingCommandlineFlag(msg)
else:
need_inst = need_auth = False
if verb == "certonly":
need_auth = True
if verb == "install":
need_inst = True
if config.authenticator:
logger.warn("Specifying an authenticator doesn't make sense in install mode")
# Try to meet the user's request and/or ask them to pick plugins
authenticator = installer = None
if verb == "run" and req_auth == req_inst:
# Unless the user has explicitly asked for different auth/install,
# only consider offering a single choice
authenticator = installer = pick_configurator(config, req_inst, plugins)
else:
if need_inst or req_inst:
installer = pick_installer(config, req_inst, plugins)
if need_auth:
authenticator = pick_authenticator(config, req_auth, plugins)
logger.debug("Selected authenticator %s and installer %s", authenticator, installer)
# Report on any failures
if need_inst and not installer:
diagnose_configurator_problem("installer", req_inst, plugins)
if need_auth and not authenticator:
diagnose_configurator_problem("authenticator", req_auth, plugins)
record_chosen_plugins(config, plugins, authenticator, installer)
return installer, authenticator
def set_configurator(previously, now):
"""
Setting configurators multiple ways is okay, as long as they all agree
:param str previously: previously identified request for the installer/authenticator
:param str requested: the request currently being processed
"""
if not now:
# we're not actually setting anything
return previously
if previously:
if previously != now:
msg = "Too many flags setting configurators/installers/authenticators {0} -> {1}"
raise errors.PluginSelectionError(msg.format(repr(previously), repr(now)))
return now
def cli_plugin_requests(config):
"""
Figure out which plugins the user requested with CLI and config options
:returns: (requested authenticator string or None, requested installer string or None)
:rtype: tuple
"""
req_inst = req_auth = config.configurator
req_inst = set_configurator(req_inst, config.installer)
req_auth = set_configurator(req_auth, config.authenticator)
if config.nginx:
req_inst = set_configurator(req_inst, "nginx")
req_auth = set_configurator(req_auth, "nginx")
if config.apache:
req_inst = set_configurator(req_inst, "apache")
req_auth = set_configurator(req_auth, "apache")
if config.standalone:
req_auth = set_configurator(req_auth, "standalone")
if config.webroot:
req_auth = set_configurator(req_auth, "webroot")
if config.manual:
req_auth = set_configurator(req_auth, "manual")
logger.debug("Requested authenticator %s and installer %s", req_auth, req_inst)
return req_auth, req_inst
def diagnose_configurator_problem(cfg_type, requested, plugins):
"""
Raise the most helpful error message about a plugin being unavailable
:param str cfg_type: either "installer" or "authenticator"
:param str requested: the plugin that was requested
:param .PluginsRegistry plugins: available plugins
:raises error.PluginSelectionError: if there was a problem
"""
if requested:
if requested not in plugins:
msg = "The requested {0} plugin does not appear to be installed".format(requested)
else:
msg = ("The {0} plugin is not working; there may be problems with "
"your existing configuration.\nThe error was: {1!r}"
.format(requested, plugins[requested].problem))
elif cfg_type == "installer":
if os.path.exists("/etc/debian_version"):
# Debian... installers are at least possible
msg = ('No installers seem to be present and working on your system; '
'fix that or try running letsencrypt with the "certonly" command')
else:
# XXX update this logic as we make progress on #788 and nginx support
msg = ('No installers are available on your OS yet; try running '
'"letsencrypt-auto certonly" to get a cert you can install manually')
else:
msg = "{0} could not be determined or is not installed".format(cfg_type)
raise errors.PluginSelectionError(msg)

149
letsencrypt/plugins/selection_test.py Normal file
View file

@ -0,0 +1,149 @@
"""Tests for letsenecrypt.plugins.selection"""
import sys
import unittest
import mock
import zope.component
from letsencrypt.display import util as display_util
from letsencrypt import interfaces
class ConveniencePickPluginTest(unittest.TestCase):
"""Tests for letsencrypt.plugins.selection.pick_*."""
def _test(self, fun, ifaces):
config = mock.Mock()
default = mock.Mock()
plugins = mock.Mock()
with mock.patch("letsencrypt.plugins.selection.pick_plugin") as mock_p:
mock_p.return_value = "foo"
self.assertEqual("foo", fun(config, default, plugins, "Question?"))
mock_p.assert_called_once_with(
config, default, plugins, "Question?", ifaces)
def test_authenticator(self):
from letsencrypt.plugins.selection import pick_authenticator
self._test(pick_authenticator, (interfaces.IAuthenticator,))
def test_installer(self):
from letsencrypt.plugins.selection import pick_installer
self._test(pick_installer, (interfaces.IInstaller,))
def test_configurator(self):
from letsencrypt.plugins.selection import pick_configurator
self._test(pick_configurator,
(interfaces.IAuthenticator, interfaces.IInstaller))
class PickPluginTest(unittest.TestCase):
"""Tests for letsencrypt.plugins.selection.pick_plugin."""
def setUp(self):
self.config = mock.Mock(noninteractive_mode=False)
self.default = None
self.reg = mock.MagicMock()
self.question = "Question?"
self.ifaces = []
def _call(self):
from letsencrypt.plugins.selection import pick_plugin
return pick_plugin(self.config, self.default, self.reg,
self.question, self.ifaces)
def test_default_provided(self):
self.default = "foo"
self._call()
self.assertEqual(1, self.reg.filter.call_count)
def test_no_default(self):
self._call()
self.assertEqual(1, self.reg.visible().ifaces.call_count)
def test_no_candidate(self):
self.assertTrue(self._call() is None)
def test_single(self):
plugin_ep = mock.MagicMock()
plugin_ep.init.return_value = "foo"
plugin_ep.misconfigured = False
self.reg.visible().ifaces().verify().available.return_value = {
"bar": plugin_ep}
self.assertEqual("foo", self._call())
def test_single_misconfigured(self):
plugin_ep = mock.MagicMock()
plugin_ep.init.return_value = "foo"
plugin_ep.misconfigured = True
self.reg.visible().ifaces().verify().available.return_value = {
"bar": plugin_ep}
self.assertTrue(self._call() is None)
def test_multiple(self):
plugin_ep = mock.MagicMock()
plugin_ep.init.return_value = "foo"
self.reg.visible().ifaces().verify().available.return_value = {
"bar": plugin_ep,
"baz": plugin_ep,
}
with mock.patch("letsencrypt.plugins.selection.choose_plugin") as mock_choose:
mock_choose.return_value = plugin_ep
self.assertEqual("foo", self._call())
mock_choose.assert_called_once_with(
[plugin_ep, plugin_ep], self.question)
def test_choose_plugin_none(self):
self.reg.visible().ifaces().verify().available.return_value = {
"bar": None,
"baz": None,
}
with mock.patch("letsencrypt.plugins.selection.choose_plugin") as mock_choose:
mock_choose.return_value = None
self.assertTrue(self._call() is None)
class ChoosePluginTest(unittest.TestCase):
"""Tests for letsencrypt.plugins.selection.choose_plugin."""
def setUp(self):
zope.component.provideUtility(display_util.FileDisplay(sys.stdout))
self.mock_apache = mock.Mock(
description_with_name="a", misconfigured=True)
self.mock_stand = mock.Mock(
description_with_name="s", misconfigured=False)
self.mock_stand.init().more_info.return_value = "standalone"
self.plugins = [
self.mock_apache,
self.mock_stand,
]
def _call(self):
from letsencrypt.plugins.selection import choose_plugin
return choose_plugin(self.plugins, "Question?")
@mock.patch("letsencrypt.plugins.selection.z_util")
def test_selection(self, mock_util):
mock_util().menu.side_effect = [(display_util.OK, 0),
(display_util.OK, 1)]
self.assertEqual(self.mock_stand, self._call())
self.assertEqual(mock_util().notification.call_count, 1)
@mock.patch("letsencrypt.plugins.selection.z_util")
def test_more_info(self, mock_util):
mock_util().menu.side_effect = [
(display_util.HELP, 0),
(display_util.HELP, 1),
(display_util.OK, 1),
]
self.assertEqual(self.mock_stand, self._call())
self.assertEqual(mock_util().notification.call_count, 2)
@mock.patch("letsencrypt.plugins.selection.z_util")
def test_no_choice(self, mock_util):
mock_util().menu.return_value = (display_util.CANCEL, 0)
self.assertTrue(self._call() is None)

0
letsencrypt/renew.py
View file

298
letsencrypt/renewal.py Normal file
View file

@ -0,0 +1,298 @@
"""Functionality for autorenewal and associated juggling of configurations"""
from __future__ import print_function
import copy
import glob
import logging
import os
import traceback
import six
import zope.component
from letsencrypt import configuration
from letsencrypt import cli
from letsencrypt import errors
from letsencrypt import storage
from letsencrypt.plugins import disco as plugins_disco
logger = logging.getLogger(__name__)
# These are the items which get pulled out of a renewal configuration
# file's renewalparams and actually used in the client configuration
# during the renewal process. We have to record their types here because
# the renewal configuration process loses this information.
STR_CONFIG_ITEMS = ["config_dir", "logs_dir", "work_dir", "user_agent",
"server", "account", "authenticator", "installer",
"standalone_supported_challenges"]
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 _reconstitute(config, full_path):
"""Try to instantiate a RenewableCert, updating config with relevant items.
This is specifically for use in renewal and enforces several checks
and policies to ensure that we can try to proceed with the renwal
request. The config argument is modified by including relevant options
read from the renewal configuration file.
:param configuration.NamespaceConfig config: configuration for the
current lineage
:param str full_path: Absolute path to the configuration file that
defines this lineage
:returns: the RenewableCert object or None if a fatal error occurred
:rtype: `storage.RenewableCert` or NoneType
"""
try:
renewal_candidate = storage.RenewableCert(
full_path, configuration.RenewerConfiguration(config))
except (errors.CertStorageError, IOError):
logger.warning("Renewal configuration file %s is broken. Skipping.", full_path)
logger.debug("Traceback was:\n%s", traceback.format_exc())
return None
if "renewalparams" not in renewal_candidate.configuration:
logger.warning("Renewal configuration file %s lacks "
"renewalparams. Skipping.", full_path)
return None
renewalparams = renewal_candidate.configuration["renewalparams"]
if "authenticator" not in renewalparams:
logger.warning("Renewal configuration file %s does not specify "
"an authenticator. Skipping.", full_path)
return None
# Now restore specific values along with their data types, if
# those elements are present.
try:
_restore_required_config_elements(config, renewalparams)
_restore_plugin_configs(config, renewalparams)
except (ValueError, errors.Error) as error:
logger.warning(
"An error occured while parsing %s. The error was %s. "
"Skipping the file.", full_path, error.message)
logger.debug("Traceback was:\n%s", traceback.format_exc())
return None
try:
for d in renewal_candidate.names():
cli.process_domain(config, d)
except errors.ConfigurationError as error:
logger.warning("Renewal configuration file %s references a cert "
"that contains an invalid domain name. The problem "
"was: %s. Skipping.", full_path, error)
return None
return renewal_candidate
def _restore_webroot_config(config, renewalparams):
"""
webroot_map is, uniquely, a dict, and the general-purpose configuration
restoring logic is not able to correctly parse it from the serialized
form.
"""
if "webroot_map" in renewalparams:
# if the user does anything that would create a new webroot map on the
# CLI, don't use the old one
if not (cli.set_by_cli("webroot_map") or cli.set_by_cli("webroot_path")):
setattr(config.namespace, "webroot_map", renewalparams["webroot_map"])
elif "webroot_path" in renewalparams:
logger.info("Ancient renewal conf file without webroot-map, restoring webroot-path")
wp = renewalparams["webroot_path"]
if isinstance(wp, str): # prior to 0.1.0, webroot_path was a string
wp = [wp]
setattr(config.namespace, "webroot_path", wp)
def _restore_plugin_configs(config, renewalparams):
"""Sets plugin specific values in config from renewalparams
:param configuration.NamespaceConfig config: configuration for the
current lineage
:param configobj.Section renewalparams: Parameters from the renewal
configuration file that defines this lineage
"""
# Now use parser to get plugin-prefixed items with correct types
# XXX: the current approach of extracting only prefixed items
# related to the actually-used installer and authenticator
# works as long as plugins don't need to read plugin-specific
# variables set by someone else (e.g., assuming Apache
# configurator doesn't need to read webroot_ variables).
# Note: if a parameter that used to be defined in the parser is no
# longer defined, stored copies of that parameter will be
# deserialized as strings by this logic even if they were
# originally meant to be some other type.
if renewalparams["authenticator"] == "webroot":
_restore_webroot_config(config, renewalparams)
plugin_prefixes = []
else:
plugin_prefixes = [renewalparams["authenticator"]]
if renewalparams.get("installer", None) is not None:
plugin_prefixes.append(renewalparams["installer"])
for plugin_prefix in set(plugin_prefixes):
for config_item, config_value in six.iteritems(renewalparams):
if config_item.startswith(plugin_prefix + "_") and not cli.set_by_cli(config_item):
# Values None, True, and False need to be treated specially,
# As their types aren't handled correctly by configobj
if config_value in ("None", "True", "False"):
# bool("False") == True
# pylint: disable=eval-used
setattr(config.namespace, config_item, eval(config_value))
else:
cast = cli.argparse_type(config_item)
setattr(config.namespace, config_item, cast(config_value))
def _restore_required_config_elements(config, renewalparams):
"""Sets non-plugin specific values in config from renewalparams
:param configuration.NamespaceConfig config: configuration for the
current lineage
:param configobj.Section renewalparams: parameters from the renewal
configuration file that defines this lineage
"""
# string-valued items to add if they're present
for config_item in STR_CONFIG_ITEMS:
if config_item in renewalparams and not cli.set_by_cli(config_item):
value = renewalparams[config_item]
# Unfortunately, we've lost type information from ConfigObj,
# so we don't know if the original was NoneType or str!
if value == "None":
value = None
setattr(config.namespace, config_item, value)
# int-valued items to add if they're present
for config_item in INT_CONFIG_ITEMS:
if config_item in renewalparams and not cli.set_by_cli(config_item):
config_value = renewalparams[config_item]
# the default value for http01_port was None during private beta
if config_item == "http01_port" and config_value == "None":
logger.info("updating legacy http01_port value")
int_value = cli.flag_default("http01_port")
else:
try:
int_value = int(config_value)
except ValueError:
raise errors.Error(
"Expected a numeric value for {0}".format(config_item))
setattr(config.namespace, config_item, int_value)
def should_renew(config, lineage):
"Return true if any of the circumstances for automatic renewal apply."
if config.renew_by_default:
logger.info("Auto-renewal forced with --force-renewal...")
return True
if lineage.should_autorenew(interactive=True):
logger.info("Cert is due for renewal, auto-renewing...")
return True
if config.dry_run:
logger.info("Cert not due for renewal, but simulating renewal for dry run")
return True
logger.info("Cert not yet due for renewal")
return False
def _renew_describe_results(config, renew_successes, renew_failures,
renew_skipped, parse_failures):
def _status(msgs, category):
return " " + "\n ".join("%s (%s)" % (m, category) for m in msgs)
if config.dry_run:
print("** DRY RUN: simulating 'letsencrypt renew' close to cert expiry")
print("** (The test certificates below have not been saved.)")
print()
if renew_skipped:
print("The following certs are not due for renewal yet:")
print(_status(renew_skipped, "skipped"))
if not renew_successes and not renew_failures:
print("No renewals were attempted.")
elif renew_successes and not renew_failures:
print("Congratulations, all renewals succeeded. The following certs "
"have been renewed:")
print(_status(renew_successes, "success"))
elif renew_failures and not renew_successes:
print("All renewal attempts failed. The following certs could not be "
"renewed:")
print(_status(renew_failures, "failure"))
elif renew_failures and renew_successes:
print("The following certs were successfully renewed:")
print(_status(renew_successes, "success"))
print("\nThe following certs could not be renewed:")
print(_status(renew_failures, "failure"))
if parse_failures:
print("\nAdditionally, the following renewal configuration files "
"were invalid: ")
print(_status(parse_failures, "parsefail"))
if config.dry_run:
print("** DRY RUN: simulating 'letsencrypt renew' close to cert expiry")
print("** (The test certificates above have not been saved.)")
def renew_all_lineages(config):
"""Examine each lineage; renew if due and report results"""
if config.domains != []:
raise errors.Error("Currently, the renew verb is only capable of "
"renewing all installed certificates that are due "
"to be renewed; individual domains cannot be "
"specified with this action. If you would like to "
"renew specific certificates, use the certonly "
"command. The renew verb may provide other options "
"for selecting certificates to renew in the future.")
renewer_config = configuration.RenewerConfiguration(config)
renew_successes = []
renew_failures = []
renew_skipped = []
parse_failures = []
for renewal_file in renewal_conf_files(renewer_config):
print("Processing " + renewal_file)
lineage_config = copy.deepcopy(config)
# Note that this modifies config (to add back the configuration
# elements from within the renewal configuration file).
try:
renewal_candidate = _reconstitute(lineage_config, renewal_file)
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)
continue
try:
if renewal_candidate is None:
parse_failures.append(renewal_file)
else:
# XXX: ensure that each call here replaces the previous one
zope.component.provideUtility(lineage_config)
if should_renew(lineage_config, renewal_candidate):
plugins = plugins_disco.PluginsRegistry.find_all()
from letsencrypt import main
main.obtain_cert(lineage_config, plugins, renewal_candidate)
renew_successes.append(renewal_candidate.fullchain)
else:
renew_skipped.append(renewal_candidate.fullchain)
except Exception as e: # pylint: disable=broad-except
# obtain_cert (presumably) encountered an unanticipated problem.
logger.warning("Attempting to renew cert from %s produced an "
"unexpected error: %s. Skipping.", renewal_file, e)
logger.debug("Traceback was:\n%s", traceback.format_exc())
renew_failures.append(renewal_candidate.fullchain)
# Describe all the results
_renew_describe_results(config, renew_successes, renew_failures,
renew_skipped, parse_failures)
if renew_failures or parse_failures:
raise errors.Error("{0} renew failure(s), {1} parse failure(s)".format(
len(renew_failures), len(parse_failures)))
else:
logger.debug("no renewal failures")

86
letsencrypt/storage.py
View file

@ -50,13 +50,12 @@ def add_time_interval(base_time, interval, textparser=parsedatetime.Calendar()):
return textparser.parseDT(interval, base_time, tzinfo=tzinfo)[0]
def write_renewal_config(filename, target, cli_config):
def write_renewal_config(filename, target, relevant_data):
"""Writes a renewal config file with the specified name and values.
:param str filename: Absolute path to the config file
:param dict target: Maps ALL_FOUR to their symlink paths
:param .RenewerConfiguration cli_config: parsed command line
arguments
:param dict relevant_data: Renewal configuration options to save
:returns: Configuration object for the new config file
:rtype: configobj.ConfigObj
@ -67,18 +66,11 @@ def write_renewal_config(filename, target, cli_config):
for kind in ALL_FOUR:
config[kind] = target[kind]
# XXX: We clearly need a more general and correct way of getting
# options into the configobj for the RenewableCert instance.
# This is a quick-and-dirty way to do it to allow integration
# testing to start. (Note that the config parameter to new_lineage
# ideally should be a ConfigObj, but in this case a dict will be
# accepted in practice.)
renewalparams = vars(cli_config.namespace)
if renewalparams:
config["renewalparams"] = renewalparams
if relevant_data:
config["renewalparams"] = relevant_data
config.comments["renewalparams"] = ["",
"Options and defaults used"
" in the renewal process"]
"Options used in "
"the renewal process"]
# TODO: add human-readable comments explaining other available
# parameters
@ -106,7 +98,10 @@ def update_configuration(lineagename, target, cli_config):
# If an existing tempfile exists, delete it
if os.path.exists(temp_filename):
os.unlink(temp_filename)
write_renewal_config(temp_filename, target, cli_config)
# Save only the config items that are relevant to renewal
values = relevant_values(vars(cli_config.namespace))
write_renewal_config(temp_filename, target, values)
os.rename(temp_filename, config_filename)
return configobj.ConfigObj(config_filename)
@ -127,6 +122,60 @@ def get_link_target(link):
return os.path.abspath(target)
def _relevant(option):
"""
Is this option one that could be restored for future renewal purposes?
:param str option: the name of the option
:rtype: bool
"""
# The list() here produces a list of the plugin names as strings.
from letsencrypt import renewal
from letsencrypt.plugins import disco as plugins_disco
plugins = list(plugins_disco.PluginsRegistry.find_all())
return (option in renewal.STR_CONFIG_ITEMS
or option in renewal.INT_CONFIG_ITEMS
or any(option.startswith(x + "_") for x in plugins))
def relevant_values(all_values):
"""Return a new dict containing only items relevant for renewal.
:param dict all_values: The original values.
:returns: A new dictionary containing items that can be used in renewal.
:rtype dict:"""
from letsencrypt import cli
def _is_cli_default(option, value):
# Look through the CLI parser defaults and see if this option is
# both present and equal to the specified value. If not, return
# False.
# pylint: disable=protected-access
for x in cli.helpful_parser.parser._actions:
if x.dest == option:
if x.default == value:
return True
else:
break
return False
values = dict()
for option, value in all_values.iteritems():
# Try to find reasons to store this item in the
# renewal config. It can be stored if it is relevant and
# (it is set_by_cli() or flag_default() is different
# from the value or flag_default() doesn't exist).
if _relevant(option):
if (cli.set_by_cli(option)
or not _is_cli_default(option, value)):
# or option not in constants.CLI_DEFAULTS
# or constants.CLI_DEFAULTS[option] != value):
values[option] = value
return values
class RenewableCert(object): # pylint: disable=too-many-instance-attributes
"""Renewable certificate.
@ -690,6 +739,7 @@ class RenewableCert(object): # pylint: disable=too-many-instance-attributes
:rtype: :class:`storage.renewableCert`
"""
# Examine the configuration and find the new lineage's name
for i in (cli_config.renewal_configs_dir, cli_config.archive_dir,
cli_config.live_dir):
@ -744,7 +794,11 @@ class RenewableCert(object): # pylint: disable=too-many-instance-attributes
# Document what we've done in a new renewal config file
config_file.close()
new_config = write_renewal_config(config_filename, target, cli_config)
# Save only the config items that are relevant to renewal
values = relevant_values(vars(cli_config.namespace))
new_config = write_renewal_config(config_filename, target, values)
return cls(new_config.filename, cli_config)
def save_successor(self, prior_version, new_cert,

44
letsencrypt/tests/cli_test.py
View file

@ -1,5 +1,4 @@
"""Tests for letsencrypt.cli."""
from __future__ import print_function
import argparse
@ -24,6 +23,7 @@ from letsencrypt import crypto_util
from letsencrypt import errors
from letsencrypt import le_util
from letsencrypt import main
from letsencrypt import renewal
from letsencrypt import storage
from letsencrypt.plugins import disco
@ -54,7 +54,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
shutil.rmtree(self.tmp_dir)
# Reset globals in cli
# pylint: disable=protected-access
cli._parser = cli._set_by_cli.detector = None
cli._parser = cli.set_by_cli.detector = None
def _call(self, args):
"Run the cli with output streams and actual client mocked out"
@ -203,12 +203,12 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
self.assertEqual(args.chain_path, os.path.abspath(chain))
self.assertEqual(args.fullchain_path, os.path.abspath(fullchain))
@mock.patch('letsencrypt.main.cli.record_chosen_plugins')
@mock.patch('letsencrypt.main.cli.display_ops')
def test_installer_selection(self, mock_display_ops, _rec):
@mock.patch('letsencrypt.main.plug_sel.record_chosen_plugins')
@mock.patch('letsencrypt.main.plug_sel.pick_installer')
def test_installer_selection(self, mock_pick_installer, _rec):
self._call(['install', '--domains', 'foo.bar', '--cert-path', 'cert',
'--key-path', 'key', '--chain-path', 'chain'])
self.assertEqual(mock_display_ops.pick_installer.call_count, 1)
self.assertEqual(mock_pick_installer.call_count, 1)
@mock.patch('letsencrypt.le_util.exe_exists')
def test_configurator_selection(self, mock_exe_exists):
@ -499,7 +499,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
self._webroot_map_test(simple_map, "/tmp2", "eg2.com,eg.com", expected_map, domains)
# test inclusion of interactively specified domains in the webroot map
with mock.patch('letsencrypt.cli.display_ops.choose_names') as mock_choose:
with mock.patch('letsencrypt.display.ops.choose_names') as mock_choose:
mock_choose.return_value = domains
expected_map["eg2.com"] = "/tmp"
self._webroot_map_test(None, "/tmp", None, expected_map, domains)
@ -523,7 +523,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
args += '-d foo.bar -a standalone certonly'.split()
self._call(args)
@mock.patch('letsencrypt.cli.zope.component.getUtility')
@mock.patch('letsencrypt.main.zope.component.getUtility')
def test_certonly_dry_run_new_request_success(self, mock_get_utility):
mock_client = mock.MagicMock()
mock_client.obtain_and_enroll_certificate.return_value = None
@ -536,7 +536,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
self.assertEqual(mock_get_utility().add_message.call_count, 1)
@mock.patch('letsencrypt.crypto_util.notAfter')
@mock.patch('letsencrypt.cli.zope.component.getUtility')
@mock.patch('letsencrypt.main.zope.component.getUtility')
def test_certonly_new_request_success(self, mock_get_utility, mock_notAfter):
cert_path = '/etc/letsencrypt/live/foo.bar'
date = '1970-01-01'
@ -561,7 +561,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
self._certonly_new_request_common, mock_client)
def _test_renewal_common(self, due_for_renewal, extra_args, log_out=None,
args=None, renew=True, error_expected=False):
args=None, should_renew=True, error_expected=False):
# pylint: disable=too-many-locals,too-many-arguments
cert_path = 'letsencrypt/tests/testdata/cert.pem'
chain_path = '/etc/letsencrypt/live/foo.bar/fullchain.pem'
@ -600,7 +600,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
"Unexpected renewal error:\n" +
traceback.format_exc())
if renew:
if should_renew:
mock_client.obtain_certificate.assert_called_once_with(['isnot.org'])
else:
self.assertEqual(mock_client.obtain_certificate.call_count, 0)
@ -635,7 +635,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
self.assertEqual(get_utility().add_message.call_count, 1)
_, _ = self._test_renewal_common(False, ['-tvv', '--debug', '--keep'],
log_out="not yet due", renew=False)
log_out="not yet due", should_renew=False)
def _dump_log(self):
with open(os.path.join(self.logs_dir, "letsencrypt.log")) as lf:
@ -658,9 +658,9 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
def test_renew_verb(self):
self._make_test_renewal_conf('sample-renewal.conf')
args = ["renew", "--dry-run", "-tvv"]
self._test_renewal_common(True, [], args=args, renew=True)
self._test_renewal_common(True, [], args=args, should_renew=True)
@mock.patch("letsencrypt.cli._set_by_cli")
@mock.patch("letsencrypt.cli.set_by_cli")
def test_ancient_webroot_renewal_conf(self, mock_set_by_cli):
mock_set_by_cli.return_value = False
rc_path = self._make_test_renewal_conf('sample-renewal-ancient.conf')
@ -670,7 +670,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
configuration.RenewerConfiguration(config))
renewalparams = lineage.configuration["renewalparams"]
# pylint: disable=protected-access
cli._restore_webroot_config(config, renewalparams)
renewal._restore_webroot_config(config, renewalparams)
self.assertEqual(config.webroot_path, ["/var/www/"])
def test_renew_verb_empty_config(self):
@ -680,7 +680,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
with open(os.path.join(rd, 'empty.conf'), 'w'):
pass # leave the file empty
args = ["renew", "--dry-run", "-tvv"]
self._test_renewal_common(False, [], args=args, renew=False, error_expected=True)
self._test_renewal_common(False, [], args=args, should_renew=False, error_expected=True)
def _make_dummy_renewal_config(self):
renewer_configs_dir = os.path.join(self.config_dir, 'renewal')
@ -701,7 +701,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
mock_rc.return_value = mock_lineage
with mock.patch('letsencrypt.main.obtain_cert') as mock_obtain_cert:
kwargs.setdefault('args', ['renew'])
self._test_renewal_common(True, None, renew=False, **kwargs)
self._test_renewal_common(True, None, should_renew=False, **kwargs)
if assert_oc_called is not None:
if assert_oc_called:
@ -749,7 +749,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
def test_renew_reconstitute_error(self):
# pylint: disable=protected-access
with mock.patch('letsencrypt.cli._reconstitute') as mock_reconstitute:
with mock.patch('letsencrypt.main.renewal._reconstitute') as mock_reconstitute:
mock_reconstitute.side_effect = Exception
self._test_renew_common(assert_oc_called=False, error_expected=True)
@ -764,15 +764,15 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
with mock.patch('letsencrypt.main.obtain_cert') as mock_obtain_cert:
mock_obtain_cert.side_effect = Exception
self._test_renewal_common(True, None, error_expected=True,
args=['renew'], renew=False)
args=['renew'], should_renew=False)
def test_renew_with_bad_cli_args(self):
self._test_renewal_common(True, None, args='renew -d example.com'.split(),
renew=False, error_expected=True)
should_renew=False, error_expected=True)
self._test_renewal_common(True, None, args='renew --csr {0}'.format(CSR).split(),
renew=False, error_expected=True)
should_renew=False, error_expected=True)
@mock.patch('letsencrypt.cli.zope.component.getUtility')
@mock.patch('letsencrypt.main.zope.component.getUtility')
@mock.patch('letsencrypt.main._treat_as_renewal')
@mock.patch('letsencrypt.main._init_le_client')
def test_certonly_reinstall(self, mock_init, mock_renewal, mock_get_utility):

5
letsencrypt/tests/client_test.py
View file

@ -460,9 +460,8 @@ class RollbackTest(unittest.TestCase):
@classmethod
def _call(cls, checkpoints, side_effect):
from letsencrypt.client import rollback
with mock.patch("letsencrypt.client"
".display_ops.pick_installer") as mock_pick_installer:
mock_pick_installer.side_effect = side_effect
with mock.patch("letsencrypt.client.plugin_selection.pick_installer") as mpi:
mpi.side_effect = side_effect
rollback(None, checkpoints, {}, mock.MagicMock())
def test_no_problems(self):

164
letsencrypt/tests/display/ops_test.py
View file

@ -22,146 +22,6 @@ from letsencrypt.tests import test_util
KEY = jose.JWKRSA.load(test_util.load_vector("rsa512_key.pem"))
class ChoosePluginTest(unittest.TestCase):
"""Tests for letsencrypt.display.ops.choose_plugin."""
def setUp(self):
zope.component.provideUtility(display_util.FileDisplay(sys.stdout))
self.mock_apache = mock.Mock(
description_with_name="a", misconfigured=True)
self.mock_stand = mock.Mock(
description_with_name="s", misconfigured=False)
self.mock_stand.init().more_info.return_value = "standalone"
self.plugins = [
self.mock_apache,
self.mock_stand,
]
def _call(self):
from letsencrypt.display.ops import choose_plugin
return choose_plugin(self.plugins, "Question?")
@mock.patch("letsencrypt.display.ops.util")
def test_selection(self, mock_util):
mock_util().menu.side_effect = [(display_util.OK, 0),
(display_util.OK, 1)]
self.assertEqual(self.mock_stand, self._call())
self.assertEqual(mock_util().notification.call_count, 1)
@mock.patch("letsencrypt.display.ops.util")
def test_more_info(self, mock_util):
mock_util().menu.side_effect = [
(display_util.HELP, 0),
(display_util.HELP, 1),
(display_util.OK, 1),
]
self.assertEqual(self.mock_stand, self._call())
self.assertEqual(mock_util().notification.call_count, 2)
@mock.patch("letsencrypt.display.ops.util")
def test_no_choice(self, mock_util):
mock_util().menu.return_value = (display_util.CANCEL, 0)
self.assertTrue(self._call() is None)
class PickPluginTest(unittest.TestCase):
"""Tests for letsencrypt.display.ops.pick_plugin."""
def setUp(self):
self.config = mock.Mock(noninteractive_mode=False)
self.default = None
self.reg = mock.MagicMock()
self.question = "Question?"
self.ifaces = []
def _call(self):
from letsencrypt.display.ops import pick_plugin
return pick_plugin(self.config, self.default, self.reg,
self.question, self.ifaces)
def test_default_provided(self):
self.default = "foo"
self._call()
self.assertEqual(1, self.reg.filter.call_count)
def test_no_default(self):
self._call()
self.assertEqual(1, self.reg.visible().ifaces.call_count)
def test_no_candidate(self):
self.assertTrue(self._call() is None)
def test_single(self):
plugin_ep = mock.MagicMock()
plugin_ep.init.return_value = "foo"
plugin_ep.misconfigured = False
self.reg.visible().ifaces().verify().available.return_value = {
"bar": plugin_ep}
self.assertEqual("foo", self._call())
def test_single_misconfigured(self):
plugin_ep = mock.MagicMock()
plugin_ep.init.return_value = "foo"
plugin_ep.misconfigured = True
self.reg.visible().ifaces().verify().available.return_value = {
"bar": plugin_ep}
self.assertTrue(self._call() is None)
def test_multiple(self):
plugin_ep = mock.MagicMock()
plugin_ep.init.return_value = "foo"
self.reg.visible().ifaces().verify().available.return_value = {
"bar": plugin_ep,
"baz": plugin_ep,
}
with mock.patch("letsencrypt.display.ops.choose_plugin") as mock_choose:
mock_choose.return_value = plugin_ep
self.assertEqual("foo", self._call())
mock_choose.assert_called_once_with(
[plugin_ep, plugin_ep], self.question)
def test_choose_plugin_none(self):
self.reg.visible().ifaces().verify().available.return_value = {
"bar": None,
"baz": None,
}
with mock.patch("letsencrypt.display.ops.choose_plugin") as mock_choose:
mock_choose.return_value = None
self.assertTrue(self._call() is None)
class ConveniencePickPluginTest(unittest.TestCase):
"""Tests for letsencrypt.display.ops.pick_*."""
def _test(self, fun, ifaces):
config = mock.Mock()
default = mock.Mock()
plugins = mock.Mock()
with mock.patch("letsencrypt.display.ops.pick_plugin") as mock_p:
mock_p.return_value = "foo"
self.assertEqual("foo", fun(config, default, plugins, "Question?"))
mock_p.assert_called_once_with(
config, default, plugins, "Question?", ifaces)
def test_authenticator(self):
from letsencrypt.display.ops import pick_authenticator
self._test(pick_authenticator, (interfaces.IAuthenticator,))
def test_installer(self):
from letsencrypt.display.ops import pick_installer
self._test(pick_installer, (interfaces.IInstaller,))
def test_configurator(self):
from letsencrypt.display.ops import pick_configurator
self._test(pick_configurator, (
interfaces.IAuthenticator, interfaces.IInstaller))
class GetEmailTest(unittest.TestCase):
"""Tests for letsencrypt.display.ops.get_email."""
@ -241,17 +101,17 @@ class ChooseAccountTest(unittest.TestCase):
from letsencrypt.display import ops
return ops.choose_account(accounts)
@mock.patch("letsencrypt.display.ops.util")
@mock.patch("letsencrypt.display.ops.z_util")
def test_one(self, mock_util):
mock_util().menu.return_value = (display_util.OK, 0)
self.assertEqual(self._call([self.acc1]), self.acc1)
@mock.patch("letsencrypt.display.ops.util")
@mock.patch("letsencrypt.display.ops.z_util")
def test_two(self, mock_util):
mock_util().menu.return_value = (display_util.OK, 1)
self.assertEqual(self._call([self.acc1, self.acc2]), self.acc2)
@mock.patch("letsencrypt.display.ops.util")
@mock.patch("letsencrypt.display.ops.z_util")
def test_cancel(self, mock_util):
mock_util().menu.return_value = (display_util.CANCEL, 1)
self.assertTrue(self._call([self.acc1, self.acc2]) is None)
@ -339,12 +199,12 @@ class ChooseNamesTest(unittest.TestCase):
self._call(None)
self.assertEqual(mock_manual.call_count, 1)
@mock.patch("letsencrypt.display.ops.util")
@mock.patch("letsencrypt.display.ops.z_util")
def test_no_installer_cancel(self, mock_util):
mock_util().input.return_value = (display_util.CANCEL, [])
self.assertEqual(self._call(None), [])
@mock.patch("letsencrypt.display.ops.util")
@mock.patch("letsencrypt.display.ops.z_util")
def test_no_names_choose(self, mock_util):
self.mock_install().get_all_names.return_value = set()
mock_util().yesno.return_value = True
@ -355,14 +215,14 @@ class ChooseNamesTest(unittest.TestCase):
self.assertEqual(mock_util().input.call_count, 1)
self.assertEqual(actual_doms, [domain])
@mock.patch("letsencrypt.display.ops.util")
@mock.patch("letsencrypt.display.ops.z_util")
def test_no_names_quit(self, mock_util):
self.mock_install().get_all_names.return_value = set()
mock_util().yesno.return_value = False
self.assertEqual(self._call(self.mock_install), [])
@mock.patch("letsencrypt.display.ops.util")
@mock.patch("letsencrypt.display.ops.z_util")
def test_filter_names_valid_return(self, mock_util):
self.mock_install.get_all_names.return_value = set(["example.com"])
mock_util().checklist.return_value = (display_util.OK, ["example.com"])
@ -371,14 +231,14 @@ class ChooseNamesTest(unittest.TestCase):
self.assertEqual(names, ["example.com"])
self.assertEqual(mock_util().checklist.call_count, 1)
@mock.patch("letsencrypt.display.ops.util")
@mock.patch("letsencrypt.display.ops.z_util")
def test_filter_names_nothing_selected(self, mock_util):
self.mock_install.get_all_names.return_value = set(["example.com"])
mock_util().checklist.return_value = (display_util.OK, [])
self.assertEqual(self._call(self.mock_install), [])
@mock.patch("letsencrypt.display.ops.util")
@mock.patch("letsencrypt.display.ops.z_util")
def test_filter_names_cancel(self, mock_util):
self.mock_install.get_all_names.return_value = set(["example.com"])
mock_util().checklist.return_value = (
@ -397,7 +257,7 @@ class ChooseNamesTest(unittest.TestCase):
self.assertEqual(get_valid_domains(all_invalid), [])
self.assertEqual(len(get_valid_domains(two_valid)), 2)
@mock.patch("letsencrypt.display.ops.util")
@mock.patch("letsencrypt.display.ops.z_util")
def test_choose_manually(self, mock_util):
from letsencrypt.display.ops import _choose_names_manually
# No retry
@ -445,7 +305,7 @@ class SuccessInstallationTest(unittest.TestCase):
from letsencrypt.display.ops import success_installation
success_installation(names)
@mock.patch("letsencrypt.display.ops.util")
@mock.patch("letsencrypt.display.ops.z_util")
def test_success_installation(self, mock_util):
mock_util().notification.return_value = None
names = ["example.com", "abc.com"]
@ -467,7 +327,7 @@ class SuccessRenewalTest(unittest.TestCase):
from letsencrypt.display.ops import success_renewal
success_renewal(names, "renew")
@mock.patch("letsencrypt.display.ops.util")
@mock.patch("letsencrypt.display.ops.z_util")
def test_success_renewal(self, mock_util):
mock_util().notification.return_value = None
names = ["example.com", "abc.com"]

55
letsencrypt/tests/storage_test.py
View file

@ -493,7 +493,12 @@ class RenewableCertTests(BaseRenewableCertTest):
self.assertTrue(self.test_rc.should_autorenew())
mock_ocsp.return_value = False
def test_save_successor(self):
@mock.patch("letsencrypt.storage.relevant_values")
def test_save_successor(self, mock_rv):
# Mock relevant_values() to claim that all values are relevant here
# (to avoid instantiating parser)
mock_rv.side_effect = lambda x: x
for ver in xrange(1, 6):
for kind in ALL_FOUR:
where = getattr(self.test_rc, kind)
@ -557,8 +562,47 @@ class RenewableCertTests(BaseRenewableCertTest):
self.assertFalse(os.path.islink(self.test_rc.version("privkey", 10)))
self.assertFalse(os.path.exists(temp_config_file))
def test_new_lineage(self):
@mock.patch("letsencrypt.cli.helpful_parser")
def test_relevant_values(self, mock_parser):
"""Test that relevant_values() can reject an irrelevant value."""
# pylint: disable=protected-access
from letsencrypt import storage
mock_parser.verb = "certonly"
mock_parser.args = ["--standalone"]
mock_action = mock.Mock(dest="rsa_key_size", default=2048)
mock_parser.parser._actions = [mock_action]
self.assertEqual(storage.relevant_values({"hello": "there"}), {})
@mock.patch("letsencrypt.cli.helpful_parser")
def test_relevant_values_default(self, mock_parser):
"""Test that relevant_values() can reject a default value."""
# pylint: disable=protected-access
from letsencrypt import storage
mock_parser.verb = "certonly"
mock_parser.args = ["--standalone"]
mock_action = mock.Mock(dest="rsa_key_size", default=2048)
mock_parser.parser._actions = [mock_action]
self.assertEqual(storage.relevant_values({"rsa_key_size": 2048}), {})
@mock.patch("letsencrypt.cli.helpful_parser")
def test_relevant_values_nondefault(self, mock_parser):
"""Test that relevant_values() can retain a non-default value."""
# pylint: disable=protected-access
from letsencrypt import storage
mock_parser.verb = "certonly"
mock_parser.args = ["--standalone"]
mock_action = mock.Mock(dest="rsa_key_size", default=2048)
mock_parser.parser._actions = [mock_action]
self.assertEqual(storage.relevant_values({"rsa_key_size": 12}),
{"rsa_key_size": 12})
@mock.patch("letsencrypt.storage.relevant_values")
def test_new_lineage(self, mock_rv):
"""Test for new_lineage() class method."""
# Mock relevant_values to say everything is relevant here (so we
# don't have to mock the parser to help it decide!)
mock_rv.side_effect = lambda x: x
from letsencrypt import storage
result = storage.RenewableCert.new_lineage(
"the-lineage.com", "cert", "privkey", "chain", self.cli_config)
@ -592,8 +636,13 @@ class RenewableCertTests(BaseRenewableCertTest):
# TODO: Conceivably we could test that the renewal parameters actually
# got saved
def test_new_lineage_nonexistent_dirs(self):
@mock.patch("letsencrypt.storage.relevant_values")
def test_new_lineage_nonexistent_dirs(self, mock_rv):
"""Test that directories can be created if they don't exist."""
# Mock relevant_values to say everything is relevant here (so we
# don't have to mock the parser to help it decide!)
mock_rv.side_effect = lambda x: x
from letsencrypt import storage
shutil.rmtree(self.cli_config.renewal_configs_dir)
shutil.rmtree(self.cli_config.archive_dir)

2
letsencrypt/tests/testdata/sample-renewal-ancient.conf vendored
View file

@ -14,7 +14,7 @@ apache_dismod = a2dismod
register_unsafely_without_email = False
apache_handle_modules = True
uir = None
installer = none
installer = None
nginx_ctl = nginx
config_dir = MAGICDIR
text_mode = False

2
letsencrypt/tests/testdata/sample-renewal.conf vendored
View file

@ -14,7 +14,7 @@ apache_dismod = a2dismod
register_unsafely_without_email = False
apache_handle_modules = True
uir = None
installer = none
installer = None
nginx_ctl = nginx
config_dir = MAGICDIR
text_mode = False