feature(*): add authorization with token (optionnal) + get permissions

geosyspy/geosys.py

@@ -26,17 +26,19 @@ class Geosys:
 
     Parameters:
         enum_env: 'Env.PROD' or 'Env.PREPROD'
-        enum_region: 'Region.NA' or 'Region.EU'
+        enum_region: 'Region.NA'
         priority_queue: 'realtime' or 'bulk'
     """
 
-    def __init__(self, client_id: str,
-                 client_secret: str,
-                 username: str,
-                 password: str,
-                 enum_env: Env,
-                 enum_region: Region,
+    def __init__(self,
+                 client_id: str = None,
+                 client_secret: str = None,
+                 username: str = None,
+                 password: str = None,
+                 enum_env: Env = Env.PROD,
+                 enum_region: Region = Region.NA,
                  priority_queue: str = "realtime",
+                 bearer_token: str = None
                  ):
         self.logger = logging.getLogger(__name__)
         self.region: str = enum_region.value
@@ -45,7 +47,7 @@ def __init__(self, client_id: str,
         self.gis_url: str = GIS_API_URLS[enum_region.value][enum_env.value]
         self.priority_queue: str = priority_queue
         self.http_client: HttpClient = HttpClient(client_id, client_secret, username, password, enum_env.value,
-                                                  enum_region.value)
+                                                  enum_region.value, bearer_token)
         self.__master_data_management_service = MasterDataManagementService(self.base_url, self.http_client)
         self.__analytics_fabric_service = AnalyticsFabricService(self.base_url, self.http_client)
         self.__analytics_processor_service = AnalyticsProcessorService(self.base_url, self.http_client)
@@ -410,6 +412,20 @@ def get_available_crops(self):
 
         return crop_enum
 
+    def get_available_permissions(self):
+        """Build the list of available permissions codes for the connected user in an enum
+
+        Returns:
+            permissions: a string array containing all available permissions of the connected user
+        """
+        # get crop code list
+        result = self.__master_data_management_service.get_permission_codes()
+
+        # build a string array with all available permission codes for the connected user
+        permissions = result["permissions"]
+
+        return permissions
+
     ###########################################
     #           AGRIQUEST                     #
     ###########################################

geosyspy/services/master_data_management_service.py

@@ -107,10 +107,36 @@ def get_available_crops_code(self) -> [str]:
 
         dict_response = response.json()
 
-        # extract unique id from response:
         if response.status_code == 200:
             return dict_response
         else:
             raise ValueError(
                 f"Cannot handle HTTP response : {str(response.status_code)} : {str(response.json())}"
             )
+
+
+    def get_permission_codes(self) -> [str]:
+        """Extracts the list of available permissions for the connected user
+
+        Args:
+
+        Returns:
+            A list of string representing the available permissions codes
+
+        Raises:
+            ValueError: The response status code is not as expected.
+        """
+        mdm_url: str = urljoin(self.base_url, GeosysApiEndpoints.MASTER_DATA_MANAGEMENT_ENDPOINT.value + f"/profile?$fields=permissions&$limit=none")
+
+        response = self.http_client.get(mdm_url)
+
+        dict_response = response.json()
+
+        if response.status_code == 200:
+            return dict_response
+        else:
+            raise ValueError(
+                f"Cannot handle HTTP response : {str(response.status_code)} : {str(response.json())}"
+            )
+
+

geosyspy/utils/__init__.py

@@ -1 +1 @@
-__all__ = ["constants", "geosys_platform_urls", "oauth2_client", "http_client", "helper"]
+__all__ = ["constants", "geosys_platform_urls", "oauth2_client", "http_client", "helper", "jwt_validator"]

geosyspy/utils/http_client.py

@@ -32,6 +32,7 @@ def __init__(
             password: str,
             enum_env: str,
             enum_region: str,
+            bearer_token: str = None
     ):
         self.__client_oauth = oauth2_client.Oauth2Api(
             client_id=client_id,
@@ -40,6 +41,7 @@ def __init__(
             username=username,
             enum_env=enum_env,
             enum_region=enum_region,
+            bearer_token=bearer_token
         )
         self.access_token = self.__client_oauth.token
 

geosyspy/utils/jwt_validator.py

@@ -0,0 +1,80 @@
+import jwt
+import base64
+import json
+from datetime import datetime, timezone
+from cryptography import x509
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import serialization
+import logging
+
+
+logger = logging.getLogger(__name__)
+
+
+def check_token_validity(token: str, certificate_key: str, algorithms=['RS256']) -> bool:
+    """
+    Check the validity of a JWT token.
+
+    Args:
+        token (str): The JWT token to check.
+        certificate_key (str): The certificate key in PEM format.
+        algorithms (list, optional): The encryption algorithms to use.
+            Default is ['RS256'].
+
+    Returns:
+        bool: True if the token is valid, False otherwise.
+    """
+    try:
+        cert_bytes = certificate_key.encode()
+        cert = x509.load_pem_x509_certificate(cert_bytes, default_backend())
+        public_key = cert.public_key()
+
+        # extract public key in PEM format
+        public_key = public_key.public_bytes(
+            encoding=serialization.Encoding.PEM,
+            format=serialization.PublicFormat.SubjectPublicKeyInfo
+        ).decode()
+        aud = __get_audience_from_token(token)
+        decoded_token = jwt.decode(token, public_key, algorithms=algorithms, audience=aud)
+        expiration_timestamp = decoded_token['exp']
+        expiration_datetime = datetime.fromtimestamp(expiration_timestamp, tz=timezone.utc)
+
+        if expiration_datetime > datetime.now(timezone.utc):
+            return True
+        else:
+            return False
+    except jwt.ExpiredSignatureError:
+        logger.error("Expired Token")
+        return False
+    except jwt.InvalidTokenError as e:
+        logger.error("Invalid Token." + str(e))
+        return False
+    except Exception as e:
+        logger.error("Invalid Token." + str(e))
+        return False
+
+
+def __get_audience_from_token(token: str) -> str:
+    """
+    Get the audience from a JWT token.
+
+    Args:
+        token (str): The JWT token.
+
+    Returns:
+        str: The audience value.
+    """
+    # parse the token (header, payload, signature)
+    header, payload, signature = token.split('.')
+
+    # add missing data
+    padding = '=' * (4 - len(payload) % 4)
+    payload += padding
+
+    # payload decoding & serialization
+    decoded_payload = base64.urlsafe_b64decode(payload.encode('utf-8'))
+    payload_json = decoded_payload.decode('utf-8')
+
+    payload_dict = json.loads(payload_json)
+    audience = payload_dict.get('aud')
+    return audience

geosyspy/utils/oauth2_client.py

@@ -13,6 +13,7 @@ def __init__(
             password: str,
             enum_env: str,
             enum_region: str,
+            bearer_token: str = None
     ):
         """Initializes a Geosys instance with the required credentials
         to connect to the GEOSYS API.
@@ -24,7 +25,11 @@ def __init__(
         self.token = None
         self.username = username
         self.password = password
-        self.__authenticate()
+
+        if bearer_token:
+            self.token = {"access_token": bearer_token}
+        else:
+            self.__authenticate()
 
     def __authenticate(self):
         """Authenticates the http_client to the API.