Implement onboarding

auth/auth.py

@@ -12,6 +12,11 @@ class Authorization(EnvironmentalService):
     TOKEN_KEY = "token"
     ERROR_KEY = "error"
 
+    def __init__(self, *args, **kwargs):
+        self._public_key = kwargs.pop("public_key")
+        self._private_key = kwargs.pop("private_key")
+        super(Authorization, self).__init__(*args, **kwargs)
+
     def get_auth_request_url(self, parameters: AuthUrlParameter) -> str:
         auth_parameters = parameters.get_parameters()
         return self._environment.get_secured_onboarding_authorization_url(**auth_parameters)
@@ -21,8 +26,9 @@ def extract_auth_response(self, url: str) -> AuthResponse:
         query_params = self._extract_query_params(parsed_url.query)
         return AuthResponse(query_params)
 
-    def verify_auth_response(self):
-        pass
+    @staticmethod
+    def verify_auth_response(response, public_key):
+        response.verify(public_key)
 
     @staticmethod
     def _extract_query_params(query_params: str) -> dict:

auth/response.py

@@ -1,6 +1,11 @@
 import base64
 import json
 from typing import Union
+from urllib.parse import unquote
+
+from cryptography.exceptions import InvalidSignature
+
+from onboarding.signature import verify_signature
 
 
 class AuthResponse:
@@ -19,7 +24,7 @@ def __init__(self, query_params):
         self.is_successful = not bool(self._error)
         self.is_valid = False
 
-    def verify(self) -> None:
+    def verify(self, public_key) -> None:
         """
         Validates signature according to docs:
         https://docs.my-agrirouter.com/agrirouter-interface-documentation/latest/integration/authorization.html#analyse-result
@@ -29,17 +34,23 @@ def verify(self) -> None:
 
         :return:
         """
-        # TODO: implement validation of response according to docs:
-        # https://docs.my-agrirouter.com/agrirouter-interface-documentation/latest/integration/authorization.html#analyse-result
+        encoded_data = self._state + self._token
+        unquoted_signature = unquote(self._signature)
+        encoded_signature = base64.b64decode(unquoted_signature.encode("utf-8"))
+        try:
+            verify_signature(encoded_data, encoded_signature, public_key)
+        except InvalidSignature:
+            print("Response is invalid: invalid signature.")
+            self.is_valid = False
 
         self.is_valid = True
 
     @staticmethod
     def decode_token(token: Union[str, bytes]) -> dict:
         if type(token) == str:
-            token = token.encode("ASCII")
+            token = token.encode("utf-8")
         base_64_decoded_token = base64.b64decode(token)
-        decoded_token = base_64_decoded_token.decode("ASCII")
+        decoded_token = base_64_decoded_token.decode("utf-8")
         return json.loads(decoded_token)
 
     def get_auth_result(self) -> dict:

constants/media_types.py

@@ -0,0 +1,5 @@
+from auth.enums import BaseEnum
+
+
+class ContentTypes(BaseEnum):
+    APPLICATION_JSON = "application/json"

environments/environmental_services.py

@@ -12,4 +12,4 @@ def _set_env(self, env) -> None:
         elif env == "Production":
             self._environment = ProductionEnvironment()
         else:
-            raise InvalidEnvironmentSetup(env=env)
+            raise InvalidEnvironmentSetup(env=env)

onboarding/enums.py

@@ -0,0 +1,11 @@
+from auth.enums import BaseEnum
+
+
+class CertificateTypes(BaseEnum):
+    PEM = "PEM"
+    P12 = "P12"
+
+
+class GateWays(BaseEnum):
+    MQTT = "2"
+    REST = "3"

onboarding/exceptions.py

@@ -0,0 +1,23 @@
+class AgriRouuterBaseException(Exception):
+    _message = ...
+
+    def __init__(self, message=None):
+        if not message:
+            message = self._message
+        self.message = message
+
+
+class WrongCertificationType(AgriRouuterBaseException):
+    _message = "Wrong Certification type. Use onboarding.enums.CertificationTypes values instead."
+
+
+class WrongGateWay(AgriRouuterBaseException):
+    _message = "Wrong Gate Way Id. Use onboarding.enums.GateWays values instead."
+
+
+class RequestNotSigned(AgriRouuterBaseException):
+    _message = """
+    Request does not contain signature header. Please sign the request with request.sign() method.\n
+    Details on: https://docs.my-agrirouter.com/agrirouter-interface-documentation/latest/
+    integration/onboarding.html#signing-requests
+    """

onboarding/headers.py

