starred/certbot-certbot
1
0
Fork 0
mirror of https://github.com/certbot/certbot.git synced 2026-07-27 16:30:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

Initial implementation of reverter, showing main.py capabilities

Browse Source
This commit is contained in:
James Kasten
2015-01-22 02:36:36 -08:00
parent 1dc7bfccc4
commit bedbd2e315
12 changed files with 445 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files
letsencrypt/client/CONFIG.py
+11 -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"""
@@ -116,3 +101,11 @@ APACHE_CTL = "/usr/sbin/apache2ctl"
APACHE2 = "/etc/init.d/apache2"
"""Command used for reload and restart."""
# Static Strings/Messages
CERT_DELETE_MSG = "This certificate has either been deleted or moved"
"""Used in revocation cert dict for 'installed'.
Indicates that the original certificate has been moved/deleted.
"""
letsencrypt/client/apache/configurator.py
-1
View File
@@ -1028,7 +1028,6 @@ 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
apache_restart()
except (OSError, subprocess.CalledProcessError) as err:
logging.error("Error enabling mod_%s", mod_name)
letsencrypt/client/apache/parser.py
+5 -6
View File
@@ -240,25 +240,24 @@ class ApacheParser(object):
regex = regex + letter
return regex
def _parse_file(self, file_path):
def _parse_file(self, filepath):
"""Parse file with Augeas
Checks to see if file_path is parsed by Augeas
If file_path isn't parsed, the file is added and Augeas is reloaded
If filepath isn't parsed, the file is added and Augeas is reloaded
:param str file_path: Apache config file path
:param str filepath: Apache config file path
"""
# Test if augeas included file for Httpd.lens
# Note: This works for augeas globs, ie. *.conf
inc_test = self.aug.match(
"/augeas/load/Httpd/incl [. ='%s']" % file_path)
"/augeas/load/Httpd/incl [. ='%s']" % filepath)
if not inc_test:
# Load up files
# self.httpd_incl.append(file_path)
# self.aug.add_transform("Httpd.lns",
# self.httpd_incl, None, self.httpd_excl)
self._add_httpd_transform(file_path)
self._add_httpd_transform(filepath)
self.aug.load()
def standardize_excl(self):
letsencrypt/client/augeas_configurator.py
+14 -7
View File
@@ -40,7 +40,10 @@ class AugeasConfigurator(object):
# TODO: this instantiation can be optimized to only load
# relevant files - I believe -> NO_MODL_AUTOLOAD
# Set Augeas flags to save backup
self.aug = augeas.Augeas(flags=augeas.Augeas.NONE)
my_flags = augeas.Augeas.NONE | augeas.Augeas.NO_MODL_AUTOLOAD
self.aug = augeas.Augeas(flags=my_flags)
self.aug.add_transform("Httpd.lns", "/etc/apache2/apache2.conf")
self.save_notes = ""
# See if any temporary changes need to be recovered
@@ -61,13 +64,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.
@@ -92,7 +95,7 @@ class AugeasConfigurator(object):
# This is a noop save
self.aug.save()
except (RuntimeError, IOError):
self._log_save_errors()
self._log_save_errors(ex_errs)
# Erase Save Notes
self.save_notes = ""
return False
@@ -125,8 +128,12 @@ class AugeasConfigurator(object):
return True
def _log_save_errors(self):
"""Log errors due to bad Augeas save."""
def _log_save_errors(self, ex_errs):
"""Log errors due to bad Augeas save.
:param list ex_errs: Existing errors before save
"""
# Check for the root of save problems
new_errs = self.aug.match("/augeas//error")
# logging.error("During Save - %s", mod_conf)
letsencrypt/client/crypto_util.py
+20 -24
View File
@@ -45,20 +45,20 @@ 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")))
return {
"nonce": le_util.jose_b64encode(nonce),
"alg": "RS256",
"jwk": {
"kty": "RSA",
"n": le_util.jose_b64encode(n_bytes),
"e": le_util.jose_b64encode(e_bytes),
'nonce': le_util.jose_b64encode(nonce),
'alg': "RS256",
'jwk': {
'kty': "RSA",
'n': le_util.jose_b64encode(n_bytes),
'e': le_util.jose_b64encode(e_bytes),
},
"sig": le_util.jose_b64encode(signature),
'sig': le_util.jose_b64encode(signature),
}
@@ -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=CONFIG.RSA_KEY_SIZE):
"""Generate PEM encoded RSA key.
@@ -154,10 +153,6 @@ def make_key(bits=CONFIG.RSA_KEY_SIZE):
: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,8 +223,9 @@ def make_ss_cert(key_str, domains, not_before=None,
def get_cert_info(filename):
"""Get certificate info.
:param str filename: Name of file containing certificate in PEM format.
.. 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
"""
@@ -237,20 +233,20 @@ def get_cert_info(filename):
cert = M2Crypto.X509.load_cert(filename)
try:
san = cert.get_ext("subjectAltName").get_value()
san = cert.get_ext('subjectAltName').get_value()
except:
san = ""
return {
"not_before": cert.get_not_before().get_datetime(),
"not_after": cert.get_not_after().get_datetime(),
"subject": cert.get_subject().as_text(),
"cn": cert.get_subject().CN,
"issuer": cert.get_issuer().as_text(),
"fingerprint": cert.get_fingerprint(md='sha1'),
"san": san,
"serial": cert.get_serial_number(),
"pub_key": "RSA " + str(cert.get_pubkey().size() * 8),
'not_before': cert.get_not_before().get_datetime(),
'not_after': cert.get_not_after().get_datetime(),
'subject': cert.get_subject().as_text(),
'cn': cert.get_subject().CN,
'issuer': cert.get_issuer().as_text(),
'fingerprint': cert.get_fingerprint(md='sha1'),
'san': san,
'serial': cert.get_serial_number(),
'pub_key': "RSA " + str(cert.get_pubkey().size() * 8),
}
letsencrypt/client/display.py
+15 -6
View File
@@ -3,6 +3,7 @@ import textwrap
import dialog
import zope.interface
from letsencrypt.client import CONFIG
from letsencrypt.client import interfaces
@@ -55,12 +56,20 @@ class NcursesDisplay(object):
+ gen_https_names(domains) + "!", width=self.width)
def display_certs(self, certs):
list_choices = [
(str(i+1), "%s | %s | %s" %
(str(c["cn"].ljust(self.width - 39)),
c["not_before"].strftime("%m-%d-%y"),
"Installed" if c["installed"] else ""))
for i, c in enumerate(certs)]
list_choices = []
for i, c in enumerate(certs):
if c['installed']:
if c['installed'] == CONFIG.CERT_DELETE_MSG:
status = "Deleted/Moved"
else:
status = "Installed"
else:
status = ""
list_choices.append((str(i+1), "{0} | {1} | {2}".format(
str(c['cn'].ljust(self.width-39)),
c['not_before'].strftime("%m-%d-%y"),
status)))
code, tag = self.dialog.menu(
"Which certificates would you like to revoke?",
letsencrypt/client/reverter.py
+100 -59
View File
@@ -13,9 +13,9 @@ 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}
direc = {'backup': CONFIG.BACKUP_DIR,
'temp': CONFIG.TEMP_CHECKPOINT_DIR,
'progress': CONFIG.IN_PROGRESS_DIR}
self.direc = direc
def revert_temporary_config(self):
@@ -28,31 +28,34 @@ class Reverter(object):
Unable to revert config
"""
if os.path.isdir(self.direc["temp"]):
result = self._recover_checkpoint(self.direc["temp"])
if result != 0:
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"])
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
:param int rollback: Number of checkpoints to reverse. A str num will be
cast to an integer. So '2' is also acceptable.
"""
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 < 1:
if rollback < 0:
logging.error("Rollback argument must be a positive integer")
return
raise errors.LetsEncryptReverterError("Invalid Input")
backups = os.listdir(self.direc["backup"])
backups = os.listdir(self.direc['backup'])
backups.sort()
if len(backups) < rollback:
@@ -60,11 +63,13 @@ class Reverter(object):
rollback, len(backups))
while rollback > 0 and backups:
cp_dir = self.direc["backup"] + backups.pop()
result = self._recover_checkpoint(cp_dir)
if result != 0:
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")
sys.exit(39)
raise errors.LetsEncryptReverterError(
"Unable to load checkpoint during rollback")
rollback -= 1
def view_config_changes(self):
@@ -77,42 +82,46 @@ class Reverter(object):
called.
"""
backups = os.listdir(self.direc["backup"])
backups = os.listdir(self.direc['backup'])
backups.sort(reverse=True)
if not backups:
print ("Letsencrypt has not saved any backups of your "
"configuration")
logging.info("Letsencrypt has not saved any backups of your "
"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']
raise errors.LetsEncryptReverterError(
"Invalid directories in {}".format(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:
cur_dir = os.path.join(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:
with open(os.path.join(cur_dir, "FILEPATHS")) as paths_fd:
filepaths = paths_fd.read().splitlines()
for path in filepaths:
print " %s" % path
print " {}".format(path)
try:
if os.path.isfile(os.path.join(cur_dir, 'NEW_FILES')):
with open(os.path.join(cur_dir, 'NEW_FILES')) as new_fd:
if os.path.isfile(os.path.join(cur_dir, "NEW_FILES")):
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
print " {}".format(path)
except (IOError, OSError) as err:
logging.warn(str(err))
print ""
logging.error(str(err))
raise errors.LetsEncryptReverterError(
"Unable to read new files in checkpoint"
"- {}".format(cur_dir))
print "\n"
def add_to_temp_checkpoint(self, save_files, save_notes):
"""Add files to temporary checkpoint
@@ -130,6 +139,8 @@ class Reverter(object):
: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)
@@ -141,7 +152,6 @@ class Reverter(object):
:param str save_notes: notes about changes made during the save
"""
self._check_tempfile_saves(save_files)
le_util.make_or_verify_dir(cp_dir, 0o755, os.geteuid())
existing_filepaths = []
@@ -156,6 +166,8 @@ class Reverter(object):
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
@@ -177,8 +189,7 @@ class Reverter(object):
:param str cp_dir: checkpoint directory file path
:returns: 0 success, 1 Unable to revert, -1 Unable to delete
:rtype: int
:raises errors.LetsEncryptReverterError: If unable to recover checkpoint
"""
if os.path.isfile(os.path.join(cp_dir, "FILEPATHS")):
@@ -192,7 +203,8 @@ class Reverter(object):
except (IOError, OSError):
# This file is required in all checkpoints.
logging.error("Unable to recover files from %s", cp_dir)
return 1
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"))
@@ -201,9 +213,8 @@ class Reverter(object):
shutil.rmtree(cp_dir)
except OSError:
logging.error("Unable to remove directory: %s", cp_dir)
return -1
return 0
raise errors.LetsEncryptReverterError(
"Unable to remove directory: %s" % cp_dir)
def _check_tempfile_saves(self, save_files): # pylint: disable=no-self-use
"""Verify save isn't overwriting any temporary files.
@@ -214,15 +225,26 @@ class Reverter(object):
when save is attempting to overwrite a temporary file.
"""
temp_path = "%sFILEPATHS" % self.direc["temp"]
protected_files = []
# Check 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 = protected_fd.read().splitlines()
for filename in protected_files:
if filename in save_files:
raise errors.LetsEncryptReverterError(
"Attempting to overwrite challenge "
"file - %s" % filename)
protected_files.extend(protected_fd.read().splitlines())
# Check 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)
# pylint: disable=no-self-use, anomalous-backslash-in-string
def register_file_creation(self, temporary, *files):
@@ -239,9 +261,9 @@ class Reverter(object):
"""
if temporary:
cp_dir = self.direc["temp"]
cp_dir = self.direc['temp']
else:
cp_dir = self.direc["progress"]
cp_dir = self.direc['progress']
le_util.make_or_verify_dir(cp_dir, 0o755, os.geteuid())
try:
@@ -262,15 +284,17 @@ class Reverter(object):
"""
self.revert_temporary_config()
if os.path.isdir(self.direc["progress"]):
result = self._recover_checkpoint(self.direc["progress"])
if result != 0:
if os.path.isdir(self.direc['progress']):
try:
self._recover_checkpoint(self.direc['progress'])
except errors.LetsEncryptReverterError:
# 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.direc['progress'])
raise errors.LetsEncryptReverterError(
"Incomplete or failed recovery for "
"%s" % self.direc['progress'])
# pylint: disable=no-self-use
def _remove_contained_files(self, file_list):
@@ -310,14 +334,14 @@ class Reverter(object):
"""Move IN_PROGRESS checkpoint to timestamped checkpoint.
Adds title to self.direc['progress'] CHANGES_SINCE
Move self.direc['progress'] to Backups directory and rename with timestamp
Move self.direc['progress'] to Backups directory and
rename the directory as a timestamp
"""
# Check to make sure an "in progress" directory exists
if not os.path.isdir(self.direc['progress']):
return
final_dir = os.path.join(self.direc['backup'], str(time.time()))
changes_since_path = os.path.join(
self.direc['progress'], 'CHANGES_SINCE')
changes_since_tmp_path = os.path.join(
@@ -334,10 +358,27 @@ class Reverter(object):
except (IOError, OSError):
logging.error("Unable to finalize checkpoint - adding title")
raise errors.LetsEncryptReverterError("Unable to add title")
try:
os.rename(self.direc['progress'], final_dir)
except OSError:
logging.error(
"Unable to finalize checkpoint, %s -> %s", cp_dir, final_dir)
raise errors.LetsEncryptReverterError(
"Unable to finalize checkpoint renaming")
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 i 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
# There will be a naming collision, 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/revoker.py
+30 -25
View File
@@ -30,16 +30,16 @@ class Revoker(object):
:rtype: dict
"""
cert_der = M2Crypto.X509.load_cert(cert["backup_cert_file"]).as_der()
with open(cert["backup_key_file"], 'rU') as backup_key_file:
cert_der = M2Crypto.X509.load_cert(cert['backup_cert_file']).as_der()
with open(cert['backup_key_file'], 'rU') as backup_key_file:
key = backup_key_file.read()
revocation = self.network.send_and_receive_expected(
acme.revocation_request(cert_der, key), "revocation")
acme.revocation_request(cert_der, key), 'revocation')
zope.component.getUtility(interfaces.IDisplay).generic_notification(
"You have successfully revoked the certificate for "
"%s" % cert["cn"])
"%s" % cert['cn'])
self.remove_cert_key(cert)
self.list_certs_keys()
@@ -57,30 +57,35 @@ class Revoker(object):
return
c_sha1_vh = {}
for (cert, _, path) in self.installer.get_all_certs_keys():
try:
c_sha1_vh[M2Crypto.X509.load_cert(
cert).get_fingerprint(md='sha1')] = path
except:
continue
if self.installer is not None:
for (cert, _, path) in self.installer.get_all_certs_keys():
try:
c_sha1_vh[M2Crypto.X509.load_cert(
cert).get_fingerprint(md='sha1')] = path
except:
continue
with open(list_file, 'rb') as csvfile:
csvreader = csv.reader(csvfile)
for row in csvreader:
cert = crypto_util.get_cert_info(row[1])
# Generate backup key/cert names
b_k = os.path.join(CONFIG.CERT_KEY_BACKUP,
os.path.basename(row[2]) + "_" + row[0])
os.path.basename(row[2]) + '_' + row[0])
b_c = os.path.join(CONFIG.CERT_KEY_BACKUP,
os.path.basename(row[1]) + "_" + row[0])
os.path.basename(row[1]) + '_' + row[0])
cert = crypto_util.get_cert_info(b_c)
if not os.path.isfile(row[1]):
cert['installed'] = CONFIG.CERT_DELETE_MSG
cert.update({
"orig_key_file": row[2],
"orig_cert_file": row[1],
"idx": int(row[0]),
"backup_key_file": b_k,
"backup_cert_file": b_c,
"installed": c_sha1_vh.get(cert["fingerprint"], ""),
'orig_key_file': row[2],
'orig_cert_file': row[1],
'idx': int(row[0]),
'backup_key_file': b_k,
'backup_cert_file': b_c,
'installed': c_sha1_vh.get(
cert['fingerprint'], cert.get('installed', "")),
})
certs.append(cert)
if certs:
@@ -129,12 +134,12 @@ class Revoker(object):
csvwriter = csv.writer(newfile)
for row in csvreader:
if not (row[0] == str(cert["idx"]) and
row[1] == cert["orig_cert_file"] and
row[2] == cert["orig_key_file"]):
if not (row[0] == str(cert['idx']) and
row[1] == cert['orig_cert_file'] and
row[2] == cert['orig_key_file']):
csvwriter.writerow(row)
shutil.copy2(list_file2, list_file)
os.remove(list_file2)
os.remove(cert["backup_cert_file"])
os.remove(cert["backup_key_file"])
os.remove(cert['backup_cert_file'])
os.remove(cert['backup_key_file'])
letsencrypt/client/tests/apache/configurator_test.py
+6 -2
View File
@@ -31,8 +31,12 @@ class TwoVhost80Test(unittest.TestCase):
self.config_path = os.path.join(
self.temp_dir, "debian_apache_2_4/two_vhost_80/apache2/")
self.config = 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 = config_util.get_apache_configurator(
self.config_path, self.config_dir, self.work_dir,
self.ssl_options)
self.vh_truth = config_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
@@ -27,8 +27,12 @@ class DvsniPerformTest(unittest.TestCase):
self.config_path = os.path.join(
self.temp_dir, "debian_apache_2_4/two_vhost_80/apache2/")
config = 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 = config_util.get_apache_configurator(
self.config_path, self.config_dir, self.work_dir,
self.ssl_options)
self.sni = dvsni.ApacheDvsni(config)
letsencrypt/client/tests/reverter_test.py
+229
View File
@@ -0,0 +1,229 @@
"""Test letsencrypt.client.reverter."""
import logging
import os
import shutil
import tempfile
import unittest
class ReverterTest(unittest.TestCase):
"""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 = tempfile.mkdtemp("work")
backup = os.path.join(self.work_dir, "backup")
self.direc = {'backup': backup,
'temp': os.path.join(self.work_dir, "temp"),
'progress': os.path.join(backup, "progress")}
self.reverter = Reverter(self.direc)
self.dir1 = tempfile.mkdtemp("dir1")
self.dir2 = tempfile.mkdtemp("dir2")
self.config1 = os.path.join(self.dir1, "config.txt")
self.config2 = os.path.join(self.dir2, "config.txt")
with open(self.config1, 'w') as file_fd:
file_fd.write("directive-dir1")
with open(self.config2, 'w') as file_fd:
file_fd.write("directive-dir2")
self.sets = [set([self.config1]),
set([self.config2]),
set([self.config1, self.config2])]
def tearDown(self):
shutil.rmtree(self.work_dir)
shutil.rmtree(self.dir1)
shutil.rmtree(self.dir2)
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_checkpoint_conflict(self):
"""Make sure that checkpoint errors are thrown appropriately."""
from letsencrypt.client.errors import LetsEncryptReverterError
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(
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(
LetsEncryptReverterError,
self.reverter.add_to_checkpoint,
set([config3]), "invalid save")
def test_multiple_saves_and_rollback(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_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")
def test_rollback_improper_inputs(self):
from letsencrypt.client.errors import LetsEncryptReverterError
self.assertRaises(
LetsEncryptReverterError,
self.reverter.rollback_checkpoints, "-1")
self.assertRaises(
LetsEncryptReverterError,
self.reverter.rollback_checkpoints, -1000)
self.assertRaises(
LetsEncryptReverterError,
self.reverter.rollback_checkpoints, "one")
def test_rollback_finalize_checkpoint_valid_inputs(self):
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")
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))
def test_view_config_changes(self):
"""This is not strict as this is subject to change."""
self._setup_three_checkpoints()
# Just make sure it doesn't throw any errors.
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
def get_save_notes(dir):
"""Read save notes"""
return read_in(os.path.join(dir, 'CHANGES_SINCE'))
def get_filepaths(dir):
"""Get Filepaths"""
return read_in(os.path.join(dir, 'FILEPATHS'))
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, str):
"""Update a file with a new value."""
with open(filename, 'w') as file_fd:
file_fd.write(str)
if __name__ == '__main__':
unittest.main()
letsencrypt/scripts/main.py
+9 -1
View File
@@ -209,6 +209,10 @@ def read_file(filename):
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.
:param int checkpoints: Number of checkpoints to revert.
"""
@@ -248,7 +252,11 @@ def revoke(server):
def view_config_changes():
"""View checkpoints and associated configuration 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()