Merge branch 'main' into docs/pkey_caveat

Author: crandmck

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "c2pa-python"
-version = "0.5.0"
+version = "0.5.1"
 edition = "2021"
 authors = ["Gavin Peacock <[email protected]"]
 
@@ -11,7 +11,7 @@ crate-type = ["cdylib"]
 
 
 [dependencies]
-c2pa = {version = "0.32.0", features = ["unstable_api", "openssl"]}
+c2pa = {version = "0.32.2", features = ["unstable_api", "openssl"]}
 pem = "3.0.3"
 serde = { version = "1.0.197", features = ["derive"] }
 serde_derive = "1.0"

c2pa/c2pa_api/c2pa_api.py

@@ -15,6 +15,7 @@
 import os
 import sys
 import tempfile
+import shutil
 PROJECT_PATH = os.getcwd()
 SOURCE_PATH = os.path.join(
     PROJECT_PATH,"target","python"
@@ -50,7 +51,7 @@ def from_file(cls, path: str, format=None):
     def get_manifest(self, label):
         manifest_store = json.loads(self.json())
         return manifest_store["manifests"].get(label)
-        return json.loads(self.json())
+    
     def get_active_manifest(self):
         manifest_store = json.loads(self.json())
         active_label = manifest_store.get("active_manifest")
@@ -128,10 +129,27 @@ def sign(self, signer, format, input, output = None):
         return super().sign(format, C2paStream(input), C2paStream(output), signer)
 
     def sign_file(self, signer, sourcePath, outputPath):
-        format = os.path.splitext(outputPath)[1][1:]
+        extension = os.path.splitext(outputPath)[1][1:]
         input = open(sourcePath, "rb")
-        output = open(outputPath, "wb")
-        return self.sign(signer, format, input, output)
+        # Check if outputPath is the same as sourcePath
+        if outputPath == sourcePath:
+            # Create a temporary file with the same extension as sourcePath
+            temp_output = tempfile.NamedTemporaryFile(suffix='.' + extension, delete=False)
+            temp_output_path = temp_output.name
+            temp_output.close()
+        else:
+            temp_output_path = outputPath
+        
+        output = open(temp_output_path, "wb")
+        result = self.sign(signer, extension, input, output)
+        output.close()
+        input.close()
+
+        # If a temporary file was used, rename or copy it to the outputPath
+        if outputPath == sourcePath:
+            shutil.move(temp_output_path, outputPath)
+
+        return result
 
 
 # Implements a C2paStream given a stream handle

tests/test_api.py

@@ -13,6 +13,7 @@
 import json
 import pytest
 import tempfile
+import shutil
 
 from c2pa import Builder, Error, Reader, SigningAlg, create_signer, sdk_version, sign_ps256, version
 
@@ -50,7 +51,7 @@ def getitem(d, key):
 }
 
 def test_version():
-    assert version() == "0.5.0"
+    assert version() == "0.5.1"
 
 def test_sdk_version():
     assert "c2pa-rs/" in sdk_version()
@@ -121,6 +122,40 @@ def sign(data: bytes) -> bytes:
         assert manifest,["format"] == "image/jpeg"
         # There should be no validation status errors
         assert manifest.get("validation_status") == None
+    except Exception as e:
+        print("Failed to sign manifest store: " + str(e))
+        exit(1)
+
+# Test signing the same source and destination file
+def test_v2_sign_file_same():
+    data_dir = "tests/fixtures/"
+    try:
+        def sign(data: bytes) -> bytes:
+            return sign_ps256(data, data_dir+"ps256.pem")
+        
+        certs = open(data_dir + "ps256.pub", "rb").read()
+        # Create a local signer from a certificate pem file
+        signer = create_signer(sign, SigningAlg.PS256, certs, "http://timestamp.digicert.com")
+
+        builder = Builder(manifest_def)
+
+        builder.add_resource_file("A.jpg", data_dir + "A.jpg")
+
+        with tempfile.TemporaryDirectory() as output_dir:
+            path = output_dir + "/A.jpg"
+            # Copy the file from data_dir to output_dir
+            shutil.copy(data_dir + "A.jpg", path)
+            c2pa_data = builder.sign_file(signer, path, path)
+            assert len(c2pa_data) > 0
+
+            reader = Reader.from_file(path)
+            manifest = reader.get_active_manifest()
+
+            # check custom title and format
+            assert manifest["title"]== "My Title" 
+            assert manifest["format"] == "image/jpeg"
+            # There should be no validation status errors
+            assert manifest.get("validation_status") == None
     except Exception as e:
         print("Failed to sign manifest store: " + str(e))
         exit(1)

tests/unit_tests.py

@@ -26,7 +26,7 @@
 class TestC2paSdk(unittest.TestCase):
 
     def test_version(self):
-        self.assertIn("0.5.0", sdk_version())
+        self.assertIn("0.5.1", sdk_version())
 
 
 class TestReader(unittest.TestCase):