feat: support KMS 4.5.0

Author: Manuthor

.github/workflows/ci.yml

@@ -31,7 +31,7 @@ jobs:
   test:
     services:
       kms:
-        image: ghcr.io/cosmian/kms:4.4.3
+        image: ghcr.io/cosmian/kms:4.5.0
         env:
           COSMIAN_SERVER_URL: http://localhost:9998
           KMS_PUBLIC_PATH: /tmp
@@ -94,7 +94,7 @@ jobs:
       extension: so
       destination: linux-x86-64
       os: ubuntu-20.04
-      kms-version: 4.4.3
+      kms-version: 4.5.0
       findex-cloud-version: 0.1.0
       copy_fresh_build: false
       copy_regression_files: |
@@ -107,7 +107,7 @@ jobs:
     with:
       branch: develop
       target: wasm32-unknown-unknown
-      kms-version: 4.4.3
+      kms-version: 4.5.0
       findex-cloud-version: 0.1.0
       copy_fresh_build: false
       copy_regression_files: |

CHANGELOG.md

@@ -2,6 +2,12 @@
 
 All notable changes to this project will be documented in this file.
 
+## [4.1.0] - 2023-08-28
+
+### Features
+
+- Support KMS v4.5.0 with tagging support
+
 ## [4.0.2] - 2023-07-17
 
 ### Miscellaneous Tasks

README.md

@@ -94,6 +94,7 @@ This table shows the minimum version correspondence between the various componen
 
 | `cloudproof_py` | CoverCrypt | Findex      | KMS   |
 | --------------- | ---------- | ----------- | ----- |
+| >=4.1.0         | 11.0.0     | 4.0.0,4.0.1 | 4.5.0 |
 | >=4.0.2         | 11.0.0     | 4.0.0,4.0.1 | 4.4.3 |
 | >=4.0.1         | 11.0.0     | 4.0.0,4.0.1 | 4.4.2 |
 | >=4.0.0         | 11.0.0     | 4.0.0,4.0.1 | 4.3.3 |

examples/cover_crypt/main.py

@@ -50,23 +50,23 @@ async def main(use_kms: bool = True):
         # Messages encryption
         protected_mkg_data = b"protected_mkg_message"
         protected_mkg_ciphertext = await kms_client.cover_crypt_encryption(
-            public_key_uid,
             "Department::MKG && Security Level::Protected",
             protected_mkg_data,
+            public_key_uid,
         )
 