@@ -0,0 +1,52 @@
+from abc import ABC, abstractmethod
+
+from constants.media_types import ContentTypes
+
+
+class BaseOnboardingHeader(ABC):
+    @abstractmethod
+    def __init__(self, *args, **kwargs):
+        self._set_params(*args, **kwargs)
+
+    @abstractmethod
+    def get_header(self) -> dict:
+        ...
+
+    @abstractmethod
+    def _set_params(self, *args, **kwargs):
+        ...
+
+    @abstractmethod
+    def sign(self, *args, **kwargs):
+        ...
+
+
+class SoftwareOnboardingHeader(BaseOnboardingHeader):
+    def __init__(self,
+                 reg_code,
+                 application_id,
+                 signature=None,
+                 content_type=ContentTypes.APPLICATION_JSON.value
+                 ):
+
+        self._set_params(reg_code, application_id, signature, content_type)
+
+    def get_header(self) -> dict:
+        return self.params
+
+    def sign(self, signature):
+        self.params["X-Agrirouter-Signature"] = signature
+
+    def _set_params(self, reg_code: str, application_id: str, signature: str, content_type: str):
+        header = dict()
+        header["Authorization"] = f"Bearer {reg_code}"
+        header["Content-Type"] = content_type
+        header["X-Agrirouter-ApplicationId"] = application_id
+        if signature:
+            header["X-Agrirouter-Signature"] = signature
+
+        self.params = header
+
+
+class CUOnboardingHeader(BaseOnboardingHeader):
+    pass

onboarding/onboarding.py

@@ -0,0 +1,83 @@
+import requests
+
+from environments.environmental_services import EnvironmentalService
+from onboarding.exceptions import RequestNotSigned
+from onboarding.headers import SoftwareOnboardingHeader, CUOnboardingHeader
+from onboarding.parameters import SoftwareOnboardingParameter, BaseOnboardingParameter, CUOnboardingParameter
+from onboarding.request import SoftwareOnboardingRequest, BaseOnboardingRequest, CUOnboardingRequest
+from onboarding.request_body import SoftwareOnboardingBody, CUOnboardingBody
+from onboarding.response import SoftwareVerifyOnboardingResponse, SoftwareOnboardingResponse, CUOnboardingResponse
+
+
+class SoftwareOnboarding(EnvironmentalService):
+
+    def __init__(self, *args, **kwargs):
+        self._public_key = kwargs.pop("public_key")
+        self._private_key = kwargs.pop("private_key")
+        super(SoftwareOnboarding, self).__init__(*args, **kwargs)
+
+    def _create_request(self, params: BaseOnboardingParameter, url: str) -> SoftwareOnboardingRequest:
+        body_params = params.get_body_params()
+        request_body = SoftwareOnboardingBody(**body_params)
+
+        header_params = params.get_header_params()
+        request_header = SoftwareOnboardingHeader(**header_params)
+
+        return SoftwareOnboardingRequest(header=request_header, body=request_body, url=url)
+
+    def _perform_request(self, params: BaseOnboardingParameter, url: str) -> requests.Response:
+        request = self._create_request(params, url)
+        request.sign(self._private_key)
+        if request.is_signed:
+            return requests.post(
+                url=request.get_url(),
+                data=request.get_data(),
+                headers=request.get_header()
+            )
+        raise RequestNotSigned
+
+    def verify(self, params: SoftwareOnboardingParameter) -> SoftwareOnboardingResponse:
+        url = self._environment.get_verify_onboard_request_url()
+        http_response = self._perform_request(params=params, url=url)
+
+        return SoftwareOnboardingResponse(http_response)
+
+    def onboard(self, params: SoftwareOnboardingParameter) -> SoftwareOnboardingResponse:
+        url = self._environment.get_secured_onboard_url()
+        http_response = self._perform_request(params=params, url=url)
+
+        return SoftwareOnboardingResponse(http_response)
+
+
+class CUOnboarding(EnvironmentalService):
+
+    def __init__(self, *args, **kwargs):
+        self._public_key = kwargs.pop("public_key")
+        self._private_key = kwargs.pop("private_key")
+        super(CUOnboarding, self).__init__(*args, **kwargs)
+
+    def _create_request(self, params: CUOnboardingParameter, url: str) -> CUOnboardingRequest:
+        body_params = params.get_body_params()
+        request_body = CUOnboardingBody(**body_params)
+
+        header_params = params.get_header_params()
+        request_header = CUOnboardingHeader(**header_params)
+
+        return CUOnboardingRequest(header=request_header, body=request_body, url=url)
+
+    def _perform_request(self, params: CUOnboardingParameter, url: str) -> requests.Response:
+        request = self._create_request(params, url)
+        request.sign(self._private_key)
+        if request.is_signed:
+            return requests.post(
+                url=request.get_url(),
+                data=request.get_data(),
+                headers=request.get_header()
+            )
+        raise RequestNotSigned
+
+    def onboard(self, params: CUOnboardingParameter) -> CUOnboardingResponse:
+        url = self._environment.get_onboard_url()
+        http_response = self._perform_request(params=params, url=url)
+
+        return CUOnboardingResponse(http_response)

