fix: Change return

Author: tmathern

src/c2pa/c2pa.py

@@ -679,6 +679,114 @@ def sign_file(
             signer.close()
 
 
+def sign_file_with_callback_signer(
+    source_path: Union[str, Path],
+    dest_path: Union[str, Path],
+    manifest: str,
+    callback: Callable[[bytes], bytes],
+    alg: C2paSigningAlg,
+    certs: str,
+    tsa_url: Optional[str] = None,
+    data_dir: Optional[Union[str, Path]] = None
+) -> bytes:
+    """Sign a file with a C2PA manifest using a callback signer.
+
+    This function provides a shortcut to sign files using a callback-based signer
+    and returns the raw manifest bytes.
+
+    Args:
+        source_path: Path to the source file
+        dest_path: Path to write the signed file to
+        manifest: The manifest JSON string
+        callback: Function that signs data and returns the signature
+        alg: The signing algorithm to use
+        certs: Certificate chain in PEM format
+        tsa_url: Optional RFC 3161 timestamp authority URL
+        data_dir: Optional directory to write binary resources to
+
+    Returns:
+        The manifest bytes (binary data)
+
+    Raises:
+        C2paError: If there was an error signing the file
+        C2paError.Encoding: If any of the string inputs contain invalid UTF-8 characters
+        C2paError.NotSupported: If the file type cannot be determined
+    """
+    try:
+        # Create a signer from the callback
+        signer = Signer.from_callback(callback, alg, certs, tsa_url)
+
+        # Create a builder from the manifest
+        builder = Builder(manifest)
+
+        # Open source and destination files
+        with open(source_path, 'rb') as source_file, open(dest_path, 'wb') as dest_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}")
+
+            # Convert Python streams to Stream objects
+            source_stream = Stream(source_file)
+            dest_stream = Stream(dest_file)
+
+            # Use the internal signing logic to get manifest bytes
+            format_str = mime_type.encode('utf-8')
+            manifest_bytes_ptr = ctypes.POINTER(ctypes.c_ubyte)()
+
+            # Call the native signing function
+            result = _lib.c2pa_builder_sign(
+                builder._builder,
+                format_str,
+                source_stream._stream,
+                dest_stream._stream,
+                signer._signer,
+                ctypes.byref(manifest_bytes_ptr)
+            )
+
+            if result < 0:
+                error = _parse_operation_result_for_error(_lib.c2pa_error())
+                if error:
+                    raise C2paError(error)
+
+            # Capture the manifest bytes if available
+            manifest_bytes = b""
+            if manifest_bytes_ptr:
+                # Convert the C pointer to Python bytes
+                manifest_bytes = bytes(manifest_bytes_ptr[:result])
+                # Free the C-allocated memory
+                _lib.c2pa_manifest_bytes_free(manifest_bytes_ptr)
+
+            # If we have manifest bytes and a data directory, write them
+            if manifest_bytes and data_dir:
+                manifest_path = os.path.join(str(data_dir), 'manifest.json')
+                with open(manifest_path, 'wb') as f:
+                    f.write(manifest_bytes)
+
+            return manifest_bytes
+
+    except Exception as e:
+        # Clean up destination file if it exists and there was an error
+        if os.path.exists(dest_path):
+            try:
+                os.remove(dest_path)
+            except OSError:
+                pass  # Ignore cleanup errors
+
+        # Re-raise the error
+        raise C2paError(f"Error signing file: {str(e)}")
+    finally:
+        # Ensure resources are cleaned up
+        if 'builder' in locals():
+            builder.close()
+        if 'signer' in locals():
+            signer.close()
+        if 'source_stream' in locals():
+            source_stream.close()
+        if 'dest_stream' in locals():
+            dest_stream.close()
+
+
 class Stream:
     # Class-level counter for generating unique stream IDs
     # (useful for tracing streams usage in debug)
@@ -2083,6 +2191,7 @@ def ed25519_sign(data: bytes, private_key: str) -> bytes:
     'read_file',
     'read_ingredient_file',
     'sign_file',
