starred/certbot-certbot
1
0
Fork 0
mirror of https://github.com/certbot/certbot.git synced 2026-07-28 16:54:59 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #2650 from letsencrypt/split-renew

Browse Source 
Split renew.py out of cli.py
This commit is contained in:
bmw
2016-03-23 16:31:55 -07:00
parent b46b54400f a9faa3b4ba
commit 764770f9e3
7 changed files with 349 additions and 320 deletions
Download Patch File Download Diff File Expand all files Collapse all files
letsencrypt/cli.py
+21 -296
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,19 +12,14 @@ 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
@@ -35,16 +28,7 @@ from letsencrypt.plugins import disco as plugins_disco
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:
@@ -115,21 +99,6 @@ 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
@@ -271,20 +240,20 @@ def record_chosen_plugins(config, plugins, auth, inst):
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)
@@ -312,264 +281,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,6 +313,10 @@ 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]
@@ -633,7 +358,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 +569,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 +844,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()