chore(crypto): add ari tests

Refs: #121

Author: piraz

automatoes/errors.py

@@ -1,6 +1,4 @@
-#!/usr/bin/env python
-#
-# Copyright 2019 Flavio Garcia
+# Copyright 2019-2025 Flavio Garcia
 # Copyright 2016-2017 Veeti Paananen under MIT License
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -25,8 +23,8 @@ def __init__(self, response):
         try:
             details = response.json()
             self.type = details.get('type', 'unknown')
-            message = "{} (type {}, HTTP {})".format(details.get('detail'),
-                    self.type, response.status_code)
+            message = (f"{details.get('detail')} (type {self.type}, HTTP "
+                       f"{response.status_code})")
         except (ValueError, TypeError, AttributeError):
             pass
         super().__init__(message)

automatoes/protocol.py

@@ -1,4 +1,4 @@
-# Copyright 2019-2024 Flavio Garcia
+# Copyright 2019-2025 Flavio Garcia
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -12,8 +12,12 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from . import get_version
+from automatoes import get_version
+from automatoes.errors import AccountAlreadyExistsError, AcmeError
+from automatoes.model import Account, RegistrationResult
+
 from peasant.client.protocol import Peasant
+from peasant.client.transport import METHOD_POST
 from peasant.client.transport_requests import RequestsTransport
 
 
@@ -23,24 +27,9 @@ def __init__(self, transport, **kwargs):
         """
         """
         super().__init__(transport)
-        self._url = kwargs.get("url")
-        self._account = kwargs.get("account")
         self._directory_path = kwargs.get("directory", "directory")
         self._verify = kwargs.get("verify")
 
-    @property
-    def url(self):
-        return self._url
-
-    @property
-    def account(self):
-        return self.account
-
-    @account.setter
-    def account(self, account):
-        # TODO: Throw an error right here if account is None
-        self._account = account
-
     @property
     def directory_path(self):
         return self._directory_path
@@ -80,8 +69,44 @@ def set_directory(self):
             raise Exception
 
     def new_nonce(self):
-        """ Returns a new nonce """
+        """ Return a new nonce """
         return self.head(self.peasant.directory()['newNonce'], headers={
             'resource': "new-reg",
             'payload': None,
         }).headers.get('Replay-Nonce')
+
+    def new_account(self, account: Account, contacts: list,
+                    terms_agreed: bool = False):
+        """ Create a new account in the Acme Server """
+        payload = {
+           "termsOfServiceAgreed": terms_agreed,
+           "contact": [f"mailto:{contact}" for contact in contacts],
+        }
+        response = self.post(
+            self.peasant.directory()['newAccount'],
+            payload
+        )
+
+        uri = response.headers.get("Location")
+
+        if response.status_code == 201:
+            self.account.uri = uri
+
+            # Find terms of service from link headers
+            terms = self.terms_from_directory()
+
+            return RegistrationResult(
+                contents=_json(response),
+                uri=uri,
+                terms=terms
+            )
+        elif response.status_code == 409:
+            raise AccountAlreadyExistsError(response, uri)
+        raise AcmeError(response)
+
+
+def _json(response):
+    try:
+        return response.json()
+    except ValueError as e:
+        raise AcmeError("Invalid JSON response. {}".format(e))

tests/__init__.py

@@ -1,4 +1,4 @@
-# Copyright 2019-2024 Flavio Garcia
+# Copyright 2019-2025 Flavio Garcia
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -12,13 +12,30 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from automatoes.protocol import AcmeV2Pesant, AcmeRequestsTransport
 import os
 
 
 TEST_ROOT = os.path.dirname(os.path.abspath(__file__))
 FIXTURES_ROOT = os.path.abspath(os.path.join(TEST_ROOT, "fixtures"))
 PROJECT_ROOT = os.path.abspath(os.path.join(TEST_ROOT, ".."))
 
