DAVACMEClient

CHANGELOG.md

@@ -12,6 +12,10 @@ To see unreleased changes, please see the [CHANGELOG on the master branch](https
 
 ## [Unreleased]
 
+## Added
+
+* DAVACMEClient: http-01 fulfillment using WebDAV
+
 ## Changed
 
 * ACMEClient has been moved into `gufo.acme.clients.base`.

README.md

@@ -21,6 +21,17 @@ for automated certificate signing, now widely used by services
 such as Let's Encrypt. Gufo ACME is a Python asyncio ACME client library that
 simplifies the protocol complexity with a straightforward and robust API.
 
+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.    
+
+## Supported Certificate Authorities
+
+* [Letsencrypt](https://letsencrypt.org)
+* Any [RFC-8555](https://tools.ietf.org/html/rfc8555) compatible CA.
+
 ## Examples
 
 ### Account Creation
@@ -83,7 +94,7 @@ async with SignACMEClient.from_state(state) as client:
 * Fully typed.
 * Clean API.
 * Robust well-tested code.
-* 97%+ test coverage.
+* 99%+ test coverage.
 
 ## On Gufo Stack
 

docs/index.md

@@ -18,6 +18,17 @@ for automated certificate signing, now widely used by services
 such as Let's Encrypt. Gufo ACME is a Python asyncio ACME client library that
 simplifies the protocol complexity with a straightforward and robust API.
 
+Gufo ACME contains various clients which can be applied to your tasks:
+
+* [ACMEClient](reference/gufo/acme/clients/base/#gufo.acme.clients.base.ACMEClient) - base client to implement any fulfillment functionality
+    by creating subclasses.
+* [DAVACMEClient](reference/gufo/acme/clients/dav/#gufo.acme.clients.dav.DAVACMEClient) - http-01 fulfillment using WebDAV methods.    
+
+## Supported Certificate Authorities
+
+* [Letsencrypt](https://letsencrypt.org)
+* Any [RFC-8555](https://tools.ietf.org/html/rfc8555) compatible CA.
+
 ## Examples
 
 ### Account Creation
@@ -80,7 +91,7 @@ async with SignACMEClient.from_state(state) as client:
 * Fully typed.
 * Clean API.
 * Robust well-tested code.
-* 97+% test coverage.
+* 99+% test coverage.
 
 ## On Gufo Stack
 

pyproject.toml

@@ -59,7 +59,7 @@ select = [
 ]
 # Always autofix, but never try to fix `F401` (unused imports).
 fix = true
-ignore = ["D203", "D212", "D107", "A002", "A003", "PLR0911", "PLR0913", "RUF012", "S603"]
+ignore = ["ANN401", "D203", "D212", "D107", "A002", "A003", "PLR0911", "PLR0913", "RUF012", "S603"]
 unfixable = ["F401"]
 
 [tool.ruff.flake8-quotes]
@@ -101,6 +101,8 @@ max-complexity = 12
   "S603", # `subprocess` call: check for execution of untrusted input
   "PLR2004", # Magic value used in comparison, consider replacing {value} with a constant variable
   "PT011", # `pytest.raises(ValueError)` is too broad, set the `match` parameter or use a more specific exception
+  "S105", # Possible hardcoded password assigned to: "..."
+  "S106", # Possible hardcoded password assigned to argument: "..."  
 ]
 
 [project]

src/gufo/acme/clients/__init__.py

@@ -9,5 +9,6 @@
 
 The following clients are provided out-of-the-box:
 
-* [base
+* [base][gufo.acme.clients.base] - Base class
+* [dav][gufo.acme.clients.dav] - http-01 WebDAV fulfillment.
 """

src/gufo/acme/clients/base.py

@@ -112,21 +112,22 @@ async def fulfill_http_01(
     NONCE_HEADER: str = "Replay-Nonce"
     RETRY_AFTER_HEADER: str = "Retry-After"
     DEFAULT_TIMEOUT: float = 40.0
+    DEFAULT_SIGNATURE = RS256
 
     def __init__(
         self: "ACMEClient",
         directory_url: str,
         *,
         key: JWK,
-        alg: JWASignature = RS256,
+        alg: Optional[JWASignature] = None,
         account_url: Optional[str] = None,
         timeout: Optional[float] = None,
         user_agent: Optional[str] = None,
     ) -> None:
         self._directory_url = directory_url
         self._directory: Optional[ACMEDirectory] = None
         self._key = key
-        self._alg = alg
+        self._alg = alg or self.DEFAULT_SIGNATURE
         self._account_url = account_url
         self._nonces: Set[bytes] = set()
         self._timeout = timeout or self.DEFAULT_TIMEOUT
@@ -556,6 +557,27 @@ def _pem_to_der(pem: bytes) -> bytes:
         csr = x509.load_pem_x509_csr(pem)
         return csr.public_bytes(encoding=Encoding.DER)
 
+    @staticmethod
+    def _get_order_status(resp: httpx.Response) -> str:
+        """
+        Check order response.
+
+        Args:
+            resp: Order response
+
+        Returns:
+            Order status
+
+        Raises:
+            ACMECertificateError: if status is invalid
+        """
+        data = resp.json()
+        status = cast(str, data["status"])
+        if status == "invalid":
+            msg = "Failed to finalize order"
+            raise ACMECertificateError(msg)
+        return status
+
     async def finalize_and_wait(
         self: "ACMEClient", order: ACMEOrder, *, csr: bytes
     ) -> bytes:
@@ -578,24 +600,17 @@ async def finalize_and_wait(
         resp = await self._post(
             order.finalize, {"csr": encode_b64jose(self._pem_to_der(csr))}
         )
-        data = resp.json()
-        status = data["status"]
-        if status == "invalid":
-            msg = "Failed to finalize order"
-            raise ACMECertificateError(msg)
+        self._get_order_status(resp)
         order_uri = resp.headers["Location"]
         # Poll for certificate
         await self._random_delay(1.0)
         while True:
             logger.warning("Polling order")
             resp = await self._post(order_uri, None)
-            data = resp.json()
-            status = data["status"]
-            if status == "invalid":
-                msg = "Failed to finalize order"
-                raise ACMECertificateError(msg)
+            status = self._get_order_status(resp)
             if status == "valid":
                 logger.warning("Order is ready. Downloading certificate")
+                data = resp.json()
                 resp = await self._post(data["certificate"], None)
                 return resp.text.encode()
 
@@ -1148,7 +1163,7 @@ def get_state(self: "ACMEClient") -> bytes:
         return json.dumps(state, indent=2).encode()
 
     @classmethod
-    def from_state(cls: Type[CT], state: bytes) -> CT:
+    def from_state(cls: Type[CT], state: bytes, **kwargs: Any) -> CT:
         """
         Restore ACMEClient from the state.
 
@@ -1158,6 +1173,7 @@ def from_state(cls: Type[CT], state: bytes) -> CT:
 
         Args:
             state: Stored state.
+            kwargs: An additional arguments to be passed to constructor.
 
         Returns:
             New ACMEClient instance.
@@ -1167,4 +1183,5 @@ def from_state(cls: Type[CT], state: bytes) -> CT:
             s["directory"],
             key=JWKRSA.fields_from_json(s["key"]),
             account_url=s.get("account_url"),
+            **kwargs,
         )

src/gufo/acme/clients/dav.py

@@ -0,0 +1,116 @@
+# ---------------------------------------------------------------------
+# Gufo ACME: ACMEClient implementation
+# ---------------------------------------------------------------------
+# Copyright (C) 2023, Gufo Labs
+# ---------------------------------------------------------------------
+"""An DAVACMEClient implementation."""
+
+# Python modules
+from typing import Any
+
+# Third-party modules
+import httpx
+
+from ..error import ACMEFulfillmentFailed
+
+# Gufo ACME modules
+from ..types import ACMEChallenge
+from .base import ACMEClient
+
+HTTP_MAX_VALID = 299
+
+
+class DAVACMEClient(ACMEClient):
+    """
+    WebDAV-compatible ACME Client.
+
+    Fulfills http-01 challenge by uploading
+    a token using HTTP PUT/DELETE methods
+    with basic authorization.
+    Works either with WebDAV modules
+    or with custom scripts.
+    """
+
+    def __init__(
+        self: "DAVACMEClient",
+        directory_url: str,
+        *,
+        username: str,
+        password: str,
+        **kwargs: Any,
+    ) -> None:
+        super().__init__(directory_url, **kwargs)
+        self.username = username
+        self.password = password
+
+    def get_auth(self: "DAVACMEClient") -> httpx.Auth:
+        """
+        Get Auth for request.
+
+        Returns:
+            Auth information to be sent along with
+            the request.
+        """
+        return httpx.BasicAuth(
+            username=self.username,
+            password=self.password,
+        )
+
+    @staticmethod
+    def _check_dav_response(resp: httpx.Response) -> None:
+        """
+        Check DAV response.
+
+        Raise an error if necessary.
+        """
+        if resp.status_code > HTTP_MAX_VALID:
+            msg = f"Failed to put challenge: code {resp.status_code}"
+            raise ACMEFulfillmentFailed(msg)
+
+    async def fulfill_http_01(
+        self: "DAVACMEClient", domain: str, challenge: ACMEChallenge
+    ) -> bool:
+        """
+        Perform http-01 fullfilment.
+
+        Execute PUT method to place a token.
+
+        Args:
+            domain: Domain name
+            challenge: ACMEChallenge instance, containing token.
+
+        Returns:
+            True - on succeess
+
+        Raises:
+            ACMEFulfillmentFailed: On error.
+        """
+        async with self._get_client() as client:
+            v = self.get_key_authorization(challenge)
+            resp = await client.put(
+                f"http://{domain}/.well-known/acme-challenge/{challenge.token}",
+                content=v,
+                auth=self.get_auth(),
+            )
+            self._check_dav_response(resp)
+        return True
+
+    async def clear_http_01(
+        self: "DAVACMEClient", domain: str, challenge: ACMEChallenge
+    ) -> None:
+        """
+        Remove provisioned token.
+
+        Args:
+            domain: Domain name
+            challenge: ACMEChallenge instance, containing token.
+
+        Raises:
+            ACMEFulfillmentFailed: On error.
+        """
+        async with self._get_client() as client:
+            resp = await client.delete(
+                f"http://{domain}/.well-known/acme-challenge/{challenge.token}",
+                auth=self.get_auth(),
+            )
+            self._check_dav_response(resp)