mirror of
https://github.com/certbot/certbot.git
synced 2026-08-01 16:19:13 +02:00
Refactor Lexicon-based DNS plugins (#9746)
* Refactor Lexicon-based DNS plugins and upgrade minimal version of Lexicon * Relax filterwarning to comply with envs where boto3 is not installed * Update pinned dependencies * Use our previous method to deprecate part of modules * Safe import internally * Add changelog Co-authored-by: Brad Warren <bmw@users.noreply.github.com>
This commit is contained in:
Adrien Ferrand
co-authored by
Brad Warren
Brad Warren
parent
694c758db7
commit
732a3ac962
@@ -2,34 +2,40 @@
|
||||
import logging
|
||||
from typing import Any
|
||||
from typing import Callable
|
||||
from typing import cast
|
||||
from typing import Optional
|
||||
|
||||
from lexicon.providers import ovh
|
||||
from requests import HTTPError
|
||||
|
||||
from certbot import errors
|
||||
from certbot.plugins import dns_common
|
||||
from certbot.plugins import dns_common_lexicon
|
||||
from certbot.plugins.dns_common import CredentialsConfiguration
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
TOKEN_URL = 'https://eu.api.ovh.com/createToken/ or https://ca.api.ovh.com/createToken/'
|
||||
|
||||
|
||||
class Authenticator(dns_common.DNSAuthenticator):
|
||||
class Authenticator(dns_common_lexicon.LexiconDNSAuthenticator):
|
||||
"""DNS Authenticator for OVH
|
||||
|
||||
This Authenticator uses the OVH API to fulfill a dns-01 challenge.
|
||||
"""
|
||||
|
||||
description = 'Obtain certificates using a DNS TXT record (if you are using OVH for DNS).'
|
||||
ttl = 60
|
||||
|
||||
def __init__(self, *args: Any, **kwargs: Any) -> None:
|
||||
super().__init__(*args, **kwargs)
|
||||
self.credentials: Optional[CredentialsConfiguration] = None
|
||||
self._add_provider_option('endpoint',
|
||||
'OVH API endpoint (ovh-eu or ovh-ca)',
|
||||
'auth_entrypoint')
|
||||
self._add_provider_option('application-key',
|
||||
f'Application key for OVH API, obtained from {TOKEN_URL}',
|
||||
'auth_application_key')
|
||||
self._add_provider_option('application-secret',
|
||||
f'Application secret for OVH API, obtained from {TOKEN_URL}',
|
||||
'auth_application_secret')
|
||||
self._add_provider_option('consumer-key',
|
||||
f'Consumer key for OVH API, obtained from {TOKEN_URL}',
|
||||
'auth_consumer_key')
|
||||
|
||||
@classmethod
|
||||
def add_parser_arguments(cls, add: Callable[..., None],
|
||||
@@ -41,58 +47,9 @@ class Authenticator(dns_common.DNSAuthenticator):
|
||||
return 'This plugin configures a DNS TXT record to respond to a dns-01 challenge using ' + \
|
||||
'the OVH API.'
|
||||
|
||||
def _setup_credentials(self) -> None:
|
||||
self.credentials = self._configure_credentials(
|
||||
'credentials',
|
||||
'OVH credentials INI file',
|
||||
{
|
||||
'endpoint': 'OVH API endpoint (ovh-eu or ovh-ca)',
|
||||
'application-key': 'Application key for OVH API, obtained from {0}'
|
||||
.format(TOKEN_URL),
|
||||
'application-secret': 'Application secret for OVH API, obtained from {0}'
|
||||
.format(TOKEN_URL),
|
||||
'consumer-key': 'Consumer key for OVH API, obtained from {0}'
|
||||
.format(TOKEN_URL),
|
||||
}
|
||||
)
|
||||
|
||||
def _perform(self, domain: str, validation_name: str, validation: str) -> None:
|
||||
self._get_ovh_client().add_txt_record(domain, validation_name, validation)
|
||||
|
||||
def _cleanup(self, domain: str, validation_name: str, validation: str) -> None:
|
||||
self._get_ovh_client().del_txt_record(domain, validation_name, validation)
|
||||
|
||||
def _get_ovh_client(self) -> "_OVHLexiconClient":
|
||||
if not self.credentials: # pragma: no cover
|
||||
raise errors.Error("Plugin has not been prepared.")
|
||||
return _OVHLexiconClient(
|
||||
cast(str, self.credentials.conf('endpoint')),
|
||||
cast(str, self.credentials.conf('application-key')),
|
||||
cast(str, self.credentials.conf('application-secret')),
|
||||
cast(str, self.credentials.conf('consumer-key')),
|
||||
self.ttl
|
||||
)
|
||||
|
||||
|
||||
class _OVHLexiconClient(dns_common_lexicon.LexiconClient):
|
||||
"""
|
||||
Encapsulates all communication with the OVH API via Lexicon.
|
||||
"""
|
||||
|
||||
def __init__(self, endpoint: str, application_key: str, application_secret: str,
|
||||
consumer_key: str, ttl: int) -> None:
|
||||
super().__init__()
|
||||
|
||||
config = dns_common_lexicon.build_lexicon_config('ovh', {
|
||||
'ttl': ttl,
|
||||
}, {
|
||||
'auth_entrypoint': endpoint,
|
||||
'auth_application_key': application_key,
|
||||
'auth_application_secret': application_secret,
|
||||
'auth_consumer_key': consumer_key,
|
||||
})
|
||||
|
||||
self.provider = ovh.Provider(config)
|
||||
@property
|
||||
def _provider_name(self) -> str:
|
||||
return 'ovh'
|
||||
|
||||
def _handle_http_error(self, e: HTTPError, domain_name: str) -> errors.PluginError:
|
||||
hint = None
|
||||
|
||||
@@ -1,8 +1,6 @@
|
||||
"""Tests for certbot_dns_ovh._internal.dns_ovh."""
|
||||
|
||||
import sys
|
||||
import unittest
|
||||
from unittest import mock
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
from requests.exceptions import HTTPError
|
||||
@@ -19,7 +17,10 @@ CONSUMER_KEY = 'spam'
|
||||
|
||||
|
||||
class AuthenticatorTest(test_util.TempDirTestCase,
|
||||
dns_test_common_lexicon.BaseLexiconAuthenticatorTest):
|
||||
dns_test_common_lexicon.BaseLexiconDNSAuthenticatorTest):
|
||||
|
||||
DOMAIN_NOT_FOUND = Exception('Domain example.com not found')
|
||||
LOGIN_ERROR = HTTPError('403 Client Error: Forbidden for url: https://eu.api.ovh.com/1.0/...')
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
@@ -38,26 +39,7 @@ class AuthenticatorTest(test_util.TempDirTestCase,
|
||||
self.config = mock.MagicMock(ovh_credentials=path,
|
||||
ovh_propagation_seconds=0) # don't wait during tests
|
||||
|
||||
self.auth = Authenticator(self.config, "ovh")
|
||||
|
||||
self.mock_client = mock.MagicMock()
|
||||
# _get_ovh_client | pylint: disable=protected-access
|
||||
self.auth._get_ovh_client = mock.MagicMock(return_value=self.mock_client)
|
||||
|
||||
|
||||
class OVHLexiconClientTest(unittest.TestCase, dns_test_common_lexicon.BaseLexiconClientTest):
|
||||
DOMAIN_NOT_FOUND = Exception('Domain example.com not found')
|
||||
LOGIN_ERROR = HTTPError('403 Client Error: Forbidden for url: https://eu.api.ovh.com/1.0/...')
|
||||
|
||||
def setUp(self):
|
||||
from certbot_dns_ovh._internal.dns_ovh import _OVHLexiconClient
|
||||
|
||||
self.client = _OVHLexiconClient(
|
||||
ENDPOINT, APPLICATION_KEY, APPLICATION_SECRET, CONSUMER_KEY, 0
|
||||
)
|
||||
|
||||
self.provider_mock = mock.MagicMock()
|
||||
self.client.provider = self.provider_mock
|
||||
self.auth = Authenticator(self.config, 'ovh')
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -7,7 +7,7 @@ from setuptools import setup
|
||||
version = '2.7.0.dev0'
|
||||
|
||||
install_requires = [
|
||||
'dns-lexicon>=3.2.1',
|
||||
'dns-lexicon>=3.14.1',
|
||||
'setuptools>=41.6.0',
|
||||
]
|
||||
|
||||
|
||||
Reference in New Issue
Block a user