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

fixed mainline merge conflict

Browse Source
This commit is contained in:
Noah Swartz
2016-01-08 02:36:35 -08:00
parent 6fedd22dc8 89296eca9d
commit e7ae025a72
95 changed files with 1711 additions and 286 deletions
Download Patch File Download Diff File Expand all files Collapse all files
letsencrypt/auth_handler.py
+3 -8
View File
@@ -540,16 +540,11 @@ def _generate_failed_chall_msg(failed_achalls):
"""
typ = failed_achalls[0].error.typ
msg = [
"The following '{0}' errors were reported by the server:".format(typ)]
msg = ["The following errors were reported by the server:"]
problems = dict()
for achall in failed_achalls:
problems.setdefault(achall.error.description, set()).add(achall.domain)
for problem in problems:
msg.append("\n\nDomains: ")
msg.append(", ".join(sorted(problems[problem])))
msg.append("\nError: {0}".format(problem))
msg.append("\n\nDomain: %s\nType: %s\nDetail: %s" % (
achall.domain, achall.error.typ, achall.error.detail))
if typ in _ERROR_HELP:
msg.append("\n\n")
letsencrypt/cli.py
+1 -1
View File
@@ -1189,7 +1189,7 @@ def _plugins_parsing(helpful, plugins):
# These would normally be a flag within the webroot plugin, but because
# they are parsed in conjunction with --domains, they live here for
# legibiility. helpful.add_plugin_ags must be called first to add the
# legibility. helpful.add_plugin_ags must be called first to add the
# "webroot" topic
helpful.add("webroot", "-w", "--webroot-path", action=WebrootPathProcessor,
help="public_html / webroot path. This can be specified multiple times to "
letsencrypt/plugins/webroot_test.py
+9 -1
View File
@@ -66,8 +66,16 @@ class AuthenticatorTest(unittest.TestCase):
def test_prepare_reraises_other_errors(self):
self.auth.full_path = os.path.join(self.path, "null")
permission_canary = os.path.join(self.path, "rnd")
with open(permission_canary, "w") as f:
f.write("thingimy")
os.chmod(self.path, 0o000)
self.assertRaises(errors.PluginError, self.auth.prepare)
try:
open(permission_canary, "r")
print "Warning, running tests as root skips permissions tests..."
except IOError:
# ok, permissions work, test away...
self.assertRaises(errors.PluginError, self.auth.prepare)
os.chmod(self.path, 0o700)
@mock.patch("letsencrypt.plugins.webroot.os.chown")
letsencrypt/renewer.py
+4 -1
View File
@@ -116,6 +116,7 @@ def renew(cert, old_version):
def _cli_log_handler(args, level, fmt): # pylint: disable=unused-argument
handler = colored_logging.StreamHandler()
handler.setFormatter(logging.Formatter(fmt))
handler.setLevel(level)
return handler
@@ -181,7 +182,9 @@ def main(cli_args=sys.argv[1:]):
# RenewableCert object for this cert at all, which could
# dramatically improve performance for large deployments
# where autorenewal is widely turned off.
cert = storage.RenewableCert(renewal_file, cli_config)
cert = storage.RenewableCert(
os.path.join(cli_config.renewal_configs_dir, renewal_file),
cli_config)
except errors.CertStorageError:
# This indicates an invalid renewal configuration file, such
# as one missing a required parameter (in the future, perhaps
letsencrypt/storage.py
+4 -1
View File
@@ -260,7 +260,7 @@ class RenewableCert(object): # pylint: disable=too-many-instance-attributes
:returns: The path to the current version of the specified
member.
:rtype: str
:rtype: str or None
"""
if kind not in ALL_FOUR:
@@ -450,12 +450,15 @@ class RenewableCert(object): # pylint: disable=too-many-instance-attributes
:param int version: the desired version number
:returns: the subject names
:rtype: `list` of `str`
:raises .CertStorageError: if could not find cert file.
"""
if version is None:
target = self.current_target("cert")
else:
target = self.version("cert", version)
if target is None:
raise errors.CertStorageError("could not find cert file")
with open(target) as f:
return crypto_util.get_sans_from_cert(f.read())
letsencrypt/tests/auth_handler_test.py
+1 -1
View File
@@ -467,7 +467,7 @@ class ReportFailedChallsTest(unittest.TestCase):
auth_handler._report_failed_challs([self.http01, self.tls_sni_same])
call_list = mock_zope().add_message.call_args_list
self.assertTrue(len(call_list) == 1)
self.assertTrue("Domains: example.com\n" in call_list[0][0][0])
self.assertTrue("Domain: example.com\nType: tls\nDetail: detail" in call_list[0][0][0])
@mock.patch("letsencrypt.auth_handler.zope.component.getUtility")
def test_different_errors_and_domains(self, mock_zope):
letsencrypt/tests/renewer_test.py
+2
View File
@@ -764,6 +764,8 @@ class RenewableCertTests(BaseRenewableCertTest):
def test_bad_config_file(self):
from letsencrypt import renewer
os.unlink(os.path.join(self.cli_config.renewal_configs_dir,
"example.org.conf"))
with open(os.path.join(self.cli_config.renewal_configs_dir,
"bad.conf"), "w") as f:
f.write("incomplete = configfile\n")
letsencrypt/tests/validator_test.py
-122
View File
@@ -1,122 +0,0 @@
"""Tests for letsencrypt.validator."""
import requests
import unittest
import mock
import OpenSSL
from acme import errors as acme_errors
from letsencrypt import validator
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)
self.assertFalse(self.validator.redirect("test.com"))
@mock.patch("letsencrypt.validator.requests.get")
def test_redirect_wrong_status_code(self, mock_get_request):
mock_get_request.return_value = create_response(
201, {"location": "https://test.com"})
self.assertFalse(self.validator.redirect("test.com"))
@mock.patch("letsencrypt.validator.requests.get")
def test_redirect_wrong_redirect_code(self, mock_get_request):
mock_get_request.return_value = create_response(
303, {"location": "https://test.com"})
self.assertFalse(self.validator.redirect("test.com"))
@mock.patch("letsencrypt.validator.requests.get")
def test_hsts_empty(self, mock_get_request):
mock_get_request.return_value = create_response(
headers={"strict-transport-security": ""})
self.assertFalse(self.validator.hsts("test.com"))
@mock.patch("letsencrypt.validator.requests.get")
def test_hsts_malformed(self, mock_get_request):
mock_get_request.return_value = create_response(
headers={"strict-transport-security": "sdfal"})
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.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.assertFalse(self.validator.hsts("test.com"))
@mock.patch("letsencrypt.validator.requests.get")
def test_hsts(self, mock_get_request):
mock_get_request.return_value = create_response(
headers={"strict-transport-security": "max-age=31536000"})
self.assertTrue(self.validator.hsts("test.com"))
@mock.patch("letsencrypt.validator.requests.get")
def test_hsts_include_subdomains(self, mock_get_request):
mock_get_request.return_value = create_response(
headers={"strict-transport-security":
"max-age=31536000;includeSubDomains"})
self.assertTrue(self.validator.hsts("test.com"))
def test_ocsp_stapling(self):
self.assertRaises(
NotImplementedError, self.validator.ocsp_stapling, "test.com")
def create_response(status_code=200, headers=None):
"""Creates a requests.Response object for testing"""
response = requests.Response()
response.status_code = status_code
if headers:
response.headers = headers
return response
if __name__ == '__main__':
unittest.main() # pragma: no cover
letsencrypt/validator.py
-84
View File
@@ -1,84 +0,0 @@
"""Validators to determine the current webserver configuration"""
import logging
import socket
import requests
import zope.interface
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 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."""
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
redirect_location = response.headers.get("location", "")
if not redirect_location.startswith("https://"):
return False
if response.status_code != 301:
logger.error("Server did not redirect with permanent code")
return False
return True
def hsts(self, name):
"""Test for HTTP Strict Transport Security header"""
headers = requests.get("https://" + name).headers
hsts_header = headers.get("strict-transport-security")
if not hsts_header:
return False
# Split directives following RFC6797, section 6.1
directives = [d.split("=") for d in hsts_header.split(";")]
max_age = [d for d in directives if d[0] == "max-age"]
if not max_age:
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:
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):
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()