Cloudflare: DNS-01 Delegation Zone

certbot-dns-cloudflare/src/certbot_dns_cloudflare/__init__.py

@@ -18,6 +18,9 @@ Named Arguments
                                           to propagate before asking the ACME
                                           server to verify the DNS record.
                                           (Default: 10)
+``--dns-cloudflare-delegate-via``         The domain of the target zone, where
+                                          existing CNAME delegation records
+                                          point to
 ========================================  =====================================
 
 
@@ -121,4 +124,17 @@ Examples
      --dns-cloudflare-propagation-seconds 60 \\
      -d example.com
 
+.. code-block:: bash
+   :caption: To acquire a certificate for ``example.com``, using Cloudflare
+             credentials with write access for the CNAME delegation zone
+             ``acme-delegation.org``
+
+    # Assuming the following CNAME record has already been set:
+    # _acme-challenge.example.com. <TTL> IN CNAME _acme-challenge.example.com.acme-delegation.org.
+   certbot certonly \\
+     --dns-cloudflare \\
+     --dns-cloudflare-credentials ~/.secrets/certbot/cloudflare.ini \\
+     --dns-cloudflare-delegate-via acme-delegation.org \\
+     -d example.com
+
 """

certbot-dns-cloudflare/src/certbot_dns_cloudflare/_internal/dns_cloudflare.py

@@ -29,12 +29,14 @@ class Authenticator(dns_common.DNSAuthenticator):
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         super().__init__(*args, **kwargs)
         self.credentials: Optional[CredentialsConfiguration] = None
+        self.delegate_zone: Optional[str] = None
 
     @classmethod
     def add_parser_arguments(cls, add: Callable[..., None],
                              default_propagation_seconds: int = 10) -> None:
         super().add_parser_arguments(add, default_propagation_seconds)
         add('credentials', help='Cloudflare credentials INI file.')
+        add('delegate-via', help='The domain (zone) of the CNAME delegation')
 
     def more_info(self) -> str:
         return 'This plugin configures a DNS TXT record to respond to a dns-01 challenge using ' + \
@@ -70,12 +72,15 @@ class Authenticator(dns_common.DNSAuthenticator):
             None,
             self._validate_credentials
         )
+        self.delegate_zone = self.conf('delegate-via')
 
     def _perform(self, domain: str, validation_name: str, validation: str) -> None:
-        self._get_cloudflare_client().add_txt_record(domain, validation_name, validation, self.ttl)
+        zone = self.delegate_zone if self.delegate_zone else domain
+        self._get_cloudflare_client().add_txt_record(zone, validation_name, validation, self.ttl)
 
     def _cleanup(self, domain: str, validation_name: str, validation: str) -> None:
-        self._get_cloudflare_client().del_txt_record(domain, validation_name, validation)
+        zone = self.delegate_zone if self.delegate_zone else domain
+        self._get_cloudflare_client().del_txt_record(zone, validation_name, validation)
 
     def _get_cloudflare_client(self) -> "_CloudflareClient":
         if not self.credentials:  # pragma: no cover

certbot-dns-cloudflare/src/certbot_dns_cloudflare/_internal/tests/dns_cloudflare_test.py

@@ -32,7 +32,8 @@ class AuthenticatorTest(test_util.TempDirTestCase, dns_test_common.BaseAuthentic
         dns_test_common.write({"cloudflare_email": EMAIL, "cloudflare_api_key": API_KEY}, path)
 
         self.config = mock.MagicMock(cloudflare_credentials=path,
-                                     cloudflare_propagation_seconds=0)  # don't wait during tests
+                                     cloudflare_propagation_seconds=0,  # don't wait during tests
+                                     cloudflare_delegate_via=None)
 
         self.auth = Authenticator(self.config, "cloudflare")
 
@@ -96,6 +97,50 @@ class AuthenticatorTest(test_util.TempDirTestCase, dns_test_common.BaseAuthentic
         with pytest.raises(errors.PluginError):
             self.auth.perform([self.achall])
 
+    @test_util.patch_display_util()
+    def test_delegation_single_domain(self, unused_mock_get_utility):
+        self.config.cloudflare_delegate_via = 'acme-zone.org'
+        self.auth.perform([self.achall])
+        expected = [mock.call.add_txt_record('acme-zone.org', '_acme-challenge.'+DOMAIN, mock.ANY, mock.ANY)]
+        assert expected == self.mock_client.mock_calls
+
+    @test_util.patch_display_util()
+    def test_delegation_multiple_domains(self, unused_mock_get_utility):
+        from certbot import achallenges
+        from certbot.tests import acme_util
+        from certbot.plugins.dns_test_common import KEY
+        # Create second challenge for different domain
+        achall2 = achallenges.KeyAuthorizationAnnotatedChallenge(
+            challb=acme_util.DNS01_P_2, domain='second-domain.com', account_key=KEY)
+        self.config.cloudflare_delegate_via = 'acme-zone.org'
+        self.auth.perform([self.achall, achall2])
+        expected = [
+            mock.call.add_txt_record('acme-zone.org', '_acme-challenge.'+DOMAIN, mock.ANY, mock.ANY),
+            mock.call.add_txt_record('acme-zone.org', '_acme-challenge.second-domain.com', mock.ANY, mock.ANY)
+        ]
+        assert expected == self.mock_client.mock_calls
+
+    @test_util.patch_display_util()
+    def test_delegation_wildcard(self, unused_mock_get_utility):
+        from certbot import achallenges
+        from certbot.tests import acme_util
+        from certbot.plugins.dns_test_common import KEY
+        wildcard_achall = achallenges.KeyAuthorizationAnnotatedChallenge(
+            challb=acme_util.DNS01_P, domain='*.'+DOMAIN, account_key=KEY)
+        self.config.cloudflare_delegate_via = 'acme-zone.org'
+        self.auth.perform([wildcard_achall])
+        # Wildcard domain creates validation name with *.  - delegation zone is still used
+        expected = [mock.call.add_txt_record('acme-zone.org', '_acme-challenge.*.'+DOMAIN, mock.ANY, mock.ANY)]
+        assert expected == self.mock_client.mock_calls
+
+    def test_cleanup_with_delegation(self):
+        # _attempt_cleanup | pylint: disable=protected-access
+        self.auth._attempt_cleanup = True
+        # delegate_zone | pylint: disable=protected-access
+        self.auth.delegate_zone = 'acme-zone.org'
+        self.auth.cleanup([self.achall])
+        expected = [mock.call.del_txt_record('acme-zone.org', '_acme-challenge.'+DOMAIN, mock.ANY)]
+        assert expected == self.mock_client.mock_calls
 
 class CloudflareClientTest(unittest.TestCase):
     record_name = "foo"