upstream/certbot
1
0
Fork 0
mirror of https://github.com/certbot/certbot.git synced 2026-05-28 04:34:11 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Merge pull request #2649 from letsencrypt/split-cli

Browse source 
Split main.py out of cli.py
This commit is contained in:
bmw 2016-03-17 16:06:13 -07:00
commit 8b2caa03b6
7 changed files with 803 additions and 806 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.pylintrc
View file

@ -38,7 +38,7 @@ load-plugins=linter_plugin
# --enable=similarities". If you want to run only the classes checker, but have
# no Warning level messages displayed, use"--disable=all --enable=classes
# --disable=W"
disable=fixme,locally-disabled,abstract-class-not-used,abstract-class-little-used,bad-continuation,too-few-public-methods,no-self-use,invalid-name,too-many-instance-attributes
disable=fixme,locally-disabled,abstract-class-not-used,abstract-class-little-used,bad-continuation,too-few-public-methods,no-self-use,invalid-name,too-many-instance-attributes,cyclic-import
# abstract-class-not-used cannot be disabled locally (at least in
# pylint 1.4.1), same for abstract-class-little-used

737
letsencrypt/cli.py
View file

@ -1,20 +1,14 @@
"""Let's Encrypt CLI."""
from __future__ import print_function
# TODO: Sanity check all input. Be sure to avoid shell code etc...
"""Let's Encrypt command line argument & config processing."""
# pylint: disable=too-many-lines
# (TODO: split this file into main.py and cli.py)
from __future__ import print_function
import argparse
import atexit
import copy
import functools
import glob
import json
import logging
import logging.handlers
import os
import sys
import time
import traceback
import configargparse
@ -24,27 +18,20 @@ import zope.component
import zope.interface.exceptions
import zope.interface.verify
from acme import jose
import letsencrypt
from letsencrypt import account
from letsencrypt import colored_logging
from letsencrypt import configuration
from letsencrypt import constants
from letsencrypt import client
from letsencrypt import crypto_util
from letsencrypt import errors
from letsencrypt import interfaces
from letsencrypt import le_util
from letsencrypt import log
from letsencrypt import reporter
from letsencrypt import storage
from letsencrypt.display import util as display_util
from letsencrypt.display import ops as display_ops
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
@ -128,158 +115,7 @@ def usage_strings(plugins):
return USAGE % (apache_doc, nginx_doc), SHORT_USAGE
def _find_domains(config, installer):
if not config.domains:
domains = display_ops.choose_names(installer)
# record in config.domains (so that it can be serialised in renewal config files),
# and set webroot_map entries if applicable
for d in domains:
_process_domain(config, d)
else:
domains = config.domains
if not domains:
raise errors.Error("Please specify --domains, or --installer that "
"will help in domain names autodiscovery")
return domains
def _determine_account(config):
"""Determine which account to use.
In order to make the renewer (configuration de/serialization) happy,
if ``config.account`` is ``None``, it will be updated based on the
user input. Same for ``config.email``.
:param argparse.Namespace config: CLI arguments
:param letsencrypt.interface.IConfig config: Configuration object
:param .AccountStorage account_storage: Account storage.
:returns: Account and optionally ACME client API (biproduct of new
registration).
:rtype: `tuple` of `letsencrypt.account.Account` and
`acme.client.Client`
"""
account_storage = account.AccountFileStorage(config)
acme = None
if config.account is not None:
acc = account_storage.load(config.account)
else:
accounts = account_storage.find_all()
if len(accounts) > 1:
acc = display_ops.choose_account(accounts)
elif len(accounts) == 1:
acc = accounts[0]
else: # no account registered yet
if config.email is None and not config.register_unsafely_without_email:
config.namespace.email = display_ops.get_email()
def _tos_cb(regr):
if config.tos:
return True
msg = ("Please read the Terms of Service at {0}. You "
"must agree in order to register with the ACME "
"server at {1}".format(
regr.terms_of_service, config.server))
obj = zope.component.getUtility(interfaces.IDisplay)
return obj.yesno(msg, "Agree", "Cancel", cli_flag="--agree-tos")
try:
acc, acme = client.register(
config, account_storage, tos_cb=_tos_cb)
except errors.MissingCommandlineFlag:
raise
except errors.Error as error:
logger.debug(error, exc_info=True)
raise errors.Error(
"Unable to register an account with ACME server")
config.namespace.account = acc.id
return acc, acme
def _init_le_client(config, authenticator, installer):
if authenticator is not None:
# if authenticator was given, then we will need account...
acc, acme = _determine_account(config)
logger.debug("Picked account: %r", acc)
# XXX
#crypto_util.validate_key_csr(acc.key)
else:
acc, acme = None, None
return client.Client(config, acc, authenticator, installer, acme=acme)
def _find_duplicative_certs(config, domains):
"""Find existing certs that duplicate the request."""
identical_names_cert, subset_names_cert = None, None
cli_config = configuration.RenewerConfiguration(config)
configs_dir = cli_config.renewal_configs_dir
# 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):
try:
candidate_lineage = storage.RenewableCert(renewal_file, cli_config)
except (errors.CertStorageError, IOError):
logger.warning("Renewal conf file %s is broken. Skipping.", renewal_file)
logger.debug("Traceback was:\n%s", traceback.format_exc())
continue
# TODO: Handle these differently depending on whether they are
# expired or still valid?
candidate_names = set(candidate_lineage.names())
if candidate_names == set(domains):
identical_names_cert = candidate_lineage
elif candidate_names.issubset(set(domains)):
# This logic finds and returns the largest subset-names cert
# in the case where there are several available.
if subset_names_cert is None:
subset_names_cert = candidate_lineage
elif len(candidate_names) > len(subset_names_cert.names()):
subset_names_cert = candidate_lineage
return identical_names_cert, subset_names_cert
def _treat_as_renewal(config, domains):
"""Determine whether there are duplicated names and how to handle
them (renew, reinstall, newcert, or raising an error to stop
the client run if the user chooses to cancel the operation when
prompted).
:returns: Two-element tuple containing desired new-certificate behavior as
a string token ("reinstall", "renew", or "newcert"), plus either
a RenewableCert instance or None if renewal shouldn't occur.
:raises .Error: If the user would like to rerun the client again.
"""
# Considering the possibility that the requested certificate is
# related to an existing certificate. (config.duplicate, which
# is set with --duplicate, skips all of this logic and forces any
# kind of certificate to be obtained with renewal = False.)
if config.duplicate:
return "newcert", None
# TODO: Also address superset case
ident_names_cert, subset_names_cert = _find_duplicative_certs(config, domains)
# XXX ^ schoen is not sure whether that correctly reads the systemwide
# configuration file.
if ident_names_cert is None and subset_names_cert is None:
return "newcert", None
if ident_names_cert is not None:
return _handle_identical_cert_request(config, ident_names_cert)
elif subset_names_cert is not None:
return _handle_subset_cert_request(config, domains, subset_names_cert)
def _should_renew(config, lineage):
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...")
@ -294,217 +130,6 @@ def _should_renew(config, lineage):
return False
def _handle_identical_cert_request(config, cert):
"""Figure out what to do if a cert has the same names as a previously obtained one
:param storage.RenewableCert cert:
:returns: Tuple of (string, cert_or_None) as per _treat_as_renewal
:rtype: tuple
"""
if _should_renew(config, cert):
return "renew", cert
if config.reinstall:
# Set with --reinstall, force an identical certificate to be
# reinstalled without further prompting.
return "reinstall", cert
question = (
"You have an existing certificate that contains exactly the same "
"domains you requested and isn't close to expiry."
"{br}(ref: {0}){br}{br}What would you like to do?"
).format(cert.configfile.filename, br=os.linesep)
if config.verb == "run":
keep_opt = "Attempt to reinstall this existing certificate"
elif config.verb == "certonly":
keep_opt = "Keep the existing certificate for now"
choices = [keep_opt,
"Renew & replace the cert (limit ~5 per 7 days)"]
display = zope.component.getUtility(interfaces.IDisplay)
response = display.menu(question, choices, "OK", "Cancel", default=0)
if response[0] == display_util.CANCEL:
# TODO: Add notification related to command-line options for
# skipping the menu for this case.
raise errors.Error(
"User chose to cancel the operation and may "
"reinvoke the client.")
elif response[1] == 0:
return "reinstall", cert
elif response[1] == 1:
return "renew", cert
else:
assert False, "This is impossible"
def _handle_subset_cert_request(config, domains, cert):
"""Figure out what to do if a previous cert had a subset of the names now requested
:param storage.RenewableCert cert:
:returns: Tuple of (string, cert_or_None) as per _treat_as_renewal
:rtype: tuple
"""
existing = ", ".join(cert.names())
question = (
"You have an existing certificate that contains a portion of "
"the domains you requested (ref: {0}){br}{br}It contains these "
"names: {1}{br}{br}You requested these names for the new "
"certificate: {2}.{br}{br}Do you want to expand and replace this existing "
"certificate with the new certificate?"
).format(cert.configfile.filename,
existing,
", ".join(domains),
br=os.linesep)
if config.expand or config.renew_by_default or zope.component.getUtility(
interfaces.IDisplay).yesno(question, "Expand", "Cancel",
cli_flag="--expand"):
return "renew", cert
else:
reporter_util = zope.component.getUtility(interfaces.IReporter)
reporter_util.add_message(
"To obtain a new certificate that contains these names without "
"replacing your existing certificate for {0}, you must use the "
"--duplicate option.{br}{br}"
"For example:{br}{br}{1} --duplicate {2}".format(
existing,
sys.argv[0], " ".join(sys.argv[1:]),
br=os.linesep
),
reporter_util.HIGH_PRIORITY)
raise errors.Error(
"User chose to cancel the operation and may "
"reinvoke the client.")
def _report_new_cert(cert_path, fullchain_path):
"""Reports the creation of a new certificate to the user.
:param str cert_path: path to cert
:param str fullchain_path: path to full chain
"""
expiry = crypto_util.notAfter(cert_path).date()
reporter_util = zope.component.getUtility(interfaces.IReporter)
if fullchain_path:
# Print the path to fullchain.pem because that's what modern webservers
# (Nginx and Apache2.4) will want.
and_chain = "and chain have"
path = fullchain_path
else:
# Unless we're in .csr mode and there really isn't one
and_chain = "has "
path = cert_path
# XXX Perhaps one day we could detect the presence of known old webservers
# and say something more informative here.
msg = ("Congratulations! Your certificate {0} been saved at {1}."
" Your cert will expire on {2}. To obtain a new version of the "
"certificate in the future, simply run Let's Encrypt again."
.format(and_chain, path, expiry))
reporter_util.add_message(msg, reporter_util.MEDIUM_PRIORITY)
def _suggest_donation_if_appropriate(config, action):
"""Potentially suggest a donation to support Let's Encrypt."""
if config.staging or config.verb == "renew":
# --dry-run implies --staging
return
if action not in ["renew", "newcert"]:
return
reporter_util = zope.component.getUtility(interfaces.IReporter)
msg = ("If you like Let's Encrypt, please consider supporting our work by:\n\n"
"Donating to ISRG / Let's Encrypt: https://letsencrypt.org/donate\n"
"Donating to EFF: https://eff.org/donate-le\n\n")
reporter_util.add_message(msg, reporter_util.LOW_PRIORITY)
def _report_successful_dry_run(config):
reporter_util = zope.component.getUtility(interfaces.IReporter)
if config.verb != "renew":
reporter_util.add_message("The dry run was successful.",
reporter_util.HIGH_PRIORITY, on_crash=False)
def _auth_from_domains(le_client, config, domains, lineage=None):
"""Authenticate and enroll certificate."""
# Note: This can raise errors... caught above us though. This is now
# a three-way case: reinstall (which results in a no-op here because
# although there is a relevant lineage, we don't do anything to it
# inside this function -- we don't obtain a new certificate), renew
# (which results in treating the request as a renewal), or newcert
# (which results in treating the request as a new certificate request).
# If lineage is specified, use that one instead of looking around for
# a matching one.
if lineage is None:
# This will find a relevant matching lineage that exists
action, lineage = _treat_as_renewal(config, domains)
else:
# Renewal, where we already know the specific lineage we're
# interested in
action = "renew"
if action == "reinstall":
# The lineage already exists; allow the caller to try installing
# it without getting a new certificate at all.
return lineage, "reinstall"
elif action == "renew":
original_server = lineage.configuration["renewalparams"]["server"]
_avoid_invalidating_lineage(config, lineage, original_server)
# TODO: schoen wishes to reuse key - discussion
# https://github.com/letsencrypt/letsencrypt/pull/777/files#r40498574
new_certr, new_chain, new_key, _ = le_client.obtain_certificate(domains)
# TODO: Check whether it worked! <- or make sure errors are thrown (jdk)
if config.dry_run:
logger.info("Dry run: skipping updating lineage at %s",
os.path.dirname(lineage.cert))
else:
lineage.save_successor(
lineage.latest_common_version(), OpenSSL.crypto.dump_certificate(
OpenSSL.crypto.FILETYPE_PEM, new_certr.body.wrapped),
new_key.pem, crypto_util.dump_pyopenssl_chain(new_chain),
configuration.RenewerConfiguration(config.namespace))
lineage.update_all_links_to(lineage.latest_common_version())
# TODO: Check return value of save_successor
# TODO: Also update lineage renewal config with any relevant
# configuration values from this attempt? <- Absolutely (jdkasten)
elif action == "newcert":
# TREAT AS NEW REQUEST
lineage = le_client.obtain_and_enroll_certificate(domains)
if lineage is False:
raise errors.Error("Certificate could not be obtained")
if not config.dry_run and not config.verb == "renew":
_report_new_cert(lineage.cert, lineage.fullchain)
return lineage, action
def _avoid_invalidating_lineage(config, lineage, original_server):
"Do not renew a valid cert with one from a staging server!"
def _is_staging(srv):
return srv == constants.STAGING_URI or "staging" in srv
# Some lineages may have begun with --staging, but then had production certs
# added to them
latest_cert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM,
open(lineage.cert).read())
# all our test certs are from happy hacker fake CA, though maybe one day
# we should test more methodically
now_valid = "fake" not in repr(latest_cert.get_issuer()).lower()
if _is_staging(config.server):
if not _is_staging(original_server) or now_valid:
if not config.break_my_certs:
names = ", ".join(lineage.names())
raise errors.Error(
"You've asked to renew/replace a seemingly valid certificate with "
"a test certificate (domains: {0}). We will not do that "
"unless you use the --break-my-certs flag!".format(names))
def diagnose_configurator_problem(cfg_type, requested, plugins):
"""
Raise the most helpful error message about a plugin being unavailable
@ -646,103 +271,6 @@ def record_chosen_plugins(config, plugins, auth, inst):
cn.installer = plugins.find_init(inst).name if inst else "none"
# TODO: Make run as close to auth + install as possible
# Possible difficulties: config.csr was hacked into auth
def run(config, plugins): # pylint: disable=too-many-branches,too-many-locals
"""Obtain a certificate and install."""
try:
installer, authenticator = choose_configurator_plugins(config, plugins, "run")
except errors.PluginSelectionError as e:
return e.message
domains = _find_domains(config, installer)
# TODO: Handle errors from _init_le_client?
le_client = _init_le_client(config, authenticator, installer)
lineage, action = _auth_from_domains(le_client, config, domains)
le_client.deploy_certificate(
domains, lineage.privkey, lineage.cert,
lineage.chain, lineage.fullchain)
le_client.enhance_config(domains, config)
if len(lineage.available_versions("cert")) == 1:
display_ops.success_installation(domains)
else:
display_ops.success_renewal(domains, action)
_suggest_donation_if_appropriate(config, action)
def obtain_cert(config, plugins, lineage=None):
"""Implements "certonly": authenticate & obtain cert, but do not install it."""
# pylint: disable=too-many-locals
try:
# installers are used in auth mode to determine domain names
installer, authenticator = choose_configurator_plugins(config, plugins, "certonly")
except errors.PluginSelectionError as e:
logger.info("Could not choose appropriate plugin: %s", e)
raise
# TODO: Handle errors from _init_le_client?
le_client = _init_le_client(config, authenticator, installer)
action = "newcert"
# This is a special case; cert and chain are simply saved
if config.csr is not None:
assert lineage is None, "Did not expect a CSR with a RenewableCert"
csr, typ = config.actual_csr
certr, chain = le_client.obtain_certificate_from_csr(config.domains, csr, typ)
if config.dry_run:
logger.info(
"Dry run: skipping saving certificate to %s", config.cert_path)
else:
cert_path, _, cert_fullchain = le_client.save_certificate(
certr, chain, config.cert_path, config.chain_path, config.fullchain_path)
_report_new_cert(cert_path, cert_fullchain)
else:
domains = _find_domains(config, installer)
_, action = _auth_from_domains(le_client, config, domains, lineage)
if config.dry_run:
_report_successful_dry_run(config)
elif config.verb == "renew":
if installer is None:
# Tell the user that the server was not restarted.
print("new certificate deployed without reload, fullchain is",
lineage.fullchain)
else:
# In case of a renewal, reload server to pick up new certificate.
# In principle we could have a configuration option to inhibit this
# from happening.
installer.restart()
print("new certificate deployed with reload of",
config.installer, "server; fullchain is", lineage.fullchain)
_suggest_donation_if_appropriate(config, action)
def install(config, plugins):
"""Install a previously obtained cert in a server."""
# XXX: Update for renewer/RenewableCert
# FIXME: be consistent about whether errors are raised or returned from
# this function ...
try:
installer, _ = choose_configurator_plugins(config, plugins, "install")
except errors.PluginSelectionError as e:
return e.message
domains = _find_domains(config, installer)
le_client = _init_le_client(config, authenticator=None, installer=installer)
assert config.cert_path is not None # required=True in the subparser
le_client.deploy_certificate(
domains, config.key_path, config.cert_path, config.chain_path,
config.fullchain_path)
le_client.enhance_config(domains, config)
def _set_by_cli(var):
"""
Return True if a particular config variable has been set by the user
@ -932,7 +460,7 @@ def _reconstitute(config, full_path):
try:
for d in renewal_candidate.names():
_process_domain(config, d)
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 "
@ -1019,9 +547,10 @@ def renew(config, unused_plugins):
else:
# XXX: ensure that each call here replaces the previous one
zope.component.provideUtility(lineage_config)
if _should_renew(lineage_config, renewal_candidate):
if should_renew(lineage_config, renewal_candidate):
plugins = plugins_disco.PluginsRegistry.find_all()
obtain_cert(lineage_config, plugins, renewal_candidate)
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)
@ -1042,64 +571,6 @@ def renew(config, unused_plugins):
else:
logger.debug("no renewal failures")
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"
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])
key = jose.JWK.load(config.key_path[1])
else: # revocation by account key
logger.debug("Revoking %s using Account Key", config.cert_path[0])
acc, _ = _determine_account(config)
key = acc.key
acme = client.acme_from_config_key(config, key)
cert = crypto_util.pyopenssl_load_certificate(config.cert_path[1])[0]
acme.revoke(jose.ComparableX509(cert))
def rollback(config, plugins):
"""Rollback server configuration changes made during install."""
client.rollback(config.installer, config.checkpoints, config, plugins)
def config_changes(config, unused_plugins):
"""Show changes made to server config during installation
View checkpoints and associated configuration changes.
"""
client.view_config_changes(config, num=config.num)
def plugins_cmd(config, plugins): # TODO: Use IDisplay rather than print
"""List server software plugins."""
logger.debug("Expected interfaces: %s", config.ifaces)
ifaces = [] if config.ifaces is None else config.ifaces
filtered = plugins.visible().ifaces(ifaces)
logger.debug("Filtered plugins: %r", filtered)
if not config.init and not config.prepare:
print(str(filtered))
return
filtered.init(config)
verified = filtered.verify(ifaces)
logger.debug("Verified plugins: %r", verified)
if not config.prepare:
print(str(verified))
return
verified.prepare()
available = verified.available()
logger.debug("Prepared plugins: %s", available)
print(str(available))
def read_file(filename, mode="rb"):
"""Returns the given file's contents.
@ -1148,7 +619,6 @@ class SilentParser(object): # pylint: disable=too-few-public-methods
kwargs["help"] = argparse.SUPPRESS
self.parser.add_argument(*args, **kwargs)
class HelpfulArgumentParser(object):
"""Argparse Wrapper.
@ -1158,19 +628,19 @@ class HelpfulArgumentParser(object):
"""
# Maps verbs/subcommands to the functions that implement them
VERBS = {"auth": obtain_cert, "certonly": obtain_cert,
"config_changes": config_changes, "everything": run,
"install": install, "plugins": plugins_cmd, "renew": renew,
"revoke": revoke, "rollback": rollback, "run": run}
# List of topics for which additional help can be provided
HELP_TOPICS = ["all", "security",
"paths", "automation", "testing"] + list(six.iterkeys(VERBS))
def __init__(self, args, plugins, detect_defaults=False):
plugin_names = list(six.iterkeys(plugins))
self.help_topics = self.HELP_TOPICS + plugin_names + [None]
from letsencrypt import main
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,
"rollback": main.rollback, "everything": main.run}
# List of topics for which additional help can be provided
HELP_TOPICS = ["all", "security", "paths", "automation", "testing"] + list(self.VERBS)
plugin_names = list(plugins)
self.help_topics = HELP_TOPICS + plugin_names + [None]
usage, short_usage = usage_strings(plugins)
self.parser = configargparse.ArgParser(
usage=short_usage,
@ -1254,7 +724,7 @@ class HelpfulArgumentParser(object):
def handle_csr(self, parsed_args):
"""
Process a --csr flag. This needs to happen early enough that the
webroot plugin can know about the calls to _process_domain
webroot plugin can know about the calls to process_domain
"""
if parsed_args.verb != "certonly":
raise errors.Error("Currently, a CSR file may only be specified "
@ -1277,7 +747,7 @@ class HelpfulArgumentParser(object):
logger.debug("PEM CSR parse error %s", traceback.format_exc())
raise errors.Error("Failed to parse CSR file: {0}".format(parsed_args.csr[0]))
for d in domains:
_process_domain(parsed_args, d)
process_domain(parsed_args, d)
for d in domains:
sanitised = le_util.enforce_domain_sanity(d)
@ -1813,7 +1283,7 @@ class WebrootPathProcessor(argparse.Action): # pylint: disable=missing-docstrin
args.webroot_path.append(webroot)
def _process_domain(args_or_config, domain_arg, webroot_path=None):
def process_domain(args_or_config, domain_arg, webroot_path=None):
"""
Process a new -d flag, helping the webroot plugin construct a map of
{domain : webrootpath} if -w / --webroot-path is in use
@ -1839,165 +1309,10 @@ class WebrootMapProcessor(argparse.Action): # pylint: disable=missing-docstring
def __call__(self, parser, args, webroot_map_arg, option_string=None):
webroot_map = json.loads(webroot_map_arg)
for domains, webroot_path in six.iteritems(webroot_map):
_process_domain(args, domains, [webroot_path])
process_domain(args, domains, [webroot_path])
class DomainFlagProcessor(argparse.Action): # pylint: disable=missing-docstring
def __call__(self, parser, args, domain_arg, option_string=None):
"""Just wrap _process_domain in argparseese."""
_process_domain(args, domain_arg)
def setup_log_file_handler(config, logfile, fmt):
"""Setup file debug logging."""
log_file_path = os.path.join(config.logs_dir, logfile)
handler = logging.handlers.RotatingFileHandler(
log_file_path, maxBytes=2 ** 20, backupCount=10)
# rotate on each invocation, rollover only possible when maxBytes
# is nonzero and backupCount is nonzero, so we set maxBytes as big
# as possible not to overrun in single CLI invocation (1MB).
handler.doRollover() # TODO: creates empty letsencrypt.log.1 file
handler.setLevel(logging.DEBUG)
handler_formatter = logging.Formatter(fmt=fmt)
handler_formatter.converter = time.gmtime # don't use localtime
handler.setFormatter(handler_formatter)
return handler, log_file_path
def _cli_log_handler(config, level, fmt):
if config.text_mode or config.noninteractive_mode or config.verb == "renew":
handler = colored_logging.StreamHandler()
handler.setFormatter(logging.Formatter(fmt))
else:
handler = log.DialogHandler()
# dialog box is small, display as less as possible
handler.setFormatter(logging.Formatter("%(message)s"))
handler.setLevel(level)
return handler
def setup_logging(config, cli_handler_factory, logfile):
"""Setup logging."""
fmt = "%(asctime)s:%(levelname)s:%(name)s:%(message)s"
level = -config.verbose_count * 10
file_handler, log_file_path = setup_log_file_handler(
config, logfile=logfile, fmt=fmt)
cli_handler = cli_handler_factory(config, level, fmt)
# TODO: use fileConfig?
root_logger = logging.getLogger()
root_logger.setLevel(logging.DEBUG) # send all records to handlers
root_logger.addHandler(cli_handler)
root_logger.addHandler(file_handler)
logger.debug("Root logging level set at %d", level)
logger.info("Saving debug log to %s", log_file_path)
def _handle_exception(exc_type, exc_value, trace, config):
"""Logs exceptions and reports them to the user.
Config is used to determine how to display exceptions to the user. In
general, if config.debug is True, then the full exception and traceback is
shown to the user, otherwise it is suppressed. If config itself is None,
then the traceback and exception is attempted to be written to a logfile.
If this is successful, the traceback is suppressed, otherwise it is shown
to the user. sys.exit is always called with a nonzero status.
"""
logger.debug(
"Exiting abnormally:%s%s",
os.linesep,
"".join(traceback.format_exception(exc_type, exc_value, trace)))
if issubclass(exc_type, Exception) and (config is None or not config.debug):
if config is None:
logfile = "letsencrypt.log"
try:
with open(logfile, "w") as logfd:
traceback.print_exception(
exc_type, exc_value, trace, file=logfd)
except: # pylint: disable=bare-except
sys.exit("".join(
traceback.format_exception(exc_type, exc_value, trace)))
if issubclass(exc_type, errors.Error):
sys.exit(exc_value)
else:
# Here we're passing a client or ACME error out to the client at the shell
# Tell the user a bit about what happened, without overwhelming
# them with a full traceback
err = traceback.format_exception_only(exc_type, exc_value)[0]
# Typical error from the ACME module:
# acme.messages.Error: urn:acme:error:malformed :: The request message was
# malformed :: Error creating new registration :: Validation of contact
# mailto:none@longrandomstring.biz failed: Server failure at resolver
if (("urn:acme" in err and ":: " in err and
config.verbose_count <= flag_default("verbose_count"))):
# prune ACME error code, we have a human description
_code, _sep, err = err.partition(":: ")
msg = "An unexpected error occurred:\n" + err + "Please see the "
if config is None:
msg += "logfile '{0}' for more details.".format(logfile)
else:
msg += "logfiles in {0} for more details.".format(config.logs_dir)
sys.exit(msg)
else:
sys.exit("".join(
traceback.format_exception(exc_type, exc_value, trace)))
def main(cli_args=sys.argv[1:]):
"""Command line argument parsing and main script execution."""
sys.excepthook = functools.partial(_handle_exception, config=None)
plugins = plugins_disco.PluginsRegistry.find_all()
# note: arg parser internally handles --help (and exits afterwards)
args = prepare_and_parse_args(plugins, cli_args)
config = configuration.NamespaceConfig(args)
zope.component.provideUtility(config)
# Setup logging ASAP, otherwise "No handlers could be found for
# logger ..." TODO: this should be done before plugins discovery
for directory in config.config_dir, config.work_dir:
le_util.make_or_verify_dir(
directory, constants.CONFIG_DIRS_MODE, os.geteuid(),
"--strict-permissions" in cli_args)
# TODO: logs might contain sensitive data such as contents of the
# private key! #525
le_util.make_or_verify_dir(
config.logs_dir, 0o700, os.geteuid(), "--strict-permissions" in cli_args)
setup_logging(config, _cli_log_handler, logfile='letsencrypt.log')
logger.debug("letsencrypt version: %s", letsencrypt.__version__)
# do not log `config`, as it contains sensitive data (e.g. revoke --key)!
logger.debug("Arguments: %r", cli_args)
logger.debug("Discovered plugins: %r", plugins)
sys.excepthook = functools.partial(_handle_exception, config=config)
# Displayer
if config.noninteractive_mode:
displayer = display_util.NoninteractiveDisplay(sys.stdout)
elif config.text_mode:
displayer = display_util.FileDisplay(sys.stdout)
elif config.verb == "renew":
config.noninteractive_mode = True
displayer = display_util.NoninteractiveDisplay(sys.stdout)
else:
displayer = display_util.NcursesDisplay()
zope.component.provideUtility(displayer)
# Reporter
report = reporter.Reporter()
zope.component.provideUtility(report)
atexit.register(report.atexit_print_messages)
return config.func(config, plugins)
if __name__ == "__main__":
err_string = main()
if err_string:
logger.warn("Exiting with message %s", err_string)
sys.exit(err_string) # pragma: no cover
"""Just wrap process_domain in argparseese."""
process_domain(args, domain_arg)

