Merge branch 'cli_new_cert_reporting' into ux

This commit is contained in:
Alex Zorin 2021-05-18 14:56:58 +10:00
commit 812b34c1e8
7 changed files with 269 additions and 86 deletions

View file

@ -506,8 +506,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)
@ -761,5 +759,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)

View file

@ -73,21 +73,6 @@ def _suggest_donation_if_appropriate(config):
"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)
def _get_and_save_cert(le_client, config, domains=None, certname=None, lineage=None):
"""Authenticate and enroll certificate.
@ -482,8 +467,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 +487,70 @@ 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, cert_path, chain_path, fullchain_path):
# type: (interfaces.IConfig, Optional[str], Optional[str], Optional[str]) -> None
""" --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 cert_path: path to cert.pem
:type cert_path: str
:param chain_path: path to chain.pem
:type chain_path: str
:param fullchain_path: path to fullchain.pem
:type fullchain_path: str
"""
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):
@ -1190,6 +1220,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 +1233,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):
@ -1281,8 +1321,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

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
@ -437,6 +438,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.

View file

@ -240,25 +240,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):

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"""

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."""

View file

@ -6,6 +6,7 @@ from importlib import reload as reload_module
import io
import itertools
import json
from operator import is_
import shutil
import sys
import tempfile
@ -1004,21 +1005,25 @@ 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.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, mock_get_utility, mock_notAfter,
mock_subscription, 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,11 @@ 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_get_utility().add_message.call_args[0][0])
self.assertIs(mock_subscription.called, True)
@mock.patch('certbot._internal.eff.handle_subscription')
@ -1123,28 +1128,28 @@ class MainTest(test_util.ConfigTestCase):
return mock_lineage, mock_get_utility, stdout
@mock.patch('certbot._internal.main._report_new_cert')
@mock.patch('certbot.crypto_util.notAfter')
def test_certonly_renewal(self, _):
def test_certonly_renewal(self, _, mock_report):
lineage, get_utility, _ = 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.assertEqual(mock_report.call_count, 1)
self.assertIn('fullchain.pem', mock_report.call_args[0][2])
self.assertIn('donate', get_utility().add_message.call_args[0][0])
@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'],
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)
@ -1404,19 +1409,22 @@ class MainTest(test_util.ConfigTestCase):
return mock_get_utility
@mock.patch('certbot._internal.main._csr_report_new_cert')
@mock.patch('certbot._internal.eff.handle_subscription')
def test_certonly_csr(self, mock_subscription):
def test_certonly_csr(self, mock_subscription, mock_csr_report):
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.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_get_utility().add_message.call_args[0][0])
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')
@ -1745,6 +1753,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.fromtimestamp(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"""