test(protocol): Introduce account management test

Refs: #141

Author: piraz

automatoes/model.py

@@ -24,13 +24,18 @@
 
 class Account:
 
-    def __init__(self, key, uri=None):
+    def __init__(self, key, **kwargs):
+        self.contact = kwargs.get("contact")
+        self.orders = kwargs.get("orders")
+        # TODO: Maybe store the key format
         self.key = key
-        self.uri = uri
+        self.uri = kwargs.get("uri")
 
     def serialize(self):
         return json.dumps({
+            'contact': self.contact,
             'key': export_private_key(self.key).decode("utf-8"),
+            'orders': self.orders,
             'uri': self.uri,
         }).encode("utf-8")
 
@@ -46,8 +51,17 @@ def deserialize(data):
             data = json.loads(data)
             if "key" not in data or "uri" not in data:
                 raise ValueError("Missing 'key' or 'uri' fields.")
-            return Account(key=load_private_key(data['key'].encode("utf8")),
-                           uri=data['uri'])
+            if "contact" in data:
+                raise ValueError("Missing 'key' or 'uri' fields.")
+            account = Account(key=load_private_key(data['key'].encode("utf8")),
+                              uri=data['uri'])
+            # TODO: Check how to deal with this after migration from 0.9.x to
+            # 1.x finishes
+            if "contact" in data:
+                account.contact = data['contact']
+            if "orders" in data:
+                account.orders = data['orders']
+            return account
         except (TypeError, ValueError, AttributeError) as e:
             raise IOError("Invalid account structure: {}".format(e))
 
@@ -142,7 +156,9 @@ def deserialize(data):
             raise IOError("Invalid account structure: {}".format(e))
 
 
-RegistrationResult = namedtuple("RegistrationResult", "contents uri terms")
+AccountResponse = namedtuple("AccountResponse", "orders status terms")
+RegistrationResponse = namedtuple("RegistrationResponse",
+                                  "contact orders status")
 NewAuthorizationResult = namedtuple("NewAuthorizationResult", "contents uri")
 IssuanceResult = namedtuple("IssuanceResult",
                             "certificate location intermediate")

automatoes/protocol.py

@@ -15,15 +15,15 @@
 from automatoes import get_version
 from automatoes.crypto import generate_protected_header, sign_request_v2
 from automatoes.errors import AccountAlreadyExistsError, AcmeError
-from automatoes.model import Account, RegistrationResult
+from automatoes.model import (Account, AccountResponse, RegistrationResponse)
 
 import copy
 
 from peasant.client.protocol import Peasant
 from peasant.client.transport import METHOD_POST
 from peasant.client.transport_requests import RequestsTransport
 
-import requests
+import warnings
 
 
 class AcmeV2Pesant(Peasant):
@@ -47,11 +47,11 @@ def directory_path(self, path):
     def verify(self):
         return self._verify
 
-    def get_registration(self, account: Account):
+    def get_registration(self, account: Account) -> RegistrationResponse:
         return self.transport.get_registration(account)
 
     def new_account(self, account: Account, contacts: list,
-                    terms_agreed: bool = False) -> RegistrationResult:
+                    terms_agreed: bool = False) -> AccountResponse:
         return self.transport.new_account(account, contacts, terms_agreed)
 
 
@@ -84,8 +84,7 @@ def __kwargs_updater(self, method, **kwargs):
                 protected['kid'] = kid
                 protected.pop("jwk")
             data = kwargs.get("data")
-            if data:
-                kwargs['data'] = sign_request_v2(key, protected, data)
+            kwargs['data'] = sign_request_v2(key, protected, data)
         if self.peasant.verify:
             kwargs['verify'] = self.peasant.verify
         return kwargs
@@ -102,21 +101,34 @@ def get_protected_headers(self, key, uri=None):
         return protected_header
 
     def post_as_get(self, path, **kwargs):
