diff options
Diffstat (limited to 'tests/test_acme_dns_tiny.py')
| -rw-r--r-- | tests/test_acme_dns_tiny.py | 121 |
1 files changed, 78 insertions, 43 deletions
diff --git a/tests/test_acme_dns_tiny.py b/tests/test_acme_dns_tiny.py index 0d61b97..5a4a479 100644 --- a/tests/test_acme_dns_tiny.py +++ b/tests/test_acme_dns_tiny.py @@ -1,9 +1,7 @@ #!python3 -import collections import contextlib -import inspect import logging import os import subprocess @@ -26,7 +24,6 @@ acme_dns_tiny.log_configure = False ACME_STAGING_DIRECTORY = \ 'https://acme-staging-v02.api.letsencrypt.org/directory' DOMAIN = os.getenv('TEST_DOMAIN') -CONTACT = os.getenv('TEST_CONTACT') # form: mailto:name@domain SCRIPT = os.getenv('TEST_SCRIPT') @@ -39,6 +36,28 @@ key_types = ( ('ecdsa', 256, False), ) +# the first case of success expectation, try also to test the anti-replay +# nonce expiration retry mechanism by waiting for it to expire (more than +# 5min). +first_success_case_nonce_timeout_done = False +original_sreq_method = acme_dns_tiny.ACME.sreq + + +nonce_expiration_sreq_wrapper_counter_max = 3 +nonce_expiration_sreq_wrapper_counter = nonce_expiration_sreq_wrapper_counter_max +def nonce_expiration_sreq_wrapper(*args, **kwargs): + global nonce_expiration_sreq_wrapper_counter + nonce_expiration_sreq_wrapper_counter -= 1 + if nonce_expiration_sreq_wrapper_counter <= 0: + nonce_expiration_sreq_wrapper_counter = nonce_expiration_sreq_wrapper_counter_max + wait = 30 + logging.info('waiting %dmin for nonce expiration test', wait) + for i in range(wait, 0, -1): + logging.info('%d min left', i) + time.sleep(60) + acme_dns_tiny.ACME.sreq = original_sreq_method # reset original method + return original_sreq_method(*args, **kwargs) + @contextlib.contextmanager def does_not_raise(): @@ -118,16 +137,16 @@ def keys_generator(): @pytest.fixture(scope='module', params=keys_generator()) -def account_domain_key(request): +def keysdef(request): return request.param @pytest.fixture(scope='module', params=[ {'subj': f'/CN={DOMAIN}', 'san': None}, {'subj': f'/CN=*.{DOMAIN}', 'san': None}, - {'subj': f'/', 'san': f'DNS:{DOMAIN},DNS:www.{DOMAIN}'}, + {'subj': '/', 'san': f'DNS:{DOMAIN},DNS:www.{DOMAIN}'}, {'subj': f'/CN={DOMAIN}', 'san': f'DNS:www.{DOMAIN}'}, - {'subj': f'/', 'san': f'DNS:{DOMAIN},DNS:*.{DOMAIN}'}, + {'subj': '/', 'san': f'DNS:{DOMAIN},DNS:*.{DOMAIN}'}, ]) def subj_fixture(request): return request.param @@ -136,42 +155,32 @@ def subj_fixture(request): CSR_ARGS_COUNTER = 0 @pytest.fixture(scope='module', params=[ - pytest.param(None, id='no_separator/no_contact'), + pytest.param({}, id='no_separator'), pytest.param({'args': {'--separator': '\\0'}}, id='separator'), - pytest.param({'args': {'--contact': [CONTACT]}}, id='contact'), - pytest.param({'args': { - '--contact': [CONTACT], - '--separator': '\\0', - }}, id='separator+contact'), ]) -def main_args(tmpdir_factory, account_domain_key, request, subj_fixture): +def main_args(tmpdir_factory, keysdef, request, subj_fixture): global CSR_ARGS_COUNTER CSR_ARGS_COUNTER += 1 tmpdir = tmpdir_factory.mktemp("data") - #keysdef = account_domain_key.values[0] - keysdef = account_domain_key - #print('debug', keysdef) - default_args = { + account_key = keysdef['account_key_path'](tmpdir) + args = { '--acme-directory': ACME_STAGING_DIRECTORY, '--script': SCRIPT, - '--account-key': keysdef['account_key_path'](tmpdir), + '--account-key': account_key, #'--ttl': 60, # already the default '--ttl': None, '--separator': None, - '--contact': None, '--verbose': None, '--quiet': None, + **request.param.get('args', {}) } - if request.param: - default_args.update(request.param.get('args', {})) - - if default_args['--separator'] is not None: - assert default_args['--separator'] == '\\0' - assert default_args['--separator'].encode( + if args['--separator'] is not None: + assert args['--separator'] == '\\0' + assert args['--separator'].encode( 'utf8').decode('unicode_escape') == '\0' name = f'{CSR_ARGS_COUNTER:02X}' @@ -191,13 +200,16 @@ def main_args(tmpdir_factory, account_domain_key, request, subj_fixture): '-out', path ]) - return {**default_args, '--csr': path}, keysdef['raise_expected'] + return {**args, '--csr': path}, { + **keysdef, + 'account_key_path': account_key, # expanded version + 'domain_key_path': domain_key, # expanded version + } def test_sanity_env(): 'check environment is correctly set before running other tests' assert bool(DOMAIN) - assert bool(CONTACT) assert bool(SCRIPT) @@ -206,19 +218,33 @@ def test_sanity_command(): subprocess.run([SCRIPT, 'add', f'_acme-challenge.{DOMAIN}.', 'dummy']) -def assert_cert(capsys, args): +def assert_cert(capsys, args, keysdef): captured = capsys.readouterr() #assert not captured.err #certlist = captured.out.split() - if args['--separator'] is None: + logging.debug('captured stdout %s', captured.out) + logging.debug('captured stderr %s', captured.err) + + # Subscriber certificates with RSA public keys are issued from our RSA + # intermediates, which are issued only from our RSA root ISRG Root X1 + # (i.e. they are not cross-signed). Therefore, all RSA subscriber certificates + # have only a single chain available. (since 2025-06-11, see here for more + # information https://letsencrypt.org/certificates/). + if ( + args['--separator'] is None + or ( + args['--separator'] is not None + and '/domain_rsa_' in keysdef['domain_key_path'] + ) + ): assert '\0' not in captured.out else: assert '\0' in captured.out for chain in captured.out.split('\0'): - assert chain.count('-----BEGIN CERTIFICATE-----') == 2 - assert chain.count('-----END CERTIFICATE-----') == 2 + assert chain.count('-----BEGIN CERTIFICATE-----') >= 2 + assert chain.count('-----END CERTIFICATE-----') >= 2 textchain = acme_dns_tiny.openssl(['x509', '-text', '-noout'], stdin=chain) @@ -228,33 +254,42 @@ def assert_cert(capsys, args): 'certtool', '-e', '--verify-profile', 'low'], input=chain, encoding='utf8') assert ' Verified.' in certtool_out - assert certtool_out.count('Subject:') == 3 + assert certtool_out.count('Subject:') >= 3 -def module_main_caller(*, capsys, args, expectation): +def module_main_caller(*, capsys, args, expectation, do_expire_nonce, keysdef): logging.info(f'module_main_caller({args}, {expectation})') logging.debug('before call to acme_dns_tiny.main()') with expectation: + acme_dns_tiny.ACME.sreq = original_sreq_method + if do_expire_nonce: + logging.info('doing expire nonce test') + acme_dns_tiny.ACME.sreq = nonce_expiration_sreq_wrapper acme_dns_tiny.main(args) # check_cert is under the expectation context manager since if # acme_dns_tiny.main() raises, following statement must not be run. - assert_cert(capsys, args) + assert_cert(capsys, args, keysdef) logging.debug('after call to acme_dns_tiny.main()') def test_main(main_args, capsys): - t_start = time.time() + #t_start = time.time() args = main_args[0] - raise_expected = main_args[1] + keysdef = main_args[1] + raise_expected = keysdef['raise_expected'] + do_expire_nonce = False #print('subj', subj, 'args', args) expectation = does_not_raise() if raise_expected: expectation = pytest.raises(ValueError) - module_main_caller(capsys=capsys, args=args, expectation=expectation) - t_stop = time.time() - - # calculate for letsencrypt rate limit (50account per hour) - t_diff = 216 - (t_stop - t_start) - if t_diff > 0: - time.sleep(t_diff) + elif not first_success_case_nonce_timeout_done: + do_expire_nonce = True + module_main_caller(capsys=capsys, args=args, expectation=expectation, + do_expire_nonce=do_expire_nonce, + keysdef=keysdef) + #t_stop = time.time() + ## calculate for letsencrypt rate limit (50account per hour) + #t_diff = 216 - (t_stop - t_start) + #if t_diff > 0 and os.environ.get('ACCOUNT_SLEEP_SKIP') != '1': + # time.sleep(t_diff) |