fix: Clean up

Author: tmathern

requirements-dev.txt

@@ -12,4 +12,7 @@ pytest-benchmark>=5.1.0
 requests>=2.0.0
 
 # Code formatting
-autopep8==2.0.4  # For automatic code formatting
+autopep8==2.0.4  # For automatic code formatting
+
+# Test dependencies (for callback signers)
+cryptography==45.0.4

src/c2pa/c2pa.py

@@ -1308,14 +1308,28 @@ def from_callback(
             C2paError: If there was an error creating the signer
             C2paError.Encoding: If the certificate data or TSA URL contains invalid UTF-8 characters
         """
+        # Define error messages locally since they're instance attributes
+        error_messages = {
+            'closed_error': "Signer is closed",
+            'cleanup_error': "Error during cleanup: {}",
+            'signer_cleanup': "Error cleaning up signer: {}",
+            'size_error': "Error getting reserve size: {}",
+            'callback_error': "Error in signer callback: {}",
+            'info_error': "Error creating signer from info: {}",
+            'invalid_data': "Invalid data for signing: {}",
+            'invalid_certs': "Invalid certificate data: {}",
+            'invalid_tsa': "Invalid TSA URL: {}",
+            'encoding_error': "Invalid UTF-8 characters in input: {}"
+        }
+
         # Validate inputs before creating
         if not certs:
             raise C2paError(
-                cls._error_messages['invalid_certs'].format("Missing certificate data"))
+                error_messages['invalid_certs'].format("Missing certificate data"))
 
         if tsa_url and not tsa_url.startswith(('http://', 'https://')):
             raise C2paError(
-                cls._error_messages['invalid_tsa'].format("Invalid TSA URL format"))
+                error_messages['invalid_tsa'].format("Invalid TSA URL format"))
 
         # Create a wrapper callback that handles errors and memory management
         def wrapped_callback(context, data_ptr, data_len, signed_bytes_ptr, signed_len):
@@ -1349,7 +1363,7 @@ def wrapped_callback(context, data_ptr, data_len, signed_bytes_ptr, signed_len):
                 return actual_len
             except Exception as e:
                 print(
-                    cls._error_messages['callback_error'].format(
+                    error_messages['callback_error'].format(
                         str(e)), file=sys.stderr)
                 return 0
 
@@ -1359,7 +1373,7 @@ def wrapped_callback(context, data_ptr, data_len, signed_bytes_ptr, signed_len):
             tsa_url_bytes = tsa_url.encode('utf-8') if tsa_url else None
         except UnicodeError as e:
             raise C2paError.Encoding(
-                cls._error_messages['encoding_error'].format(
+                error_messages['encoding_error'].format(
                     str(e)))
 
         # Create the signer with the wrapped callback
@@ -1385,7 +1399,7 @@ def wrapped_callback(context, data_ptr, data_len, signed_bytes_ptr, signed_len):
         # Initialize the signer instance
         signer_instance._signer = signer_ptr
         signer_instance._closed = False
-        signer_instance._error_messages = cls._error_messages
+        signer_instance._error_messages = error_messages
 
         return signer_instance
 
@@ -1973,7 +1987,7 @@ def wrapped_callback(context, data_ptr, data_len, signed_bytes_ptr, signed_len):
     create_signer._callbacks.append(c_callback)  # Keep it alive
 
     signer_ptr = _lib.c2pa_signer_create(
-        None,  # context
+        None,
         c_callback,  # Use the stored callback
         alg,
         certs_bytes,

tests/test_unit_tests.py

@@ -21,6 +21,8 @@
 from cryptography.hazmat.primitives import hashes, serialization
 from cryptography.hazmat.primitives.asymmetric import padding
 from cryptography.hazmat.backends import default_backend
+import tempfile
+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
@@ -720,9 +722,6 @@ def test_builder_set_remote_url_no_embed(self):
 
     def test_sign_file(self):
         """Test signing a file using the sign_file method."""
-        import tempfile
-        import shutil
-
         # Create a temporary directory for the test
         temp_dir = tempfile.mkdtemp()
         try:
@@ -753,9 +752,6 @@ def test_sign_file(self):
 
     def test_sign_file_callback_signer(self):
         """Test signing a file using the sign_file method."""
-        import tempfile
-        import shutil
-
         # Create a temporary directory for the test
         temp_dir = tempfile.mkdtemp()
         try:
@@ -819,6 +815,70 @@ def sign_callback(data: bytes) -> bytes:
             # Clean up the temporary directory
             shutil.rmtree(temp_dir)
 
+    def test_sign_file_callback_signer_from_callback(self):
+        """Test signing a file using the sign_file method with Signer.from_callback."""
+        # 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_from_callback.jpg")
+
+            # Use the sign_file method
+            builder = Builder(self.manifestDefinition)
+
+            # 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)
+                # For ECDSA, we use the signature_algorithm_constructor
+                from cryptography.hazmat.primitives import hashes
+                from cryptography.hazmat.primitives.asymmetric import ec
+
+                signature = private_key.sign(
+                    data,
+                    ec.ECDSA(hashes.SHA256())
+                )
+
+                return signature
+
+            # Create signer with callback using Signer.from_callback
+            signer = Signer.from_callback(
+                callback=sign_callback,
+                alg=SigningAlg.ES256,
+                certs=self.certs.decode('utf-8'),
+                tsa_url="http://timestamp.digicert.com"
+            )
+
+            result = builder.sign_file(
+                source_path=self.testPath,
+                dest_path=output_path,
+                signer=signer
+            )
+
+            # 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/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)
 
 class TestStream(unittest.TestCase):
     def setUp(self):