upstream/certbot
1
0
Fork 0
mirror of https://github.com/certbot/certbot.git synced 2026-05-28 04:34:11 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Merge pull request #2361 from letsencrypt/update-successor-config

Browse source 
Update config in save_successor
This commit is contained in:
schoen 2016-02-04 12:56:34 -08:00
commit fd1a503aee
4 changed files with 115 additions and 74 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
letsencrypt/cli.py
View file

@ -428,8 +428,8 @@ def _auth_from_domains(le_client, config, domains):
lineage.save_successor(
lineage.latest_common_version(), OpenSSL.crypto.dump_certificate(
OpenSSL.crypto.FILETYPE_PEM, new_certr.body.wrapped),
new_key.pem, crypto_util.dump_pyopenssl_chain(new_chain))
new_key.pem, crypto_util.dump_pyopenssl_chain(new_chain),
configuration.RenewerConfiguration(config.namespace))
lineage.update_all_links_to(lineage.latest_common_version())
# TODO: Check return value of save_successor
# TODO: Also update lineage renewal config with any relevant

18
letsencrypt/client.py
View file

@ -282,23 +282,13 @@ class Client(object):
"""
certr, chain, key, _ = self.obtain_certificate(domains)
# XXX: We clearly need a more general and correct way of getting
# options into the configobj for the RenewableCert instance.
# This is a quick-and-dirty way to do it to allow integration
# testing to start. (Note that the config parameter to new_lineage
# ideally should be a ConfigObj, but in this case a dict will be
# accepted in practice.)
params = vars(self.config.namespace)
config = {}
cli_config = configuration.RenewerConfiguration(self.config.namespace)
if (cli_config.config_dir != constants.CLI_DEFAULTS["config_dir"] or
cli_config.work_dir != constants.CLI_DEFAULTS["work_dir"]):
if (self.config.config_dir != constants.CLI_DEFAULTS["config_dir"] or
self.config.work_dir != constants.CLI_DEFAULTS["work_dir"]):
logger.warning(
"Non-standard path(s), might not work with crontab installed "
"by your operating system package manager")
if cli_config.dry_run:
if self.config.dry_run:
logger.info("Dry run: Skipping creating new lineage for %s",
domains[0])
return None
@ -307,7 +297,7 @@ class Client(object):
domains[0], OpenSSL.crypto.dump_certificate(
OpenSSL.crypto.FILETYPE_PEM, certr.body.wrapped),
key.pem, crypto_util.dump_pyopenssl_chain(chain),
params, config, cli_config)
configuration.RenewerConfiguration(self.config.namespace))
def save_certificate(self, certr, chain_cert,
cert_path, chain_path, fullchain_path):

110
letsencrypt/storage.py
View file

@ -50,6 +50,68 @@ def add_time_interval(base_time, interval, textparser=parsedatetime.Calendar()):
return textparser.parseDT(interval, base_time, tzinfo=tzinfo)[0]
def write_renewal_config(filename, target, cli_config):
"""Writes a renewal config file with the specified name and values.
:param str filename: Absolute path to the config file
:param dict target: Maps ALL_FOUR to their symlink paths
:param .RenewerConfiguration cli_config: parsed command line
arguments
:returns: Configuration object for the new config file
:rtype: configobj.ConfigObj
"""
# create_empty creates a new config file if filename does not exist
config = configobj.ConfigObj(filename, create_empty=True)
for kind in ALL_FOUR:
config[kind] = target[kind]
# XXX: We clearly need a more general and correct way of getting
# options into the configobj for the RenewableCert instance.
# This is a quick-and-dirty way to do it to allow integration
# testing to start. (Note that the config parameter to new_lineage
# ideally should be a ConfigObj, but in this case a dict will be
# accepted in practice.)
renewalparams = vars(cli_config.namespace)
if renewalparams:
config["renewalparams"] = renewalparams
config.comments["renewalparams"] = ["",
"Options and defaults used"
" in the renewal process"]
# TODO: add human-readable comments explaining other available
# parameters
logger.debug("Writing new config %s.", filename)
config.write()
return config
def update_configuration(lineagename, target, cli_config):
"""Modifies lineagename's config to contain the specified values.
:param str lineagename: Name of the lineage being modified
:param dict target: Maps ALL_FOUR to their symlink paths
:param .RenewerConfiguration cli_config: parsed command line
arguments
:returns: Configuration object for the updated config file
:rtype: configobj.ConfigObj
"""
config_filename = os.path.join(
cli_config.renewal_configs_dir, lineagename) + ".conf"
temp_filename = config_filename + ".new"
# If an existing tempfile exists, delete it
if os.path.exists(temp_filename):
os.unlink(temp_filename)
write_renewal_config(temp_filename, target, cli_config)
os.rename(temp_filename, config_filename)
return configobj.ConfigObj(config_filename)
class RenewableCert(object): # pylint: disable=too-many-instance-attributes
"""Renewable certificate.
@ -589,9 +651,8 @@ class RenewableCert(object): # pylint: disable=too-many-instance-attributes
return False
@classmethod
def new_lineage(cls, lineagename, cert, privkey, chain,
renewalparams=None, config=None, cli_config=None):
# pylint: disable=too-many-locals,too-many-arguments
def new_lineage(cls, lineagename, cert, privkey, chain, cli_config):
# pylint: disable=too-many-locals
"""Create a new certificate lineage.
Attempts to create a certificate lineage -- enrolled for
@ -611,26 +672,13 @@ class RenewableCert(object): # pylint: disable=too-many-instance-attributes
:param str cert: the initial certificate version in PEM format
:param str privkey: the private key in PEM format
:param str chain: the certificate chain in PEM format
:param configobj.ConfigObj renewalparams: parameters that
should be used when instantiating authenticator and installer
objects in the future to attempt to renew this cert or deploy
new versions of it
:param configobj.ConfigObj config: renewal configuration
defaults, affecting, for example, the locations of the
directories where the associated files will be saved
:param .RenewerConfiguration cli_config: parsed command line
arguments
:returns: the newly-created RenewalCert object
:rtype: :class:`storage.renewableCert`"""
config = config_with_defaults(config)
# This attempts to read the renewer config file and augment or replace
# the renewer defaults with any options contained in that file. If
# renewer_config_file is undefined or if the file is nonexistent or
# empty, this .merge() will have no effect.
config.merge(configobj.ConfigObj(cli_config.renewer_config_file))
:rtype: :class:`storage.renewableCert`
"""
# Examine the configuration and find the new lineage's name
for i in (cli_config.renewal_configs_dir, cli_config.archive_dir,
cli_config.live_dir):
@ -685,21 +733,11 @@ class RenewableCert(object): # pylint: disable=too-many-instance-attributes
# Document what we've done in a new renewal config file
config_file.close()
new_config = configobj.ConfigObj(config_filename, create_empty=True)
for kind in ALL_FOUR:
new_config[kind] = target[kind]
if renewalparams:
new_config["renewalparams"] = renewalparams
new_config.comments["renewalparams"] = ["",
"Options and defaults used"
" in the renewal process"]
# TODO: add human-readable comments explaining other available
# parameters
logger.debug("Writing new config %s.", config_filename)
new_config.write()
new_config = write_renewal_config(config_filename, target, cli_config)
return cls(new_config.filename, cli_config)
def save_successor(self, prior_version, new_cert, new_privkey, new_chain):
def save_successor(self, prior_version, new_cert,
new_privkey, new_chain, cli_config):
"""Save new cert and chain as a successor of a prior version.
Returns the new version number that was created.
@ -715,6 +753,8 @@ class RenewableCert(object): # pylint: disable=too-many-instance-attributes
:param str new_privkey: the new private key, in PEM format,
or ``None``, if the private key has not changed
:param str new_chain: the new chain, in PEM format
:param .RenewerConfiguration cli_config: parsed command line
arguments
:returns: the new version number that was created
:rtype: int
@ -726,6 +766,7 @@ class RenewableCert(object): # pylint: disable=too-many-instance-attributes
# if needed (ensuring their permissions are correct)
# Figure out what the new version is and hence where to save things
self.cli_config = cli_config
target_version = self.next_free_version()
archive = self.cli_config.archive_dir
prefix = os.path.join(archive, self.lineagename)
@ -763,4 +804,11 @@ class RenewableCert(object): # pylint: disable=too-many-instance-attributes
with open(target["fullchain"], "w") as f:
logger.debug("Writing full chain to %s.", target["fullchain"])
f.write(new_cert + new_chain)
symlinks = dict((kind, self.configuration[kind]) for kind in ALL_FOUR)
# Update renewal config file
self.configfile = update_configuration(
self.lineagename, symlinks, cli_config)
self.configuration = config_with_defaults(self.configfile)
return target_version

