starred/certbot-certbot-2
1
0
Fork 0
mirror of https://github.com/certbot/certbot.git synced 2026-07-28 00:24:35 +02:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #185 from letsencrypt/reverter

Browse Source 
Reverter
This commit is contained in:
James Kasten
2015-01-26 17:28:46 -08:00
parent 248407c70e eeb58d9232
commit 8c80d0bedf
19 changed files with 1519 additions and 498 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.pylintrc
+1 -1
View File
@@ -201,7 +201,7 @@ ignore-comments=yes
ignore-docstrings=yes
# Ignore imports when computing similarities.
ignore-imports=no
ignore-imports=yes
[FORMAT]
docs/api/client/reverter.rst
+5
View File
@@ -0,0 +1,5 @@
:mod:`letsencrypt.client.reverter`
----------------------------------
.. automodule:: letsencrypt.client.reverter
:members:
letsencrypt/client/CONFIG.py
+3 -18
View File
@@ -24,7 +24,7 @@ BACKUP_DIR = os.path.join(WORK_DIR, "backups/")
"""Directory where configuration backups are stored"""
TEMP_CHECKPOINT_DIR = os.path.join(WORK_DIR, "temp_checkpoint/")
"""Replaces MODIFIED_FILES, directory where temp checkpoint is created"""
"""Directory where temp checkpoint is created"""
IN_PROGRESS_DIR = os.path.join(BACKUP_DIR, "IN_PROGRESS/")
"""Directory used before a permanent checkpoint is finalized"""
@@ -57,6 +57,7 @@ CHAIN_PATH = CERT_DIR + "chain-letsencrypt.pem"
INVALID_EXT = ".acme.invalid"
"""Invalid Extension"""
# Challenge Sets
EXCLUSIVE_CHALLENGES = [frozenset(["dvsni", "simpleHttps"])]
"""Mutually Exclusive Challenges - only solve 1"""
@@ -89,23 +90,7 @@ ocsp-stapling, TODO
spdy, TODO
"""
# ENHANCEMENTS = [
# {
# "type": "redirect",
# "description": ("Please choose whether HTTPS access is required or "
# "optional."),
# "options": [
# ("Easy", "Allow both HTTP and HTTPS access to thses sites"),
# ("Secure", "Make all requests redirect to secure HTTPS access"),
# ],
# },
# {
# "type": ""
# }
# ]
# Config Optimizations
# Apache Enhancement Arguments
REWRITE_HTTPS_ARGS = [
"^.*$", "https://%{SERVER_NAME}%{REQUEST_URI}", "[L,R=permanent]"]
"""Rewrite rule arguments used for redirections to https vhost"""
letsencrypt/client/apache/configurator.py
+32 -24
View File
@@ -43,6 +43,7 @@ from letsencrypt.client.apache import parser
class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
# pylint: disable=too-many-instance-attributes
"""Apache configurator.
State of Configurator: This code has been tested under Ubuntu 12.04
@@ -95,12 +96,7 @@ class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
"work": CONFIG.WORK_DIR}
super(ApacheConfigurator, self).__init__(direc)
# See if any temporary changes need to be recovered
# This needs to occur before VirtualHost objects are setup...
# because this will change the underlying configuration and potential
# vhosts
self.recovery_routine()
self.direc = direc
# Verify that all directories and files exist with proper permissions
if os.geteuid() == 0:
@@ -184,7 +180,8 @@ class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
self.aug.set(path["cert_chain"][0], cert_chain)
self.save_notes += ("Changed vhost at %s with addresses of %s\n" %
(vhost.filep, vhost.addrs))
(vhost.filep,
", ".join(str(addr) for addr in vhost.addrs)))
self.save_notes += "\tSSLCertificateFile %s\n" % cert
self.save_notes += "\tSSLCertificateKeyFile %s\n" % key
if cert_chain:
@@ -449,7 +446,7 @@ class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
# First register the creation so that it is properly removed if
# configuration is rolled back
self.register_file_creation(False, ssl_fp)
self.reverter.register_file_creation(False, ssl_fp)
try:
orig_file = open(avail_fp, 'r')
@@ -703,7 +700,7 @@ class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
# Register the new file that will be created
# Note: always register the creation before writing to ensure file will
# be removed in case of unexpected program exit
self.register_file_creation(False, redirect_filepath)
self.reverter.register_file_creation(False, redirect_filepath)
# Write out file
with open(redirect_filepath, 'w') as redirect_fd:
@@ -871,7 +868,7 @@ class ApacheConfigurator(augeas_configurator.AugeasConfigurator):
if "/sites-available/" in vhost.filep:
enabled_path = ("%ssites-enabled/%s" %
(self.parser.root, os.path.basename(vhost.filep)))
self.register_file_creation(False, enabled_path)
self.reverter.register_file_creation(False, enabled_path)
os.symlink(vhost.filep, enabled_path)
vhost.enabled = True
logging.info("Enabling available site: %s", vhost.filep)
@@ -1029,10 +1026,7 @@ def enable_mod(mod_name):
subprocess.check_call(["sudo", "a2enmod", mod_name],
stdout=open("/dev/null", 'w'),
stderr=open("/dev/null", 'w'))
# Hopefully this waits for output
subprocess.check_call(["sudo", CONFIG.APACHE2, "restart"],
stdout=open("/dev/null", 'w'),
stderr=open("/dev/null", 'w'))
apache_restart()
except (OSError, subprocess.CalledProcessError) as err:
logging.error("Error enabling mod_%s", mod_name)
logging.error("Exception: %s", err)
@@ -1042,7 +1036,8 @@ def enable_mod(mod_name):
def mod_loaded(module):
"""Checks to see if mod_ssl is loaded
Uses CONFIG.APACHE_CTL to get loaded module list
Uses CONFIG.APACHE_CTL to get loaded module list. This also effectively
serves as a config_test.
:returns: If ssl_module is included and active in Apache
:rtype: bool
@@ -1052,15 +1047,22 @@ def mod_loaded(module):
proc = subprocess.Popen(
[CONFIG.APACHE_CTL, '-M'],
stdout=subprocess.PIPE,
stderr=open("/dev/null", 'w')).communicate()[0]
stderr=subprocess.PIPE)
stdout, stderr = proc.communicate()
except (OSError, ValueError):
logging.error(
"Error accessing %s for loaded modules!", CONFIG.APACHE_CTL)
logging.error("This may be caused by an Apache Configuration Error")
return False
raise errors.LetsEncryptConfiguratorError(
"Error accessing loaded modules")
# Small errors that do not impede
if proc.returncode != 0:
logging.warn("Error in checking loaded module list: %s", stderr)
raise errors.LetsEncryptMisconfigurationError(
"Apache is unable to check whether or not the module is "
"loaded because Apache is misconfigured.")
if module in proc:
if module in stdout:
return True
return False
@@ -1069,20 +1071,26 @@ def apache_restart():
"""Restarts the Apache Server.
.. todo:: Try to use reload instead. (This caused timing problems before)
.. todo:: This should be written to use the process return code.
.. todo:: On failure, this should be a recovery_routine call with another
restart. This will confuse and inhibit developers from testing code
though. This change should happen after
the ApacheConfigurator has been thoroughly tested. The function will
need to be moved into the class again. Perhaps
this version can live on... for testing purposes.
"""
try:
proc = subprocess.Popen([CONFIG.APACHE2, 'restart'],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
text = proc.communicate()
stdout, stderr = proc.communicate()
if proc.returncode != 0:
# Enter recovery routine...
logging.error("Configtest failed")
logging.error(text[0])
logging.error(text[1])
logging.error("Apache Restart Failed!")
logging.error(stdout)
logging.error(stderr)
return False
except (OSError, ValueError):
letsencrypt/client/apache/dvsni.py
+2 -2
View File
@@ -88,7 +88,7 @@ class ApacheDvsni(object):
# Create all of the challenge certs
for chall in self.dvsni_chall:
cert_path = self.get_cert_file(chall.nonce)
self.config.register_file_creation(cert_path)
self.config.reverter.register_file_creation(True, cert_path)
s_b64 = challenge_util.dvsni_gen_cert(
cert_path, chall.domain, chall.r_b64, chall.nonce, chall.key)
@@ -120,7 +120,7 @@ class ApacheDvsni(object):
config_text += "</IfModule>\n"
self._conf_include_check(self.config.parser.loc["default"])
self.config.register_file_creation(True, self.challenge_conf)
self.config.reverter.register_file_creation(True, self.challenge_conf)
with open(self.challenge_conf, 'w') as new_conf:
new_conf.write(config_text)
letsencrypt/client/apache/parser.py
+1 -1
View File
@@ -342,7 +342,7 @@ class ApacheParser(object):
if os.path.isfile(os.path.join(self.root, name)):
return os.path.join(self.root, name)
raise errors.LetsEncryptConfiguratorError(
raise errors.LetsEncryptNoInstallationError(
"Could not find configuration root")
def _set_user_config_file(self, root):
letsencrypt/client/augeas_configurator.py
+49 -326
View File
@@ -1,14 +1,10 @@
"""Class of Augeas Configurators."""
import logging
import os
import sys
import shutil
import time
import augeas
from letsencrypt.client import CONFIG
from letsencrypt.client import le_util
from letsencrypt.client import reverter
class AugeasConfigurator(object):
@@ -19,6 +15,8 @@ class AugeasConfigurator(object):
:ivar str save_notes: Human-readable configuration change notes
:ivar dict direc: dictionary containing save directory paths
:ivar reverter: saves and reverts checkpoints
:type reverter: :class:`letsencrypt.client.reverter.Reverter`
"""
@@ -35,14 +33,19 @@ class AugeasConfigurator(object):
"temp": CONFIG.TEMP_CHECKPOINT_DIR,
"progress": CONFIG.IN_PROGRESS_DIR}
self.direc = direc
# Set Augeas flags to not save backup (we do it ourselves)
# Set Augeas to not load anything by default
my_flags = augeas.Augeas.NONE | augeas.Augeas.NO_MODL_AUTOLOAD
self.aug = augeas.Augeas(flags=my_flags)
self.save_notes = ""
# See if any temporary changes need to be recovered
# This needs to occur before VirtualHost objects are setup...
# because this will change the underlying configuration and potential
# vhosts
self.reverter = reverter.Reverter(direc)
self.reverter.recovery_routine()
def check_parsing_errors(self, lens):
"""Verify Augeas can parse all of the lens files.
@@ -54,13 +57,13 @@ class AugeasConfigurator(object):
for path in error_files:
# Check to see if it was an error resulting from the use of
# the httpd lens
lens_path = self.aug.get(path + '/lens')
lens_path = self.aug.get(path + "/lens")
# As aug.get may return null
if lens_path and lens in lens_path:
# Strip off /augeas/files and /error
logging.error('There has been an error in parsing the file: %s',
logging.error("There has been an error in parsing the file: %s",
path[13:len(path) - 6])
logging.error(self.aug.get(path + '/message'))
logging.error(self.aug.get(path + "/message"))
def save(self, title=None, temporary=False):
"""Saves all changes to the configuration files.
@@ -85,16 +88,7 @@ class AugeasConfigurator(object):
# This is a noop save
self.aug.save()
except (RuntimeError, IOError):
# Check for the root of save problems
new_errs = self.aug.match("/augeas//error")
# logging.error("During Save - %s", mod_conf)
# Only print new errors caused by recent save
for err in new_errs:
if err not in ex_errs:
logging.error(
"Unable to save file - %s", err[13:len(err) - 6])
logging.error("Attempted Save Notes")
logging.error(self.save_notes)
self._log_save_errors(ex_errs)
# Erase Save Notes
self.save_notes = ""
return False
@@ -111,27 +105,15 @@ class AugeasConfigurator(object):
for path in save_paths:
save_files.add(self.aug.get(path)[6:])
valid, message = self.check_tempfile_saves(save_files)
if not valid:
logging.fatal(message)
# What is the protocol in this situation?
# This shouldn't happen if the challenge codebase is correct
return False
# Create Checkpoint
if temporary:
self.add_to_checkpoint(self.direc["temp"], save_files)
self.reverter.add_to_temp_checkpoint(
save_files, self.save_notes)
else:
self.add_to_checkpoint(self.direc["progress"], save_files)
self.reverter.add_to_checkpoint(save_files, self.save_notes)
if title and not temporary and os.path.isdir(self.direc["progress"]):
success = self._finalize_checkpoint(self.direc["progress"], title)
if not success:
# This should never happen
# This will be hopefully be cleaned up on the recovery
# routine startup
sys.exit(9)
if title and not temporary:
self.reverter.finalize_checkpoint(title)
self.aug.set("/augeas/save", save_state)
self.save_notes = ""
@@ -139,307 +121,48 @@ class AugeasConfigurator(object):
return True
def revert_challenge_config(self):
"""Reload users original configuration files after a challenge.
def _log_save_errors(self, ex_errs):
"""Log errors due to bad Augeas save.
This function should reload the users original configuration files
for all saves with temporary=True
:param list ex_errs: Existing errors before save
"""
if os.path.isdir(self.direc["temp"]):
result = self._recover_checkpoint(self.direc["temp"])
if result != 0:
# We have a partial or incomplete recovery
logging.fatal("Incomplete or failed recovery for %s",
self.direc["temp"])
sys.exit(67)
# Remember to reload Augeas
self.aug.load()
def rollback_checkpoints(self, rollback=1):
"""Revert 'rollback' number of configuration checkpoints.
:param int rollback: Number of checkpoints to reverse
"""
try:
rollback = int(rollback)
except ValueError:
logging.error("Rollback argument must be a positive integer")
# Sanity check input
if rollback < 1:
logging.error("Rollback argument must be a positive integer")
return
backups = os.listdir(self.direc["backup"])
backups.sort()
if len(backups) < rollback:
logging.error("Unable to rollback %d checkpoints, only %d exist",
rollback, len(backups))
while rollback > 0 and backups:
cp_dir = self.direc["backup"] + backups.pop()
result = self._recover_checkpoint(cp_dir)
if result != 0:
logging.fatal("Failed to load checkpoint during rollback")
sys.exit(39)
rollback -= 1
self.aug.load()
def view_config_changes(self):
"""Displays all saved checkpoints.
All checkpoints are printed to the console.
Note: Any 'IN_PROGRESS' checkpoints will be removed by the cleanup
script found in the constructor, before this function would ever be
called.
"""
backups = os.listdir(self.direc["backup"])
backups.sort(reverse=True)
if not backups:
print ("Letsencrypt has not saved any backups of your "
"apache configuration")
# Make sure there isn't anything unexpected in the backup folder
# There should only be timestamped (float) directories
try:
for bkup in backups:
float(bkup)
except ValueError:
assert False, "Invalid files in %s" % self.direc["backup"]
for bkup in backups:
print time.ctime(float(bkup))
cur_dir = self.direc["backup"] + bkup
with open(os.path.join(cur_dir, "CHANGES_SINCE")) as changes_fd:
print changes_fd.read()
print "Affected files:"
with open(os.path.join(cur_dir, "FILEPATHS")) as paths_fd:
filepaths = paths_fd.read().splitlines()
for path in filepaths:
print " %s" % path
try:
with open(os.path.join(cur_dir, "NEW_FILES")) as new_fd:
print "New Configuration Files:"
filepaths = new_fd.read().splitlines()
for path in filepaths:
print " %s" % path
except (IOError, OSError) as exc:
print exc
print ""
def add_to_checkpoint(self, cp_dir, save_files):
"""Add save files to checkpoint directory.
:param str cp_dir: Checkpoint directory filepath
:param set save_files: set of files to save
"""
le_util.make_or_verify_dir(cp_dir, 0o755, os.geteuid())
existing_filepaths = []
filepaths_path = os.path.join(cp_dir, "FILEPATHS")
# Open up FILEPATHS differently depending on if it already exists
if os.path.isfile(filepaths_path):
op_fd = open(filepaths_path, 'r+')
existing_filepaths = op_fd.read().splitlines()
else:
op_fd = open(filepaths_path, 'w')
idx = len(existing_filepaths)
for filename in save_files:
if filename not in existing_filepaths:
# Tag files with index so multiple files can
# have the same filename
logging.debug("Creating backup of %s", filename)
shutil.copy2(filename, os.path.join(
cp_dir, os.path.basename(filename) + "_" + str(idx)))
op_fd.write(filename + '\n')
idx += 1
op_fd.close()
with open(os.path.join(cp_dir, "CHANGES_SINCE"), 'a') as notes_fd:
notes_fd.write(self.save_notes)
def _recover_checkpoint(self, cp_dir):
"""Recover a specific checkpoint.
Recover a specific checkpoint provided by cp_dir
Note: this function does not reload augeas.
:param str cp_dir: checkpoint directory file path
:returns: 0 success, 1 Unable to revert, -1 Unable to delete
:rtype: int
"""
if os.path.isfile(os.path.join(cp_dir, "FILEPATHS")):
try:
with open(os.path.join(cp_dir, "FILEPATHS")) as paths_fd:
filepaths = paths_fd.read().splitlines()
for idx, path in enumerate(filepaths):
shutil.copy2(os.path.join(
cp_dir,
os.path.basename(path) + '_' + str(idx)), path)
except (IOError, OSError):
# This file is required in all checkpoints.
logging.error("Unable to recover files from %s", cp_dir)
return 1
# Remove any newly added files if they exist
self._remove_contained_files(os.path.join(cp_dir, "NEW_FILES"))
try:
shutil.rmtree(cp_dir)
except OSError:
logging.error("Unable to remove directory: %s", cp_dir)
return -1
return 0
def check_tempfile_saves(self, save_files): # pylint: disable=no-self-use
"""Verify save isn't overwriting any temporary files.
:param set save_files: Set of files about to be saved.
:returns: Success, error message
:rtype: bool, str
"""
temp_path = "%sFILEPATHS" % self.direc["temp"]
if os.path.isfile(temp_path):
with open(temp_path, 'r') as protected_fd:
protected_files = protected_fd.read().splitlines()
for filename in protected_files:
if filename in save_files:
return False, ("Attempting to overwrite challenge "
"file - %s" % filename)
return True, ""
def register_file_creation(self, temporary, *files):
# pylint: disable=no-self-use
r"""Register the creation of all files during letsencrypt execution.
Call this method before writing to the file to make sure that the
file will be cleaned up if the program exits unexpectedly.
(Before a save occurs)
:param bool temporary: If the file creation registry is for
a temp or permanent save.
:param \*files: file paths (str) to be registered
"""
if temporary:
cp_dir = self.direc["temp"]
else:
cp_dir = self.direc["progress"]
le_util.make_or_verify_dir(cp_dir, 0o755, os.geteuid())
try:
with open(os.path.join(cp_dir, "NEW_FILES"), 'a') as new_fd:
for file_path in files:
new_fd.write("%s\n" % file_path)
except (IOError, OSError):
logging.error("ERROR: Unable to register file creation")
# Check for the root of save problems
new_errs = self.aug.match("/augeas//error")
# logging.error("During Save - %s", mod_conf)
# Only print new errors caused by recent save
for err in new_errs:
if err not in ex_errs:
logging.error(
"Unable to save file - %s", err[13:len(err) - 6])
logging.error("Attempted Save Notes")
logging.error(self.save_notes)
# Wrapper functions for Reverter class
def recovery_routine(self):
"""Revert all previously modified files.
First, any changes found in self.direc["temp"] are removed,
then IN_PROGRESS changes are removed The order is important.
IN_PROGRESS is unable to add files that are already added by a TEMP
change. Thus TEMP must be rolled back first because that will be the
'latest' occurrence of the file.
Reverts all modified files that have not been saved as a checkpoint
"""
self.revert_challenge_config()
if os.path.isdir(self.direc["progress"]):
result = self._recover_checkpoint(self.direc["progress"])
if result != 0:
# We have a partial or incomplete recovery
# Not as egregious
# TODO: Additional tests? recovery
logging.fatal("Incomplete or failed recovery for %s",
self.direc["progress"])
sys.exit(68)
self.reverter.recovery_routine()
# Need to reload configuration after these changes take effect
self.aug.load()
# Need to reload configuration after these changes take effect
self.aug.load()
def revert_challenge_config(self):
"""Used to cleanup challenge configurations."""
self.reverter.revert_temporary_config()
self.aug.load()
def _remove_contained_files(self, file_list): # pylint: disable=no-self-use
"""Erase all files contained within file_list.
def rollback_checkpoints(self, rollback=1):
"""Rollback saved checkpoints.
:param str file_list: file containing list of file paths to be deleted
:returns: Success
:rtype: bool
:param int rollback: Number of checkpoints to revert
"""
# Check to see that file exists to differentiate can't find file_list
# and can't remove filepaths within file_list errors.
if not os.path.isfile(file_list):
return False
try:
with open(file_list, 'r') as list_fd:
filepaths = list_fd.read().splitlines()
for path in filepaths:
# Files are registered before they are added... so
# check to see if file exists first
if os.path.lexists(path):
os.remove(path)
else:
logging.warn(
"File: %s - Could not be found to be deleted\n"
"LE probably shut down unexpectedly", path)
except (IOError, OSError):
logging.fatal(
"Unable to remove filepaths contained within %s", file_list)
sys.exit(41)
self.reverter.rollback_checkpoints(rollback)
self.aug.load()
return True
def _finalize_checkpoint(self, cp_dir, title):
# pylint: disable=no-self-use
"""Move IN_PROGRESS checkpoint to timestamped checkpoint.
Adds title to cp_dir CHANGES_SINCE
Move cp_dir to Backups directory and rename with timestamp
:param cp_dir: "IN PROGRESS" directory
:type cp_dir: str
:returns: Success
:rtype: bool
"""
final_dir = os.path.join(self.direc["backup"], str(time.time()))
changes_since_path = os.path.join(cp_dir, "CHANGES_SINCE")
changes_since_tmp_path = os.path.join(cp_dir, "CHANGES_SINCE.tmp")
try:
with open(changes_since_tmp_path, 'w') as changes_tmp:
changes_tmp.write("-- %s --\n" % title)
with open(changes_since_path, 'r') as changes_orig:
changes_tmp.write(changes_orig.read())
shutil.move(changes_since_tmp_path, changes_since_path)
except (IOError, OSError):
logging.error("Unable to finalize checkpoint - adding title")
return False
try:
os.rename(cp_dir, final_dir)
except OSError:
logging.error(
"Unable to finalize checkpoint, %s -> %s", cp_dir, final_dir)
return False
return True
def view_config_changes(self):
"""Show all of the configuration changes that have taken place."""
self.reverter.view_config_changes()
letsencrypt/client/challenge_util.py
+1
View File
@@ -8,6 +8,7 @@ from letsencrypt.client import CONFIG
from letsencrypt.client import crypto_util
from letsencrypt.client import le_util
# Authenticator Challenges
DvsniChall = collections.namedtuple("DvsniChall", "domain, r_b64, nonce, key")
SimpleHttpsChall = collections.namedtuple(
letsencrypt/client/client.py
+153 -14
View File
@@ -20,9 +20,12 @@ from letsencrypt.client import errors
from letsencrypt.client import interfaces
from letsencrypt.client import le_util
from letsencrypt.client import network
from letsencrypt.client import reverter
from letsencrypt.client import revoker
from letsencrypt.client.apache import configurator
# it's weird to point to chocolate servers via raw IPv6 addresses, and
# it's weird to point to ACME servers via raw IPv6 addresses, and
# such addresses can be %SCARY in some contexts, so out of paranoia
# let's disable them by default
ALLOW_RAW_IPV6_SERVER = False
@@ -48,6 +51,7 @@ class Client(object):
zope.interface.implements(interfaces.IAuthenticator)
Key = collections.namedtuple("Key", "file pem")
# Note: form is the type of data, "pem" or "der"
CSR = collections.namedtuple("CSR", "file data form")
def __init__(self, server, authkey, dv_auth, installer):
@@ -64,9 +68,12 @@ class Client(object):
self.installer = installer
client_auth = client_authenticator.ClientAuthenticator(server)
self.auth_handler = auth_handler.AuthHandler(
dv_auth, client_auth, self.network)
if dv_auth is not None:
client_auth = client_authenticator.ClientAuthenticator(server)
self.auth_handler = auth_handler.AuthHandler(
dv_auth, client_auth, self.network)
else:
self.auth_handler = None
def obtain_certificate(self, domains, csr=None,
cert_path=CONFIG.CERT_PATH,
@@ -85,7 +92,12 @@ class Client(object):
:rtype: `tuple` of `str`
"""
if self.auth_handler is None:
logging.warning("Unable to obtain a certificate, because client "
"does not have a valid auth handler.")
sanity_check_names(domains)
# Request Challenges
for name in domains:
self.auth_handler.add_chall_msg(
@@ -179,6 +191,11 @@ class Client(object):
:param str chain_file: chain file path
"""
if self.installer is None:
logging.warning("No installer specified, client is unable to deploy"
"the certificate")
raise errors.LetsEncryptClientError("No installer available")
chain = None if chain_file is None else os.path.abspath(chain_file)
for dom in domains:
@@ -197,12 +214,24 @@ class Client(object):
def enhance_config(self, domains, redirect=None):
"""Enhance the configuration.
.. todo:: This needs to handle the specific enhancements offered by the
installer. We will also have to find a method to pass in the chosen
values efficiently.
:param list domains: list of domains to configure
:param redirect: If traffic should be forwarded from HTTP to HTTPS.
:type redirect: bool or None
:raises :class:`letsencrypt.client.errors.LetsEncryptClientError`: if
no installer is specified in the client.
"""
if self.installer is None:
logging.warning("No installer is specified, there isn't any "
"configuration to enhance.")
raise errors.LetsEncryptClientError("No installer available")
if redirect is None:
redirect = zope.component.getUtility(
interfaces.IDisplay).redirect_by_default()
@@ -210,16 +239,6 @@ class Client(object):
if redirect:
self.redirect_to_ssl(domains)
# if self.ocsp_stapling is None:
# q = ("Would you like to protect the privacy of your users "
# "by enabling OCSP stapling? If so, your users will not have "
# "to query the Let's Encrypt CA separately about the current "
# "revocation status of your certificate.")
# self.ocsp_stapling = self.ocsp_stapling = display.ocsp_stapling(q)
# if self.ocsp_stapling:
# # TODO enable OCSP Stapling
# continue
def store_cert_key(self, cert_file, encrypt=False):
"""Store certificate key. (Used to allow quick revocation)
@@ -421,3 +440,123 @@ def is_hostname_sane(hostname):
return True
except socket.error:
return False
# This should be controlled by commandline parameters
def determine_authenticator():
"""Returns a valid IAuthenticator."""
try:
return configurator.ApacheConfigurator()
except errors.LetsEncryptNoInstallationError:
logging.info("Unable to determine a way to authenticate the server")
def determine_installer():
"""Returns a valid installer if one exists."""
try:
return configurator.ApacheConfigurator()
except errors.LetsEncryptNoInstallationError:
logging.info("Unable to find a way to install the certificate.")
def rollback(checkpoints):
"""Revert configuration the specified number of checkpoints.
.. note:: If another installer uses something other than the reverter class
to do their configuration changes, the correct reverter will have to be
determined.
.. note:: This function restarts the server even if there weren't any
rollbacks. The user may be confused or made an error and simply needs
to restart the server.
.. todo:: This function will have to change depending on the functionality
of future installers. Perhaps the interface should define errors that
are thrown for the various functions.
:param int checkpoints: Number of checkpoints to revert.
"""
# Misconfigurations are only a slight problems... allow the user to rollback
try:
installer = determine_installer()
except errors.LetsEncryptMisconfigurationError:
_misconfigured_rollback(checkpoints)
return
# No Errors occurred during init... proceed normally
# If installer is None... couldn't find an installer... there shouldn't be
# anything to rollback
if installer is not None:
installer.rollback_checkpoints(checkpoints)
installer.restart()
def _misconfigured_rollback(checkpoints):
"""Handles the case where the Installer is misconfigured."""
yes = zope.component.getUtility(interfaces.IDisplay).generic_yesno(
"Oh, no! The web server is currently misconfigured.{0}{0}"
"Would you still like to rollback the "
"configuration?".format(os.linesep))
if not yes:
logging.info("The error message is above.")
logging.info("Configuration was not rolled back.")
return
logging.info("Rolling back using the Reverter module")
# recovery routine has probably already been run by installer
# in the__init__ attempt, run it again for safety... it shouldn't hurt
# Also... not sure how future installers will handle recovery.
rev = reverter.Reverter()
rev.recovery_routine()
rev.rollback_checkpoints(checkpoints)
# We should try to restart the server
try:
installer = determine_installer()
installer.restart()
logging.info("Hooray! Rollback solved the misconfiguration!")
logging.info("Your web server is back up and running.")
except errors.LetsEncryptMisconfigurationError:
logging.warning(
"Rollback was unable to solve the misconfiguration issues")
def revoke(server):
"""Revoke certificates.
:param str server: ACME server the client wishes to revoke certificates from
"""
# Misconfigurations don't really matter. Determine installer better choose
# correctly though.
try:
installer = determine_installer()
except errors.LetsEncryptMisconfigurationError:
zope.component.getUtility(interfaces.IDisplay).generic_notification(
"The web server is currently misconfigured. Some "
"abilities like seeing which certificates are currently "
"installed may not be available.")
installer = None
# This is a temporary fix to avoid errors. The Revoker is not fully
# developed.
if installer is None:
zope.component.getUtility(interfaces.IDisplay).generic_notification(
"The Let's Encrypt Revoker module does not currently support "
"revocation without a valid installer. This feature should come "
"soon.")
return
revoc = revoker.Revoker(server, installer)
revoc.list_certs_keys()
def view_config_changes():
"""View checkpoints and associated configuration changes.
.. note:: This assumes that the installation is using a Reverter object.
"""
rev = reverter.Reverter()
rev.recovery_routine()
rev.view_config_changes()
letsencrypt/client/crypto_util.py
+4 -7
View File
@@ -45,7 +45,7 @@ def create_sig(msg, key_str, nonce=None, nonce_len=CONFIG.NONCE_SIZE):
hashed = Crypto.Hash.SHA256.new(msg_with_nonce)
signature = Crypto.Signature.PKCS1_v1_5.new(key).sign(hashed)
logging.debug('%s signed as %s', msg_with_nonce, signature)
logging.debug("%s signed as %s", msg_with_nonce, signature)
n_bytes = binascii.unhexlify(_leading_zeros(hex(key.n)[2:].rstrip("L")))
e_bytes = binascii.unhexlify(_leading_zeros(hex(key.e)[2:].rstrip("L")))
@@ -95,7 +95,7 @@ def make_csr(key_str, domains):
extstack = M2Crypto.X509.X509_Extension_Stack()
ext = M2Crypto.X509.new_extension(
'subjectAltName', ", ".join("DNS:%s" % d for d in domains))
"subjectAltName", ", ".join("DNS:%s" % d for d in domains))
extstack.push(ext)
csr.add_extensions(extstack)
@@ -144,7 +144,6 @@ def csr_matches_pubkey(csr, privkey):
return csr_obj.get_pubkey().get_rsa().pub() == privkey_obj.pub()
# based on M2Crypto unit test written by Toby Allsopp
def make_key(bits):
"""Generate PEM encoded RSA key.
@@ -154,10 +153,6 @@ def make_key(bits):
:rtype: str
"""
# rsa = M2Crypto.RSA.gen_key(bits, 65537)
# key_pem = rsa.as_pem(cipher=None)
# rsa = None # should not be freed here
# Python Crypto module doesn't produce any stdout
return Crypto.PublicKey.RSA.generate(bits).exportKey(format='PEM')
@@ -228,6 +223,8 @@ def make_ss_cert(key_str, domains, not_before=None,
def get_cert_info(filename):
"""Get certificate info.
.. todo:: Pub key is assumed to be RSA... find a good solution to allow EC.
:param str filename: Name of file containing certificate in PEM format.
:rtype: dict
letsencrypt/client/display.py
+224 -50
View File
@@ -1,4 +1,5 @@
"""Lets Encrypt display."""
import os
import textwrap
import dialog
@@ -14,7 +15,13 @@ HEIGHT = 20
class CommonDisplayMixin(object): # pylint: disable=too-few-public-methods
"""Mixin with methods common to classes implementing IDisplay."""
def redirect_by_default(self): # pylint: disable=missing-docstring
def redirect_by_default(self):
"""Determines whether the user would like to redirect to HTTPS.
:returns: True if redirect is desired, False otherwise
:rtype: bool
"""
choices = [
("Easy", "Allow both HTTP and HTTPS access to these sites"),
("Secure", "Make all requests redirect to secure HTTPS access")]
@@ -41,12 +48,29 @@ class NcursesDisplay(CommonDisplayMixin):
self.width = width
self.height = height
def generic_notification(self, message):
# pylint: disable=missing-docstring
self.dialog.msgbox(message, width=self.width)
def generic_notification(self, message, height=10):
"""Display a notification to the user and wait for user acceptance.
:param str message: Message to display
:param int height: Height of the dialog box
"""
self.dialog.msgbox(message, height, width=self.width)
def generic_menu(self, message, choices, unused_input_text=""):
# pylint: disable=missing-docstring
"""Display a menu.
:param str message: title of menu
:param choices: menu lines
:type choices: list of tuples (tag, item) or
list of items (tags will be enumerated)
:returns: tuple of the form (code, tag) where
code is a display exit code
tag is the tag string corresponding to the item chosen
:rtype: tuple
"""
# Can accept either tuples or just the actual choices
if choices and isinstance(choices[0], tuple):
code, selection = self.dialog.menu(
@@ -57,18 +81,46 @@ class NcursesDisplay(CommonDisplayMixin):
code, tag = self.dialog.menu(
message, choices=choices, width=self.width, height=self.height)
return code(int(tag) - 1)
return code, int(tag) - 1
def generic_input(self, message): # pylint: disable=missing-docstring
def generic_input(self, message):
"""Display an input box to the user.
:param str message: Message to display that asks for input.
:returns: tuple of the form (code, string) where
code is a display exit code
string is the input entered by the user
"""
return self.dialog.inputbox(message)
def generic_yesno(self, message, yes_label="Yes", no_label="No"):
# pylint: disable=missing-docstring
"""Display a Yes/No dialog box
:param str message: message to display to user
:param str yes_label: label on the 'yes' button
:param str no_label: label on the 'no' button
:returns: if yes_label was selected
:rtype: bool
"""
return self.dialog.DIALOG_OK == self.dialog.yesno(
message, self.height, self.width,
yes_label=yes_label, no_label=no_label)
def filter_names(self, names): # pylint: disable=missing-docstring
def filter_names(self, names):
"""Determine which names the user would like to select from a list.
:param list names: domain names
:returns: tuple of the form (code, names) where
code is a display exit code
names is a list of names selected
:rtype: tuple
"""
choices = [(n, "", 0) for n in names]
code, names = self.dialog.checklist(
"Which names would you like to activate HTTPS for?",
@@ -76,12 +128,26 @@ class NcursesDisplay(CommonDisplayMixin):
return code, [str(s) for s in names]
def success_installation(self, domains):
# pylint: disable=missing-docstring
"""Display a box confirming the installation of HTTPS.
:param list domains: domain names which were enabled
"""
self.dialog.msgbox(
"\nCongratulations! You have successfully enabled "
+ gen_https_names(domains) + "!", width=self.width)
def display_certs(self, certs): # pylint: disable=missing-docstring
def display_certs(self, certs):
"""Display certificates for revocation.
:param list certs: `list` of `dict` used throughout revoker.py
:returns: tuple of the form (code, selection) where
code is a display exit code
selection is the user's int selection
:rtype: tuple
"""
list_choices = [
(str(i+1), "%s | %s | %s" %
(str(c["cn"].ljust(self.width - 39)),
@@ -98,7 +164,15 @@ class NcursesDisplay(CommonDisplayMixin):
tag = -1
return code, (int(tag) - 1)
def confirm_revocation(self, cert): # pylint: disable=missing-docstring
def confirm_revocation(self, cert):
"""Confirm revocation screen.
:param dict cert: cert dict used throughout revoker.py
:returns: True if user would like to revoke, False otherwise
:rtype: bool
"""
text = ("Are you sure you would like to revoke the following "
"certificate:\n")
text += cert_info_frame(cert)
@@ -106,10 +180,14 @@ class NcursesDisplay(CommonDisplayMixin):
return self.dialog.DIALOG_OK == self.dialog.yesno(
text, width=self.width, height=self.height)
def more_info_cert(self, cert): # pylint: disable=missing-docstring
def more_info_cert(self, cert):
"""Displays more information about the certificate.
:param dict cert: cert dict used throughout revoker.py
"""
text = "Certificate Information:\n"
text += cert_info_frame(cert)
print text
self.dialog.msgbox(text, width=self.width, height=self.height)
@@ -122,15 +200,36 @@ class FileDisplay(CommonDisplayMixin):
super(FileDisplay, self).__init__()
self.outfile = outfile
def generic_notification(self, message):
# pylint: disable=missing-docstring
def generic_notification(self, message, unused_height):
"""Displays a notification and waits for user acceptance.
:param str message: Message to display
"""
side_frame = '-' * 79
msg = textwrap.fill(message, 80)
self.outfile.write("\n%s\n%s\n%s\n" % (side_frame, msg, side_frame))
lines = message.splitlines()
fixed_l = []
for line in lines:
fixed_l.append(textwrap.fill(line, 80))
self.outfile.write(
"{0}{1}{0}{2}{0}{1}{0}".format(
os.linesep, side_frame, os.linesep.join(fixed_l)))
raw_input("Press Enter to Continue")
def generic_menu(self, message, choices, input_text=""):
# pylint: disable=missing-docstring
"""Display a menu.
:param str message: title of menu
:param choices: Menu lines
:type choices: list of tuples (tag, item) or
list of items (tags will be enumerated)
:returns: tuple of the form (code, tag) where
code is a display exit code
tag is the tag string corresponding to the item chosen
:rtype: tuple
"""
# Can take either tuples or single items in choices list
if choices and isinstance(choices[0], tuple):
choices = ["%s - %s" % (c[0], c[1]) for c in choices]
@@ -145,27 +244,54 @@ class FileDisplay(CommonDisplayMixin):
self.outfile.write("%s\n" % side_frame)
code, selection = self.__get_valid_int_ans(
code, selection = self._get_valid_int_ans(
"%s (c to cancel): " % input_text)
return code, (selection - 1)
def generic_input(self, message):
# pylint: disable=no-self-use,missing-docstring
# pylint: disable=no-self-use
"""Accept input from the user
:param str message: message to display to the user
:returns: tuple of (code, input) where
code is a display exit code
input is a str of the user's input
:rtype: tuple
"""
ans = raw_input("%s (Enter c to cancel)\n" % message)
if ans.startswith('c') or ans.startswith('C'):
return CANCEL, -1
if ans == 'c' or ans == 'C':
return CANCEL, "-1"
else:
return OK, ans
def generic_yesno(self, message, unused_yes_label="", unused_no_label=""):
# pylint: disable=missing-docstring
"""Query the user with a yes/no question.
:param str message: question for the user
:returns: True for 'Yes', False for 'No"
:rtype: bool
"""
self.outfile.write("\n%s\n" % textwrap.fill(message, 80))
ans = raw_input("y/n: ")
return ans.startswith('y') or ans.startswith('Y')
def filter_names(self, names): # pylint: disable=missing-docstring
def filter_names(self, names):
"""Determine which names the user would like to select from a list.
:param list names: domain names
:returns: tuple of the form (code, names) where
code is a display exit code
names is a list of names selected
:rtype: tuple
"""
code, tag = self.generic_menu(
"Choose the names would you like to upgrade to HTTPS?",
names, "Select the number of the name: ")
@@ -173,7 +299,28 @@ class FileDisplay(CommonDisplayMixin):
# Make sure to return a list...
return code, [names[tag]]
def display_certs(self, certs): # pylint: disable=missing-docstring
def success_installation(self, domains):
"""Display a box confirming the installation of HTTPS.
:param list domains: domain names which were enabled
"""
side_frame = '*' * 79
msg = textwrap.fill("Congratulations! You have successfully "
"enabled %s!" % gen_https_names(domains))
self.outfile.write("%s\n%s\n%s\n" % (side_frame, msg, side_frame))
def display_certs(self, certs):
"""Display certificates for revocation.
:param list certs: `list` of `dict` used throughout revoker.py
:returns: tuple of the form (code, selection) where
code is a display exit code
selection is the user's int selection
:rtype: tuple
"""
menu_choices = [(str(i+1), str(c["cn"]) + " - " + c["pub_key"] +
" - " + str(c["not_before"])[:-6])
for i, c in enumerate(certs)]
@@ -183,11 +330,20 @@ class FileDisplay(CommonDisplayMixin):
self.outfile.write(textwrap.fill(
"%s: %s - %s Signed (UTC): %s\n" % choice[:4]))
return self.__get_valid_int_ans("Revoke Number (c to cancel): ") - 1
return self._get_valid_int_ans("Revoke Number (c to cancel): ") - 1
def __get_valid_int_ans(self, input_string):
def _get_valid_int_ans(self, input_string):
"""Get a numerical selection.
:param str input_string: Instructions for the user to make a selection.
:returns: tuple of the form (code, selection) where
code is a display exit code
selection is the user's int selection
:rtype: tuple
"""
valid_ans = False
e_msg = "Please input a number or the letter c to cancel\n"
while not valid_ans:
@@ -210,14 +366,15 @@ class FileDisplay(CommonDisplayMixin):
return code, selection
def success_installation(self, domains):
# pylint: disable=missing-docstring
side_frame = '*' * 79
msg = textwrap.fill("Congratulations! You have successfully "
"enabled %s!" % gen_https_names(domains))
self.outfile.write("%s\n%s\n%s\n" % (side_frame, msg, side_frame))
def confirm_revocation(self, cert):
"""Confirm revocation screen.
def confirm_revocation(self, cert): # pylint: disable=missing-docstring
:param dict cert: cert dict used throughout revoker.py
:returns: True if user would like to revoke, False otherwise
:rtype: bool
"""
self.outfile.write("Are you sure you would like to revoke "
"the following certificate:\n")
self.outfile.write(cert_info_frame(cert))
@@ -225,39 +382,56 @@ class FileDisplay(CommonDisplayMixin):
ans = raw_input("y/n")
return ans.startswith('y') or ans.startswith('Y')
def more_info_cert(self, cert): # pylint: disable=missing-docstring
def more_info_cert(self, cert):
"""Displays more info about the cert.
:param dict cert: cert dict used throughout revoker.py
"""
self.outfile.write("\nCertificate Information:\n")
self.outfile.write(cert_info_frame(cert))
# Display exit codes
OK = "ok"
"""Display exit code indicating user acceptance"""
CANCEL = "cancel"
"""Display exit code for a user canceling the display"""
HELP = "help"
"""Display exit code when for when the user requests more help."""
def cert_info_frame(cert): # pylint: disable=missing-docstring
text = "-" * (WIDTH - 4) + "\n"
def cert_info_frame(cert):
"""Nicely frames a cert dict used in revoker.py"""
text = "-" * (WIDTH - 4) + os.linesep
text += cert_info_string(cert)
text += "-" * (WIDTH - 4)
return text
def cert_info_string(cert): # pylint: disable=missing-docstring
text = "Subject: %s\n" % cert["subject"]
text += "SAN: %s\n" % cert["san"]
text += "Issuer: %s\n" % cert["issuer"]
text += "Public Key: %s\n" % cert["pub_key"]
text += "Not Before: %s\n" % str(cert["not_before"])
text += "Not After: %s\n" % str(cert["not_after"])
text += "Serial Number: %s\n" % cert["serial"]
text += "SHA1: %s\n" % cert["fingerprint"]
text += "Installed: %s\n" % cert["installed"]
return text
def cert_info_string(cert):
"""Turn a cert dict into a string."""
text = []
text.append("Subject: %s" % cert["subject"])
text.append("SAN: %s" % cert["san"])
text.append("Issuer: %s" % cert["issuer"])
text.append("Public Key: %s" % cert["pub_key"])
text.append("Not Before: %s" % str(cert["not_before"]))
text.append("Not After: %s" % str(cert["not_after"]))
text.append("Serial Number: %s" % cert["serial"])
text.append("SHA1: %s" % cert["fingerprint"])
text.append("Installed: %s" % cert["installed"])
return os.linesep.join(text)
def gen_https_names(domains):
"""Returns a string of the https domains.
Domains are formatted nicely with https:// prepended to each.
.. todo:: This should not use +=, rewrite this with unittests
"""
result = ""
if len(domains) > 2:
letsencrypt/client/errors.py
+13 -1
View File
@@ -5,17 +5,29 @@ class LetsEncryptClientError(Exception):
"""Generic Let's Encrypt client error."""
class LetsEncryptReverterError(LetsEncryptClientError):
"""Let's Encrypt Reverter error."""
class LetsEncryptAuthHandlerError(LetsEncryptClientError):
"""Let's Encrypt Auth Handler error."""
class LetsEncryptClientAuthError(LetsEncryptAuthHandlerError):
"""Let's Encrypt Client Authenticator Error."""
"""Let's Encrypt Client Authenticator error."""
class LetsEncryptConfiguratorError(LetsEncryptClientError):
"""Let's Encrypt Configurator error."""
class LetsEncryptNoInstallationError(LetsEncryptConfiguratorError):
"""Let's Encrypt No Installation error."""
class LetsEncryptMisconfigurationError(LetsEncryptConfiguratorError):
"""Let's Encrypt Misconfiguration error."""
class LetsEncryptDvsniError(LetsEncryptConfiguratorError):
"""Let's Encrypt DVSNI error."""
letsencrypt/client/reverter.py
+442
View File
@@ -0,0 +1,442 @@
"""Reverter class saves configuration checkpoints and allows for recovery."""
import logging
import os
import shutil
import time
import zope.component
from letsencrypt.client import CONFIG
from letsencrypt.client import display
from letsencrypt.client import errors
from letsencrypt.client import interfaces
from letsencrypt.client import le_util
class Reverter(object):
"""Reverter Class - save and revert configuration checkpoints"""
def __init__(self, direc=None):
if not direc:
direc = {'backup': CONFIG.BACKUP_DIR,
'temp': CONFIG.TEMP_CHECKPOINT_DIR,
'progress': CONFIG.IN_PROGRESS_DIR}
self.direc = direc
def revert_temporary_config(self):
"""Reload users original configuration files after a temporary save.
This function should reinstall the users original configuration files
for all saves with temporary=True
:raises :class:`errors.LetsEncryptReverterError`:
Unable to revert config
"""
if os.path.isdir(self.direc['temp']):
try:
self._recover_checkpoint(self.direc['temp'])
except errors.LetsEncryptReverterError:
# We have a partial or incomplete recovery
logging.fatal("Incomplete or failed recovery for %s",
self.direc['temp'])
raise errors.LetsEncryptReverterError(
"Unable to revert temporary config")
def rollback_checkpoints(self, rollback=1):
"""Revert 'rollback' number of configuration checkpoints.
:param int rollback: Number of checkpoints to reverse. A str num will be
cast to an integer. So '2' is also acceptable.
:raises :class:`letsencrypt.client.errors.LetsEncryptReverterError`: If
there is a problem with the input or if the function is unable to
correctly revert the configuration checkpoints.
"""
try:
rollback = int(rollback)
except ValueError:
logging.error("Rollback argument must be a positive integer")
raise errors.LetsEncryptReverterError("Invalid Input")
# Sanity check input
if rollback < 0:
logging.error("Rollback argument must be a positive integer")
raise errors.LetsEncryptReverterError("Invalid Input")
backups = os.listdir(self.direc['backup'])
backups.sort()
if len(backups) < rollback:
logging.warning("Unable to rollback %d checkpoints, only %d exist",
rollback, len(backups))
while rollback > 0 and backups:
cp_dir = os.path.join(self.direc['backup'], backups.pop())
try:
self._recover_checkpoint(cp_dir)
except errors.LetsEncryptReverterError:
logging.fatal("Failed to load checkpoint during rollback")
raise errors.LetsEncryptReverterError(
"Unable to load checkpoint during rollback")
rollback -= 1
def view_config_changes(self):
"""Displays all saved checkpoints.
All checkpoints are printed to the console.
.. todo:: Decide on a policy for error handling, OSError IOError...
"""
backups = os.listdir(self.direc['backup'])
backups.sort(reverse=True)
if not backups:
logging.info("The Let's Encrypt client has not saved any backups "
"of your configuration")
return
# Make sure there isn't anything unexpected in the backup folder
# There should only be timestamped (float) directories
try:
for bkup in backups:
float(bkup)
except ValueError:
raise errors.LetsEncryptReverterError(
"Invalid directories in {0}".format(self.direc['backup']))
output = []
for bkup in backups:
output.append(time.ctime(float(bkup)))
cur_dir = os.path.join(self.direc['backup'], bkup)
with open(os.path.join(cur_dir, "CHANGES_SINCE")) as changes_fd:
output.append(changes_fd.read())
output.append("Affected files:")
with open(os.path.join(cur_dir, "FILEPATHS")) as paths_fd:
filepaths = paths_fd.read().splitlines()
for path in filepaths:
output.append(" {0}".format(path))
if os.path.isfile(os.path.join(cur_dir, "NEW_FILES")):
with open(os.path.join(cur_dir, "NEW_FILES")) as new_fd:
output.append("New Configuration Files:")
filepaths = new_fd.read().splitlines()
for path in filepaths:
output.append(" {0}".format(path))
output.append(os.linesep)
zope.component.getUtility(interfaces.IDisplay).generic_notification(
os.linesep.join(output), display.HEIGHT)
def add_to_temp_checkpoint(self, save_files, save_notes):
"""Add files to temporary checkpoint
param set save_files: set of filepaths to save
param str save_notes: notes about changes during the save
"""
self._add_to_checkpoint_dir(self.direc['temp'], save_files, save_notes)
def add_to_checkpoint(self, save_files, save_notes):
"""Add files to a permanent checkpoint
:param set save_files: set of filepaths to save
:param str save_notes: notes about changes during the save
"""
# Check to make sure we are not overwriting a temp file
self._check_tempfile_saves(save_files)
self._add_to_checkpoint_dir(
self.direc['progress'], save_files, save_notes)
def _add_to_checkpoint_dir(self, cp_dir, save_files, save_notes):
"""Add save files to checkpoint directory.
:param str cp_dir: Checkpoint directory filepath
:param set save_files: set of files to save
:param str save_notes: notes about changes made during the save
:raises IOError: If unable to open cp_dir + FILEPATHS file
:raises :class:`letsencrypt.client.errors.LetsEncryptReverterError: If
unable to add checkpoint
"""
le_util.make_or_verify_dir(cp_dir, 0o755, os.geteuid())
op_fd, existing_filepaths = self._read_and_append(
os.path.join(cp_dir, "FILEPATHS"))
idx = len(existing_filepaths)
for filename in save_files:
# No need to copy/index already existing files
# The oldest copy already exists in the directory...
if filename not in existing_filepaths:
# Tag files with index so multiple files can
# have the same filename
logging.debug("Creating backup of %s", filename)
try:
shutil.copy2(filename, os.path.join(
cp_dir, os.path.basename(filename) + "_" + str(idx)))
op_fd.write(filename + '\n')
# http://stackoverflow.com/questions/4726260/effective-use-of-python-shutil-copy2
except IOError:
op_fd.close()
logging.error(
"Unable to add file %s to checkpoint %s",
filename, cp_dir)
raise errors.LetsEncryptReverterError(
"Unable to add file {0} to checkpoint "
"{1}".format(filename, cp_dir))
idx += 1
op_fd.close()
with open(os.path.join(cp_dir, "CHANGES_SINCE"), 'a') as notes_fd:
notes_fd.write(save_notes)
def _read_and_append(self, filepath): # pylint: disable=no-self-use
"""Reads the file lines and returns a fd.
Read the file returning the lines, and a pointer to the end of the file.
"""
# Open up filepath differently depending on if it already exists
if os.path.isfile(filepath):
op_fd = open(filepath, 'r+')
lines = op_fd.read().splitlines()
else:
lines = []
op_fd = open(filepath, 'w')
return op_fd, lines
def _recover_checkpoint(self, cp_dir):
"""Recover a specific checkpoint.
Recover a specific checkpoint provided by cp_dir
Note: this function does not reload augeas.
:param str cp_dir: checkpoint directory file path
:raises errors.LetsEncryptReverterError: If unable to recover checkpoint
"""
if os.path.isfile(os.path.join(cp_dir, "FILEPATHS")):
try:
with open(os.path.join(cp_dir, "FILEPATHS")) as paths_fd:
filepaths = paths_fd.read().splitlines()
for idx, path in enumerate(filepaths):
shutil.copy2(os.path.join(
cp_dir,
os.path.basename(path) + '_' + str(idx)), path)
except (IOError, OSError):
# This file is required in all checkpoints.
logging.error("Unable to recover files from %s", cp_dir)
raise errors.LetsEncryptReverterError(
"Unable to recover files from %s" % cp_dir)
# Remove any newly added files if they exist
self._remove_contained_files(os.path.join(cp_dir, "NEW_FILES"))
try:
shutil.rmtree(cp_dir)
except OSError:
logging.error("Unable to remove directory: %s", cp_dir)
raise errors.LetsEncryptReverterError(
"Unable to remove directory: %s" % cp_dir)
def _check_tempfile_saves(self, save_files):
"""Verify save isn't overwriting any temporary files.
:param set save_files: Set of files about to be saved.
:raises :class:`letsencrypt.client.errors.LetsEncryptReverterError`:
when save is attempting to overwrite a temporary file.
"""
protected_files = []
# Get temp modified files
temp_path = os.path.join(self.direc['temp'], "FILEPATHS")
if os.path.isfile(temp_path):
with open(temp_path, 'r') as protected_fd:
protected_files.extend(protected_fd.read().splitlines())
# Get temp new files
new_path = os.path.join(self.direc['temp'], "NEW_FILES")
if os.path.isfile(new_path):
with open(new_path, 'r') as protected_fd:
protected_files.extend(protected_fd.read().splitlines())
# Verify no save_file is in protected_files
for filename in protected_files:
if filename in save_files:
raise errors.LetsEncryptReverterError(
"Attempting to overwrite challenge "
"file - %s" % filename)
def register_file_creation(self, temporary, *files):
r"""Register the creation of all files during letsencrypt execution.
Call this method before writing to the file to make sure that the
file will be cleaned up if the program exits unexpectedly.
(Before a save occurs)
:param bool temporary: If the file creation registry is for
a temp or permanent save.
:param \*files: file paths (str) to be registered
:raises :class:`letsencrypt.client.errors.LetsEncryptReverterError`: If
call does not contain necessary parameters or if the file creation
is unable to be registered.
"""
# Make sure some files are provided... as this is an error
# Made this mistake in my initial implementation of apache.dvsni.py
if not files:
raise errors.LetsEncryptReverterError(
"Forgot to provide files to registration call")
if temporary:
cp_dir = self.direc['temp']
else:
cp_dir = self.direc['progress']
le_util.make_or_verify_dir(cp_dir, 0o755, os.geteuid())
# Append all new files (that aren't already registered)
new_fd = None
try:
new_fd, ex_files = self._read_and_append(
os.path.join(cp_dir, "NEW_FILES"))
for path in files:
if path not in ex_files:
new_fd.write("{0}{1}".format(path, os.linesep))
except (IOError, OSError):
logging.error("Unable to register file creation(s) - %s", files)
raise errors.LetsEncryptReverterError(
"Unable to register file creation(s) - {0}".format(files))
finally:
if new_fd is not None:
new_fd.close()
def recovery_routine(self):
"""Revert all previously modified files.
First, any changes found in self.direc['temp'] are removed,
then IN_PROGRESS changes are removed The order is important.
IN_PROGRESS is unable to add files that are already added by a TEMP
change. Thus TEMP must be rolled back first because that will be the
'latest' occurrence of the file.
"""
self.revert_temporary_config()
if os.path.isdir(self.direc['progress']):
try:
self._recover_checkpoint(self.direc['progress'])
except errors.LetsEncryptReverterError:
# We have a partial or incomplete recovery
logging.fatal("Incomplete or failed recovery for IN_PROGRESS "
"checkpoint - %s",
self.direc['progress'])
raise errors.LetsEncryptReverterError(
"Incomplete or failed recovery for IN_PROGRESS checkpoint "
"- %s" % self.direc['progress'])
def _remove_contained_files(self, file_list): # pylint: disable=no-self-use
"""Erase all files contained within file_list.
:param str file_list: file containing list of file paths to be deleted
:returns: Success
:rtype: bool
:raises :class:`letsencrypt.client.errors.LetsEncryptReverterError`: If
all files within file_list cannot be removed
"""
# Check to see that file exists to differentiate can't find file_list
# and can't remove filepaths within file_list errors.
if not os.path.isfile(file_list):
return False
try:
with open(file_list, 'r') as list_fd:
filepaths = list_fd.read().splitlines()
for path in filepaths:
# Files are registered before they are added... so
# check to see if file exists first
if os.path.lexists(path):
os.remove(path)
else:
logging.warning(
"File: %s - Could not be found to be deleted%s"
"LE probably shut down unexpectedly",
os.linesep, path)
except (IOError, OSError):
logging.fatal(
"Unable to remove filepaths contained within %s", file_list)
raise errors.LetsEncryptReverterError(
"Unable to remove filepaths contained within "
"{0}".format(file_list))
return True
def finalize_checkpoint(self, title):
"""Move IN_PROGRESS checkpoint to timestamped checkpoint.
Adds title to self.direc['progress'] CHANGES_SINCE
Move self.direc['progress'] to Backups directory and
rename the directory as a timestamp
:param str title: Title describing checkpoint
:raises :class:`letsencrypt.client.errors.LetsEncryptReverterError`
"""
# Check to make sure an "in progress" directory exists
if not os.path.isdir(self.direc['progress']):
logging.warning("No IN_PROGRESS checkpoint to finalize")
return
changes_since_path = os.path.join(
self.direc['progress'], 'CHANGES_SINCE')
changes_since_tmp_path = os.path.join(
self.direc['progress'], 'CHANGES_SINCE.tmp')
try:
with open(changes_since_tmp_path, 'w') as changes_tmp:
changes_tmp.write("-- %s --\n" % title)
with open(changes_since_path, 'r') as changes_orig:
changes_tmp.write(changes_orig.read())
shutil.move(changes_since_tmp_path, changes_since_path)
except (IOError, OSError):
logging.error("Unable to finalize checkpoint - adding title")
raise errors.LetsEncryptReverterError("Unable to add title")
self._timestamp_progress_dir()
def _timestamp_progress_dir(self):
"""Timestamp the checkpoint."""
# It is possible save checkpoints faster than 1 per second resulting in
# collisions in the naming convention.
cur_time = time.time()
for _ in range(10):
final_dir = os.path.join(self.direc['backup'], str(cur_time))
try:
os.rename(self.direc['progress'], final_dir)
return
except OSError:
# It is possible if the checkpoints are made extremely quickly
# that will result in a name collision.
# If so, increment and try again
cur_time += .01
# After 10 attempts... something is probably wrong here...
logging.error(
"Unable to finalize checkpoint, %s -> %s",
self.direc['progress'], final_dir)
raise errors.LetsEncryptReverterError(
"Unable to finalize checkpoint renaming")
letsencrypt/client/tests/apache/configurator_test.py
+6 -2
View File
@@ -23,8 +23,12 @@ class TwoVhost80Test(util.ApacheTest):
def setUp(self):
super(TwoVhost80Test, self).setUp()
self.config = util.get_apache_configurator(
self.config_path, self.config_dir, self.work_dir, self.ssl_options)
with mock.patch("letsencrypt.client.apache.configurator."
"mod_loaded") as mock_load:
mock_load.return_value = True
self.config = util.get_apache_configurator(
self.config_path, self.config_dir, self.work_dir,
self.ssl_options)
self.vh_truth = util.get_vh_truth(
self.temp_dir, "debian_apache_2_4/two_vhost_80")
letsencrypt/client/tests/apache/dvsni_test.py
+6 -2
View File
@@ -18,8 +18,12 @@ class DvsniPerformTest(util.ApacheTest):
def setUp(self):
super(DvsniPerformTest, self).setUp()
config = util.get_apache_configurator(
self.config_path, self.config_dir, self.work_dir, self.ssl_options)
with mock.patch("letsencrypt.client.apache.configurator."
"mod_loaded") as mock_load:
mock_load.return_value = True
config = util.get_apache_configurator(
self.config_path, self.config_dir, self.work_dir,
self.ssl_options)
from letsencrypt.client.apache import dvsni
self.sni = dvsni.ApacheDvsni(config)
letsencrypt/client/tests/client_test.py
+90
View File
@@ -0,0 +1,90 @@
"""letsencrypt.client.client.py tests."""
import unittest
import mock
from letsencrypt.client import errors
class RollbackTest(unittest.TestCase):
"""Test the rollback function."""
def setUp(self):
self.m_install = mock.MagicMock()
@classmethod
def _call(cls, checkpoints):
from letsencrypt.client.client import rollback
rollback(checkpoints)
@mock.patch("letsencrypt.client.client.determine_installer")
def test_no_problems(self, mock_det):
mock_det.side_effect = self.m_install
self._call(1)
self.assertEqual(self.m_install().rollback_checkpoints.call_count, 1)
self.assertEqual(self.m_install().restart.call_count, 1)
@mock.patch("letsencrypt.client.client.zope.component.getUtility")
@mock.patch("letsencrypt.client.reverter.Reverter")
@mock.patch("letsencrypt.client.client.determine_installer")
def test_misconfiguration_fixed(self, mock_det, mock_rev, mock_input):
mock_det.side_effect = [errors.LetsEncryptMisconfigurationError,
self.m_install]
mock_input().generic_yesno.return_value = True
self._call(1)
# Don't rollback twice... (only on one object)
self.assertEqual(self.m_install().rollback_checkpoints.call_count, 0)
self.assertEqual(mock_rev().rollback_checkpoints.call_count, 1)
# Only restart once
self.assertEqual(self.m_install.restart.call_count, 1)
@mock.patch("letsencrypt.client.client.zope.component.getUtility")
@mock.patch("letsencrypt.client.client.logging.warning")
@mock.patch("letsencrypt.client.reverter.Reverter")
@mock.patch("letsencrypt.client.client.determine_installer")
def test_misconfiguration_remains(
self, mock_det, mock_rev, mock_warn, mock_input):
mock_det.side_effect = errors.LetsEncryptMisconfigurationError
mock_input().generic_yesno.return_value = True
self._call(1)
# Don't rollback twice... (only on one object)
self.assertEqual(self.m_install().rollback_checkpoints.call_count, 0)
self.assertEqual(mock_rev().rollback_checkpoints.call_count, 1)
# Never call restart because init never succeeds
self.assertEqual(self.m_install().restart.call_count, 0)
# There should be a warning about the remaining problem
self.assertEqual(mock_warn.call_count, 1)
@mock.patch("letsencrypt.client.client.zope.component.getUtility")
@mock.patch("letsencrypt.client.reverter.Reverter")
@mock.patch("letsencrypt.client.client.determine_installer")
def test_user_decides_to_manually_investigate(
self, mock_det, mock_rev, mock_input):
mock_det.side_effect = errors.LetsEncryptMisconfigurationError
mock_input().generic_yesno.return_value = False
self._call(1)
# Neither is ever called
self.assertEqual(self.m_install().rollback_checkpoints.call_count, 0)
self.assertEqual(mock_rev().rollback_checkpoints.call_count, 0)
@mock.patch("letsencrypt.client.client.determine_installer")
def test_no_installer(self, mock_det):
mock_det.return_value = None
# Just make sure no exceptions are raised
self._call(1)
if __name__ == '__main__':
unittest.main()
letsencrypt/client/tests/reverter_test.py
+457
View File
@@ -0,0 +1,457 @@
"""Test letsencrypt.client.reverter."""
import logging
import os
import shutil
import tempfile
import unittest
import mock
from letsencrypt.client import errors
class ReverterCheckpointLocalTest(unittest.TestCase):
# pylint: disable=too-many-instance-attributes
"""Test the Reverter Class."""
def setUp(self):
from letsencrypt.client.reverter import Reverter
# Disable spurious errors... we are trying to test for them
logging.disable(logging.CRITICAL)
self.work_dir, self.direc = setup_work_direc()
self.reverter = Reverter(self.direc)
tup = setup_test_files()
self.config1, self.config2, self.dir1, self.dir2, self.sets = tup
def tearDown(self):
shutil.rmtree(self.work_dir)
shutil.rmtree(self.dir1)
shutil.rmtree(self.dir2)
logging.disable(logging.NOTSET)
def test_basic_add_to_temp_checkpoint(self):
# These shouldn't conflict even though they are both named config.txt
self.reverter.add_to_temp_checkpoint(self.sets[0], "save1")
self.reverter.add_to_temp_checkpoint(self.sets[1], "save2")
self.assertTrue(os.path.isdir(self.reverter.direc['temp']))
self.assertEqual(get_save_notes(self.direc['temp']), "save1save2")
self.assertFalse(os.path.isfile(
os.path.join(self.direc['temp'], "NEW_FILES")))
self.assertEqual(
get_filepaths(self.direc['temp']),
"{0}\n{1}\n".format(self.config1, self.config2))
def test_add_to_checkpoint_copy_failure(self):
with mock.patch("letsencrypt.client.reverter."
"shutil.copy2") as mock_copy2:
mock_copy2.side_effect = IOError("bad copy")
self.assertRaises(errors.LetsEncryptReverterError,
self.reverter.add_to_checkpoint,
self.sets[0],
"save1")
def test_checkpoint_conflict(self):
"""Make sure that checkpoint errors are thrown appropriately."""
config3 = os.path.join(self.dir1, "config3.txt")
self.reverter.register_file_creation(True, config3)
update_file(config3, "This is a new file!")
self.reverter.add_to_checkpoint(self.sets[2], "save1")
# This shouldn't throw an error
self.reverter.add_to_temp_checkpoint(self.sets[0], "save2")
# Raise error
self.assertRaises(
errors.LetsEncryptReverterError, self.reverter.add_to_checkpoint,
self.sets[2], "save3")
# Should not cause an error
self.reverter.add_to_checkpoint(self.sets[1], "save4")
# Check to make sure new files are also checked...
self.assertRaises(
errors.LetsEncryptReverterError,
self.reverter.add_to_checkpoint,
set([config3]), "invalid save")
def test_multiple_saves_and_temp_revert(self):
self.reverter.add_to_temp_checkpoint(self.sets[0], "save1")
update_file(self.config1, "updated-directive")
self.reverter.add_to_temp_checkpoint(self.sets[0], "save2-updated dir")
update_file(self.config1, "new directive change that we won't keep")
self.reverter.revert_temporary_config()
self.assertEqual(read_in(self.config1), "directive-dir1")
def test_multiple_registration_fail_and_revert(self):
# pylint: disable=invalid-name
config3 = os.path.join(self.dir1, "config3.txt")
update_file(config3, "Config3")
config4 = os.path.join(self.dir2, "config4.txt")
update_file(config4, "Config4")
# Test multiple registrations and two registrations at once
self.reverter.register_file_creation(True, self.config1)
self.reverter.register_file_creation(True, self.config2)
self.reverter.register_file_creation(True, config3, config4)
# Simulate Let's Encrypt crash... recovery routine is run
self.reverter.recovery_routine()
self.assertFalse(os.path.isfile(self.config1))
self.assertFalse(os.path.isfile(self.config2))
self.assertFalse(os.path.isfile(config3))
self.assertFalse(os.path.isfile(config4))
def test_multiple_registration_same_file(self):
self.reverter.register_file_creation(True, self.config1)
self.reverter.register_file_creation(True, self.config1)
self.reverter.register_file_creation(True, self.config1)
self.reverter.register_file_creation(True, self.config1)
files = get_new_files(self.direc['temp'])
self.assertEqual(len(files), 1)
def test_register_file_creation_write_error(self):
m_open = mock.mock_open()
with mock.patch("letsencrypt.client.reverter.open",
m_open, create=True):
m_open.side_effect = OSError("bad open")
self.assertRaises(errors.LetsEncryptReverterError,
self.reverter.register_file_creation,
True, self.config1)
def test_bad_registration(self):
# Made this mistake and want to make sure it doesn't happen again...
self.assertRaises(errors.LetsEncryptReverterError,
self.reverter.register_file_creation,
"filepath")
def test_recovery_routine_in_progress_failure(self):
self.reverter.add_to_checkpoint(self.sets[0], "perm save")
# pylint: disable=protected-access
self.reverter._recover_checkpoint = mock.MagicMock(
side_effect=errors.LetsEncryptReverterError)
self.assertRaises(errors.LetsEncryptReverterError,
self.reverter.recovery_routine)
def test_recover_checkpoint_revert_temp_failures(self):
# pylint: disable=invalid-name
mock_recover = mock.MagicMock(
side_effect=errors.LetsEncryptReverterError("e"))
# pylint: disable=protected-access
self.reverter._recover_checkpoint = mock_recover
self.reverter.add_to_temp_checkpoint(self.sets[0], "config1 save")
self.assertRaises(errors.LetsEncryptReverterError,
self.reverter.revert_temporary_config)
def test_recover_checkpoint_rollback_failure(self):
mock_recover = mock.MagicMock(
side_effect=errors.LetsEncryptReverterError("e"))
# pylint: disable=protected-access
self.reverter._recover_checkpoint = mock_recover
self.reverter.add_to_checkpoint(self.sets[0], "config1 save")
self.reverter.finalize_checkpoint("Title")
self.assertRaises(errors.LetsEncryptReverterError,
self.reverter.rollback_checkpoints, 1)
def test_recover_checkpoint_copy_failure(self):
self.reverter.add_to_temp_checkpoint(self.sets[0], "save1")
with mock.patch("letsencrypt.client.reverter.shutil."
"copy2") as mock_copy2:
mock_copy2.side_effect = OSError("bad copy")
self.assertRaises(errors.LetsEncryptReverterError,
self.reverter.revert_temporary_config)
def test_recover_checkpoint_rm_failure(self):
self.reverter.add_to_temp_checkpoint(self.sets[0], "temp save")
with mock.patch("letsencrypt.client.reverter.shutil."
"rmtree") as mock_rmtree:
mock_rmtree.side_effect = OSError("Cannot remove tree")
self.assertRaises(errors.LetsEncryptReverterError,
self.reverter.revert_temporary_config)
@mock.patch("letsencrypt.client.reverter.logging.warning")
def test_recover_checkpoint_missing_new_files(self, mock_warn):
self.reverter.register_file_creation(
True, os.path.join(self.dir1, "missing_file.txt"))
self.reverter.revert_temporary_config()
self.assertEqual(mock_warn.call_count, 1)
@mock.patch("letsencrypt.client.reverter.os.remove")
def test_recover_checkpoint_remove_failure(self, mock_remove):
self.reverter.register_file_creation(True, self.config1)
mock_remove.side_effect = OSError("Can't remove")
self.assertRaises(errors.LetsEncryptReverterError,
self.reverter.revert_temporary_config)
def test_recovery_routine_temp_and_perm(self):
# Register a new perm checkpoint file
config3 = os.path.join(self.dir1, "config3.txt")
self.reverter.register_file_creation(False, config3)
update_file(config3, "This is a new perm file!")
# Add changes to perm checkpoint
self.reverter.add_to_checkpoint(self.sets[0], "perm save1")
update_file(self.config1, "updated perm config1")
self.reverter.add_to_checkpoint(self.sets[1], "perm save2")
update_file(self.config2, "updated perm config2")
# Add changes to a temporary checkpoint
self.reverter.add_to_temp_checkpoint(self.sets[0], "temp save1")
update_file(self.config1, "second update now temp config1")
# Register a new temp checkpoint file
config4 = os.path.join(self.dir2, "config4.txt")
self.reverter.register_file_creation(True, config4)
update_file(config4, "New temporary file!")
# Now erase everything
self.reverter.recovery_routine()
# Now Run tests
# These were new files.. they should be removed
self.assertFalse(os.path.isfile(config3))
self.assertFalse(os.path.isfile(config4))
# Check to make sure everything got rolled back appropriately
self.assertEqual(read_in(self.config1), "directive-dir1")
self.assertEqual(read_in(self.config2), "directive-dir2")
class TestFullCheckpointsReverter(unittest.TestCase):
# pylint: disable=too-many-instance-attributes
"""Tests functions having to deal with full checkpoints."""
def setUp(self):
from letsencrypt.client.reverter import Reverter
# Disable spurious errors...
logging.disable(logging.CRITICAL)
self.work_dir, self.direc = setup_work_direc()
self.reverter = Reverter(self.direc)
tup = setup_test_files()
self.config1, self.config2, self.dir1, self.dir2, self.sets = tup
def tearDown(self):
shutil.rmtree(self.work_dir)
shutil.rmtree(self.dir1)
shutil.rmtree(self.dir2)
logging.disable(logging.NOTSET)
def test_rollback_improper_inputs(self):
self.assertRaises(
errors.LetsEncryptReverterError,
self.reverter.rollback_checkpoints, "-1")
self.assertRaises(
errors.LetsEncryptReverterError,
self.reverter.rollback_checkpoints, -1000)
self.assertRaises(
errors.LetsEncryptReverterError,
self.reverter.rollback_checkpoints, "one")
def test_rollback_finalize_checkpoint_valid_inputs(self):
# pylint: disable=invalid-name
config3 = self._setup_three_checkpoints()
# Check resulting backup directory
self.assertEqual(len(os.listdir(self.direc['backup'])), 3)
# Check rollbacks
# First rollback
self.reverter.rollback_checkpoints(1)
self.assertEqual(read_in(self.config1), "update config1")
self.assertEqual(read_in(self.config2), "update config2")
# config3 was not included in checkpoint
self.assertEqual(read_in(config3), "Final form config3")
# Second rollback
self.reverter.rollback_checkpoints(1)
self.assertEqual(read_in(self.config1), "update config1")
self.assertEqual(read_in(self.config2), "directive-dir2")
self.assertFalse(os.path.isfile(config3))
# One dir left... check title
all_dirs = os.listdir(self.direc['backup'])
self.assertEqual(len(all_dirs), 1)
self.assertTrue(
"First Checkpoint" in get_save_notes(
os.path.join(self.direc['backup'], all_dirs[0])))
# Final rollback
self.reverter.rollback_checkpoints(1)
self.assertEqual(read_in(self.config1), "directive-dir1")
@mock.patch("letsencrypt.client.reverter.logging.warning")
def test_finalize_checkpoint_no_in_progress(self, mock_warn):
self.reverter.finalize_checkpoint("No checkpoint... should warn")
self.assertEqual(mock_warn.call_count, 1)
@mock.patch("letsencrypt.client.reverter.shutil.move")
def test_finalize_checkpoint_cannot_title(self, mock_move):
self.reverter.add_to_checkpoint(self.sets[0], "perm save")
mock_move.side_effect = OSError("cannot move")
self.assertRaises(errors.LetsEncryptReverterError,
self.reverter.finalize_checkpoint,
"Title")
@mock.patch("letsencrypt.client.reverter.os.rename")
def test_finalize_checkpoint_no_rename_directory(self, mock_rename):
# pylint: disable=invalid-name
self.reverter.add_to_checkpoint(self.sets[0], "perm save")
mock_rename.side_effect = OSError
self.assertRaises(errors.LetsEncryptReverterError,
self.reverter.finalize_checkpoint,
"Title")
@mock.patch("letsencrypt.client.reverter.logging")
def test_rollback_too_many(self, mock_logging):
self.reverter.rollback_checkpoints(1)
self.assertEqual(mock_logging.warning.call_count, 1)
def test_multi_rollback(self):
config3 = self._setup_three_checkpoints()
self.reverter.rollback_checkpoints(3)
self.assertEqual(read_in(self.config1), "directive-dir1")
self.assertEqual(read_in(self.config2), "directive-dir2")
self.assertFalse(os.path.isfile(config3))
@mock.patch("letsencrypt.client.client.zope.component.getUtility")
def test_view_config_changes(self, mock_output):
"""This is not strict as this is subject to change."""
self._setup_three_checkpoints()
# Make sure it doesn't throw any errors
self.reverter.view_config_changes()
# Make sure notification is output
self.assertEqual(mock_output().generic_notification.call_count, 1)
@mock.patch("letsencrypt.client.reverter.logging")
def test_view_config_changes_no_backups(self, mock_logging):
self.reverter.view_config_changes()
self.assertTrue(mock_logging.info.call_count > 0)
def test_view_config_changes_bad_backups_dir(self):
# There shouldn't be any "in progess directories when this is called
# It must just be clean checkpoints
os.makedirs(os.path.join(self.direc['backup'], "in_progress"))
self.assertRaises(errors.LetsEncryptReverterError,
self.reverter.view_config_changes)
def _setup_three_checkpoints(self):
"""Generate some finalized checkpoints."""
# Checkpoint1 - config1
self.reverter.add_to_checkpoint(self.sets[0], "first save")
self.reverter.finalize_checkpoint("First Checkpoint")
update_file(self.config1, "update config1")
# Checkpoint2 - new file config3, update config2
config3 = os.path.join(self.dir1, "config3.txt")
self.reverter.register_file_creation(False, config3)
update_file(config3, "directive-config3")
self.reverter.add_to_checkpoint(self.sets[1], "second save")
self.reverter.finalize_checkpoint("Second Checkpoint")
update_file(self.config2, "update config2")
update_file(config3, "update config3")
# Checkpoint3 - update config1, config2
self.reverter.add_to_checkpoint(self.sets[2], "third save")
self.reverter.finalize_checkpoint("Third Checkpoint - Save both")
update_file(self.config1, "Final form config1")
update_file(self.config2, "Final form config2")
update_file(config3, "Final form config3")
return config3
class QuickInitReverterTest(unittest.TestCase):
"""Quick test of init."""
def test_init(self):
from letsencrypt.client.reverter import Reverter
rev = Reverter()
# Verify direc is set
self.assertTrue(rev.direc['backup'])
self.assertTrue(rev.direc['temp'])
self.assertTrue(rev.direc['progress'])
def setup_work_direc():
"""Setup directories."""
work_dir = tempfile.mkdtemp("work")
backup = os.path.join(work_dir, "backup")
os.makedirs(backup)
direc = {'backup': backup,
'temp': os.path.join(work_dir, "temp"),
'progress': os.path.join(backup, "progress")}
return work_dir, direc
def setup_test_files():
"""Setup sample configuration files."""
dir1 = tempfile.mkdtemp("dir1")
dir2 = tempfile.mkdtemp("dir2")
config1 = os.path.join(dir1, "config.txt")
config2 = os.path.join(dir2, "config.txt")
with open(config1, 'w') as file_fd:
file_fd.write("directive-dir1")
with open(config2, 'w') as file_fd:
file_fd.write("directive-dir2")
sets = [set([config1]),
set([config2]),
set([config1, config2])]
return config1, config2, dir1, dir2, sets
def get_save_notes(dire):
"""Read save notes"""
return read_in(os.path.join(dire, 'CHANGES_SINCE'))
def get_filepaths(dire):
"""Get Filepaths"""
return read_in(os.path.join(dire, 'FILEPATHS'))
def get_new_files(dire):
"""Get new files."""
return read_in(os.path.join(dire, 'NEW_FILES')).splitlines()
def read_in(path):
"""Read in a file, return the str"""
with open(path, 'r') as file_fd:
return file_fd.read()
def update_file(filename, string):
"""Update a file with a new value."""
with open(filename, 'w') as file_fd:
file_fd.write(string)
if __name__ == '__main__':
unittest.main()
letsencrypt/scripts/main.py
+29 -49
View File
@@ -11,13 +11,12 @@ import zope.interface
from letsencrypt.client import CONFIG
from letsencrypt.client import client
from letsencrypt.client import display
from letsencrypt.client import errors
from letsencrypt.client import interfaces
from letsencrypt.client import log
from letsencrypt.client import revoker
from letsencrypt.client.apache import configurator
def main(): # pylint: disable=too-many-statements
def main(): # pylint: disable=too-many-statements,too-many-branches
"""Command line argument parsing and main script execution."""
parser = argparse.ArgumentParser(
description="An ACME client that can update Apache configurations.")
@@ -77,29 +76,36 @@ def main(): # pylint: disable=too-many-statements
displayer = display.FileDisplay(sys.stdout)
zope.component.provideUtility(displayer)
installer = determine_installer()
if args.view_config_changes:
client.view_config_changes()
sys.exit()
if args.revoke:
revoc = revoker.Revoker(args.server, installer)
revoc.list_certs_keys()
client.revoke(args.server)
sys.exit()
if args.rollback > 0:
rollback(installer, args.rollback)
sys.exit()
if args.view_config_changes:
view_config_changes(installer)
client.rollback(args.rollback)
sys.exit()
if not args.eula:
display_eula()
# Make sure we actually get an installer that is functioning properly
# before we begin to try to use it.
try:
installer = client.determine_installer()
except errors.LetsEncryptMisconfigurationError as err:
logging.fatal("Please fix your configuration before proceeding. "
"The Installer exited with the following message: "
"%s", err)
sys.exit(1)
# Use the same object if possible
if interfaces.IAuthenticator.providedBy(installer): # pylint: disable=no-member
auth = installer
else:
auth = determine_authenticator()
auth = client.determine_authenticator()
domains = choose_names(installer) if args.domains is None else args.domains
@@ -114,9 +120,16 @@ def main(): # pylint: disable=too-many-statements
# Validate the key and csr
client.validate_key_csr(privkey)
cert_file, chain_file = acme.obtain_certificate(domains)
acme.deploy_certificate(domains, privkey, cert_file, chain_file)
acme.enhance_config(domains, args.redirect)
# This more closely mimics the capabilities of the CLI
# It should be possible for reconfig only, install-only, no-install
# I am not sure the best way to handle all of the unimplemented abilities,
# but this code should be safe on all environments.
if auth is not None:
cert_file, chain_file = acme.obtain_certificate(domains)
if installer is not None and cert_file is not None:
acme.deploy_certificate(domains, privkey, cert_file, chain_file)
if installer is not None:
acme.enhance_config(domains, args.redirect)
def display_eula():
@@ -134,8 +147,7 @@ def choose_names(installer):
:type installer: :class:`letsencrypt.client.interfaces.IInstaller`
"""
# This function adds all names
# found within the config to self.names
# This function adds all names found in the installer configuration
# Then filters them based on user selection
code, names = zope.component.getUtility(
interfaces.IDisplay).filter_names(get_all_names(installer))
@@ -165,16 +177,6 @@ def get_all_names(installer):
return names
# This should be controlled by commandline parameters
def determine_authenticator():
"""Returns a valid IAuthenticator."""
return configurator.ApacheConfigurator()
def determine_installer():
"""Returns a valid IInstaller."""
return configurator.ApacheConfigurator()
def read_file(filename):
"""Returns the given file's contents with universal new line support.
@@ -193,27 +195,5 @@ def read_file(filename):
raise argparse.ArgumentTypeError(exc.strerror)
def rollback(installer, checkpoints):
"""Revert configuration the specified number of checkpoints.
:param installer: Installer object
:type installer: :class:`letsencrypt.client.interfaces.IInstaller`
:param int checkpoints: Number of checkpoints to revert.
"""
installer.rollback_checkpoints(checkpoints)
installer.restart()
def view_config_changes(installer):
"""View checkpoints and associated configuration changes.
:param installer: Installer object
:type installer: :class:`letsencrypt.client.interfaces.IInstaller`
"""
installer.view_config_changes()
if __name__ == "__main__":
main()
tox.ini
+1 -1
View File
@@ -14,7 +14,7 @@ commands =
[testenv:cover]
commands =
python setup.py dev
python setup.py nosetests --with-coverage --cover-min-percentage=61
python setup.py nosetests --with-coverage --cover-min-percentage=66
[testenv:lint]
commands =