[ON-HOLD] feat: Add some filetype guessing capabilities

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "c2pa-python"
-version = "0.12.0"
+version = "0.12.1"
 requires-python = ">=3.10"
 description = "Python bindings for the C2PA Content Authenticity Initiative (CAI) library"
 readme = { file = "README.md", content-type = "text/markdown" }

src/c2pa/c2pa.py

@@ -150,6 +150,75 @@ class C2paSigningAlg(enum.IntEnum):
             ctypes.c_ubyte), ctypes.c_size_t)
 
 
+def _guess_mime_type_using_magic_number(file_path: Union[str, Path]) -> Optional[tuple[str, str]]:
+    """Guess MIME type by reading file header bytes.
+    Currently supports a limited set of files, as this is best effort.
+    You should consider adding extensions to filepaths to ensure we don't
+    have to guess too much.
+
+    Args:
+        file_path: Path to the file to check
+    Returns:
+        Tuple of (extension, mime_type) if detected, None otherwise
+    """
+    try:
+        with open(file_path, 'rb') as f:
+            # Read first 1024 bytes to check for file signatures
+            header = f.read(1024)
+
+        # Convert to string for easier pattern matching (for text-based formats)
+        header_str = header.decode('utf-8', errors='ignore').strip()
+
+        # Check for SVG signatures
+        if header_str.startswith('<?xml') and '<svg' in header_str:
+            return ('svg', 'image/svg+xml')
+        elif header_str.startswith('<svg'):
+            return ('svg', 'image/svg+xml')
+
+        # Check for PDF signature
+        if header.startswith(b'%PDF'):
+            return ('pdf', 'application/pdf')
+
+        # Check for image formats
+        if header.startswith(b'\x89PNG\r\n\x1a\n'):
+            return ('png', 'image/png')
+        elif header.startswith(b'\xff\xd8\xff'):
+            return ('jpg', 'image/jpeg')
+        elif header.startswith(b'GIF87a') or header.startswith(b'GIF89a'):
+            return ('gif', 'image/gif')
+        elif header.startswith(b'II*\x00') or header.startswith(b'MM\x00*'):
+            return ('tiff', 'image/tiff')
+        elif header.startswith(b'RIFF') and header[8:12] == b'WEBP':
+            return ('webp', 'image/webp')
+        elif header.startswith(b'\x00\x00\x00\x20ftypavif'):
+            return ('avif', 'image/avif')
+        elif header.startswith(b'\x00\x00\x00\x18ftypmif1') or header.startswith(b'\x00\x00\x00\x18ftypmsf1') or header.startswith(b'\x00\x00\x00\x18ftypheic') or header.startswith(b'\x00\x00\x00\x18ftypheix'):
+            return ('heic', 'image/heic')
+
+        # Check for audio formats
+        elif header.startswith(b'ID3') or header.startswith(b'\xff\xfb') or header.startswith(b'\xff\xf3'):
+            return ('mp3', 'audio/mpeg')
+        elif header.startswith(b'\x00\x00\x00\x20ftypM4A') or header.startswith(b'\x00\x00\x00\x1cftypM4A'):
+            return ('m4a', 'audio/mp4')
+        elif header.startswith(b'RIFF') and header[8:12] == b'WAVE':
+            return ('wav', 'audio/wav')
+
+        # Check for video formats
+        # MP4: look for 'ftyp' at offset 4 and major brand in common MP4 video brands
+        # Generally catches MP4-based formats that were not caught above
+        elif header[4:8] == b'ftyp' and header[8:12] in {b'mp41', b'mp42', b'isom', b'iso2', b'avc1', b'dash', b'M4V '}:
+            return ('mp4', 'video/mp4')
+        # MOV: look for 'ftyp' at offset 4 and major brand in common QuickTime brands
+        elif header[4:8] == b'ftyp' and header[8:12] in {b'qt  ', b'M4V '}:
+            return ('mov', 'video/quicktime')
+        elif header.startswith(b'RIFF') and header[8:12] == b'AVI ':
+            return ('avi', 'video/x-msvideo')
+
+        return None
+    except Exception:
+        return None
+
+
 class StreamContext(ctypes.Structure):
     """Opaque structure for stream context."""
     _fields_ = []  # Empty as it's opaque in the C API