+    'sign_file_with_callback_signer',
     'format_embeddable',
     'ed25519_sign',
     'sdk_version'

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, sign_file_with_callback_signer
 
 # Suppress deprecation warnings
 warnings.filterwarnings("ignore", category=DeprecationWarning)
@@ -541,16 +541,16 @@ def test_builder_sign_with_duplicate_ingredient(self):
             # Verify the first ingredient's title matches what we set
             first_ingredient = active_manifest["ingredients"][0]
             self.assertEqual(first_ingredient["title"], "Test Ingredient")
-            
+
             # Verify subsequent labels are unique and have a double underscore with a monotonically inc. index
             second_ingredient = active_manifest["ingredients"][1]
             self.assertTrue(second_ingredient["label"].endswith("__1"))
 
             third_ingredient = active_manifest["ingredients"][2]
             self.assertTrue(third_ingredient["label"].endswith("__2"))
-            
+
         builder.close()
-   
+
     def test_builder_sign_with_ingredient_from_stream(self):
         """Test Builder class operations with a real file using stream for ingredient."""
         # Test creating builder from JSON
@@ -880,6 +880,80 @@ def sign_callback(data: bytes) -> bytes:
             # Clean up the temporary directory
             shutil.rmtree(temp_dir)
 
+    def test_sign_file_with_callback_signer(self):
+        """Test signing a file using the sign_file_with_callback_signer function."""
+        # Create a temporary directory for the test
+        temp_dir = tempfile.mkdtemp()
+        try:
+            # Create a temporary output file path
+            output_path = os.path.join(temp_dir, "signed_output_callback.jpg")
+
+            # Create a real ES256 signing callback
+            def sign_callback(data: bytes) -> bytes:
+                """Real ES256 signing callback that creates actual signatures."""
+                # Load the private key from the test fixtures
+                with open(os.path.join(self.data_dir, "es256_private.key"), "rb") as key_file:
+                    private_key_data = key_file.read()
+
+                # Load the private key using cryptography
+                private_key = serialization.load_pem_private_key(
+                    private_key_data,
+                    password=None,
+                    backend=default_backend()
+                )
+
+                # Create the signature using ES256 (ECDSA with SHA-256)
+                from cryptography.hazmat.primitives import hashes
+                from cryptography.hazmat.primitives.asymmetric import ec
+
+                signature = private_key.sign(
+                    data,
+                    ec.ECDSA(hashes.SHA256())
+                )
+
+                return signature
+
+            # Import the new function
+            from c2pa.c2pa import sign_file_with_callback_signer
+
+            # Use the sign_file_with_callback_signer function
+            manifest_bytes = sign_file_with_callback_signer(
+                source_path=self.testPath,
+                dest_path=output_path,
+                manifest=self.manifestDefinition,
+                callback=sign_callback,
+                alg=SigningAlg.ES256,
+                certs=self.certs.decode('utf-8'),
+                tsa_url="http://timestamp.digicert.com"
+            )
+
+            # Verify the output file was created
+            self.assertTrue(os.path.exists(output_path))
+
+            # Verify the manifest bytes are binary data (not JSON text)
+            self.assertIsInstance(manifest_bytes, bytes)
+            self.assertGreater(len(manifest_bytes), 0)
+            
+            # Try to decode as UTF-8 to see if it's text-based (it shouldn't be)
+            try:
+                manifest_bytes.decode('utf-8')
+                # If we get here, it's text-based, which is unexpected
+                self.fail("Manifest bytes should be binary data, not UTF-8 text")
+            except UnicodeDecodeError:
+                # This is expected - manifest bytes should be binary
+                pass
+
+            # Read the signed file and verify the manifest contains expected content
+            with open(output_path, "rb") as file:
+                reader = Reader("image/jpeg", file)
+                file_manifest_json = reader.json()
+                self.assertIn("Python Test", file_manifest_json)
+                self.assertNotIn("validation_status", file_manifest_json)
+
+        finally:
+            # Clean up the temporary directory
+            shutil.rmtree(temp_dir)
+
 class TestStream(unittest.TestCase):
     def setUp(self):
         # Create a temporary file for testing