57
letsencrypt/tests/storage_test.py
View file

@ -504,8 +504,9 @@ class RenewableCertTests(BaseRenewableCertTest):
with open(where, "w") as f:
f.write(kind)
self.test_rc.update_all_links_to(3)
self.assertEqual(6, self.test_rc.save_successor(3, "new cert", None,
"new chain"))
self.assertEqual(
6, self.test_rc.save_successor(3, "new cert", None,
"new chain", self.cli_config))
with open(self.test_rc.version("cert", 6)) as f:
self.assertEqual(f.read(), "new cert")
with open(self.test_rc.version("chain", 6)) as f:
@ -516,10 +517,12 @@ class RenewableCertTests(BaseRenewableCertTest):
self.assertFalse(os.path.islink(self.test_rc.version("privkey", 3)))
self.assertTrue(os.path.islink(self.test_rc.version("privkey", 6)))
# Let's try two more updates
self.assertEqual(7, self.test_rc.save_successor(6, "again", None,
"newer chain"))
self.assertEqual(8, self.test_rc.save_successor(7, "hello", None,
"other chain"))
self.assertEqual(
7, self.test_rc.save_successor(6, "again", None,
"newer chain", self.cli_config))
self.assertEqual(
8, self.test_rc.save_successor(7, "hello", None,
"other chain", self.cli_config))
# All of the subsequent versions should link directly to the original
# privkey.
for i in (6, 7, 8):
@ -532,27 +535,33 @@ class RenewableCertTests(BaseRenewableCertTest):
self.assertEqual(self.test_rc.current_version(kind), 3)
# Test updating from latest version rather than old version
self.test_rc.update_all_links_to(8)
self.assertEqual(9, self.test_rc.save_successor(8, "last", None,
"attempt"))
self.assertEqual(
9, self.test_rc.save_successor(8, "last", None,
"attempt", self.cli_config))
for kind in ALL_FOUR:
self.assertEqual(self.test_rc.available_versions(kind),
range(1, 10))
self.assertEqual(self.test_rc.current_version(kind), 8)
with open(self.test_rc.version("fullchain", 9)) as f:
self.assertEqual(f.read(), "last" + "attempt")
temp_config_file = os.path.join(self.cli_config.renewal_configs_dir,
self.test_rc.lineagename) + ".conf.new"
with open(temp_config_file, "w") as f:
f.write("We previously crashed while writing me :(")
# Test updating when providing a new privkey. The key should
# be saved in a new file rather than creating a new symlink.
self.assertEqual(10, self.test_rc.save_successor(9, "with", "a",
"key"))
self.assertEqual(
10, self.test_rc.save_successor(9, "with", "a",
"key", self.cli_config))
self.assertTrue(os.path.exists(self.test_rc.version("privkey", 10)))
self.assertFalse(os.path.islink(self.test_rc.version("privkey", 10)))
self.assertFalse(os.path.exists(temp_config_file))
def test_new_lineage(self):
"""Test for new_lineage() class method."""
from letsencrypt import storage
result = storage.RenewableCert.new_lineage(
"the-lineage.com", "cert", "privkey", "chain", None,
self.defaults, self.cli_config)
"the-lineage.com", "cert", "privkey", "chain", self.cli_config)
# This consistency check tests most relevant properties about the
# newly created cert lineage.
# pylint: disable=protected-access
@ -563,27 +572,23 @@ class RenewableCertTests(BaseRenewableCertTest):
self.assertEqual(f.read(), "cert" + "chain")
# Let's do it again and make sure it makes a different lineage
result = storage.RenewableCert.new_lineage(
"the-lineage.com", "cert2", "privkey2", "chain2", None,
self.defaults, self.cli_config)
"the-lineage.com", "cert2", "privkey2", "chain2", self.cli_config)
self.assertTrue(os.path.exists(os.path.join(
self.cli_config.renewal_configs_dir, "the-lineage.com-0001.conf")))
# Now trigger the detection of already existing files
os.mkdir(os.path.join(
self.cli_config.live_dir, "the-lineage.com-0002"))
self.assertRaises(errors.CertStorageError,
storage.RenewableCert.new_lineage,
"the-lineage.com", "cert3", "privkey3", "chain3",
None, self.defaults, self.cli_config)
storage.RenewableCert.new_lineage, "the-lineage.com",
"cert3", "privkey3", "chain3", self.cli_config)
os.mkdir(os.path.join(self.cli_config.archive_dir, "other-example.com"))
self.assertRaises(errors.CertStorageError,
storage.RenewableCert.new_lineage,
"other-example.com", "cert4", "privkey4", "chain4",
None, self.defaults, self.cli_config)
"other-example.com", "cert4",
"privkey4", "chain4", self.cli_config)
# Make sure it can accept renewal parameters
params = {"stuff": "properties of stuff", "great": "awesome"}
result = storage.RenewableCert.new_lineage(
"the-lineage.com", "cert2", "privkey2", "chain2",
params, self.defaults, self.cli_config)
"the-lineage.com", "cert2", "privkey2", "chain2", self.cli_config)
# TODO: Conceivably we could test that the renewal parameters actually
# got saved
@ -595,8 +600,7 @@ class RenewableCertTests(BaseRenewableCertTest):
shutil.rmtree(self.cli_config.live_dir)
storage.RenewableCert.new_lineage(
"the-lineage.com", "cert2", "privkey2", "chain2",
None, self.defaults, self.cli_config)
"the-lineage.com", "cert2", "privkey2", "chain2", self.cli_config)
self.assertTrue(os.path.exists(
os.path.join(
self.cli_config.renewal_configs_dir, "the-lineage.com.conf")))
@ -610,9 +614,8 @@ class RenewableCertTests(BaseRenewableCertTest):
from letsencrypt import storage
mock_uln.return_value = "this_does_not_end_with_dot_conf", "yikes"
self.assertRaises(errors.CertStorageError,
storage.RenewableCert.new_lineage,
"example.com", "cert", "privkey", "chain",
None, self.defaults, self.cli_config)
storage.RenewableCert.new_lineage, "example.com",
"cert", "privkey", "chain", self.cli_config)
def test_bad_kind(self):
self.assertRaises(