feat(jumbf-parser): #46: add manifest parsing functionality

.gitignore

@@ -4,6 +4,8 @@ __pycache__
 build
 dist
 **/signed_*
+# This file is needed for unit tests
+!tests/test_files/signed_signed_test_doc.pdf
 .vscode/launch.json
 .env
 **/.ipynb_checkpoints

c2pie/c2pa_parsing/jumbf_parsing.py

@@ -0,0 +1,106 @@
+from __future__ import annotations
+
+from c2pie.jumbf_boxes.box import Box, iter_boxes
+from c2pie.jumbf_boxes.constants import (
+    JUMB_TYPE,
+    JUMD_TYPE,
+    LABEL_OFFSET,
+)
+
+
+def jumd_label(payload: bytes) -> str | None:
+    """
+    Extracts the UTF-8 label from a JUMD (description box) payload.
+
+    The label starts at LABEL_OFFSET and is null-terminated.
+    Returns None if the payload is too short or the null terminator is missing.
+    """
+    if len(payload) < LABEL_OFFSET + 1:
+        return None
+
+    raw_label = payload[LABEL_OFFSET:]
+    null_terminator = raw_label.find(b"\x00")
+
+    if null_terminator == -1:
+        return None
+
+    return raw_label[:null_terminator].decode("utf-8")
+
+
+def find_box_by_label(
+    data: bytes,
+    wanted_label: str,
+) -> Box | None:
+    """
+    Searches the JUMBF box tree for the first superbox whose JUMD label
+    matches `wanted_label`.
+
+    Performs a depth-first traversal of all top-level boxes in `data`.
+    Returns the matching superbox or None if not found.
+    """
+    for box in iter_boxes(data):
+        found = _find_in_box(box, wanted_label)
+        if found is not None:
+            return found
+    return None
+
+
+def _find_in_box(
+    box: Box,
+    wanted_label: str,
+) -> Box | None:
+    """
+    Recursively searches inside a single superbox for a JUMD label match.
+
+    Skips non-JUMB boxes and boxes without a leading JUMD child.
+    Returns the matching superbox or None.
+    """
+    if box.get_type() != JUMB_TYPE:
+        return None
+
+    children = list(iter_boxes(box.get_payload()))
+    if not children or children[0].get_type() != JUMD_TYPE:
+        return None
+
+    if jumd_label(children[0].get_payload()) == wanted_label:
+        return box
+
+    for child in children[1:]:
+        found = _find_in_box(child, wanted_label)
+        if found is not None:
+            return found
+
+    return None
+
+
+def get_active_manifest_uuid(manifest_store_bytes: bytes) -> str | None:
+    """
+    Returns the URN label of the active manifest from a raw JUMBF Manifest Store.
+
+    The active manifest is defined as the last manifest box (JUMB with a JUMD label)
+    within the Manifest Store superbox, as per the C2PA specification.
+    Returns None if the data is not a valid JUMBF box or contains no manifests.
+    """
+    try:
+        manifest_store_box, _ = Box.parse_from_bytes(manifest_store_bytes, 0)
+    except ValueError:
+        return None
+
+    if manifest_store_box.get_type() != JUMB_TYPE:
+        return None
+
+    active_manifest_urn: str | None = None
+
+    for child in iter_boxes(manifest_store_box.get_payload()):
+        if child.get_type() != JUMB_TYPE:
+            continue
+
+        manifest = list(iter_boxes(child.get_payload()))
+        if not manifest or manifest[0].get_type() != JUMD_TYPE:
+            continue
+
+        label = jumd_label(manifest[0].get_payload())
+        if label:
+            active_manifest_urn = label
+
+    return active_manifest_urn

c2pie/c2pa_parsing/manifest_extractor.py

