starred/certbot-certbot
1
0
Fork 0
mirror of https://github.com/certbot/certbot.git synced 2026-08-01 19:02:52 +02:00
Code Issues Packages Projects Releases Wiki Activity

Work in progress: make renewer renew

Browse Source
This commit is contained in:
Seth Schoen
2015-05-10 22:43:58 -07:00
parent 30a4f72649
commit b235dc4c80
5 changed files with 536 additions and 458 deletions
Download Patch File Download Diff File Expand all files Collapse all files
letsencrypt/client/client.py
+2 -2
View File
@@ -17,9 +17,9 @@ from letsencrypt.client import errors
from letsencrypt.client import interfaces from letsencrypt.client import interfaces
from letsencrypt.client import le_util from letsencrypt.client import le_util
from letsencrypt.client import network2 from letsencrypt.client import network2
from letsencrypt.client import renewer
from letsencrypt.client import reverter from letsencrypt.client import reverter
from letsencrypt.client import revoker from letsencrypt.client import revoker
from letsencrypt.client import storage
from letsencrypt.client.display import ops as display_ops from letsencrypt.client.display import ops as display_ops
from letsencrypt.client.display import enhancements from letsencrypt.client.display import enhancements
@@ -160,7 +160,7 @@ class Client(object):
# of assuming that each plugin has a .name attribute # of assuming that each plugin has a .name attribute
self.config.namespace.authenticator = authenticator.name self.config.namespace.authenticator = authenticator.name
self.config.namespace.installer = installer.name self.config.namespace.installer = installer.name
return renewer.RenewableCert.new_lineage(domains[0], cert_pem, return storage.RenewableCert.new_lineage(domains[0], cert_pem,
privkey, chain_pem, privkey, chain_pem,
vars(self.config.namespace)) vars(self.config.namespace))
letsencrypt/client/crypto_util.py
+11
View File
@@ -231,3 +231,14 @@ def make_ss_cert(key_str, domains, not_before=None,
assert cert.verify() assert cert.verify()
# print check_purpose(,0 # print check_purpose(,0
return cert.as_pem() return cert.as_pem()
def get_sans_from_cert(pem):
"""Extracts the DNS subjectAltName values from a cert in PEM format.
"""
x509 = M2Crypto.X509.load_cert_string(pem)
try:
ext=x509.get_ext("subjectAltName")
except LookupError:
return []
return [x[4:] for x in ext.get_value().split(", ") if x.startswith("DNS:")]
letsencrypt/client/renewer.py
+55 -425
View File
@@ -14,461 +14,81 @@ configuration."""
# TODO: when renewing or deploying, update config file to # TODO: when renewing or deploying, update config file to
# memorialize the fact that it happened # memorialize the fact that it happened
import code #XXX: remove
import configobj import configobj
import copy import copy
import datetime import datetime
import os import os
import OpenSSL import OpenSSL
import parsedatetime
import pkg_resources import pkg_resources
import pyrfc3339 import pyrfc3339
import pytz import pytz
import re import re
import time import time
from letsencrypt.client import configuration
from letsencrypt.client import client
from letsencrypt.client import crypto_util
from letsencrypt.client import le_util from letsencrypt.client import le_util
from letsencrypt.client import notify from letsencrypt.client import notify
from letsencrypt.client import storage
from letsencrypt.client.plugins import disco as plugins_disco
DEFAULTS = configobj.ConfigObj("renewal.conf") DEFAULTS = configobj.ConfigObj("renewal.conf")
DEFAULTS["renewal_configs_dir"] = "/tmp/etc/letsencrypt/configs" DEFAULTS["renewal_configs_dir"] = "/tmp/etc/letsencrypt/configs"
DEFAULTS["official_archive_dir"] = "/tmp/etc/letsencrypt/archive" DEFAULTS["official_archive_dir"] = "/tmp/etc/letsencrypt/archive"
DEFAULTS["live_dir"] = "/tmp/etc/letsencrypt/live" DEFAULTS["live_dir"] = "/tmp/etc/letsencrypt/live"
ALL_FOUR = ("cert", "privkey", "chain", "fullchain")
def parse_time_interval(interval, textparser=parsedatetime.Calendar()): class AttrDict(dict):
"""Parse the time specified time interval, which can be in the def __init__(self, *args, **kwargs):
English-language format understood by parsedatetime, e.g., '10 days', super(AttrDict, self).__init__(*args, **kwargs)
'3 weeks', '6 months', '9 hours', or a sequence of such intervals self.__dict__ = self
like '6 months 1 week' or '3 days 12 hours'. If an integer is found
with no associated unit, it is interpreted by default as a number of
days."""
if interval.strip().isdigit():
interval += " days"
return datetime.timedelta(0, time.mktime(textparser.parse(
interval, time.localtime(0))[0]))
class RenewableCert(object): # pylint: disable=too-many-instance-attributes
"""Represents a lineage of certificates that is under the management
of the Let's Encrypt client, indicated by the existence of an
associated renewal configuration file."""
def __init__(self, configfile, defaults=DEFAULTS):
# self.configuration should be used to read parameters that
# may have been chosen based on default values from the
# systemwide renewal configuration; self.configfile should be
# used to make and save changes.
self.configuration = copy.deepcopy(defaults)
self.configfile = configobj.ConfigObj(configfile)
self.configuration.merge(self.configfile)
if not configfile.filename.endswith(".conf"):
raise ValueError("renewal config file name must end in .conf")
self.lineagename = os.path.basename(configfile.filename)[:-5]
self.configfilename = configfile.filename
self.cert = self.configuration["cert"]
self.privkey = self.configuration["privkey"]
self.chain = self.configuration["chain"]
self.fullchain = self.configuration["fullchain"]
def consistent(self):
"""Is the structure of the archived files and links related to this
lineage correct and self-consistent?"""
# Each element must be referenced with an absolute path
if any(not os.path.isabs(x) for x in
(self.cert, self.privkey, self.chain, self.fullchain)):
return False
# Each element must exist and be a symbolic link
if any(not os.path.islink(x) for x in
(self.cert, self.privkey, self.chain, self.fullchain)):
return False
for kind in ALL_FOUR:
link = self.__getattribute__(kind)
where = os.path.dirname(link)
target = os.readlink(link)
if not os.path.isabs(target):
target = os.path.join(where, target)
# Each element's link must point within the cert lineage's
# directory within the official archive directory
desired_directory = os.path.join(
self.configuration["official_archive_dir"], self.lineagename)
if not os.path.samefile(os.path.dirname(target),
desired_directory):
return False
# The link must point to a file that exists
if not os.path.exists(target):
return False
# The link must point to a file that follows the archive
# naming convention
pattern = re.compile(r"^{0}([0-9]+)\.pem$".format(kind))
if not pattern.match(os.path.basename(target)):
return False
# It is NOT required that the link's target be a regular
# file (it may itself be a symlink). But we should probably
# do a recursive check that ultimately the target does
# exist?
# XXX: Additional possible consistency checks (e.g.
# cryptographic validation of the chain being a chain,
# the chain matching the cert, and the cert matching
# the subject key)
# XXX: All four of the targets are in the same directory
# (This check is redundant with the check that they
# are all in the desired directory!)
# len(set(os.path.basename(self.current_target(x)
# for x in ALL_FOUR))) == 1
return True
def fix(self):
"""Attempt to fix some kinds of defects or inconsistencies
in the symlink structure, if possible."""
# TODO: Figure out what kinds of fixes are possible. For
# example, checking if there is a valid version that
# we can update the symlinks to. (Maybe involve
# parsing keys and certs to see if they exist and
# if a key corresponds to the subject key of a cert?)
# TODO: In general, the symlink-reading functions below are not
# cautious enough about the possibility that links or their
# targets may not exist. (This shouldn't happen, but might
# happen as a result of random tampering by a sysadmin, or
# filesystem errors, or crashes.)
def current_target(self, kind):
"""Returns the full path to which the link of the specified
kind currently points."""
if kind not in ALL_FOUR:
raise ValueError("unknown kind of item")
link = self.__getattribute__(kind)
if not os.path.exists(link):
return None
target = os.readlink(link)
if not os.path.isabs(target):
target = os.path.join(os.path.dirname(link), target)
return target
def current_version(self, kind):
"""Returns the numerical version of the object to which the link
of the specified kind currently points. For example, if kind
is "chain" and the current chain link points to a file named
"chain7.pem", returns the integer 7."""
if kind not in ALL_FOUR:
raise ValueError("unknown kind of item")
pattern = re.compile(r"^{0}([0-9]+)\.pem$".format(kind))
target = self.current_target(kind)
if not target or not os.path.exists(target):
target = ""
matches = pattern.match(os.path.basename(target))
if matches:
return int(matches.groups()[0])
else:
return None
def version(self, kind, version):
"""Constructs the filename that would correspond to the
specified version of the specified kind of item in this
lineage. Warning: the specified version may not exist."""
if kind not in ALL_FOUR:
raise ValueError("unknown kind of item")
where = os.path.dirname(self.current_target(kind))
return os.path.join(where, "{0}{1}.pem".format(kind, version))
def available_versions(self, kind):
"""Which alternative versions of the specified kind of item
exist in the archive directory where the current version is
stored?"""
if kind not in ALL_FOUR:
raise ValueError("unknown kind of item")
where = os.path.dirname(self.current_target(kind))
files = os.listdir(where)
pattern = re.compile(r"^{0}([0-9]+)\.pem$".format(kind))
matches = [pattern.match(f) for f in files]
return sorted([int(m.groups()[0]) for m in matches if m])
def newest_available_version(self, kind):
"""What is the newest available version of the specified
kind of item?"""
return max(self.available_versions(kind))
def latest_common_version(self):
"""What is the largest version number for which versions
of cert, privkey, chain, and fullchain are all available?"""
# TODO: this can raise ValueError if there is no version overlap
# (it should probably return None instead)
# TODO: this can raise a spurious AttributeError if the current
# link for any kind is missing (it should probably return None)
versions = [self.available_versions(x) for x in ALL_FOUR]
return max(n for n in versions[0] if all(n in v for v in versions[1:]))
def next_free_version(self):
"""What is the smallest new version number that is larger than
any available version of any managed item?"""
# TODO: consider locking/mutual exclusion between updating processes
# This isn't self.latest_common_version() + 1 because we don't want
# collide with a version that might exist for one file type but not
# for the others.
return max(self.newest_available_version(x) for x in ALL_FOUR) + 1
def has_pending_deployment(self):
"""Is there a later version of all of the managed items?"""
# TODO: consider whether to assume consistency or treat
# inconsistent/consistent versions differently
smallest_current = min(self.current_version(x) for x in ALL_FOUR)
return smallest_current < self.latest_common_version()
def update_link_to(self, kind, version):
"""Change the target of the link of the specified item to point
to the specified version. (Note that this method doesn't verify
that the specified version exists.)"""
if kind not in ALL_FOUR:
raise ValueError("unknown kind of item")
link = self.__getattribute__(kind)
filename = "{0}{1}.pem".format(kind, version)
# Relative rather than absolute target directory
target_directory = os.path.dirname(os.readlink(link))
# TODO: it could be safer to make the link first under a temporary
# filename, then unlink the old link, then rename the new link
# to the old link; this ensures that this process is able to
# create symlinks.
# TODO: we might also want to check consistency of related links
# for the other corresponding items
os.unlink(link)
os.symlink(os.path.join(target_directory, filename), link)
def update_all_links_to(self, version):
"""Change the target of the cert, privkey, chain, and fullchain links
to point to the specified version."""
for kind in ALL_FOUR:
self.update_link_to(kind, version)
def notbefore(self, version=None):
"""When is the beginning validity time of the specified version of the
cert in this lineage? (If no version is specified, use the current
version.)"""
if version == None:
target = self.current_target("cert")
else:
target = self.version("cert", version)
pem = open(target).read()
x509 = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM,
pem)
i = x509.get_notBefore()
return pyrfc3339.parse(i[0:4] + "-" + i[4:6] + "-" + i[6:8] + "T" +
i[8:10] + ":" + i[10:12] +":" +i[12:])
def notafter(self, version=None):
"""When is the ending validity time of the specified version of the
cert in this lineage? (If no version is specified, use the current
version.)"""
if version == None:
target = self.current_target("cert")
else:
target = self.version("cert", version)
pem = open(target).read()
x509 = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM,
pem)
i = x509.get_notAfter()
return pyrfc3339.parse(i[0:4] + "-" + i[4:6] + "-" + i[6:8] + "T" +
i[8:10] + ":" + i[10:12] +":" +i[12:])
def should_autodeploy(self):
"""Should this certificate lineage be updated automatically to
point to an existing pending newer version? (Considers whether
autodeployment is enabled, whether a relevant newer version
exists, and whether the time interval for autodeployment has
been reached.)"""
if (not self.configuration.has_key("autodeploy") or
self.configuration.as_bool("autodeploy")):
if self.has_pending_deployment():
interval = self.configuration.get("deploy_before_expiry",
"5 days")
autodeploy_interval = parse_time_interval(interval)
expiry = self.notafter()
now = datetime.datetime.utcnow().replace(tzinfo=pytz.UTC)
remaining = expiry - now
if remaining < autodeploy_interval:
return True
return False
def ocsp_revoked(self, version=None):
# pylint: disable=no-self-use,unused-argument
"""Is the specified version of this certificate lineage revoked
according to OCSP or intended to be revoked according to Let's
Encrypt OCSP extensions? (If no version is specified, use the
current version.)"""
# XXX: This query and its associated network service aren't
# implemented yet, so we currently return False (indicating that the
# certificate is not revoked).
return False
def should_autorenew(self):
"""Should an attempt be made to automatically renew the most
recent certificate in this certificate lineage right now?"""
if (not self.configuration.has_key("autorenew")
or self.configuration.as_bool("autorenew")):
# Consider whether to attempt to autorenew this cert now
# XXX: both self.ocsp_revoked() and self.notafter() are bugs
# here because we should be looking at the latest version, not
# the current version!
# Renewals on the basis of revocation
if self.ocsp_revoked():
return True
# Renewals on the basis of expiry time
interval = self.configuration.get("renew_before_expiry", "10 days")
autorenew_interval = parse_time_interval(interval)
expiry = self.notafter()
now = datetime.datetime.utcnow().replace(tzinfo=pytz.UTC)
remaining = expiry - now
if remaining < autorenew_interval:
return True
return False
@classmethod
def new_lineage(cls, lineagename, cert, privkey, chain,
renewalparams=None, config=DEFAULTS):
# pylint: disable=too-many-locals
"""Create a new certificate lineage with the (suggested) lineage name
lineagename, and the associated cert, privkey, and chain (the
associated fullchain will be created automatically). Optional
configurator and renewalparams record the configuration that was
originally used to obtain this cert, so that it can be reused later
during automated renewal.
Returns a new RenewableCert object referring to the created
lineage. (The actual lineage name, as well as all the relevant
file paths, will be available within this object.)"""
print config
configs_dir = config["renewal_configs_dir"]
archive_dir = config["official_archive_dir"]
live_dir = config["live_dir"]
for i in (configs_dir, archive_dir, live_dir):
if not os.path.exists(i):
os.makedirs(i, 0700)
config_file, config_filename = le_util.unique_lineage_name(configs_dir,
lineagename)
if not config_filename.endswith(".conf"):
raise ValueError("renewal config file name must end in .conf")
# lineagename will now potentially be modified based on what
# renewal configuration file could actually be created
lineagename = os.path.basename(config_filename)[:-5]
archive = os.path.join(archive_dir, lineagename)
live_dir = os.path.join(live_dir, lineagename)
if os.path.exists(archive):
raise ValueError("archive directory exists for " + lineagename)
if os.path.exists(live_dir):
raise ValueError("live directory exists for " + lineagename)
os.mkdir(archive)
os.mkdir(live_dir)
relative_archive = os.path.join("..", "..", "archive", lineagename)
cert_target = os.path.join(live_dir, "cert.pem")
privkey_target = os.path.join(live_dir, "privkey.pem")
chain_target = os.path.join(live_dir, "chain.pem")
fullchain_target = os.path.join(live_dir, "fullchain.pem")
os.symlink(os.path.join(relative_archive, "cert1.pem"),
cert_target)
os.symlink(os.path.join(relative_archive, "privkey1.pem"),
privkey_target)
os.symlink(os.path.join(relative_archive, "chain1.pem"),
chain_target)
os.symlink(os.path.join(relative_archive, "fullchain1.pem"),
fullchain_target)
with open(cert_target, "w") as f:
f.write(cert)
with open(privkey_target, "w") as f:
f.write(privkey)
# XXX: Let's make sure to get the file permissions right here
with open(chain_target, "w") as f:
f.write(chain)
with open(fullchain_target, "w") as f:
f.write(cert + chain)
config_file.close()
new_config = configobj.ConfigObj(config_filename, create_empty=True)
new_config["cert"] = cert_target
new_config["privkey"] = privkey_target
new_config["chain"] = chain_target
new_config["fullchain"] = fullchain_target
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
new_config.write()
return cls(new_config, config)
def save_successor(self, prior_version, new_cert, new_chain):
"""Save a new cert and chain as a successor of a specific prior
version in this lineage. Returns the new version number that was
created. Note: does NOT update links to deploy this version."""
# XXX: no private key change: should be extended with a key=None
# default argument that allows changing the private key; also we
# should perhaps allow new_chain=None which makes a link to
# the prior chain's target
# XXX: assumes official archive location rather than examining links
# XXX: consider using os.open for availablity of os.O_EXCL
# XXX: ensure file permissions are correct; also create directories
# if needed (ensuring their permissions are correct)
target_version = self.next_free_version()
archive = self.configuration["official_archive_dir"]
prefix = os.path.join(archive, self.lineagename)
cert_target = os.path.join(
prefix, "cert{0}.pem".format(target_version))
privkey_target = os.path.join(
prefix, "privkey{0}.pem".format(target_version))
chain_target = os.path.join(
prefix, "chain{0}.pem".format(target_version))
fullchain_target = os.path.join(
prefix, "fullchain{0}.pem".format(target_version))
with open(cert_target, "w") as f:
f.write(new_cert)
# The behavior below always keeps the prior key by creating a new
# symlink to the old key or the target of the old key symlink.
old_privkey = os.path.join(
prefix, "privkey{0}.pem".format(prior_version))
if os.path.islink(old_privkey):
old_privkey = os.readlink(old_privkey)
else:
old_privkey = "privkey{0}.pem".format(prior_version)
os.symlink(old_privkey, privkey_target)
with open(chain_target, "w") as f:
f.write(new_chain)
with open(fullchain_target, "w") as f:
f.write(new_cert + new_chain)
return target_version
def renew(cert, old_version): def renew(cert, old_version):
"""Perform automated renewal of the referenced cert, if possible.""" """Perform automated renewal of the referenced cert, if possible."""
# TODO: handle partial success # TODO: handle partial success
# TODO: handle obligatory key rotation # TODO: handle obligatory key rotation vs. optional key rotation vs.
# XXX: Deserialize config here # requested key rotation
if not cert.configfile.has_key("renewalparams"):
for entrypoint in pkg_resources.iter_entry_points( # TODO: notify user?
SETUPTOOLS_AUTHENTICATORS_ENTRY_POINT): return False
auth_cls = entrypoint.load() renewalparams = cert.configfile["renewalparams"]
# XXX: need to regenerate "config" from serialized authenticator if not renewalparams.has_key("authenticator"):
# config! # TODO: notify user?
auth = auth_cls(config) return False
try: # Instantiate the appropriate authenticator
zope.interface.verify.verifyObject(interfaces.IAuthenticator, auth) plugins = plugins_disco.PluginsRegistry.find_all()
except zope.interface.exceptions.BrokenImplementation: try:
pass config = configuration.NamespaceConfig(AttrDict(renewalparams))
else: # XXX: this loses type data (for example, the fact that key_size
if entrypoint.name == cert.configuration["authenticator"]: # was an int, not a str)
break config.rsa_key_size = int(config.rsa_key_size)
else: authenticator = plugins[renewalparams["authenticator"]]
# TODO: Notify failure to instantiate the authenticator authenticator = authenticator.init(config)
except KeyError:
# TODO: Notify user? (authenticator could not be found)
return False return False
auth.prepare()
client = Client(config, None, auth, None)
new_cert, new_key, new_chain = client.obtain_certificate(domains) authenticator.prepare()
account = client.determine_account(config)
# TODO: are there other ways to get the right account object, e.g.
# based on the email parameter that might be present in
# renewalparams?
our_client = client.Client(config, account, authenticator, None)
# XXX: find the domains
with open(cert.version("cert", old_version)) as f:
sans = crypto_util.get_sans_from_cert(f.read())
new_cert, new_key, new_chain = our_client.obtain_certificate(sans)
if new_cert and new_key and new_chain: if new_cert and new_key and new_chain:
# XXX: Assumes that there was no key change. We need logic # XXX: Assumes that there was no key change. We need logic
# for figuring out whether there was or not. Probably # for figuring out whether there was or not. Probably
# best is to have obtain_certificate return None for # best is to have obtain_certificate return None for
# new_key if the old key is to be used (since save_successor # new_key if the old key is to be used (since save_successor
# already understands this distinction!) # already understands this distinction!)
self.save_successor(old_version, new_cert, new_chain) cert.save_successor(old_version, new_cert, new_key, new_chain)
# Notify results # Notify results
else: else:
# Notify negative results # Notify negative results
@@ -481,7 +101,17 @@ def main(config=DEFAULTS):
print "Processing", i print "Processing", i
if not i.endswith(".conf"): if not i.endswith(".conf"):
continue continue
cert = RenewableCert(i) try:
cert = storage.RenewableCert(
os.path.join(config["renewal_configs_dir"], i))
except ValueError:
# This indicates an invalid renewal configuration file, such
# as one missing a required parameter (in the future, perhaps
# also one that is internally inconsistent or is missing a
# required parameter). As a TODO, maybe we should warn the
# user about the existence of an invalid or corrupt renewal
# config rather than simply ignoring it.
continue
if cert.should_autodeploy(): if cert.should_autodeploy():
cert.update_all_links_to(cert.latest_common_version()) cert.update_all_links_to(cert.latest_common_version())
# TODO: restart web server # TODO: restart web server
letsencrypt/client/storage.py
+432
View File
@@ -0,0 +1,432 @@
import configobj
import copy
import datetime
import os
import OpenSSL
import parsedatetime
import pyrfc3339
import pytz
import re
import time
from letsencrypt.client import le_util
DEFAULTS = configobj.ConfigObj("renewal.conf")
DEFAULTS["renewal_configs_dir"] = "/tmp/etc/letsencrypt/configs"
DEFAULTS["official_archive_dir"] = "/tmp/etc/letsencrypt/archive"
DEFAULTS["live_dir"] = "/tmp/etc/letsencrypt/live"
ALL_FOUR = ("cert", "privkey", "chain", "fullchain")
def parse_time_interval(interval, textparser=parsedatetime.Calendar()):
"""Parse the time specified time interval, which can be in the
English-language format understood by parsedatetime, e.g., '10 days',
'3 weeks', '6 months', '9 hours', or a sequence of such intervals
like '6 months 1 week' or '3 days 12 hours'. If an integer is found
with no associated unit, it is interpreted by default as a number of
days."""
if interval.strip().isdigit():
interval += " days"
return datetime.timedelta(0, time.mktime(textparser.parse(
interval, time.localtime(0))[0]))
class RenewableCert(object): # pylint: disable=too-many-instance-attributes
"""Represents a lineage of certificates that is under the management
of the Let's Encrypt client, indicated by the existence of an
associated renewal configuration file."""
def __init__(self, configfile, defaults=DEFAULTS):
if isinstance(configfile, str):
if not os.path.exists(configfile):
raise ValueError(
"renewal config file {0} doesn't exist".format(configfile))
if not configfile.endswith(".conf"):
raise ValueError("renewal config file name must end in .conf")
self.lineagename = os.path.basename(configfile)[:-5]
self.configfilename = os.path.basename(configfile)
elif isinstance(configfile, configobj.ConfigObj):
self.lineagename = os.path.basename(configfile.filename)[:-5]
self.configfilename = os.path.basename(configfile.filename)
else:
raise TypeError("RenewableCert config must be file path "
"or ConfigObj object")
# self.configuration should be used to read parameters that
# may have been chosen based on default values from the
# systemwide renewal configuration; self.configfile should be
# used to make and save changes.
self.configuration = copy.deepcopy(defaults)
self.configfile = configobj.ConfigObj(configfile)
self.configuration.merge(self.configfile)
if not all(self.configuration.has_key(x) for x in ALL_FOUR):
raise ValueError("renewal config file {0} is missing a required "
"file reference".format(configfile))
self.cert = self.configuration["cert"]
self.privkey = self.configuration["privkey"]
self.chain = self.configuration["chain"]
self.fullchain = self.configuration["fullchain"]
def consistent(self):
"""Is the structure of the archived files and links related to this
lineage correct and self-consistent?"""
# Each element must be referenced with an absolute path
if any(not os.path.isabs(x) for x in
(self.cert, self.privkey, self.chain, self.fullchain)):
return False
# Each element must exist and be a symbolic link
if any(not os.path.islink(x) for x in
(self.cert, self.privkey, self.chain, self.fullchain)):
return False
for kind in ALL_FOUR:
link = self.__getattribute__(kind)
where = os.path.dirname(link)
target = os.readlink(link)
if not os.path.isabs(target):
target = os.path.join(where, target)
# Each element's link must point within the cert lineage's
# directory within the official archive directory
desired_directory = os.path.join(
self.configuration["official_archive_dir"], self.lineagename)
if not os.path.samefile(os.path.dirname(target),
desired_directory):
return False
# The link must point to a file that exists
if not os.path.exists(target):
return False
# The link must point to a file that follows the archive
# naming convention
pattern = re.compile(r"^{0}([0-9]+)\.pem$".format(kind))
if not pattern.match(os.path.basename(target)):
return False
# It is NOT required that the link's target be a regular
# file (it may itself be a symlink). But we should probably
# do a recursive check that ultimately the target does
# exist?
# XXX: Additional possible consistency checks (e.g.
# cryptographic validation of the chain being a chain,
# the chain matching the cert, and the cert matching
# the subject key)
# XXX: All four of the targets are in the same directory
# (This check is redundant with the check that they
# are all in the desired directory!)
# len(set(os.path.basename(self.current_target(x)
# for x in ALL_FOUR))) == 1
return True
def fix(self):
"""Attempt to fix some kinds of defects or inconsistencies
in the symlink structure, if possible."""
# TODO: Figure out what kinds of fixes are possible. For
# example, checking if there is a valid version that
# we can update the symlinks to. (Maybe involve
# parsing keys and certs to see if they exist and
# if a key corresponds to the subject key of a cert?)
# TODO: In general, the symlink-reading functions below are not
# cautious enough about the possibility that links or their
# targets may not exist. (This shouldn't happen, but might
# happen as a result of random tampering by a sysadmin, or
# filesystem errors, or crashes.)
def current_target(self, kind):
"""Returns the full path to which the link of the specified
kind currently points."""
if kind not in ALL_FOUR:
raise ValueError("unknown kind of item")
link = self.__getattribute__(kind)
if not os.path.exists(link):
return None
target = os.readlink(link)
if not os.path.isabs(target):
target = os.path.join(os.path.dirname(link), target)
return target
def current_version(self, kind):
"""Returns the numerical version of the object to which the link
of the specified kind currently points. For example, if kind
is "chain" and the current chain link points to a file named
"chain7.pem", returns the integer 7."""
if kind not in ALL_FOUR:
raise ValueError("unknown kind of item")
pattern = re.compile(r"^{0}([0-9]+)\.pem$".format(kind))
target = self.current_target(kind)
if not target or not os.path.exists(target):
target = ""
matches = pattern.match(os.path.basename(target))
if matches:
return int(matches.groups()[0])
else:
return None
def version(self, kind, version):
"""Constructs the filename that would correspond to the
specified version of the specified kind of item in this
lineage. Warning: the specified version may not exist."""
if kind not in ALL_FOUR:
raise ValueError("unknown kind of item")
where = os.path.dirname(self.current_target(kind))
return os.path.join(where, "{0}{1}.pem".format(kind, version))
def available_versions(self, kind):
"""Which alternative versions of the specified kind of item
exist in the archive directory where the current version is
stored?"""
if kind not in ALL_FOUR:
raise ValueError("unknown kind of item")
where = os.path.dirname(self.current_target(kind))
files = os.listdir(where)
pattern = re.compile(r"^{0}([0-9]+)\.pem$".format(kind))
matches = [pattern.match(f) for f in files]
return sorted([int(m.groups()[0]) for m in matches if m])
def newest_available_version(self, kind):
"""What is the newest available version of the specified
kind of item?"""
return max(self.available_versions(kind))
def latest_common_version(self):
"""What is the largest version number for which versions
of cert, privkey, chain, and fullchain are all available?"""
# TODO: this can raise ValueError if there is no version overlap
# (it should probably return None instead)
# TODO: this can raise a spurious AttributeError if the current
# link for any kind is missing (it should probably return None)
versions = [self.available_versions(x) for x in ALL_FOUR]
return max(n for n in versions[0] if all(n in v for v in versions[1:]))
def next_free_version(self):
"""What is the smallest new version number that is larger than
any available version of any managed item?"""
# TODO: consider locking/mutual exclusion between updating processes
# This isn't self.latest_common_version() + 1 because we don't want
# collide with a version that might exist for one file type but not
# for the others.
return max(self.newest_available_version(x) for x in ALL_FOUR) + 1
def has_pending_deployment(self):
"""Is there a later version of all of the managed items?"""
# TODO: consider whether to assume consistency or treat
# inconsistent/consistent versions differently
smallest_current = min(self.current_version(x) for x in ALL_FOUR)
return smallest_current < self.latest_common_version()
def update_link_to(self, kind, version):
"""Change the target of the link of the specified item to point
to the specified version. (Note that this method doesn't verify
that the specified version exists.)"""
if kind not in ALL_FOUR:
raise ValueError("unknown kind of item")
link = self.__getattribute__(kind)
filename = "{0}{1}.pem".format(kind, version)
# Relative rather than absolute target directory
target_directory = os.path.dirname(os.readlink(link))
# TODO: it could be safer to make the link first under a temporary
# filename, then unlink the old link, then rename the new link
# to the old link; this ensures that this process is able to
# create symlinks.
# TODO: we might also want to check consistency of related links
# for the other corresponding items
os.unlink(link)
os.symlink(os.path.join(target_directory, filename), link)
def update_all_links_to(self, version):
"""Change the target of the cert, privkey, chain, and fullchain links
to point to the specified version."""
for kind in ALL_FOUR:
self.update_link_to(kind, version)
def notbefore(self, version=None):
"""When is the beginning validity time of the specified version of the
cert in this lineage? (If no version is specified, use the current
version.)"""
if version == None:
target = self.current_target("cert")
else:
target = self.version("cert", version)
pem = open(target).read()
x509 = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM,
pem)
i = x509.get_notBefore()
return pyrfc3339.parse(i[0:4] + "-" + i[4:6] + "-" + i[6:8] + "T" +
i[8:10] + ":" + i[10:12] +":" +i[12:])
def notafter(self, version=None):
"""When is the ending validity time of the specified version of the
cert in this lineage? (If no version is specified, use the current
version.)"""
if version == None:
target = self.current_target("cert")
else:
target = self.version("cert", version)
pem = open(target).read()
x509 = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM,
pem)
i = x509.get_notAfter()
return pyrfc3339.parse(i[0:4] + "-" + i[4:6] + "-" + i[6:8] + "T" +
i[8:10] + ":" + i[10:12] +":" +i[12:])
def should_autodeploy(self):
"""Should this certificate lineage be updated automatically to
point to an existing pending newer version? (Considers whether
autodeployment is enabled, whether a relevant newer version
exists, and whether the time interval for autodeployment has
been reached.)"""
if (not self.configuration.has_key("autodeploy") or
self.configuration.as_bool("autodeploy")):
if self.has_pending_deployment():
interval = self.configuration.get("deploy_before_expiry",
"5 days")
autodeploy_interval = parse_time_interval(interval)
expiry = self.notafter()
now = datetime.datetime.utcnow().replace(tzinfo=pytz.UTC)
remaining = expiry - now
if remaining < autodeploy_interval:
return True
return False
def ocsp_revoked(self, version=None):
# pylint: disable=no-self-use,unused-argument
"""Is the specified version of this certificate lineage revoked
according to OCSP or intended to be revoked according to Let's
Encrypt OCSP extensions? (If no version is specified, use the
current version.)"""
# XXX: This query and its associated network service aren't
# implemented yet, so we currently return False (indicating that the
# certificate is not revoked).
return False
def should_autorenew(self):
"""Should an attempt be made to automatically renew the most
recent certificate in this certificate lineage right now?"""
if (not self.configuration.has_key("autorenew")
or self.configuration.as_bool("autorenew")):
# Consider whether to attempt to autorenew this cert now
# XXX: both self.ocsp_revoked() and self.notafter() are bugs
# here because we should be looking at the latest version, not
# the current version!
# Renewals on the basis of revocation
if self.ocsp_revoked():
return True
# Renewals on the basis of expiry time
interval = self.configuration.get("renew_before_expiry", "10 days")
autorenew_interval = parse_time_interval(interval)
expiry = self.notafter()
now = datetime.datetime.utcnow().replace(tzinfo=pytz.UTC)
remaining = expiry - now
if remaining < autorenew_interval:
return True
return False
@classmethod
def new_lineage(cls, lineagename, cert, privkey, chain,
renewalparams=None, config=DEFAULTS):
# pylint: disable=too-many-locals
"""Create a new certificate lineage with the (suggested) lineage name
lineagename, and the associated cert, privkey, and chain (the
associated fullchain will be created automatically). Optional
configurator and renewalparams record the configuration that was
originally used to obtain this cert, so that it can be reused later
during automated renewal.
Returns a new RenewableCert object referring to the created
lineage. (The actual lineage name, as well as all the relevant
file paths, will be available within this object.)"""
configs_dir = config["renewal_configs_dir"]
archive_dir = config["official_archive_dir"]
live_dir = config["live_dir"]
for i in (configs_dir, archive_dir, live_dir):
if not os.path.exists(i):
os.makedirs(i, 0700)
config_file, config_filename = le_util.unique_lineage_name(configs_dir,
lineagename)
if not config_filename.endswith(".conf"):
raise ValueError("renewal config file name must end in .conf")
# lineagename will now potentially be modified based on what
# renewal configuration file could actually be created
lineagename = os.path.basename(config_filename)[:-5]
archive = os.path.join(archive_dir, lineagename)
live_dir = os.path.join(live_dir, lineagename)
if os.path.exists(archive):
raise ValueError("archive directory exists for " + lineagename)
if os.path.exists(live_dir):
raise ValueError("live directory exists for " + lineagename)
os.mkdir(archive)
os.mkdir(live_dir)
relative_archive = os.path.join("..", "..", "archive", lineagename)
cert_target = os.path.join(live_dir, "cert.pem")
privkey_target = os.path.join(live_dir, "privkey.pem")
chain_target = os.path.join(live_dir, "chain.pem")
fullchain_target = os.path.join(live_dir, "fullchain.pem")
os.symlink(os.path.join(relative_archive, "cert1.pem"),
cert_target)
os.symlink(os.path.join(relative_archive, "privkey1.pem"),
privkey_target)
os.symlink(os.path.join(relative_archive, "chain1.pem"),
chain_target)
os.symlink(os.path.join(relative_archive, "fullchain1.pem"),
fullchain_target)
with open(cert_target, "w") as f:
f.write(cert)
with open(privkey_target, "w") as f:
f.write(privkey)
# XXX: Let's make sure to get the file permissions right here
with open(chain_target, "w") as f:
f.write(chain)
with open(fullchain_target, "w") as f:
f.write(cert + chain)
config_file.close()
new_config = configobj.ConfigObj(config_filename, create_empty=True)
new_config["cert"] = cert_target
new_config["privkey"] = privkey_target
new_config["chain"] = chain_target
new_config["fullchain"] = fullchain_target
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
new_config.write()
return cls(new_config, config)
def save_successor(self, prior_version, new_cert, new_privkey, new_chain):
"""Save a new cert and chain as a successor of a specific prior
version in this lineage. Returns the new version number that was
created. Note: does NOT update links to deploy this version."""
# XXX: assumes official archive location rather than examining links
# XXX: consider using os.open for availablity of os.O_EXCL
# XXX: ensure file permissions are correct; also create directories
# if needed (ensuring their permissions are correct)
target_version = self.next_free_version()
archive = self.configuration["official_archive_dir"]
prefix = os.path.join(archive, self.lineagename)
cert_target = os.path.join(
prefix, "cert{0}.pem".format(target_version))
privkey_target = os.path.join(
prefix, "privkey{0}.pem".format(target_version))
chain_target = os.path.join(
prefix, "chain{0}.pem".format(target_version))
fullchain_target = os.path.join(
prefix, "fullchain{0}.pem".format(target_version))
with open(cert_target, "w") as f:
f.write(new_cert)
if new_privkey is None:
# The behavior below keeps the prior key by creating a new
# symlink to the old key or the target of the old key symlink.
old_privkey = os.path.join(
prefix, "privkey{0}.pem".format(prior_version))
if os.path.islink(old_privkey):
old_privkey = os.readlink(old_privkey)
else:
old_privkey = "privkey{0}.pem".format(prior_version)
os.symlink(old_privkey, privkey_target)
else:
with open(privkey_target, "w") as f:
f.write(new_privkey)
with open(chain_target, "w") as f:
f.write(new_chain)
with open(fullchain_target, "w") as f:
f.write(new_cert + new_chain)
return target_version
letsencrypt/client/tests/renewer_test.py
+36 -31
View File
@@ -17,7 +17,7 @@ class RenewableCertTests(unittest.TestCase):
"""Tests for the RenewableCert class as well as other functions """Tests for the RenewableCert class as well as other functions
within renewer.py.""" within renewer.py."""
def setUp(self): def setUp(self):
from letsencrypt.client import renewer from letsencrypt.client import storage
self.tempdir = tempfile.mkdtemp() self.tempdir = tempfile.mkdtemp()
os.makedirs(os.path.join(self.tempdir, "live", "example.org")) os.makedirs(os.path.join(self.tempdir, "live", "example.org"))
os.makedirs(os.path.join(self.tempdir, "archive", "example.org")) os.makedirs(os.path.join(self.tempdir, "archive", "example.org"))
@@ -38,7 +38,7 @@ class RenewableCertTests(unittest.TestCase):
config.filename = os.path.join(self.tempdir, "configs", config.filename = os.path.join(self.tempdir, "configs",
"example.org.conf") "example.org.conf")
self.defaults = defaults # for main() test self.defaults = defaults # for main() test
self.test_rc = renewer.RenewableCert(config, defaults) self.test_rc = storage.RenewableCert(config, defaults)
def tearDown(self): def tearDown(self):
shutil.rmtree(self.tempdir) shutil.rmtree(self.tempdir)
@@ -65,6 +65,7 @@ class RenewableCertTests(unittest.TestCase):
"""Test that the RenewableCert constructor will complain if """Test that the RenewableCert constructor will complain if
the renewal configuration file doesn't end in ".conf".""" the renewal configuration file doesn't end in ".conf"."""
from letsencrypt.client import renewer from letsencrypt.client import renewer
from letsencrypt.client import storage
defaults = configobj.ConfigObj() defaults = configobj.ConfigObj()
config = configobj.ConfigObj() config = configobj.ConfigObj()
config["cert"] = "/tmp/cert.pem" config["cert"] = "/tmp/cert.pem"
@@ -72,7 +73,7 @@ class RenewableCertTests(unittest.TestCase):
config["chain"] = "/tmp/chain.pem" config["chain"] = "/tmp/chain.pem"
config["fullchain"] = "/tmp/fullchain.pem" config["fullchain"] = "/tmp/fullchain.pem"
config.filename = "/tmp/sillyfile" config.filename = "/tmp/sillyfile"
self.assertRaises(ValueError, renewer.RenewableCert, config, defaults) self.assertRaises(ValueError, storage.RenewableCert, config, defaults)
def test_consistent(self): # pylint: disable=too-many-statements def test_consistent(self): # pylint: disable=too-many-statements
oldcert = self.test_rc.cert oldcert = self.test_rc.cert
@@ -404,7 +405,7 @@ class RenewableCertTests(unittest.TestCase):
self.assertTrue(self.test_rc.should_autodeploy()) self.assertTrue(self.test_rc.should_autodeploy())
@mock.patch("letsencrypt.client.renewer.datetime") @mock.patch("letsencrypt.client.renewer.datetime")
@mock.patch("letsencrypt.client.renewer.RenewableCert.ocsp_revoked") @mock.patch("letsencrypt.client.storage.RenewableCert.ocsp_revoked")
def test_should_autorenew(self, mock_ocsp, mock_datetime): def test_should_autorenew(self, mock_ocsp, mock_datetime):
# pylint: disable=too-many-statements # pylint: disable=too-many-statements
# Autorenewal turned off # Autorenewal turned off
@@ -483,7 +484,7 @@ class RenewableCertTests(unittest.TestCase):
with open(where, "w") as f: with open(where, "w") as f:
f.write(kind) f.write(kind)
self.test_rc.update_all_links_to(3) self.test_rc.update_all_links_to(3)
self.assertEqual(6, self.test_rc.save_successor(3, "new cert", self.assertEqual(6, self.test_rc.save_successor(3, "new cert", "key",
"new chain")) "new chain"))
with open(self.test_rc.version("cert", 6)) as f: with open(self.test_rc.version("cert", 6)) as f:
self.assertEqual(f.read(), "new cert") self.assertEqual(f.read(), "new cert")
@@ -495,9 +496,9 @@ class RenewableCertTests(unittest.TestCase):
self.assertFalse(os.path.islink(self.test_rc.version("privkey", 3))) self.assertFalse(os.path.islink(self.test_rc.version("privkey", 3)))
self.assertTrue(os.path.islink(self.test_rc.version("privkey", 6))) self.assertTrue(os.path.islink(self.test_rc.version("privkey", 6)))
# Let's try two more updates # Let's try two more updates
self.assertEqual(7, self.test_rc.save_successor(6, "again", self.assertEqual(7, self.test_rc.save_successor(6, "again", "key",
"newer chain")) "newer chain"))
self.assertEqual(8, self.test_rc.save_successor(7, "hello", self.assertEqual(8, self.test_rc.save_successor(7, "hello", "key",
"other chain")) "other chain"))
# All of the subsequent versions should link directly to the original # All of the subsequent versions should link directly to the original
# privkey. # privkey.
@@ -518,7 +519,8 @@ class RenewableCertTests(unittest.TestCase):
self.assertEqual(self.test_rc.current_version(kind), 3) self.assertEqual(self.test_rc.current_version(kind), 3)
# Test updating from latest version rather than old version # Test updating from latest version rather than old version
self.test_rc.update_all_links_to(8) self.test_rc.update_all_links_to(8)
self.assertEqual(9, self.test_rc.save_successor(8, "last", "attempt")) self.assertEqual(9, self.test_rc.save_successor(8, "a", "last",
"attempt"))
for kind in ALL_FOUR: for kind in ALL_FOUR:
self.assertEqual(self.test_rc.available_versions(kind), self.assertEqual(self.test_rc.available_versions(kind),
range(1, 10)) range(1, 10))
@@ -529,12 +531,13 @@ class RenewableCertTests(unittest.TestCase):
def test_new_lineage(self): def test_new_lineage(self):
"""Test for new_lineage() class method.""" """Test for new_lineage() class method."""
from letsencrypt.client import renewer from letsencrypt.client import renewer
from letsencrypt.client import storage
config_dir = self.defaults["renewal_configs_dir"] config_dir = self.defaults["renewal_configs_dir"]
archive_dir = self.defaults["official_archive_dir"] archive_dir = self.defaults["official_archive_dir"]
live_dir = self.defaults["live_dir"] live_dir = self.defaults["live_dir"]
result = renewer.RenewableCert.new_lineage("the-lineage.com", "cert", result = storage.RenewableCert.new_lineage("the-lineage.com", "cert",
"privkey", "chain", None, "privkey", "chain", None,
None, self.defaults) self.defaults)
# This consistency check tests most relevant properties about the # This consistency check tests most relevant properties about the
# newly created cert lineage. # newly created cert lineage.
self.assertTrue(result.consistent()) self.assertTrue(result.consistent())
@@ -543,33 +546,34 @@ class RenewableCertTests(unittest.TestCase):
with open(result.fullchain) as f: with open(result.fullchain) as f:
self.assertEqual(f.read(), "cert" + "chain") self.assertEqual(f.read(), "cert" + "chain")
# Let's do it again and make sure it makes a different lineage # Let's do it again and make sure it makes a different lineage
result = renewer.RenewableCert.new_lineage("the-lineage.com", "cert2", result = storage.RenewableCert.new_lineage("the-lineage.com", "cert2",
"privkey2", "chain2", None, "privkey2", "chain2", None,
None, self.defaults) self.defaults)
self.assertTrue(os.path.exists( self.assertTrue(os.path.exists(
os.path.join(config_dir, "the-lineage.com-0001.conf"))) os.path.join(config_dir, "the-lineage.com-0001.conf")))
# Now trigger the detection of already existing files # Now trigger the detection of already existing files
os.mkdir(os.path.join(live_dir, "the-lineage.com-0002")) os.mkdir(os.path.join(live_dir, "the-lineage.com-0002"))
self.assertRaises(ValueError, renewer.RenewableCert.new_lineage, self.assertRaises(ValueError, storage.RenewableCert.new_lineage,
"the-lineage.com", "cert3", "privkey3", "chain3", "the-lineage.com", "cert3", "privkey3", "chain3",
None, None, self.defaults) None, self.defaults)
os.mkdir(os.path.join(archive_dir, "other-example.com")) os.mkdir(os.path.join(archive_dir, "other-example.com"))
self.assertRaises(ValueError, renewer.RenewableCert.new_lineage, self.assertRaises(ValueError, storage.RenewableCert.new_lineage,
"other-example.com", "cert4", "privkey4", "chain4", "other-example.com", "cert4", "privkey4", "chain4",
None, None, self.defaults) None, self.defaults)
def test_new_lineage_nonexistent_dirs(self): def test_new_lineage_nonexistent_dirs(self):
"""Test that directories can be created if they don't exist.""" """Test that directories can be created if they don't exist."""
from letsencrypt.client import renewer from letsencrypt.client import renewer
from letsencrypt.client import storage
config_dir = self.defaults["renewal_configs_dir"] config_dir = self.defaults["renewal_configs_dir"]
archive_dir = self.defaults["official_archive_dir"] archive_dir = self.defaults["official_archive_dir"]
live_dir = self.defaults["live_dir"] live_dir = self.defaults["live_dir"]
shutil.rmtree(config_dir) shutil.rmtree(config_dir)
shutil.rmtree(archive_dir) shutil.rmtree(archive_dir)
shutil.rmtree(live_dir) shutil.rmtree(live_dir)
result = renewer.RenewableCert.new_lineage("the-lineage.com", "cert2", result = storage.RenewableCert.new_lineage("the-lineage.com", "cert2",
"privkey2", "chain2", "privkey2", "chain2",
None, None, self.defaults) None, self.defaults)
self.assertTrue(os.path.exists( self.assertTrue(os.path.exists(
os.path.join(config_dir, "the-lineage.com.conf"))) os.path.join(config_dir, "the-lineage.com.conf")))
self.assertTrue(os.path.exists( self.assertTrue(os.path.exists(
@@ -580,10 +584,11 @@ class RenewableCertTests(unittest.TestCase):
@mock.patch("letsencrypt.client.renewer.le_util.unique_lineage_name") @mock.patch("letsencrypt.client.renewer.le_util.unique_lineage_name")
def test_invalid_config_filename(self, mock_uln): def test_invalid_config_filename(self, mock_uln):
from letsencrypt.client import renewer from letsencrypt.client import renewer
from letsencrypt.client import storage
mock_uln.return_value = "this_does_not_end_with_dot_conf", "yikes" mock_uln.return_value = "this_does_not_end_with_dot_conf", "yikes"
self.assertRaises(ValueError, renewer.RenewableCert.new_lineage, self.assertRaises(ValueError, storage.RenewableCert.new_lineage,
"example.com", "cert", "privkey", "chain", "example.com", "cert", "privkey", "chain",
None, None, self.defaults) None, self.defaults)
def test_bad_kind(self): def test_bad_kind(self):
self.assertRaises(ValueError, self.test_rc.current_target, "elephant") self.assertRaises(ValueError, self.test_rc.current_target, "elephant")
@@ -602,33 +607,33 @@ class RenewableCertTests(unittest.TestCase):
self.assertEqual(self.test_rc.ocsp_revoked(), False) self.assertEqual(self.test_rc.ocsp_revoked(), False)
def test_parse_time_interval(self): def test_parse_time_interval(self):
from letsencrypt.client import renewer from letsencrypt.client import storage
# XXX: I'm not sure if intervals related to years and months # XXX: I'm not sure if intervals related to years and months
# take account of the current date (if so, some of these # take account of the current date (if so, some of these
# may fail in the future, like in leap years or even in # may fail in the future, like in leap years or even in
# months of different lengths!) # months of different lengths!)
self.assertEqual(renewer.parse_time_interval(""), self.assertEqual(storage.parse_time_interval(""),
datetime.timedelta(0)) datetime.timedelta(0))
self.assertEqual(renewer.parse_time_interval("1 hour"), self.assertEqual(storage.parse_time_interval("1 hour"),
datetime.timedelta(0, 3600)) datetime.timedelta(0, 3600))
self.assertEqual(renewer.parse_time_interval("17 days"), self.assertEqual(storage.parse_time_interval("17 days"),
datetime.timedelta(17)) datetime.timedelta(17))
# Days are assumed if no unit is specified. # Days are assumed if no unit is specified.
self.assertEqual(renewer.parse_time_interval("23"), self.assertEqual(storage.parse_time_interval("23"),
datetime.timedelta(23)) datetime.timedelta(23))
self.assertEqual(renewer.parse_time_interval("1 month"), self.assertEqual(storage.parse_time_interval("1 month"),
datetime.timedelta(31)) datetime.timedelta(31))
self.assertEqual(renewer.parse_time_interval("7 weeks"), self.assertEqual(storage.parse_time_interval("7 weeks"),
datetime.timedelta(49)) datetime.timedelta(49))
self.assertEqual(renewer.parse_time_interval("1 year 1 day"), self.assertEqual(storage.parse_time_interval("1 year 1 day"),
datetime.timedelta(366)) datetime.timedelta(366))
self.assertEqual(renewer.parse_time_interval("1 year-1 day"), self.assertEqual(storage.parse_time_interval("1 year-1 day"),
datetime.timedelta(364)) datetime.timedelta(364))
self.assertEqual(renewer.parse_time_interval("4 years"), self.assertEqual(storage.parse_time_interval("4 years"),
datetime.timedelta(1461)) datetime.timedelta(1461))
@mock.patch("letsencrypt.client.renewer.notify") @mock.patch("letsencrypt.client.renewer.notify")
@mock.patch("letsencrypt.client.renewer.RenewableCert") @mock.patch("letsencrypt.client.storage.RenewableCert")
@mock.patch("letsencrypt.client.renewer.renew") @mock.patch("letsencrypt.client.renewer.renew")
def test_main(self, mock_renew, mock_rc, mock_notify): def test_main(self, mock_renew, mock_rc, mock_notify):
"""Test for main() function.""" """Test for main() function."""