mirror of
https://github.com/certbot/certbot.git
synced 2026-07-28 00:35:50 +02:00
Merge branch 'candidate-0.1.1' into release-engineering
This commit is contained in:
+3
-1
@@ -20,7 +20,9 @@ from acme import messages
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Python does not validate certificates by default before version 2.7.9
|
||||
# Prior to Python 2.7.9 the stdlib SSL module did not allow a user to configure
|
||||
# many important security related options. On these platforms we use PyOpenSSL
|
||||
# for SSL, which does allow these options to be configured.
|
||||
# https://urllib3.readthedocs.org/en/latest/security.html#insecureplatformwarning
|
||||
if sys.version_info < (2, 7, 9): # pragma: no cover
|
||||
requests.packages.urllib3.contrib.pyopenssl.inject_into_urllib3()
|
||||
|
||||
@@ -28,8 +28,7 @@ acme = client.Client(DIRECTORY_URL, key)
|
||||
|
||||
regr = acme.register()
|
||||
logging.info('Auto-accepting TOS: %s', regr.terms_of_service)
|
||||
acme.update_registration(regr.update(
|
||||
body=regr.body.update(agreement=regr.terms_of_service)))
|
||||
acme.agree_to_tos(regr)
|
||||
logging.debug(regr)
|
||||
|
||||
authzr = acme.request_challenges(
|
||||
|
||||
+166
-65
@@ -205,76 +205,131 @@ def _find_duplicative_certs(config, domains):
|
||||
if candidate_names == set(domains):
|
||||
identical_names_cert = candidate_lineage
|
||||
elif candidate_names.issubset(set(domains)):
|
||||
subset_names_cert = candidate_lineage
|
||||
# This logic finds and returns the largest subset-names cert
|
||||
# in the case where there are several available.
|
||||
if subset_names_cert is None:
|
||||
subset_names_cert = candidate_lineage
|
||||
elif len(candidate_names) > len(subset_names_cert.names()):
|
||||
subset_names_cert = candidate_lineage
|
||||
|
||||
return identical_names_cert, subset_names_cert
|
||||
|
||||
|
||||
def _treat_as_renewal(config, domains):
|
||||
"""Determine whether or not the call should be treated as a renewal.
|
||||
"""Determine whether there are duplicated names and how to handle them.
|
||||
|
||||
:returns: RenewableCert or None if renewal shouldn't occur.
|
||||
:rtype: :class:`.storage.RenewableCert`
|
||||
:returns: Two-element tuple containing desired new-certificate behavior as
|
||||
a string token ("reinstall", "renew", or "newcert"), plus either
|
||||
a RenewableCert instance or None if renewal shouldn't occur.
|
||||
|
||||
:raises .Error: If the user would like to rerun the client again.
|
||||
|
||||
"""
|
||||
renewal = False
|
||||
|
||||
# Considering the possibility that the requested certificate is
|
||||
# related to an existing certificate. (config.duplicate, which
|
||||
# is set with --duplicate, skips all of this logic and forces any
|
||||
# kind of certificate to be obtained with renewal = False.)
|
||||
if not config.duplicate:
|
||||
ident_names_cert, subset_names_cert = _find_duplicative_certs(
|
||||
config, domains)
|
||||
# I am not sure whether that correctly reads the systemwide
|
||||
# configuration file.
|
||||
question = None
|
||||
if ident_names_cert is not None:
|
||||
question = (
|
||||
"You have an existing certificate that contains exactly the "
|
||||
"same domains you requested (ref: {0}){br}{br}Do you want to "
|
||||
"renew and replace this certificate with a newly-issued one?"
|
||||
).format(ident_names_cert.configfile.filename, br=os.linesep)
|
||||
elif subset_names_cert is not None:
|
||||
question = (
|
||||
"You have an existing certificate that contains a portion of "
|
||||
"the domains you requested (ref: {0}){br}{br}It contains these "
|
||||
"names: {1}{br}{br}You requested these names for the new "
|
||||
"certificate: {2}.{br}{br}Do you want to replace this existing "
|
||||
"certificate with the new certificate?"
|
||||
).format(subset_names_cert.configfile.filename,
|
||||
", ".join(subset_names_cert.names()),
|
||||
", ".join(domains),
|
||||
br=os.linesep)
|
||||
if question is None:
|
||||
# We aren't in a duplicative-names situation at all, so we don't
|
||||
# have to tell or ask the user anything about this.
|
||||
pass
|
||||
elif config.renew_by_default or zope.component.getUtility(
|
||||
interfaces.IDisplay).yesno(question, "Replace", "Cancel"):
|
||||
renewal = True
|
||||
else:
|
||||
reporter_util = zope.component.getUtility(interfaces.IReporter)
|
||||
reporter_util.add_message(
|
||||
"To obtain a new certificate that {0} an existing certificate "
|
||||
"in its domain-name coverage, you must use the --duplicate "
|
||||
"option.{br}{br}For example:{br}{br}{1} --duplicate {2}".format(
|
||||
"duplicates" if ident_names_cert is not None else
|
||||
"overlaps with",
|
||||
sys.argv[0], " ".join(sys.argv[1:]),
|
||||
br=os.linesep
|
||||
),
|
||||
reporter_util.HIGH_PRIORITY)
|
||||
raise errors.Error(
|
||||
"User did not use proper CLI and would like "
|
||||
"to reinvoke the client.")
|
||||
if config.duplicate:
|
||||
return "newcert", None
|
||||
# TODO: Also address superset case
|
||||
ident_names_cert, subset_names_cert = _find_duplicative_certs(config, domains)
|
||||
# XXX ^ schoen is not sure whether that correctly reads the systemwide
|
||||
# configuration file.
|
||||
if ident_names_cert is None and subset_names_cert is None:
|
||||
return "newcert", None
|
||||
|
||||
if renewal:
|
||||
return ident_names_cert if ident_names_cert is not None else subset_names_cert
|
||||
if ident_names_cert is not None:
|
||||
return _handle_identical_cert_request(config, ident_names_cert)
|
||||
elif subset_names_cert is not None:
|
||||
return _handle_subset_cert_request(config, domains, subset_names_cert)
|
||||
|
||||
return None
|
||||
def _handle_identical_cert_request(config, cert):
|
||||
"""Figure out what to do if a cert has the same names as a perviously obtained one
|
||||
|
||||
:param storage.RenewableCert cert:
|
||||
|
||||
:returns: Tuple of (string, cert_or_None) as per _treat_as_renewal
|
||||
:rtype: tuple
|
||||
|
||||
"""
|
||||
if config.renew_by_default:
|
||||
logger.info("Auto-renewal forced with --renew-by-default...")
|
||||
return "renew", cert
|
||||
if cert.should_autorenew(interactive=True):
|
||||
logger.info("Cert is due for renewal, auto-renewing...")
|
||||
return "renew", cert
|
||||
if config.reinstall:
|
||||
# Set with --reinstall, force an identical certificate to be
|
||||
# reinstalled without further prompting.
|
||||
return "reinstall", cert
|
||||
|
||||
question = (
|
||||
"You have an existing certificate that contains exactly the same "
|
||||
"domains you requested and isn't close to expiry."
|
||||
"{br}(ref: {0}){br}{br}What would you like to do?"
|
||||
).format(cert.configfile.filename, br=os.linesep)
|
||||
|
||||
if config.verb == "run":
|
||||
keep_opt = "Attempt to reinstall this existing certificate"
|
||||
elif config.verb == "certonly":
|
||||
keep_opt = "Keep the existing certificate for now"
|
||||
choices = [keep_opt,
|
||||
"Renew & replace the cert (limit ~5 per 7 days)",
|
||||
"Cancel this operation and do nothing"]
|
||||
|
||||
display = zope.component.getUtility(interfaces.IDisplay)
|
||||
response = display.menu(question, choices, "OK", "Cancel")
|
||||
if response[0] == "cancel" or response[1] == 2:
|
||||
# TODO: Add notification related to command-line options for
|
||||
# skipping the menu for this case.
|
||||
raise errors.Error(
|
||||
"User chose to cancel the operation and may "
|
||||
"reinvoke the client.")
|
||||
elif response[1] == 0:
|
||||
return "reinstall", cert
|
||||
elif response[1] == 1:
|
||||
return "renew", cert
|
||||
else:
|
||||
assert False, "This is impossible"
|
||||
|
||||
def _handle_subset_cert_request(config, domains, cert):
|
||||
"""Figure out what to do if a previous cert had a subset of the names now requested
|
||||
|
||||
:param storage.RenewableCert cert:
|
||||
|
||||
:returns: Tuple of (string, cert_or_None) as per _treat_as_renewal
|
||||
:rtype: tuple
|
||||
|
||||
"""
|
||||
existing = ", ".join(cert.names())
|
||||
question = (
|
||||
"You have an existing certificate that contains a portion of "
|
||||
"the domains you requested (ref: {0}){br}{br}It contains these "
|
||||
"names: {1}{br}{br}You requested these names for the new "
|
||||
"certificate: {2}.{br}{br}Do you want to expand and replace this existing "
|
||||
"certificate with the new certificate?"
|
||||
).format(cert.configfile.filename,
|
||||
existing,
|
||||
", ".join(domains),
|
||||
br=os.linesep)
|
||||
if config.expand or config.renew_by_default or zope.component.getUtility(
|
||||
interfaces.IDisplay).yesno(question, "Expand", "Cancel"):
|
||||
return "renew", cert
|
||||
else:
|
||||
reporter_util = zope.component.getUtility(interfaces.IReporter)
|
||||
reporter_util.add_message(
|
||||
"To obtain a new certificate that contains these names without "
|
||||
"replacing your existing certificate for {0}, you must use the "
|
||||
"--duplicate option.{br}{br}"
|
||||
"For example:{br}{br}{1} --duplicate {2}".format(
|
||||
existing,
|
||||
sys.argv[0], " ".join(sys.argv[1:]),
|
||||
br=os.linesep
|
||||
),
|
||||
reporter_util.HIGH_PRIORITY)
|
||||
raise errors.Error(
|
||||
"User chose to cancel the operation and may "
|
||||
"reinvoke the client.")
|
||||
|
||||
|
||||
def _report_new_cert(cert_path, fullchain_path):
|
||||
@@ -314,10 +369,21 @@ def _suggest_donate():
|
||||
|
||||
def _auth_from_domains(le_client, config, domains):
|
||||
"""Authenticate and enroll certificate."""
|
||||
# Note: This can raise errors... caught above us though.
|
||||
lineage = _treat_as_renewal(config, domains)
|
||||
# Note: This can raise errors... caught above us though. This is now
|
||||
# a three-way case: reinstall (which results in a no-op here because
|
||||
# although there is a relevant lineage, we don't do anything to it
|
||||
# inside this function -- we don't obtain a new certificate), renew
|
||||
# (which results in treating the request as a renewal), or newcert
|
||||
# (which results in treating the request as a new certificate request).
|
||||
|
||||
if lineage is not None:
|
||||
action, lineage = _treat_as_renewal(config, domains)
|
||||
if action == "reinstall":
|
||||
# The lineage already exists; allow the caller to try installing
|
||||
# it without getting a new certificate at all.
|
||||
return lineage
|
||||
elif action == "renew":
|
||||
original_server = lineage.configuration["renewalparams"]["server"]
|
||||
_avoid_invalidating_lineage(config, lineage, original_server)
|
||||
# TODO: schoen wishes to reuse key - discussion
|
||||
# https://github.com/letsencrypt/letsencrypt/pull/777/files#r40498574
|
||||
new_certr, new_chain, new_key, _ = le_client.obtain_certificate(domains)
|
||||
@@ -331,7 +397,7 @@ def _auth_from_domains(le_client, config, domains):
|
||||
# TODO: Check return value of save_successor
|
||||
# TODO: Also update lineage renewal config with any relevant
|
||||
# configuration values from this attempt? <- Absolutely (jdkasten)
|
||||
else:
|
||||
elif action == "newcert":
|
||||
# TREAT AS NEW REQUEST
|
||||
lineage = le_client.obtain_and_enroll_certificate(domains)
|
||||
if not lineage:
|
||||
@@ -341,6 +407,27 @@ def _auth_from_domains(le_client, config, domains):
|
||||
|
||||
return lineage
|
||||
|
||||
def _avoid_invalidating_lineage(config, lineage, original_server):
|
||||
"Do not renew a valid cert with one from a staging server!"
|
||||
def _is_staging(srv):
|
||||
return srv == constants.STAGING_URI or "staging" in srv
|
||||
|
||||
# Some lineages may have begun with --staging, but then had production certs
|
||||
# added to them
|
||||
latest_cert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM,
|
||||
open(lineage.cert).read())
|
||||
# all our test certs are from happy hacker fake CA, though maybe one day
|
||||
# we should test more methodically
|
||||
now_valid = not "fake" in repr(latest_cert.get_issuer()).lower()
|
||||
|
||||
if _is_staging(config.server):
|
||||
if not _is_staging(original_server) or now_valid:
|
||||
if not config.break_my_certs:
|
||||
names = ", ".join(lineage.names())
|
||||
raise errors.Error(
|
||||
"You've asked to renew/replace a seemingly valid certificate with "
|
||||
"a test certificate (domains: {0}). We will not do that "
|
||||
"unless you use the --break-my-certs flag!".format(names))
|
||||
|
||||
def set_configurator(previously, now):
|
||||
"""
|
||||
@@ -485,7 +572,6 @@ def run(args, config, plugins): # pylint: disable=too-many-branches,too-many-lo
|
||||
|
||||
def obtain_cert(args, config, plugins):
|
||||
"""Authenticate & obtain cert, but do not install it."""
|
||||
|
||||
if args.domains and args.csr is not None:
|
||||
# TODO: --csr could have a priority, when --domains is
|
||||
# supplied, check if CSR matches given domains?
|
||||
@@ -695,6 +781,7 @@ class HelpfulArgumentParser(object):
|
||||
"""
|
||||
parsed_args = self.parser.parse_args(self.args)
|
||||
parsed_args.func = self.VERBS[self.verb]
|
||||
parsed_args.verb = self.verb
|
||||
|
||||
# Do any post-parsing homework here
|
||||
|
||||
@@ -877,13 +964,19 @@ def prepare_and_parse_args(plugins, args):
|
||||
help="Domain names to apply. For multiple domains you can use "
|
||||
"multiple -d flags or enter a comma separated list of domains "
|
||||
"as a parameter.")
|
||||
helpful.add(
|
||||
None, "--duplicate", dest="duplicate", action="store_true",
|
||||
help="Allow getting a certificate that duplicates an existing one")
|
||||
|
||||
helpful.add_group(
|
||||
"automation",
|
||||
description="Arguments for automating execution & other tweaks")
|
||||
helpful.add(
|
||||
"automation", "--keep-until-expiring", "--keep", "--reinstall",
|
||||
dest="reinstall", action="store_true",
|
||||
help="If the requested cert matches an existing cert, always keep the "
|
||||
"existing one until it is due for renewal (for the "
|
||||
"'run' subcommand this means reinstall the existing cert)")
|
||||
helpful.add(
|
||||
"automation", "--expand", action="store_true",
|
||||
help="If an existing cert covers some subset of the requested names, "
|
||||
"always expand and replace it with the additional names.")
|
||||
helpful.add(
|
||||
"automation", "--version", action="version",
|
||||
version="%(prog)s {0}".format(letsencrypt.__version__),
|
||||
@@ -891,13 +984,18 @@ def prepare_and_parse_args(plugins, args):
|
||||
helpful.add(
|
||||
"automation", "--renew-by-default", "--replace", action="store_true",
|
||||
help="Select renewal by default when domains are a superset of a "
|
||||
"previously attained cert")
|
||||
"previously attained cert (often --keep-until-expiring is "
|
||||
"more appropriate). Implies --expand.")
|
||||
helpful.add(
|
||||
"automation", "--agree-tos", dest="tos", action="store_true",
|
||||
help="Agree to the Let's Encrypt Subscriber Agreement")
|
||||
helpful.add(
|
||||
"automation", "--account", metavar="ACCOUNT_ID",
|
||||
help="Account ID to use")
|
||||
helpful.add(
|
||||
"automation", "--duplicate", dest="duplicate", action="store_true",
|
||||
help="Allow making a certificate lineage that duplicates an existing one "
|
||||
"(both can be renewed in parallel)")
|
||||
|
||||
helpful.add_group(
|
||||
"testing", description="The following flags are meant for "
|
||||
@@ -918,7 +1016,10 @@ def prepare_and_parse_args(plugins, args):
|
||||
helpful.add(
|
||||
"testing", "--http-01-port", type=int, dest="http01_port",
|
||||
default=flag_default("http01_port"), help=config_help("http01_port"))
|
||||
|
||||
helpful.add(
|
||||
"testing", "--break-my-certs", action="store_true",
|
||||
help="Be willing to replace or renew valid certs with invalid "
|
||||
"(testing/staging) certs")
|
||||
helpful.add_group(
|
||||
"security", description="Security parameters & server settings")
|
||||
helpful.add(
|
||||
|
||||
@@ -2,7 +2,6 @@
|
||||
import argparse
|
||||
import collections
|
||||
import logging
|
||||
import random
|
||||
import socket
|
||||
import threading
|
||||
|
||||
@@ -108,7 +107,7 @@ class ServerManager(object):
|
||||
in six.iteritems(self._instances))
|
||||
|
||||
|
||||
SUPPORTED_CHALLENGES = set([challenges.TLSSNI01, challenges.HTTP01])
|
||||
SUPPORTED_CHALLENGES = [challenges.TLSSNI01, challenges.HTTP01]
|
||||
|
||||
|
||||
def supported_challenges_validator(data):
|
||||
@@ -166,16 +165,16 @@ class Authenticator(common.Plugin):
|
||||
|
||||
@classmethod
|
||||
def add_parser_arguments(cls, add):
|
||||
add("supported-challenges", help="Supported challenges, "
|
||||
"order preferences are randomly chosen.",
|
||||
type=supported_challenges_validator, default=",".join(
|
||||
sorted(chall.typ for chall in SUPPORTED_CHALLENGES)))
|
||||
add("supported-challenges",
|
||||
help="Supported challenges. Preferred in the order they are listed.",
|
||||
type=supported_challenges_validator,
|
||||
default=",".join(chall.typ for chall in SUPPORTED_CHALLENGES))
|
||||
|
||||
@property
|
||||
def supported_challenges(self):
|
||||
"""Challenges supported by this plugin."""
|
||||
return set(challenges.Challenge.TYPES[name] for name in
|
||||
self.conf("supported-challenges").split(","))
|
||||
return [challenges.Challenge.TYPES[name] for name in
|
||||
self.conf("supported-challenges").split(",")]
|
||||
|
||||
@property
|
||||
def _necessary_ports(self):
|
||||
@@ -198,9 +197,7 @@ class Authenticator(common.Plugin):
|
||||
|
||||
def get_chall_pref(self, domain):
|
||||
# pylint: disable=unused-argument,missing-docstring
|
||||
chall_pref = list(self.supported_challenges)
|
||||
random.shuffle(chall_pref) # 50% for each challenge
|
||||
return chall_pref
|
||||
return self.supported_challenges
|
||||
|
||||
def perform(self, achalls): # pylint: disable=missing-docstring
|
||||
if any(util.already_listening(port) for port in self._necessary_ports):
|
||||
|
||||
@@ -98,17 +98,27 @@ class AuthenticatorTest(unittest.TestCase):
|
||||
|
||||
def test_supported_challenges(self):
|
||||
self.assertEqual(self.auth.supported_challenges,
|
||||
set([challenges.TLSSNI01, challenges.HTTP01]))
|
||||
[challenges.TLSSNI01, challenges.HTTP01])
|
||||
|
||||
def test_supported_challenges_configured(self):
|
||||
self.config.standalone_supported_challenges = "tls-sni-01"
|
||||
self.assertEqual(self.auth.supported_challenges,
|
||||
[challenges.TLSSNI01])
|
||||
|
||||
def test_more_info(self):
|
||||
self.assertTrue(isinstance(self.auth.more_info(), six.string_types))
|
||||
|
||||
def test_get_chall_pref(self):
|
||||
self.assertEqual(set(self.auth.get_chall_pref(domain=None)),
|
||||
set([challenges.TLSSNI01, challenges.HTTP01]))
|
||||
self.assertEqual(self.auth.get_chall_pref(domain=None),
|
||||
[challenges.TLSSNI01, challenges.HTTP01])
|
||||
|
||||
def test_get_chall_pref_configured(self):
|
||||
self.config.standalone_supported_challenges = "tls-sni-01"
|
||||
self.assertEqual(self.auth.get_chall_pref(domain=None),
|
||||
[challenges.TLSSNI01])
|
||||
|
||||
@mock.patch("letsencrypt.plugins.standalone.util")
|
||||
def test_perform_alredy_listening(self, mock_util):
|
||||
def test_perform_already_listening(self, mock_util):
|
||||
for chall, port in ((challenges.TLSSNI01.typ, 1234),
|
||||
(challenges.HTTP01.typ, 4321)):
|
||||
mock_util.already_listening.return_value = True
|
||||
|
||||
@@ -2,7 +2,6 @@
|
||||
import errno
|
||||
import logging
|
||||
import os
|
||||
import stat
|
||||
|
||||
import zope.interface
|
||||
|
||||
@@ -59,24 +58,38 @@ to serve all files under specified web root ({0})."""
|
||||
|
||||
logger.debug("Creating root challenges validation dir at %s",
|
||||
self.full_roots[name])
|
||||
|
||||
# Change the permissions to be writable (GH #1389)
|
||||
# Umask is used instead of chmod to ensure the client can also
|
||||
# run as non-root (GH #1795)
|
||||
old_umask = os.umask(0o022)
|
||||
|
||||
try:
|
||||
os.makedirs(self.full_roots[name])
|
||||
# Set permissions as parent directory (GH #1389)
|
||||
# We don't use the parameters in makedirs because it
|
||||
# may not always work
|
||||
# This is coupled with the "umask" call above because
|
||||
# os.makedirs's "mode" parameter may not always work:
|
||||
# https://stackoverflow.com/questions/5231901/permission-problems-when-creating-a-dir-with-os-makedirs-python
|
||||
stat_path = os.stat(path)
|
||||
filemode = stat.S_IMODE(stat_path.st_mode)
|
||||
os.chmod(self.full_roots[name], filemode)
|
||||
# Set owner and group, too
|
||||
os.chown(self.full_roots[name], stat_path.st_uid,
|
||||
stat_path.st_gid)
|
||||
os.makedirs(self.full_roots[name], 0o0755)
|
||||
|
||||
# Set owner as parent directory if possible
|
||||
try:
|
||||
stat_path = os.stat(path)
|
||||
os.chown(self.full_roots[name], stat_path.st_uid,
|
||||
stat_path.st_gid)
|
||||
except OSError as exception:
|
||||
if exception.errno == errno.EACCES:
|
||||
logger.debug("Insufficient permissions to change owner and uid - ignoring")
|
||||
else:
|
||||
raise errors.PluginError(
|
||||
"Couldn't create root for {0} http-01 "
|
||||
"challenge responses: {1}", name, exception)
|
||||
|
||||
except OSError as exception:
|
||||
if exception.errno != errno.EEXIST:
|
||||
raise errors.PluginError(
|
||||
"Couldn't create root for {0} http-01 "
|
||||
"challenge responses: {1}", name, exception)
|
||||
finally:
|
||||
os.umask(old_umask)
|
||||
|
||||
def perform(self, achalls): # pylint: disable=missing-docstring
|
||||
assert self.full_roots, "Webroot plugin appears to be missing webroot map"
|
||||
@@ -87,26 +100,26 @@ to serve all files under specified web root ({0})."""
|
||||
path = self.full_roots[achall.domain]
|
||||
except IndexError:
|
||||
raise errors.PluginError("Missing --webroot-path for domain: {1}"
|
||||
.format(achall.domain))
|
||||
.format(achall.domain))
|
||||
if not os.path.exists(path):
|
||||
raise errors.PluginError("Mysteriously missing path {0} for domain: {1}"
|
||||
.format(path, achall.domain))
|
||||
.format(path, achall.domain))
|
||||
return os.path.join(path, achall.chall.encode("token"))
|
||||
|
||||
def _perform_single(self, achall):
|
||||
response, validation = achall.response_and_validation()
|
||||
|
||||
path = self._path_for_achall(achall)
|
||||
logger.debug("Attempting to save validation to %s", path)
|
||||
with open(path, "w") as validation_file:
|
||||
validation_file.write(validation.encode())
|
||||
|
||||
# Set permissions as parent directory (GH #1389)
|
||||
parent_path = self.full_roots[achall.domain]
|
||||
stat_parent_path = os.stat(parent_path)
|
||||
filemode = stat.S_IMODE(stat_parent_path.st_mode)
|
||||
# Remove execution bit (not needed for this file)
|
||||
os.chmod(path, filemode & ~stat.S_IEXEC)
|
||||
os.chown(path, stat_parent_path.st_uid, stat_parent_path.st_gid)
|
||||
# Change permissions to be world-readable, owner-writable (GH #1795)
|
||||
old_umask = os.umask(0o022)
|
||||
|
||||
try:
|
||||
with open(path, "w") as validation_file:
|
||||
validation_file.write(validation.encode())
|
||||
finally:
|
||||
os.umask(old_umask)
|
||||
|
||||
return response
|
||||
|
||||
|
||||
@@ -1,9 +1,10 @@
|
||||
"""Tests for letsencrypt.plugins.webroot."""
|
||||
import errno
|
||||
import os
|
||||
import shutil
|
||||
import stat
|
||||
import tempfile
|
||||
import unittest
|
||||
import stat
|
||||
|
||||
import mock
|
||||
|
||||
@@ -35,7 +36,6 @@ class AuthenticatorTest(unittest.TestCase):
|
||||
self.config = mock.MagicMock(webroot_path=self.path,
|
||||
webroot_map={"thing.com": self.path})
|
||||
self.auth = Authenticator(self.config, "webroot")
|
||||
self.auth.prepare()
|
||||
|
||||
def tearDown(self):
|
||||
shutil.rmtree(self.path)
|
||||
@@ -48,7 +48,7 @@ class AuthenticatorTest(unittest.TestCase):
|
||||
def test_add_parser_arguments(self):
|
||||
add = mock.MagicMock()
|
||||
self.auth.add_parser_arguments(add)
|
||||
self.assertEqual(0, add.call_count) # became 0 when we moved the args to cli.py!
|
||||
self.assertEqual(0, add.call_count) # args moved to cli.py!
|
||||
|
||||
def test_prepare_bad_root(self):
|
||||
self.config.webroot_path = os.path.join(self.path, "null")
|
||||
@@ -70,17 +70,33 @@ class AuthenticatorTest(unittest.TestCase):
|
||||
self.assertRaises(errors.PluginError, self.auth.prepare)
|
||||
os.chmod(self.path, 0o700)
|
||||
|
||||
@mock.patch("letsencrypt.plugins.webroot.os.chown")
|
||||
def test_failed_chown_eacces(self, mock_chown):
|
||||
mock_chown.side_effect = OSError(errno.EACCES, "msg")
|
||||
self.auth.prepare() # exception caught and logged
|
||||
|
||||
@mock.patch("letsencrypt.plugins.webroot.os.chown")
|
||||
def test_failed_chown_not_eacces(self, mock_chown):
|
||||
mock_chown.side_effect = OSError()
|
||||
self.assertRaises(errors.PluginError, self.auth.prepare)
|
||||
|
||||
def test_prepare_permissions(self):
|
||||
self.auth.prepare()
|
||||
|
||||
# Remove exec bit from permission check, so that it
|
||||
# matches the file
|
||||
self.auth.perform([self.achall])
|
||||
parent_permissions = (stat.S_IMODE(os.stat(self.path).st_mode) &
|
||||
~stat.S_IEXEC)
|
||||
path_permissions = stat.S_IMODE(os.stat(self.validation_path).st_mode)
|
||||
self.assertEqual(path_permissions, 0o644)
|
||||
|
||||
actual_permissions = stat.S_IMODE(os.stat(self.validation_path).st_mode)
|
||||
# Check permissions of the directories
|
||||
|
||||
for dirpath, dirnames, _ in os.walk(self.path):
|
||||
for directory in dirnames:
|
||||
full_path = os.path.join(dirpath, directory)
|
||||
dir_permissions = stat.S_IMODE(os.stat(full_path).st_mode)
|
||||
self.assertEqual(dir_permissions, 0o755)
|
||||
|
||||
self.assertEqual(parent_permissions, actual_permissions)
|
||||
parent_gid = os.stat(self.path).st_gid
|
||||
parent_uid = os.stat(self.path).st_uid
|
||||
|
||||
@@ -88,6 +104,7 @@ class AuthenticatorTest(unittest.TestCase):
|
||||
self.assertEqual(os.stat(self.validation_path).st_uid, parent_uid)
|
||||
|
||||
def test_perform_cleanup(self):
|
||||
self.auth.prepare()
|
||||
responses = self.auth.perform([self.achall])
|
||||
self.assertEqual(1, len(responses))
|
||||
self.assertTrue(os.path.exists(self.validation_path))
|
||||
|
||||
+14
-6
@@ -471,7 +471,7 @@ class RenewableCert(object): # pylint: disable=too-many-instance-attributes
|
||||
return ("autodeploy" not in self.configuration or
|
||||
self.configuration.as_bool("autodeploy"))
|
||||
|
||||
def should_autodeploy(self):
|
||||
def should_autodeploy(self, interactive=False):
|
||||
"""Should this lineage now automatically deploy a newer version?
|
||||
|
||||
This is a policy question and does not only depend on whether
|
||||
@@ -480,12 +480,16 @@ class RenewableCert(object): # pylint: disable=too-many-instance-attributes
|
||||
exists, and whether the time interval for autodeployment has
|
||||
been reached.)
|
||||
|
||||
:param bool interactive: set to True to examine the question
|
||||
regardless of whether the renewal configuration allows
|
||||
automated deployment (for interactive use). Default False.
|
||||
|
||||
:returns: whether the lineage now ought to autodeploy an
|
||||
existing newer cert version
|
||||
:rtype: bool
|
||||
|
||||
"""
|
||||
if self.autodeployment_is_enabled():
|
||||
if interactive or self.autodeployment_is_enabled():
|
||||
if self.has_pending_deployment():
|
||||
interval = self.configuration.get("deploy_before_expiry",
|
||||
"5 days")
|
||||
@@ -529,7 +533,7 @@ class RenewableCert(object): # pylint: disable=too-many-instance-attributes
|
||||
return ("autorenew" not in self.configuration or
|
||||
self.configuration.as_bool("autorenew"))
|
||||
|
||||
def should_autorenew(self):
|
||||
def should_autorenew(self, interactive=False):
|
||||
"""Should we now try to autorenew the most recent cert version?
|
||||
|
||||
This is a policy question and does not only depend on whether
|
||||
@@ -540,12 +544,16 @@ class RenewableCert(object): # pylint: disable=too-many-instance-attributes
|
||||
Note that this examines the numerically most recent cert version,
|
||||
not the currently deployed version.
|
||||
|
||||
:param bool interactive: set to True to examine the question
|
||||
regardless of whether the renewal configuration allows
|
||||
automated renewal (for interactive use). Default False.
|
||||
|
||||
:returns: whether an attempt should now be made to autorenew the
|
||||
most current cert version in this lineage
|
||||
:rtype: bool
|
||||
|
||||
"""
|
||||
if self.autorenewal_is_enabled():
|
||||
if interactive or self.autorenewal_is_enabled():
|
||||
# Consider whether to attempt to autorenew this cert now
|
||||
|
||||
# Renewals on the basis of revocation
|
||||
@@ -559,8 +567,8 @@ class RenewableCert(object): # pylint: disable=too-many-instance-attributes
|
||||
"cert", self.latest_common_version()))
|
||||
now = pytz.UTC.fromutc(datetime.datetime.utcnow())
|
||||
if expiry < add_time_interval(now, interval):
|
||||
logger.debug("Should renew, certificate "
|
||||
"has been expired since %s.",
|
||||
logger.debug("Should renew, less than %s before certificate "
|
||||
"expiry %s.", interval,
|
||||
expiry.strftime("%Y-%m-%d %H:%M:%S %Z"))
|
||||
return True
|
||||
return False
|
||||
|
||||
@@ -403,7 +403,7 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
|
||||
|
||||
def _certonly_new_request_common(self, mock_client):
|
||||
with mock.patch('letsencrypt.cli._treat_as_renewal') as mock_renewal:
|
||||
mock_renewal.return_value = None
|
||||
mock_renewal.return_value = ("newcert", None)
|
||||
with mock.patch('letsencrypt.cli._init_le_client') as mock_init:
|
||||
mock_init.return_value = mock_client
|
||||
self._call(['-d', 'foo.bar', '-a', 'standalone', 'certonly'])
|
||||
@@ -413,13 +413,13 @@ class CLITest(unittest.TestCase): # pylint: disable=too-many-public-methods
|
||||
@mock.patch('letsencrypt.cli._treat_as_renewal')
|
||||
@mock.patch('letsencrypt.cli._init_le_client')
|
||||
def test_certonly_renewal(self, mock_init, mock_renewal, mock_get_utility, _suggest):
|
||||
cert_path = '/etc/letsencrypt/live/foo.bar/cert.pem'
|
||||
cert_path = 'letsencrypt/tests/testdata/cert.pem'
|
||||
chain_path = '/etc/letsencrypt/live/foo.bar/fullchain.pem'
|
||||
|
||||
mock_lineage = mock.MagicMock(cert=cert_path, fullchain=chain_path)
|
||||
mock_cert = mock.MagicMock(body='body')
|
||||
mock_key = mock.MagicMock(pem='pem_key')
|
||||
mock_renewal.return_value = mock_lineage
|
||||
mock_renewal.return_value = ("renew", mock_lineage)
|
||||
mock_client = mock.MagicMock()
|
||||
mock_client.obtain_certificate.return_value = (mock_cert, 'chain',
|
||||
mock_key, 'csr')
|
||||
|
||||
Reference in New Issue
Block a user