mirror of
https://github.com/certbot/certbot.git
synced 2026-07-28 08:45:22 +02:00
Move validator to compatibility-test — Refs #1997
This commit is contained in:
Rémy HUBSCHER
@@ -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
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user