starred/certbot-certbot
1
0
Fork 0
mirror of https://github.com/certbot/certbot.git synced 2026-07-28 08:45:22 +02:00
Code Issues Packages Projects Releases Wiki Activity

route53: make sleep duration configurable like other DNS authenticators (#4771)

Browse Source 
* Re-structure perform to allow for easier refactoring

* Refactor to use dns_common

* Make ttl a class variable, like other plugins
This commit is contained in:
Zach Shepherd
2017-06-05 21:20:17 -07:00
committed by Brad Warren
parent a926d42bd6
commit f0e1be55d6
2 changed files with 35 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
certbot-route53/certbot_route53/authenticator.py
+21 -39
View File
@@ -4,17 +4,14 @@ import time
import boto3
import zope.interface
from acme import challenges
from botocore.exceptions import NoCredentialsError, ClientError
from certbot import errors
from certbot import interfaces
from certbot.plugins import common
from certbot.plugins import dns_common
logger = logging.getLogger(__name__)
TTL = 10
INSTRUCTIONS = (
"To use certbot-route53, configure credentials as described at "
"https://boto3.readthedocs.io/en/latest/guide/configuration.html#best-practices-for-configuring-credentials " # pylint: disable=line-too-long
@@ -22,53 +19,41 @@ INSTRUCTIONS = (
@zope.interface.implementer(interfaces.IAuthenticator)
@zope.interface.provider(interfaces.IPluginFactory)
class Authenticator(common.Plugin):
class Authenticator(dns_common.DNSAuthenticator):
"""Route53 Authenticator
This authenticator solves a DNS01 challenge by uploading the answer to AWS
Route53.
"""
description = "Obtain certificates using a DNS TXT record (if you are using AWS Route53 for DNS)."
description = ("Obtain certificates using a DNS TXT record (if you are using AWS Route53 for "
"DNS).")
ttl = 10
def __init__(self, *args, **kwargs):
super(Authenticator, self).__init__(*args, **kwargs)
self.r53 = boto3.client("route53")
def prepare(self): # pylint: disable=missing-docstring,no-self-use
pass # pragma: no cover
def more_info(self): # pylint: disable=missing-docstring,no-self-use
return "Solve a DNS01 challenge using AWS Route53"
def get_chall_pref(self, domain):
# pylint: disable=missing-docstring,no-self-use,unused-argument
return [challenges.DNS01]
def _setup_credentials(self):
pass
def perform(self, achalls): # pylint: disable=missing-docstring
def _perform(self, domain, validation_domain_name, validation):
try:
change_ids = [
self._change_txt_record("UPSERT", achall)
for achall in achalls
]
change_id = self._change_txt_record("UPSERT", validation_domain_name, validation)
for change_id in change_ids:
self._wait_for_change(change_id)
# Sleep for at least the TTL, to ensure that any records cached by
# the ACME server after previous validation attempts are gone. In
# most cases we'll need to wait at least this long for the Route53
# records to propagate, so this doesn't delay us much.
time.sleep(TTL)
return [achall.response(achall.account_key) for achall in achalls]
self._wait_for_change(change_id)
except (NoCredentialsError, ClientError) as e:
logger.debug('Encountered error during perform: %s', e, exc_info=True)
raise errors.PluginError("\n".join([str(e), INSTRUCTIONS]))
def cleanup(self, achalls): # pylint: disable=missing-docstring
for achall in achalls:
try:
self._change_txt_record("DELETE", achall)
except (NoCredentialsError, ClientError) as e:
logger.debug('Encountered error during cleanup: %s', e, exc_info=True)
def _cleanup(self, domain, validation_domain_name, validation):
try:
self._change_txt_record("DELETE", validation_domain_name, validation)
except (NoCredentialsError, ClientError) as e:
logger.debug('Encountered error during cleanup: %s', e, exc_info=True)
def _find_zone_id_for_domain(self, domain):
"""Find the zone id responsible a given FQDN.
@@ -100,11 +85,8 @@ class Authenticator(common.Plugin):
zones.sort(key=lambda z: len(z[0]), reverse=True)
return zones[0][1]
def _change_txt_record(self, action, achall):
domain = achall.validation_domain_name(achall.domain)
value = achall.validation(achall.account_key)
zone_id = self._find_zone_id_for_domain(domain)
def _change_txt_record(self, action, validation_domain_name, validation):
zone_id = self._find_zone_id_for_domain(validation_domain_name)
response = self.r53.change_resource_record_sets(
HostedZoneId=zone_id,
@@ -114,13 +96,13 @@ class Authenticator(common.Plugin):
{
"Action": action,
"ResourceRecordSet": {
"Name": domain,
"Name": validation_domain_name,
"Type": "TXT",
"TTL": TTL,
"TTL": self.ttl,
"ResourceRecords": [
# For some reason TXT records need to be
# manually quoted.
{"Value": '"{0}"'.format(value)}
{"Value": '"{0}"'.format(validation)}
],
}
}
certbot-route53/certbot_route53/authenticator_test.py
+14 -6
View File
@@ -7,6 +7,7 @@ from botocore.exceptions import NoCredentialsError, ClientError
from certbot import errors
from certbot.plugins import dns_test_common
from certbot.plugins.dns_test_common import DOMAIN
class AuthenticatorTest(unittest.TestCase, dns_test_common.BaseAuthenticatorTest):
@@ -21,16 +22,15 @@ class AuthenticatorTest(unittest.TestCase, dns_test_common.BaseAuthenticatorTest
self.auth = Authenticator(self.config, "route53")
def test_parser_arguments(self):
pass # TODO: follow convention of defining optional argument for DNS propagation delay
def test_perform(self):
self.auth._change_txt_record = mock.MagicMock()
self.auth._wait_for_change = mock.MagicMock()
self.auth.perform([self.achall])
self.auth._change_txt_record.assert_called_once_with("UPSERT", self.achall)
self.auth._change_txt_record.assert_called_once_with("UPSERT",
'_acme-challenge.' + DOMAIN,
mock.ANY)
self.auth._wait_for_change.assert_called_once()
def test_perform_no_credentials_error(self):
@@ -49,18 +49,26 @@ class AuthenticatorTest(unittest.TestCase, dns_test_common.BaseAuthenticatorTest
[self.achall])
def test_cleanup(self):
self.auth._attempt_cleanup = True
self.auth._change_txt_record = mock.MagicMock()
self.auth.cleanup([self.achall])
self.auth._change_txt_record.assert_called_once_with("DELETE", self.achall)
self.auth._change_txt_record.assert_called_once_with("DELETE",
'_acme-challenge.'+DOMAIN,
mock.ANY)
def test_cleanup_no_credentials_error(self):
self.auth._attempt_cleanup = True
self.auth._change_txt_record = mock.MagicMock(side_effect=NoCredentialsError)
self.auth.cleanup([self.achall])
def test_cleanup_client_error(self):
self.auth._attempt_cleanup = True
self.auth._change_txt_record = mock.MagicMock(
side_effect=ClientError({"Error": {"Code": "foo"}}, "bar"))
@@ -173,7 +181,7 @@ class ClientTest(unittest.TestCase):
self.client.r53.change_resource_record_sets = mock.MagicMock(
return_value={"ChangeInfo": {"Id": 1}})
self.client._change_txt_record("FOO", dns_test_common.BaseAuthenticatorTest.achall)
self.client._change_txt_record("FOO", DOMAIN, "foo")
self.client.r53.change_resource_record_sets.assert_called_once()