PowerDnsAcmeClient

CHANGELOG.md

@@ -15,6 +15,8 @@ To see unreleased changes, please see the [CHANGELOG on the master branch](https
 ## Added
 
 * DAVACMEClient: http-01 fulfillment using WebDAV
+* PowerDnsAcmeClient: dns-01 fulfillment using PowerDNS.
+* WEBACMEClient: http-01 fulfillment using static files.
 
 ## Changed
 

README.md

@@ -25,7 +25,8 @@ Gufo ACME contains various clients which can be applied to your tasks:
 
 * ACMEClient - base client to implement any fulfillment functionality
     by creating subclasses.
-* DAVACMEClient - http-01 fulfillment using WebDAV methods.    
+* DAVACMEClient - http-01 fulfillment using WebDAV methods.
+* PowerDnsAcmeClient - dns-01 PowerDNS fulfillment.
 * WebACMEClient - http-01 static file fulfillment.
 
 ## Supported Certificate Authorities
@@ -95,6 +96,7 @@ async with SignACMEClient.from_state(state) as client:
 * Fully typed.
 * Clean API.
 * Robust well-tested code.
+* Batteries included.
 * 99%+ test coverage.
 
 ## On Gufo Stack

docs/index.md

@@ -23,6 +23,7 @@ Gufo ACME contains various clients which can be applied to your tasks:
 * [ACMEClient][gufo.acme.clients.base.ACMEClient] - base client to implement any fulfillment functionality
     by creating subclasses.
 * [DAVACMEClient][gufo.acme.clients.dav.DAVACMEClient] - http-01 fulfillment using WebDAV methods.
+* [PowerDnsAcmeClient][gufo.acme.clients.powerdns.PowerDnsAcmeClient] - dns-01 PowerDNS fulfillment.
 * [WebACMEClient][gufo.acme.clients.web.WebACMEClient] - http-01 static file fulfillment.
 
 ## Supported Certificate Authorities

docs/testing.md

@@ -78,6 +78,8 @@ to run tests in the CI environment. On local environments, the test is skipped b
 To enable the test in your local environment, additional
 infrastructure is needed.
 
+### DAVACMEClient
+
 1. Have control over a DNS zone (later `<mydomain>`).
 2. Set up an Nginx server.
 
@@ -158,9 +160,41 @@ commands in your development environment:
 
 ```
 export CI_ACME_TEST_DOMAIN=acme-ci.<domain>
-export CI_ACME_TEST_USER=<user>
-export CI_ACME_TEST_PASS=<password>
+export CI_DAV_TEST_DOMAIN=acme-ci.<domain>
+export CI_DAV_TEST_USER=<user>
+export CI_DAV_TEST_PASSWORD=<password>
+```
+
+### PowerDnsAcmeClient
+
+We're considering:
+
+* We're perform testig on csr-proxy-test.<domain>
+* Your PowerDNS server's name is pdns.<domain>
+
+First, in zone `<domain>` create a glue record pointing to your PowerDNS server:
+
+```
+csr-proxy-test IN IS pdns.<domain>
+```
+
+Create `csr-proxy-test.<domain>` zone:
+
 ```
+pdnsutil create-zone csr-proxy-test.gufolabs.com
+```
+
+Your environment is now ready. Before running the test suite, execute the following
+commands in your development environment:
+
+```
+export CI_POWERDNS_TEST_DOMAIN=csr-proxy-test.<domain>
+export CI_POWERDNS_TEST_API_URL=https://<power-dns-url>
+export CI_POWERDNS_TEST_API_KEY=<api key>
+
+```
+
+
 
 ## Python Test Code Coverage Check
 

src/gufo/acme/clients/__init__.py

@@ -11,5 +11,6 @@
 
 * [base][gufo.acme.clients.base] - Base class
 * [dav][gufo.acme.clients.dav] - http-01 WebDAV fulfillment.
+* [powerdns][gufo.acme.clients.powerdns] - dns-01 PowerDNS fulfillment.
 * [web][gufo.acme.clients.web] - http-01 static file fulfillment.
 """

src/gufo/acme/clients/dav.py

@@ -29,6 +29,10 @@ class DAVACMEClient(ACMEClient):
     with basic authorization.
     Works either with WebDAV modules
     or with custom scripts.
+
+    Args:
+        username: DAV user name.
+        password: DAV password.
     """
 
     def __init__(

src/gufo/acme/clients/powerdns.py

@@ -0,0 +1,118 @@
+# ---------------------------------------------------------------------
+# Gufo ACME: PowerDnsAcmeClient implementation
+# ---------------------------------------------------------------------
+# Copyright (C) 2023, Gufo Labs
+# ---------------------------------------------------------------------
+"""A PowerDnsAcmeClient implementation."""
+
+# Python modules
+import hashlib
+from typing import Any
+
+import httpx
+
+# Third-party modules
+from josepy.json_util import encode_b64jose
+
+from ..error import ACMEFulfillmentFailed
+
+# Gufo ACME modules
+from ..log import logger
+from ..types import ACMEChallenge
+from .base import ACMEClient
+
+RESP_NO_CONTENT = 204
+
+
+class PowerDnsAcmeClient(ACMEClient):
+    """
+    PowerDNS compatible ACME Client.
+
+    Fulfills dns-01 challenge by manipulating
+    DNS RR via PowerDNS API.
+
+    Args:
+        api_url: Root url of the PowerDNS web.
+        api_key: PowerDNS API key.
+    """
+
+    def __init__(
+        self: "PowerDnsAcmeClient",
+        directory_url: str,
+        *,
+        api_url: str,
+        api_key: str,
+        **kwargs: Any,
+    ) -> None:
+        super().__init__(directory_url, **kwargs)
+        self.api_url = self._normalize_url(api_url)
+        self.api_key = api_key
+
+    @staticmethod
+    def _normalize_url(url: str) -> str:
+        if url.endswith("/"):
+            return url[:-1]
+        return url
+
+    @staticmethod
+    def _check_api_response(resp: httpx.Response) -> None:
+        if resp.status_code != RESP_NO_CONTENT:
+            msg = f"Failed to fulfill: Server returned {resp}"
+            logger.error(msg)
+            raise ACMEFulfillmentFailed(msg)
+
+    async def fulfill_dns_01(
+        self: "PowerDnsAcmeClient", domain: str, challenge: ACMEChallenge
+    ) -> bool:
+        """
+        Fulfill dns-01 challenge.
+
+        Update token via PowerDNS API.
+
+        Args:
+            domain: Domain name
+            challenge: ACMEChallenge instance, containing token.
+
+        Returns:
+            True - on succeess.
+
+        Raises:
+            ACMEFulfillmentFailed: On error.
+        """
+        # Calculate value
+        v = encode_b64jose(
+            hashlib.sha256(self.get_key_authorization(challenge)).digest()
+        )
+        # Set PDNS challenge
+        async with self._get_client() as client:
+            # Construct the API endpoint for updating a record in a specific zone
+            endpoint = (
+                f"{self.api_url}/api/v1/servers/localhost/zones/{domain}"
+            )
+            # Set up the headers, including the API key for authentication
+            headers = {
+                "X-API-Key": self.api_key,
+                "Content-Type": "application/json",
+            }
+            # Prepare the payload for the update
+            update_payload = {
+                "rrsets": [
+                    {
+                        "name": f"_acme-challenge.{domain}.",
+                        "type": "TXT",
+                        "ttl": 1,
+                        "changetype": "REPLACE",
+                        "records": [
+                            {
+                                "content": f'"{v}"',
+                                "disabled": False,
+                            }
+                        ],
+                    }
+                ]
+            }
+            resp = await client.patch(
+                endpoint, json=update_payload, headers=headers
+            )
+            self._check_api_response(resp)
+            return True

tests/clients/test_powerdns.py

@@ -0,0 +1,86 @@
+# ---------------------------------------------------------------------
+# CSR Proxy: PowerDnsAcmeClient client tests
+# ---------------------------------------------------------------------
+# Copyright (C) 2023, Gufo Labs
+# ---------------------------------------------------------------------
+
+# Python modules
+import asyncio
+import os
+
+# Third-party modules
+import httpx
+import pytest
+
+# Gufo ACME modules
+from gufo.acme.clients.base import ACMEClient
+from gufo.acme.clients.powerdns import PowerDnsAcmeClient
+from gufo.acme.error import ACMEFulfillmentFailed
+
+from .utils import DIRECTORY, EMAIL, get_csr_pem, not_set, not_set_reason
+
+ENV_CI_POWERDNS_TEST_DOMAIN = "CI_POWERDNS_TEST_DOMAIN"
+ENV_CI_POWERDNS_TEST_API_URL = "CI_POWERDNS_TEST_API_URL"
+ENV_CI_POWERDNS_TEST_API_KEY = "CI_POWERDNS_TEST_API_KEY"
+SCENARIO_ENV = [
+    ENV_CI_POWERDNS_TEST_DOMAIN,
+    ENV_CI_POWERDNS_TEST_API_URL,
+    ENV_CI_POWERDNS_TEST_API_KEY,
+]
+
+
+@pytest.mark.skipif(not_set(SCENARIO_ENV), reason=not_set_reason(SCENARIO_ENV))
+def test_sign():
+    async def inner():
+        csr_pem = get_csr_pem(domain)
+        #
+        pk = PowerDnsAcmeClient.get_key()
+        async with PowerDnsAcmeClient(
+            DIRECTORY,
+            api_url=os.getenv(ENV_CI_POWERDNS_TEST_API_URL),
+            api_key=os.getenv(ENV_CI_POWERDNS_TEST_API_KEY),
+            key=pk,
+        ) as client:
+            # Register account
+            uri = await client.new_account(EMAIL)
+            assert uri
+            # Create new order
+            cert = await client.sign(domain, csr_pem)
+            # Deactivate account
+            await client.deactivate_account()
+        assert cert
+        assert b"BEGIN CERTIFICATE" in cert
+        assert b"END CERTIFICATE" in cert
+
+    domain = os.getenv(ENV_CI_POWERDNS_TEST_DOMAIN) or ""
+    asyncio.run(inner())
+
+
+def test_state():
+    client = ACMEClient(
+        DIRECTORY,
+        key=PowerDnsAcmeClient.get_key(),
+    )
+    state = client.get_state()
+    client2 = PowerDnsAcmeClient.from_state(
+        state, api_url="https://127.0.0.1/", api_key="xxx"
+    )
+    assert isinstance(client2, PowerDnsAcmeClient)
+
+
+@pytest.mark.parametrize(
+    ("url", "expected"),
+    [
+        ("https://example.com", "https://example.com"),
+        ("https://example.com/", "https://example.com"),
+    ],
+)
+def test_normalize_url(url: str, expected: str) -> None:
+    r = PowerDnsAcmeClient._normalize_url(url)
+    assert r == expected
+
+
+def test_invalid_response():
+    resp = httpx.Response(200)
+    with pytest.raises(ACMEFulfillmentFailed):
+        PowerDnsAcmeClient._check_api_response(resp)