cli: redesign output of new certificate reporting

Changes the output of run, certonly and certonly --csr. No longer uses
IReporter.
This commit is contained in:
Alex Zorin 2020-12-07 10:30:43 +11:00
parent 38017473c5
commit ea21f558c1
5 changed files with 230 additions and 68 deletions

View file

@ -503,8 +503,6 @@ class Client(object):
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)
@ -758,5 +756,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

@ -67,21 +67,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.
@ -479,8 +464,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
@ -495,29 +484,65 @@ 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."
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 cert 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{key_msg}{symlink_msg}"
"This certificate expires on {expiry}.").format(
cert_path=fullchain_path,
expiry=expiry,
key_msg="Key is saved at: {}\n".format(key_path) if key_path else "",
symlink_msg="These files will be automatically updated "
"every time the certificate is renewed.\n",
)
)
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()
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):
@ -1177,6 +1202,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
@ -1189,20 +1215,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):
@ -1276,8 +1311,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

@ -26,7 +26,7 @@ import six
import zope.component
from acme import crypto_util as acme_crypto_util
from acme.magic_typing import IO # pylint: disable=unused-import
from acme.magic_typing import IO, Any, List # pylint: disable=unused-import
from certbot import errors
from certbot import interfaces
from certbot import util
@ -437,6 +437,19 @@ def get_names_from_cert(csr, typ=crypto.FILETYPE_PEM):
csr, crypto.load_certificate, typ)
def get_names_from_req(csr, typ=crypto.FILETYPE_PEM):
# type: (str, Any) -> 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

@ -379,6 +379,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

@ -6,6 +6,7 @@ from __future__ import print_function
import datetime
import itertools
import json
from operator import is_
import shutil
import sys
import tempfile
@ -971,22 +972,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.assertTrue(
'dry run' in mock_get_utility().add_message.call_args[0][0])
self.assertEqual(mock_report.call_count, 1)
self.assertEqual(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'
@ -999,10 +1004,9 @@ 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.assertTrue(cert_path in cert_msg)
self.assertTrue(date in cert_msg)
self.assertTrue(key_path in cert_msg)
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.assertTrue(
'donate' in mock_get_utility().add_message.call_args[0][0])
self.assertTrue(mock_subscription.called)
@ -1092,28 +1096,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.assertTrue('fullchain.pem' in cert_msg)
self.assertEqual(mock_report.call_count, 1)
self.assertIn('fullchain.pem', mock_report.call_args[0][2])
self.assertTrue('donate' in 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.assertTrue('dry run' in 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)
@ -1373,21 +1377,23 @@ 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.assertTrue('fullchain.pem' in cert_msg)
self.assertFalse('Your key file has been saved at' in 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.assertTrue(
'donate' in mock_get_utility().add_message.call_args[0][0])
self.assertTrue(mock_subscription.called)
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.assertTrue(
'dry run' in 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.assertEqual(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')
@ -1795,5 +1801,86 @@ 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'
'These files will be automatically updated every time the certificate is renewed.\n'
'This certificate expires on 1970-01-01.'
)
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'
'These files will be automatically updated every time the certificate is renewed.\n'
'This certificate expires on 1970-01-01.'
)
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.'
)
if __name__ == '__main__':
unittest.main() # pragma: no cover