refactoring/small fixes for PR

This commit is contained in:
James Kasten
2015-01-25 04:24:05 -08:00
parent 8c6cfaded0
commit 208e7ec34b
11 changed files with 201 additions and 188 deletions
+1 -1
View File
@@ -1,5 +1,5 @@
:mod:`letsencrypt.client.reverter`
---------------------------------
----------------------------------
.. automodule:: letsencrypt.client.reverter
:members:
+119 -10
View File
@@ -20,7 +20,10 @@ 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 ACME servers via raw IPv6 addresses, and
# such addresses can be %SCARY in some contexts, so out of paranoia
@@ -211,6 +214,10 @@ 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.
@@ -232,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)
@@ -443,3 +440,115 @@ 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 at this time.")
installer = None
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()
+17 -17
View File
@@ -51,14 +51,14 @@ def create_sig(msg, key_str, nonce=None, nonce_len=CONFIG.NONCE_SIZE):
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),
}
@@ -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)
@@ -239,15 +239,15 @@ def get_cert_info(filename):
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),
}
+5 -5
View File
@@ -80,9 +80,9 @@ class NcursesDisplay(CommonDisplayMixin):
def display_certs(self, certs): # pylint: disable=missing-docstring
list_choices = []
for i, c in enumerate(certs):
if c['installed']:
if c['installed'] == CONFIG.CERT_DELETE_MSG:
for i, cert in enumerate(certs):
if cert["installed"]:
if cert["installed"] == CONFIG.CERT_DELETE_MSG:
status = "Deleted/Moved"
else:
status = "Installed"
@@ -90,8 +90,8 @@ class NcursesDisplay(CommonDisplayMixin):
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"),
str(cert["cn"].ljust(self.width-39)),
cert["not_before"].strftime("%m-%d-%y"),
status)))
code, tag = self.dialog.menu(
+3 -6
View File
@@ -234,7 +234,7 @@ class Reverter(object):
raise errors.LetsEncryptReverterError(
"Unable to remove directory: %s" % cp_dir)
def _check_tempfile_saves(self, save_files): # pylint: disable=no-self-use
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.
@@ -264,9 +264,8 @@ class Reverter(object):
"Attempting to overwrite challenge "
"file - %s" % filename)
# pylint: disable=no-self-use, anomalous-backslash-in-string
def register_file_creation(self, temporary, *files):
"""Register the creation of all files during letsencrypt execution.
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.
@@ -334,8 +333,7 @@ class Reverter(object):
"Incomplete or failed recovery for IN_PROGRESS checkpoint "
"- %s" % self.direc['progress'])
# pylint: disable=no-self-use
def _remove_contained_files(self, file_list):
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
@@ -373,7 +371,6 @@ class Reverter(object):
return True
# pylint: disable=no-self-use
def finalize_checkpoint(self, title):
"""Move IN_PROGRESS checkpoint to timestamped checkpoint.
+19 -19
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()
@@ -71,22 +71,22 @@ class Revoker(object):
for row in csvreader:
# 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["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'], cert.get('installed', "")),
"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:
@@ -135,12 +135,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"])
@@ -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")
@@ -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)
@@ -1,4 +1,4 @@
"""letsencrypt.scripts.main.py tests."""
"""letsencrypt.client.client.py tests."""
import unittest
import mock
@@ -12,11 +12,11 @@ class RollbackTest(unittest.TestCase):
self.m_input = mock.MagicMock()
zope.component.getUtility = self.m_input
def _call(self, checkpoints):
from letsencrypt.scripts.main import rollback
def _call(self, checkpoints): # pylint: disable=no-self-use
from letsencrypt.client.client import rollback
rollback(checkpoints)
@mock.patch("letsencrypt.scripts.main.determine_installer")
@mock.patch("letsencrypt.client.client.determine_installer")
def test_no_problems(self, mock_det):
mock_det.side_effect = self.m_install
@@ -26,7 +26,7 @@ class RollbackTest(unittest.TestCase):
self.assertEqual(self.m_install().restart.call_count, 1)
@mock.patch("letsencrypt.client.reverter.Reverter")
@mock.patch("letsencrypt.scripts.main.determine_installer")
@mock.patch("letsencrypt.client.client.determine_installer")
def test_misconfiguration_fixed(self, mock_det, mock_rev):
from letsencrypt.client.errors import LetsEncryptMisconfigurationError
mock_det.side_effect = [LetsEncryptMisconfigurationError,
@@ -42,9 +42,9 @@ class RollbackTest(unittest.TestCase):
# Only restart once
self.assertEqual(self.m_install.restart.call_count, 1)
@mock.patch("letsencrypt.scripts.main.logging.warning")
@mock.patch("letsencrypt.client.client.logging.warning")
@mock.patch("letsencrypt.client.reverter.Reverter")
@mock.patch("letsencrypt.scripts.main.determine_installer")
@mock.patch("letsencrypt.client.client.determine_installer")
def test_misconfiguration_remains(self, mock_det, mock_rev, mock_warn):
from letsencrypt.client.errors import LetsEncryptMisconfigurationError
mock_det.side_effect = LetsEncryptMisconfigurationError
@@ -63,7 +63,7 @@ class RollbackTest(unittest.TestCase):
self.assertEqual(mock_warn.call_count, 1)
@mock.patch("letsencrypt.client.reverter.Reverter")
@mock.patch("letsencrypt.scripts.main.determine_installer")
@mock.patch("letsencrypt.client.client.determine_installer")
def test_user_decides_to_manually_investigate(self, mock_det, mock_rev):
from letsencrypt.client.errors import LetsEncryptMisconfigurationError
mock_det.side_effect = LetsEncryptMisconfigurationError
@@ -76,7 +76,7 @@ class RollbackTest(unittest.TestCase):
self.assertEqual(self.m_install().rollback_checkpoints.call_count, 0)
self.assertEqual(mock_rev().rollback_checkpoints.call_count, 0)
@mock.patch("letsencrypt.scripts.main.determine_installer")
@mock.patch("letsencrypt.client.client.determine_installer")
def test_no_installer(self, mock_det):
mock_det.return_value = None
+10 -2
View File
@@ -8,8 +8,8 @@ import unittest
import mock
# pylint: disable=invalid-name,protected-access,too-many-instance-attributes
class ReverterCheckpointLocalTest(unittest.TestCase):
# pylint: disable=too-many-instance-attributes
"""Test the Reverter Class."""
def setUp(self):
from letsencrypt.client.reverter import Reverter
@@ -88,6 +88,7 @@ class ReverterCheckpointLocalTest(unittest.TestCase):
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")
@@ -138,15 +139,18 @@ class ReverterCheckpointLocalTest(unittest.TestCase):
from letsencrypt.client.errors import LetsEncryptReverterError
self.reverter.add_to_checkpoint(self.sets[0], "perm save")
# pylint: disable=protected-access
self.reverter._recover_checkpoint = mock.MagicMock(
side_effect=LetsEncryptReverterError)
self.assertRaises(LetsEncryptReverterError,
self.reverter.recovery_routine)
def test_recover_checkpoint_revert_temp_failures(self):
# pylint: disable=invalid-name
from letsencrypt.client.errors import LetsEncryptReverterError
mock_recover = mock.MagicMock(side_effect=LetsEncryptReverterError("e"))
# pylint: disable=protected-access
self.reverter._recover_checkpoint = mock_recover
self.reverter.add_to_temp_checkpoint(self.sets[0], "config1 save")
@@ -158,6 +162,7 @@ class ReverterCheckpointLocalTest(unittest.TestCase):
from letsencrypt.client.errors import LetsEncryptReverterError
mock_recover = mock.MagicMock(side_effect=LetsEncryptReverterError("e"))
# pylint: disable=protected-access
self.reverter._recover_checkpoint = mock_recover
self.reverter.add_to_checkpoint(self.sets[0], "config1 save")
@@ -237,8 +242,9 @@ class ReverterCheckpointLocalTest(unittest.TestCase):
self.assertEqual(read_in(self.config1), "directive-dir1")
self.assertEqual(read_in(self.config2), "directive-dir2")
# pylint: disable=invalid-name,protected-access,too-many-instance-attributes
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
@@ -269,6 +275,7 @@ class TestFullCheckpointsReverter(unittest.TestCase):
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
@@ -315,6 +322,7 @@ class TestFullCheckpointsReverter(unittest.TestCase):
@mock.patch("letsencrypt.client.reverter.os.rename")
def test_finalize_checkpoint_no_rename_directory(self, mock_rename):
# pylint: disable=invalid-name
from letsencrypt.client.errors import LetsEncryptReverterError
self.reverter.add_to_checkpoint(self.sets[0], "perm save")
+6 -115
View File
@@ -14,12 +14,9 @@ from letsencrypt.client import display
from letsencrypt.client import interfaces
from letsencrypt.client import errors
from letsencrypt.client import log
from letsencrypt.client import reverter
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."""
if not os.geteuid() == 0:
sys.exit(
@@ -78,15 +75,15 @@ def main(): # pylint: disable=too-many-statements
zope.component.provideUtility(displayer)
if args.view_config_changes:
view_config_changes()
client.view_config_changes()
sys.exit()
if args.revoke:
revoke(args.server)
client.revoke(args.server)
sys.exit()
if args.rollback > 0:
rollback(args.rollback)
client.rollback(args.rollback)
sys.exit()
if not args.eula:
@@ -95,7 +92,7 @@ def main(): # pylint: disable=too-many-statements
# Make sure we actually get an installer that is functioning properly
# before we begin to try to use it.
try:
installer = determine_installer()
installer = client.determine_installer()
except errors.LetsEncryptMisconfigurationError as err:
logging.fatal("Please fix your configuration before proceeding. "
"The Installer exited with the following message: "
@@ -106,7 +103,7 @@ def main(): # pylint: disable=too-many-statements
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
@@ -178,22 +175,6 @@ def get_all_names(installer):
return names
# 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 read_file(filename):
"""Returns the given file's contents with universal new line support.
@@ -212,95 +193,5 @@ def read_file(filename):
raise argparse.ArgumentTypeError(exc.strerror)
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:
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")
finally:
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 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 at this time.")
installer = None
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()
if __name__ == "__main__":
main()