Command-line UX overhaul (#8852)

Streamline and reorganize Certbot's CLI output.

This change is a substantial command-line UX overhaul,
based on previous user research. The main goal was to streamline
and clarify output. To see more verbose output, use the -v or -vv flags.

---

* nginx,apache: CLI logging changes

- Add "Successfully deployed ..." message using display_util
- Remove IReporter usage and replace with display_util
- Standardize "... could not find a VirtualHost ..." error

This changes also bumps the version of certbot required by certbot-nginx
and certbot-apache to take use of the new display_util function.

* fix certbot_compatibility_test

since the http plugins now require IDisplay, we need to inject it

* fix dependency version on certbot

* use better asserts

* try fix oldest deps

because certbot 1.10.0 depends on acme>=1.8.0, we need to use
acme==1.8.0 in the -oldest tests

* cli: redesign output of new certificate reporting

Changes the output of run, certonly and certonly --csr. No longer uses
IReporter.

* cli: redesign output of failed authz reporting

* fix problem sorting to be stable between py2 & 3

* add some catch-all error text

* cli: dont use IReporter for EFF donation prompt

* add per-authenticator hints

* pass achalls to auth_hint, write some tests

* exclude static auth hints from coverage

* dont call auth_hint unless derived from .Plugin

* dns fallback hint: dont assume --dns-blah works

--dns-blah won't work for third-party plugins, they need to be specified
using --authenticator dns-blah.

* add code comments about the auth_hint interface

* renew: don't restart the installer for dry-runs

Prevents Certbot from superfluously invoking the installer restart
during dry-run renewals. (This does not affect authenticator restarts).

Additionally removes some CLI output that was reporting the fullchain
path of the renewed certificate.

* update CHANGELOG.md

* cli: redesign output when cert installation failed

- Display a message when certificate installation begins.
- Don't use IReporter, just log errors immediately if restart/rollback
  fails.
- Prompt the user with a command to retry the installation process once
  they have fixed any underlying problems.

* vary by preconfigured_renewal

and move expiry date to be above the renewal advice

* update code comment

Co-authored-by: ohemorange <ebportnoy@gmail.com>

* update code comment

Co-authored-by: ohemorange <ebportnoy@gmail.com>

* fix lint

* derve cert name from cert_path, if possible

* fix type annotation

* text change in nginx hint

Co-authored-by: ohemorange <ebportnoy@gmail.com>

* print message when restarting server after renewal

* log: print "advice" when exiting with an error

When running in non-quiet mode.

* try fix -oldest lock_test.py

* fix docstring

* s/Restarting/Reloading/ when notifying the user

* fix test name

Co-authored-by: ohemorange <ebportnoy@gmail.com>

* type annotations

* s/using the {} plugin/installer: {}/

* copy: avoid "plugin" where possible

* link to user guide#automated-renewals

when not running with --preconfigured-renewal

* cli: reduce default logging verbosity

* fix lock_test: -vv is needed to see logger.debug

* Change comment in log.py to match the change to default verbosity

* Audit and adjust logging levels in apache module

* Audit and adjust logging levels in nginx module

* Audit, adjust logging levels, and improve logging calls in certbot module

* Fix tests to mock correct methods and classes

* typo in non-preconfigured-renewal message

Co-authored-by: ohemorange <ebportnoy@gmail.com>

* fix test

* revert acme version bump

* catch up to python3 changes

* Revert "revert acme version bump"

This reverts commit fa83d6a51c.

* Change ocsp check error to warning since it's non-fatal

* Update storage_test in parallel with last change

* get rid of leading newline on "Deploying [...]"

* shrink renewal and installation success messages

* print logfile rather than logdir in exit handler

* Decrease logging level to info for idempotent operation where enhancement is already set

* Display cert not yet due for renewal message when renewing and no other action will be taken, and change cert to certificate

* also write to logger so it goes in the log file

* Don't double write to log file; fix main test

* cli: remove trailing newline on new cert reporting

* ignore type error

* revert accidental changes to dependencies

* Pass tests in any timezone by using utcfromtimestamp

* Add changelog entry

* fix nits

* Improve wording of try again message

* minor wording change to changelog

* hooks: send hook stdout to CLI stdout

includes both --manual and --{pre,post,renew} hooks

* update docstrings and remove TODO

* add a pending deprecation on execute_command

* add test coverage for both

* update deprecation text

Co-authored-by: ohemorange <ebportnoy@gmail.com>

Co-authored-by: Alex Zorin <alex@zorin.id.au>
Co-authored-by: alexzorin <alex@zor.io>
This commit is contained in:
ohemorange
2021-05-25 10:47:39 +10:00
committed by GitHub
co-authored by Alex Zorin alexzorin
parent 099c6c8b24
commit 6f27c32db1
55 changed files with 1010 additions and 386 deletions
+6
View File
@@ -14,10 +14,16 @@ Certbot adheres to [Semantic Versioning](https://semver.org/).
* Use UTF-8 encoding for renewal configuration files
* Windows installer now cleans up old Certbot dependency packages
before installing the new ones to avoid version conflicts.
* This release contains a substantial command-line UX overhaul,
based on previous user research. The main goal was to streamline
and clarify output. If you would like to see more verbose output, use
the -v or -vv flags. UX improvements are an iterative process and
the Certbot team welcomes constructive feedback.
### Fixed
* Fix TypeError due to incompatibility with lexicon >= v3.6.0
* Installers (e.g. nginx, Apache) were being restarted unnecessarily after dry-run renewals.
More details about these changes can be found on our GitHub repo.
+31 -57
View File
@@ -15,6 +15,8 @@ from certbot import achallenges
from certbot import errors
from certbot import interfaces
from certbot._internal import error_handler
from certbot.display import util as display_util
from certbot.plugins import common as plugin_common
logger = logging.getLogger(__name__)
@@ -70,7 +72,6 @@ class AuthHandler:
resps = self.auth.perform(achalls)
# If debug is on, wait for user input before starting the verification process.
logger.info('Waiting for verification...')
config = zope.component.getUtility(interfaces.IConfig)
if config.debug_challenges:
notify = zope.component.getUtility(interfaces.IDisplay).notification
@@ -88,6 +89,7 @@ class AuthHandler:
self.acme.answer_challenge(achall.challb, resp)
# Wait for authorizations to be checked.
logger.info('Waiting for verification...')
self._poll_authorizations(authzrs, max_retries, best_effort)
# Keep validated authorizations only. If there is none, no certificate can be issued.
@@ -148,7 +150,7 @@ class AuthHandler:
authzrs_failed = [authzr for authzr, _ in authzrs_to_check.values()
if authzr.body.status == messages.STATUS_INVALID]
for authzr_failed in authzrs_failed:
logger.warning('Challenge failed for domain %s',
logger.info('Challenge failed for domain %s',
authzr_failed.body.identifier.value)
# Accumulating all failed authzrs to build a consolidated report
# on them at the end of the polling.
@@ -173,7 +175,7 @@ class AuthHandler:
# In case of failed authzrs, create a report to the user.
if authzrs_failed_to_report:
_report_failed_authzrs(authzrs_failed_to_report, self.account.key)
self._report_failed_authzrs(authzrs_failed_to_report)
if not best_effort:
# Without best effort, having failed authzrs is critical and fail the process.
raise errors.AuthorizationError('Some challenges have failed.')
@@ -264,6 +266,29 @@ class AuthHandler:
return achalls
def _report_failed_authzrs(self, failed_authzrs: List[messages.AuthorizationResource]) -> None:
"""Notifies the user about failed authorizations."""
problems: Dict[str, List[achallenges.AnnotatedChallenge]] = {}
failed_achalls = [challb_to_achall(challb, self.account.key, authzr.body.identifier.value)
for authzr in failed_authzrs for challb in authzr.body.challenges
if challb.error]
for achall in failed_achalls:
problems.setdefault(achall.error.typ, []).append(achall)
msg = [f"\nCertbot failed to authenticate some domains (authenticator: {self.auth.name})."
" The Certificate Authority reported these problems:"]
for _, achalls in sorted(problems.items(), key=lambda item: item[0]):
msg.append(_generate_failed_chall_msg(achalls))
# auth_hint will only be called on authenticators that subclass
# plugin_common.Plugin. Refer to comment on that function.
if failed_achalls and isinstance(self.auth, plugin_common.Plugin):
msg.append(f"\nHint: {self.auth.auth_hint(failed_achalls)}\n")
display_util.notify("".join(msg))
def challb_to_achall(challb, account_key, domain):
"""Converts a ChallengeBody object to an AnnotatedChallenge.
@@ -393,60 +418,13 @@ def _report_no_chall_path(challbs):
raise errors.AuthorizationError(msg)
_ERROR_HELP_COMMON = (
"To fix these errors, please make sure that your domain name was entered "
"correctly and the DNS A/AAAA record(s) for that domain contain(s) the "
"right IP address.")
_ERROR_HELP = {
"connection":
_ERROR_HELP_COMMON + " Additionally, please check that your computer "
"has a publicly routable IP address and that no firewalls are preventing "
"the server from communicating with the client. If you're using the "
"webroot plugin, you should also verify that you are serving files "
"from the webroot path you provided.",
"dnssec":
_ERROR_HELP_COMMON + " Additionally, if you have DNSSEC enabled for "
"your domain, please ensure that the signature is valid.",
"malformed":
"To fix these errors, please make sure that you did not provide any "
"invalid information to the client, and try running Certbot "
"again.",
"serverInternal":
"Unfortunately, an error on the ACME server prevented you from completing "
"authorization. Please try again later.",
"tls":
_ERROR_HELP_COMMON + " Additionally, please check that you have an "
"up-to-date TLS configuration that allows the server to communicate "
"with the Certbot client.",
"unauthorized": _ERROR_HELP_COMMON,
"unknownHost": _ERROR_HELP_COMMON,
}
def _report_failed_authzrs(failed_authzrs, account_key):
"""Notifies the user about failed authorizations."""
problems: Dict[str, List[achallenges.AnnotatedChallenge]] = {}
failed_achalls = [challb_to_achall(challb, account_key, authzr.body.identifier.value)
for authzr in failed_authzrs for challb in authzr.body.challenges
if challb.error]
for achall in failed_achalls:
problems.setdefault(achall.error.typ, []).append(achall)
reporter = zope.component.getUtility(interfaces.IReporter)
for achalls in problems.values():
reporter.add_message(_generate_failed_chall_msg(achalls), reporter.MEDIUM_PRIORITY)
def _generate_failed_chall_msg(failed_achalls):
# type: (List[achallenges.AnnotatedChallenge]) -> str
"""Creates a user friendly error message about failed challenges.
:param list failed_achalls: A list of failed
:class:`certbot.achallenges.AnnotatedChallenge` with the same error
type.
:returns: A formatted error message for the client.
:rtype: str
@@ -455,14 +433,10 @@ def _generate_failed_chall_msg(failed_achalls):
typ = error.typ
if messages.is_acme_error(error):
typ = error.code
msg = ["The following errors were reported by the server:"]
msg = []
for achall in failed_achalls:
msg.append("\n\nDomain: %s\nType: %s\nDetail: %s" % (
msg.append("\n Domain: %s\n Type: %s\n Detail: %s\n" % (
achall.domain, typ, achall.error.detail))
if typ in _ERROR_HELP:
msg.append("\n\n")
msg.append(_ERROR_HELP[typ])
return "".join(msg)
+24 -24
View File
@@ -9,7 +9,6 @@ from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric.rsa import generate_private_key # type: ignore
import josepy as jose
import OpenSSL
import zope.component
from acme import client as acme_client
from acme import crypto_util as acme_crypto_util
@@ -18,7 +17,6 @@ from acme import messages
import certbot
from certbot import crypto_util
from certbot import errors
from certbot import interfaces
from certbot import util
from certbot._internal import account
from certbot._internal import auth_handler
@@ -30,6 +28,7 @@ from certbot._internal import storage
from certbot._internal.plugins import selection as plugin_selection
from certbot.compat import os
from certbot.display import ops as display_ops
from certbot.display import util as display_util
logger = logging.getLogger(__name__)
@@ -154,7 +153,7 @@ def register(config, account_storage, tos_cb=None):
if not config.register_unsafely_without_email:
msg = ("No email was provided and "
"--register-unsafely-without-email was not present.")
logger.warning(msg)
logger.error(msg)
raise errors.Error(msg)
if not config.dry_run:
logger.debug("Registering without email!")
@@ -276,7 +275,7 @@ class Client:
if self.auth_handler is None:
msg = ("Unable to obtain certificate because authenticator is "
"not set.")
logger.warning(msg)
logger.error(msg)
raise errors.Error(msg)
if self.account.regr is None:
raise errors.Error("Please register with the ACME server first.")
@@ -506,8 +505,6 @@ class Client:
cert_file.write(cert_pem)
finally:
cert_file.close()
logger.info("Server issued certificate; certificate written to %s",
abs_cert_path)
chain_file, abs_chain_path =\
_open_pem_file('chain_path', chain_path)
@@ -519,10 +516,11 @@ class Client:
return abs_cert_path, abs_chain_path, abs_fullchain_path
def deploy_certificate(self, domains, privkey_path,
def deploy_certificate(self, cert_name, domains, privkey_path,
cert_path, chain_path, fullchain_path):
"""Install certificate
:param str cert_name: name of the certificate lineage (optional)
:param list domains: list of domains to install the certificate
:param str privkey_path: path to certificate private key
:param str cert_path: certificate file path (optional)
@@ -530,13 +528,19 @@ class Client:
"""
if self.installer is None:
logger.warning("No installer specified, client is unable to deploy"
logger.error("No installer specified, client is unable to deploy"
"the certificate")
raise errors.Error("No installer available")
chain_path = None if chain_path is None else os.path.abspath(chain_path)
msg = ("Unable to install the certificate")
display_util.notify("Deploying certificate")
msg = f"Failed to install the certificate (installer: {self.config.installer})."
if cert_name:
msg += (" Try again after fixing errors by running:\n\n"
f" {cli.cli_constants.cli_command} install --cert-name {cert_name}\n")
with error_handler.ErrorHandler(self._recovery_routine_with_msg, msg):
for dom in domains:
self.installer.deploy_cert(
@@ -568,7 +572,7 @@ class Client:
"""
if self.installer is None:
logger.warning("No installer is specified, there isn't any "
logger.error("No installer is specified, there isn't any "
"configuration to enhance.")
raise errors.Error("No installer available")
@@ -589,7 +593,7 @@ class Client:
self.apply_enhancement(domains, enhancement_name, option)
enhanced = True
elif config_value:
logger.warning(
logger.error(
"Option %s is not supported by the selected installer. "
"Skipping enhancement.", config_name)
@@ -621,13 +625,13 @@ class Client:
self.installer.enhance(dom, enhancement, options)
except errors.PluginEnhancementAlreadyPresent:
if enhancement == "ensure-http-header":
logger.warning("Enhancement %s was already set.",
logger.info("Enhancement %s was already set.",
options)
else:
logger.warning("Enhancement %s was already set.",
logger.info("Enhancement %s was already set.",
enhancement)
except errors.PluginError:
logger.warning("Unable to set enhancement %s for %s",
logger.error("Unable to set enhancement %s for %s",
enhancement, dom)
raise
@@ -640,8 +644,7 @@ class Client:
"""
self.installer.recovery_routine()
reporter = zope.component.getUtility(interfaces.IReporter)
reporter.add_message(success_msg, reporter.HIGH_PRIORITY)
display_util.notify(success_msg)
def _rollback_and_restart(self, success_msg):
"""Rollback the most recent checkpoint and restart the webserver
@@ -649,20 +652,19 @@ class Client:
:param str success_msg: message to show on successful rollback
"""
logger.critical("Rolling back to previous server configuration...")
reporter = zope.component.getUtility(interfaces.IReporter)
logger.info("Rolling back to previous server configuration...")
try:
self.installer.rollback_checkpoints()
self.installer.restart()
except:
reporter.add_message(
logger.error(
"An error occurred and we failed to restore your config and "
"restart your server. Please post to "
"https://community.letsencrypt.org/c/help "
"with details about your configuration and this error you received.",
reporter.HIGH_PRIORITY)
"with details about your configuration and this error you received."
)
raise
reporter.add_message(success_msg, reporter.HIGH_PRIORITY)
display_util.notify(success_msg)
def validate_key_csr(privkey, csr=None):
@@ -761,5 +763,3 @@ def _save_chain(chain_pem, chain_file):
chain_file.write(chain_pem)
finally:
chain_file.close()
logger.info("Cert chain written to %s", chain_file.name)
+2 -2
View File
@@ -22,7 +22,7 @@ CLI_DEFAULTS = dict(
],
# Main parser
verbose_count=-int(logging.INFO / 10),
verbose_count=-int(logging.WARNING / 10),
text_mode=False,
max_log_backups=1000,
preconfigured_renewal=False,
@@ -139,7 +139,7 @@ REVOCATION_REASONS = {
"""Defaults for CLI flags and `.IConfig` attributes."""
QUIET_LOGGING_LEVEL = logging.WARNING
QUIET_LOGGING_LEVEL = logging.ERROR
"""Logging level to use in quiet mode."""
RENEWER_DEFAULTS = dict(
+5 -2
View File
@@ -9,6 +9,7 @@ from certbot import util
from certbot.compat import filesystem
from certbot.compat import misc
from certbot.compat import os
from certbot.display import ops as display_ops
from certbot.plugins import util as plug_util
logger = logging.getLogger(__name__)
@@ -210,7 +211,7 @@ def _run_deploy_hook(command, domains, lineage_path, dry_run):
"""
if dry_run:
logger.warning("Dry run: skipping deploy hook command: %s",
logger.info("Dry run: skipping deploy hook command: %s",
command)
return
@@ -227,7 +228,9 @@ def _run_hook(cmd_name, shell_cmd):
:type shell_cmd: `list` of `str` or `str`
:returns: stderr if there was any"""
err, _ = misc.execute_command(cmd_name, shell_cmd, env=util.env_no_snap_for_external_calls())
returncode, err, out = misc.execute_command_status(
cmd_name, shell_cmd, env=util.env_no_snap_for_external_calls())
display_ops.report_executed_command(f"Hook '{cmd_name}'", returncode, out, err)
return err
+24 -14
View File
@@ -13,7 +13,7 @@ and properly flushed before program exit.
The `logging` module is useful for recording messages about about what
Certbot is doing under the hood, but do not necessarily need to be shown
to the user on the terminal. The default verbosity is INFO.
to the user on the terminal. The default verbosity is WARNING.
The preferred method to display important information to the user is to
use `certbot.display.util` and `certbot.display.ops`.
@@ -28,6 +28,7 @@ import shutil
import sys
import tempfile
import traceback
from types import TracebackType
from acme import messages
from certbot import errors
@@ -78,7 +79,9 @@ def pre_arg_parse_setup():
util.atexit_register(logging.shutdown)
sys.excepthook = functools.partial(
pre_arg_parse_except_hook, memory_handler,
debug='--debug' in sys.argv, log_path=temp_handler.path)
debug='--debug' in sys.argv,
quiet='--quiet' in sys.argv or '-q' in sys.argv,
log_path=temp_handler.path)
def post_arg_parse_setup(config):
@@ -95,7 +98,6 @@ def post_arg_parse_setup(config):
"""
file_handler, file_path = setup_log_file_handler(
config, 'letsencrypt.log', FILE_FMT)
logs_dir = os.path.dirname(file_path)
root_logger = logging.getLogger()
memory_handler = stderr_handler = None
@@ -122,10 +124,13 @@ def post_arg_parse_setup(config):
level = -config.verbose_count * 10
stderr_handler.setLevel(level)
logger.debug('Root logging level set at %d', level)
logger.info('Saving debug log to %s', file_path)
if not config.quiet:
print(f'Saving debug log to {file_path}', file=sys.stderr)
sys.excepthook = functools.partial(
post_arg_parse_except_hook, debug=config.debug, log_path=logs_dir)
post_arg_parse_except_hook,
debug=config.debug, quiet=config.quiet, log_path=file_path)
def setup_log_file_handler(config, logfile, fmt):
@@ -307,7 +312,8 @@ def pre_arg_parse_except_hook(memory_handler, *args, **kwargs):
memory_handler.flush(force=True)
def post_arg_parse_except_hook(exc_type, exc_value, trace, debug, log_path):
def post_arg_parse_except_hook(exc_type: type, exc_value: BaseException, trace: TracebackType,
debug: bool, quiet: bool, log_path: str):
"""Logs fatal exceptions and reports them to the user.
If debug is True, the full exception and traceback is shown to the
@@ -318,10 +324,13 @@ def post_arg_parse_except_hook(exc_type, exc_value, trace, debug, log_path):
:param BaseException exc_value: raised exception
:param traceback trace: traceback of where the exception was raised
:param bool debug: True if the traceback should be shown to the user
:param bool quiet: True if Certbot is running in quiet mode
:param str log_path: path to file or directory containing the log
"""
exc_info = (exc_type, exc_value, trace)
# Only print human advice if not running under --quiet
exit_func = lambda: sys.exit(1) if quiet else exit_with_advice(log_path)
# constants.QUIET_LOGGING_LEVEL or higher should be used to
# display message the user, otherwise, a lower level like
# logger.DEBUG should be used
@@ -337,7 +346,7 @@ def post_arg_parse_except_hook(exc_type, exc_value, trace, debug, log_path):
# our logger printing warnings and errors in red text.
if issubclass(exc_type, errors.Error):
logger.error(str(exc_value))
sys.exit(1)
exit_func()
logger.error('An unexpected error occurred:')
if messages.is_acme_error(exc_value):
# Remove the ACME error prefix from the exception
@@ -350,11 +359,11 @@ def post_arg_parse_except_hook(exc_type, exc_value, trace, debug, log_path):
# and remove the final newline before passing it to
# logger.error.
logger.error(''.join(output).rstrip())
exit_with_log_path(log_path)
exit_func()
def exit_with_log_path(log_path):
"""Print a message about the log location and exit.
def exit_with_advice(log_path: str):
"""Print a link to the community forums, the debug log path, and exit
The message is printed to stderr and the program will exit with a
nonzero status.
@@ -362,10 +371,11 @@ def exit_with_log_path(log_path):
:param str log_path: path to file or directory containing the log
"""
msg = 'Please see the '
msg = ("Ask for help or search for solutions at https://community.letsencrypt.org. "
"See the ")
if os.path.isdir(log_path):
msg += 'logfiles in {0} '.format(log_path)
msg += f'logfiles in {log_path} '
else:
msg += "logfile '{0}' ".format(log_path)
msg += 'for more details.'
msg += f"logfile {log_path} "
msg += 'or re-run Certbot with -v for more details.'
sys.exit(msg)
+100 -56
View File
@@ -67,26 +67,14 @@ def _suggest_donation_if_appropriate(config):
if config.staging:
# --dry-run implies --staging
return
reporter_util = zope.component.getUtility(interfaces.IReporter)
msg = ("If you like Certbot, 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):
"""Reports on successful dry run
:param config: Configuration object
:type config: interfaces.IConfig
:returns: `None`
:rtype: None
"""
reporter_util = zope.component.getUtility(interfaces.IReporter)
assert config.verb != "renew"
reporter_util.add_message("The dry run was successful.",
reporter_util.HIGH_PRIORITY, on_crash=False)
disp = zope.component.getUtility(interfaces.IDisplay)
util.atexit_register(
disp.notification,
"If you like Certbot, please consider supporting our work by:\n"
" * Donating to ISRG / Let's Encrypt: https://letsencrypt.org/donate\n"
" * Donating to EFF: https://eff.org/donate-le",
pause=False
)
def _get_and_save_cert(le_client, config, domains=None, certname=None, lineage=None):
@@ -482,8 +470,12 @@ def _find_domains_or_certname(config, installer, question=None):
def _report_new_cert(config, cert_path, fullchain_path, key_path=None):
# type: (interfaces.IConfig, Optional[str], Optional[str], Optional[str]) -> None
"""Reports the creation of a new certificate to the user.
:param config: Configuration object
:type config: interfaces.IConfig
:param cert_path: path to certificate
:type cert_path: str
@@ -498,29 +490,67 @@ def _report_new_cert(config, cert_path, fullchain_path, key_path=None):
"""
if config.dry_run:
_report_successful_dry_run(config)
display_util.notify("The dry run was successful.")
return
assert cert_path and fullchain_path, "No certificates saved to report."
display_util.notify(
("\nSuccessfully received certificate.\n"
"Certificate is saved at: {cert_path}\n{key_msg}"
"This certificate expires on {expiry}.\n"
"These files will be updated when the certificate renews.\n{renew_msg}{nl}").format(
cert_path=fullchain_path,
expiry=crypto_util.notAfter(cert_path).date(),
key_msg="Key is saved at: {}\n".format(key_path) if key_path else "",
renew_msg="Certbot will automatically renew this certificate in the background."
if config.preconfigured_renewal else
(f'Run "{cli.cli_constants.cli_command} renew" to renew '
"expiring certificates. "
"We recommend setting up a scheduled task for renewal; see "
"https://certbot.eff.org/docs/using.html#automated-renewals "
"for instructions."),
nl="\n" if config.verb == "run" else "" # visually split output if also deploying
)
)
def _csr_report_new_cert(config: interfaces.IConfig, cert_path: Optional[str],
chain_path: Optional[str], fullchain_path: Optional[str]):
""" --csr variant of _report_new_cert.
Until --csr is overhauled (#8332) this is transitional function to report the creation
of a new certificate using --csr.
TODO: remove this function and just call _report_new_cert when --csr is overhauled.
:param config: Configuration object
:type config: interfaces.IConfig
:param str cert_path: path to cert.pem
:param str chain_path: path to chain.pem
:param str fullchain_path: path to fullchain.pem
"""
if config.dry_run:
display_util.notify("The dry run was successful.")
return
assert cert_path and fullchain_path, "No certificates saved to report."
expiry = crypto_util.notAfter(cert_path).date()
reporter_util = zope.component.getUtility(interfaces.IReporter)
# Print the path to fullchain.pem because that's what modern webservers
# (Nginx and Apache2.4) will want.
verbswitch = ' with the "certonly" option' if config.verb == "run" else ""
privkey_statement = 'Your key file has been saved at:{br}{0}{br}'.format(
key_path, br=os.linesep) if key_path else ""
# XXX Perhaps one day we could detect the presence of known old webservers
# and say something more informative here.
msg = ('Congratulations! Your certificate and chain have been saved at:{br}'
'{0}{br}{1}'
'Your certificate will expire on {2}. To obtain a new or tweaked version of this '
'certificate in the future, simply run {3} again{4}. '
'To non-interactively renew *all* of your certificates, run "{3} renew"'
.format(fullchain_path, privkey_statement, expiry, cli.cli_command, verbswitch,
br=os.linesep))
reporter_util.add_message(msg, reporter_util.MEDIUM_PRIORITY)
display_util.notify(
("\nSuccessfully received certificate.\n"
"Certificate is saved at: {cert_path}\n"
"Intermediate CA chain is saved at: {chain_path}\n"
"Full certificate chain is saved at: {fullchain_path}\n"
"This certificate expires on {expiry}.").format(
cert_path=cert_path, chain_path=chain_path,
fullchain_path=fullchain_path, expiry=expiry,
)
)
def _determine_account(config):
@@ -805,7 +835,19 @@ def _install_cert(config, le_client, domains, lineage=None):
path_provider = lineage if lineage else config
assert path_provider.cert_path is not None
le_client.deploy_certificate(domains, path_provider.key_path,
cert_name: Optional[str] = None
if isinstance(path_provider, storage.RenewableCert):
cert_name = path_provider.lineagename
elif path_provider.certname:
cert_name = path_provider.certname
else:
# Check if the cert path happens to be part of an existing lineage
try:
cert_name = cert_manager.cert_path_to_lineage(config)
except errors.Error:
pass
le_client.deploy_certificate(cert_name, domains, path_provider.key_path,
path_provider.cert_path, path_provider.chain_path, path_provider.fullchain_path)
le_client.enhance_config(domains, path_provider.chain_path)
@@ -951,7 +993,7 @@ def enhance(config, plugins):
if not enhancements.are_requested(config) and not oldstyle_enh:
msg = ("Please specify one or more enhancement types to configure. To list "
"the available enhancement types, run:\n\n%s --help enhance\n")
logger.warning(msg, sys.argv[0])
logger.error(msg, sys.argv[0])
raise errors.MisconfigurationError("No enhancements requested, exiting.")
try:
@@ -1190,6 +1232,7 @@ def run(config, plugins):
def _csr_get_and_save_cert(config, le_client):
# type: (interfaces.IConfig, client.Client) -> Tuple[Optional[str], Optional[str], Optional[str]] # pylint: disable=line-too-long
"""Obtain a cert using a user-supplied CSR
This works differently in the CSR case (for now) because we don't
@@ -1202,20 +1245,29 @@ def _csr_get_and_save_cert(config, le_client):
:param client: Client object
:type client: client.Client
:returns: `cert_path` and `fullchain_path` as absolute paths to the actual files
:returns: `cert_path`, `chain_path` and `fullchain_path` as absolute
paths to the actual files, or None for each if it's a dry-run.
:rtype: `tuple` of `str`
"""
csr, _ = config.actual_csr
csr_names = crypto_util.get_names_from_req(csr.data)
display_util.notify(
"{action} for {domains}".format(
action="Simulating a certificate request" if config.dry_run else
"Requesting a certificate",
domains=display_util.summarize_domain_list(csr_names)
)
)
cert, chain = le_client.obtain_certificate_from_csr(csr)
if config.dry_run:
logger.debug(
"Dry run: skipping saving certificate to %s", config.cert_path)
return None, None
cert_path, _, fullchain_path = le_client.save_certificate(
return None, None, None
cert_path, chain_path, fullchain_path = le_client.save_certificate(
cert, chain, os.path.normpath(config.cert_path),
os.path.normpath(config.chain_path), os.path.normpath(config.fullchain_path))
return cert_path, fullchain_path
return cert_path, chain_path, fullchain_path
def renew_cert(config, plugins, lineage):
@@ -1242,19 +1294,11 @@ def renew_cert(config, plugins, lineage):
renewed_lineage = _get_and_save_cert(le_client, config, lineage=lineage)
notify = zope.component.getUtility(interfaces.IDisplay).notification
if installer is None:
notify("new certificate deployed without reload, fullchain is {0}".format(
lineage.fullchain), pause=False)
else:
if installer and not config.dry_run:
# 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.
# Run deployer
updater.run_renewal_deployer(config, renewed_lineage, installer)
installer.restart()
notify("new certificate deployed with reload of {0} server; fullchain is {1}".format(
config.installer, lineage.fullchain), pause=False)
display_util.notify(f"Reloading {config.installer} server after certificate renewal")
installer.restart() # type: ignore
def certonly(config, plugins):
@@ -1281,8 +1325,8 @@ def certonly(config, plugins):
le_client = _init_le_client(config, auth, installer)
if config.csr:
cert_path, fullchain_path = _csr_get_and_save_cert(config, le_client)
_report_new_cert(config, cert_path, fullchain_path)
cert_path, chain_path, fullchain_path = _csr_get_and_save_cert(config, le_client)
_csr_report_new_cert(config, cert_path, chain_path, fullchain_path)
_suggest_donation_if_appropriate(config)
eff.handle_subscription(config, le_client.account)
return
+53 -6
View File
@@ -1,4 +1,5 @@
"""Manual authenticator plugin"""
import logging
from typing import Dict
import zope.component
@@ -10,11 +11,14 @@ from certbot import errors
from certbot import interfaces
from certbot import reverter
from certbot import util
from certbot._internal.cli import cli_constants
from certbot._internal import hooks
from certbot.compat import misc
from certbot.compat import os
from certbot.display import ops as display_ops
from certbot.plugins import common
logger = logging.getLogger(__name__)
@zope.interface.implementer(interfaces.IAuthenticator)
@zope.interface.provider(interfaces.IPluginFactory)
@@ -60,7 +64,7 @@ with the following value:
{validation}
"""
_DNS_VERIFY_INSTRUCTIONS = """
Before continuing, verify the TXT record has been deployed. Depending on the DNS
Before continuing, verify the TXT record has been deployed. Depending on the DNS
provider, this may take some time, from a few seconds to multiple minutes. You can
check if it has finished deploying with aid of online tools, such as the Google
Admin Toolbox: https://toolbox.googleapps.com/apps/dig/#TXT/{domain}.
@@ -125,6 +129,42 @@ permitted by DNS standards.)
'validation challenges either through shell scripts provided by '
'the user or by performing the setup manually.')
def auth_hint(self, failed_achalls):
has_chall = lambda cls: any(isinstance(achall.chall, cls) for achall in failed_achalls)
has_dns = has_chall(challenges.DNS01)
resource_names = {
challenges.DNS01: 'DNS TXT records',
challenges.HTTP01: 'challenge files',
challenges.TLSALPN01: 'TLS-ALPN certificates'
}
resources = ' and '.join(sorted([v for k, v in resource_names.items() if has_chall(k)]))
if self.conf('auth-hook'):
return (
'The Certificate Authority failed to verify the {resources} created by the '
'--manual-auth-hook. Ensure that this hook is functioning correctly{dns_hint}. '
'Refer to "{certbot} --help manual" and the Certbot User Guide.'
.format(
certbot=cli_constants.cli_command,
resources=resources,
dns_hint=(
' and that it waits a sufficient duration of time for DNS propagation'
) if has_dns else ''
)
)
else:
return (
'The Certificate Authority failed to verify the manually created {resources}. '
'Ensure that you created these in the correct location{dns_hint}.'
.format(
resources=resources,
dns_hint=(
', or try waiting longer for DNS propagation on the next attempt'
) if has_dns else ''
)
)
def get_chall_pref(self, domain):
# pylint: disable=unused-argument,missing-function-docstring
return [challenges.HTTP01, challenges.DNS01]
@@ -153,7 +193,7 @@ permitted by DNS standards.)
else:
os.environ.pop('CERTBOT_TOKEN', None)
os.environ.update(env)
_, out = self._execute_hook('auth-hook')
_, out = self._execute_hook('auth-hook', achall.domain)
env['CERTBOT_AUTH_OUTPUT'] = out.strip()
self.env[achall] = env
@@ -196,9 +236,16 @@ permitted by DNS standards.)
if 'CERTBOT_TOKEN' not in env:
os.environ.pop('CERTBOT_TOKEN', None)
os.environ.update(env)
self._execute_hook('cleanup-hook')
self._execute_hook('cleanup-hook', achall.domain)
self.reverter.recovery_routine()
def _execute_hook(self, hook_name):
return misc.execute_command(self.option_name(hook_name), self.conf(hook_name),
env=util.env_no_snap_for_external_calls())
def _execute_hook(self, hook_name, achall_domain):
returncode, err, out = misc.execute_command_status(
self.option_name(hook_name), self.conf(hook_name),
env=util.env_no_snap_for_external_calls()
)
display_ops.report_executed_command(
f"Hook '--manual-{hook_name}' for {achall_domain}", returncode, out, err)
return err, out
@@ -2,10 +2,12 @@
import logging
from typing import Optional, Tuple
import zope.component
from certbot import errors
from certbot import interfaces
from certbot._internal.plugins import disco
from certbot.compat import os
from certbot.display import util as display_util
@@ -164,7 +166,9 @@ def record_chosen_plugins(config, plugins, auth, inst):
config.authenticator, config.installer)
def choose_configurator_plugins(config, plugins, verb):
def choose_configurator_plugins(config: interfaces.IConfig, plugins: disco.PluginsRegistry,
verb: str) -> Tuple[Optional[interfaces.IInstaller],
Optional[interfaces.IAuthenticator]]:
"""
Figure out which configurator we're going to use, modifies
config.authenticator and config.installer strings to reflect that choice if
@@ -172,7 +176,7 @@ def choose_configurator_plugins(config, plugins, verb):
:raises errors.PluginSelectionError if there was a problem
:returns: (an `IAuthenticator` or None, an `IInstaller` or None)
:returns: tuple of (`IInstaller` or None, `IAuthenticator` or None)
:rtype: tuple
"""
+7 -1
View File
@@ -61,6 +61,12 @@ to serve all files under specified web root ({0})."""
"file, it needs to be on a single line, like: webroot-map = "
'{"example.com":"/var/www"}.')
def auth_hint(self, failed_achalls): # pragma: no cover
return ("The Certificate Authority failed to download the temporary challenge files "
"created by Certbot. Ensure that the listed domains serve their content from "
"the provided --webroot-path/-w and that files created there can be downloaded "
"from the internet.")
def get_chall_pref(self, domain): # pragma: no cover
# pylint: disable=unused-argument,missing-function-docstring
return [challenges.HTTP01]
@@ -183,7 +189,7 @@ to serve all files under specified web root ({0})."""
filesystem.copy_ownership_and_apply_mode(
path, prefix, 0o755, copy_user=True, copy_group=True)
except (OSError, AttributeError) as exception:
logger.info("Unable to change owner and uid of webroot directory")
logger.warning("Unable to change owner and uid of webroot directory")
logger.debug("Error was: %s", exception)
except OSError as exception:
raise errors.PluginError(
+11 -11
View File
@@ -67,18 +67,18 @@ def _reconstitute(config, full_path):
"""
try:
renewal_candidate = storage.RenewableCert(full_path, config)
except (errors.CertStorageError, IOError):
logger.warning("", exc_info=True)
logger.warning("Renewal configuration file %s is broken. Skipping.", full_path)
except (errors.CertStorageError, IOError) as error:
logger.error("Renewal configuration file %s is broken.", full_path)
logger.error("The error was: %s\nSkipping.", str(error))
logger.debug("Traceback was:\n%s", traceback.format_exc())
return None
if "renewalparams" not in renewal_candidate.configuration:
logger.warning("Renewal configuration file %s lacks "
logger.error("Renewal configuration file %s lacks "
"renewalparams. Skipping.", full_path)
return None
renewalparams = renewal_candidate.configuration["renewalparams"]
if "authenticator" not in renewalparams:
logger.warning("Renewal configuration file %s does not specify "
logger.error("Renewal configuration file %s does not specify "
"an authenticator. Skipping.", full_path)
return None
# Now restore specific values along with their data types, if
@@ -88,7 +88,7 @@ def _reconstitute(config, full_path):
restore_required_config_elements(config, renewalparams)
_restore_plugin_configs(config, renewalparams)
except (ValueError, errors.Error) as error:
logger.warning(
logger.error(
"An error occurred while parsing %s. The error was %s. "
"Skipping the file.", full_path, str(error))
logger.debug("Traceback was:\n%s", traceback.format_exc())
@@ -98,7 +98,7 @@ def _reconstitute(config, full_path):
config.domains = [util.enforce_domain_sanity(d)
for d in renewal_candidate.names()]
except errors.ConfigurationError as error:
logger.warning("Renewal configuration file %s references a certificate "
logger.error("Renewal configuration file %s references a certificate "
"that contains an invalid domain name. The problem "
"was: %s. Skipping.", full_path, error)
return None
@@ -294,12 +294,12 @@ def should_renew(config, lineage):
logger.debug("Auto-renewal forced with --force-renewal...")
return True
if lineage.should_autorenew():
logger.info("Cert is due for renewal, auto-renewing...")
logger.info("Certificate is due for renewal, auto-renewing...")
return True
if config.dry_run:
logger.info("Cert not due for renewal, but simulating renewal for dry run")
logger.info("Certificate not due for renewal, but simulating renewal for dry run")
return True
logger.info("Cert not yet due for renewal")
display_util.notify("Certificate not yet due for renewal")
return False
@@ -439,7 +439,7 @@ def handle_renewal_request(config):
try:
renewal_candidate = _reconstitute(lineage_config, renewal_file)
except Exception as e: # pylint: disable=broad-except
logger.warning("Renewal configuration file %s (cert: %s) "
logger.error("Renewal configuration file %s (cert: %s) "
"produced an unexpected error: %s. Skipping.",
renewal_file, lineagename, e)
logger.debug("Traceback was:\n%s", traceback.format_exc())
+2 -2
View File
@@ -331,7 +331,7 @@ def delete_files(config, certname):
renewal_filename, encoding='utf-8', default_encoding='utf-8')
except configobj.ConfigObjError:
# config is corrupted
logger.warning("Could not parse %s. You may wish to manually "
logger.error("Could not parse %s. You may wish to manually "
"delete the contents of %s and %s.", renewal_filename,
full_default_live_dir, full_default_archive_dir)
raise errors.CertStorageError(
@@ -340,7 +340,7 @@ def delete_files(config, certname):
# we couldn't read it, but let's at least delete it
# if this was going to fail, it already would have.
os.remove(renewal_filename)
logger.debug("Removed %s", renewal_filename)
logger.info("Removed %s", renewal_filename)
# cert files and (hopefully) live directory
# it's not guaranteed that the files are in our default storage
+1 -1
View File
@@ -29,7 +29,7 @@ def run_generic_updaters(config, lineage, plugins):
try:
installer = plug_sel.get_unprepared_installer(config, plugins)
except errors.Error as e:
logger.warning("Could not choose appropriate plugin for updaters: %s", e)
logger.error("Could not choose appropriate plugin for updaters: %s", e)
return
if installer:
_run_updaters(lineage, installer, config)
+34 -4
View File
@@ -8,6 +8,7 @@ import logging
import select
import subprocess
import sys
import warnings
from typing import Optional
from typing import Tuple
@@ -112,18 +113,22 @@ def underscores_for_unsupported_characters_in_path(path: str) -> str:
return drive + tail.replace(':', '_')
def execute_command(cmd_name: str, shell_cmd: str, env: Optional[dict] = None) -> Tuple[str, str]:
def execute_command_status(cmd_name: str, shell_cmd: str,
env: Optional[dict] = None) -> Tuple[int, str, str]:
"""
Run a command:
- on Linux command will be run by the standard shell selected with
subprocess.run(shell=True)
- on Windows command will be run in a Powershell shell
This differs from execute_command: it returns the exit code, and does not log the result
and output of the command.
:param str cmd_name: the user facing name of the hook being run
:param str shell_cmd: shell command to execute
:param dict env: environ to pass into subprocess.run
:returns: `tuple` (`str` stderr, `str` stdout)
:returns: `tuple` (`int` returncode, `str` stderr, `str` stdout)
"""
logger.info("Running %s command: %s", cmd_name, shell_cmd)
@@ -139,12 +144,37 @@ def execute_command(cmd_name: str, shell_cmd: str, env: Optional[dict] = None) -
# universal_newlines causes stdout and stderr to be str objects instead of
# bytes in Python 3
out, err = proc.stdout, proc.stderr
return proc.returncode, err, out
def execute_command(cmd_name: str, shell_cmd: str, env: Optional[dict] = None) -> Tuple[str, str]:
"""
Run a command:
- on Linux command will be run by the standard shell selected with
subprocess.run(shell=True)
- on Windows command will be run in a Powershell shell
This differs from execute_command: it returns the exit code, and does not log the result
and output of the command.
:param str cmd_name: the user facing name of the hook being run
:param str shell_cmd: shell command to execute
:param dict env: environ to pass into subprocess.run
:returns: `tuple` (`str` stderr, `str` stdout)
"""
# Deprecation per https://github.com/certbot/certbot/issues/8854
warnings.warn(
"execute_command will be deprecated in the future, use execute_command_status instead",
PendingDeprecationWarning
)
returncode, err, out = execute_command_status(cmd_name, shell_cmd, env)
base_cmd = os.path.basename(shell_cmd.split(None, 1)[0])
if out:
logger.info('Output from %s command %s:\n%s', cmd_name, base_cmd, out)
if proc.returncode != 0:
if returncode != 0:
logger.error('%s command "%s" returned error code %d',
cmd_name, shell_cmd, proc.returncode)
cmd_name, shell_cmd, returncode)
if err:
logger.error('Error output from %s command %s:\n%s', cmd_name, base_cmd, err)
return err, out
+19 -4
View File
@@ -9,6 +9,7 @@ import logging
import re
import warnings
from typing import List
# See https://github.com/pyca/cryptography/issues/4275
from cryptography import x509 # type: ignore
from cryptography.exceptions import InvalidSignature
@@ -64,7 +65,8 @@ def init_save_key(
bits=key_size, elliptic_curve=elliptic_curve or "secp256r1", key_type=key_type,
)
except ValueError as err:
logger.error("", exc_info=True)
logger.debug("", exc_info=True)
logger.error("Encountered error while making key: %s", str(err))
raise err
config = zope.component.getUtility(interfaces.IConfig)
@@ -387,8 +389,9 @@ def _load_cert_or_req(cert_or_req_str, load_func,
typ=crypto.FILETYPE_PEM):
try:
return load_func(typ, cert_or_req_str)
except crypto.Error:
logger.error("", exc_info=True)
except crypto.Error as err:
logger.debug("", exc_info=True)
logger.error("Encountered error while loading certificate or csr: %s", str(err))
raise
@@ -437,6 +440,18 @@ def get_names_from_cert(csr, typ=crypto.FILETYPE_PEM):
csr, crypto.load_certificate, typ)
def get_names_from_req(csr: str, typ: int = crypto.FILETYPE_PEM) -> List[str]:
"""Get a list of domains from a CSR, including the CN if it is set.
:param str cert: CSR (encoded).
:param typ: `crypto.FILETYPE_PEM` or `crypto.FILETYPE_ASN1`
:returns: A list of domain names.
:rtype: list
"""
return _get_names_from_cert_or_req(csr, crypto.load_certificate_request, typ)
def dump_pyopenssl_chain(chain, filetype=crypto.FILETYPE_PEM):
"""Dump certificate chain into a bundle.
@@ -589,7 +604,7 @@ def find_chain_with_issuer(fullchains, issuer_cn, warn_on_no_match=False):
# Nothing matched, return whatever was first in the list.
if warn_on_no_match:
logger.info("Certbot has been configured to prefer certificate chains with "
logger.warning("Certbot has been configured to prefer certificate chains with "
"issuer '%s', but no chain from the CA matched this issuer. Using "
"the default certificate chain instead.", issuer_cn)
return fullchains[0]
+27 -11
View File
@@ -1,5 +1,6 @@
"""Contains UI methods for LE user operations."""
import logging
from textwrap import indent
import zope.component
@@ -240,25 +241,22 @@ def success_installation(domains):
:param list domains: domain names which were enabled
"""
z_util(interfaces.IDisplay).notification(
"Congratulations! You have successfully enabled {0}".format(
_gen_https_names(domains)),
pause=False)
display_util.notify(
"Congratulations! You have successfully enabled HTTPS on {0}"
.format(_gen_https_names(domains))
)
def success_renewal(domains):
def success_renewal(unused_domains):
"""Display a box confirming the renewal of an existing certificate.
:param list domains: domain names which were renewed
"""
z_util(interfaces.IDisplay).notification(
display_util.notify(
"Your existing certificate has been successfully renewed, and the "
"new certificate has been installed.{1}{1}"
"The new certificate covers the following domains: {0}".format(
_gen_https_names(domains),
os.linesep),
pause=False)
"new certificate has been installed."
)
def success_revocation(cert_path):
@@ -273,6 +271,24 @@ def success_revocation(cert_path):
)
def report_executed_command(command_name: str, returncode: int, stdout: str, stderr: str) -> None:
"""Display a message describing the success or failure of an executed process (e.g. hook).
:param str command_name: Human-readable description of the executed command
:param int returncode: The exit code of the executed command
:param str stdout: The stdout output of the executed command
:param str stderr: The stderr output of the executed command
"""
out_s, err_s = stdout.strip(), stderr.strip()
if returncode != 0:
logger.warning("%s reported error code %d", command_name, returncode)
if out_s:
display_util.notify(f"{command_name} ran with output:\n{indent(out_s, ' ')}")
if err_s:
logger.warning("%s ran with error output:\n%s", command_name, indent(err_s, ' '))
def _gen_https_names(domains):
"""Returns a string of the https domains.
+5 -5
View File
@@ -192,7 +192,7 @@ def _check_ocsp_cryptography(cert_path: str, chain_path: str, url: str, timeout:
# Check OCSP response validity
if response_ocsp.response_status != ocsp.OCSPResponseStatus.SUCCESSFUL:
logger.error("Invalid OCSP response status for %s: %s",
logger.warning("Invalid OCSP response status for %s: %s",
cert_path, response_ocsp.response_status)
return False
@@ -200,13 +200,13 @@ def _check_ocsp_cryptography(cert_path: str, chain_path: str, url: str, timeout:
try:
_check_ocsp_response(response_ocsp, request, issuer, cert_path)
except UnsupportedAlgorithm as e:
logger.error(str(e))
logger.warning(str(e))
except errors.Error as e:
logger.error(str(e))
logger.warning(str(e))
except InvalidSignature:
logger.error('Invalid signature on OCSP response for %s', cert_path)
logger.warning('Invalid signature on OCSP response for %s', cert_path)
except AssertionError as error:
logger.error('Invalid OCSP response for %s: %s.', cert_path, str(error))
logger.warning('Invalid OCSP response for %s: %s.', cert_path, str(error))
else:
# Check OCSP certificate status
logger.debug("OCSP certificate status for %s is: %s",
+27
View File
@@ -97,6 +97,33 @@ class Plugin:
"""Find a configuration value for variable ``var``."""
return getattr(self.config, self.dest(var))
def auth_hint(self, failed_achalls):
# type: (List[achallenges.AnnotatedChallenge]) -> str
"""Human-readable string to help the user troubleshoot the authenticator.
Shown to the user if one or more of the attempted challenges were not a success.
Should describe, in simple language, what the authenticator tried to do, what went
wrong and what the user should try as their "next steps".
TODO: auth_hint belongs in IAuthenticator but can't be added until the next major
version of Certbot. For now, it lives in .Plugin and auth_handler will only call it
on authenticators that subclass .Plugin. For now, inherit from `.Plugin` to implement
and/or override the method.
:param list failed_achalls: List of one or more failed challenges
(:class:`achallenges.AnnotatedChallenge` subclasses).
:rtype str:
"""
# This is a fallback hint. Authenticators should implement their own auth_hint that
# addresses the specific mechanics of that authenticator.
challs = " and ".join(sorted({achall.typ for achall in failed_achalls}))
return ("The Certificate Authority couldn't exterally verify that the {name} plugin "
"completed the required {challs} challenges. Ensure the plugin is configured "
"correctly and that the changes it makes are accessible from the internet."
.format(name=self.name, challs=challs))
class Installer(Plugin):
"""An installer base class with reverter and ssl_dhparam methods defined.
+10 -1
View File
@@ -37,6 +37,15 @@ class DNSAuthenticator(common.Plugin):
help='The number of seconds to wait for DNS to propagate before asking the ACME server '
'to verify the DNS record.')
def auth_hint(self, failed_achalls):
delay = self.conf('propagation-seconds')
return (
'The Certificate Authority failed to verify the DNS TXT records created by --{name}. '
'Ensure the above domains are hosted by this DNS provider, or try increasing '
'--{name}-propagation-seconds (currently {secs} second{suffix}).'
.format(name=self.name, secs=delay, suffix='s' if delay != 1 else '')
)
def get_chall_pref(self, unused_domain): # pylint: disable=missing-function-docstring
return [challenges.DNS01]
@@ -63,7 +72,7 @@ class DNSAuthenticator(common.Plugin):
# DNS updates take time to propagate and checking to see if the update has occurred is not
# reliable (the machine this code is running on might be able to see an update before
# the ACME server). So: we sleep for a short amount of time we believe to be long enough.
logger.info("Waiting %d seconds for DNS changes to propagate",
display_util.notify("Waiting %d seconds for DNS changes to propagate" %
self.conf('propagation-seconds'))
sleep(self.conf('propagation-seconds'))
@@ -67,7 +67,8 @@ class _LexiconAwareTestCase(Protocol):
class BaseLexiconAuthenticatorTest(dns_test_common.BaseAuthenticatorTest):
def test_perform(self: _AuthenticatorCallableLexiconTestCase):
@test_util.patch_get_utility()
def test_perform(self: _AuthenticatorCallableLexiconTestCase, unused_mock_get_utility):
self.auth.perform([self.achall])
expected = [mock.call.add_txt_record(DOMAIN, '_acme-challenge.'+DOMAIN, mock.ANY)]
+1 -1
View File
@@ -514,7 +514,7 @@ class Reverter:
filesystem.replace(self.config.in_progress_dir, final_dir)
return
except OSError:
logger.warning("Extreme, unexpected race condition, retrying (%s)", timestamp)
logger.warning("Unexpected race condition, retrying (%s)", timestamp)
# After 10 attempts... something is probably wrong here...
logger.error(
+3 -2
View File
@@ -16,6 +16,7 @@ from typing import Dict
from typing import Text
from typing import Tuple
from typing import Union
import warnings
import configargparse
@@ -433,14 +434,14 @@ def safe_email(email):
"""Scrub email address before using it."""
if EMAIL_REGEX.match(email) is not None:
return not email.startswith(".") and ".." not in email
logger.warning("Invalid email address: %s.", email)
logger.error("Invalid email address: %s.", email)
return False
class DeprecatedArgumentAction(argparse.Action):
"""Action to log a warning when an argument is used."""
def __call__(self, unused1, unused2, unused3, option_string=None):
logger.warning("Use of %s is deprecated.", option_string)
warnings.warn("Use of %s is deprecated." % option_string, DeprecationWarning)
def add_deprecated_argument(add_argument, argument_name, nargs):
+60 -14
View File
@@ -17,6 +17,7 @@ from certbot import achallenges
from certbot import errors
from certbot import interfaces
from certbot import util
from certbot.plugins import common as plugin_common
from certbot.tests import acme_util
from certbot.tests import util as test_util
@@ -327,7 +328,8 @@ class HandleAuthorizationsTest(unittest.TestCase):
mock_order = mock.MagicMock(authorizations=authzrs)
with mock.patch('certbot._internal.auth_handler._report_failed_authzrs') as mock_report:
with mock.patch('certbot._internal.auth_handler.AuthHandler._report_failed_authzrs') \
as mock_report:
valid_authzr = self.handler.handle_authorizations(mock_order, True)
# Because best_effort=True, we did not blow up. Instead ...
@@ -474,10 +476,18 @@ class GenChallengePathTest(unittest.TestCase):
class ReportFailedAuthzrsTest(unittest.TestCase):
"""Tests for certbot._internal.auth_handler._report_failed_authzrs."""
"""Tests for certbot._internal.auth_handler.AuthHandler._report_failed_authzrs."""
# pylint: disable=protected-access
def setUp(self):
from certbot._internal.auth_handler import AuthHandler
self.mock_auth = mock.MagicMock(spec=plugin_common.Plugin, name="buzz")
self.mock_auth.name = "buzz"
self.mock_auth.auth_hint.return_value = "the buzz hint"
self.handler = AuthHandler(self.mock_auth, mock.MagicMock(), mock.MagicMock(), [])
kwargs = {
"chall": acme_util.HTTP01,
"uri": "uri",
@@ -504,21 +514,57 @@ class ReportFailedAuthzrsTest(unittest.TestCase):
self.authzr2.body.identifier.value = 'foo.bar'
self.authzr2.body.challenges = [http_01_diff]
@test_util.patch_get_utility()
def test_same_error_and_domain(self, mock_zope):
from certbot._internal import auth_handler
@mock.patch('certbot._internal.auth_handler.display_util.notify')
def test_same_error_and_domain(self, mock_notify):
self.handler._report_failed_authzrs([self.authzr1])
mock_notify.assert_called_with(
'\n'
'Certbot failed to authenticate some domains (authenticator: buzz). '
'The Certificate Authority reported these problems:\n'
' Domain: example.com\n'
' Type: tls\n'
' Detail: detail\n'
'\n'
' Domain: example.com\n'
' Type: tls\n'
' Detail: detail\n'
'\nHint: the buzz hint\n'
)
auth_handler._report_failed_authzrs([self.authzr1], 'key')
call_list = mock_zope().add_message.call_args_list
self.assertEqual(len(call_list), 1)
self.assertIn("Domain: example.com\nType: tls\nDetail: detail", call_list[0][0][0])
@mock.patch('certbot._internal.auth_handler.display_util.notify')
def test_different_errors_and_domains(self, mock_notify):
self.mock_auth.name = "quux"
self.mock_auth.auth_hint.return_value = "quuuuuux"
self.handler._report_failed_authzrs([self.authzr1, self.authzr2])
mock_notify.assert_called_with(
'\n'
'Certbot failed to authenticate some domains (authenticator: quux). '
'The Certificate Authority reported these problems:\n'
' Domain: foo.bar\n'
' Type: dnssec\n'
' Detail: detail\n'
'\n'
' Domain: example.com\n'
' Type: tls\n'
' Detail: detail\n'
'\n'
' Domain: example.com\n'
' Type: tls\n'
' Detail: detail\n'
'\nHint: quuuuuux\n'
)
@test_util.patch_get_utility()
def test_different_errors_and_domains(self, mock_zope):
from certbot._internal import auth_handler
@mock.patch('certbot._internal.auth_handler.display_util.notify')
def test_non_subclassed_authenticator(self, mock_notify):
"""If authenticator not derived from common.Plugin, we shouldn't call .auth_hint"""
from certbot._internal.auth_handler import AuthHandler
auth_handler._report_failed_authzrs([self.authzr1, self.authzr2], 'key')
self.assertEqual(mock_zope().add_message.call_count, 2)
self.mock_auth = mock.MagicMock(name="quuz")
self.mock_auth.name = "quuz"
self.mock_auth.auth_hint.side_effect = Exception
self.handler = AuthHandler(self.mock_auth, mock.MagicMock(), mock.MagicMock(), [])
self.handler._report_failed_authzrs([self.authzr1])
self.assertEqual(mock_notify.call_count, 1)
def gen_auth_resp(chall_list):
+4
View File
@@ -578,6 +578,10 @@ class CertPathToLineageTest(storage_test.BaseRenewableCertTest):
self._archive_files(x, 'fullchain')]
self.assertEqual('example.org', self._call(self.config))
def test_only_path(self):
self.config.cert_path = self.fullchain
self.assertEqual('example.org', self._call(self.config))
class MatchAndCheckOverlaps(storage_test.BaseRenewableCertTest):
"""Tests for certbot._internal.cert_manager.match_and_check_overlaps w/o overlapping
+72 -26
View File
@@ -99,7 +99,8 @@ class RegisterTest(test_util.ConfigTestCase):
self._call()
self.assertIs(mock_prepare.called, True)
def test_it(self):
@test_util.patch_get_utility()
def test_it(self, unused_mock_get_utility):
with mock.patch("certbot._internal.client.acme_client.BackwardsCompatibleClientV2") as mock_client:
mock_client().external_account_required.side_effect = self._false_mock
with mock.patch("certbot._internal.eff.handle_subscription"):
@@ -159,7 +160,8 @@ class RegisterTest(test_util.ConfigTestCase):
# check Certbot created an account with no email. Contact should return empty
self.assertFalse(mock_client().new_account_and_tos.call_args[0][0].contact)
def test_with_eab_arguments(self):
@test_util.patch_get_utility()
def test_with_eab_arguments(self, unused_mock_get_utility):
with mock.patch("certbot._internal.client.acme_client.BackwardsCompatibleClientV2") as mock_client:
mock_client().client.directory.__getitem__ = mock.Mock(
side_effect=self._new_acct_dir_mock
@@ -174,7 +176,8 @@ class RegisterTest(test_util.ConfigTestCase):
self.assertIs(mock_eab_from_data.called, True)
def test_without_eab_arguments(self):
@test_util.patch_get_utility()
def test_without_eab_arguments(self, unused_mock_get_utility):
with mock.patch("certbot._internal.client.acme_client.BackwardsCompatibleClientV2") as mock_client:
mock_client().external_account_required.side_effect = self._false_mock
with mock.patch("certbot._internal.eff.handle_subscription"):
@@ -315,7 +318,7 @@ class ClientTest(ClientTestCommon):
errors.Error,
self.client.obtain_certificate_from_csr,
test_csr)
mock_logger.warning.assert_called_once_with(mock.ANY)
mock_logger.error.assert_called_once_with(mock.ANY)
@mock.patch("certbot._internal.client.crypto_util")
def test_obtain_certificate(self, mock_crypto_util):
@@ -515,15 +518,16 @@ class ClientTest(ClientTestCommon):
shutil.rmtree(tmp_path)
def test_deploy_certificate_success(self):
@test_util.patch_get_utility()
def test_deploy_certificate_success(self, mock_util):
self.assertRaises(errors.Error, self.client.deploy_certificate,
["foo.bar"], "key", "cert", "chain", "fullchain")
"foo.bar", ["foo.bar"], "key", "cert", "chain", "fullchain")
installer = mock.MagicMock()
self.client.installer = installer
self.client.deploy_certificate(
["foo.bar"], "key", "cert", "chain", "fullchain")
"foo.bar", ["foo.bar"], "key", "cert", "chain", "fullchain")
installer.deploy_cert.assert_called_once_with(
cert_path=os.path.abspath("cert"),
chain_path=os.path.abspath("chain"),
@@ -533,46 +537,81 @@ class ClientTest(ClientTestCommon):
self.assertEqual(installer.save.call_count, 2)
installer.restart.assert_called_once_with()
def test_deploy_certificate_failure(self):
@mock.patch('certbot._internal.client.display_util.notify')
@test_util.patch_get_utility()
def test_deploy_certificate_failure(self, mock_util, mock_notify):
installer = mock.MagicMock()
self.client.installer = installer
self.config.installer = "foobar"
installer.deploy_cert.side_effect = errors.PluginError
self.assertRaises(errors.PluginError, self.client.deploy_certificate,
["foo.bar"], "key", "cert", "chain", "fullchain")
"foo.bar", ["foo.bar"], "key", "cert", "chain", "fullchain")
installer.recovery_routine.assert_called_once_with()
def test_deploy_certificate_save_failure(self):
mock_notify.assert_any_call('Deploying certificate')
mock_notify.assert_any_call(
'Failed to install the certificate (installer: foobar). '
'Try again after fixing errors by running:\n\n certbot install --cert-name foo.bar\n'
)
@mock.patch('certbot._internal.client.display_util.notify')
@test_util.patch_get_utility()
def test_deploy_certificate_failure_no_certname(self, mock_util, mock_notify):
installer = mock.MagicMock()
self.client.installer = installer
self.config.installer = "foobar"
installer.deploy_cert.side_effect = errors.PluginError
self.assertRaises(errors.PluginError, self.client.deploy_certificate,
None, ["foo.bar"], "key", "cert", "chain", "fullchain")
installer.recovery_routine.assert_called_once_with()
mock_notify.assert_any_call('Deploying certificate')
mock_notify.assert_any_call(
'Failed to install the certificate (installer: foobar).'
)
@test_util.patch_get_utility()
def test_deploy_certificate_save_failure(self, mock_util):
installer = mock.MagicMock()
self.client.installer = installer
installer.save.side_effect = errors.PluginError
self.assertRaises(errors.PluginError, self.client.deploy_certificate,
["foo.bar"], "key", "cert", "chain", "fullchain")
"foo.bar", ["foo.bar"], "key", "cert", "chain", "fullchain")
installer.recovery_routine.assert_called_once_with()
@mock.patch('certbot._internal.client.display_util.notify')
@test_util.patch_get_utility()
def test_deploy_certificate_restart_failure(self, mock_get_utility):
def test_deploy_certificate_restart_failure(self, mock_get_utility, mock_notify):
installer = mock.MagicMock()
installer.restart.side_effect = [errors.PluginError, None]
self.client.installer = installer
self.assertRaises(errors.PluginError, self.client.deploy_certificate,
["foo.bar"], "key", "cert", "chain", "fullchain")
self.assertEqual(mock_get_utility().add_message.call_count, 1)
"foo.bar", ["foo.bar"], "key", "cert", "chain", "fullchain")
mock_notify.assert_called_with(
'We were unable to install your certificate, however, we successfully restored '
'your server to its prior configuration.')
installer.rollback_checkpoints.assert_called_once_with()
self.assertEqual(installer.restart.call_count, 2)
@mock.patch('certbot._internal.client.logger')
@test_util.patch_get_utility()
def test_deploy_certificate_restart_failure2(self, mock_get_utility):
def test_deploy_certificate_restart_failure2(self, mock_get_utility, mock_logger):
installer = mock.MagicMock()
installer.restart.side_effect = errors.PluginError
installer.rollback_checkpoints.side_effect = errors.ReverterError
self.client.installer = installer
self.assertRaises(errors.PluginError, self.client.deploy_certificate,
["foo.bar"], "key", "cert", "chain", "fullchain")
self.assertEqual(mock_get_utility().add_message.call_count, 1)
"foo.bar", ["foo.bar"], "key", "cert", "chain", "fullchain")
self.assertEqual(mock_logger.error.call_count, 1)
self.assertIn(
'An error occurred and we failed to restore your config',
mock_logger.error.call_args[0][0])
installer.rollback_checkpoints.assert_called_once_with()
self.assertEqual(installer.restart.call_count, 1)
@@ -601,23 +640,23 @@ class EnhanceConfigTest(ClientTestCommon):
self.config.hsts = True
with mock.patch("certbot._internal.client.logger") as mock_logger:
self.client.enhance_config([self.domain], None)
self.assertEqual(mock_logger.warning.call_count, 1)
self.assertEqual(mock_logger.error.call_count, 1)
self.client.installer.enhance.assert_not_called()
@mock.patch("certbot._internal.client.logger")
def test_already_exists_header(self, mock_log):
self.config.hsts = True
self._test_with_already_existing()
self.assertIs(mock_log.warning.called, True)
self.assertEqual(mock_log.warning.call_args[0][1],
self.assertIs(mock_log.info.called, True)
self.assertEqual(mock_log.info.call_args[0][1],
'Strict-Transport-Security')
@mock.patch("certbot._internal.client.logger")
def test_already_exists_redirect(self, mock_log):
self.config.redirect = True
self._test_with_already_existing()
self.assertIs(mock_log.warning.called, True)
self.assertEqual(mock_log.warning.call_args[0][1],
self.assertIs(mock_log.info.called, True)
self.assertEqual(mock_log.info.call_args[0][1],
'redirect')
@mock.patch("certbot._internal.client.logger")
@@ -659,7 +698,7 @@ class EnhanceConfigTest(ClientTestCommon):
def test_enhance_failure(self):
self.client.installer = mock.MagicMock()
self.client.installer.enhance.side_effect = errors.PluginError
self._test_error()
self._test_error(enhance_error=True)
self.client.installer.recovery_routine.assert_called_once_with()
def test_save_failure(self):
@@ -685,12 +724,19 @@ class EnhanceConfigTest(ClientTestCommon):
self._test_error()
self.assertIs(self.client.installer.restart.called, True)
def _test_error(self):
def _test_error(self, enhance_error=False, restart_error=False):
self.config.redirect = True
with test_util.patch_get_utility() as mock_gu:
with mock.patch('certbot._internal.client.logger') as mock_logger, \
test_util.patch_get_utility() as mock_gu:
self.assertRaises(
errors.PluginError, self._test_with_all_supported)
self.assertEqual(mock_gu().add_message.call_count, 1)
if enhance_error:
self.assertEqual(mock_logger.error.call_count, 1)
self.assertIn('Unable to set enhancement', mock_logger.error.call_args_list[0][0][0])
if restart_error:
mock_logger.critical.assert_called_with(
'Rolling back to previous server configuration...')
def _test_with_all_supported(self):
if self.client.installer is None:
+41 -1
View File
@@ -4,17 +4,22 @@ try:
except ImportError: # pragma: no cover
from unittest import mock # type: ignore
import unittest
import warnings
from certbot.compat import os
class ExecuteTest(unittest.TestCase):
"""Tests for certbot.compat.misc.execute_command."""
@classmethod
def _call(cls, *args, **kwargs):
from certbot.compat.misc import execute_command
return execute_command(*args, **kwargs)
# execute_command is superseded by execute_command_status
with warnings.catch_warnings():
warnings.simplefilter('ignore', category=PendingDeprecationWarning)
return execute_command(*args, **kwargs)
def test_it(self):
for returncode in range(0, 2):
@@ -47,3 +52,38 @@ class ExecuteTest(unittest.TestCase):
mock.ANY, stdout)
if stderr or returncode:
self.assertIs(mock_logger.error.called, True)
class ExecuteStatusTest(ExecuteTest):
"""Tests for certbot.compat.misc.execute_command_status."""
@classmethod
def _call(cls, *args, **kwargs):
from certbot.compat.misc import execute_command_status
return execute_command_status(*args, **kwargs)
def _test_common(self, returncode, stdout, stderr):
given_command = "foo"
given_name = "foo-hook"
with mock.patch("certbot.compat.misc.subprocess.run") as mock_run:
mock_run.return_value.stdout = stdout
mock_run.return_value.stderr = stderr
mock_run.return_value.returncode = returncode
with mock.patch("certbot.compat.misc.logger") as mock_logger:
self.assertEqual(self._call(given_name, given_command), (returncode, stderr, stdout))
executed_command = mock_run.call_args[1].get(
"args", mock_run.call_args[0][0])
if os.name == 'nt':
expected_command = ['powershell.exe', '-Command', given_command]
else:
expected_command = given_command
self.assertEqual(executed_command, expected_command)
self.assertEqual(executed_command, expected_command)
mock_logger.info.assert_any_call("Running %s command: %s",
given_name, given_command)
if __name__ == "__main__":
unittest.main() # pragma: no cover
+34 -3
View File
@@ -381,6 +381,37 @@ class GetNamesFromCertTest(unittest.TestCase):
self.assertRaises(OpenSSL.crypto.Error, self._call, "hello there")
class GetNamesFromReqTest(unittest.TestCase):
"""Tests for certbot.crypto_util.get_names_from_req."""
@classmethod
def _call(cls, *args, **kwargs):
from certbot.crypto_util import get_names_from_req
return get_names_from_req(*args, **kwargs)
def test_nonames(self):
self.assertEqual(
[],
self._call(test_util.load_vector('csr-nonames_512.pem')))
def test_nosans(self):
self.assertEqual(
['example.com'],
self._call(test_util.load_vector('csr-nosans_512.pem')))
def test_sans(self):
self.assertEqual(
['example.com', 'example.org', 'example.net', 'example.info',
'subdomain.example.com', 'other.subdomain.example.com'],
self._call(test_util.load_vector('csr-6sans_512.pem')))
def test_der(self):
from OpenSSL.crypto import FILETYPE_ASN1
self.assertEqual(
['Example.com'],
self._call(test_util.load_vector('csr_512.der'), typ=FILETYPE_ASN1))
class CertLoaderTest(unittest.TestCase):
"""Tests for certbot.crypto_util.pyopenssl_load_certificate"""
@@ -493,13 +524,13 @@ class FindChainWithIssuerTest(unittest.TestCase):
self.assertEqual(matched, fullchains[0])
mock_info.assert_not_called()
@mock.patch('certbot.crypto_util.logger.info')
def test_warning_on_no_match(self, mock_info):
@mock.patch('certbot.crypto_util.logger.warning')
def test_warning_on_no_match(self, mock_warning):
fullchains = self._all_fullchains()
matched = self._call(fullchains, "non-existent issuer",
warn_on_no_match=True)
self.assertEqual(matched, fullchains[0])
mock_info.assert_called_once_with("Certbot has been configured to prefer "
mock_warning.assert_called_once_with("Certbot has been configured to prefer "
"certificate chains with issuer '%s', but no chain from the CA matched "
"this issuer. Using the default certificate chain instead.",
"non-existent issuer")
+31 -9
View File
@@ -340,15 +340,16 @@ class SuccessInstallationTest(unittest.TestCase):
from certbot.display.ops import success_installation
success_installation(names)
@test_util.patch_get_utility("certbot.display.util.notify")
@test_util.patch_get_utility("certbot.display.ops.z_util")
def test_success_installation(self, mock_util):
def test_success_installation(self, mock_util, mock_notify):
mock_util().notification.return_value = None
names = ["example.com", "abc.com"]
self._call(names)
self.assertEqual(mock_util().notification.call_count, 1)
arg = mock_util().notification.call_args_list[0][0][0]
self.assertEqual(mock_notify.call_count, 1)
arg = mock_notify.call_args_list[0][0][0]
for name in names:
self.assertIn(name, arg)
@@ -361,18 +362,15 @@ class SuccessRenewalTest(unittest.TestCase):
from certbot.display.ops import success_renewal
success_renewal(names)
@test_util.patch_get_utility("certbot.display.util.notify")
@test_util.patch_get_utility("certbot.display.ops.z_util")
def test_success_renewal(self, mock_util):
def test_success_renewal(self, mock_util, mock_notify):
mock_util().notification.return_value = None
names = ["example.com", "abc.com"]
self._call(names)
self.assertEqual(mock_util().notification.call_count, 1)
arg = mock_util().notification.call_args_list[0][0][0]
for name in names:
self.assertIn(name, arg)
self.assertEqual(mock_notify.call_count, 1)
class SuccessRevocationTest(unittest.TestCase):
"""Test the success revocation message."""
@@ -500,5 +498,29 @@ class ChooseValuesTest(unittest.TestCase):
self.assertEqual(mock_util().checklist.call_args[0][0], question)
@mock.patch('certbot.display.ops.logger')
@mock.patch('certbot.display.util.notify')
class ReportExecutedCommand(unittest.TestCase):
"""Test report_executed_command"""
@classmethod
def _call(cls, cmd_name: str, rc: int, out: str, err: str):
from certbot.display.ops import report_executed_command
report_executed_command(cmd_name, rc, out, err)
def test_mixed_success(self, mock_notify, mock_logger):
self._call("some-hook", 0, "Did a thing", "Some warning")
self.assertEqual(mock_logger.warning.call_count, 1)
self.assertEqual(mock_notify.call_count, 1)
def test_mixed_error(self, mock_notify, mock_logger):
self._call("some-hook", -127, "Did a thing", "Some warning")
self.assertEqual(mock_logger.warning.call_count, 2)
self.assertEqual(mock_notify.call_count, 1)
def test_empty_success(self, mock_notify, mock_logger):
self._call("some-hook", 0, "\n", " ")
self.assertEqual(mock_logger.warning.call_count, 0)
self.assertEqual(mock_notify.call_count, 0)
if __name__ == "__main__":
unittest.main() # pragma: no cover
+8 -8
View File
@@ -72,14 +72,14 @@ class HookTest(test_util.ConfigTestCase):
@classmethod
def _call_with_mock_execute(cls, *args, **kwargs):
"""Calls self._call after mocking out certbot.compat.misc.execute_command.
"""Calls self._call after mocking out certbot.compat.misc.execute_command_status.
The mock execute object is returned rather than the return value
of self._call.
"""
with mock.patch("certbot.compat.misc.execute_command") as mock_execute:
mock_execute.return_value = ("", "")
with mock.patch("certbot.compat.misc.execute_command_status") as mock_execute:
mock_execute.return_value = (0, "", "")
cls._call(*args, **kwargs)
return mock_execute
@@ -292,7 +292,7 @@ class RenewalHookTest(HookTest):
# pylint: disable=abstract-method
def _call_with_mock_execute(self, *args, **kwargs):
"""Calls self._call after mocking out certbot.compat.misc.execute_command.
"""Calls self._call after mocking out certbot.compat.misc.execute_command_status.
The mock execute object is returned rather than the return value
of self._call. The mock execute object asserts that environment
@@ -311,9 +311,9 @@ class RenewalHookTest(HookTest):
"""
self.assertEqual(os.environ["RENEWED_DOMAINS"], " ".join(domains))
self.assertEqual(os.environ["RENEWED_LINEAGE"], lineage)
return ("", "")
return (0, "", "")
with mock.patch("certbot.compat.misc.execute_command") as mock_execute:
with mock.patch("certbot.compat.misc.execute_command_status") as mock_execute:
mock_execute.side_effect = execute_side_effect
self._call(*args, **kwargs)
return mock_execute
@@ -345,7 +345,7 @@ class DeployHookTest(RenewalHookTest):
mock_execute = self._call_with_mock_execute(
self.config, ["example.org"], "/foo/bar")
self.assertIs(mock_execute.called, False)
self.assertTrue(mock_logger.warning.called)
self.assertTrue(mock_logger.info.called)
@mock.patch("certbot._internal.hooks.logger")
def test_no_hook(self, mock_logger):
@@ -393,7 +393,7 @@ class RenewHookTest(RenewalHookTest):
mock_execute = self._call_with_mock_execute(
self.config, ["example.org"], "/foo/bar")
self.assertIs(mock_execute.called, False)
self.assertEqual(mock_logger.warning.call_count, 2)
self.assertEqual(mock_logger.info.call_count, 2)
def test_no_hooks(self):
self.config.renew_hook = None
+19 -11
View File
@@ -57,7 +57,7 @@ class PreArgParseSetupTest(unittest.TestCase):
mock_register.assert_called_once_with(logging.shutdown)
mock_sys.excepthook(1, 2, 3)
mock_except_hook.assert_called_once_with(
memory_handler, 1, 2, 3, debug=True, log_path=mock.ANY)
memory_handler, 1, 2, 3, debug=True, quiet=False, log_path=mock.ANY)
class PostArgParseSetupTest(test_util.ConfigTestCase):
@@ -101,15 +101,17 @@ class PostArgParseSetupTest(test_util.ConfigTestCase):
mock_sys.version_info = sys.version_info
self._call(self.config)
log_path = os.path.join(self.config.logs_dir, 'letsencrypt.log')
self.root_logger.removeHandler.assert_called_once_with(
self.memory_handler)
self.assertTrue(self.root_logger.addHandler.called)
self.assertTrue(os.path.exists(os.path.join(
self.config.logs_dir, 'letsencrypt.log')))
self.assertTrue(os.path.exists(log_path))
self.assertFalse(os.path.exists(self.temp_path))
mock_sys.excepthook(1, 2, 3)
mock_except_hook.assert_called_once_with(
1, 2, 3, debug=self.config.debug, log_path=self.config.logs_dir)
1, 2, 3, debug=self.config.debug,
quiet=self.config.quiet, log_path=log_path)
level = self.stream_handler.level
if self.config.quiet:
@@ -319,6 +321,12 @@ class PostArgParseExceptHookTest(unittest.TestCase):
self._assert_exception_logged(mock_logger.error, exc_type)
self._assert_logfile_output(output)
def test_quiet(self):
exc_type = ValueError
mock_logger, output = self._test_common(exc_type, debug=True, quiet=True)
self._assert_exception_logged(mock_logger.error, exc_type)
self.assertNotIn('See the logfile', output)
def test_custom_error(self):
exc_type = errors.PluginError
mock_logger, output = self._test_common(exc_type, debug=False)
@@ -349,7 +357,7 @@ class PostArgParseExceptHookTest(unittest.TestCase):
mock_logger, output = self._test_common(exc_type, debug=False)
mock_logger.error.assert_called_once_with('Exiting due to user request.')
def _test_common(self, error_type, debug):
def _test_common(self, error_type, debug, quiet=False):
"""Returns the mocked logger and stderr output."""
mock_err = io.StringIO()
@@ -366,7 +374,7 @@ class PostArgParseExceptHookTest(unittest.TestCase):
with mock.patch('certbot._internal.log.sys.stderr', mock_err):
try:
self._call(
*exc_info, debug=debug, log_path=self.log_path)
*exc_info, debug=debug, quiet=quiet, log_path=self.log_path)
except SystemExit as exit_err:
mock_err.write(str(exit_err))
else: # pragma: no cover
@@ -385,7 +393,7 @@ class PostArgParseExceptHookTest(unittest.TestCase):
self.assertEqual(actual_exc_info, expected_exc_info)
def _assert_logfile_output(self, output):
self.assertIn('Please see the logfile', output)
self.assertIn('See the logfile', output)
self.assertIn(self.log_path, output)
def _assert_quiet_output(self, mock_logger, output):
@@ -394,12 +402,12 @@ class PostArgParseExceptHookTest(unittest.TestCase):
self.assertIn(self.error_msg, output)
class ExitWithLogPathTest(test_util.TempDirTestCase):
"""Tests for certbot._internal.log.exit_with_log_path."""
class ExitWithAdviceTest(test_util.TempDirTestCase):
"""Tests for certbot._internal.log.exit_with_advice."""
@classmethod
def _call(cls, *args, **kwargs):
from certbot._internal.log import exit_with_log_path
return exit_with_log_path(*args, **kwargs)
from certbot._internal.log import exit_with_advice
return exit_with_advice(*args, **kwargs)
def test_log_file(self):
log_file = os.path.join(self.tempdir, 'test.log')
+159 -31
View File
@@ -1004,21 +1004,26 @@ class MainTest(test_util.ConfigTestCase):
args += '-d foo.bar -a standalone certonly'.split()
self._call(args)
@mock.patch('certbot._internal.main._report_new_cert')
@test_util.patch_get_utility()
def test_certonly_dry_run_new_request_success(self, mock_get_utility):
def test_certonly_dry_run_new_request_success(self, mock_get_utility, mock_report):
mock_client = mock.MagicMock()
mock_client.obtain_and_enroll_certificate.return_value = None
self._certonly_new_request_common(mock_client, ['--dry-run'])
self.assertEqual(
mock_client.obtain_and_enroll_certificate.call_count, 1)
self.assertIn('dry run', mock_get_utility().add_message.call_args[0][0])
self.assertEqual(mock_report.call_count, 1)
self.assertIs(mock_report.call_args[0][0].dry_run, True)
# Asserts we don't suggest donating after a successful dry run
self.assertEqual(mock_get_utility().add_message.call_count, 1)
self.assertEqual(mock_get_utility().add_message.call_count, 0)
@mock.patch('certbot._internal.main._report_new_cert')
@mock.patch('certbot._internal.main.util.atexit_register')
@mock.patch('certbot._internal.eff.handle_subscription')
@mock.patch('certbot.crypto_util.notAfter')
@test_util.patch_get_utility()
def test_certonly_new_request_success(self, mock_get_utility, mock_notAfter, mock_subscription):
def test_certonly_new_request_success(self, unused_mock_get_utility, mock_notAfter,
mock_subscription, mock_register, mock_report):
cert_path = os.path.normpath(os.path.join(self.config.config_dir, 'live/foo.bar'))
key_path = os.path.normpath(os.path.join(self.config.config_dir, 'live/baz.qux'))
date = '1970-01-01'
@@ -1031,11 +1036,10 @@ class MainTest(test_util.ConfigTestCase):
self._certonly_new_request_common(mock_client)
self.assertEqual(
mock_client.obtain_and_enroll_certificate.call_count, 1)
cert_msg = mock_get_utility().add_message.call_args_list[0][0][0]
self.assertIn(cert_path, cert_msg)
self.assertIn(date, cert_msg)
self.assertIn(key_path, cert_msg)
self.assertIn('donate', mock_get_utility().add_message.call_args[0][0])
self.assertEqual(mock_report.call_count, 1)
self.assertIn(cert_path, mock_report.call_args[0][2])
self.assertIn(key_path, mock_report.call_args[0][3])
self.assertIn('donate', mock_register.call_args[0][1])
self.assertIs(mock_subscription.called, True)
@mock.patch('certbot._internal.eff.handle_subscription')
@@ -1119,31 +1123,33 @@ class MainTest(test_util.ConfigTestCase):
return mock_lineage, mock_get_utility, stdout
@mock.patch('certbot._internal.main._report_new_cert')
@mock.patch('certbot._internal.main.util.atexit_register')
@mock.patch('certbot.crypto_util.notAfter')
def test_certonly_renewal(self, _):
lineage, get_utility, _ = self._test_renewal_common(True, [])
def test_certonly_renewal(self, _, mock_register, mock_report):
lineage, _, _ = self._test_renewal_common(True, [])
self.assertEqual(lineage.save_successor.call_count, 1)
lineage.update_all_links_to.assert_called_once_with(
lineage.latest_common_version())
cert_msg = get_utility().add_message.call_args_list[0][0][0]
self.assertIn('fullchain.pem', cert_msg)
self.assertIn('donate', get_utility().add_message.call_args[0][0])
self.assertEqual(mock_report.call_count, 1)
self.assertIn('fullchain.pem', mock_report.call_args[0][2])
self.assertIn('donate', mock_register.call_args[0][1])
@mock.patch('certbot._internal.main.display_util.notify')
@mock.patch('certbot._internal.log.logging.handlers.RotatingFileHandler.doRollover')
@mock.patch('certbot.crypto_util.notAfter')
def test_certonly_renewal_triggers(self, _, __):
def test_certonly_renewal_triggers(self, _, __, mock_notify):
# --dry-run should force renewal
_, get_utility, _ = self._test_renewal_common(False, ['--dry-run', '--keep'],
_, _, _ = self._test_renewal_common(False, ['--dry-run', '--keep'],
log_out="simulating renewal")
self.assertEqual(get_utility().add_message.call_count, 1)
self.assertIn('dry run', get_utility().add_message.call_args[0][0])
mock_notify.assert_any_call('The dry run was successful.')
self._test_renewal_common(False, ['--renew-by-default', '-tvv', '--debug'],
log_out="Auto-renewal forced")
self.assertEqual(get_utility().add_message.call_count, 1)
self._test_renewal_common(False, ['-tvv', '--debug', '--keep'],
log_out="not yet due", should_renew=False)
_, get_utility, _ = self._test_renewal_common(False, ['-tvv', '--debug', '--keep'],
should_renew=False)
self.assertIn('not yet due', get_utility().notification.call_args[0][0])
def _dump_log(self):
print("Logs:")
@@ -1400,19 +1406,23 @@ class MainTest(test_util.ConfigTestCase):
return mock_get_utility
@mock.patch('certbot._internal.main._csr_report_new_cert')
@mock.patch('certbot._internal.main.util.atexit_register')
@mock.patch('certbot._internal.eff.handle_subscription')
def test_certonly_csr(self, mock_subscription):
mock_get_utility = self._test_certonly_csr_common()
cert_msg = mock_get_utility().add_message.call_args_list[0][0][0]
self.assertIn('fullchain.pem', cert_msg)
self.assertNotIn('Your key file has been saved at', cert_msg)
self.assertIn('donate', mock_get_utility().add_message.call_args[0][0])
def test_certonly_csr(self, mock_subscription, mock_register, mock_csr_report):
_ = self._test_certonly_csr_common()
self.assertEqual(mock_csr_report.call_count, 1)
self.assertIn('cert_512.pem', mock_csr_report.call_args[0][1])
self.assertIsNone(mock_csr_report.call_args[0][2])
self.assertIn('fullchain.pem', mock_csr_report.call_args[0][3])
self.assertIn('donate', mock_register.call_args[0][1])
self.assertIs(mock_subscription.called, True)
def test_certonly_csr_dry_run(self):
mock_get_utility = self._test_certonly_csr_common(['--dry-run'])
self.assertEqual(mock_get_utility().add_message.call_count, 1)
self.assertIn('dry run', mock_get_utility().add_message.call_args[0][0])
@mock.patch('certbot._internal.main._csr_report_new_cert')
def test_certonly_csr_dry_run(self, mock_csr_report):
_ = self._test_certonly_csr_common(['--dry-run'])
self.assertEqual(mock_csr_report.call_count, 1)
self.assertIs(mock_csr_report.call_args[0][0].dry_run, True)
@mock.patch('certbot._internal.main._delete_if_appropriate')
@mock.patch('certbot._internal.main.client.acme_client')
@@ -1482,6 +1492,24 @@ class MainTest(test_util.ConfigTestCase):
# without installer
self.assertIs(mock_run.called, False)
@mock.patch('certbot._internal.main.updater.run_renewal_deployer')
@mock.patch('certbot._internal.plugins.selection.choose_configurator_plugins')
@mock.patch('certbot._internal.main._init_le_client')
@mock.patch('certbot._internal.main._get_and_save_cert')
def test_renew_doesnt_restart_on_dryrun(self, mock_get_cert, mock_init, mock_choose,
mock_run_renewal_deployer):
"""A dry-run renewal shouldn't try to restart the installer"""
self.config.dry_run = True
installer = mock.MagicMock()
mock_choose.return_value = (installer, mock.MagicMock())
main.renew_cert(self.config, None, None)
self.assertEqual(mock_init.call_count, 1)
self.assertEqual(mock_get_cert.call_count, 1)
installer.restart.assert_not_called()
mock_run_renewal_deployer.assert_not_called()
class UnregisterTest(unittest.TestCase):
def setUp(self):
@@ -1741,6 +1769,106 @@ class InstallTest(test_util.ConfigTestCase):
self.config, plugins)
class ReportNewCertTest(unittest.TestCase):
"""Tests for certbot._internal.main._report_new_cert and
certbot._internal.main._csr_report_new_cert.
"""
def setUp(self):
from datetime import datetime
self.notify_patch = mock.patch('certbot._internal.main.display_util.notify')
self.mock_notify = self.notify_patch.start()
self.notafter_patch = mock.patch('certbot._internal.main.crypto_util.notAfter')
self.mock_notafter = self.notafter_patch.start()
self.mock_notafter.return_value = datetime.utcfromtimestamp(0)
def tearDown(self):
self.notify_patch.stop()
self.notafter_patch.stop()
@classmethod
def _call(cls, *args, **kwargs):
from certbot._internal.main import _report_new_cert
return _report_new_cert(*args, **kwargs)
@classmethod
def _call_csr(cls, *args, **kwargs):
from certbot._internal.main import _csr_report_new_cert
return _csr_report_new_cert(*args, **kwargs)
def test_report_dry_run(self):
self._call(mock.Mock(dry_run=True), None, None, None)
self.mock_notify.assert_called_with("The dry run was successful.")
def test_csr_report_dry_run(self):
self._call_csr(mock.Mock(dry_run=True), None, None, None)
self.mock_notify.assert_called_with("The dry run was successful.")
def test_report_no_paths(self):
with self.assertRaises(AssertionError):
self._call(mock.Mock(dry_run=False), None, None, None)
with self.assertRaises(AssertionError):
self._call_csr(mock.Mock(dry_run=False), None, None, None)
def test_report(self):
self._call(mock.Mock(dry_run=False),
'/path/to/cert.pem', '/path/to/fullchain.pem',
'/path/to/privkey.pem')
self.mock_notify.assert_called_with(
'\nSuccessfully received certificate.\n'
'Certificate is saved at: /path/to/fullchain.pem\n'
'Key is saved at: /path/to/privkey.pem\n'
'This certificate expires on 1970-01-01.\n'
'These files will be updated when the certificate renews.\n'
'Certbot will automatically renew this certificate in the background.'
)
def test_report_no_key(self):
self._call(mock.Mock(dry_run=False),
'/path/to/cert.pem', '/path/to/fullchain.pem',
None)
self.mock_notify.assert_called_with(
'\nSuccessfully received certificate.\n'
'Certificate is saved at: /path/to/fullchain.pem\n'
'This certificate expires on 1970-01-01.\n'
'These files will be updated when the certificate renews.\n'
'Certbot will automatically renew this certificate in the background.'
)
def test_report_no_preconfigured_renewal(self):
self._call(mock.Mock(dry_run=False, preconfigured_renewal=False),
'/path/to/cert.pem', '/path/to/fullchain.pem',
'/path/to/privkey.pem')
self.mock_notify.assert_called_with(
'\nSuccessfully received certificate.\n'
'Certificate is saved at: /path/to/fullchain.pem\n'
'Key is saved at: /path/to/privkey.pem\n'
'This certificate expires on 1970-01-01.\n'
'These files will be updated when the certificate renews.\n'
'Run "certbot renew" to renew expiring certificates. We recommend setting up a '
'scheduled task for renewal; see https://certbot.eff.org/docs/using.html#automated'
'-renewals for instructions.'
)
def test_csr_report(self):
self._call_csr(mock.Mock(dry_run=False), '/path/to/cert.pem',
'/path/to/chain.pem', '/path/to/fullchain.pem')
self.mock_notify.assert_called_with(
'\nSuccessfully received certificate.\n'
'Certificate is saved at: /path/to/cert.pem\n'
'Intermediate CA chain is saved at: /path/to/chain.pem\n'
'Full certificate chain is saved at: /path/to/fullchain.pem\n'
'This certificate expires on 1970-01-01.'
)
class UpdateAccountTest(test_util.ConfigTestCase):
"""Tests for certbot._internal.main.update_account"""
+7
View File
@@ -83,6 +83,13 @@ class PluginTest(unittest.TestCase):
parser.add_argument.assert_called_once_with(
"--mock-foo-bar", dest="different_to_foo_bar", x=1, y=None)
def test_fallback_auth_hint(self):
self.assertIn("the mock plugin completed the required dns-01 challenges",
self.plugin.auth_hint([acme_util.DNS01_A, acme_util.DNS01_A]))
self.assertIn("the mock plugin completed the required dns-01 and http-01 challenges",
self.plugin.auth_hint([acme_util.DNS01_A, acme_util.HTTP01_A,
acme_util.DNS01_A]))
class InstallerTest(test_util.ConfigTestCase):
"""Tests for certbot.plugins.common.Installer."""
+8 -1
View File
@@ -42,7 +42,8 @@ class DNSAuthenticatorTest(test_util.TempDirTestCase, dns_test_common.BaseAuthen
self.auth = DNSAuthenticatorTest._FakeDNSAuthenticator(self.config, "fake")
def test_perform(self):
@test_util.patch_get_utility()
def test_perform(self, unused_mock_get_utility):
self.auth.perform([self.achall])
self.auth._perform.assert_called_once_with(dns_test_common.DOMAIN, mock.ANY, mock.ANY)
@@ -119,6 +120,12 @@ class DNSAuthenticatorTest(test_util.TempDirTestCase, dns_test_common.BaseAuthen
credentials = self.auth._configure_credentials("credentials", "", {"test": ""})
self.assertEqual(credentials.conf("test"), "value")
def test_auth_hint(self):
self.assertIn(
'try increasing --fake-propagation-seconds (currently 0 seconds).',
self.auth.auth_hint([mock.MagicMock()])
)
class CredentialsConfigurationTest(test_util.TempDirTestCase):
class _MockLoggingHandler(logging.Handler):
+44 -3
View File
@@ -1,5 +1,6 @@
"""Tests for certbot._internal.plugins.manual"""
import sys
import textwrap
import unittest
try:
@@ -20,6 +21,10 @@ class AuthenticatorTest(test_util.TempDirTestCase):
def setUp(self):
super().setUp()
get_utility_patch = test_util.patch_get_utility()
self.mock_get_utility = get_utility_patch.start()
self.addCleanup(get_utility_patch.stop)
self.http_achall = acme_util.HTTP01_A
self.dns_achall = acme_util.DNS01_A
self.dns_achall_2 = acme_util.DNS01_A_2
@@ -89,12 +94,19 @@ class AuthenticatorTest(test_util.TempDirTestCase):
self.auth.env[self.http_achall]['CERTBOT_AUTH_OUTPUT'],
http_expected)
@test_util.patch_get_utility()
def test_manual_perform(self, mock_get_utility):
# Successful hook output should be sent to notify
self.assertEqual(self.mock_get_utility().notification.call_count, len(self.achalls))
for i, (args, _) in enumerate(self.mock_get_utility().notification.call_args_list):
needle = textwrap.indent(self.auth.env[self.achalls[i]]['CERTBOT_AUTH_OUTPUT'], ' ')
self.assertIn(needle, args[0])
def test_manual_perform(self):
self.assertEqual(
self.auth.perform(self.achalls),
[achall.response(achall.account_key) for achall in self.achalls])
for i, (args, kwargs) in enumerate(mock_get_utility().notification.call_args_list):
self.assertEqual(self.mock_get_utility().notification.call_count, len(self.achalls))
for i, (args, kwargs) in enumerate(self.mock_get_utility().notification.call_args_list):
achall = self.achalls[i]
self.assertIn(achall.validation(achall.account_key), args[0])
self.assertIs(kwargs['wrap'], False)
@@ -120,6 +132,35 @@ class AuthenticatorTest(test_util.TempDirTestCase):
else:
self.assertNotIn('CERTBOT_TOKEN', os.environ)
def test_auth_hint_hook(self):
self.config.manual_auth_hook = '/bin/true'
self.assertEqual(
self.auth.auth_hint([acme_util.DNS01_A, acme_util.HTTP01_A]),
'The Certificate Authority failed to verify the DNS TXT records and challenge '
'files created by the --manual-auth-hook. Ensure that this hook is functioning '
'correctly and that it waits a sufficient duration of time for DNS propagation. '
'Refer to "certbot --help manual" and the Certbot User Guide.'
)
self.assertEqual(
self.auth.auth_hint([acme_util.HTTP01_A]),
'The Certificate Authority failed to verify the challenge files created by the '
'--manual-auth-hook. Ensure that this hook is functioning correctly. Refer to '
'"certbot --help manual" and the Certbot User Guide.'
)
def test_auth_hint_no_hook(self):
self.assertEqual(
self.auth.auth_hint([acme_util.DNS01_A, acme_util.HTTP01_A]),
'The Certificate Authority failed to verify the manually created DNS TXT records '
'and challenge files. Ensure that you created these in the correct location, or '
'try waiting longer for DNS propagation on the next attempt.'
)
self.assertEqual(
self.auth.auth_hint([acme_util.HTTP01_A, acme_util.HTTP01_A, acme_util.HTTP01_A]),
'The Certificate Authority failed to verify the manually created challenge files. '
'Ensure that you created these in the correct location.'
)
if __name__ == '__main__':
unittest.main() # pragma: no cover
+4 -4
View File
@@ -346,19 +346,19 @@ class AddDeprecatedArgumentTest(unittest.TestCase):
def test_warning_no_arg(self):
self._call("--old-option", 0)
with mock.patch("certbot.util.logger.warning") as mock_warn:
with mock.patch("warnings.warn") as mock_warn:
self.parser.parse_args(["--old-option"])
self.assertEqual(mock_warn.call_count, 1)
self.assertIn("is deprecated", mock_warn.call_args[0][0])
self.assertEqual("--old-option", mock_warn.call_args[0][1])
self.assertIn("--old-option", mock_warn.call_args[0][0])
def test_warning_with_arg(self):
self._call("--old-option", 1)
with mock.patch("certbot.util.logger.warning") as mock_warn:
with mock.patch("warnings.warn") as mock_warn:
self.parser.parse_args(["--old-option", "42"])
self.assertEqual(mock_warn.call_count, 1)
self.assertIn("is deprecated", mock_warn.call_args[0][0])
self.assertEqual("--old-option", mock_warn.call_args[0][1])
self.assertIn("--old-option", mock_warn.call_args[0][0])
def test_help(self):
self._call("--old-option", 2)