709
letsencrypt/main.py Normal file
View file

@ -0,0 +1,709 @@
"""Let's Encrypt main entry point."""
from __future__ import print_function
import atexit
import functools
import logging.handlers
import os
import sys
import time
import traceback
import OpenSSL
import zope.component
from acme import jose
import letsencrypt
from letsencrypt import account
from letsencrypt import client
from letsencrypt import cli
from letsencrypt import crypto_util
from letsencrypt import colored_logging
from letsencrypt import configuration
from letsencrypt import constants
from letsencrypt import errors
from letsencrypt import interfaces
from letsencrypt import le_util
from letsencrypt import log
from letsencrypt import reporter
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
logger = logging.getLogger(__name__)
def _suggest_donation_if_appropriate(config, action):
"""Potentially suggest a donation to support Let's Encrypt."""
if config.staging or config.verb == "renew":
# --dry-run implies --staging
return
if action not in ["renew", "newcert"]:
return
reporter_util = zope.component.getUtility(interfaces.IReporter)
msg = ("If you like Let's Encrypt, please consider supporting our work by:\n\n"
"Donating to ISRG / Let's Encrypt: https://letsencrypt.org/donate\n"
"Donating to EFF: https://eff.org/donate-le\n\n")
reporter_util.add_message(msg, reporter_util.LOW_PRIORITY)
def _avoid_invalidating_lineage(config, lineage, original_server):
"Do not renew a valid cert with one from a staging server!"
def _is_staging(srv):
return srv == constants.STAGING_URI or "staging" in srv
# Some lineages may have begun with --staging, but then had production certs
# added to them
latest_cert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM,
open(lineage.cert).read())
# all our test certs are from happy hacker fake CA, though maybe one day
# we should test more methodically
now_valid = "fake" not in repr(latest_cert.get_issuer()).lower()
if _is_staging(config.server):
if not _is_staging(original_server) or now_valid:
if not config.break_my_certs:
names = ", ".join(lineage.names())
raise errors.Error(
"You've asked to renew/replace a seemingly valid certificate with "
"a test certificate (domains: {0}). We will not do that "
"unless you use the --break-my-certs flag!".format(names))
def _report_successful_dry_run(config):
reporter_util = zope.component.getUtility(interfaces.IReporter)
if config.verb != "renew":
reporter_util.add_message("The dry run was successful.",
reporter_util.HIGH_PRIORITY, on_crash=False)
def _auth_from_domains(le_client, config, domains, lineage=None):
"""Authenticate and enroll certificate."""
# Note: This can raise errors... caught above us though. This is now
# a three-way case: reinstall (which results in a no-op here because
# although there is a relevant lineage, we don't do anything to it
# inside this function -- we don't obtain a new certificate), renew
# (which results in treating the request as a renewal), or newcert
# (which results in treating the request as a new certificate request).
# If lineage is specified, use that one instead of looking around for
# a matching one.
if lineage is None:
# This will find a relevant matching lineage that exists
action, lineage = _treat_as_renewal(config, domains)
else:
# Renewal, where we already know the specific lineage we're
# interested in
action = "renew"
if action == "reinstall":
# The lineage already exists; allow the caller to try installing
# it without getting a new certificate at all.
return lineage, "reinstall"
elif action == "renew":
original_server = lineage.configuration["renewalparams"]["server"]
_avoid_invalidating_lineage(config, lineage, original_server)
# TODO: schoen wishes to reuse key - discussion
# https://github.com/letsencrypt/letsencrypt/pull/777/files#r40498574
new_certr, new_chain, new_key, _ = le_client.obtain_certificate(domains)
# TODO: Check whether it worked! <- or make sure errors are thrown (jdk)
if config.dry_run:
logger.info("Dry run: skipping updating lineage at %s",
os.path.dirname(lineage.cert))
else:
lineage.save_successor(
lineage.latest_common_version(), OpenSSL.crypto.dump_certificate(
OpenSSL.crypto.FILETYPE_PEM, new_certr.body.wrapped),
new_key.pem, crypto_util.dump_pyopenssl_chain(new_chain),
configuration.RenewerConfiguration(config.namespace))
lineage.update_all_links_to(lineage.latest_common_version())
# TODO: Check return value of save_successor
# TODO: Also update lineage renewal config with any relevant
# configuration values from this attempt? <- Absolutely (jdkasten)
elif action == "newcert":
# TREAT AS NEW REQUEST
lineage = le_client.obtain_and_enroll_certificate(domains)
if lineage is False:
raise errors.Error("Certificate could not be obtained")
if not config.dry_run and not config.verb == "renew":
_report_new_cert(lineage.cert, lineage.fullchain)
return lineage, action
def _handle_subset_cert_request(config, domains, cert):
"""Figure out what to do if a previous cert had a subset of the names now requested
:param storage.RenewableCert cert:
:returns: Tuple of (string, cert_or_None) as per _treat_as_renewal
:rtype: tuple
"""
existing = ", ".join(cert.names())
question = (
"You have an existing certificate that contains a portion of "
"the domains you requested (ref: {0}){br}{br}It contains these "
"names: {1}{br}{br}You requested these names for the new "
"certificate: {2}.{br}{br}Do you want to expand and replace this existing "
"certificate with the new certificate?"
).format(cert.configfile.filename,
existing,
", ".join(domains),
br=os.linesep)
if config.expand or config.renew_by_default or zope.component.getUtility(
interfaces.IDisplay).yesno(question, "Expand", "Cancel",
cli_flag="--expand"):
return "renew", cert
else:
reporter_util = zope.component.getUtility(interfaces.IReporter)
reporter_util.add_message(
"To obtain a new certificate that contains these names without "
"replacing your existing certificate for {0}, you must use the "
"--duplicate option.{br}{br}"
"For example:{br}{br}{1} --duplicate {2}".format(
existing,
sys.argv[0], " ".join(sys.argv[1:]),
br=os.linesep
),
reporter_util.HIGH_PRIORITY)
raise errors.Error(
"User chose to cancel the operation and may "
"reinvoke the client.")
def _handle_identical_cert_request(config, cert):
"""Figure out what to do if a cert has the same names as a previously obtained one
:param storage.RenewableCert cert:
:returns: Tuple of (string, cert_or_None) as per _treat_as_renewal
:rtype: tuple
"""
if should_renew(config, cert):
return "renew", cert
if config.reinstall:
# Set with --reinstall, force an identical certificate to be
# reinstalled without further prompting.
return "reinstall", cert
question = (
"You have an existing certificate that contains exactly the same "
"domains you requested and isn't close to expiry."
"{br}(ref: {0}){br}{br}What would you like to do?"
).format(cert.configfile.filename, br=os.linesep)
if config.verb == "run":
keep_opt = "Attempt to reinstall this existing certificate"
elif config.verb == "certonly":
keep_opt = "Keep the existing certificate for now"
choices = [keep_opt,
"Renew & replace the cert (limit ~5 per 7 days)"]
display = zope.component.getUtility(interfaces.IDisplay)
response = display.menu(question, choices, "OK", "Cancel", default=0)
if response[0] == display_util.CANCEL:
# TODO: Add notification related to command-line options for
# skipping the menu for this case.
raise errors.Error(
"User chose to cancel the operation and may "
"reinvoke the client.")
elif response[1] == 0:
return "reinstall", cert
elif response[1] == 1:
return "renew", cert
else:
assert False, "This is impossible"
def _treat_as_renewal(config, domains):
"""Determine whether there are duplicated names and how to handle
them (renew, reinstall, newcert, or raising an error to stop
the client run if the user chooses to cancel the operation when
prompted).
:returns: Two-element tuple containing desired new-certificate behavior as
a string token ("reinstall", "renew", or "newcert"), plus either
a RenewableCert instance or None if renewal shouldn't occur.
:raises .Error: If the user would like to rerun the client again.
"""
# Considering the possibility that the requested certificate is
# related to an existing certificate. (config.duplicate, which
# is set with --duplicate, skips all of this logic and forces any
# kind of certificate to be obtained with renewal = False.)
if config.duplicate:
return "newcert", None
# TODO: Also address superset case
ident_names_cert, subset_names_cert = _find_duplicative_certs(config, domains)
# XXX ^ schoen is not sure whether that correctly reads the systemwide
# configuration file.
if ident_names_cert is None and subset_names_cert is None:
return "newcert", None
if ident_names_cert is not None:
return _handle_identical_cert_request(config, ident_names_cert)
elif subset_names_cert is not None:
return _handle_subset_cert_request(config, domains, subset_names_cert)
def _find_duplicative_certs(config, domains):
"""Find existing certs that duplicate the request."""
identical_names_cert, subset_names_cert = None, None
cli_config = configuration.RenewerConfiguration(config)
configs_dir = cli_config.renewal_configs_dir
# 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):
try:
candidate_lineage = storage.RenewableCert(renewal_file, cli_config)
except (errors.CertStorageError, IOError):
logger.warning("Renewal conf file %s is broken. Skipping.", renewal_file)
logger.debug("Traceback was:\n%s", traceback.format_exc())
continue
# TODO: Handle these differently depending on whether they are
# expired or still valid?
candidate_names = set(candidate_lineage.names())
if candidate_names == set(domains):
identical_names_cert = candidate_lineage
elif candidate_names.issubset(set(domains)):
# This logic finds and returns the largest subset-names cert
# in the case where there are several available.
if subset_names_cert is None:
subset_names_cert = candidate_lineage
elif len(candidate_names) > len(subset_names_cert.names()):
subset_names_cert = candidate_lineage
return identical_names_cert, subset_names_cert
def _find_domains(config, installer):
if not config.domains:
domains = display_ops.choose_names(installer)
# record in config.domains (so that it can be serialised in renewal config files),
# and set webroot_map entries if applicable
for d in domains:
cli.process_domain(config, d)
else:
domains = config.domains
if not domains:
raise errors.Error("Please specify --domains, or --installer that "
"will help in domain names autodiscovery")
return domains
def _report_new_cert(cert_path, fullchain_path):
"""Reports the creation of a new certificate to the user.
:param str cert_path: path to cert
:param str fullchain_path: path to full chain
"""
expiry = crypto_util.notAfter(cert_path).date()
reporter_util = zope.component.getUtility(interfaces.IReporter)
if fullchain_path:
# Print the path to fullchain.pem because that's what modern webservers
# (Nginx and Apache2.4) will want.
and_chain = "and chain have"
path = fullchain_path
else:
# Unless we're in .csr mode and there really isn't one
and_chain = "has "
path = cert_path
# XXX Perhaps one day we could detect the presence of known old webservers
# and say something more informative here.
msg = ("Congratulations! Your certificate {0} been saved at {1}."
" Your cert will expire on {2}. To obtain a new version of the "
"certificate in the future, simply run Let's Encrypt again."
.format(and_chain, path, expiry))
reporter_util.add_message(msg, reporter_util.MEDIUM_PRIORITY)
def _determine_account(config):
"""Determine which account to use.
In order to make the renewer (configuration de/serialization) happy,
if ``config.account`` is ``None``, it will be updated based on the
user input. Same for ``config.email``.
:param argparse.Namespace config: CLI arguments
:param letsencrypt.interface.IConfig config: Configuration object
:param .AccountStorage account_storage: Account storage.
:returns: Account and optionally ACME client API (biproduct of new
registration).
:rtype: `tuple` of `letsencrypt.account.Account` and
`acme.client.Client`
"""
account_storage = account.AccountFileStorage(config)
acme = None
if config.account is not None:
acc = account_storage.load(config.account)
else:
accounts = account_storage.find_all()
if len(accounts) > 1:
acc = display_ops.choose_account(accounts)
elif len(accounts) == 1:
acc = accounts[0]
else: # no account registered yet
if config.email is None and not config.register_unsafely_without_email:
config.namespace.email = display_ops.get_email()
def _tos_cb(regr):
if config.tos:
return True
msg = ("Please read the Terms of Service at {0}. You "
"must agree in order to register with the ACME "
"server at {1}".format(
regr.terms_of_service, config.server))
obj = zope.component.getUtility(interfaces.IDisplay)
return obj.yesno(msg, "Agree", "Cancel", cli_flag="--agree-tos")
try:
acc, acme = client.register(
config, account_storage, tos_cb=_tos_cb)
except errors.MissingCommandlineFlag:
raise
except errors.Error as error:
logger.debug(error, exc_info=True)
raise errors.Error(
"Unable to register an account with ACME server")
config.namespace.account = acc.id
return acc, acme
def _init_le_client(config, authenticator, installer):
if authenticator is not None:
# if authenticator was given, then we will need account...
acc, acme = _determine_account(config)
logger.debug("Picked account: %r", acc)
# XXX
#crypto_util.validate_key_csr(acc.key)
else:
acc, acme = None, None
return client.Client(config, acc, authenticator, installer, acme=acme)
def install(config, plugins):
"""Install a previously obtained cert in a server."""
# XXX: Update for renewer/RenewableCert
# FIXME: be consistent about whether errors are raised or returned from
# this function ...
try:
installer, _ = choose_configurator_plugins(config, plugins, "install")
except errors.PluginSelectionError as e:
return e.message
domains = _find_domains(config, installer)
le_client = _init_le_client(config, authenticator=None, installer=installer)
assert config.cert_path is not None # required=True in the subparser
le_client.deploy_certificate(
domains, config.key_path, config.cert_path, config.chain_path,
config.fullchain_path)
le_client.enhance_config(domains, config)
def plugins_cmd(config, plugins): # TODO: Use IDisplay rather than print
"""List server software plugins."""
logger.debug("Expected interfaces: %s", config.ifaces)
ifaces = [] if config.ifaces is None else config.ifaces
filtered = plugins.visible().ifaces(ifaces)
logger.debug("Filtered plugins: %r", filtered)
if not config.init and not config.prepare:
print(str(filtered))
return
filtered.init(config)
verified = filtered.verify(ifaces)
logger.debug("Verified plugins: %r", verified)
if not config.prepare:
print(str(verified))
return
verified.prepare()
available = verified.available()
logger.debug("Prepared plugins: %s", available)
print(str(available))
def rollback(config, plugins):
"""Rollback server configuration changes made during install."""
client.rollback(config.installer, config.checkpoints, config, plugins)
def config_changes(config, unused_plugins):
"""Show changes made to server config during installation
View checkpoints and associated configuration changes.
"""
client.view_config_changes(config, num=config.num)
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"
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])
key = jose.JWK.load(config.key_path[1])
else: # revocation by account key
logger.debug("Revoking %s using Account Key", config.cert_path[0])
acc, _ = _determine_account(config)
key = acc.key
acme = client.acme_from_config_key(config, key)
cert = crypto_util.pyopenssl_load_certificate(config.cert_path[1])[0]
acme.revoke(jose.ComparableX509(cert))
def run(config, plugins): # pylint: disable=too-many-branches,too-many-locals
"""Obtain a certificate and install."""
# 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")
except errors.PluginSelectionError as e:
return e.message
domains = _find_domains(config, installer)
# TODO: Handle errors from _init_le_client?
le_client = _init_le_client(config, authenticator, installer)
lineage, action = _auth_from_domains(le_client, config, domains)
le_client.deploy_certificate(
domains, lineage.privkey, lineage.cert,
lineage.chain, lineage.fullchain)
le_client.enhance_config(domains, config)
if len(lineage.available_versions("cert")) == 1:
display_ops.success_installation(domains)
else:
display_ops.success_renewal(domains, action)
_suggest_donation_if_appropriate(config, action)
def obtain_cert(config, plugins, lineage=None):
"""Implements "certonly": authenticate & obtain cert, but do not install it."""
# pylint: disable=too-many-locals
try:
# installers are used in auth mode to determine domain names
installer, authenticator = choose_configurator_plugins(config, plugins, "certonly")
except errors.PluginSelectionError as e:
logger.info("Could not choose appropriate plugin: %s", e)
raise
# TODO: Handle errors from _init_le_client?
le_client = _init_le_client(config, authenticator, installer)
action = "newcert"
# This is a special case; cert and chain are simply saved
if config.csr is not None:
assert lineage is None, "Did not expect a CSR with a RenewableCert"
csr, typ = config.actual_csr
certr, chain = le_client.obtain_certificate_from_csr(config.domains, csr, typ)
if config.dry_run:
logger.info(
"Dry run: skipping saving certificate to %s", config.cert_path)
else:
cert_path, _, cert_fullchain = le_client.save_certificate(
certr, chain, config.cert_path, config.chain_path, config.fullchain_path)
_report_new_cert(cert_path, cert_fullchain)
else:
domains = _find_domains(config, installer)
_, action = _auth_from_domains(le_client, config, domains, lineage)
if config.dry_run:
_report_successful_dry_run(config)
elif config.verb == "renew":
if installer is None:
# Tell the user that the server was not restarted.
print("new certificate deployed without reload, fullchain is",
lineage.fullchain)
else:
# In case of a renewal, reload server to pick up new certificate.
# In principle we could have a configuration option to inhibit this
# from happening.
installer.restart()
print("new certificate deployed with reload of",
config.installer, "server; fullchain is", lineage.fullchain)
_suggest_donation_if_appropriate(config, action)
def setup_log_file_handler(config, logfile, fmt):
"""Setup file debug logging."""
log_file_path = os.path.join(config.logs_dir, logfile)
handler = logging.handlers.RotatingFileHandler(
log_file_path, maxBytes=2 ** 20, backupCount=10)
# rotate on each invocation, rollover only possible when maxBytes
# is nonzero and backupCount is nonzero, so we set maxBytes as big
# as possible not to overrun in single CLI invocation (1MB).
handler.doRollover() # TODO: creates empty letsencrypt.log.1 file
handler.setLevel(logging.DEBUG)
handler_formatter = logging.Formatter(fmt=fmt)
handler_formatter.converter = time.gmtime # don't use localtime
handler.setFormatter(handler_formatter)
return handler, log_file_path
def _cli_log_handler(config, level, fmt):
if config.text_mode or config.noninteractive_mode or config.verb == "renew":
handler = colored_logging.StreamHandler()
handler.setFormatter(logging.Formatter(fmt))
else:
handler = log.DialogHandler()
# dialog box is small, display as less as possible
handler.setFormatter(logging.Formatter("%(message)s"))
handler.setLevel(level)
return handler
def setup_logging(config, cli_handler_factory, logfile):
"""Setup logging."""
fmt = "%(asctime)s:%(levelname)s:%(name)s:%(message)s"
level = -config.verbose_count * 10
file_handler, log_file_path = setup_log_file_handler(
config, logfile=logfile, fmt=fmt)
cli_handler = cli_handler_factory(config, level, fmt)
# TODO: use fileConfig?
root_logger = logging.getLogger()
root_logger.setLevel(logging.DEBUG) # send all records to handlers
root_logger.addHandler(cli_handler)
root_logger.addHandler(file_handler)
logger.debug("Root logging level set at %d", level)
logger.info("Saving debug log to %s", log_file_path)
def _handle_exception(exc_type, exc_value, trace, config):
"""Logs exceptions and reports them to the user.
Config is used to determine how to display exceptions to the user. In
general, if config.debug is True, then the full exception and traceback is
shown to the user, otherwise it is suppressed. If config itself is None,
then the traceback and exception is attempted to be written to a logfile.
If this is successful, the traceback is suppressed, otherwise it is shown
to the user. sys.exit is always called with a nonzero status.
"""
logger.debug(
"Exiting abnormally:%s%s",
os.linesep,
"".join(traceback.format_exception(exc_type, exc_value, trace)))
if issubclass(exc_type, Exception) and (config is None or not config.debug):
if config is None:
logfile = "letsencrypt.log"
try:
with open(logfile, "w") as logfd:
traceback.print_exception(
exc_type, exc_value, trace, file=logfd)
except: # pylint: disable=bare-except
sys.exit("".join(
traceback.format_exception(exc_type, exc_value, trace)))
if issubclass(exc_type, errors.Error):
sys.exit(exc_value)
else:
# Here we're passing a client or ACME error out to the client at the shell
# Tell the user a bit about what happened, without overwhelming
# them with a full traceback
err = traceback.format_exception_only(exc_type, exc_value)[0]
# Typical error from the ACME module:
# acme.messages.Error: urn:acme:error:malformed :: The request message was
# malformed :: Error creating new registration :: Validation of contact
# mailto:none@longrandomstring.biz failed: Server failure at resolver
if (("urn:acme" in err and ":: " in err and
config.verbose_count <= cli.flag_default("verbose_count"))):
# prune ACME error code, we have a human description
_code, _sep, err = err.partition(":: ")
msg = "An unexpected error occurred:\n" + err + "Please see the "
if config is None:
msg += "logfile '{0}' for more details.".format(logfile)
else:
msg += "logfiles in {0} for more details.".format(config.logs_dir)
sys.exit(msg)
else:
sys.exit("".join(
traceback.format_exception(exc_type, exc_value, trace)))
def main(cli_args=sys.argv[1:]):
"""Command line argument parsing and main script execution."""
sys.excepthook = functools.partial(_handle_exception, config=None)
plugins = plugins_disco.PluginsRegistry.find_all()
# note: arg parser internally handles --help (and exits afterwards)
args = cli.prepare_and_parse_args(plugins, cli_args)
config = configuration.NamespaceConfig(args)
zope.component.provideUtility(config)
# Setup logging ASAP, otherwise "No handlers could be found for
# logger ..." TODO: this should be done before plugins discovery
for directory in config.config_dir, config.work_dir:
le_util.make_or_verify_dir(
directory, constants.CONFIG_DIRS_MODE, os.geteuid(),
"--strict-permissions" in cli_args)
# TODO: logs might contain sensitive data such as contents of the
# private key! #525
le_util.make_or_verify_dir(
config.logs_dir, 0o700, os.geteuid(), "--strict-permissions" in cli_args)
setup_logging(config, _cli_log_handler, logfile='letsencrypt.log')
logger.debug("letsencrypt version: %s", letsencrypt.__version__)
# do not log `config`, as it contains sensitive data (e.g. revoke --key)!
logger.debug("Arguments: %r", cli_args)
logger.debug("Discovered plugins: %r", plugins)
sys.excepthook = functools.partial(_handle_exception, config=config)
# Displayer
if config.noninteractive_mode:
displayer = display_util.NoninteractiveDisplay(sys.stdout)
elif config.text_mode:
displayer = display_util.FileDisplay(sys.stdout)
elif config.verb == "renew":
config.noninteractive_mode = True
displayer = display_util.NoninteractiveDisplay(sys.stdout)
else:
displayer = display_util.NcursesDisplay()
zope.component.provideUtility(displayer)
# Reporter
report = reporter.Reporter()
zope.component.provideUtility(report)
atexit.register(report.atexit_print_messages)
return config.func(config, plugins)
if __name__ == "__main__":
err_string = main()
if err_string:
logger.warn("Exiting with message %s", err_string)
sys.exit(err_string) # pragma: no cover