-        """Send a POST request.
+        """Send a POST as GET request.
+
+        This method enforces no data/body being passed into kwargs.
+
+        If either data or body is present into kwargs a warning will be
+        displayed and the value will be discarded.
+
+        This post will be sent without body into.
 
         :param path: absolute or relative URL for the new
         :class:`requests.Request` object.
         :param **kwargs: Optional arguments that ``request`` takes.
         :return: :class:`requests.Response <Response>` object
         :rtype: requests.Response
         """
-        url = self.get_url(path, **kwargs)
-        headers = self.get_headers(**kwargs)
-        kwargs['headers'] = headers
-        kwargs = self.update_kwargs(METHOD_POST, **kwargs)
-        with requests.post(url, **kwargs) as result:
-            result.raise_for_status()
-        return result
+        if "data" in kwargs:
+            kwargs.pop("data")
+            warnings.warn("Do not pass the key 'data' in kwargs. The "
+                          "post_as_get method won't add body to the request "
+                          "and the value will be ignored!",
+                          category=RuntimeWarning)
+        if "body" in kwargs:
+            kwargs.pop("body")
+            warnings.warn("Do not pass the key 'body' in kwargs. The "
+                          "post_as_get method won't add body to the request "
+                          "and the value will be ignored!",
+                          category=RuntimeWarning)
+        return self.post(path, **kwargs)
 
     def set_directory(self):
         response = self.get("/%s" % self.peasant.directory_path)
@@ -125,15 +137,21 @@ def set_directory(self):
         else:
             raise Exception
 
-    def get_registration(self, account: Account):
+    def get_registration(self, account: Account) -> RegistrationResponse:
         """
         Get available account information from the server.
         """
         headers = {'Content-Type': "application/jose+json"}
         response = self.post_as_get(account.uri, headers=headers,
-                                    kid=self.account.uri,
-                                    key=account.key, uri=account.uri)
+                                    kid=account.uri, key=account.key,
+                                    uri=account.uri)
         if str(response.status_code).startswith("2"):
+            response_json = _json(response)
+            return RegistrationResponse(
+                contact=response_json['contact'],
+                orders=response_json['orders'],
+                status=response_json['status'],
+            )
             return _json(response)
         raise AcmeError(response)
 
@@ -146,37 +164,39 @@ def new_nonce(self):
 
     # TODO: Document that this method used to be called register
     def new_account(self, account: Account, contacts: list,
-                    terms_agreed: bool = False) -> RegistrationResult:
+                    terms_agreed: bool = False) -> AccountResponse:
         """ Create a new account in the Acme Server """
         payload = {
            "termsOfServiceAgreed": terms_agreed,
            "contact": [f"mailto:{contact}" for contact in contacts],
         }
         headers = {'Content-Type': "application/jose+json"}
         path = self.peasant.directory()['newAccount']
-        response = self.post_as_get(path, data=payload, headers=headers,
-                                    key=account.key, uri=path)
-
-        uri = response.headers.get("Location")
-
+        response = self.post(path, data=payload, headers=headers,
+                             key=account.key, uri=path)
         if response.status_code == 201:
+            uri = response.headers.get("Location")
             # Find terms of service from link headers
             terms = self.terms_from_directory()
-            return RegistrationResult(
-                contents=_json(response),
-                uri=uri,
-                terms=terms
+            response_json = _json(response)
+            account.contact = response_json['contact']
+            account.orders = response_json['orders']
+            account.uri = uri
+            return AccountResponse(
+                orders=response_json['orders'],
+                status=response_json['status'],
+                terms=terms,
             )
         elif response.status_code == 409:
             raise AccountAlreadyExistsError(response, uri)
         raise AcmeError(response)
 
     def terms_from_directory(self):
-        response = self.get_directory()
-        if response.status_code == 200:
-            if "meta" in response.json():
-                if "termsOfService" in response.json()['meta']:
-                    return response.json()['meta']['termsOfService']
+        # TODO: Check how to trigger an error here
+        directory = self.peasant.directory()
+        if "meta" in directory:
+            if "termsOfService" in directory['meta']:
+                return directory['meta']['termsOfService']
         return None
 
 

tests/account_management_test.py

@@ -0,0 +1,61 @@
+# 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.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from automatoes.crypto import generate_rsa_key
+from automatoes.model import Account
+from cartola import security
+from tests import get_protocol, get_transport, PEEBLE_URL
+import unittest
+
+
+class AccountTestCase(unittest.TestCase):
+    """ Test letsencrypt nonce
+    """
+
+    def setUp(self) -> None:
+        self.protocol = get_protocol(get_transport())
+
+    def test_new_acount(self):
+        contact = "candango_{}_{}@candango.org".format(
+                security.random_string(5, False, False),
+                security.random_string(5, False, False)
+                )
+        # To check against the get_registration method after
+        # TODO: check against more than one emails in the contacts
+        peeble_term = ("data:text/plain,Do%20what%20thou%20wilt")
+        account = Account(key=generate_rsa_key(4096))
+        response = self.protocol.new_account(account, [contact], True)
+        self.assertEqual(peeble_term, response.terms)
+        self.assertEqual("valid", response.status)
+        urix = account.uri.split("/")
+        self.assertEqual(PEEBLE_URL, "/".join(urix[0:3]))
+        self.assertEqual("my-account", "/".join(urix[3:4]))
+        self.assertIsInstance(urix[4:5][0], str)
+        ordersx = account.orders.split("/")
+        self.assertEqual(PEEBLE_URL, "/".join(ordersx[0:3]))
+        self.assertEqual("list-orderz", "/".join(ordersx[3:4]))
+        self.assertIsInstance(ordersx[4:5][0], str)
+        self.assertEqual([f"mailto:{contact}"], account.contact)
+
+    def test_get_acount(self):
+        contact = "candango_{}_{}@candango.org".format(
+                security.random_string(5, False, False),
+                security.random_string(5, False, False)
+                )
+        account = Account(key=generate_rsa_key(4096))
+        self.protocol.new_account(account, [contact], True)
+        response = self.protocol.get_registration(account)
+        self.assertEqual("valid", response.status)
+        self.assertEqual(account.contact, response.contact)
+        self.assertEqual(account.orders, response.orders)