feat: #80: investigation and correction of hash mismatches during the signing process

README.md

@@ -8,9 +8,9 @@
 
 [![Linting](https://github.com/TourmalineCore/c2pie/actions/workflows/lint-on-pull-request.yml/badge.svg?branch=develop)](https://github.com/TourmalineCore/c2pie/actions/workflows/lint-on-pull-request.yml)
 [![c2pa](https://img.shields.io/badge/c2pa-v1.4-seagreen.svg)](https://c2pa.org/)
-[![coverage](https://img.shields.io/badge/e2e_coverage-71.13%25-yellow)](https://github.com/TourmalineCore/c2pie/actions/workflows/calculate-tests-coverage-on-pull-request.yml)
-[![coverage](https://img.shields.io/badge/units_coverage-79.65%25-yellow)](https://github.com/TourmalineCore/c2pie/actions/workflows/calculate-tests-coverage-on-pull-request.yml)
-[![coverage](https://img.shields.io/badge/full_coverage-91.80%25-forestgreen)](https://github.com/TourmalineCore/c2pie/actions/workflows/calculate-tests-coverage-on-pull-request.yml)
+[![coverage](https://img.shields.io/badge/e2e_coverage-71.15%25-yellow)](https://github.com/TourmalineCore/c2pie/actions/workflows/calculate-tests-coverage-on-pull-request.yml)
+[![coverage](https://img.shields.io/badge/units_coverage-85.01%25-olivedrab)](https://github.com/TourmalineCore/c2pie/actions/workflows/calculate-tests-coverage-on-pull-request.yml)
+[![coverage](https://img.shields.io/badge/full_coverage-92.51%25-forestgreen)](https://github.com/TourmalineCore/c2pie/actions/workflows/calculate-tests-coverage-on-pull-request.yml)
 [![latest](https://img.shields.io/pypi/v/c2pie?label=latest&colorB=fc8021)](https://pypi.org/project/c2pie/)
 
 <br>

c2pie/c2pa/assertion.py

@@ -37,7 +37,12 @@ def __init__(
         if not content_boxes:
             payload = self.get_payload_from_schema()
             box_type_hex = get_assertion_content_box_type(self.type)
-            content_boxes = [ContentBox(box_type=box_type_hex, payload=payload)]
+            content_boxes = [
+                ContentBox(
+                    box_type=box_type_hex,
+                    payload=payload,
+                )
+            ]
 
         super().__init__(
             content_type=get_assertion_content_type(self.type),
@@ -64,56 +69,71 @@ class HashDataAssertion(Assertion):
 
     def __init__(
         self,
-        cai_offset: int,
         hashed_data: bytes,
-        additional_exclusions: list[dict[str, int]] | None = None,
     ):
-        exclusions: list[dict[str, int]] = [
-            {
-                "start": cai_offset,
-                "length": 65535,
-            },
-        ]
-
-        if additional_exclusions:
-            exclusions.extend(additional_exclusions)
+        exclusions: list[dict[str, int]] = []
 
         schema: dict[str, Any] = {
-            "name": "jumbf manifest",
             "exclusions": exclusions,
             "alg": "sha256",
             "hash": hashed_data,
-            "pad": [],
+            # The specification recommends setting the pad to at least 16 bytes. We use 64 bytes
+            # to allow for some extra space before the 23-byte limit is exceeded, since otherwise
+            # the CBOR header of the pad field would be reduced by 1 byte.
+            "pad": b"\x00" * 64,
         }
-        super().__init__(C2PA_AssertionTypes.data_hash, schema)
 
-    def set_hash_data_length(
+        super().__init__(
+            C2PA_AssertionTypes.data_hash,
+            schema,
+        )
+
+    def add_full_c2pa_structure_exclusion(
         self,
+        offset: int,
         length: int,
     ) -> None:
-        if self.schema.get("name") != "jumbf manifest":
-            raise ValueError("c2pa.hash.data: jumbf manifest is missing")
+        exclusions = self.schema["exclusions"]
+        previous_exclusion_length = len(cbor_to_bytes(exclusions))
+
+        self.schema["exclusions"].extend(
+            [
+                {
+                    "start": offset,
+                    "length": length,
+                },
+            ]
+        )
+
+        # NOTE: If the number of exclusions exceeds 23, an additional length byte
+        # will be added to the CBOR header of serialized exclusions array. This byte
+        # is included in the recalculation of the serialized exclusions.
+        current_exclusion_length = len(cbor_to_bytes(exclusions))
+
+        difference = previous_exclusion_length - current_exclusion_length
 
-        exclusions = self.schema.get("exclusions", [])
+        if -difference > len(self.schema["pad"]):
+            raise ValueError("Difference in length exceeds the predefined pad")
 
-        if not exclusions:
-            raise ValueError("c2pa.hash.data: exclusions are missing")
+        # If the pad is less than 24 bytes the size of the cbor header
+        # will change during conversion to cbor and will occupy less than 2 bytes.
+        updated_pad_length = len(self.schema["pad"]) + difference
 
-        exclusions[0]["length"] = int(length)
+        # If a CBOR overflow is not handled, the extra length byte that
+        # would be added in this case will not be taken into account.
+        if updated_pad_length < 24:
+            updated_pad_length -= 1
+
+        self.schema["pad"] = b"\x00" * updated_pad_length
 
         payload = self.get_payload_from_schema()
-        if self.content_boxes:
-            self.content_boxes[0] = ContentBox(
+
+        self.content_boxes = [
+            ContentBox(
                 box_type=get_assertion_content_box_type(self.type),
                 payload=payload,
             )
-        else:
-            self.content_boxes = [
-                ContentBox(
-                    box_type=get_assertion_content_box_type(self.type),
-                    payload=payload,
-                )
-            ]
+        ]
 
         self.sync_payload()
 

c2pie/c2pa/assertion_store.py

@@ -20,11 +20,16 @@ def __init__(
     def get_assertions(self) -> list:
         return self.assertions
 
-    def set_hash_data_length(
+    def add_full_c2pa_structure_exclusion(
         self,
+        offset: int,
         length: int,
     ) -> None:
         for assertion in self.assertions:
             if assertion.type == C2PA_AssertionTypes.data_hash:
-                assertion.set_hash_data_length(length)
+                assertion.add_full_c2pa_structure_exclusion(
+                    offset,
+                    length,
+                )
+
         self.sync_payload()

c2pie/c2pa/claim_signature.py

@@ -64,6 +64,8 @@ def __init__(
         self.require_tsa = require_tsa
         self.tsa_log_dir = tsa_log_dir
 
+        self.serialized_cose_sign1_length = 0
+
         content_boxes = self._generate_payload()
 
         super().__init__(
@@ -129,10 +131,49 @@ def _generate_unprotected_header(self, serialized_sig_structure: bytes) -> bytes
                         },
                     ],
                 },
+                # The specification recommends setting the pad to at least 16 bytes. We use 64 bytes
+                # to allow for some extra space before the 23-byte limit is exceeded, since otherwise
+                # the CBOR header of the pad field would be reduced by 1 byte.
+                "pad": b"\x00" * 8,
             }
 
         return unprotected_header
 
+    def serialize_cose_sign1_tagged_with_alignment(
+        self,
+        cose_sign1: list,
+    ) -> bytes:
+        cose_sign1_tagged_cbor = cbor2.dumps(
+            cbor2.CBORTag(18, cose_sign1),
+            canonical=True,
+        )
+
+        # The length of a TSA token can be variable. To ensure that a new token does not exceed
+        # the exclusion boundary for the C2PA structure, we need to align the length of
+        # the Claim Signature using the pad field, similar to the Data Hash Assertion.
+        if self.serialized_cose_sign1_length == 0:
+            self.serialized_cose_sign1_length = len(cose_sign1_tagged_cbor)
+        elif self.serialized_cose_sign1_length != len(cose_sign1_tagged_cbor):
+            difference = self.serialized_cose_sign1_length - len(cose_sign1_tagged_cbor)
+
+            if -difference > len(cose_sign1[1]["pad"]):
+                raise ValueError("Difference in length exceeds the predefined pad")
+
+            updated_pad_length = len(cose_sign1[1]["pad"]) + difference
+
+            # If a CBOR overflow is not handled, the extra length byte that
+            # would be added in this case will not be taken into account.
+            if updated_pad_length > 23:
+                updated_pad_length += 1
+
+            cose_sign1[1]["pad"] = b"\x00" * updated_pad_length
+            cose_sign1_tagged_cbor = cbor2.dumps(
+                cbor2.CBORTag(18, cose_sign1),
+                canonical=True,
+            )
+
+        return cose_sign1_tagged_cbor
+
     def _create_cose_sign1_tagged(self) -> bytes:
         """
         COSE_Sign1 = [
@@ -179,4 +220,6 @@ def _create_cose_sign1_tagged(self) -> bytes:
 
         cose_sign1 = [serialized_protected_header, unprotected_header, None, signature]
 
-        return cbor2.dumps(cbor2.CBORTag(18, cose_sign1), canonical=True)
+        cose_sign1_tagged_cbor = self.serialize_cose_sign1_tagged_with_alignment(cose_sign1)
+
+        return cose_sign1_tagged_cbor

c2pie/c2pa/manifest.py

@@ -44,16 +44,20 @@ def get_assertions(self):
             return self.assertion_store.get_assertions()
         return
 
-    def set_hash_data_length(
+    def add_full_c2pa_structure_exclusion(
         self,
+        offset: int,
         length: int,
     ):
         """
         Updates the length of exceptions in HashData, reassembles Claim (assertion hashes)
         and ClaimSignature (COSE Sign1 detached over Claim CBOR).
         """
         if self.assertion_store and self.claim and self.claim_signature:
-            self.assertion_store.set_hash_data_length(length)
+            self.assertion_store.add_full_c2pa_structure_exclusion(
+                offset,
+                length,
+            )
             self.claim.set_assertion_store(self.assertion_store)
             self.claim_signature.set_claim(self.claim)
 

c2pie/c2pa/manifest_store.py

@@ -25,11 +25,15 @@ def __init__(
     def sync_payload(self):
         super().sync_payload()
 
-    def set_hash_data_length_for_all(
+    def add_full_c2pa_structure_exclusion(
         self,
+        offset: int,
         length: int,
     ) -> None:
-        self.manifests[-1].set_hash_data_length(length)
+        self.manifests[-1].add_full_c2pa_structure_exclusion(
+            offset,
+            length,
+        )
 
         super().sync_payload()
 

c2pie/c2pa_injection/jpg_injection.py

@@ -1,3 +1,13 @@
+from c2pie.c2pa.manifest_store import ManifestStore
+
+# JPG_SEGMENT_MAX_PAYLOAD_LENGTH =
+#   65535 (max segment length)
+#   - 2 (bytes of length)
+#   - 2 (bytes of CI)
+#   - 2 (bytes of EN)
+#   - 4 (bytes of Z)
+#   - 4 (bytes of LBox)
+#   - 4 (bytes of TBox)
 JPG_SEGMENT_MAX_PAYLOAD_LENGTH = 65517
 
 
@@ -98,3 +108,36 @@ def serialize(self):
 
         self.serialized_length = len(serialized_storage_data)
         return serialized_storage_data
+
+
+def create_and_serialize_app11_storage(
+    manifest_store: ManifestStore,
+) -> bytes:
+    serialized_manifest_store = manifest_store.serialize()
+
+    app11_storage = JpgSegmentApp11Storage(
+        app11_segment_box_length=manifest_store.get_length(),
+        app11_segment_box_type=manifest_store.get_type(),
+        payload=serialized_manifest_store,
+    )
+
+    return app11_storage.serialize()
+
+
+def emplace_manifest_into_jpeg(
+    content_bytes: bytes,
+    manifest_store: ManifestStore,
+    c2pa_offset: int,
+) -> bytes:
+    serialized_app11_storage = create_and_serialize_app11_storage(manifest_store)
+
+    serialized_app11_storage_length = len(serialized_app11_storage)
+
+    manifest_store.add_full_c2pa_structure_exclusion(
+        c2pa_offset,
+        serialized_app11_storage_length,
+    )
+
+    tail = create_and_serialize_app11_storage(manifest_store)
+
+    return content_bytes[:c2pa_offset] + tail + content_bytes[c2pa_offset:]