@@ -676,8 +745,14 @@ def sign_file(
             # Get the MIME type from the file extension
             mime_type = mimetypes.guess_type(str(source_path))[0]
             if not mime_type:
-                raise C2paError.NotSupported(
-                    f"Could not determine MIME type for file: {source_path}")
+                # If the file is extensionless, we may not be able to properly guess
+                # So we attempt one more guessing round here
+                other_mimetype_guess = _guess_mime_type_using_magic_number(source_path)
+                if other_mimetype_guess:
+                    mime_type, _ = other_mimetype_guess
+                else:
+                  raise C2paError.NotSupported(
+                      f"Could not determine MIME type for file: {source_path}")
 
             if return_manifest_as_bytes:
                 # Convert Python streams to Stream objects for internal signing
@@ -1039,7 +1114,7 @@ def __init__(self,
         """Create a new Reader.
 
         Args:
-            format_or_path: The format or path to read from
+            format_or_path: The format (eg. file extension) or path to read from
             stream: Optional stream to read from (any Python stream-like object)
             manifest_data: Optional manifest data in bytes
 
@@ -1952,8 +2027,14 @@ def sign_file(self,
         # Get the MIME type from the file extension
         mime_type = mimetypes.guess_type(str(source_path))[0]
         if not mime_type:
-            raise C2paError.NotSupported(
-                f"Could not determine MIME type for file: {source_path}")
+            # If the file is extensionless, we may not be able to properly guess
+            # So we attempt one more guessing round here
+            other_mimetype_guess = _guess_mime_type_using_magic_number(source_path)
+            if other_mimetype_guess:
+                mime_type, _ = other_mimetype_guess
+            else:
+              raise C2paError.NotSupported(
+                  f"Could not determine MIME type for file: {source_path}")
 
         # Open source and destination files
         with open(source_path, 'rb') as source_file, open(dest_path, 'wb') as dest_file:

tests/fixtures/extensionless-files/readme.md

@@ -0,0 +1,3 @@
+# About the these test files
+
+The test files in these folder are a subseet of files found in the [c2pa-rs repo's test fixtures](https://github.com/contentauth/c2pa-rs/tree/main/sdk/tests/fixtures), renamed to loose the file extension.

tests/test_unit_tests.py

@@ -25,7 +25,7 @@
 import shutil
 
 from c2pa import Builder, C2paError as Error, Reader, C2paSigningAlg as SigningAlg, C2paSignerInfo, Signer, sdk_version
-from c2pa.c2pa import Stream, read_ingredient_file, read_file, sign_file, load_settings, create_signer
+from c2pa.c2pa import Stream, read_ingredient_file, read_file, sign_file, load_settings, create_signer, _guess_mime_type_using_magic_number
 
 # Suppress deprecation warnings
 warnings.filterwarnings("ignore", category=DeprecationWarning)
@@ -783,6 +783,150 @@ def test_sign_file(self):
             # Clean up the temporary directory
             shutil.rmtree(temp_dir)
 
+    def test_sign_extensionless_jpg_file(self):
+        """Test signing a file using the sign_file method."""
+        # Create a temporary directory for the test
+        temp_dir = tempfile.mkdtemp()
+        try:
+            # Use the extensionless jpg from the test files
+            extensionless_dir = os.path.join(FIXTURES_DIR, "extensionless-files")
+            source_path = os.path.join(extensionless_dir, "jpg")
+
+            output_path = os.path.join(temp_dir, "signed_output.jpg")
+
+            # Use the sign_file method
+            builder = Builder(self.manifestDefinition)
+            manifest_bytes = builder.sign_file(
+                source_path=source_path,
+                dest_path=output_path,
+                signer=self.signer
+            )
+
+            # Verify the output file was created
+            self.assertTrue(os.path.exists(output_path))
+
+            # Verify
+            self.assertIsInstance(manifest_bytes, bytes)
+            self.assertGreater(len(manifest_bytes), 0)
+
+            # Read the signed file and verify the manifest
+            with open(output_path, "rb") as file:
+                reader = Reader("image/jpeg", file)
+                json_data = reader.json()
+                self.assertIn("Python Test", json_data)
+                self.assertNotIn("validation_status", json_data)
+
+        finally:
+            # Clean up the temporary directory
+            shutil.rmtree(temp_dir)
+
+    def test_sign_extensionless_svg_file(self):
+        """Test signing an extensionless SVG file using the sign_file method."""
+        # Create a temporary directory for the test
+        temp_dir = tempfile.mkdtemp()
+        try:
+            # Use the extensionless svg from the test files
+            extensionless_dir = os.path.join(FIXTURES_DIR, "extensionless-files")
+            source_path = os.path.join(extensionless_dir, "svg")
+
+            output_path = os.path.join(temp_dir, "signed_output.svg")
+
+            # Use the sign_file method
+            builder = Builder(self.manifestDefinition)
+            manifest_bytes = builder.sign_file(
+                source_path=source_path,
+                dest_path=output_path,
+                signer=self.signer
+            )
+
+            # Verify the output file was created
+            self.assertTrue(os.path.exists(output_path))
+
+            # Verify output
+            self.assertIsInstance(manifest_bytes, bytes)
+            self.assertGreater(len(manifest_bytes), 0)
+
+            # Read the signed file and verify the manifest
+            with open(output_path, "rb") as file:
+                reader = Reader("image/svg+xml", file)
+                json_data = reader.json()
+                self.assertIn("Python Test", json_data)
+                self.assertNotIn("validation_status", json_data)
+
+        finally:
+            # Clean up the temporary directory
+            shutil.rmtree(temp_dir)
+
+    def test_builder_sign_extensionless_svg_file_mimetype(self):
+        """Test signing an extensionless SVG file using the builder.sign method."""
+        # Create a temporary directory for the test
+        temp_dir = tempfile.mkdtemp()
+        try:
+            # Use the extensionless svg from the test files
+            extensionless_dir = os.path.join(FIXTURES_DIR, "extensionless-files")
+            source_path = os.path.join(extensionless_dir, "svg")
+
+            output_path = os.path.join(temp_dir, "signed_output.svg")
+
+            # Use the builder.sign method
+            builder = Builder(self.manifestDefinition)
+            with open(source_path, 'rb') as source_file, open(output_path, 'wb') as dest_file:
+                builder.sign(
+                    signer=self.signer,
+                    format="image/svg+xml",  # Use extension instead of MIME type
+                    source=source_file,
+                    dest=dest_file
+                )
+
+            # Verify the output file was created
+            self.assertTrue(os.path.exists(output_path))
+
+            # Read the signed file and verify the manifest
+            with open(output_path, "rb") as file:
+                reader = Reader("image/svg+xml", file)
+                json_data = reader.json()
+                self.assertIn("Python Test", json_data)
+                self.assertNotIn("validation_status", json_data)
+
+        finally:
+            # Clean up the temporary directory
+            shutil.rmtree(temp_dir)
+
+    def test_builder_sign_extensionless_svg_file_ext(self):
+        """Test signing an extensionless SVG file using the builder.sign method."""
+        # Create a temporary directory for the test
+        temp_dir = tempfile.mkdtemp()
+        try:
+            # Use the extensionless svg from the test files
+            extensionless_dir = os.path.join(FIXTURES_DIR, "extensionless-files")
+            source_path = os.path.join(extensionless_dir, "svg")
+
+            output_path = os.path.join(temp_dir, "signed_output.svg")
+
+            # Use the builder.sign method
+            builder = Builder(self.manifestDefinition)
+            with open(source_path, 'rb') as source_file, open(output_path, 'wb') as dest_file:
+                builder.sign(
+                    signer=self.signer,
+                    format="svg",  # Use extension instead of MIME type
+                    source=source_file,
+                    dest=dest_file
+                )
+
+            # Verify the output file was created
+            self.assertTrue(os.path.exists(output_path))
+
+            # Read the signed file and verify the manifest
+            with open(output_path, "rb") as file:
+                reader = Reader("image/svg+xml", file)
+                json_data = reader.json()
+                self.assertIn("Python Test", json_data)
+                self.assertNotIn("validation_status", json_data)
+
+        finally:
+            # Clean up the temporary directory
+            shutil.rmtree(temp_dir)
+
     def test_sign_file_callback_signer(self):
         """Test signing a file using the sign_file method."""
 
@@ -1447,5 +1591,50 @@ def test_sign_file(self):
                 os.remove(output_path)
 
 
+class TestHelpers(unittest.TestCase):
+    def test_guess_mime_type_using_magic_number(self):
+        """Test the _guess_mime_type_using_magic_number function with various file formats."""
+        extensionless_dir = os.path.join(FIXTURES_DIR, "extensionless-files")
+
+        # Test cases with explicit file paths and expected results
+        test_cases = [
+            (os.path.join(extensionless_dir, "svg"), ('svg', 'image/svg+xml')),
+            (os.path.join(extensionless_dir, "png"), ('png', 'image/png')),
+            (os.path.join(extensionless_dir, "jpg"), ('jpg', 'image/jpeg')),
+            (os.path.join(extensionless_dir, "gif"), ('gif', 'image/gif')),
+            (os.path.join(extensionless_dir, "heic"), ('heic', 'image/heic')),
+            (os.path.join(extensionless_dir, "tiff"), ('tiff', 'image/tiff')),
+            (os.path.join(extensionless_dir, "webp"), ('webp', 'image/webp')),
+            (os.path.join(extensionless_dir, "avif"), ('avif', 'image/avif')),
+            (os.path.join(extensionless_dir, "mp4"), ('mp4', 'video/mp4')),
+            (os.path.join(extensionless_dir, "avi"), ('avi', 'video/x-msvideo')),
+            (os.path.join(extensionless_dir, "mp3"), ('mp3', 'audio/mpeg')),
+            (os.path.join(extensionless_dir, "m4a"), ('m4a', 'audio/mp4')),
+            (os.path.join(extensionless_dir, "wav"), ('wav', 'audio/wav')),
+            (os.path.join(extensionless_dir, "pdf"), ('pdf', 'application/pdf')),
+        ]
+
+        # Test each file explicitly
+        for file_path, expected_result in test_cases:
+            filename = os.path.basename(file_path)
+            with self.subTest(filename=filename):
+                result = _guess_mime_type_using_magic_number(file_path)
+
+                # Verify the result matches expectations
+                self.assertIsNotNone(result, f"Failed to detect type for {filename}")
+                self.assertEqual(result, expected_result,
+                               f"Expected {expected_result} for {filename}, got {result}")
+
+                # Verify extension matches filename
+                expected_extension = filename
+                self.assertEqual(result[0], expected_extension,
+                               f"Extension mismatch for {filename}: expected {expected_extension}, got {result[0]}")
+
+        # Test with non-existent file
+        non_existent_path = os.path.join(extensionless_dir, "non_existent_file")
+        result = _guess_mime_type_using_magic_number(non_existent_path)
+        self.assertIsNone(result, "Should return None for non-existent file")
+
+
 if __name__ == '__main__':
     unittest.main()