Hy main

cert_manager/__init__.py

@@ -10,7 +10,8 @@
 from .report import Report
 from .smime import SMIME
 from .ssl import SSL
+from .dcv import DomainControlValidation
 
 __all__ = [
-    "ACMEAccount", "Admin", "Client", "Domain", "Organization", "PendingError", "Person", "Report", "SMIME", "SSL"
+    "ACMEAccount", "Admin", "Client", "Domain", "DomainControlValidation", "Organization", "PendingError", "Person", "Report", "SMIME", "SSL"
 ]

cert_manager/acme.py

@@ -175,6 +175,29 @@ def add_domains(self, acme_id, domains):
 
         return result.json()
 
+    def get_domains (self, acme_id, **kwargs):
+        """List ACME account’s domains
+
+        :param int acme_id: The ID of the acme account to list domains
+        """
+        #self._change_api_version("v2")
+        self.__acc_domains = []
+        result = self.__find_domains(acme_id)
+        for dom in result:
+            self.__acc_domains.append(dom)
+        return self.__acc_domains
+
+    @paginate
+    def __find_domains(self, acme_id, **kwargs):
+
+        params = {
+          self._find_params_to_api[param]: kwargs.get(param)
+          for param in self._find_params_to_api  # pylint:disable=consider-using-dict-items
+        }
+        url = self._url(f"{acme_id}", "domain")
+        result = self._client.get(url, params=params)
+        return result.json()
+
     def remove_domains(self, acme_id, domains):
         """Remove domains from an acme account.
 

cert_manager/dcv.py

@@ -5,18 +5,33 @@
 from requests.exceptions import HTTPError
 
 from ._endpoint import Endpoint
+from ._helpers import paginate
 
 
 class DomainControlValidation(Endpoint):
     """Query the Sectigo Cert Manager REST API for Domain Control Validation (DCV) data."""
-
     def __init__(self, client, api_version="v1"):
         """Initialize the class.
 
         :param object client: An instantiated cert_manager.Client object
         :param string api_version: The API version to use; the default is "v1"
         """
         super().__init__(client=client, endpoint="/dcv", api_version=api_version)
+        self.__dcv_domains = None
+
+    def all(self, force=False):
+        """Return list of all domain control validations.
+        :param bool force: If set to True, force refreshing the data from the API
+        """
+        if (self.__dcv_domains) and (not force):
+            return self.__dcv_domains
+
+        self.__dcv_domains = []
+        result = self.__search()
+        for dom in result:
+            self.__dcv_domains.append(dom)
+
+        return self.__dcv_domains
 
     def search(self, **kwargs):
         """Search the DCV statuses of domains.
@@ -34,6 +49,11 @@ def search(self, **kwargs):
 
         return result.json()
 
+    @paginate
+    def __search(self, **kwargs):
+        """Paginated wrapper for search"""
+        return self.search(**kwargs)
+
     def get_validation_status(self, domain: str):
         """Get the DCV statuses of a domain.
 
@@ -85,6 +105,26 @@ def start_validation_cname(self, domain: str):
 
         return result.json()
 
+    def start_validation_email(self, domain: str):
+        """Start Domain Control Validation using Email method.
+
+        :param string domain: The domain to validate
+        :return response: List of valid email addresses
+        """
+        url = self._url("validation", "start", "domain", "email")
+        data = {"domain": domain}
+
+        try:
+            result = self._client.post(url, data=data)
+        except HTTPError as exc:
+            status_code = exc.response.status_code
+            if status_code == HTTPStatus.BAD_REQUEST:
+                err_response = exc.response.json()
+                raise ValueError(err_response["description"]) from exc
+            raise exc
+
+        return result.json()
+
     def submit_validation_cname(self, domain: str):
         """Finish Domain Control Validation using the CNAME method.
 
@@ -111,3 +151,29 @@ def submit_validation_cname(self, domain: str):
             raise exc
 
         return result.json()
+
+    def submit_validation_email(self, domain: str, email: str):
+        """Finish Domain Control Validation using the email method.
+
+        :param string domain: The domain to validate
+        :param string email: validation email sent to
+
+        :return response: a dictionary containing
+            status: The status of the validation
+            orderStatus: The status of the validation request
+            message: An optional message to help with debugging
+        """
+        url = self._url("validation", "submit", "domain", "email")
+        data = {"domain": domain,
+                "email": email}
+
+        try:
+            result = self._client.post(url, data=data)
+        except HTTPError as exc:
+            status_code = exc.response.status_code
+            if status_code == HTTPStatus.BAD_REQUEST:
+                err_response = exc.response.json()
+                raise ValueError(err_response["description"]) from exc
+            raise exc
+
+        return result.json()

tests/test_acme.py

@@ -354,6 +354,32 @@ def test_ne_acme_id(self):
         self.assertRaises(HTTPError, acme.get, acme_id)
 
 
+class TestGet_Domains(TestACMEAccount):
+    """Test the .get_domains method."""
+
+    @responses.activate
+    def test_acme_id(self):
+        """Return all domains from the specified ACME ID."""
+        acme_id = 1234
+        api_url = self.get_acme_account_url(acme_id) + "/domain?size=200&position=0"
+        valid_response = []
+
+        # Setup the mocked response
+        responses.add(responses.GET, api_url, json=valid_response, status=200)
+
+        acme = ACMEAccount(client=self.client)
+        data = acme.get_domains(acme_id)
+
+        self.assertEqual(len(responses.calls), 1)
+        self.assertEqual(responses.calls[0].request.url, api_url)
+        self.assertEqual(data, valid_response)
+
+
+    def test_need_acme_id(self):
+        """Raise an exception without an acme_id parameter."""
+        acme = ACMEAccount(client=self.client)
+        self.assertRaises(TypeError, acme.get_domains)
+
 def _test_create_test_factory(acme_id=1234, header="location", **kwargs):
     """Act as a wrapper to inject headers and parameters."""
     params = ["name", "acmeServer", "organizationId", "evDetails"]