feat: remove data policies from files

.changeset/happy-birds-shout.md

@@ -0,0 +1,5 @@
+---
+"evervault-python": minor
+---
+
+remove data policies from file encryption

e2e/encrypt_test.py

@@ -9,7 +9,7 @@
     {"role": "forbid-all", "decryption_should_succeed": False},
     {"role": None, "decryption_should_succeed": True},
 ]
-METADATA_ENCRYPTED_STRING_REGEX = r"((ev(:|%3A))(debug(:|%3A))?((QlJV|TENZ|)(:|%3A))?((number|boolean|string)(:|%3A))?(([A-z0-9+\/=%]+)(:|%3A)){3}(\$|%24))|(((eyJ[A-z0-9+=.]+){2})([\w]{8}(-[\w]{4}){3}-[\w]{12}))"
+METADATA_ENCRYPTED_STRING_REGEX = r"((ev(:|%3A))(debug(:|%3A))?((QkTC|S0lS|)(:|%3A))?((number|boolean|string)(:|%3A))?(([A-z0-9+\/=%]+)(:|%3A)){3}(\$|%24))|(((eyJ[A-z0-9+=.]+){2})([\w]{8}(-[\w]{4}){3}-[\w]{12}))"
 
 
 def check_object_has_strings_with_correct_versions(value):
@@ -178,14 +178,13 @@ def test_encrypt_dict(self, curve, role_and_success):
                 == "Decryption of the provided data is restricted by your current policies. Please check and modify your policies, if needed, to enable decryption in this context."
             )
 
-    @parameterized.expand(generate_combinations(CURVES, ROLES_AND_SUCCESSES))
-    def test_encrypt_file(self, curve, role_and_success):
+    @parameterized.expand(CURVES)
+    def test_encrypt_file(self, curve):
         self.setUp(curve)
-        role = role_and_success["role"]
-        decryption_should_succeed = role_and_success["decryption_should_succeed"]
+        decryption_should_succeed = True
         file = b"hello world"
         try:
-            encrypted = self.evervault.encrypt(file, role)
+            encrypted = self.evervault.encrypt(file)
             decrypted = self.evervault.decrypt(encrypted)
             assert decryption_should_succeed
             assert "hello world" == decrypted

evervault/__init__.py

@@ -12,7 +12,10 @@
 from evervault.http.pcrManager import PcrManager
 from importlib import metadata
 
-__version__ = metadata.version(__package__ or __name__)
+try:
+    __version__ = metadata.version(__package__ or __name__)
+except:
+    __version__ = "0.0.0"
 
 ev_client = None
 _app_uuid = None

evervault/crypto/client.py

@@ -33,17 +33,11 @@ def __init__(self, api_key=None, curve="SECP256K1", max_file_size_in_mb=25):
         self.decoded_team_cage_key = None
         self.compressed_public_key = None
         self.uncompressed_public_key = None
-        self.shared_key = None
+        self.shared_key: bytes | None = None
         self.start_time = int(time.time())
         self.api_key = api_key
-        self.ev_version = self.__base_64_remove_padding(
-            base64.b64encode(bytes(VERSION[self.curve], "utf8")).decode("utf")
-        )
-        self.ev_version_with_metadata = self.__base_64_remove_padding(
-            base64.b64encode(bytes(VERSION_WITH_METADATA[self.curve], "utf8")).decode(
-                "utf"
-            )
-        )
+        self.ev_version = VERSION[self.curve]
+        self.ev_version_with_metadata = VERSION_WITH_METADATA[self.curve]
         self.max_file_size_in_mb = max_file_size_in_mb
         self.max_file_size_in_bytes = max_file_size_in_mb * 1024 * 1024
 
@@ -105,6 +99,25 @@ def __encrypt_set(self, data, role):
                 encrypted_set.add(self.__traverse_and_encrypt(item, role))
         return encrypted_set
 
+    def __create_v2_aad(
+        self, datatype, ephemeral_public_key_bytes, app_public_key_bytes
+    ) -> bytes:
+        datatype_number = 0
+        if datatype == "number":
+            datatype_number = 1
+        elif datatype == "boolean":
+            datatype_number = 2
+
+        version_prefix = 0x00 if self.curve == SECP256K1 else 0x01
+
+        aad = bytearray()
+        aad.append(version_prefix | (datatype_number << 4))
+
+        aad.extend(ephemeral_public_key_bytes)
+        aad.extend(app_public_key_bytes)
+
+        return aad
+
     def __encrypt_string(self, data, role):
         header_type = map_header_type(data)
         coerced_data = self.__coerce_type(data)
