Encrpytion Module: Added ability for multiple providers through type …

…param

g2p_encryption/models/encryption_provider.py

@@ -12,26 +12,52 @@ def encrypt_data(self, data: bytes, **kwargs) -> bytes:
         """
         Both input and output are NOT base64 encoded
         """
-        raise NotImplementedError()
+        try:
+            encrypt_func = getattr(self, f"encrypt_data_{self.type}")
+        except Exception as e:
+            raise NotImplementedError() from e
+        return encrypt_func(data, **kwargs)
 
     def decrypt_data(self, data: bytes, **kwargs) -> bytes:
         """
         Both input and output are NOT base64 encoded
         """
-        raise NotImplementedError()
+        try:
+            decrypt_func = getattr(self, f"decrypt_data_{self.type}")
+        except Exception as e:
+            raise NotImplementedError() from e
+        return decrypt_func(data, **kwargs)
 
     def jwt_sign(
         self,
         data,
         include_payload=True,
         include_certificate=False,
         include_cert_hash=False,
-        **kwargs
+        **kwargs,
     ) -> str:
-        raise NotImplementedError()
+        try:
+            jwt_func = getattr(self, f"jwt_sign_{self.type}")
+        except Exception as e:
+            raise NotImplementedError() from e
+        return jwt_func(
+            data,
+            include_payload=True,
+            include_certificate=False,
+            include_cert_hash=False,
+            **kwargs,
+        )
 
     def jwt_verify(self, data: str, **kwargs):
-        raise NotImplementedError()
+        try:
+            jwt_func = getattr(self, f"jwt_verify_{self.type}")
+        except Exception as e:
+            raise NotImplementedError() from e
+        return jwt_func(data, **kwargs)
 
     def get_jwks(self, **kwargs):
-        raise NotImplementedError()
+        try:
+            jwk_func = getattr(self, f"get_jwks_{self.type}")
+        except Exception as e:
+            raise NotImplementedError() from e
+        return jwk_func(**kwargs)

g2p_encryption_keymanager/models/encryption_provider.py

@@ -63,7 +63,7 @@ def km_generate_current_time(self):
     keymanager_encrypt_salt = fields.Char(default=_km_random_secret)
     keymanager_encrypt_aad = fields.Char(default=_km_random_secret)
 
-    def encrypt_data(self, data: bytes, **kwargs) -> bytes:
+    def encrypt_data_keymanager(self, data: bytes, **kwargs) -> bytes:
         self.ensure_one()
         access_token = self.km_get_access_token()
         current_time = self.km_generate_current_time()
@@ -94,7 +94,7 @@ def encrypt_data(self, data: bytes, **kwargs) -> bytes:
             return self.km_urlsafe_b64decode(response.get("data"))
         raise ValueError("Could not encrypt data, invalid keymanager response")
 
-    def decrypt_data(self, data: bytes, **kwargs) -> bytes:
+    def decrypt_data_keymanager(self, data: bytes, **kwargs) -> bytes:
         self.ensure_one()
         access_token = self.km_get_access_token()
         current_time = self.km_generate_current_time()
@@ -125,7 +125,7 @@ def decrypt_data(self, data: bytes, **kwargs) -> bytes:
             return self.km_urlsafe_b64decode(response.get("data"))
         raise ValueError("Could not decrypt data, invalid keymanager response")
 
-    def jwt_sign(
+    def jwt_sign_keymanager(
         self,
         data,
         include_payload=True,
@@ -168,7 +168,7 @@ def jwt_sign(
             return response.get("jwtSignedData")
         raise ValueError("Could not sign jwt, invalid keymanager response")
 
-    def jwt_verify(self, data: str, **kwargs):
+    def jwt_verify_keymanager(self, data: str, **kwargs):
         self.ensure_one()
         access_token = self.km_get_access_token()
         current_time = self.km_generate_current_time()
@@ -201,7 +201,7 @@ def jwt_verify(self, data: str, **kwargs):
             return jwt.get_unverified_claims(data)
         raise ValueError("invalid jwt signature")
 
-    def get_jwks(self, **kwargs):
+    def get_jwks_keymanager(self, **kwargs):
         # TODO: Cache this JWKS response somehow
         self.ensure_one()
         access_token = self.km_get_access_token()