Merge pull request #371 from OpenSPP/spp_encryption

VCI Issuer

Author: gonzalesedwin1123

requirements.txt

@@ -5,9 +5,11 @@ bravado_core
 faker
 geojson
 jsonschema
+jwcrypto
 pyjwt>=2.4.0
 pyproj
 python-magic
+qrcode
 shapely
 simplejson
 swagger_spec_validator

spp_encryption/__init__.py

@@ -0,0 +1 @@
+from . import models

spp_encryption/__manifest__.py

@@ -0,0 +1,27 @@
+# Part of OpenSPP. See LICENSE file for full copyright and licensing details.
+
+
+{
+    "name": "SPP Encryption",
+    "category": "OpenSPP",
+    "version": "17.0.1.0.0",
+    "sequence": 1,
+    "author": "OpenSPP.org",
+    "website": "https://github.com/OpenSPP/openspp-modules",
+    "license": "LGPL-3",
+    "development_status": "Beta",
+    "maintainers": ["jeremi", "gonzalesedwin1123"],
+    "depends": [
+        "g2p_encryption",
+    ],
+    "external_dependencies": {"python": ["jwcrypto"]},
+    "data": [
+        "views/encryption_provider.xml",
+    ],
+    "assets": {},
+    "demo": [],
+    "images": [],
+    "application": False,
+    "installable": True,
+    "auto_install": False,
+}

spp_encryption/models/__init__.py

@@ -0,0 +1 @@
+from . import encryption_provider

spp_encryption/models/encryption_provider.py

@@ -0,0 +1,85 @@
+import json
+import uuid
+
+from jwcrypto import jwe, jwk, jwt
+from jwcrypto.common import json_decode, json_encode
+from jwcrypto.jws import InvalidJWSSignature
+
+from odoo import fields, models
+
+
+class JWCryptoEncryptionProvider(models.Model):
+    _inherit = "g2p.encryption.provider"
+
+    type = fields.Selection(selection_add=[("jwcrypto", "JWCrypto")])
+
+    jwcrypto_key = fields.Char(help="JWK key in JSON format for encryption, decryption, signing, and verification")
+
+    def _get_jwk_key(self):
+        self.ensure_one()
+        if not self.jwcrypto_key:
+            raise ValueError("JWCrypto key is not set.")
+        return jwk.JWK.from_json(self.jwcrypto_key)
+
+    def encrypt_data_jwcrypto(self, data: bytes, **kwargs) -> bytes:
+        self.ensure_one()
+        key = self._get_jwk_key()
+        enc = jwe.JWE(data, json_encode({"alg": "RSA-OAEP", "enc": "A256GCM"}))
+        enc.add_recipient(key)
+        return enc.serialize(compact=True).encode("utf-8")
+
+    def decrypt_data_jwcrypto(self, data: bytes, **kwargs) -> bytes:
+        self.ensure_one()
+        key = self._get_jwk_key()
+        enc = jwe.JWE()
+        enc.deserialize(data.decode("utf-8"), key=key)
+        return enc.payload
+
+    def jwt_sign_jwcrypto(self, data, **kwargs) -> str:
+        self.ensure_one()
+        key = self._get_jwk_key()
+        token = jwt.JWT(header={"alg": "RS256"}, claims=data)
+        token.make_signed_token(key)
+        return token.serialize()
+
+    def jwt_verify_jwcrypto(self, token: str, **kwargs):
+        self.ensure_one()
+        key = self._get_jwk_key()
+        try:
+            received_jwt = jwt.JWT(key=key, jwt=token)
+            verified = True
+        except InvalidJWSSignature:
+            received_jwt = None
+            verified = False
+        return verified, received_jwt
+
+    def get_jwks_jwcrypto(self, **kwargs):
+        self.ensure_one()
+        key = self._get_jwk_key()
+        public_key = key.export_public()
+        jwks = {"keys": [json_decode(public_key)]}
+        return jwks
+
+    def generate_and_store_jwcrypto_key(self, key_type="RSA", size=2048):
+        """
+        Generates a new JWK (JSON Web Key) for the current record and stores it in the `jwcrypto_key` field.
+        :param key_type: The type of key to generate, e.g., 'RSA'.
+        :param size: The size of the key (applies to RSA keys).
+        :return: None
+        """
+        if key_type != "RSA":
+            raise ValueError("Unsupported key type. Currently, only 'RSA' is supported.")
+
+        key = jwk.JWK.generate(kty=key_type, size=size)
+
+        kid = str(uuid.uuid4())
+
+        key_export = key.export()
+
+        export_data = json.loads(key_export)
+        export_data["kid"] = kid
+
+        key_export = json.dumps(export_data)
+
+        # Assuming this method is called on a specific record, not on the model class itself
+        self.jwcrypto_key = key_export

