diff --git a/acme/acme/challenges.py b/acme/acme/challenges.py index 89bc68966..723c51317 100644 --- a/acme/acme/challenges.py +++ b/acme/acme/challenges.py @@ -207,7 +207,7 @@ class DVSNI(DVChallenge): kwargs["name"] = self.nonce_domain # TODO: try different methods? # pylint: disable=protected-access - return crypto_util._probe_sni(**kwargs) + return crypto_util.probe_sni(**kwargs) @ChallengeResponse.register diff --git a/acme/acme/challenges_test.py b/acme/acme/challenges_test.py index e4ec37362..68492fbea 100644 --- a/acme/acme/challenges_test.py +++ b/acme/acme/challenges_test.py @@ -179,7 +179,7 @@ class DVSNITest(unittest.TestCase): jose.DeserializationError, DVSNI.from_json, self.jmsg) @mock.patch('acme.challenges.socket.gethostbyname') - @mock.patch('acme.challenges.crypto_util._probe_sni') + @mock.patch('acme.challenges.crypto_util.probe_sni') def test_probe_cert(self, mock_probe_sni, mock_gethostbyname): mock_gethostbyname.return_value = '127.0.0.1' self.msg.probe_cert('foo.com') diff --git a/acme/acme/crypto_util.py b/acme/acme/crypto_util.py index cb796cb88..624d371e1 100644 --- a/acme/acme/crypto_util.py +++ b/acme/acme/crypto_util.py @@ -69,8 +69,8 @@ def _serve_sni(certs, sock, reuseaddr=True, method=_DEFAULT_DVSNI_SSL_METHOD, raise errors.Error(error) -def _probe_sni(name, host, port=443, timeout=300, - method=_DEFAULT_DVSNI_SSL_METHOD, source_address=('0', 0)): +def probe_sni(name, host, port=443, timeout=300, + method=_DEFAULT_DVSNI_SSL_METHOD, source_address=('0', 0)): """Probe SNI server for SSL certificate. :param bytes name: Byte string to send as the server name in the diff --git a/acme/acme/crypto_util_test.py b/acme/acme/crypto_util_test.py index 10d62fbf5..49aacfa1b 100644 --- a/acme/acme/crypto_util_test.py +++ b/acme/acme/crypto_util_test.py @@ -13,7 +13,7 @@ from acme import test_util class ServeProbeSNITest(unittest.TestCase): - """Tests for acme.crypto_util._serve_sni/_probe_sni.""" + """Tests for acme.crypto_util._serve_sni/probe_sni.""" def setUp(self): self.cert = test_util.load_cert('cert.pem') @@ -45,8 +45,8 @@ class ServeProbeSNITest(unittest.TestCase): self.server.join() def _probe(self, name): - from acme.crypto_util import _probe_sni - return jose.ComparableX509(_probe_sni( + from acme.crypto_util import probe_sni + return jose.ComparableX509(probe_sni( name, host='127.0.0.1', port=self.port)) def test_probe_ok(self): diff --git a/letsencrypt/interfaces.py b/letsencrypt/interfaces.py index 54f0dc92b..52f23ab88 100644 --- a/letsencrypt/interfaces.py +++ b/letsencrypt/interfaces.py @@ -384,17 +384,52 @@ class IDisplay(zope.interface.Interface): class IValidator(zope.interface.Interface): """Configuration validator.""" - def redirect(hostname, port=80, headers=None): - """Verify redirect to HTTPS.""" + def certificate(cert, name, alt_host=None, port=443): + """Verifies the certificate presented at name is cert - def https(hostname, port=443, headers=None): - """Verify HTTPS is enabled for domain.""" + :param OpenSSL.crypto.X509 cert: Expected certificate + :param str name: Server's domain name + :param bytes alt_host: Host to connect to instead of the IP + address of host + :param int port: Port to connect to - def hsts(hostname): - """Verify HSTS header is enabled.""" + :returns: True if the certificate was verified successfully + :rtype: bool - def ocsp_stapling(hostname): - """Verify ocsp stapling for domain.""" + """ + + def redirect(name, port=80, headers=None): + """Verify redirect to HTTPS + + :param str name: Server's domain name + :param int port: Port to connect to + :param dict headers: HTTP headers to include in request + + :returns: True if redirect is successfully enabled + :rtype: bool + + """ + + + def hsts(name): + """Verify HSTS header is enabled + + :param str name: Server's domain name + + :returns: True if HSTS header is successfully enabled + :rtype: bool + + """ + + def ocsp_stapling(name): + """Verify ocsp stapling for domain + + :param str name: Server's domain name + + :returns: True if ocsp stapling is successfully enabled + :rtype: bool + + """ class IReporter(zope.interface.Interface): diff --git a/letsencrypt/tests/validator_test.py b/letsencrypt/tests/validator_test.py index c9cb19ec2..c02a7d865 100644 --- a/letsencrypt/tests/validator_test.py +++ b/letsencrypt/tests/validator_test.py @@ -3,8 +3,9 @@ import requests import unittest import mock +import OpenSSL -from letsencrypt import errors +from acme import errors as acme_errors from letsencrypt import validator @@ -12,12 +13,41 @@ class ValidatorTest(unittest.TestCase): def setUp(self): self.validator = validator.Validator() + @mock.patch("letsencrypt.validator.crypto_util.probe_sni") + def test_certificate_success(self, mock_probe_sni): + cert = OpenSSL.crypto.X509() + mock_probe_sni.return_value = cert + self.assertTrue(self.validator.certificate( + cert, "test.com", "127.0.0.1")) + + @mock.patch("letsencrypt.validator.crypto_util.probe_sni") + def test_certificate_error(self, mock_probe_sni): + cert = OpenSSL.crypto.X509() + mock_probe_sni.side_effect = [acme_errors.Error] + self.assertFalse(self.validator.certificate( + cert, "test.com", "127.0.0.1")) + + @mock.patch("letsencrypt.validator.crypto_util.probe_sni") + def test_certificate_failure(self, mock_probe_sni): + cert = OpenSSL.crypto.X509() + cert.set_serial_number(1337) + mock_probe_sni.return_value = OpenSSL.crypto.X509() + self.assertFalse(self.validator.certificate( + cert, "test.com", "127.0.0.1")) + @mock.patch("letsencrypt.validator.requests.get") def test_succesful_redirect(self, mock_get_request): mock_get_request.return_value = create_response( 301, {"location" : "https://test.com"}) self.assertTrue(self.validator.redirect("test.com")) + @mock.patch("letsencrypt.validator.requests.get") + def test_redirect_with_headers(self, mock_get_request): + mock_get_request.return_value = create_response( + 301, {"location" : "https://test.com"}) + self.assertTrue(self.validator.redirect( + "test.com", headers={"Host" : "test.com"})) + @mock.patch("letsencrypt.validator.requests.get") def test_redirect_missing_location(self, mock_get_request): mock_get_request.return_value = create_response(301) @@ -33,19 +63,7 @@ class ValidatorTest(unittest.TestCase): def test_redirect_wrong_redirect_code(self, mock_get_request): mock_get_request.return_value = create_response( 303, {"location" : "https://test.com"}) - self.assertRaises( - errors.ValidationError, self.validator.redirect, "test.com") - - @mock.patch("letsencrypt.validator.requests.get") - def test_https_fail(self, mock_get_request): - mock_get_request.side_effect = [requests.exceptions.ConnectionError] - self.assertRaises( - requests.exceptions.ConnectionError, self.validator.https, "test.com") - - def test_https_success(self): - with mock.patch("letsencrypt.validator.requests.get"): - self.assertTrue(self.validator.https( - "test.com", headers={"Host" : "test.com"})) + self.assertFalse(self.validator.redirect("test.com")) @mock.patch("letsencrypt.validator.requests.get") def test_hsts_empty(self, mock_get_request): @@ -57,22 +75,19 @@ class ValidatorTest(unittest.TestCase): def test_hsts_malformed(self, mock_get_request): mock_get_request.return_value = create_response( headers={"strict-transport-security": "sdfal"}) - self.assertRaises( - errors.ValidationError, self.validator.hsts, "test.com") + self.assertFalse(self.validator.hsts("test.com")) @mock.patch("letsencrypt.validator.requests.get") def test_hsts_bad_max_age(self, mock_get_request): mock_get_request.return_value = create_response( headers={"strict-transport-security": "max-age=not-an-int"}) - self.assertRaises( - errors.ValidationError, self.validator.hsts, "test.com") + self.assertFalse(self.validator.hsts("test.com")) @mock.patch("letsencrypt.validator.requests.get") def test_hsts_expire(self, mock_get_request): mock_get_request.return_value = create_response( headers={"strict-transport-security": "max-age=3600"}) - self.assertRaises( - errors.ValidationError, self.validator.hsts, "test.com") + self.assertFalse(self.validator.hsts("test.com")) @mock.patch("letsencrypt.validator.requests.get") def test_hsts(self, mock_get_request): @@ -103,4 +118,4 @@ def create_response(status_code=200, headers=None): if __name__ == '__main__': - unittest.main() + unittest.main() # pragma: no cover diff --git a/letsencrypt/validator.py b/letsencrypt/validator.py index 6ef88d566..e5386f290 100644 --- a/letsencrypt/validator.py +++ b/letsencrypt/validator.py @@ -1,19 +1,40 @@ """Validators to determine the current webserver configuration""" +import logging +import socket import requests import zope.interface -from letsencrypt import errors +from acme import crypto_util +from acme import errors as acme_errors from letsencrypt import interfaces +logger = logging.getLogger(__name__) + + class Validator(object): # pylint: disable=no-self-use """Collection of functions to test a live webserver's configuration""" zope.interface.implements(interfaces.IValidator) - def redirect(self, hostname, port=80, headers=None): + def certificate(self, cert, name, alt_host=None, port=443): + """Verifies the certificate presented at name is cert""" + host = alt_host if alt_host else socket.gethostbyname(name) + try: + presented_cert = crypto_util.probe_sni(name, host, port) + except acme_errors.Error as error: + logger.exception(error) + return False + + return presented_cert.digest("sha256") == cert.digest("sha256") + + def redirect(self, name, port=80, headers=None): """Test whether webserver redirects to secure connection.""" - response = _get("http", hostname, port, headers) + url = "http://{0}:{1}".format(name, port) + if headers: + response = requests.get(url, headers=headers, allow_redirects=False) + else: + response = requests.get(url, allow_redirects=False) if response.status_code not in (301, 303): return False @@ -23,19 +44,14 @@ class Validator(object): return False if response.status_code != 301: - error_msg = "Server did not redirect with permanent code." - raise errors.ValidationError(error_msg) + logger.error("Server did not redirect with permanent code") + return False return True - def https(self, hostname, port=443, headers=None): - """Test whether webserver supports HTTPS""" - _get("https", hostname, port, headers) - return True - - def hsts(self, hostname): + def hsts(self, name): """Test for HTTP Strict Transport Security header""" - headers = requests.get("https://" + hostname).headers + headers = requests.get("https://" + name).headers hsts_header = headers.get("strict-transport-security") if not hsts_header: @@ -46,32 +62,23 @@ class Validator(object): max_age = [d for d in directives if d[0] == "max-age"] if not max_age: - error_msg = "Server responded with invalid HSTS header field." - raise errors.ValidationError(error_msg) + logger.error("Server responded with invalid HSTS header field") + return False try: _, max_age_value = max_age[0] max_age_value = int(max_age_value) except ValueError: - error_msg = "Server responded with invalid HSTS header field." - raise errors.ValidationError(error_msg) + logger.error("Server responded with invalid HSTS header field") + return False # Test whether HSTS does not expire for at least two weeks. if max_age_value <= (2 * 7 * 24 * 3600): - error_msg = "HSTS should not expire in less than two weeks." - raise errors.ValidationError(error_msg) + logger.error("HSTS should not expire in less than two weeks") + return False return True def ocsp_stapling(self, name): """Verify ocsp stapling for domain.""" raise NotImplementedError() - - -def _get(scheme, hostname, port, headers, **kwargs): - """Makes a GET request for specified resource""" - url = "{0}://{1}:{2}".format(scheme, hostname, port) - if headers: - return requests.get(url, headers=headers, **kwargs) - else: - return requests.get(url, **kwargs) diff --git a/tests/compatibility/configurators/apache/common.py b/tests/compatibility/configurators/apache/common.py index 99d78904a..a3b7ddd95 100644 --- a/tests/compatibility/configurators/apache/common.py +++ b/tests/compatibility/configurators/apache/common.py @@ -113,13 +113,6 @@ class Proxy(configurators_common.Proxy): super(Proxy, self).cleanup_from_tests() self._patch.stop() - def get_testable_domain_names(self): - """Returns the set of domain names that can be tested against""" - if self._test_names: - return self._test_names - else: - raise errors.Error("No configuration file loaded") - def get_all_names_answer(self): """Returns the set of domain names that the plugin should find""" if self._all_names: @@ -127,6 +120,12 @@ class Proxy(configurators_common.Proxy): else: raise errors.Error("No configuration file loaded") + def get_testable_domain_names(self): + """Returns the set of domain names that can be tested against""" + if self._test_names: + return self._test_names + else: + return {"example.com"} def deploy_cert(self, domain, cert_path, key_path, chain_path=None): """Installs cert""" cert_path, key_path, chain_path = self.copy_certs_and_keys( diff --git a/tests/compatibility/configurators/common.py b/tests/compatibility/configurators/common.py index 549b2f272..e517d71e7 100644 --- a/tests/compatibility/configurators/common.py +++ b/tests/compatibility/configurators/common.py @@ -7,6 +7,7 @@ import threading import docker +from letsencrypt import constants from tests.compatibility import errors from tests.compatibility import util @@ -61,6 +62,9 @@ class Proxy(object): def load_config(self): """Returns the next config directory to be tested""" + shutil.rmtree(self.le_config.work_dir, ignore_errors=True) + backup = os.path.join(self.le_config.work_dir, constants.BACKUP_DIR) + os.makedirs(backup) return self._configs.pop() def start_docker(self, image_name, command): @@ -136,7 +140,8 @@ class Proxy(object): def copy_certs_and_keys(self, cert_path, key_path, chain_path=None): """Copies certs and keys into the temporary directory""" cert_and_key_dir = os.path.join(self._temp_dir, "certs_and_keys") - os.mkdir(cert_and_key_dir) + if not os.path.isdir(cert_and_key_dir): + os.mkdir(cert_and_key_dir) cert = os.path.join(cert_and_key_dir, "cert") shutil.copy(cert_path, cert) diff --git a/tests/compatibility/test_driver.py b/tests/compatibility/test_driver.py index da49868b1..65ad36f18 100644 --- a/tests/compatibility/test_driver.py +++ b/tests/compatibility/test_driver.py @@ -1,6 +1,7 @@ """Tests Let's Encrypt plugins against different server configurations.""" import argparse import filecmp +import functools import logging import os import shutil @@ -12,6 +13,8 @@ from acme import challenges from acme import crypto_util from acme import messages from letsencrypt import achallenges +from letsencrypt import errors as le_errors +from letsencrypt import validator from letsencrypt.tests import acme_util from tests.compatibility import errors from tests.compatibility import util @@ -34,36 +37,41 @@ logger = logging.getLogger(__name__) def test_authenticator(plugin, config, temp_dir): """Tests plugin as an authenticator""" - backup = os.path.join(temp_dir, "backup") - shutil.copytree(config, backup, symlinks=True) + backup = _create_backup(config, temp_dir) achalls = _create_achalls(plugin) - if achalls: - try: - responses = plugin.perform(achalls) - for i in xrange(len(responses)): - if not responses[i]: - raise errors.Error( - "Plugin returned 'None' or 'False' response to " - "challenge") - elif isinstance(responses[i], challenges.DVSNIResponse): - if responses[i].simple_verify(achalls[i], - achalls[i].domain, - util.JWK.key.public_key(), - host="127.0.0.1", - port=plugin.https_port): - logger.info( - "Verification of DVSNI response for %s succeeded", - achalls[i].domain) - else: - raise errors.Error( - "Verification of DVSNI response for {0} " - "failed".format(achalls[i].domain)) - finally: - plugin.cleanup(achalls) + if not achalls: + return + + try: + responses = plugin.perform(achalls) + for i in xrange(len(responses)): + if not responses[i]: + logger.error( + "Plugin returned `None` or `False` response to challenge " + "for config `%s`", config) + elif isinstance(responses[i], challenges.DVSNIResponse): + if responses[i].simple_verify(achalls[i], + achalls[i].domain, + util.JWK.key.public_key(), + host="127.0.0.1", + port=plugin.https_port): + logger.info( + "Verification of DVSNI response for %s succeeded", + achalls[i].domain) + else: + logger.error( + "Verification of DVSNI response for %s in config '%s' " + "failed ", achalls[i].domain, config) + except le_errors.Error as error: + logger.info( + "Plugin raised %s during authentication with config '%s'", + error, config) + finally: + plugin.cleanup(achalls) if _dirs_are_unequal(config, backup): - raise errors.Error("Challenge cleanup failed") + logger.error("Challenge cleanup failed for config %s", config) else: logger.info("Challenge cleanup succeeded") @@ -88,17 +96,24 @@ def _create_achalls(plugin): return achalls -def test_installer(plugin, config, temp_dir): +def test_installer(args, plugin, config, temp_dir): """Tests plugin as an installer""" - backup = os.path.join(temp_dir, "backup") - shutil.copytree(config, backup, symlinks=True) + backup = _create_backup(config, temp_dir) - if plugin.get_all_names() != plugin.get_all_names_answer(): - raise errors.Error("get_all_names test failed") + if plugin.get_all_names().issubset(plugin.get_all_names_answer()): + logger.info("get_all_names test succeeded") else: - logging.info("get_all_names test succeeded") + logger.error("get_all_names test failed for config `%s`", config) domains = list(plugin.get_testable_domain_names()) + if test_deploy_cert(plugin, temp_dir, domains) and args.enhance: + test_enhancements(plugin, domains) + + test_rollback(plugin, config, backup) + + +def test_deploy_cert(plugin, temp_dir, domains): + """Tests deploy_cert returning True if the tests are successful""" cert = crypto_util.gen_ss_cert(util.KEY, domains) cert_path = os.path.join(temp_dir, "cert.pem") with open(cert_path, "w") as f: @@ -107,9 +122,65 @@ def test_installer(plugin, config, temp_dir): for domain in domains: plugin.deploy_cert(domain, cert_path, util.KEY_PATH) - plugin.save() + plugin.save("deployed") plugin.restart() + verify_cert = validator.Validator().certificate + success = True + for domain in domains: + if not verify_cert(cert, domain, "127.0.0.1", plugin.https_port): + logger.error("Could not verify certificate for domain %s", domain) + success = False + + if success: + logger.info("HTTPS validation succeeded") + + return success + + +def test_enhancements(plugin, domains): + """Tests enhancements supported by the plugin""" + supported = plugin.supported_enhancements() + + if "redirect" not in supported: + return + + for domain in domains: + plugin.enhance(domain, "redirect") + + plugin.save("enhanced") + plugin.restart() + + verify_redirect = functools.partial( + validator.Validator().redirect, "localhost", plugin.http_port) + success = True + for domain in domains: + if not verify_redirect(headers={"Host" : domain}): + logger.error("Improper redirect for domain %s", domain) + success = False + + if success: + logger.info("Enhancments test succeeded") + + +def test_rollback(plugin, config, backup): + """Tests the rollback checkpoints function""" + plugin.rollback_checkpoints(2) + + if _dirs_are_unequal(config, backup): + logger.error("Rollback failed for config `%s`", config) + else: + logger.info("Rollback succeeded") + + +def _create_backup(config, temp_dir): + """Creates a backup of config in temp_dir""" + backup = os.path.join(temp_dir, "backup") + shutil.rmtree(backup, ignore_errors=True) + shutil.copytree(config, backup, symlinks=True) + + return backup + def _dirs_are_unequal(dir1, dir2): """Returns True if dir1 and dir2 are equal""" @@ -151,7 +222,7 @@ def get_args(): if args.enhance: args.install = True elif not (args.auth or args.install): - args.auth = args.install = args.redirect = True + args.auth = args.install = args.enhance = True return args @@ -161,7 +232,7 @@ def setup_logging(args): handler = logging.StreamHandler() root_logger = logging.getLogger() - root_logger.setLevel(logging.INFO - args.verbose_count * 10) + root_logger.setLevel(logging.WARNING - args.verbose_count * 10) root_logger.addHandler(handler) @@ -183,10 +254,10 @@ def main(): logger.info("Loaded configuration: %s", config) if args.auth: test_authenticator(plugin, config, temp_dir) - #if args.install: - #test_installer(plugin, temp_dir) + if args.install: + test_installer(args, plugin, config, temp_dir) except errors.Error as error: - logger.warning("Test failed: %s", error) + logger.error("Tests on config `%s` raised: %s", config, error) finally: plugin.cleanup_from_tests()