mirror of
https://github.com/certbot/certbot.git
synced 2026-08-01 00:35:03 +02:00
Add a test to cover all assertions. Remove the CSR logic for now
This commit is contained in:
Adrien Ferrand
@@ -1,10 +1,14 @@
|
||||
"""Module executing integration tests against certbot core."""
|
||||
from __future__ import print_function
|
||||
import os
|
||||
import shutil
|
||||
from os.path import join
|
||||
|
||||
import pytest
|
||||
from certbot_integration_tests.certbot_tests import context as certbot_context
|
||||
from certbot_integration_tests.certbot_tests.assertions import (
|
||||
assert_hook_execution, assert_save_renew_hook
|
||||
assert_hook_execution, assert_save_renew_hook, assert_certs_count_for_lineage,
|
||||
assert_world_permissions, assert_equals_group_owner, assert_equals_permissions,
|
||||
)
|
||||
from certbot_integration_tests.utils import misc
|
||||
|
||||
@@ -19,21 +23,63 @@ def context(request):
|
||||
integration_test_context.cleanup()
|
||||
|
||||
|
||||
def test_manual_http_auth(context):
|
||||
"""Test the HTTP-01 challenge using manual plugin."""
|
||||
with misc.create_http_server(context.http_01_port) as webroot:
|
||||
manual_http_hooks = misc.manual_http_hooks(webroot)
|
||||
def test_manual_dns_auth(context):
|
||||
"""Test the DNS-01 challenge using manual plugin."""
|
||||
certname = context.domain('dns')
|
||||
context.certbot([
|
||||
'-a', 'manual', '-d', certname, '--preferred-challenges', 'dns',
|
||||
'run', '--cert-name', certname,
|
||||
'--manual-auth-hook', context.manual_dns_auth_hook,
|
||||
'--manual-cleanup-hook', context.manual_dns_cleanup_hook,
|
||||
'--pre-hook', 'echo wtf.pre >> "{0}"'.format(context.hook_probe),
|
||||
'--post-hook', 'echo wtf.post >> "{0}"'.format(context.hook_probe),
|
||||
'--renew-hook', 'echo renew >> "{0}"'.format(context.hook_probe)
|
||||
])
|
||||
|
||||
certname = context.domain()
|
||||
context.certbot([
|
||||
'certonly', '-a', 'manual', '-d', certname,
|
||||
'--cert-name', certname,
|
||||
'--manual-auth-hook', manual_http_hooks[0],
|
||||
'--manual-cleanup-hook', manual_http_hooks[1],
|
||||
'--pre-hook', 'echo wtf.pre >> "{0}"'.format(context.hook_probe),
|
||||
'--post-hook', 'echo wtf.post >> "{0}"'.format(context.hook_probe),
|
||||
'--deploy-hook', 'echo deploy >> "{0}"'.format(context.hook_probe)
|
||||
])
|
||||
|
||||
assert_hook_execution(context.hook_probe, 'deploy')
|
||||
with pytest.raises(AssertionError):
|
||||
assert_hook_execution(context.hook_probe, 'renew')
|
||||
assert_save_renew_hook(context.config_dir, certname)
|
||||
|
||||
|
||||
def test_renew(context):
|
||||
"""Test various certificate renew scenarios."""
|
||||
# First, we create a target certificate, with all hook dirs instantiated.
|
||||
# We should have a new certificate, with hooks executed.
|
||||
# Check also file permissions.
|
||||
certname = context.domain('renew')
|
||||
context.certbot([
|
||||
'certonly', '-d', certname, '--rsa-key-size', '4096',
|
||||
'--preferred-challenges', 'http-01'
|
||||
])
|
||||
|
||||
assert_certs_count_for_lineage(context.config_dir, certname, 1)
|
||||
assert_world_permissions(
|
||||
join(context.config_dir, 'archive/{0}/privkey1.pem'.format(certname)), 0)
|
||||
|
||||
# Second, we force renew, and ensure that renewal hooks files are executed.
|
||||
# Also check that file permissions are correct.
|
||||
misc.generate_test_file_hooks(context.config_dir, context.hook_probe)
|
||||
context.certbot(['renew'])
|
||||
|
||||
assert_certs_count_for_lineage(context.config_dir, certname, 2)
|
||||
assert_hook_execution(context.hook_probe, 'deploy')
|
||||
assert_world_permissions(
|
||||
join(context.config_dir, 'archive/{0}/privkey2.pem'.format(certname)), 0)
|
||||
assert_equals_group_owner(
|
||||
join(context.config_dir, 'archive/{0}/privkey1.pem'.format(certname)),
|
||||
join(context.config_dir, 'archive/{0}/privkey2.pem'.format(certname)))
|
||||
assert_equals_permissions(
|
||||
join(context.config_dir, 'archive/{0}/privkey1.pem'.format(certname)),
|
||||
join(context.config_dir, 'archive/{0}/privkey2.pem'.format(certname)), 0o074)
|
||||
|
||||
os.chmod(join(context.config_dir, 'archive/{0}/privkey2.pem'.format(certname)), 0o444)
|
||||
|
||||
# Third, we try to renew without force.
|
||||
# It is not time, so no renew should occur, and no hooks should be executed.
|
||||
open(context.hook_probe, 'w').close()
|
||||
misc.generate_test_file_hooks(context.config_dir, context.hook_probe)
|
||||
context.certbot_no_force_renew(['renew'])
|
||||
|
||||
assert_certs_count_for_lineage(context.config_dir, certname, 2)
|
||||
with pytest.raises(AssertionError):
|
||||
assert_hook_execution(context.hook_probe, 'deploy')
|
||||
|
||||
@@ -40,71 +40,6 @@ def check_until_timeout(url):
|
||||
raise ValueError('Error, url did not respond after 150 attempts: {0}'.format(url))
|
||||
|
||||
|
||||
def find_certbot_root_directory():
|
||||
"""
|
||||
Find the certbot root directory. To do so, this method will recursively move up in the directory
|
||||
hierarchy, until finding the git root file, that corresponds to the certbot root directory.
|
||||
:return str: the path to the certbot root directory
|
||||
"""
|
||||
script_path = os.path.realpath(__file__)
|
||||
current_dir = os.path.dirname(script_path)
|
||||
|
||||
while '.git' not in os.listdir(current_dir) and current_dir != os.path.dirname(current_dir):
|
||||
current_dir = os.path.dirname(current_dir)
|
||||
|
||||
dirs = os.listdir(current_dir)
|
||||
if '.git' not in dirs:
|
||||
raise ValueError('Error, could not find certbot sources root directory')
|
||||
|
||||
return current_dir
|
||||
|
||||
|
||||
def generate_csr(domains, key_path, csr_path, key_type='RSA'):
|
||||
"""
|
||||
Generate a CSR for the given domains, using the provided private key path.
|
||||
This method uses the script demo to generate CSR that is available in `examples/generate-csr.py`.
|
||||
:param str[] domains: the domain names to include in the CSR
|
||||
:param str key_path: path to the private key
|
||||
:param str csr_path: path to the CSR that will be generated
|
||||
:param str key_type: type of the key (RSA or ECDSA)
|
||||
"""
|
||||
certbot_root_directory = find_certbot_root_directory()
|
||||
script_path = os.path.join(certbot_root_directory, 'examples', 'generate-csr.py')
|
||||
|
||||
command = [
|
||||
sys.executable, script_path, '--key-path', key_path, '--csr-path', csr_path,
|
||||
'--key-type', key_type
|
||||
]
|
||||
command.extend(domains)
|
||||
subprocess.check_call(command)
|
||||
|
||||
|
||||
def load_sample_data_path(workspace):
|
||||
"""
|
||||
Load the certbot configuration example designed to make OCSP tests, and return its path
|
||||
:param str workspace: current test workspace directory path
|
||||
:return str: the path to the loaded sample data directory
|
||||
"""
|
||||
certbot_root_directory = find_certbot_root_directory()
|
||||
original = os.path.join(certbot_root_directory, 'tests', 'integration', 'sample-config')
|
||||
copied = os.path.join(workspace, 'sample-config')
|
||||
shutil.copytree(original, copied, symlinks=True)
|
||||
return copied
|
||||
|
||||
|
||||
def read_certificate(cert_path):
|
||||
"""
|
||||
Load the certificate from the provided path, and return a human readable version of it (TEXT mode).
|
||||
:param str cert_path: the path to the certificate
|
||||
:return: the TEXT version of the certificate, as it would be displayed by openssl binary
|
||||
"""
|
||||
with open(cert_path, 'r') as file:
|
||||
data = file.read()
|
||||
|
||||
cert = crypto.load_certificate(crypto.FILETYPE_PEM, data.encode('utf-8'))
|
||||
return crypto.dump_certificate(crypto.FILETYPE_TEXT, cert).decode('utf-8')
|
||||
|
||||
|
||||
class GracefulTCPServer(socketserver.TCPServer):
|
||||
"""
|
||||
This subclass of TCPServer allows to gracefully reuse an address that has
|
||||
|
||||
@@ -1,72 +0,0 @@
|
||||
#!/usr/bin/env python
|
||||
# -*- coding: utf-8 -*-
|
||||
# This script generates a simple SAN CSR to be used with Let's Encrypt CA.
|
||||
import os
|
||||
import argparse
|
||||
from OpenSSL import crypto
|
||||
|
||||
try:
|
||||
from cryptography.hazmat.backends import default_backend
|
||||
from cryptography.hazmat.primitives.asymmetric import rsa, ec
|
||||
from cryptography.hazmat.primitives.serialization import Encoding, PrivateFormat, NoEncryption
|
||||
cryptography = True
|
||||
except ImportError:
|
||||
cryptography = False
|
||||
|
||||
CWD = os.getcwd()
|
||||
|
||||
|
||||
def generate_parser():
|
||||
one_parser = argparse.ArgumentParser(description='Generate a simple SAN CSR')
|
||||
one_parser.add_argument('domain', nargs='+', help='A FQDN to add in the CSR.')
|
||||
one_parser.add_argument('--key-path', default=os.path.join(CWD, 'key.pem'),
|
||||
help='Path for the generated key '
|
||||
'(default is key.pem in current directory)')
|
||||
one_parser.add_argument('--csr-path', default=os.path.join(CWD, 'csr.der'),
|
||||
help='Path for the generated certificate '
|
||||
'(default is csr.der in current directory)')
|
||||
one_parser.add_argument('--key-type', default='RSA',
|
||||
choices=['RSA', 'ECDSA'],
|
||||
help='Key type to use, ECDSA is supported only '
|
||||
'if cryptography module is available')
|
||||
return one_parser
|
||||
|
||||
|
||||
def main(domains, key_path, csr_path, key_type):
|
||||
san = ','.join(['DNS:{0}'.format(item) for item in domains])
|
||||
|
||||
if key_type == 'RSA':
|
||||
key = crypto.PKey()
|
||||
key.generate_key(crypto.TYPE_RSA, 2048)
|
||||
elif key_type == 'ECDSA':
|
||||
if not cryptography:
|
||||
raise ValueError('Error, cryptography module is not installed,'
|
||||
'but is required to use ECDSA')
|
||||
|
||||
key = ec.generate_private_key(ec.SECP384R1(), default_backend())
|
||||
key = key.private_bytes(encoding=Encoding.PEM, format=PrivateFormat.TraditionalOpenSSL,
|
||||
encryption_algorithm=NoEncryption())
|
||||
key = crypto.load_privatekey(crypto.FILETYPE_PEM, key)
|
||||
else:
|
||||
raise ValueError('Invalid key type: {0}'.format(key_type))
|
||||
|
||||
with open(key_path, 'wb') as file:
|
||||
file.write(crypto.dump_privatekey(crypto.FILETYPE_PEM, key))
|
||||
|
||||
req = crypto.X509Req()
|
||||
san_constraint = crypto.X509Extension(b'subjectAltName', False, san.encode('utf-8'))
|
||||
req.add_extensions([san_constraint])
|
||||
|
||||
req.set_pubkey(key)
|
||||
req.sign(key, 'sha256')
|
||||
|
||||
with open(csr_path, 'wb') as file:
|
||||
file.write(crypto.dump_certificate_request(crypto.FILETYPE_ASN1, req))
|
||||
|
||||
print('You can now run: certbot auth --csr {0}'.format(csr_path))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
parser = generate_parser()
|
||||
namespace = parser.parse_args()
|
||||
main(namespace.domain, namespace.key_path, namespace.csr_path, namespace.key_type)
|
||||
Reference in New Issue
Block a user