spp_encryption/pyproject.toml

@@ -0,0 +1,3 @@
+[build-system]
+requires = ["whool"]
+build-backend = "whool.buildapi"

spp_encryption/tests/__init__.py

@@ -0,0 +1 @@
+from . import test_encryption_provider

spp_encryption/tests/test_encryption_provider.py

@@ -0,0 +1,116 @@
+from unittest.mock import patch
+
+from odoo.tests.common import TransactionCase
+
+
+class EncryptionProviderTest(TransactionCase):
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.provide_no_type = cls.env["g2p.encryption.provider"].create({"name": "Test Provider"})
+        cls.provider = cls.env["g2p.encryption.provider"].create(
+            {"name": "Test Provider with Type", "type": "jwcrypto"}
+        )
+        cls.provider_with_key = cls.env["g2p.encryption.provider"].create(
+            {"name": "Test Provider with Key", "type": "jwcrypto"}
+        )
+        cls.provider_with_key.generate_and_store_jwcrypto_key()
+
+    def test_get_jwk_key(self):
+        self.assertEqual(self.provider.jwcrypto_key, False)
+        with self.assertRaises(ValueError):
+            self.provider._get_jwk_key()
+
+        expected_keys = ["kty", "kid", "n", "e", "d", "p", "q", "dp", "dq", "qi"]
+        self.assertTrue(all(elem in expected_keys for elem in self.provider_with_key._get_jwk_key().keys()))
+
+    def test_encrypt_data_jwcrypto(self):
+        with self.assertRaises(NotImplementedError):
+            self.provide_no_type.encrypt_data(b"test")
+
+        with patch.object(type(self.provider), "encrypt_data_jwcrypto", return_value=None) as encrypt_data_jwcrypto:
+            result = self.provider.encrypt_data(b"test")
+            encrypt_data_jwcrypto.assert_called_once()
+
+        result_with_key = self.provider_with_key.encrypt_data(b"test")
+
+        self.assertIsNone(result)
+        self.assertIsNotNone(result_with_key)
+        self.assertIsInstance(result_with_key, bytes)
+
+    def test_decrypt_data_jwcrypto(self):
+        with self.assertRaises(NotImplementedError):
+            self.provide_no_type.decrypt_data(b"test")
+
+        with patch.object(type(self.provider), "decrypt_data_jwcrypto", return_value=None) as decrypt_data_jwcrypto:
+            self.provider.decrypt_data(b"test")
+            decrypt_data_jwcrypto.assert_called_once()
+
+        data = "test"
+        encoded_data = data.encode("utf-8")
+
+        encrypted_data = self.provider_with_key.encrypt_data(encoded_data)
+        decrypted_data = self.provider_with_key.decrypt_data(encrypted_data)
+
+        self.assertEqual(decrypted_data, encoded_data)
+
+    def test_jwt_sign_jwcrypto(self):
+        with self.assertRaises(NotImplementedError):
+            self.provide_no_type.jwt_sign({})
+
+        with patch.object(type(self.provider), "jwt_sign_jwcrypto", return_value=None) as jwt_sign_jwcrypto:
+            self.provider.jwt_sign({})
+            jwt_sign_jwcrypto.assert_called_once()
+
+        data = {"test": "test"}
+        result = self.provider_with_key.jwt_sign(data)
+
+        self.assertIsNotNone(result)
+        self.assertIsInstance(result, str)
+
+    def test_jwt_verify_jwcrypto(self):
+        with self.assertRaises(NotImplementedError):
+            self.provide_no_type.jwt_verify("test")
+
+        with patch.object(type(self.provider), "jwt_verify_jwcrypto", return_value=None) as jwt_verify_jwcrypto:
+            self.provider.jwt_verify("test")
+            jwt_verify_jwcrypto.assert_called_once()
+
+        data = {"test": "test"}
+        token = self.provider_with_key.jwt_sign(data)
+        verified, received_jwt = self.provider_with_key.jwt_verify(token)
+
+        self.assertTrue(verified)
+        self.assertIsNotNone(received_jwt)
+
+    def test_get_jwks_jwcrypto(self):
+        with self.assertRaises(NotImplementedError):
+            self.provide_no_type.get_jwks()
+
+        with patch.object(type(self.provider), "get_jwks_jwcrypto", return_value=None) as get_jwks_jwcrypto:
+            self.provider.get_jwks()
+            get_jwks_jwcrypto.assert_called_once()
+
+        jwks = self.provider_with_key.get_jwks()
+
+        self.assertIsNotNone(jwks)
+        self.assertTrue("keys" in jwks)
+        self.assertIsInstance(jwks["keys"], list)
+        self.assertEqual(len(jwks["keys"]), 1)
+        self.assertIsInstance(jwks["keys"][0], dict)
+
+        expected_keys = ["kty", "kid", "n", "e", "d", "p", "q", "dp", "dq", "qi"]
+        self.assertTrue(all(elem in expected_keys for elem in jwks["keys"][0].keys()))
+
+    def test_generate_and_store_jwcrypto_key(self):
+        with self.assertRaisesRegex(ValueError, "Unsupported key type. Currently, only 'RSA' is supported."):
+            self.provider.generate_and_store_jwcrypto_key(key_type="DSA")
+
+        self.assertIn(self.provider.jwcrypto_key, [False, None])
+
+        self.provider.generate_and_store_jwcrypto_key()
+
+        expected_keys = ["kty", "kid", "n", "e", "d", "p", "q", "dp", "dq", "qi"]
+
+        self.assertIsNotNone(self.provider.jwcrypto_key)
+        self.assertTrue(all(elem in expected_keys for elem in self.provider_with_key._get_jwk_key().keys()))

spp_encryption/views/encryption_provider.xml

@@ -0,0 +1,27 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<odoo>
+    <record id="view_encryption_provider_form" model="ir.ui.view">
+        <field name="name">view_encryption_provider_form</field>
+        <field name="model">g2p.encryption.provider</field>
+        <field name="inherit_id" ref="g2p_encryption.view_encryption_provider_form" />
+        <field name="arch" type="xml">
+            <xpath expr="//group[@name='Base']" position="after">
+                <field name="jwcrypto_key" invisible="1" />
+                <button
+                    name="generate_and_store_jwcrypto_key"
+                    string="Create Jwcrypto Key"
+                    type="object"
+                    class="oe_highlight btn-primary"
+                    invisible="jwcrypto_key"
+                />
+                <button
+                    name="generate_and_store_jwcrypto_key"
+                    string="Update Jwcrypto Key"
+                    type="object"
+                    class="oe_highlight btn-primary"
+                    invisible="not jwcrypto_key"
+                />
+            </xpath>
+        </field>
+    </record>
+</odoo>

spp_openid_vci/__init__.py

@@ -0,0 +1,2 @@
+from . import models
+from . import wizard