+PEEBLE_URL = "https://localhost:14000"
+
+
+def get_transport() -> AcmeRequestsTransport:
+    return AcmeRequestsTransport(PEEBLE_URL)
+
+
+def get_protocol(transport: AcmeRequestsTransport = None) -> AcmeV2Pesant:
+    if transport is None:
+        transport = get_transport()
+    return AcmeV2Pesant(
+        transport,
+        directory="dir",
+        verify=get_absolute_path("certs/candango.minica.pem")
+    )
+
 
 def get_absolute_path(directory):
     return os.path.realpath(

tests/ari_test.py

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from . import FIXTURES_ROOT
+from tests import FIXTURES_ROOT, get_protocol
 from automatoes.crypto import (generate_ari_data,
                                get_certificate_aki,
                                get_certificate_serial,
@@ -21,33 +21,49 @@
 from cartola import fs
 import unittest
 import os
+from tornado import testing
 
 
-class ARITestCase(unittest.TestCase):
-    """ Tests the crypto module from automatoes
+class ARITestCase(testing.AsyncTestCase):
+    """ Test letsencrypt nonce
     """
 
-    def test_aki(self):
-        """ Test the strip_certificate function """
+    @testing.gen_test
+    async def test_nonce(self):
+        protocol = get_protocol()
+        # print(protocol.directory()['renewalInfo'])
+        self.assertIsNotNone(protocol.new_nonce())
+
+
+class AKISerialTestCase(unittest.TestCase):
+    """ Test aki and serial fucntion to create the data to be sent for aki
+    request
+    """
+
+    def setUp(self) -> None:
         key_directory = os.path.join(FIXTURES_ROOT, "keys", "candango.org",
                                      "another")
-
         key_crt = fs.read(
             os.path.join(key_directory, "another.candango.org.crt"),
             True
         )
+        self.cert = load_pem_certificate(key_crt)
 
-        pem_crt = load_pem_certificate(key_crt)
+        self.expected_aki = ("c0:cc:03:46:b9:58:20:cc:5c:72:70:f3:e1:2e:cb:"
+                             "20:a6:f5:68:3a")
+        self.expected_serial = ("fa:f3:97:73:26:ea:e8:44:e7:14:00:20:ae:90:"
+                                "60:af:ba:44")
 
-        aki = get_certificate_aki(pem_crt)
-        serial = get_certificate_serial(pem_crt)
-        expected_aki = ("c0:cc:03:46:b9:58:20:cc:5c:72:70:f3:e1:2e:cb:20:a6:"
-                        "f5:68:3a")
-        expected_serial = ("fa:f3:97:73:26:ea:e8:44:e7:14:00:20:ae:90:60:af:"
-                           "ba:44")
-        aki_b64 = base64.urlsafe_b64encode(expected_aki.encode())
-        serial_b64 = base64.urlsafe_b64encode(expected_serial.encode())
+    def test_aki_serial(self):
+        """ Test cert aki and serial """
+        aki = get_certificate_aki(self.cert)
+        serial = get_certificate_serial(self.cert)
+        self.assertEqual(self.expected_aki, aki)
+        self.assertEqual(self.expected_serial, serial)
 
-        self.assertEqual(expected_aki, aki)
-        self.assertEqual(expected_serial, serial)
-        self.assertEqual(f"{aki_b64}.{serial_b64}", generate_ari_data(pem_crt))
+    def test_ari_data(self):
+        """ Test ari_data from aki and serial """
+        aki_b64 = base64.urlsafe_b64encode(self.expected_aki.encode())
+        serial_b64 = base64.urlsafe_b64encode(self.expected_serial.encode())
+        self.assertEqual(f"{aki_b64}.{serial_b64}",
+                         generate_ari_data(self.cert))

tests/nonce_test.py

@@ -1,4 +1,4 @@
-# Copyright 2019-2024 Flavio Garcia
+# Copyright 2019-2025 Flavio Garcia
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -12,8 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from . import get_absolute_path
-from automatoes.protocol import AcmeV2Pesant, AcmeRequestsTransport
+from tests import get_protocol, get_transport
 from tornado import testing
 
 
@@ -22,11 +21,7 @@ class NonceTestCase(testing.AsyncTestCase):
     """
 
     @testing.gen_test
-    async def test_auth(self):
-        transport = AcmeRequestsTransport("https://localhost:14000")
-        protocol = AcmeV2Pesant(
-                transport,
-                directory="dir",
-                verify=get_absolute_path("certs/candango.minica.pem")
-        )
+    async def test_nonce(self):
+        protocol = get_protocol(get_transport())
+        print(protocol.new_nonce())
         self.assertIsNotNone(protocol.new_nonce())

tests/runtests.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 #
-# Copyright 2019-2024 Flavio Garcia
+# Copyright 2019-2025 Flavio Garcia
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -15,12 +15,13 @@
 # limitations under the License.
 
 import unittest
-from tests import crypto_test
+from tests import ari_test, crypto_test
 
 
 def suite():
     testLoader = unittest.TestLoader()
     alltests = unittest.TestSuite()
+    alltests.addTests(testLoader.loadTestsFromModule(ari_test))
     alltests.addTests(testLoader.loadTestsFromModule(crypto_test))
     return alltests