From e64fd392dc60fb01a37ccdfd9110d115c2b3f7d5 Mon Sep 17 00:00:00 2001 From: Peter Eckersley Date: Sun, 28 Feb 2016 23:34:44 -0800 Subject: [PATCH] Rope refactor: cli -> main --- letsencrypt/cli.py | 712 ++-------------------------------- letsencrypt/main.py | 696 +++++++++++++++++++++++++++++++++ letsencrypt/tests/cli_test.py | 30 +- 3 files changed, 736 insertions(+), 702 deletions(-) create mode 100644 letsencrypt/main.py diff --git a/letsencrypt/cli.py b/letsencrypt/cli.py index 3551d5a10..d308db72e 100644 --- a/letsencrypt/cli.py +++ b/letsencrypt/cli.py @@ -1,9 +1,5 @@ """Let's Encrypt CLI.""" from __future__ import print_function - -# TODO: Sanity check all input. Be sure to avoid shell code etc... -# pylint: disable=too-many-lines -# (TODO: split this file into main.py and cli.py) import argparse import atexit import copy @@ -23,10 +19,6 @@ 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 @@ -37,6 +29,7 @@ from letsencrypt import errors from letsencrypt import interfaces from letsencrypt import le_util from letsencrypt import log +from letsencrypt import main from letsencrypt import reporter from letsencrypt import storage @@ -44,6 +37,15 @@ from letsencrypt.display import util as display_util from letsencrypt.display import ops as display_ops from letsencrypt.plugins import disco as plugins_disco +# TODO: Sanity check all input. Be sure to avoid shell code etc... +# pylint: disable=too-many-lines +# (TODO: split this file into main.py and cli.py) + + + + + + logger = logging.getLogger(__name__) # Global, to save us from a lot of argument passing within the scope of this module @@ -127,158 +129,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...") @@ -293,217 +144,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 (or in some cases, --duplicate)"): - 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 @@ -645,103 +285,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 @@ -925,7 +468,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 " @@ -1012,9 +555,9 @@ 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) + main.obtain_cert(lineage_config, plugins, renewal_candidate) renew_successes.append(renewal_candidate.fullchain) else: renew_skipped.append(renewal_candidate.fullchain) @@ -1036,63 +579,6 @@ def renew(config, unused_plugins): 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) - - -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. @@ -1152,10 +638,10 @@ 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} + VERBS = {"auth": main.obtain_cert, "certonly": main.obtain_cert, + "config_changes": main.config_changes, "everything": main.run, + "install": main.install, "plugins": main.plugins_cmd, "renew": renew, + "revoke": main.revoke, "rollback": main.rollback, "run": main.run} # List of topics for which additional help can be provided HELP_TOPICS = ["all", "security", @@ -1247,7 +733,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 " @@ -1270,7 +756,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) @@ -1802,7 +1288,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 @@ -1828,165 +1314,17 @@ 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 webroot_map.iteritems(): - _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) + """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() + err_string = main.main() if err_string: logger.warn("Exiting with message %s", err_string) sys.exit(err_string) # pragma: no cover diff --git a/letsencrypt/main.py b/letsencrypt/main.py new file mode 100644 index 000000000..89e7a85a3 --- /dev/null +++ b/letsencrypt/main.py @@ -0,0 +1,696 @@ +from __future__ import print_function +import atexit +import functools +import os +import sys +import zope.component + +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 + +import traceback +import logging.handlers +import time +from acme import jose +import OpenSSL + +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 (or in some cases, --duplicate)"): + 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) + + +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) diff --git a/letsencrypt/tests/cli_test.py b/letsencrypt/tests/cli_test.py index aef3447c3..f17f0fb73 100644 --- a/letsencrypt/tests/cli_test.py +++ b/letsencrypt/tests/cli_test.py @@ -13,7 +13,7 @@ import mock from acme import jose -from letsencrypt import account +from letsencrypt import account, main from letsencrypt import cli from letsencrypt import configuration from letsencrypt import constants @@ -60,7 +60,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods 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! + ret = main.main(args[:]) # NOTE: parser can alter its args! return ret, stdout, stderr def _call_stdout(self, args): @@ -71,7 +71,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods 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! + ret = main.main(args[:]) # NOTE: parser can alter its args! return ret, None, stderr, client def test_no_flags(self): @@ -136,7 +136,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods exc = None try: with mock.patch('letsencrypt.cli.sys.stderr'): - cli.main(self.standard_args + args[:]) # NOTE: parser can alter its args! + 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) @@ -459,7 +459,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)) @@ -835,7 +835,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods with mock.patch('letsencrypt.cli.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))) @@ -845,7 +845,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods with mock.patch('letsencrypt.cli.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( @@ -854,7 +854,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) @@ -862,7 +862,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) @@ -870,7 +870,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))) @@ -909,7 +909,7 @@ class DetermineAccountTest(unittest.TestCase): from letsencrypt.cli import _determine_account with mock.patch('letsencrypt.cli.account.AccountFileStorage') as mock_storage: mock_storage.return_value = self.account_storage - return _determine_account(self.config) + return main._determine_account(self.config) def test_args_account_set(self): self.account_storage.save(self.accs[1]) @@ -977,24 +977,24 @@ class DuplicativeCertsTest(storage_test.BaseRenewableCertTest): f.write(test_cert) # No overlap at all - result = _find_duplicative_certs( + result = main._find_duplicative_certs( self.cli_config, ['wow.net', 'hooray.org']) self.assertEqual(result, (None, None)) # Totally identical - result = _find_duplicative_certs( + result = main._find_duplicative_certs( self.cli_config, ['example.com', 'www.example.com']) self.assertTrue(result[0].configfile.filename.endswith('example.org.conf')) self.assertEqual(result[1], None) # Superset - result = _find_duplicative_certs( + result = main._find_duplicative_certs( self.cli_config, ['example.com', 'www.example.com', 'something.new']) self.assertEqual(result[0], None) self.assertTrue(result[1].configfile.filename.endswith('example.org.conf')) # Partial overlap doesn't count - result = _find_duplicative_certs( + result = main._find_duplicative_certs( self.cli_config, ['example.com', 'something.new']) self.assertEqual(result, (None, None))