diff options
| -rw-r--r-- | acme_dns_tiny.py | 17 | ||||
| -rw-r--r-- | tests/test_acme_dns_tiny.py | 44 |
2 files changed, 52 insertions, 9 deletions
diff --git a/acme_dns_tiny.py b/acme_dns_tiny.py index 051e6ba..4903c63 100644 --- a/acme_dns_tiny.py +++ b/acme_dns_tiny.py @@ -177,17 +177,22 @@ class ACME: self.jwk_thumbprint = b64sha256(json.dumps(header['jwk'], sort_keys=True, separators=(',', ':'))) + def request_new_nonce(self): + log.info('Request new nonce') + self.jws_header['nonce'] = requests.get(self.directory["newNonce"] + ).headers['Replay-Nonce'] + def init_sreqs(self, directory_url=None): self.generate_jws_header(self.account_key_path) log.info('Fetch information from the ACME directory.') self.directory = dirmap = requests.get(directory_url, headers=self.dirheaders).json() - log.info('Request first nonce') - self.jws_header['nonce'] = requests.get(dirmap["newNonce"] - ).headers['Replay-Nonce'] + self.request_new_nonce() - def sreq(self, url=None, payload=None, headers=None): + def sreq(self, url=None, payload=None, headers=None, max_retry=10): """Sends signed requests to ACME server.""" + if max_retry < 0: + raise ValueError('max_retry exceeded') # POST-as-GET (payload=None != {}), payload64 is an empty string payload64 = '' if payload is None else b64json(payload) protected64 = b64json({**self.jws_header, 'url': url}) @@ -207,6 +212,10 @@ class ACME: with contextlib.suppress(ValueError): jmap = req.json() if req.status_code not in (200, 201): + if jmap['type'] == 'urn:ietf:params:acme:error:badNonce': + log.info('nonce expired or invalid, retry (left %d)', max_retry) + self.request_new_nonce() + return self.sreq(url, payload, headers, max_retry-1) raise ValueError(f'Request failed: {req.status_code} {jmap}, ' 'hint: {sreq.headers.get("Link", "empty")}.') return collections.namedtuple('sreq', ['code', 'headers', 'map', diff --git a/tests/test_acme_dns_tiny.py b/tests/test_acme_dns_tiny.py index 0d61b97..3a29ce4 100644 --- a/tests/test_acme_dns_tiny.py +++ b/tests/test_acme_dns_tiny.py @@ -39,6 +39,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(): @@ -211,14 +233,17 @@ def assert_cert(capsys, args): #assert not captured.err #certlist = captured.out.split() + logging.debug('captured stdout %s', captured.out) + logging.debug('captured stderr %s', captured.err) + if args['--separator'] is None: 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,14 +253,19 @@ 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): 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') + first_success_case_nonce_timeout_done = True + 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. @@ -247,11 +277,15 @@ def test_main(main_args, capsys): t_start = time.time() args = main_args[0] raise_expected = main_args[1] + 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) + 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) t_stop = time.time() # calculate for letsencrypt rate limit (50account per hour) |