@@ -125,7 +138,16 @@ def __encrypt_string(self, data, role):
         if self.curve == SECP256K1 and not has_role:
             encrypted_bytes = aesgcm.encrypt(iv, payload, None)
         else:
-            encrypted_bytes = aesgcm.encrypt(iv, payload, self.decoded_team_cage_key)
+            aad = (
+                self.__create_v2_aad(
+                    header_type,
+                    self.compressed_public_key,
+                    self.decoded_team_cage_key,
+                )
+                if has_role
+                else self.decoded_team_cage_key
+            )
+            encrypted_bytes = aesgcm.encrypt(iv, payload, aad)
 
         return self.__format(
             header_type,
@@ -168,18 +190,15 @@ def __encrypt_file(self, data, role):
             raise ExceededMaxFileSizeError(
                 f"File size must be less than {self.max_file_size_in_mb}MB"
             )
+
         iv = token_bytes(12)
         aesgcm = AESGCM(self.shared_key)
 
         encrypted_bytes = None
         encrypted_metadata = None
 
         if role is not None:
-            metadata = self.__generate_metadata(role)
-            encrypted_metadata = aesgcm.encrypt(
-                iv, metadata, self.decoded_team_cage_key
-            )
-            encrypted_bytes = aesgcm.encrypt(iv, data, self.decoded_team_cage_key)
+            raise EvervaultError("Data roles are not supported for file encryption")
         elif self.curve == SECP256K1:
             encrypted_bytes = aesgcm.encrypt(iv, data, None)
         else:
@@ -208,17 +227,10 @@ def __format(self, header, iv, public_key, encrypted_payload, has_role=False):
     def __format_file(self, iv, public_key, encrypted_metadata, encrypted_bytes):
         encrypted_file_identifier = bytes(b"\x25\x45\x56\x45\x4e\x43")
         flags = bytes(b"\00")
-        if encrypted_metadata is not None:
-            version_number = bytes(b"\04") if self.curve == SECP256K1 else bytes(b"\05")
-            metadata_offset = len(encrypted_metadata).to_bytes(2, byteorder="little")
-            offset_to_data = (55 + 2 + len(encrypted_metadata)).to_bytes(
-                2, byteorder="little"
-            )
-        else:
-            version_number = bytes(b"\02") if self.curve == SECP256K1 else bytes(b"\03")
-            metadata_offset = bytes(b"")
-            encrypted_metadata = bytes(b"")
-            offset_to_data = bytes([55, 00])
+        version_number = bytes(b"\02") if self.curve == SECP256K1 else bytes(b"\03")
+        metadata_offset = bytes(b"")
+        encrypted_metadata = bytes(b"")
+        offset_to_data = bytes([55, 00])
 
         file_bytes = (
             encrypted_file_identifier
@@ -270,7 +282,7 @@ def __derive_shared_key(self, has_role=False):
                 return self.__generate_shared_key(has_role)
             return self.shared_key
 
-    def __generate_shared_key(self, has_role):
+    def __generate_shared_key(self, has_role) -> bytes:
         generated_key = ec.generate_private_key(CURVES[self.curve]())
         public_key = generated_key.public_key()
         self.compressed_public_key = public_key.public_bytes(

evervault/crypto/version.py

@@ -1,4 +1,4 @@
 EV_VERSION = "DUB"
 
-VERSION = {"SECP256R1": "NOC", "SECP256K1": "DUB"}
-VERSION_WITH_METADATA = {"SECP256R1": "LCY", "SECP256K1": "BRU"}
+VERSION = {"SECP256R1": "Tk9D", "SECP256K1": "RFVC"}
+VERSION_WITH_METADATA = {"SECP256R1": "QkTC", "SECP256K1": "S0lS"}

tests/test_evervault.py

@@ -132,45 +132,45 @@ def test_encrypt_dicts(self, curve, role, mock_request):
         assert type(encrypted_data["dict"]) == dict
         assert self.__is_evervault_string(encrypted_data["dict"]["subnumber"], "number")
 
-    @parameterized.expand(generate_combinations(CURVES.keys(), ROLES))
+    @parameterized.expand(CURVES.keys())
     @requests_mock.Mocker()
-    def test_encrypt_files(self, curve, role, mock_request):
+    def test_encrypt_files(self, curve, mock_request):
         self.setUp(curve)
         self.mock_fetch_cage_key(mock_request)
 
         test_payload = b"\00\01\03"
-        encrypted_data = self.evervault.encrypt(test_payload, role)
+        encrypted_data = self.evervault.encrypt(test_payload)
         assert self.__is_evervault_file(encrypted_data)
 
         # Check that curve is set correctly
         if curve == "SECP256K1":
-            assert encrypted_data[6:7] == b"\04"
+            assert encrypted_data[6:7] == b"\02"
         else:
-            assert encrypted_data[6:7] == b"\05"
+            assert encrypted_data[6:7] == b"\03"
 
         # Re-calculate the crc32 and ensure it matches
         crc32 = binascii.crc32(encrypted_data[:-4])
         assert encrypted_data[-4:] == crc32.to_bytes(4, byteorder="little")
 
-    @parameterized.expand(generate_combinations(CURVES.keys(), ROLES))
+    @parameterized.expand(CURVES.keys())
     @requests_mock.Mocker()
-    def test_encrypt_files_with_bytearray(self, curve, role, mock_request):
+    def test_encrypt_files_with_bytearray(self, curve, mock_request):
         self.setUp(curve)
         self.mock_fetch_cage_key(mock_request)
 
         test_payload = bytearray(10)
-        encrypted_data = self.evervault.encrypt(test_payload, role)
+        encrypted_data = self.evervault.encrypt(test_payload)
         assert self.__is_evervault_file(encrypted_data)
 
         # Check that curve is set correctly
         if curve == "SECP256K1":
-            assert encrypted_data[6:7] == b"\04"
+            assert encrypted_data[6:7] == b"\02"
         else:
-            assert encrypted_data[6:7] == b"\05"
+            assert encrypted_data[6:7] == b"\03"
 
-    @parameterized.expand(generate_combinations(CURVES.keys(), ROLES))
+    @parameterized.expand(CURVES.keys())
     @requests_mock.Mocker()
-    def test_encrypt_large_files_throws_exception(self, curve, role, mock_request):
+    def test_encrypt_large_files_throws_exception(self, curve, mock_request):
         self.setUp(curve)
         self.mock_fetch_cage_key(mock_request)
 
@@ -179,24 +179,33 @@ def test_encrypt_large_files_throws_exception(self, curve, role, mock_request):
             ExceededMaxFileSizeError, self.evervault.encrypt, test_payload
         )
 
-    @parameterized.expand(generate_combinations(CURVES.keys(), ROLES))
+    @parameterized.expand(CURVES.keys())
+    @requests_mock.Mocker()
+    def test_encrypt_file_with_role_throws_exception(self, curve, mock_request):
+        self.setUp(curve)
+        self.mock_fetch_cage_key(mock_request)
+
+        test_payload = bytes(bytearray(10 * 1024 * 1024))
+        self.assertRaises(EvervaultError, self.evervault.encrypt, test_payload, "role")
+
+    @parameterized.expand(CURVES.keys())
     @requests_mock.Mocker()
     def test_encrypt_large_files_succeeds_with_max_size_override(
-        self, curve, role, mock_request
+        self, curve, mock_request
     ):
         os.environ["EV_MAX_FILE_SIZE_IN_MB"] = "30"
         self.setUp(curve)
         self.mock_fetch_cage_key(mock_request)
 
         test_payload = bytes(bytearray(26 * 1024 * 1024))
-        encrypted_data = self.evervault.encrypt(test_payload, role)
+        encrypted_data = self.evervault.encrypt(test_payload)
         assert self.__is_evervault_file(encrypted_data)
 
         # Check that curve is set correctly
         if curve == "SECP256K1":
-            assert encrypted_data[6:7] == b"\04"
+            assert encrypted_data[6:7] == b"\02"
         else:
-            assert encrypted_data[6:7] == b"\05"
+            assert encrypted_data[6:7] == b"\03"
 
     @parameterized.expand(CURVES)
     @requests_mock.Mocker()