Defines the RenewableCert API (#7603)

This is my proposed fix for #7540. I would ideally like this to be included in our 1.0 release.

I came up with this design by adding all attributes used either in our own plugins, 3rd party plugins listed at https://certbot.eff.org/docs/using.html#third-party-plugins, or our public API code.

Despite me thinking that zope is unneeded nowadays, I initially tried to use it to define this interface since we have it and it gives us a way to define expected attributes, but it doesn't work because zope interface objects also have a method called `names` which conflict with the API.

I talked about this with Adrien out of band and did some of my own research and there are some minor benefits with this new approach of using properties:

1. It's more conventional.
2. If you also change the implementation to inherit from the class, Python will error if all properties aren't defined.
3. The PEP 526 style type annotations with mypy seem to (currently) only be used to validate code using the class, not the class implementation itself. You can add a type annotation saying the class needs to have this attribute, never define it, and mypy won't complain.

With this new approach, I had to fix `names` because pylint was complaining that the arguments differed, however, we never used the optional parameter to `names` outside of tests so I just deleted the code altogether.

* fixes #7540

* move to properties
This commit is contained in:
Brad Warren
2019-11-27 11:32:00 -08:00
committed by GitHub
parent 6c1dfe43c7
commit 84b770b56e
6 changed files with 101 additions and 52 deletions
+14 -10
View File
@@ -17,6 +17,7 @@ from certbot._internal import constants
from certbot import crypto_util
from certbot._internal import error_handler
from certbot import errors
from certbot import interfaces
from certbot import util
from certbot.compat import os
from certbot.compat import filesystem
@@ -376,7 +377,7 @@ def delete_files(config, certname):
logger.debug("Unable to remove %s", archive_path)
class RenewableCert(object):
class RenewableCert(interfaces.RenewableCert):
"""Renewable certificate.
Represents a lineage of certificates that is under the management of
@@ -423,7 +424,7 @@ class RenewableCert(object):
"""
self.cli_config = cli_config
self.lineagename = lineagename_for_filename(config_filename)
self._lineagename = lineagename_for_filename(config_filename)
# self.configuration should be used to read parameters that
# may have been chosen based on default values from the
@@ -483,6 +484,15 @@ class RenewableCert(object):
"""Duck type for self.fullchain"""
return self.fullchain
@property
def lineagename(self):
"""Name given to the certificate lineage.
:rtype: str
"""
return self._lineagename
@property
def target_expiry(self):
"""The current target certificate's expiration datetime
@@ -858,21 +868,15 @@ class RenewableCert(object):
for _, link in previous_links:
os.unlink(link)
def names(self, version=None):
def names(self):
"""What are the subject names of this certificate?
(If no version is specified, use the current version.)
:param int version: the desired version number
:returns: the subject names
:rtype: `list` of `str`
:raises .CertStorageError: if could not find cert file.
"""
if version is None:
target = self.current_target("cert")
else:
target = self.version("cert", version)
target = self.current_target("cert")
if target is None:
raise errors.CertStorageError("could not find cert file")
with open(target) as f:
+14 -11
View File
@@ -213,26 +213,28 @@ def verify_renewable_cert(renewable_cert):
2. That fullchain matches cert and chain when concatenated.
3. Check that the private key matches the certificate.
:param `.storage.RenewableCert` renewable_cert: cert to verify
:param renewable_cert: cert to verify
:type renewable_cert: certbot.interfaces.RenewableCert
:raises errors.Error: If verification fails.
"""
verify_renewable_cert_sig(renewable_cert)
verify_fullchain(renewable_cert)
verify_cert_matches_priv_key(renewable_cert.cert, renewable_cert.privkey)
verify_cert_matches_priv_key(renewable_cert.cert_path, renewable_cert.key_path)
def verify_renewable_cert_sig(renewable_cert):
"""Verifies the signature of a `.storage.RenewableCert` object.
"""Verifies the signature of a RenewableCert object.
:param `.storage.RenewableCert` renewable_cert: cert to verify
:param renewable_cert: cert to verify
:type renewable_cert: certbot.interfaces.RenewableCert
:raises errors.Error: If signature verification fails.
"""
try:
with open(renewable_cert.chain, 'rb') as chain_file: # type: IO[bytes]
with open(renewable_cert.chain_path, 'rb') as chain_file: # type: IO[bytes]
chain = x509.load_pem_x509_certificate(chain_file.read(), default_backend())
with open(renewable_cert.cert, 'rb') as cert_file: # type: IO[bytes]
with open(renewable_cert.cert_path, 'rb') as cert_file: # type: IO[bytes]
cert = x509.load_pem_x509_certificate(cert_file.read(), default_backend())
pk = chain.public_key()
with warnings.catch_warnings():
@@ -240,7 +242,7 @@ def verify_renewable_cert_sig(renewable_cert):
cert.signature_hash_algorithm)
except (IOError, ValueError, InvalidSignature) as e:
error_str = "verifying the signature of the cert located at {0} has failed. \
Details: {1}".format(renewable_cert.cert, e)
Details: {1}".format(renewable_cert.cert_path, e)
logger.exception(error_str)
raise errors.Error(error_str)
@@ -301,16 +303,17 @@ def verify_cert_matches_priv_key(cert_path, key_path):
def verify_fullchain(renewable_cert):
""" Verifies that fullchain is indeed cert concatenated with chain.
:param `.storage.RenewableCert` renewable_cert: cert to verify
:param renewable_cert: cert to verify
:type renewable_cert: certbot.interfaces.RenewableCert
:raises errors.Error: If cert and chain do not combine to fullchain.
"""
try:
with open(renewable_cert.chain) as chain_file: # type: IO[str]
with open(renewable_cert.chain_path) as chain_file: # type: IO[str]
chain = chain_file.read()
with open(renewable_cert.cert) as cert_file: # type: IO[str]
with open(renewable_cert.cert_path) as cert_file: # type: IO[str]
cert = cert_file.read()
with open(renewable_cert.fullchain) as fullchain_file: # type: IO[str]
with open(renewable_cert.fullchain_path) as fullchain_file: # type: IO[str]
fullchain = fullchain_file.read()
if (cert + chain) != fullchain:
error_str = "fullchain does not match cert + chain for {0}!"
+58 -2
View File
@@ -532,6 +532,62 @@ class IReporter(zope.interface.Interface):
"""Prints messages to the user and clears the message queue."""
@six.add_metaclass(abc.ABCMeta)
class RenewableCert(object):
"""Interface to a certificate lineage."""
@abc.abstractproperty
def cert_path(self):
"""Path to the certificate file.
:rtype: str
"""
@abc.abstractproperty
def key_path(self):
"""Path to the private key file.
:rtype: str
"""
@abc.abstractproperty
def chain_path(self):
"""Path to the certificate chain file.
:rtype: str
"""
@abc.abstractproperty
def fullchain_path(self):
"""Path to the full chain file.
The full chain is the certificate file plus the chain file.
:rtype: str
"""
@abc.abstractproperty
def lineagename(self):
"""Name given to the certificate lineage.
:rtype: str
"""
@abc.abstractmethod
def names(self):
"""What are the subject names of this certificate?
:returns: the subject names
:rtype: `list` of `str`
:raises .CertStorageError: if could not find cert file.
"""
# Updater interfaces
#
# When "certbot renew" is run, Certbot will iterate over each lineage and check
@@ -570,7 +626,7 @@ class GenericUpdater(object):
This method is called once for each lineage.
:param lineage: Certificate lineage object
:type lineage: storage.RenewableCert
:type lineage: RenewableCert
"""
@@ -599,6 +655,6 @@ class RenewDeployer(object):
This method is called once for each lineage renewed
:param lineage: Certificate lineage object
:type lineage: storage.RenewableCert
:type lineage: RenewableCert
"""
+4 -4
View File
@@ -62,7 +62,7 @@ def enable(lineage, domains, installer, config):
Run enable method for each requested enhancement that is supported.
:param lineage: Certificate lineage object
:type lineage: certbot._internal.storage.RenewableCert
:type lineage: certbot.interfaces.RenewableCert
:param domains: List of domains in certificate to enhance
:type domains: str
@@ -123,7 +123,7 @@ class AutoHSTSEnhancement(object):
Implementation of this method should increase the max-age value.
:param lineage: Certificate lineage object
:type lineage: certbot._internal.storage.RenewableCert
:type lineage: certbot.interfaces.RenewableCert
.. note:: prepare() method inherited from `interfaces.IPlugin` might need
to be called manually within implementation of this interface method
@@ -137,7 +137,7 @@ class AutoHSTSEnhancement(object):
Long max-age value should be set in implementation of this method.
:param lineage: Certificate lineage object
:type lineage: certbot._internal.storage.RenewableCert
:type lineage: certbot.interfaces.RenewableCert
"""
@abc.abstractmethod
@@ -148,7 +148,7 @@ class AutoHSTSEnhancement(object):
over the subsequent runs of Certbot renew.
:param lineage: Certificate lineage object
:type lineage: certbot._internal.storage.RenewableCert
:type lineage: certbot.interfaces.RenewableCert
:param domains: List of domains in certificate to enhance
:type domains: str
+11 -11
View File
@@ -181,15 +181,15 @@ class VerifyCertSetup(unittest.TestCase):
super(VerifyCertSetup, self).setUp()
self.renewable_cert = mock.MagicMock()
self.renewable_cert.cert = SS_CERT_PATH
self.renewable_cert.chain = SS_CERT_PATH
self.renewable_cert.privkey = RSA2048_KEY_PATH
self.renewable_cert.fullchain = test_util.vector_path('cert_fullchain_2048.pem')
self.renewable_cert.cert_path = SS_CERT_PATH
self.renewable_cert.chain_path = SS_CERT_PATH
self.renewable_cert.key_path = RSA2048_KEY_PATH
self.renewable_cert.fullchain_path = test_util.vector_path('cert_fullchain_2048.pem')
self.bad_renewable_cert = mock.MagicMock()
self.bad_renewable_cert.chain = SS_CERT_PATH
self.bad_renewable_cert.cert = SS_CERT_PATH
self.bad_renewable_cert.fullchain = SS_CERT_PATH
self.bad_renewable_cert.chain_path = SS_CERT_PATH
self.bad_renewable_cert.cert_path = SS_CERT_PATH
self.bad_renewable_cert.fullchain_path = SS_CERT_PATH
class VerifyRenewableCertTest(VerifyCertSetup):
@@ -219,13 +219,13 @@ class VerifyRenewableCertSigTest(VerifyCertSetup):
def test_cert_sig_match_ec(self):
renewable_cert = mock.MagicMock()
renewable_cert.cert = P256_CERT_PATH
renewable_cert.chain = P256_CERT_PATH
renewable_cert.privkey = P256_KEY
renewable_cert.cert_path = P256_CERT_PATH
renewable_cert.chain_path = P256_CERT_PATH
renewable_cert.key_path = P256_KEY
self.assertEqual(None, self._call(renewable_cert))
def test_cert_sig_mismatch(self):
self.bad_renewable_cert.cert = test_util.vector_path('cert_512_bad.pem')
self.bad_renewable_cert.cert_path = test_util.vector_path('cert_512_bad.pem')
self.assertRaises(errors.Error, self._call, self.bad_renewable_cert)
-14
View File
@@ -404,20 +404,6 @@ class RenewableCertTests(BaseRenewableCertTest):
self.assertEqual(self.test_rc.names(),
["example.com", "www.example.com"])
# Trying a non-current version
self._write_out_kind("cert", 15, test_util.load_vector("cert_512.pem"))
self.assertEqual(self.test_rc.names(12),
["example.com", "www.example.com"])
# Testing common name is listed first
self._write_out_kind(
"cert", 12, test_util.load_vector("cert-5sans_512.pem"))
self.assertEqual(
self.test_rc.names(12),
["example.com"] + ["{0}.example.com".format(c) for c in "abcd"])
# Trying missing cert
os.unlink(self.test_rc.cert)
self.assertRaises(errors.CertStorageError, self.test_rc.names)