-        topSecret_mkg_data = b"top_secret_mkg_message"
-        topSecret_mkg_ciphertext = await kms_client.cover_crypt_encryption(
-            public_key_uid,
+        top_secret_mkg_data = b"top_secret_mkg_message"
+        top_secret_mkg_ciphertext = await kms_client.cover_crypt_encryption(
             "Department::MKG && Security Level::Top Secret",
-            topSecret_mkg_data,
+            top_secret_mkg_data,
+            public_key_uid,
         )
 
         protected_fin_data = b"protected_fin_message"
         protected_fin_ciphertext = await kms_client.cover_crypt_encryption(
-            public_key_uid,
             "Department::FIN && Security Level::Protected",
             protected_fin_data,
+            public_key_uid,
         )
 
         # Generating user keys
@@ -86,15 +86,15 @@ async def main(use_kms: bool = True):
 
         # Decryption with the right access policy
         protected_mkg_plaintext, _ = await kms_client.cover_crypt_decryption(
-            confidential_mkg_user_uid, protected_mkg_ciphertext
+            protected_mkg_ciphertext, confidential_mkg_user_uid
         )
         assert protected_mkg_plaintext == protected_mkg_data
 
         # Decryption without the right access will fail
         try:
             # will throw
             await kms_client.cover_crypt_decryption(
-                confidential_mkg_user_uid, topSecret_mkg_ciphertext
+                top_secret_mkg_ciphertext, confidential_mkg_user_uid
             )
         except Exception as e:
             # ==> the user is not be able to decrypt
@@ -103,7 +103,7 @@ async def main(use_kms: bool = True):
         try:
             # will throw
             await kms_client.cover_crypt_decryption(
-                confidential_mkg_user_uid, protected_fin_ciphertext
+                protected_fin_ciphertext, confidential_mkg_user_uid
             )
         except Exception as e:
             # ==> the user is not be able to decrypt
@@ -113,17 +113,17 @@ async def main(use_kms: bool = True):
         # of all Security Level within the right Department
 
         protected_mkg_plaintext2, _ = await kms_client.cover_crypt_decryption(
-            topSecret_mkg_fin_user_uid, protected_mkg_ciphertext
+            protected_mkg_ciphertext, topSecret_mkg_fin_user_uid
         )
         assert protected_mkg_plaintext2 == protected_mkg_data
 
         topSecret_mkg_plaintext, _ = await kms_client.cover_crypt_decryption(
-            topSecret_mkg_fin_user_uid, topSecret_mkg_ciphertext
+            top_secret_mkg_ciphertext, topSecret_mkg_fin_user_uid
         )
-        assert topSecret_mkg_plaintext == topSecret_mkg_data
+        assert topSecret_mkg_plaintext == top_secret_mkg_data
 
         protected_fin_plaintext, _ = await kms_client.cover_crypt_decryption(
-            topSecret_mkg_fin_user_uid, protected_fin_ciphertext
+            protected_fin_ciphertext, topSecret_mkg_fin_user_uid
         )
         assert protected_fin_plaintext == protected_fin_data
 
@@ -132,29 +132,29 @@ async def main(use_kms: bool = True):
         # rotate MKG attribute
         # all active keys will be rekeyed automatically
         await kms_client.rotate_cover_crypt_attributes(
-            private_key_uid, ["Department::MKG"]
+            ["Department::MKG"], private_key_uid
         )
 
         # New confidential marketing message
 
         confidential_mkg_data = b"confidential_secret_mkg_message"
         confidential_mkg_ciphertext = await kms_client.cover_crypt_encryption(
-            public_key_uid,
             "Department::MKG && Security Level::Confidential",
             confidential_mkg_data,
+            public_key_uid,
         )
 
         # Decrypting the messages with the rekeyed key
 
         # decrypting the "old" `protected marketing` message
         old_protected_mkg_plaintext, _ = await kms_client.cover_crypt_decryption(
-            confidential_mkg_user_uid, protected_mkg_ciphertext
+            protected_mkg_ciphertext, confidential_mkg_user_uid
         )
         assert old_protected_mkg_plaintext == protected_mkg_data
 
         # decrypting the "new" `confidential marketing` message
         new_confidential_mkg_plaintext, _ = await kms_client.cover_crypt_decryption(
-            confidential_mkg_user_uid, confidential_mkg_ciphertext
+            confidential_mkg_ciphertext, confidential_mkg_user_uid
         )
         assert new_confidential_mkg_plaintext == confidential_mkg_data
 
@@ -173,12 +173,12 @@ async def main(use_kms: bool = True):
             protected_mkg_data,
         )
 
