mirror of
https://github.com/certbot/certbot.git
synced 2026-07-28 08:45:22 +02:00
Second part: integration tests for certbot core
This commit is contained in:
Adrien Ferrand
@@ -0,0 +1,45 @@
|
||||
import os
|
||||
import grp
|
||||
|
||||
|
||||
__all__ = ['assert_hook_execution', 'assert_save_renew_hook', 'assert_certs_count_for_lineage',
|
||||
'assert_equals_permissions', 'assert_equals_group_owner', 'assert_world_permissions']
|
||||
|
||||
|
||||
def assert_hook_execution(probe_path, probe_content):
|
||||
with open(probe_path, 'r') as file:
|
||||
lines = file.readlines()
|
||||
|
||||
# Comparing pattern to each line avoids to match "pre" for a line with "pre-override"
|
||||
assert [line for line in lines if line == '{0}{1}'.format(probe_content, os.linesep)]
|
||||
|
||||
|
||||
def assert_save_renew_hook(config_dir, lineage):
|
||||
assert os.path.isfile(os.path.join(config_dir, 'renewal/{0}.conf'.format(lineage)))
|
||||
|
||||
|
||||
def assert_certs_count_for_lineage(config_dir, lineage, count):
|
||||
archive_dir = os.path.join(config_dir, 'archive')
|
||||
lineage_dir = os.path.join(archive_dir, lineage)
|
||||
certs = [file for file in os.listdir(lineage_dir) if file.startswith('cert')]
|
||||
assert len(certs) == count
|
||||
|
||||
|
||||
def assert_equals_permissions(file1, file2, mask):
|
||||
mode_file1 = os.stat(file1).st_mode & mask
|
||||
mode_file2 = os.stat(file2).st_mode & mask
|
||||
|
||||
assert mode_file1 == mode_file2
|
||||
|
||||
|
||||
def assert_equals_group_owner(file1, file2):
|
||||
group_owner_file1 = grp.getgrgid(os.stat(file1).st_gid)[0]
|
||||
group_owner_file2 = grp.getgrgid(os.stat(file2).st_gid)[0]
|
||||
|
||||
assert group_owner_file1 == group_owner_file2
|
||||
|
||||
|
||||
def assert_world_permissions(file, mode):
|
||||
mode_file_all = os.stat(file).st_mode & 0o007
|
||||
|
||||
assert mode_file_all == mode
|
||||
@@ -0,0 +1,5 @@
|
||||
import pytest
|
||||
|
||||
# Custom assertions defined in following package needs to be registered to be properly
|
||||
# displayed in a pytest report when they are failing.
|
||||
pytest.register_assert_rewrite('certbot_integration_tests.certbot_tests.assertions')
|
||||
@@ -1,16 +1,105 @@
|
||||
import os
|
||||
import tempfile
|
||||
import subprocess
|
||||
import shutil
|
||||
import sys
|
||||
from distutils.version import LooseVersion
|
||||
|
||||
from certbot_integration_tests.utils import misc
|
||||
|
||||
|
||||
class IntegrationTestsContext(object):
|
||||
"""General fixture describing a certbot integration tests context"""
|
||||
def __init__(self, request):
|
||||
self.request = request
|
||||
|
||||
self.worker_id = request.config.slaveinput['slaveid'] if hasattr(request.config, 'slaveinput') else 'master'
|
||||
if hasattr(request.config, 'slaveinput'): # Worker node
|
||||
self.worker_id = request.config.slaveinput['slaveid']
|
||||
self.acme_xdist = request.config.slaveinput['acme_xdist']
|
||||
else: # Primary node
|
||||
self.worker_id = 'primary'
|
||||
self.acme_xdist = request.config.acme_xdist
|
||||
|
||||
self.acme_server =self.acme_xdist['acme_server']
|
||||
self.directory_url = self.acme_xdist['directory_url']
|
||||
self.tls_alpn_01_port = self.acme_xdist['https_port'][self.worker_id]
|
||||
self.http_01_port = self.acme_xdist['http_port'][self.worker_id]
|
||||
self.challtestsrv_mgt_port = self.acme_xdist['challtestsrv_port']
|
||||
|
||||
self.certbot_version = misc.get_certbot_version()
|
||||
|
||||
self.workspace = tempfile.mkdtemp()
|
||||
self.config_dir = os.path.join(self.workspace, 'conf')
|
||||
self.hook_probe = tempfile.mkstemp(dir=self.workspace)[1]
|
||||
|
||||
self.manual_dns_auth_hook = (
|
||||
'{0} -c "import os; import requests; import json; '
|
||||
"assert not os.environ.get('CERTBOT_DOMAIN').startswith('fail'); "
|
||||
"data = {{'host':'_acme-challenge.{{0}}.'.format(os.environ.get('CERTBOT_DOMAIN')),"
|
||||
"'value':os.environ.get('CERTBOT_VALIDATION')}}; "
|
||||
"request = requests.post('http://localhost:{1}/set-txt', data=json.dumps(data)); "
|
||||
"request.raise_for_status(); "
|
||||
'"'
|
||||
).format(sys.executable, self.challtestsrv_mgt_port)
|
||||
self.manual_dns_cleanup_hook = (
|
||||
'{0} -c "import os; import requests; import json; '
|
||||
"data = {{'host':'_acme-challenge.{{0}}.'.format(os.environ.get('CERTBOT_DOMAIN'))}}; "
|
||||
"request = requests.post('http://localhost:{1}/clear-txt', data=json.dumps(data)); "
|
||||
"request.raise_for_status(); "
|
||||
'"'
|
||||
).format(sys.executable, self.challtestsrv_mgt_port)
|
||||
|
||||
def cleanup(self):
|
||||
pass
|
||||
shutil.rmtree(self.workspace)
|
||||
|
||||
def certbot_test_no_force_renew(self, args):
|
||||
new_environ = os.environ.copy()
|
||||
new_environ['TMPDIR'] = self.workspace
|
||||
|
||||
additional_args = []
|
||||
if self.certbot_version >= LooseVersion('0.30.0'):
|
||||
additional_args.append('--no-random-sleep-on-renew')
|
||||
|
||||
command = [
|
||||
'certbot',
|
||||
'--server', self.directory_url,
|
||||
'--no-verify-ssl',
|
||||
'--tls-sni-01-port', str(self.tls_alpn_01_port),
|
||||
'--http-01-port', str(self.http_01_port),
|
||||
'--manual-public-ip-logging-ok',
|
||||
'--config-dir', self.config_dir,
|
||||
'--work-dir', os.path.join(self.workspace, 'work'),
|
||||
'--logs-dir', os.path.join(self.workspace, 'logs'),
|
||||
'--non-interactive',
|
||||
'--no-redirect',
|
||||
'--agree-tos',
|
||||
'--register-unsafely-without-email',
|
||||
'--debug',
|
||||
'-vv'
|
||||
]
|
||||
|
||||
command.extend(args)
|
||||
command.extend(additional_args)
|
||||
|
||||
print('Invoke command:\n{0}'.format(subprocess.list2cmdline(command)))
|
||||
return subprocess.check_output(command, universal_newlines=True,
|
||||
cwd=self.workspace, env=new_environ)
|
||||
|
||||
def certbot_test(self, args):
|
||||
command = ['--renew-by-default']
|
||||
command.extend(args)
|
||||
return self.certbot_test_no_force_renew(command)
|
||||
|
||||
def common_no_force_renew(self, args):
|
||||
command = ['--authenticator', 'standalone', '--installer', 'null']
|
||||
command.extend(args)
|
||||
return self.certbot_test_no_force_renew(command)
|
||||
|
||||
def common(self, args):
|
||||
command = ['--renew-by-default']
|
||||
command.extend(args)
|
||||
return self.common_no_force_renew(command)
|
||||
|
||||
def wtf(self, prefix='le'):
|
||||
return '{0}.{1}.wtf'.format(prefix, self.worker_id)
|
||||
|
||||
@@ -1,9 +1,15 @@
|
||||
import requests
|
||||
import urllib3
|
||||
from __future__ import print_function
|
||||
import subprocess
|
||||
import shutil
|
||||
import re
|
||||
import os
|
||||
from os.path import join, exists
|
||||
|
||||
import pytest
|
||||
|
||||
from certbot_integration_tests.certbot_tests.assertions import *
|
||||
from certbot_integration_tests.certbot_tests import context as certbot_context
|
||||
from certbot_integration_tests.utils import misc
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
@@ -16,25 +22,443 @@ def context(request):
|
||||
integration_test_context.cleanup()
|
||||
|
||||
|
||||
def test_hello_1(context):
|
||||
assert context.http_01_port
|
||||
assert context.tls_alpn_01_port
|
||||
try:
|
||||
response = requests.get(context.directory_url, verify=False)
|
||||
response.raise_for_status()
|
||||
assert response.json()
|
||||
response.close()
|
||||
except urllib3.exceptions.InsecureRequestWarning:
|
||||
pass
|
||||
def test_basic_commands(context):
|
||||
# TMPDIR env variable is set to workspace for the certbot subprocess.
|
||||
# So tempdir module will create any temporary files/dirs in workspace,
|
||||
# and its content can be tested to check correct certbot cleanup.
|
||||
initial_count_tmpfiles = len(os.listdir(context.workspace))
|
||||
|
||||
context.common(['--help'])
|
||||
context.common(['--help', 'all'])
|
||||
context.common(['--version'])
|
||||
|
||||
with pytest.raises(subprocess.CalledProcessError):
|
||||
context.common(['--csr'])
|
||||
|
||||
new_count_tmpfiles = len(os.listdir(context.workspace))
|
||||
assert initial_count_tmpfiles == new_count_tmpfiles
|
||||
|
||||
|
||||
def test_hello_2(context):
|
||||
assert context.http_01_port
|
||||
assert context.tls_alpn_01_port
|
||||
try:
|
||||
response = requests.get(context.directory_url, verify=False)
|
||||
response.raise_for_status()
|
||||
assert response.json()
|
||||
response.close()
|
||||
except urllib3.exceptions.InsecureRequestWarning:
|
||||
pass
|
||||
def test_hook_dirs_creation(context):
|
||||
context.common(['register'])
|
||||
|
||||
for hook_dir in misc.list_renewal_hooks_dirs(context.config_dir):
|
||||
assert os.path.isdir(hook_dir)
|
||||
|
||||
|
||||
def test_registration_override(context):
|
||||
context.common(['register'])
|
||||
context.common(['unregister'])
|
||||
context.common(['register', '--email', 'ex1@domain.org,ex2@domain.org'])
|
||||
|
||||
# TODO: When `certbot register --update-registration` is fully deprecated,
|
||||
# delete the two following deprecated uses
|
||||
context.common(['register', '--update-registration', '--email', 'ex1@domain.org'])
|
||||
context.common(['register', '--update-registration', '--email', 'ex1@domain.org,ex2@domain.org'])
|
||||
|
||||
context.common(['update_account', '--email', 'example@domain.org'])
|
||||
context.common(['update_account', '--email', 'ex1@domain.org,ex2@domain.org'])
|
||||
|
||||
|
||||
def test_prepare_plugins(context):
|
||||
output = context.common(['plugins', '--init', '--prepare'])
|
||||
|
||||
assert 'webroot' in output
|
||||
|
||||
|
||||
def test_http_01(context):
|
||||
with misc.create_tcp_server(context.tls_alpn_01_port):
|
||||
certname = context.wtf('le2')
|
||||
context.common([
|
||||
'--domains', certname, '--preferred-challenges', 'http-01', 'run',
|
||||
'--cert-name', certname,
|
||||
'--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')
|
||||
assert_save_renew_hook(context.config_dir, certname)
|
||||
|
||||
|
||||
def test_manual_http_auth(context):
|
||||
with misc.create_tcp_server(context.http_01_port) as webroot:
|
||||
manual_http_hooks = misc.manual_http_hooks(webroot)
|
||||
|
||||
certname = context.wtf()
|
||||
context.common([
|
||||
'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),
|
||||
'--renew-hook', 'echo renew >> "{0}"'.format(context.hook_probe)
|
||||
])
|
||||
|
||||
with pytest.raises(AssertionError):
|
||||
assert_hook_execution(context.hook_probe, 'renew')
|
||||
assert_save_renew_hook(context.config_dir, certname)
|
||||
|
||||
|
||||
def test_manual_dns_auth(context):
|
||||
certname = context.wtf('dns')
|
||||
context.common([
|
||||
'-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)
|
||||
])
|
||||
|
||||
with pytest.raises(AssertionError):
|
||||
assert_hook_execution(context.hook_probe, 'renew')
|
||||
assert_save_renew_hook(context.config_dir, certname)
|
||||
|
||||
|
||||
def test_certonly(context):
|
||||
context.common(['certonly', '--cert-name', 'newname', '-d', context.wtf('newname')])
|
||||
|
||||
|
||||
def test_auth_and_install_with_csr(context):
|
||||
certname = context.wtf('le3')
|
||||
key_path = join(context.workspace, 'key.pem')
|
||||
csr_path = join(context.workspace, 'csr.der')
|
||||
|
||||
misc.generate_csr([certname], key_path, csr_path)
|
||||
|
||||
cert_path = join(context.workspace, 'csr/cert.pem')
|
||||
chain_path = join(context.workspace, 'csr/chain.pem')
|
||||
|
||||
context.common([
|
||||
'auth', '--csr', csr_path,
|
||||
'--cert-path', cert_path,
|
||||
'--chain-path', chain_path
|
||||
])
|
||||
|
||||
print(misc.read_certificate(cert_path))
|
||||
print(misc.read_certificate(chain_path))
|
||||
|
||||
context.common([
|
||||
'--domains', certname, 'install',
|
||||
'--cert-path', cert_path,
|
||||
'--key-path', key_path
|
||||
])
|
||||
|
||||
|
||||
def test_renew(context):
|
||||
# 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.wtf('renew')
|
||||
context.common([
|
||||
'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.common(['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.common_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')
|
||||
|
||||
# Fourth, we modify the time when renew occur to 4 years before expiration.
|
||||
# When trying renew without force, then renew should occur for this large time.
|
||||
# Also we specify to not use hooks dir, so no hook should be run during this renew.
|
||||
# Also this renew should use explicitly a 2048 key size.
|
||||
# And finally we check the file permissions.
|
||||
open(context.hook_probe, 'w').close()
|
||||
with open(join(context.config_dir, 'renewal/{0}.conf'.format(certname)), 'r') as file:
|
||||
lines = file.readlines()
|
||||
lines.insert(4, 'renew_before_expiry = 100 years{0}'.format(os.linesep))
|
||||
with open(join(context.config_dir, 'renewal/{0}.conf'.format(certname)), 'w') as file:
|
||||
file.writelines(lines)
|
||||
context.common_no_force_renew(['renew', '--no-directory-hooks',
|
||||
'--rsa-key-size', '2048'])
|
||||
|
||||
assert_certs_count_for_lineage(context.config_dir, certname, 3)
|
||||
with pytest.raises(AssertionError):
|
||||
assert_hook_execution(context.hook_probe, 'deploy')
|
||||
key2 = join(context.config_dir, 'archive/{0}/privkey2.pem'.format(certname))
|
||||
key3 = join(context.config_dir, 'archive/{0}/privkey3.pem'.format(certname))
|
||||
assert os.stat(key2).st_size > 3000 # 4096 bits keys takes more than 3000 bytes
|
||||
assert os.stat(key3).st_size < 1800 # 2048 bits keys takes less than 1800 bytes
|
||||
|
||||
assert_world_permissions(
|
||||
join(context.config_dir, 'archive/{0}/privkey3.pem'.format(certname)), 4)
|
||||
assert_equals_group_owner(
|
||||
join(context.config_dir, 'archive/{0}/privkey2.pem'.format(certname)),
|
||||
join(context.config_dir, 'archive/{0}/privkey3.pem'.format(certname)))
|
||||
assert_equals_permissions(
|
||||
join(context.config_dir, 'archive/{0}/privkey2.pem'.format(certname)),
|
||||
join(context.config_dir, 'archive/{0}/privkey3.pem'.format(certname)), 0o074)
|
||||
|
||||
# Fifth, we clean every dir hook, and replace their content by empty dir and empty files.
|
||||
# Everything should renew correctly.
|
||||
for hook_dir in misc.list_renewal_hooks_dirs(context.config_dir):
|
||||
shutil.rmtree(hook_dir)
|
||||
os.makedirs(join(hook_dir, 'dir'))
|
||||
open(join(hook_dir, 'file'), 'w').close()
|
||||
context.common(['renew'])
|
||||
|
||||
assert_certs_count_for_lineage(context.config_dir, certname, 4)
|
||||
|
||||
|
||||
def test_hook_override(context):
|
||||
certname = context.wtf('override')
|
||||
context.common([
|
||||
'certonly', '-d', certname,
|
||||
'--preferred-challenges', 'http-01',
|
||||
'--pre-hook', 'echo pre >> "{0}"'.format(context.hook_probe),
|
||||
'--post-hook', 'echo post >> "{0}"'.format(context.hook_probe),
|
||||
'--deploy-hook', 'echo deploy >> "{0}"'.format(context.hook_probe)
|
||||
])
|
||||
|
||||
assert_hook_execution(context.hook_probe, 'pre')
|
||||
assert_hook_execution(context.hook_probe, 'post')
|
||||
assert_hook_execution(context.hook_probe, 'deploy')
|
||||
|
||||
open(context.hook_probe, 'w').close()
|
||||
context.common([
|
||||
'renew', '--cert-name', certname,
|
||||
'--pre-hook', 'echo pre-override >> "{0}"'.format(context.hook_probe),
|
||||
'--post-hook', 'echo post-override >> "{0}"'.format(context.hook_probe),
|
||||
'--deploy-hook', 'echo deploy-override >> "{0}"'.format(context.hook_probe)
|
||||
])
|
||||
|
||||
assert_hook_execution(context.hook_probe, 'pre-override')
|
||||
assert_hook_execution(context.hook_probe, 'post-override')
|
||||
assert_hook_execution(context.hook_probe, 'deploy-override')
|
||||
with pytest.raises(AssertionError):
|
||||
assert_hook_execution(context.hook_probe, 'pre')
|
||||
with pytest.raises(AssertionError):
|
||||
assert_hook_execution(context.hook_probe, 'post')
|
||||
with pytest.raises(AssertionError):
|
||||
assert_hook_execution(context.hook_probe, 'deploy')
|
||||
|
||||
open(context.hook_probe, 'w').close()
|
||||
context.common(['renew', '--cert-name', certname])
|
||||
|
||||
assert_hook_execution(context.hook_probe, 'pre-override')
|
||||
assert_hook_execution(context.hook_probe, 'post-override')
|
||||
assert_hook_execution(context.hook_probe, 'deploy-override')
|
||||
|
||||
|
||||
def test_invalid_domain_with_dns_challenge(context):
|
||||
certs = ','.join([context.wtf('dns1'), context.wtf('fail-dns1')])
|
||||
context.common([
|
||||
'-a', 'manual', '-d', certs,
|
||||
'--allow-subset-of-names',
|
||||
'--preferred-challenges', 'dns',
|
||||
'--manual-auth-hook', context.manual_dns_auth_hook,
|
||||
'--manual-cleanup-hook', context.manual_dns_cleanup_hook
|
||||
])
|
||||
|
||||
output = context.common(['certificates'])
|
||||
|
||||
assert context.wtf('fail-dns1') not in output
|
||||
|
||||
|
||||
def test_reuse_key(context):
|
||||
certname = context.wtf('reusekey')
|
||||
context.common(['--domains', certname, '--reuse-key'])
|
||||
context.common(['renew', '--cert-name', certname])
|
||||
|
||||
with open(join(context.config_dir, 'archive/{0}/privkey1.pem').format(certname), 'r') as file:
|
||||
privkey1 = file.read()
|
||||
with open(join(context.config_dir, 'archive/{0}/privkey2.pem').format(certname), 'r') as file:
|
||||
privkey2 = file.read()
|
||||
assert privkey1 == privkey2
|
||||
|
||||
context.common(['--cert-name', certname, '--domains', certname, '--force-renewal'])
|
||||
|
||||
with open(join(context.config_dir, 'archive/{0}/privkey3.pem').format(certname), 'r') as file:
|
||||
privkey3 = file.read()
|
||||
assert privkey2 != privkey3
|
||||
|
||||
with open(join(context.config_dir, 'archive/{0}/cert1.pem').format(certname), 'r') as file:
|
||||
cert1 = file.read()
|
||||
with open(join(context.config_dir, 'archive/{0}/cert2.pem').format(certname), 'r') as file:
|
||||
cert2 = file.read()
|
||||
with open(join(context.config_dir, 'archive/{0}/cert3.pem').format(certname), 'r') as file:
|
||||
cert3 = file.read()
|
||||
|
||||
assert len({cert1, cert2, cert3}) == 3
|
||||
|
||||
|
||||
def test_ecdsa(context):
|
||||
key_path = join(context.workspace, 'privkey-p384.pem')
|
||||
csr_path = join(context.workspace, 'csr-p384.der')
|
||||
cert_path = join(context.workspace, 'cert-p384.pem')
|
||||
chain_path = join(context.workspace, 'chain-p384.pem')
|
||||
|
||||
misc.generate_csr([context.wtf('ecdsa')], key_path, csr_path, key_type='ECDSA')
|
||||
context.common(['auth', '--csr', csr_path, '--cert-path', cert_path, '--chain-path', chain_path])
|
||||
|
||||
certificate = misc.read_certificate(cert_path)
|
||||
assert 'ASN1 OID: secp384r1' in certificate
|
||||
|
||||
|
||||
def test_ocsp_must_staple(context):
|
||||
certname = context.wtf('must-staple')
|
||||
context.common(['auth', '--must-staple', '--domains', certname])
|
||||
|
||||
certificate = misc.read_certificate(join(context.config_dir,
|
||||
'live/{0}/cert.pem').format(certname))
|
||||
assert 'status_request' in certificate or '1.3.6.1.5.5.7.1.24'
|
||||
|
||||
|
||||
def test_revoke_simple(context):
|
||||
certname = context.wtf()
|
||||
cert_path = join(context.config_dir, 'live/{0}/cert.pem'.format(certname))
|
||||
context.common(['-d', certname])
|
||||
context.common(['revoke', '--cert-path', cert_path, '--delete-after-revoke'])
|
||||
|
||||
assert not exists(cert_path)
|
||||
|
||||
certname = context.wtf('le1')
|
||||
cert_path = join(context.config_dir, 'live/{0}/cert.pem'.format(certname))
|
||||
context.common(['-d', certname])
|
||||
context.common(['revoke', '--cert-path', cert_path, '--no-delete-after-revoke'])
|
||||
|
||||
assert exists(cert_path)
|
||||
context.common(['delete', '--cert-name', certname])
|
||||
|
||||
assert not exists(join(context.config_dir, 'archive/{0}'.format(certname)))
|
||||
assert not exists(join(context.config_dir, 'live/{0}'.format(certname)))
|
||||
assert not exists(join(context.config_dir, 'renewal/{0}.conf').format(certname))
|
||||
|
||||
certname = context.wtf('le2')
|
||||
key_path = join(context.config_dir, 'live/{0}/privkey.pem'.format(certname))
|
||||
cert_path = join(context.config_dir, 'live/{0}/cert.pem'.format(certname))
|
||||
context.common(['-d', certname])
|
||||
context.common(['revoke', '--cert-path', cert_path, '--key-path', key_path])
|
||||
|
||||
|
||||
def test_revoke_and_unregister(context):
|
||||
cert1 = context.wtf('le1')
|
||||
cert2 = context.wtf('le2')
|
||||
cert3 = context.wtf('le3')
|
||||
|
||||
cert_path1 = join(context.config_dir, 'live/{0}/cert.pem'.format(cert1))
|
||||
key_path2 = join(context.config_dir, 'live/{0}/privkey.pem'.format(cert2))
|
||||
cert_path2 = join(context.config_dir, 'live/{0}/cert.pem'.format(cert2))
|
||||
|
||||
context.common(['-d', cert1])
|
||||
context.common(['-d', cert2])
|
||||
context.common(['-d', cert3])
|
||||
|
||||
context.common(['revoke', '--cert-path', cert_path1,
|
||||
'--reason', 'cessationOfOperation'])
|
||||
context.common(['revoke', '--cert-path', cert_path2, '--key-path', key_path2,
|
||||
'--reason', 'keyCompromise'])
|
||||
|
||||
context.common(['unregister'])
|
||||
|
||||
output = context.common(['certificates'])
|
||||
|
||||
assert cert1 not in output
|
||||
assert cert2 not in output
|
||||
assert cert3 in output
|
||||
|
||||
|
||||
def test_revoke_corner_cases(context):
|
||||
cert1 = context.wtf('le1')
|
||||
context.common(['-d', cert1])
|
||||
with pytest.raises(subprocess.CalledProcessError) as error:
|
||||
context.common([
|
||||
'revoke', '--cert-name', cert1,
|
||||
'--cert-path', join(context.config_dir, 'live/{0}/fullchain.pem'.format(cert1))
|
||||
])
|
||||
assert 'Exactly one of --cert-path or --cert-name must be specified' in error.out
|
||||
|
||||
assert os.path.isfile(join(context.config_dir, 'renewal/{0}.conf'.format(cert1)))
|
||||
|
||||
cert2 = context.wtf('le2')
|
||||
context.common(['-d', cert2])
|
||||
with open(join(context.config_dir, 'renewal/{0}.conf'.format(cert2)), 'r') as file:
|
||||
data = file.read()
|
||||
|
||||
data = re.sub('archive_dir = .*{0}'.format(os.linesep),
|
||||
'archive_dir = {0}{1}'.format(os.path.normpath(
|
||||
join(context.config_dir, 'archive/{0}'.format(cert1))), os.linesep),
|
||||
data)
|
||||
|
||||
with open(join(context.config_dir, 'renewal/{0}.conf'.format(cert2)), 'w') as file:
|
||||
file.write(data)
|
||||
|
||||
output = context.common([
|
||||
'revoke', '--cert-path', join(context.config_dir, 'live/{0}/cert.pem'.format(cert1))
|
||||
])
|
||||
|
||||
assert 'Not deleting revoked certs due to overlapping archive dirs' in output
|
||||
|
||||
|
||||
def test_wildcard_certificates(context):
|
||||
if context.acme_server == 'boulder-v1':
|
||||
pytest.skip('Wildcard certificates are not supported on ACME v1')
|
||||
|
||||
certname = context.wtf('wild')
|
||||
|
||||
context.common([
|
||||
'-a', 'manual', '-d', '*.{0},{0}'.format(certname),
|
||||
'--preferred-challenge', 'dns',
|
||||
'--manual-auth-hook', context.manual_dns_auth_hook,
|
||||
'--manual-cleanup-hook', context.manual_dns_cleanup_hook
|
||||
])
|
||||
|
||||
assert exists(join(context.config_dir, 'live/{0}/fullchain.pem'.format(certname)))
|
||||
|
||||
|
||||
def test_ocsp_status(context):
|
||||
if context.acme_server == 'pebble':
|
||||
pytest.skip('Pebble does not support OCSP status requests.')
|
||||
|
||||
# OCSP 1: Check stale OCSP status
|
||||
sample_data_path = misc.load_sample_data_path(context.workspace)
|
||||
output = context.common(['certificates', '--config-dir', sample_data_path])
|
||||
|
||||
assert output.count('TEST_CERT') == 2, ('Did not find two test certs as expected ({0})'
|
||||
.format(output.count('TEST_CERT')))
|
||||
assert output.count('EXPIRED') == 2, ('Did not find two expired certs as expected ({0})'
|
||||
.format(output.count('EXPIRED')))
|
||||
|
||||
# OSCP 2: Check live certificate OCSP status (VALID)
|
||||
output = context.common(['--domains', 'le-ocsp-check.wtf'])
|
||||
|
||||
assert output.count('VALID') == 1, 'Expected le-ocsp-check.wtf to be VALID'
|
||||
assert output.count('EXPIRED') == 0, 'Did not expect le-ocsp-check.wtf to be EXPIRED'
|
||||
|
||||
# OSCP 3: Check live certificate OCSP status (REVOKED)
|
||||
output = context.common(['certificates'])
|
||||
|
||||
assert output.count('INVALID') == 1, 'Expected le-ocsp-check.wtf to be INVALID'
|
||||
assert output.count('REVOKED') == 1, 'Expected le-ocsp-check.wtf to be REVOKED'
|
||||
|
||||
@@ -2,10 +2,20 @@
|
||||
Misc module contains stateless functions that could be used during pytest execution,
|
||||
or outside during setup/teardown of the integration tests environment.
|
||||
"""
|
||||
import subprocess
|
||||
import os
|
||||
import time
|
||||
import contextlib
|
||||
import tempfile
|
||||
import shutil
|
||||
import multiprocessing
|
||||
import sys
|
||||
import stat
|
||||
import errno
|
||||
from distutils.version import LooseVersion
|
||||
|
||||
from six.moves import socketserver, SimpleHTTPServer
|
||||
from OpenSSL import crypto
|
||||
import requests
|
||||
|
||||
|
||||
@@ -30,16 +40,142 @@ def check_until_timeout(url):
|
||||
raise ValueError('Error, url did not respond after 150 attempts: {0}'.format(url))
|
||||
|
||||
|
||||
def find_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'):
|
||||
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):
|
||||
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):
|
||||
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 GraceFullTCPServer(socketserver.TCPServer):
|
||||
allow_reuse_address = True
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def execute_in_given_cwd(cwd):
|
||||
"""
|
||||
Context manager that will execute any command in the given cwd after entering context,
|
||||
and restore current cwd when context is destroyed.
|
||||
:param str cwd: the path to use as the temporary current workspace for python execution
|
||||
"""
|
||||
def create_tcp_server(port):
|
||||
current_cwd = os.getcwd()
|
||||
webroot = tempfile.mkdtemp()
|
||||
|
||||
def run():
|
||||
GraceFullTCPServer(('', port), SimpleHTTPServer.SimpleHTTPRequestHandler).serve_forever()
|
||||
|
||||
process = multiprocessing.Process(target=run)
|
||||
|
||||
try:
|
||||
os.chdir(cwd)
|
||||
yield
|
||||
try:
|
||||
os.chdir(webroot)
|
||||
process.start()
|
||||
finally:
|
||||
os.chdir(current_cwd)
|
||||
|
||||
check_until_timeout('http://localhost:{0}/'.format(port))
|
||||
|
||||
yield webroot
|
||||
finally:
|
||||
os.chdir(current_cwd)
|
||||
try:
|
||||
if process.is_alive():
|
||||
process.terminate()
|
||||
process.join() # Block until process is effectively terminated
|
||||
finally:
|
||||
shutil.rmtree(webroot)
|
||||
|
||||
|
||||
def list_renewal_hooks_dirs(config_dir):
|
||||
renewal_hooks_root = os.path.join(config_dir, 'renewal-hooks')
|
||||
return [os.path.join(renewal_hooks_root, item) for item in ['pre', 'deploy', 'post']]
|
||||
|
||||
|
||||
def generate_test_file_hooks(config_dir, hook_probe):
|
||||
if sys.platform == 'win32':
|
||||
extension = 'bat'
|
||||
else:
|
||||
extension = 'sh'
|
||||
|
||||
renewal_hooks_dirs = list_renewal_hooks_dirs(config_dir)
|
||||
|
||||
for hook_dir in renewal_hooks_dirs:
|
||||
try:
|
||||
os.makedirs(hook_dir)
|
||||
except OSError as error:
|
||||
if error.errno != errno.EEXIST:
|
||||
raise
|
||||
hook_path = os.path.join(hook_dir, 'hook.{0}'.format(extension))
|
||||
if extension == 'sh':
|
||||
data = '''\
|
||||
#!/bin/bash -xe
|
||||
if [ "$0" == "{0}" ]; then
|
||||
if [ -z "$RENEWED_DOMAINS" -o -z "$RENEWED_LINEAGE" ]; then
|
||||
echo "Environment variables not properly set!" >&2
|
||||
exit 1
|
||||
fi
|
||||
fi
|
||||
echo $(basename $(dirname "$0")) >> "{1}"\
|
||||
'''.format(hook_path, hook_probe)
|
||||
else:
|
||||
data = '''\
|
||||
|
||||
'''
|
||||
with open(hook_path, 'w') as file:
|
||||
file.write(data)
|
||||
os.chmod(hook_path, os.stat(hook_path).st_mode | stat.S_IEXEC)
|
||||
|
||||
|
||||
def manual_http_hooks(http_server_root):
|
||||
auth = (
|
||||
'{0} -c "import os; '
|
||||
"challenge_dir = os.path.join('{1}', '.well-known/acme-challenge'); "
|
||||
'os.makedirs(challenge_dir); '
|
||||
"challenge_file = os.path.join(challenge_dir, os.environ.get('CERTBOT_TOKEN')); "
|
||||
"open(challenge_file, 'w').write(os.environ.get('CERTBOT_VALIDATION')); "
|
||||
'"'
|
||||
).format(sys.executable, http_server_root)
|
||||
cleanup = (
|
||||
'{0} -c "import os; import shutil; '
|
||||
"well_known = os.path.join('{1}', '.well-known'); "
|
||||
'shutil.rmtree(well_known); '
|
||||
'"'
|
||||
).format(sys.executable, http_server_root)
|
||||
|
||||
return auth, cleanup
|
||||
|
||||
|
||||
def get_certbot_version():
|
||||
output = subprocess.check_output(['certbot', '--version'], universal_newlines=True)
|
||||
# Typical response is: output = 'certbot 0.31.0.dev0'
|
||||
# version_str = output.split(' ')[1]
|
||||
return LooseVersion('0.31.0.dev0')
|
||||
|
||||
@@ -8,8 +8,12 @@ install_requires = [
|
||||
'pytest',
|
||||
'pytest-cov',
|
||||
'pytest-xdist',
|
||||
'pytest-rerunfailures==4.2',
|
||||
'pytest-sugar',
|
||||
'coverage',
|
||||
'six',
|
||||
'pyopenssl',
|
||||
'cryptography',
|
||||
'requests',
|
||||
'pyyaml',
|
||||
]
|
||||
|
||||
@@ -0,0 +1,76 @@
|
||||
#!/usr/bin/env python
|
||||
# -*- coding: utf-8 -*-
|
||||
# This script generates a simple SAN CSR to be used with Let's Encrypt
|
||||
# CA. Mostly intended for "auth --csr" testing, but, since it's easily
|
||||
# auditable, feel free to adjust it and use it on your production web
|
||||
# server.
|
||||
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)
|
||||
@@ -52,6 +52,7 @@ pytest-cov==2.5.1
|
||||
pytest-forked==0.2
|
||||
pytest-xdist==1.22.5
|
||||
pytest-sugar==0.9.2
|
||||
pytest-rerunfailures==4.2
|
||||
python-dateutil==2.6.1
|
||||
python-digitalocean==1.11
|
||||
PyYAML==3.13
|
||||
|
||||
Reference in New Issue
Block a user