onboarding/parameters.py

@@ -0,0 +1,95 @@
+from abc import ABC, abstractmethod
+
+from constants.media_types import ContentTypes
+from onboarding.enums import CertificateTypes
+
+
+class BaseOnboardingParameter(ABC):
+    @abstractmethod
+    def __init__(self, *args, **kwargs):
+        ...
+
+    @abstractmethod
+    def get_header_params(self, *args, **kwargs):
+        ...
+
+    @abstractmethod
+    def get_body_params(self, *args, **kwargs):
+        ...
+
+
+class SoftwareOnboardingParameter(BaseOnboardingParameter):
+    def __init__(self,
+                 id_,
+                 application_id,
+                 certification_version_id,
+                 gateway_id,
+                 utc_timestamp,
+                 time_zone,
+                 reg_code,
+                 content_type=ContentTypes.APPLICATION_JSON.value,
+                 certificate_type=CertificateTypes.P12.value,
+                 ):
+
+        self.id_ = id_
+        self.application_id = application_id
+        self.content_type = content_type
+        self.certification_version_id = certification_version_id
+        self.gateway_id = str(gateway_id)
+        self.certificate_type = certificate_type
+        self.utc_timestamp = str(utc_timestamp)
+        self.time_zone = str(time_zone)
+        self.reg_code = reg_code
+
+    def get_header_params(self):
+        return {
+            "content_type": self.content_type,
+            "reg_code": self.reg_code,
+            "application_id": self.application_id,
+        }
+
+    def get_body_params(self):
+        return {
+            "id_": self.id_,
+            "application_id": self.application_id,
+            "certification_version_id": self.certification_version_id,
+            "gateway_id": self.gateway_id,
+            "certificate_type": self.certificate_type,
+            "utc_timestamp": self.utc_timestamp,
+            "time_zone": self.time_zone,
+        }
+
+
+class CUOnboardingParameter(BaseOnboardingParameter):
+    def __init__(self,
+                 id_,
+                 application_id,
+                 certification_version_id,
+                 gateway_id,
+                 reg_code,
+                 content_type=ContentTypes.APPLICATION_JSON.value,
+                 certificate_type=CertificateTypes.P12.value,
+                 ):
+
+        self.id_ = id_
+        self.application_id = application_id
+        self.content_type = content_type
+        self.certification_version_id = certification_version_id
+        self.gateway_id = gateway_id
+        self.certificate_type = certificate_type
+        self.reg_code = reg_code
+
+    def get_header_params(self):
+        return {
+            "content_type": self.content_type,
+            "reg_code": self.reg_code,
+        }
+
+    def get_body_params(self):
+        return {
+            "id_": self.id_,
+            "application_id": self.application_id,
+            "certification_version_id": self.certification_version_id,
+            "gateway_id": self.gateway_id,
+            "certificate_type": self.certificate_type,
+        }

onboarding/request.py

@@ -0,0 +1,44 @@
+from onboarding.headers import SoftwareOnboardingHeader, BaseOnboardingHeader
+from onboarding.request_body import SoftwareOnboardingBody, BaseOnboardingBody
+from onboarding.signature import create_signature
+
+
+class BaseOnboardingRequest:
+    def __init__(self, header: BaseOnboardingHeader, body: BaseOnboardingBody, url: str):
+        self.header = header
+        self.body = body
+        self.url = url
+
+    def get_url(self):
+        return self.url
+
+    def get_data(self):
+        return self.body.get_parameters()
+
+    def get_header(self):
+        return self.header.get_header()
+
+    def sign(self, private_key):
+        signature = create_signature(self.body.json(new_lines=False), private_key)
+        self.header.sign(signature)
+
+    @property
+    def is_signed(self):
+        header_has_signature = self.get_header().get("X-Agrirouter-Signature", None)
+        if header_has_signature:
+            return True
+        return False
+
+
+class SoftwareOnboardingRequest(BaseOnboardingRequest):
+    """
+    Request must be used to onboard Farming Software or Telemetry Platform
+    """
+    pass
+
+
+class CUOnboardingRequest(BaseOnboardingRequest):
+    """
+    Request must be used to onboard CUs
+    """
+    pass