mirror of
https://github.com/certbot/certbot.git
synced 2026-07-28 08:45:22 +02:00
Cloudflare DNS Authenticator
Implement an Authenticator which can fulfill a dns-01 challenge using the
Cloudflare API. Applicable only for domains using Cloudflare for DNS.
Testing Done:
* `tox -e py27`
* `tox -e lint`
* Manual testing:
* Used `certbot certonly --dns-cloudflare -d`, specifying a
credentials file as a command line argument. Verified that a
certificate was successfully obtained without user interaction.
* Used `certbot certonly --dns-cloudflare -d`, without specifying a
credentials file as a command line argument. Verified that the user
was prompted and that a certificate was successfully obtained.
* Used `certbot certonly -d`. Verified that the user was prompted for
a credentials file after selecting cloudflare interactively and
that a certificate was successfully obtained.
* Used `certbot renew --force-renewal`. Verified that certificates
were renewed without user interaction.
* Negative testing:
* Path to non-existent credentials file.
* Credentials file with unsafe permissions (644).
* Credentials file missing e-mail address.
* Credentials file with blank API key.
* Credentials file with incorrect e-mail address.
* Credentials file with malformed API key.
* Credentials file with invalid API key.
* Domain name not registered to Cloudflare account.
This commit is contained in:
Zach Shepherd
@@ -28,6 +28,7 @@ class PluginEntryPoint(object):
|
||||
PREFIX_FREE_DISTRIBUTIONS = [
|
||||
"certbot",
|
||||
"certbot-apache",
|
||||
"certbot-dns-cloudflare",
|
||||
"certbot-nginx",
|
||||
]
|
||||
"""Distributions for which prefix will be omitted."""
|
||||
|
||||
@@ -0,0 +1,323 @@
|
||||
"""Common code for DNS Authenticator Plugins."""
|
||||
|
||||
import abc
|
||||
import logging
|
||||
import os
|
||||
import stat
|
||||
from time import sleep
|
||||
|
||||
import configobj
|
||||
import zope.interface
|
||||
from acme import challenges
|
||||
|
||||
from certbot import errors
|
||||
from certbot import interfaces
|
||||
from certbot.display import ops
|
||||
from certbot.display import util as display_util
|
||||
from certbot.plugins import common
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@zope.interface.implementer(interfaces.IAuthenticator)
|
||||
@zope.interface.provider(interfaces.IPluginFactory)
|
||||
class DNSAuthenticator(common.Plugin):
|
||||
"""Base class for DNS Authenticators"""
|
||||
|
||||
def __init__(self, config, name):
|
||||
super(DNSAuthenticator, self).__init__(config, name)
|
||||
|
||||
self._attempt_cleanup = False
|
||||
|
||||
@classmethod
|
||||
def add_parser_arguments(cls, add):
|
||||
add('propagation-seconds',
|
||||
default=10,
|
||||
type=int,
|
||||
help='The number of seconds to wait for DNS to propagate before asking the ACME server '
|
||||
'to verify the DNS record.')
|
||||
|
||||
def get_chall_pref(self, unused_domain): # pylint: disable=missing-docstring,no-self-use
|
||||
return [challenges.DNS01]
|
||||
|
||||
def prepare(self): # pylint: disable=missing-docstring
|
||||
pass
|
||||
|
||||
def perform(self, achalls): # pylint: disable=missing-docstring
|
||||
self._setup_credentials()
|
||||
|
||||
self._attempt_cleanup = True
|
||||
|
||||
responses = []
|
||||
for achall in achalls:
|
||||
domain = achall.domain
|
||||
validation_domain_name = achall.validation_domain_name(domain)
|
||||
validation = achall.validation(achall.account_key)
|
||||
|
||||
self._perform(domain, validation_domain_name, validation)
|
||||
responses.append(achall.response(achall.account_key))
|
||||
|
||||
# DNS updates take time to propagate and checking to see if the update has occurred is not
|
||||
# reliable (the machine this code is running on might be able to see an update before
|
||||
# the ACME server). So: we sleep for a short amount of time we believe to be long enough.
|
||||
logger.info("Waiting %d seconds for DNS changes to propagate",
|
||||
self.conf('propagation-seconds'))
|
||||
sleep(self.conf('propagation-seconds'))
|
||||
|
||||
return responses
|
||||
|
||||
def cleanup(self, achalls): # pylint: disable=missing-docstring
|
||||
if self._attempt_cleanup:
|
||||
for achall in achalls:
|
||||
domain = achall.domain
|
||||
validation_domain_name = achall.validation_domain_name(domain)
|
||||
validation = achall.validation(achall.account_key)
|
||||
|
||||
self._cleanup(domain, validation_domain_name, validation)
|
||||
|
||||
@abc.abstractmethod
|
||||
def _setup_credentials(self): # pragma: no cover
|
||||
"""
|
||||
Establish credentials, prompting if necessary.
|
||||
"""
|
||||
raise NotImplementedError()
|
||||
|
||||
@abc.abstractmethod
|
||||
def _perform(self, domain, validation_domain_name, validation): # pragma: no cover
|
||||
"""
|
||||
Performs a dns-01 challenge by creating a DNS TXT record.
|
||||
|
||||
:param str domain: The domain being validated.
|
||||
:param str validation_domain_name: The validation record domain name.
|
||||
:param str validation: The validation record content.
|
||||
:raises errors.PluginError: If the challenge cannot be performed
|
||||
"""
|
||||
raise NotImplementedError()
|
||||
|
||||
@abc.abstractmethod
|
||||
def _cleanup(self, domain, validation_domain_name, validation): # pragma: no cover
|
||||
"""
|
||||
Deletes the DNS TXT record which would have been created by `_perform_achall`.
|
||||
|
||||
Fails gracefully if no such record exists.
|
||||
|
||||
:param str domain: The domain being validated.
|
||||
:param str validation_domain_name: The validation record domain name.
|
||||
:param str validation: The validation record content.
|
||||
"""
|
||||
raise NotImplementedError()
|
||||
|
||||
def _configure(self, key, label):
|
||||
"""
|
||||
Ensure that a configuration value is available.
|
||||
|
||||
If necessary, prompts the user and stores the result.
|
||||
|
||||
:param str key: The configuration key.
|
||||
:param str label: The user-friendly label for this piece of information.
|
||||
"""
|
||||
|
||||
configured_value = self.conf(key)
|
||||
if not configured_value:
|
||||
new_value = self._prompt_for_data(label)
|
||||
|
||||
setattr(self.config, self.dest(key), new_value)
|
||||
|
||||
def _configure_file(self, key, label, validator=None):
|
||||
"""
|
||||
Ensure that a configuration value is available for a path.
|
||||
|
||||
If necessary, prompts the user and stores the result.
|
||||
|
||||
:param str key: The configuration key.
|
||||
:param str label: The user-friendly label for this piece of information.
|
||||
"""
|
||||
|
||||
configured_value = self.conf(key)
|
||||
if not configured_value:
|
||||
new_value = self._prompt_for_file(label, validator)
|
||||
|
||||
setattr(self.config, self.dest(key), os.path.abspath(os.path.expanduser(new_value)))
|
||||
|
||||
def _configure_credentials(self, key, label, required_variables=None):
|
||||
"""
|
||||
As `_configure_file`, but for a credential configuration file.
|
||||
|
||||
If necessary, prompts the user and stores the result.
|
||||
|
||||
Always stores absolute paths to avoid issues during renewal.
|
||||
|
||||
:param str key: The configuration key.
|
||||
:param str label: The user-friendly label for this piece of information.
|
||||
:param dict required_variables: Map of variable which must be present to error to display.
|
||||
"""
|
||||
|
||||
def __validator(filename):
|
||||
if required_variables:
|
||||
CredentialsConfiguration(filename, self.dest).require(required_variables)
|
||||
|
||||
self._configure_file(key, label, __validator)
|
||||
|
||||
credentials_configuration = CredentialsConfiguration(self.conf(key), self.dest)
|
||||
if required_variables:
|
||||
credentials_configuration.require(required_variables)
|
||||
|
||||
return credentials_configuration
|
||||
|
||||
@staticmethod
|
||||
def _prompt_for_data(label):
|
||||
"""
|
||||
Prompt the user for a piece of information.
|
||||
|
||||
:param str label: The user-friendly label for this piece of information.
|
||||
:returns: The user's response (guaranteed non-empty).
|
||||
:rtype: str
|
||||
"""
|
||||
|
||||
def __validator(i):
|
||||
if not i:
|
||||
raise errors.PluginError('Please enter your {0}.'.format(label))
|
||||
|
||||
code, response = ops.validated_input(
|
||||
__validator,
|
||||
'Input your {0}'.format(label),
|
||||
force_interactive=True)
|
||||
|
||||
if code == display_util.OK:
|
||||
return response
|
||||
else:
|
||||
raise errors.PluginError('{0} required to proceed.'.format(label))
|
||||
|
||||
@staticmethod
|
||||
def _prompt_for_file(label, validator=None):
|
||||
"""
|
||||
Prompt the user for a path.
|
||||
|
||||
:param str label: The user-friendly label for the file.
|
||||
:param callable validator: A method which will be called to validate the supplied input
|
||||
after it has been validated to be a non-empty path to an existing file. Should throw a
|
||||
`~certbot.errors.PluginError` to indicate any issue.
|
||||
:returns: The user's response (guaranteed to exist).
|
||||
:rtype: str
|
||||
"""
|
||||
|
||||
def __validator(filename):
|
||||
if not filename:
|
||||
raise errors.PluginError('Please enter a valid path to your {0}.'.format(label))
|
||||
|
||||
filename = os.path.expanduser(filename)
|
||||
|
||||
validate_file(filename)
|
||||
|
||||
if validator:
|
||||
validator(filename)
|
||||
|
||||
code, response = ops.validated_directory(
|
||||
__validator,
|
||||
'Input the path to your {0}'.format(label),
|
||||
force_interactive=True)
|
||||
|
||||
if code == display_util.OK:
|
||||
return response
|
||||
else:
|
||||
raise errors.PluginError('{0} required to proceed.'.format(label))
|
||||
|
||||
|
||||
class CredentialsConfiguration(object):
|
||||
"""Represents a user-supplied filed which stores API credentials."""
|
||||
|
||||
def __init__(self, filename, mapper=lambda x: x):
|
||||
"""
|
||||
:param str filename: A path to the configuration file.
|
||||
:param callable mapper: A transformation to apply to configuration key names
|
||||
:raises errors.PluginError: If the file does not exist or is not a valid format.
|
||||
"""
|
||||
validate_file_permissions(filename)
|
||||
|
||||
try:
|
||||
self.confobj = configobj.ConfigObj(filename)
|
||||
except configobj.ConfigObjError as e:
|
||||
logger.debug("Error parsing credentials configuration: %s", e, exc_info=True)
|
||||
raise errors.PluginError("Error parsing credentials configuration: {0}".format(e))
|
||||
|
||||
self.mapper = mapper
|
||||
|
||||
def require(self, required_variables):
|
||||
"""Ensures that the supplied set of variables are all present in the file.
|
||||
|
||||
:param dict required_variables: Map of variable which must be present to error to display.
|
||||
:raises errors.PluginError: If one or more are missing.
|
||||
"""
|
||||
messages = []
|
||||
|
||||
for var in required_variables:
|
||||
if not self._has(var):
|
||||
messages.append('Property "{0}" not found (should be {1}).'
|
||||
.format(self.mapper(var), required_variables[var]))
|
||||
elif not self._get(var):
|
||||
messages.append('Property "{0}" not set (should be {1}).'
|
||||
.format(self.mapper(var), required_variables[var]))
|
||||
|
||||
if messages:
|
||||
raise errors.PluginError(
|
||||
'Missing {0} in credentials configuration file {1}:\n * {2}'.format(
|
||||
'property' if len(messages) == 1 else 'properties',
|
||||
self.confobj.filename,
|
||||
'\n * '.join(messages)
|
||||
)
|
||||
)
|
||||
|
||||
def conf(self, var):
|
||||
"""Find a configuration value for variable `var`, as transformed by `mapper`.
|
||||
|
||||
:param str var: The variable to get.
|
||||
:returns: The value of the variable.
|
||||
:rtype: str
|
||||
"""
|
||||
|
||||
return self._get(var)
|
||||
|
||||
def _has(self, var):
|
||||
return self.mapper(var) in self.confobj
|
||||
|
||||
def _get(self, var):
|
||||
return self.confobj.get(self.mapper(var))
|
||||
|
||||
|
||||
def validate_file(filename):
|
||||
"""Ensure that the specified file exists."""
|
||||
|
||||
if not os.path.exists(filename):
|
||||
raise errors.PluginError('File not found: {0}'.format(filename))
|
||||
|
||||
if not os.path.isfile(filename):
|
||||
raise errors.PluginError('Path is not a file: {0}'.format(filename))
|
||||
|
||||
|
||||
def validate_file_permissions(filename):
|
||||
"""Ensure that the specified file exists and warn about unsafe permissions."""
|
||||
|
||||
validate_file(filename)
|
||||
|
||||
permissions = stat.S_IMODE(os.stat(filename).st_mode)
|
||||
if permissions & stat.S_IRWXO:
|
||||
logger.warning('Unsafe permissions on credentials configuration file: %s', filename)
|
||||
|
||||
|
||||
def base_domain_name_guesses(domain):
|
||||
"""Return a list of progressively less-specific domain names.
|
||||
|
||||
One of these will probably be the domain name known to the DNS provider.
|
||||
|
||||
:Example:
|
||||
|
||||
>>> base_domain_name_guesses('foo.bar.baz.example.com')
|
||||
['foo.bar.baz.example.com', 'bar.baz.example.com', 'baz.example.com', 'example.com', 'com']
|
||||
|
||||
:param str domain: The domain for which to return guesses.
|
||||
:returns: The a list of less specific domain names.
|
||||
:rtype: list
|
||||
"""
|
||||
|
||||
fragments = domain.split('.')
|
||||
return ['.'.join(fragments[i:]) for i in range(0, len(fragments))]
|
||||
@@ -0,0 +1,233 @@
|
||||
"""Tests for certbot.plugins.dns_common."""
|
||||
|
||||
import collections
|
||||
import logging
|
||||
import os
|
||||
import unittest
|
||||
|
||||
import mock
|
||||
|
||||
from certbot import errors
|
||||
from certbot.display import util as display_util
|
||||
from certbot.plugins import dns_common
|
||||
from certbot.plugins import dns_test_common
|
||||
from certbot.tests import util
|
||||
|
||||
|
||||
class DNSAuthenticatorTest(util.TempDirTestCase, dns_test_common.BaseAuthenticatorTest):
|
||||
# pylint: disable=protected-access
|
||||
|
||||
class _FakeDNSAuthenticator(dns_common.DNSAuthenticator):
|
||||
_setup_credentials = mock.MagicMock()
|
||||
_perform = mock.MagicMock()
|
||||
_cleanup = mock.MagicMock()
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
# pylint: disable=protected-access
|
||||
super(DNSAuthenticatorTest._FakeDNSAuthenticator, self).__init__(*args, **kwargs)
|
||||
|
||||
def more_info(self): # pylint: disable=missing-docstring,no-self-use
|
||||
return 'A fake authenticator for testing.'
|
||||
|
||||
class _FakeConfig(object):
|
||||
fake_propagation_seconds = 0
|
||||
fake_config_key = 1
|
||||
fake_other_key = None
|
||||
fake_file_path = None
|
||||
|
||||
def setUp(self):
|
||||
super(DNSAuthenticatorTest, self).setUp()
|
||||
|
||||
self.config = DNSAuthenticatorTest._FakeConfig()
|
||||
|
||||
self.auth = DNSAuthenticatorTest._FakeDNSAuthenticator(self.config, "fake")
|
||||
|
||||
def test_perform(self):
|
||||
self.auth.perform([self.achall])
|
||||
|
||||
self.auth._perform.assert_called_once_with(dns_test_common.DOMAIN, mock.ANY, mock.ANY)
|
||||
|
||||
def test_cleanup(self):
|
||||
self.auth._attempt_cleanup = True
|
||||
|
||||
self.auth.cleanup([self.achall])
|
||||
|
||||
self.auth._cleanup.assert_called_once_with(dns_test_common.DOMAIN, mock.ANY, mock.ANY)
|
||||
|
||||
@util.patch_get_utility()
|
||||
def test_prompt(self, mock_get_utility):
|
||||
mock_display = mock_get_utility()
|
||||
mock_display.input.side_effect = ((display_util.OK, "",),
|
||||
(display_util.OK, "value",))
|
||||
|
||||
self.auth._configure("other_key", "")
|
||||
self.assertEqual(self.auth.config.fake_other_key, "value")
|
||||
|
||||
@util.patch_get_utility()
|
||||
def test_prompt_canceled(self, mock_get_utility):
|
||||
mock_display = mock_get_utility()
|
||||
mock_display.input.side_effect = ((display_util.CANCEL, "c",),)
|
||||
|
||||
self.assertRaises(errors.PluginError, self.auth._configure, "other_key", "")
|
||||
|
||||
@util.patch_get_utility()
|
||||
def test_prompt_file(self, mock_get_utility):
|
||||
path = os.path.join(self.tempdir, 'file.ini')
|
||||
open(path, "wb").close()
|
||||
|
||||
mock_display = mock_get_utility()
|
||||
mock_display.directory_select.side_effect = ((display_util.OK, "",),
|
||||
(display_util.OK, "not-a-file.ini",),
|
||||
(display_util.OK, self.tempdir),
|
||||
(display_util.OK, path,))
|
||||
|
||||
self.auth._configure_file("file_path", "")
|
||||
self.assertEqual(self.auth.config.fake_file_path, path)
|
||||
|
||||
@util.patch_get_utility()
|
||||
def test_prompt_file_canceled(self, mock_get_utility):
|
||||
mock_display = mock_get_utility()
|
||||
mock_display.directory_select.side_effect = ((display_util.CANCEL, "c",),)
|
||||
|
||||
self.assertRaises(errors.PluginError, self.auth._configure_file, "file_path", "")
|
||||
|
||||
def test_configure_credentials(self):
|
||||
path = os.path.join(self.tempdir, 'file.ini')
|
||||
dns_test_common.write({"fake_test": "value"}, path)
|
||||
setattr(self.config, "fake_credentials", path)
|
||||
|
||||
credentials = self.auth._configure_credentials("credentials", "", {"test": ""})
|
||||
|
||||
self.assertEqual(credentials.conf("test"), "value")
|
||||
|
||||
@util.patch_get_utility()
|
||||
def test_prompt_credentials(self, mock_get_utility):
|
||||
bad_path = os.path.join(self.tempdir, 'bad-file.ini')
|
||||
dns_test_common.write({"fake_other": "other_value"}, bad_path)
|
||||
|
||||
path = os.path.join(self.tempdir, 'file.ini')
|
||||
dns_test_common.write({"fake_test": "value"}, path)
|
||||
setattr(self.config, "fake_credentials", "")
|
||||
|
||||
mock_display = mock_get_utility()
|
||||
mock_display.directory_select.side_effect = ((display_util.OK, "",),
|
||||
(display_util.OK, "not-a-file.ini",),
|
||||
(display_util.OK, self.tempdir),
|
||||
(display_util.OK, bad_path),
|
||||
(display_util.OK, path,))
|
||||
|
||||
credentials = self.auth._configure_credentials("credentials", "", {"test": ""})
|
||||
self.assertEqual(credentials.conf("test"), "value")
|
||||
|
||||
|
||||
class CredentialsConfigurationTest(util.TempDirTestCase):
|
||||
class _MockLoggingHandler(logging.Handler):
|
||||
messages = None
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
self.reset()
|
||||
logging.Handler.__init__(self, *args, **kwargs)
|
||||
|
||||
def emit(self, record):
|
||||
self.messages[record.levelname.lower()].append(record.getMessage())
|
||||
|
||||
def reset(self):
|
||||
"""Allows the handler to be reset between tests."""
|
||||
self.messages = collections.defaultdict(list)
|
||||
|
||||
def test_valid_file(self):
|
||||
path = os.path.join(self.tempdir, 'too-permissive-file.ini')
|
||||
|
||||
dns_test_common.write({"test": "value", "other": 1}, path)
|
||||
|
||||
credentials_configuration = dns_common.CredentialsConfiguration(path)
|
||||
self.assertEqual("value", credentials_configuration.conf("test"))
|
||||
self.assertEqual("1", credentials_configuration.conf("other"))
|
||||
|
||||
def test_nonexistent_file(self):
|
||||
path = os.path.join(self.tempdir, 'not-a-file.ini')
|
||||
|
||||
self.assertRaises(errors.PluginError, dns_common.CredentialsConfiguration, path)
|
||||
|
||||
def test_valid_file_with_unsafe_permissions(self):
|
||||
log = self._MockLoggingHandler()
|
||||
dns_common.logger.addHandler(log)
|
||||
|
||||
path = os.path.join(self.tempdir, 'too-permissive-file.ini')
|
||||
open(path, "wb").close()
|
||||
|
||||
dns_common.CredentialsConfiguration(path)
|
||||
|
||||
self.assertEqual(1, len([_ for _ in log.messages['warning'] if _.startswith("Unsafe")]))
|
||||
|
||||
|
||||
class CredentialsConfigurationRequireTest(util.TempDirTestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(CredentialsConfigurationRequireTest, self).setUp()
|
||||
|
||||
self.path = os.path.join(self.tempdir, 'file.ini')
|
||||
|
||||
def _write(self, values):
|
||||
dns_test_common.write(values, self.path)
|
||||
|
||||
def test_valid(self):
|
||||
self._write({"test": "value", "other": 1})
|
||||
|
||||
credentials_configuration = dns_common.CredentialsConfiguration(self.path)
|
||||
credentials_configuration.require({"test": "", "other": ""})
|
||||
|
||||
def test_valid_but_extra(self):
|
||||
self._write({"test": "value", "other": 1})
|
||||
|
||||
credentials_configuration = dns_common.CredentialsConfiguration(self.path)
|
||||
credentials_configuration.require({"test": ""})
|
||||
|
||||
def test_valid_empty(self):
|
||||
self._write({})
|
||||
|
||||
credentials_configuration = dns_common.CredentialsConfiguration(self.path)
|
||||
credentials_configuration.require({})
|
||||
|
||||
def test_missing(self):
|
||||
self._write({})
|
||||
|
||||
credentials_configuration = dns_common.CredentialsConfiguration(self.path)
|
||||
self.assertRaises(errors.PluginError, credentials_configuration.require, {"test": ""})
|
||||
|
||||
def test_blank(self):
|
||||
self._write({"test": ""})
|
||||
|
||||
credentials_configuration = dns_common.CredentialsConfiguration(self.path)
|
||||
self.assertRaises(errors.PluginError, credentials_configuration.require, {"test": ""})
|
||||
|
||||
def test_typo(self):
|
||||
self._write({"tets": "typo!"})
|
||||
|
||||
credentials_configuration = dns_common.CredentialsConfiguration(self.path)
|
||||
self.assertRaises(errors.PluginError, credentials_configuration.require, {"test": ""})
|
||||
|
||||
|
||||
class DomainNameGuessTest(unittest.TestCase):
|
||||
|
||||
def test_simple_case(self):
|
||||
self.assertTrue(
|
||||
'example.com' in
|
||||
dns_common.base_domain_name_guesses("example.com")
|
||||
)
|
||||
|
||||
def test_sub_domain(self):
|
||||
self.assertTrue(
|
||||
'example.com' in
|
||||
dns_common.base_domain_name_guesses("foo.bar.baz.example.com")
|
||||
)
|
||||
|
||||
def test_second_level_domain(self):
|
||||
self.assertTrue(
|
||||
'example.co.uk' in
|
||||
dns_common.base_domain_name_guesses("foo.bar.baz.example.co.uk")
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main() # pragma: no cover
|
||||
@@ -0,0 +1,63 @@
|
||||
"""Base test class for DNS authenticators."""
|
||||
|
||||
import os
|
||||
|
||||
import configobj
|
||||
import mock
|
||||
import six
|
||||
from acme import challenges
|
||||
from acme import jose
|
||||
|
||||
from certbot import achallenges
|
||||
from certbot.tests import acme_util
|
||||
from certbot.tests import util as test_util
|
||||
|
||||
DOMAIN = 'example.com'
|
||||
KEY = jose.JWKRSA.load(test_util.load_vector("rsa512_key.pem"))
|
||||
|
||||
|
||||
class BaseAuthenticatorTest(object):
|
||||
"""
|
||||
A base test class to reduce duplication between test code for DNS Authenticator Plugins.
|
||||
|
||||
Assumes:
|
||||
* That subclasses also subclass unittest.TestCase
|
||||
* That the authenticator is stored as self.auth
|
||||
"""
|
||||
|
||||
achall = achallenges.KeyAuthorizationAnnotatedChallenge(
|
||||
challb=acme_util.DNS01, domain=DOMAIN, account_key=KEY)
|
||||
|
||||
def test_more_info(self):
|
||||
# pylint: disable=no-member
|
||||
self.assertTrue(isinstance(self.auth.more_info(), six.string_types))
|
||||
|
||||
def test_get_chall_pref(self):
|
||||
# pylint: disable=no-member
|
||||
self.assertEqual(self.auth.get_chall_pref(None), [challenges.DNS01])
|
||||
|
||||
def test_parser_arguments(self):
|
||||
m = mock.MagicMock()
|
||||
|
||||
# pylint: disable=no-member
|
||||
self.auth.add_parser_arguments(m)
|
||||
|
||||
m.assert_any_call('propagation-seconds', type=int, default=mock.ANY, help=mock.ANY)
|
||||
|
||||
|
||||
def write(values, path):
|
||||
"""Write the specified values to a config file.
|
||||
|
||||
:param dict values: A map of values to write.
|
||||
:param str path: Where to write the values.
|
||||
"""
|
||||
|
||||
config = configobj.ConfigObj()
|
||||
|
||||
for key in values:
|
||||
config[key] = values[key]
|
||||
|
||||
with open(path, "wb") as f:
|
||||
config.write(outfile=f)
|
||||
|
||||
os.chmod(path, 0o600)
|
||||
@@ -133,7 +133,7 @@ def choose_plugin(prepared, question):
|
||||
else:
|
||||
return None
|
||||
|
||||
noninstaller_plugins = ["webroot", "manual", "standalone"]
|
||||
noninstaller_plugins = ["webroot", "manual", "standalone", "dns-cloudflare"]
|
||||
|
||||
def record_chosen_plugins(config, plugins, auth, inst):
|
||||
"Update the config entries to reflect the plugins we actually selected."
|
||||
@@ -237,6 +237,8 @@ def cli_plugin_requests(config):
|
||||
req_auth = set_configurator(req_auth, "webroot")
|
||||
if config.manual:
|
||||
req_auth = set_configurator(req_auth, "manual")
|
||||
if config.dns_cloudflare:
|
||||
req_auth = set_configurator(req_auth, "dns-cloudflare")
|
||||
logger.debug("Requested authenticator %s and installer %s", req_auth, req_inst)
|
||||
return req_auth, req_inst
|
||||
|
||||
|
||||
Reference in New Issue
Block a user