0
letsencrypt/renew.py Normal file
View file

157
letsencrypt/tests/cli_test.py
View file

@ -23,6 +23,7 @@ from letsencrypt import constants
from letsencrypt import crypto_util
from letsencrypt import errors
from letsencrypt import le_util
from letsencrypt import main
from letsencrypt import storage
from letsencrypt.plugins import disco
@ -54,16 +55,16 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
def _call(self, args):
"Run the cli with output streams and actual client mocked out"
with mock.patch('letsencrypt.cli.client') as client:
with mock.patch('letsencrypt.main.client') as client:
ret, stdout, stderr = self._call_no_clientmock(args)
return ret, stdout, stderr, client
def _call_no_clientmock(self, args):
"Run the client with output streams mocked out"
args = self.standard_args + args
with mock.patch('letsencrypt.cli.sys.stdout') as stdout:
with mock.patch('letsencrypt.cli.sys.stderr') as stderr:
ret = cli.main(args[:]) # NOTE: parser can alter its args!
with mock.patch('letsencrypt.main.sys.stdout') as stdout:
with mock.patch('letsencrypt.main.sys.stderr') as stderr:
ret = main.main(args[:]) # NOTE: parser can alter its args!
return ret, stdout, stderr
def _call_stdout(self, args):
@ -72,20 +73,21 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
caller.
"""
args = self.standard_args + args
with mock.patch('letsencrypt.cli.sys.stderr') as stderr:
with mock.patch('letsencrypt.cli.client') as client:
ret = cli.main(args[:]) # NOTE: parser can alter its args!
with mock.patch('letsencrypt.main.sys.stderr') as stderr:
with mock.patch('letsencrypt.main.client') as client:
ret = main.main(args[:]) # NOTE: parser can alter its args!
return ret, None, stderr, client
def test_no_flags(self):
with MockedVerb("run") as mock_run:
with mock.patch('letsencrypt.main.run') as mock_run:
self._call([])
self.assertEqual(1, mock_run.call_count)
def _help_output(self, args):
"Run a command, and return the ouput string for scrutiny"
output = six.StringIO()
with mock.patch('letsencrypt.cli.sys.stdout', new=output):
with mock.patch('letsencrypt.main.sys.stdout', new=output):
self.assertRaises(SystemExit, self._call_stdout, args)
out = output.getvalue()
return out
@ -138,8 +140,8 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
"Ensure that a particular error raises a missing cli flag error containing message"
exc = None
try:
with mock.patch('letsencrypt.cli.sys.stderr'):
cli.main(self.standard_args + args[:]) # NOTE: parser can alter its args!
with mock.patch('letsencrypt.main.sys.stderr'):
main.main(self.standard_args + args[:]) # NOTE: parser can alter its args!
except errors.MissingCommandlineFlag as exc:
self.assertTrue(message in str(exc))
self.assertTrue(exc is not None)
@ -149,15 +151,15 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
self._cli_missing_flag(args, "specify a plugin")
args.extend(['--standalone', '-d', 'eg.is'])
self._cli_missing_flag(args, "register before running")
with mock.patch('letsencrypt.cli._auth_from_domains'):
with mock.patch('letsencrypt.cli.client.acme_from_config_key'):
with mock.patch('letsencrypt.main._auth_from_domains'):
with mock.patch('letsencrypt.main.client.acme_from_config_key'):
args.extend(['--email', 'io@io.is'])
self._cli_missing_flag(args, "--agree-tos")
@mock.patch('letsencrypt.cli.client.acme_client.Client')
@mock.patch('letsencrypt.cli._determine_account')
@mock.patch('letsencrypt.cli.client.Client.obtain_and_enroll_certificate')
@mock.patch('letsencrypt.cli._auth_from_domains')
@mock.patch('letsencrypt.main.client.acme_client.Client')
@mock.patch('letsencrypt.main._determine_account')
@mock.patch('letsencrypt.main.client.Client.obtain_and_enroll_certificate')
@mock.patch('letsencrypt.main._auth_from_domains')
def test_user_agent(self, afd, _obt, det, _client):
# Normally the client is totally mocked out, but here we need more
# arguments to automate it...
@ -166,7 +168,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
det.return_value = mock.MagicMock(), None
afd.return_value = mock.MagicMock(), "newcert"
with mock.patch('letsencrypt.cli.client.acme_client.ClientNetwork') as acme_net:
with mock.patch('letsencrypt.main.client.acme_client.ClientNetwork') as acme_net:
self._call_no_clientmock(args)
os_ver = " ".join(le_util.get_os_info())
ua = acme_net.call_args[1]["user_agent"]
@ -176,7 +178,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
if "linux" in plat.lower():
self.assertTrue(platform.linux_distribution()[0] in ua)
with mock.patch('letsencrypt.cli.client.acme_client.ClientNetwork') as acme_net:
with mock.patch('letsencrypt.main.client.acme_client.ClientNetwork') as acme_net:
ua = "bandersnatch"
args += ["--user-agent", ua]
self._call_no_clientmock(args)
@ -188,7 +190,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
chain = 'chain'
fullchain = 'fullchain'
with MockedVerb('install') as mock_install:
with mock.patch('letsencrypt.main.install') as mock_install:
self._call(['install', '--cert-path', cert, '--key-path', 'key',
'--chain-path', 'chain',
'--fullchain-path', 'fullchain'])
@ -199,8 +201,8 @@ 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.cli.record_chosen_plugins')
@mock.patch('letsencrypt.cli.display_ops')
@mock.patch('letsencrypt.main.cli.record_chosen_plugins')
@mock.patch('letsencrypt.main.cli.display_ops')
def test_installer_selection(self, mock_display_ops, _rec):
self._call(['install', '--domains', 'foo.bar', '--cert-path', 'cert',
'--key-path', 'key', '--chain-path', 'chain'])
@ -239,14 +241,14 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
self._cli_missing_flag(["--standalone"], "With the standalone plugin, you probably")
with mock.patch("letsencrypt.cli._init_le_client") as mock_init:
with mock.patch("letsencrypt.cli._auth_from_domains") as mock_afd:
with mock.patch("letsencrypt.main._init_le_client") as mock_init:
with mock.patch("letsencrypt.main._auth_from_domains") as mock_afd:
mock_afd.return_value = (mock.MagicMock(), mock.MagicMock())
self._call(["certonly", "--manual", "-d", "foo.bar"])
unused_config, auth, unused_installer = mock_init.call_args[0]
self.assertTrue(isinstance(auth, manual.Authenticator))
with MockedVerb("certonly") as mock_certonly:
with mock.patch('letsencrypt.main.obtain_cert') as mock_certonly:
self._call(["auth", "--standalone"])
self.assertEqual(1, mock_certonly.call_count)
@ -269,8 +271,8 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
for r in xrange(len(flags)))):
self._call(['plugins'] + list(args))
@mock.patch('letsencrypt.cli.plugins_disco')
@mock.patch('letsencrypt.cli.HelpfulArgumentParser.determine_help_topics')
@mock.patch('letsencrypt.main.plugins_disco')
@mock.patch('letsencrypt.main.cli.HelpfulArgumentParser.determine_help_topics')
def test_plugins_no_args(self, _det, mock_disco):
ifaces = []
plugins = mock_disco.PluginsRegistry.find_all()
@ -281,8 +283,8 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
filtered = plugins.visible().ifaces()
stdout.write.called_once_with(str(filtered))
@mock.patch('letsencrypt.cli.plugins_disco')
@mock.patch('letsencrypt.cli.HelpfulArgumentParser.determine_help_topics')
@mock.patch('letsencrypt.main.plugins_disco')
@mock.patch('letsencrypt.main.cli.HelpfulArgumentParser.determine_help_topics')
def test_plugins_init(self, _det, mock_disco):
ifaces = []
plugins = mock_disco.PluginsRegistry.find_all()
@ -296,8 +298,8 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
verified = filtered.verify()
stdout.write.called_once_with(str(verified))
@mock.patch('letsencrypt.cli.plugins_disco')
@mock.patch('letsencrypt.cli.HelpfulArgumentParser.determine_help_topics')
@mock.patch('letsencrypt.main.plugins_disco')
@mock.patch('letsencrypt.main.cli.HelpfulArgumentParser.determine_help_topics')
def test_plugins_prepare(self, _det, mock_disco):
ifaces = []
plugins = mock_disco.PluginsRegistry.find_all()
@ -319,7 +321,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
chain = 'chain'
fullchain = 'fullchain'
with MockedVerb('certonly') as mock_obtaincert:
with mock.patch('letsencrypt.main.obtain_cert') as mock_obtaincert:
self._call(['certonly', '--cert-path', cert, '--key-path', 'key',
'--chain-path', 'chain',
'--fullchain-path', 'fullchain'])
@ -462,7 +464,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
if domains_arg:
webroot_map_args.extend(["-d", domains_arg])
namespace = parse(webroot_map_args)
domains = cli._find_domains(namespace, mock.MagicMock()) # pylint: disable=protected-access
domains = main._find_domains(namespace, mock.MagicMock()) # pylint: disable=protected-access
self.assertEqual(namespace.webroot_map, expected_map)
self.assertEqual(set(domains), set(expectect_domains))
@ -506,9 +508,9 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
{"eg.com": "/tmp", "www.eg.com": "/tmp", "eg.is": "/tmp2"})
def _certonly_new_request_common(self, mock_client, args=None):
with mock.patch('letsencrypt.cli._treat_as_renewal') as mock_renewal:
with mock.patch('letsencrypt.main._treat_as_renewal') as mock_renewal:
mock_renewal.return_value = ("newcert", None)
with mock.patch('letsencrypt.cli._init_le_client') as mock_init:
with mock.patch('letsencrypt.main._init_le_client') as mock_init:
mock_init.return_value = mock_client
if args is None:
args = []
@ -565,17 +567,17 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
mock_client.obtain_certificate.return_value = (mock_certr, 'chain',
mock_key, 'csr')
try:
with mock.patch('letsencrypt.cli._find_duplicative_certs') as mock_fdc:
with mock.patch('letsencrypt.main._find_duplicative_certs') as mock_fdc:
mock_fdc.return_value = (mock_lineage, None)
with mock.patch('letsencrypt.cli._init_le_client') as mock_init:
with mock.patch('letsencrypt.main._init_le_client') as mock_init:
mock_init.return_value = mock_client
get_utility_path = 'letsencrypt.cli.zope.component.getUtility'
get_utility_path = 'letsencrypt.main.zope.component.getUtility'
with mock.patch(get_utility_path) as mock_get_utility:
with mock.patch('letsencrypt.cli.OpenSSL') as mock_ssl:
with mock.patch('letsencrypt.main.OpenSSL') as mock_ssl:
mock_latest = mock.MagicMock()
mock_latest.get_issuer.return_value = "Fake fake"
mock_ssl.crypto.load_certificate.return_value = mock_latest
with mock.patch('letsencrypt.cli.crypto_util'):
with mock.patch('letsencrypt.main.crypto_util'):
if not args:
args = ['-d', 'isnot.org', '-a', 'standalone', 'certonly']
if extra_args:
@ -691,7 +693,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
if names is not None:
mock_lineage.names.return_value = names
mock_rc.return_value = mock_lineage
with mock.patch('letsencrypt.cli.obtain_cert') as mock_obtain_cert:
with mock.patch('letsencrypt.main.obtain_cert') as mock_obtain_cert:
self._test_renewal_common(True, None, error_expected=error_expected,
args=['renew'], renew=False)
if assert_oc_called is not None:
@ -746,7 +748,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
mock_rc.return_value = mock_lineage
mock_lineage.configuration = {
'renewalparams': {'authenticator': 'webroot'}}
with mock.patch('letsencrypt.cli.obtain_cert') as mock_obtain_cert:
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)
@ -758,8 +760,8 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
renew=False, error_expected=True)
@mock.patch('letsencrypt.cli.zope.component.getUtility')
@mock.patch('letsencrypt.cli._treat_as_renewal')
@mock.patch('letsencrypt.cli._init_le_client')
@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):
mock_renewal.return_value = ('reinstall', mock.MagicMock())
mock_init.return_value = mock_client = mock.MagicMock()
@ -776,9 +778,9 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
mock_client.obtain_certificate_from_csr.return_value = (certr, chain)
cert_path = '/etc/letsencrypt/live/example.com/cert.pem'
mock_client.save_certificate.return_value = cert_path, None, None
with mock.patch('letsencrypt.cli._init_le_client') as mock_init:
with mock.patch('letsencrypt.main._init_le_client') as mock_init:
mock_init.return_value = mock_client
get_utility_path = 'letsencrypt.cli.zope.component.getUtility'
get_utility_path = 'letsencrypt.main.zope.component.getUtility'
with mock.patch(get_utility_path) as mock_get_utility:
chain_path = '/etc/letsencrypt/live/example.com/chain.pem'
full_path = '/etc/letsencrypt/live/example.com/fullchain.pem'
@ -787,7 +789,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
CSR, cert_path, chain_path, full_path).split()
if extra_args:
args += extra_args
with mock.patch('letsencrypt.cli.crypto_util'):
with mock.patch('letsencrypt.main.crypto_util'):
self._call(args)
if '--dry-run' in args:
@ -811,7 +813,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
self.assertTrue(
'dry run' in mock_get_utility().add_message.call_args[0][0])
@mock.patch('letsencrypt.cli.client.acme_client')
@mock.patch('letsencrypt.main.client.acme_client')
def test_revoke_with_key(self, mock_acme_client):
server = 'foo.bar'
self._call_no_clientmock(['--cert-path', CERT, '--key-path', KEY,
@ -824,7 +826,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
mock_revoke = mock_acme_client.Client().revoke
mock_revoke.assert_called_once_with(jose.ComparableX509(cert))
@mock.patch('letsencrypt.cli._determine_account')
@mock.patch('letsencrypt.main._determine_account')
def test_revoke_without_key(self, mock_determine_account):
mock_determine_account.return_value = (mock.MagicMock(), None)
_, _, _, client = self._call(['--cert-path', CERT, 'revoke'])
@ -833,7 +835,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
mock_revoke = client.acme_from_config_key().revoke
mock_revoke.assert_called_once_with(jose.ComparableX509(cert))
@mock.patch('letsencrypt.cli.sys')
@mock.patch('letsencrypt.main.sys')
def test_handle_exception(self, mock_sys):
# pylint: disable=protected-access
from acme import messages
@ -841,20 +843,20 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
config = mock.MagicMock()
mock_open = mock.mock_open()
with mock.patch('letsencrypt.cli.open', mock_open, create=True):
with mock.patch('letsencrypt.main.open', mock_open, create=True):
exception = Exception('detail')
config.verbose_count = 1
cli._handle_exception(
main._handle_exception(
Exception, exc_value=exception, trace=None, config=None)
mock_open().write.assert_called_once_with(''.join(
traceback.format_exception_only(Exception, exception)))
error_msg = mock_sys.exit.call_args_list[0][0][0]
self.assertTrue('unexpected error' in error_msg)
with mock.patch('letsencrypt.cli.open', mock_open, create=True):
with mock.patch('letsencrypt.main.open', mock_open, create=True):
mock_open.side_effect = [KeyboardInterrupt]
error = errors.Error('detail')
cli._handle_exception(
main._handle_exception(
errors.Error, exc_value=error, trace=None, config=None)
# assert_any_call used because sys.exit doesn't exit in cli.py
mock_sys.exit.assert_any_call(''.join(
@ -863,7 +865,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
exception = messages.Error(detail='alpha', typ='urn:acme:error:triffid',
title='beta')
config = mock.MagicMock(debug=False, verbose_count=-3)
cli._handle_exception(
main._handle_exception(
messages.Error, exc_value=exception, trace=None, config=config)
error_msg = mock_sys.exit.call_args_list[-1][0][0]
self.assertTrue('unexpected error' in error_msg)
@ -871,7 +873,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
self.assertTrue('alpha' in error_msg)
self.assertTrue('beta' in error_msg)
config = mock.MagicMock(debug=False, verbose_count=1)
cli._handle_exception(
main._handle_exception(
messages.Error, exc_value=exception, trace=None, config=config)
error_msg = mock_sys.exit.call_args_list[-1][0][0]
self.assertTrue('unexpected error' in error_msg)
@ -879,7 +881,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
self.assertTrue('alpha' in error_msg)
interrupt = KeyboardInterrupt('detail')
cli._handle_exception(
main._handle_exception(
KeyboardInterrupt, exc_value=interrupt, trace=None, config=None)
mock_sys.exit.assert_called_with(''.join(
traceback.format_exception_only(KeyboardInterrupt, interrupt)))
@ -898,7 +900,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
self.assertEqual(contents, test_contents)
def test_agree_dev_preview_config(self):
with MockedVerb('run') as mocked_run:
with mock.patch('letsencrypt.main.run') as mocked_run:
self._call(['-c', test_util.vector_path('cli.ini')])
self.assertTrue(mocked_run.called)
@ -915,8 +917,8 @@ class DetermineAccountTest(unittest.TestCase):
def _call(self):
# pylint: disable=protected-access
from letsencrypt.cli import _determine_account
with mock.patch('letsencrypt.cli.account.AccountFileStorage') as mock_storage:
from letsencrypt.main import _determine_account
with mock.patch('letsencrypt.main.account.AccountFileStorage') as mock_storage:
mock_storage.return_value = self.account_storage
return _determine_account(self.config)
@ -948,7 +950,7 @@ class DetermineAccountTest(unittest.TestCase):
def test_no_accounts_no_email(self, mock_get_email):
mock_get_email.return_value = 'foo@bar.baz'
with mock.patch('letsencrypt.cli.client') as client:
with mock.patch('letsencrypt.main.client') as client:
client.register.return_value = (
self.accs[0], mock.sentinel.acme)
self.assertEqual((self.accs[0], mock.sentinel.acme), self._call())
@ -960,7 +962,7 @@ class DetermineAccountTest(unittest.TestCase):
def test_no_accounts_email(self):
self.config.email = 'other email'
with mock.patch('letsencrypt.cli.client') as client:
with mock.patch('letsencrypt.main.client') as client:
client.register.return_value = (self.accs[1], mock.sentinel.acme)
self._call()
self.assertEqual(self.accs[1].id, self.config.account)
@ -980,7 +982,7 @@ class DuplicativeCertsTest(storage_test.BaseRenewableCertTest):
@mock.patch('letsencrypt.le_util.make_or_verify_dir')
def test_find_duplicative_names(self, unused_makedir):
from letsencrypt.cli import _find_duplicative_certs
from letsencrypt.main import _find_duplicative_certs
test_cert = test_util.load_vector('cert-san.pem')
with open(self.test_rc.cert, 'w') as f:
f.write(test_cert)
@ -1008,34 +1010,5 @@ class DuplicativeCertsTest(storage_test.BaseRenewableCertTest):
self.assertEqual(result, (None, None))
class MockedVerb(object):
"""Simple class that can be used for mocking out verbs/subcommands.
Storing a dictionary of verbs and the functions that implement them
in letsencrypt.cli makes mocking much more complicated. This class
can be used as a simple context manager for mocking out verbs in CLI
tests. For example:
with MockedVerb("run") as mock_run:
self._call([])
self.assertEqual(1, mock_run.call_count)
"""
def __init__(self, verb_name):
self.verb_dict = cli.HelpfulArgumentParser.VERBS
self.verb_func = None
self.verb_name = verb_name
def __enter__(self):
self.verb_func = self.verb_dict[self.verb_name]
mocked_func = mock.MagicMock()
self.verb_dict[self.verb_name] = mocked_func
return mocked_func
def __exit__(self, unused_type, unused_value, unused_trace):
self.verb_dict[self.verb_name] = self.verb_func
if __name__ == '__main__':
unittest.main() # pragma: no cover

2
letsencrypt/tests/client_test.py
View file

@ -128,7 +128,7 @@ class ClientTest(unittest.TestCase):
# FIXME move parts of this to test_cli.py...
@mock.patch("letsencrypt.client.logger")
@mock.patch("letsencrypt.cli._process_domain")
@mock.patch("letsencrypt.cli.process_domain")
def test_obtain_certificate_from_csr(self, mock_process_domain, mock_logger):
self._mock_obtain_certificate()
from letsencrypt import cli

2
setup.py
View file

@ -127,7 +127,7 @@ setup(
entry_points={
'console_scripts': [
'letsencrypt = letsencrypt.cli:main',
'letsencrypt = letsencrypt.main:main',
],
'letsencrypt.plugins': [
'manual = letsencrypt.plugins.manual:Authenticator',