@@ -0,0 +1,139 @@
+from __future__ import annotations
+
+import re
+from collections import defaultdict
+from typing import Callable
+
+from c2pie.utils.content_types import C2PA_ContentTypes
+
+_APP11_MARKER = 0xEB
+_CI_MAGIC = b"JP"
+
+# 0xD8 - SOI
+# 0xD9 - EOI
+# 0x01 - TEM marker
+# set(range(0xD0, 0xD8) - restart markers, using in decode
+# This dictionary contains all markers that are not followed by a length byte.
+_STANDALONE_MARKERS = {0xD8, 0xD9, 0x01} | set(range(0xD0, 0xD8))
+
+
+def extract_manifest_store_bytes_from_jpeg(jpeg_bytes: bytes) -> bytes | None:
+    """Returns raw JUMBF ManifestStore bytes from the APP11 segment."""
+
+    """
+    A dictionary containing the sequence number (Z) and bytes of the APP11 
+    segment chunk, grouped by the APP11 segment identifier (EN).
+
+    app11_chunks = {
+        0: (             ; first APP11 segment identifier (EN)
+            0: b'...',   ; first APP11 chunk identifier (Z)
+            1: b'...',   ; second APP11 chunk identifier (Z)
+            ...
+        )
+    }
+
+    For more info see: docs/JPG-structure-overview.md
+    """
+    app11_chunks: dict[int, list[tuple[int, bytes]]] = defaultdict(list)
+
+    i = 2  # Jump forward 2 bytes to skip the SOI marker (0xFF, 0xD8)
+    while i + 3 < len(jpeg_bytes):
+        if jpeg_bytes[i] != 0xFF:
+            next_ff = jpeg_bytes.find(b"\xff", i + 1)
+
+            if next_ff == -1:
+                break
+
+            i = next_ff
+            continue
+
+        marker = jpeg_bytes[i + 1]
+
+        # Exit when the raw image data marker is encountered. There may be cases where the SOS (0xDA)
+        # marker is missing (corrupted image), so we also track the EOI (0xD9) marker
+        if marker == 0xD9 or marker == 0xDA:
+            break
+
+        # If a marker without a length is encountered, jump over it
+        if marker in _STANDALONE_MARKERS:
+            i += 2
+            continue
+
+        # Checking for the existence of bytes of a length after the marker
+        if i + 4 > len(jpeg_bytes):
+            break
+
+        # The segment length does not include marker bytes
+        seg_len = int.from_bytes(jpeg_bytes[i + 2 : i + 4], "big")
+
+        if marker == _APP11_MARKER:
+            chunk_payload = jpeg_bytes[i + 4 : i + 2 + seg_len]
+
+            if len(chunk_payload) >= 8 and chunk_payload[:2] == _CI_MAGIC:
+                en = int.from_bytes(chunk_payload[2:4], "big")
+                z = int.from_bytes(chunk_payload[4:8], "big")
+
+                app11_chunks[en].append((z, chunk_payload[8:]))
+
+        i += 2 + seg_len
+
+    if not app11_chunks:
+        return None
+
+    app11_chunks = dict(sorted(app11_chunks.items()))
+
+    manifest_stores: list[bytes] = []
+    for _, chunks in app11_chunks.items():
+        chunks.sort(key=lambda x: x[0])
+        manifest_store = b"".join(chunk_bytes for _, chunk_bytes in chunks)
+
+        if b"urn:c2pa:" in manifest_store:
+            manifest_stores.append(manifest_store)
+
+    if not manifest_stores:
+        return None
+
+    return manifest_stores[-1]
+
+
+def extract_manifest_store_bytes_from_pdf(pdf_bytes: bytes) -> bytes | None:
+    """Returns raw JUMBF ManifestStore bytes from the LAST C2PA EmbeddedFile stream."""
+
+    # (\d+)  - captures the stream length in bytes (e.g. "4096")
+    # .*?    - skips any additional PDF keys between /Length and stream (non-greedy)
+    # stream - literal keyword marking the start of the binary data
+    # [\r\n]+ - matches line ending after stream (LF, CRLF, or CR)
+    matches = list(
+        re.finditer(
+            rb"/Type /EmbeddedFile /Subtype /application#2Fc2pa /Length (\d+).*?stream[\r\n]+",
+            pdf_bytes,
+            re.DOTALL,
+        )
+    )
+
+    if not matches:
+        return None
+
+    # We retrieve the Manifest Store from the last section, as it contains the current Manifest Store
+    last = matches[-1]
+    # Retrieves the value of the first group (\d+) and converts it to a number
+    length = int(last.group(1))
+    # Saves the byte position after a match
+    start = last.end()
+
+    return pdf_bytes[start : start + length]
+
+
+_EXTRACTORS: dict[C2PA_ContentTypes, Callable[[bytes], bytes | None]] = {
+    C2PA_ContentTypes.jpg: extract_manifest_store_bytes_from_jpeg,
+    C2PA_ContentTypes.jpeg: extract_manifest_store_bytes_from_jpeg,
+    C2PA_ContentTypes.pdf: extract_manifest_store_bytes_from_pdf,
+}
+
+
+def extract_manifest_store_bytes(
+    content_type: C2PA_ContentTypes,
+    raw_data: bytes,
+) -> bytes | None:
+    extractor = _EXTRACTORS[content_type]
+    return extractor(raw_data)