-        topSecret_mkg_data = b"top_secret_mkg_message"
-        topSecret_mkg_ciphertext = cover_crypt.encrypt(
+        top_secret_mkg_data = b"top_secret_mkg_message"
+        top_secret_mkg_ciphertext = cover_crypt.encrypt(
             policy,
             "Department::MKG && Security Level::Top Secret",
             public_key,
-            topSecret_mkg_data,
+            top_secret_mkg_data,
         )
 
         protected_fin_data = b"protected_fin_message"
@@ -211,7 +211,7 @@ async def main(use_kms: bool = True):
         # Decryption without the right access will fail
         try:
             # will throw
-            cover_crypt.decrypt(confidential_mkg_user_key, topSecret_mkg_ciphertext)
+            cover_crypt.decrypt(confidential_mkg_user_key, top_secret_mkg_ciphertext)
         except Exception as e:
             # ==> the user is not be able to decrypt
             print("Expected error:", e)
@@ -232,9 +232,9 @@ async def main(use_kms: bool = True):
         assert protected_mkg_plaintext2 == protected_mkg_data
 
         topSecret_mkg_plaintext, _ = cover_crypt.decrypt(
-            topSecret_mkg_fin_user_key, topSecret_mkg_ciphertext
+            topSecret_mkg_fin_user_key, top_secret_mkg_ciphertext
         )
-        assert topSecret_mkg_plaintext == topSecret_mkg_data
+        assert topSecret_mkg_plaintext == top_secret_mkg_data
 
         protected_fin_plaintext, _ = cover_crypt.decrypt(
             topSecret_mkg_fin_user_key, protected_fin_ciphertext

pyproject.toml

@@ -7,7 +7,7 @@ cloudproof_py = ["py.typed", "*.pyi"]
 
 [project]
 name = "cloudproof_py"
-version = "4.0.2"
+version = "4.1.0"
 authors = [{ name = "Cosmian Tech", email = "tech@cosmian.com" }]
 description = "Python library for Cosmian Cloudproof"
 readme = "README.md"
@@ -20,7 +20,7 @@ dependencies = [
   "cloudproof-anonymization >= 0.1.0, < 1.0.0",
   "cover-crypt >= 11.0.0, < 12.0.0",
   "findex >= 4.0.0, < 5.0.0",
-  "cosmian-kms >= 4.4.3, < 5.0.0",
+  "cosmian-kms >= 4.5.0, < 5.0.0",
   "cloudproof-fpe >= 0.1.0, < 1.0.0",
 ]
 

src/cloudproof_py/kms/__init__.py

@@ -29,42 +29,49 @@ async def create_cover_crypt_master_key_pair(
         return await super().create_cover_crypt_master_key_pair(policy)
 
     async def rotate_cover_crypt_attributes(
-        self, master_secret_key_identifier: str, attributes: List[Union[Attribute, str]]
+        self,
+        attributes: List[Union[Attribute, str]],
+        master_secret_key_identifier: Optional[str],
+        tags: Optional[List[str]] = None,
     ) -> Tuple[str, str]:
         """Rotate the given policy attributes. This will rekey in the KMS:
             - the Master Keys
             - all User Decryption Keys that contain one of these attributes in their policy.
 
         Args:
-            master_secret_key_identifier (str): master secret key UID
             attributes (List[Union[Attribute, str]]): attributes to rotate e.g. ["Department::HR"]
+            master_secret_key_identifier (str): master secret key UID
+            tags: (Optional[List[str][]) tags to retrieve the master secret key if it the id is not satisfied
 
         Returns:
             Tuple[str, str]: (Public key UID, Master secret key UID)
         """
         return await super().rotate_cover_crypt_attributes(
-            master_secret_key_identifier,
             [
                 attr.to_string() if isinstance(attr, Attribute) else attr
                 for attr in attributes
             ],
+            master_secret_key_identifier,
+            tags,
         )
 
     async def cover_crypt_encryption(
         self,
-        public_key_identifier: str,
-        access_policy_str: str,
+        encryption_policy_str: str,
         data: bytes,
+        public_key_identifier: Optional[str],
+        tags: Optional[List[str]] = None,
         header_metadata: Optional[bytes] = None,
         authentication_data: Optional[bytes] = None,
     ) -> bytes:
         """Hybrid encryption. Concatenates the encrypted header and the symmetric
         ciphertext.
 
         Args:
-            public_key_identifier (str): identifier of the public key
-            access_policy_str (str): the access policy to use for encryption
+            encryption_policy_str (str): the access policy to use for encryption
             data (bytes): data to encrypt
+            public_key_identifier (str): identifier of the public key
+            tags: (Optional[List[str]]): tags to use to find the public key
             header_metadata (Optional[bytes]): additional data to encrypt in the header
             authentication_data (Optional[bytes]): authentication data to use in the encryption
 
@@ -73,32 +80,35 @@ async def cover_crypt_encryption(
         """
         return bytes(
             await super().cover_crypt_encryption(
-                public_key_identifier,
-                access_policy_str,
+                encryption_policy_str,
                 data,
+                public_key_identifier,
+                tags,
                 header_metadata,
                 authentication_data,
             )
         )
 
     async def cover_crypt_decryption(
         self,
-        user_key_identifier: str,
         encrypted_data: bytes,
+        user_key_identifier: Optional[str],
+        tags: Optional[List[str]] = None,
         authentication_data: Optional[bytes] = None,
     ) -> Tuple[bytes, bytes]:
         """Hybrid decryption.
 
         Args:
-            user_key_identifier (str): user secret key identifier
             encrypted_data (bytes): encrypted header || symmetric ciphertext
+            user_key_identifier (str): user secret key identifier
+            tags: (Optional[List[str]]): tags to use to find the user key
             authentication_data (Optional[bytes]): authentication data to use in the decryption
 
         Returns:
             Future[Tuple[bytes, bytes]]: (plaintext bytes, header metadata bytes)
         """
         plaintext, header = await super().cover_crypt_decryption(
-            user_key_identifier, encrypted_data, authentication_data
+            encrypted_data, user_key_identifier, tags, authentication_data
         )
         return bytes(plaintext), bytes(header)