c2pie/jumbf_boxes/box.py

@@ -1,5 +1,11 @@
 # Base jumbf box class
 
+from c2pie.jumbf_boxes.constants import (
+    BYTE_ORDER,
+    HEADER_SIZE,
+    LBOX_SIZE,
+)
+
 
 class Box:
     def __init__(
@@ -26,3 +32,39 @@ def serialize(self):
         t_box = bytes.fromhex(self.t_box)
         l_box = self.l_box.to_bytes(4, "big")
         return l_box + t_box + self.payload
+
+    @classmethod
+    def parse_from_bytes(
+        cls,
+        data: bytes,
+        offset: int = 0,
+    ) -> tuple["Box", int]:
+        if offset + HEADER_SIZE > len(data):
+            raise ValueError("Not enough data for box header")
+
+        l_box = int.from_bytes(
+            data[offset : offset + LBOX_SIZE],
+            BYTE_ORDER,
+        )
+        t_box = data[offset + LBOX_SIZE : offset + HEADER_SIZE].hex()
+
+        # If the length of the box is unknown, the encoders may set the LBox value to 0.
+        # In this case, you must read until the end of the available bytes.
+        if l_box == 0:
+            end = len(data)
+        else:
+            end = offset + l_box
+
+        if end > len(data):
+            raise ValueError("Box length exceeds available data")
+
+        payload = data[offset + HEADER_SIZE : end]
+        box = cls(t_box, payload)
+        return box, end
+
+
+def iter_boxes(data: bytes):
+    offset = 0
+    while offset < len(data):
+        box, offset = Box.parse_from_bytes(data, offset)
+        yield box

c2pie/jumbf_boxes/constants.py

@@ -0,0 +1,18 @@
+# JUMBF Box Header
+LBOX_SIZE = 4  # Length of the LBox field in bytes
+TBOX_SIZE = 4  # Length of the TBox field in bytes
+HEADER_SIZE = LBOX_SIZE + TBOX_SIZE  # Total box header size in bytes
+
+# JUMBF Description Box (jumd)
+CONTENT_TYPE_SIZE = 16  # UUID content type size
+TOGGLE_SIZE = 1  # Description box flags size
+JUMD_MIN_PAYLOAD_SIZE = CONTENT_TYPE_SIZE + TOGGLE_SIZE + 1  # +1 for minimum null-terminator
+LABEL_OFFSET = CONTENT_TYPE_SIZE + TOGGLE_SIZE  # Label offset within the jumd payload
+
+# Box Type Hex Values
+JUMB_TYPE = b"jumb".hex()
+JUMD_TYPE = b"jumd".hex()
+JSON_TYPE = b"json".hex()
+
+# Endianness
+BYTE_ORDER = "big"

c2pie/jumbf_boxes/description_box.py

@@ -1,6 +1,10 @@
 # Description jumbf box class
 
 from c2pie.jumbf_boxes.box import Box
+from c2pie.jumbf_boxes.constants import (
+    CONTENT_TYPE_SIZE,
+    LABEL_OFFSET,
+)
 from c2pie.utils.content_types import jumbf_content_types
 
 
@@ -18,6 +22,29 @@ def __init__(
 
         super().__init__(b"jumd".hex(), payload=payload)
 
+    @classmethod
+    def from_box(
+        cls,
+        box: Box,
+    ) -> "DescriptionBox":
+        payload = box.get_payload()
+        if len(payload) < LABEL_OFFSET + 1:
+            raise ValueError("JUMD payload is too short")
+
+        null_terminator = payload.find(b"\x00", LABEL_OFFSET)
+        if null_terminator == -1:
+            raise ValueError("JUMD label is not null-terminated")
+
+        instance = cls.__new__(cls)
+        instance.t_box = box.get_type()
+        instance.l_box = box.get_length()
+        instance.payload = payload
+
+        instance.content_type = payload[:CONTENT_TYPE_SIZE]
+        instance.toggle = payload[CONTENT_TYPE_SIZE]
+        instance.label = payload[LABEL_OFFSET:null_terminator].decode("utf-8")
+        return instance
+
     def get_label(self) -> str:
         return self.label