feat(signer): add direct COSE callback binding

Expose Signer.from_cose_callback through the native direct COSE API.

Author: s35-lee

src/c2pa/c2pa.py

@@ -82,6 +82,7 @@
     'c2pa_format_embeddable',
     # Signer bindings
     'c2pa_signer_create',
+    'c2pa_signer_create_direct_cose',
     'c2pa_signer_from_info',
     'c2pa_signer_reserve_size',
     'c2pa_ed25519_sign',
@@ -1424,6 +1425,9 @@ def _setup_function(func, argtypes, restype=None):
                  ctypes.c_char_p,
                  ctypes.c_char_p],
                 ctypes.POINTER(C2paSigner))
+_setup_function(_lib.c2pa_signer_create_direct_cose,
+                [ctypes.c_void_p, SignerCallback, ctypes.c_size_t],
+                ctypes.POINTER(C2paSigner))
 _setup_function(_lib.c2pa_signer_from_info,
                 [ctypes.POINTER(C2paSignerInfo)],
                 ctypes.POINTER(C2paSigner))
@@ -3885,9 +3889,84 @@ def wrapped_callback(
 
         return signer_instance
 
+    @classmethod
+    def from_cose_callback(
+        cls,
+        callback: Callable[[bytes], bytes],
+        reserve_size: int,
+    ) -> 'Signer':
+        """Create a signer whose callback returns a complete tagged COSE_Sign1.
+
+        The callback receives the serialized C2PA claim bytes and must return
+        exactly reserve_size bytes, including any required COSE padding. It
+        supplies the signing algorithm, certificate chain, and any timestamps.
+        The SDK bypasses normal COSE construction and post-signing verification
+        for this signer. Callback failures are reported by the signing operation
+        as a native C2PA error.
+
+        Callbacks must be safe for concurrent calls and remote callbacks must
+        enforce their own timeout. The callback is kept alive for the lifetime
+        of the native signer, including when transferred to a Context.
+
+        Args:
+            callback: Function receiving claim bytes and returning COSE bytes.
+            reserve_size: Positive size of both the signature box and output
+                buffer, representable by the native callback's signed size type.
+
+        Returns:
+            A new Signer instance.
+
+        Raises:
+            TypeError: If callback is not callable or reserve_size is not an int.
+            ValueError: If reserve_size is outside the supported range.
+            C2paError: If native signer creation fails.
+        """
+        if not callable(callback):
+            raise TypeError("callback must be callable")
+        if isinstance(reserve_size, bool) or not isinstance(reserve_size, int):
+            raise TypeError("reserve_size must be an integer")
+        if not 0 < reserve_size <= sys.maxsize:
+            raise ValueError(
+                "reserve_size must be positive and fit in a native signed size")
+
+        def wrapped_callback(context, data_ptr, data_len, output_ptr, capacity):
+            try:
+                if (
+                    not data_ptr
+                    or data_len <= 0
+                    or not output_ptr
+                    or capacity != reserve_size
+                ):
+                    return -1
+                claim = ctypes.string_at(data_ptr, data_len)
+                cose = callback(claim)
+                if not isinstance(cose, bytes) or len(cose) != capacity:
+                    return -1
+                ctypes.memmove(output_ptr, cose, capacity)
+                return len(cose)
+            except Exception as e:
+                logger.error(cls._ERROR_MESSAGES['callback_error'].format(str(e)))
+                return -1
+
+        callback_cb = SignerCallback(wrapped_callback)
+        with _native_section():
+            signer_ptr = _lib.c2pa_signer_create_direct_cose(
+                None, callback_cb, reserve_size)
+            _check_ffi_operation_result(
+                signer_ptr, "Failed to create direct COSE signer")
+
+        try:
+            signer_instance = cls(signer_ptr)
+            signer_instance._callback_cb = callback_cb
+        except Exception:
+            ManagedResource._free_native_ptr(signer_ptr)
+            raise
+        return signer_instance
+
     def __init__(self, signer_ptr: ctypes.POINTER(C2paSigner)):
         """Initialize a new Signer instance. This constructor is not meant
-        to be called directly. Use from_info() or from_callback() instead.
+        to be called directly. Use from_info(), from_callback(), or
+        from_cose_callback() instead.
 
         Args:
             signer_ptr: Pointer to the native C2PA signer
@@ -3905,7 +3984,7 @@ def __init__(self, signer_ptr: ctypes.POINTER(C2paSigner)):
 
     def _init_attrs(self):
         super()._init_attrs()
-        # from_callback() replaces this with the real callback, which has to
+        # Callback constructors replace this with the real callback, which has to
         # outlive the signer that calls it.
         self._callback_cb = None
 

tests/test_unit_tests.py

@@ -24,6 +24,8 @@
 import unittest
 import ctypes
 import warnings
+import weakref
+from cryptography import x509
 from cryptography.hazmat.primitives import hashes, serialization
 from cryptography.hazmat.primitives.asymmetric import ec
 from cryptography.hazmat.backends import default_backend
@@ -6102,6 +6104,207 @@ def test_instance_id_as_ingredient_identifier_in_catalog(self):
         archive.close()
 
 
+class TestDirectCoseSigner(unittest.TestCase):
+    """Direct COSE callback contract and native signing integration tests."""
+
+    @staticmethod
+    def _invoke_callback(signer, claim=b"claim bytes", capacity=4):
+        data = (ctypes.c_ubyte * len(claim)).from_buffer_copy(claim)
+        # A sentinel after the advertised capacity detects out-of-bounds writes.
+        output = (ctypes.c_ubyte * (capacity + 1))(*([0xaa] * (capacity + 1)))
+        written = signer._callback_cb(None, data, len(claim), output, capacity)
+        return written, bytes(output)
+
+    @staticmethod
+    def _cbor_bytes(value):
+        """Encode a CBOR byte string for the test's detached COSE object."""
+        length = len(value)
+        if length < 24:
+            header = bytes([0x40 + length])
+        elif length <= 0xff:
+            header = b"\x58" + bytes([length])
+        elif length <= 0xffff:
+            header = b"\x59" + length.to_bytes(2, "big")
+        else:
+            header = b"\x5a" + length.to_bytes(4, "big")
+        return header + value
+
+    @classmethod
+    def _make_cose_callback(cls, reserve_size, claims):
+        """Use fixture credentials to simulate a service constructing COSE."""
+        key = serialization.load_pem_private_key(
+            (Path(FIXTURES_DIR) / "ed25519.pem").read_bytes(), password=None)
+        pem_certs = re.findall(
+            rb"-----BEGIN CERTIFICATE-----.*?-----END CERTIFICATE-----",
+            (Path(FIXTURES_DIR) / "ed25519.pub").read_bytes(), re.DOTALL)
+        certs = [x509.load_pem_x509_certificate(cert).public_bytes(
+            serialization.Encoding.DER) for cert in pem_certs]
+        assert 0 < len(certs) < 24
+        chain = (cls._cbor_bytes(certs[0]) if len(certs) == 1 else
+                 bytes([0x80 + len(certs)]) + b"".join(map(cls._cbor_bytes, certs)))
+        # Protected map: algorithm = EdDSA (-8), x5chain = fixture certificates.
+        protected = b"\xa2\x01\x27\x18\x21" + chain
+
+        def callback(claim):
+            claims.append(claim)
+            # Sign the detached claim as the COSE Sig_structure payload.
+            to_sign = (b"\x84\x6aSignature1" + cls._cbor_bytes(protected)
+                       + b"\x40" + cls._cbor_bytes(claim))
+            signature = key.sign(to_sign)
+            prefix = b"\xd2\x84" + cls._cbor_bytes(protected) + b"\xa1\x63pad"
+            suffix = b"\xf6" + cls._cbor_bytes(signature)
+            # This reservation puts pad's length in the CBOR uint16 range.
+            padding = reserve_size - len(prefix) - len(suffix) - 3
+            assert 0xff < padding <= 0xffff
+            cose = prefix + cls._cbor_bytes(bytes(padding)) + suffix
+            assert len(cose) == reserve_size
+            return cose
+
+        return callback
+
+    def test_creation_and_reserve_size(self):
+        with Signer.from_cose_callback(lambda claim: b"1234", 4) as signer:
+            self.assertEqual(signer.reserve_size(), 4)
+
+    def test_invalid_reserve_size_rejected_before_native_call(self):
+        with patch.object(c2pa_module._lib, "c2pa_signer_create_direct_cose") as create:
+            for size in (True, False, 1.5, "4", None):
+                with self.subTest(size=size), self.assertRaises(TypeError):
+                    Signer.from_cose_callback(lambda claim: b"1234", size)
+            for size in (0, -1, sys.maxsize + 1):
+                with self.subTest(size=size), self.assertRaises(ValueError):
+                    Signer.from_cose_callback(lambda claim: b"1234", size)
+            create.assert_not_called()
+
+    def test_callback_must_be_callable(self):
+        with self.assertRaises(TypeError):
+            Signer.from_cose_callback(None, 4)
+
+    def test_claim_and_cose_bytes_are_preserved(self):
+        claims = []
+
+        def callback(claim):
+            claims.append(claim)
+            return b"\xd2\x84\x00\x00"
+
+        with Signer.from_cose_callback(callback, 4) as signer:
+            written, output = self._invoke_callback(signer)
+            self.assertEqual(claims, [b"claim bytes"])
+            self.assertEqual(written, 4)
+            self.assertEqual(output, b"\xd2\x84\x00\x00\xaa")
+
+    def test_invalid_results_do_not_write_or_truncate(self):
+        for result in (b"", b"123", b"12345", None, "1234",
+                       bytearray(b"1234"), memoryview(b"1234")):
+            with self.subTest(result=result):
+                with Signer.from_cose_callback(lambda claim: result, 4) as signer:
+                    written, output = self._invoke_callback(signer)
+                    self.assertEqual(written, -1)
+                    self.assertEqual(output, b"\xaa" * 5)
+
+    def test_native_capacity_must_match_reservation(self):
+        with Signer.from_cose_callback(lambda claim: b"1234", 4) as signer:
+            for capacity in (3, 5):
+                written, output = self._invoke_callback(signer, capacity=capacity)
+                self.assertEqual(written, -1)
+                self.assertEqual(output, b"\xaa" * (capacity + 1))
+
+    def test_callback_exception_returns_failure(self):
+        def fail(claim):
+            raise RuntimeError("remote signing failed")
+
+        with Signer.from_cose_callback(fail, 4) as signer:
+            with self.assertLogs(c2pa_module.logger, level="ERROR"):
+                written, output = self._invoke_callback(signer)
+            self.assertEqual(written, -1)
+            self.assertEqual(output, b"\xaa" * 5)
+
+    def test_callback_is_retained_until_signer_closes(self):
+        class Callback:
+            def __call__(self, claim):
+                return b"1234"
+
+        callback = Callback()
+        ref = weakref.ref(callback)
+        with Signer.from_cose_callback(callback, 4) as signer:
+            del callback
+            gc.collect()
+            self.assertIsNotNone(ref())
+            self.assertEqual(self._invoke_callback(signer)[0], 4)
+        gc.collect()
+        self.assertIsNone(ref())
+        self.assertIsNone(signer._callback_cb)
+
+    def test_context_retains_consumed_signer_callback(self):
+        signer = Signer.from_cose_callback(lambda claim: b"1234", 4)
+        callback = signer._callback_cb
+        with Context(signer=signer) as context:
+            self.assertIsNone(signer._callback_cb)
+            self.assertIs(context._signer_callback_cb, callback)
+            del signer
+            gc.collect()
+            self.assertIs(context._signer_callback_cb, callback)
+        self.assertIsNone(context._signer_callback_cb)
+
+    def test_native_pointer_freed_if_wrapping_fails(self):
+        class BrokenSigner(Signer):
+            def __init__(self, pointer):
+                raise RuntimeError("wrapping failed")
+
+        with patch.object(ManagedResource, "_free_native_ptr",
+                          wraps=ManagedResource._free_native_ptr) as free:
+            with self.assertRaisesRegex(RuntimeError, "wrapping failed"):
+                BrokenSigner.from_cose_callback(lambda claim: b"1234", 4)
+            free.assert_called_once()
+
+    def test_callback_failure_reaches_native_signing_error(self):
+        claims = []
+
+        def callback(claim):
+            claims.append(claim)
+            return b"too short"
+
+        with Signer.from_cose_callback(callback, 16384) as signer:
+            with Builder({"title": "Direct COSE failure"}) as builder:
+                with (Path(FIXTURES_DIR) / "C.jpg").open("rb") as source:
+                    with self.assertRaisesRegex(Error, "COSE signature invalid"):
+                        builder.sign(signer, "image/jpeg", source, io.BytesIO())
+        self.assertEqual(len(claims), 1)
+
+    def test_sign_and_read_valid_cose_after_context_transfer(self):
+        claims = []
+        signer = Signer.from_cose_callback(self._make_cose_callback(16384, claims), 16384)
+        settings_json = json.loads((Path(__file__).parent /
+                                    "trust_config_test_settings.json").read_text())
+        manifest = {
+            "title": "Direct COSE integration",
+            "claim_generator": "direct_cose_test/1.0",
+            "assertions": [{"label": "c2pa.actions", "data": {"actions": [{
+                "action": "c2pa.created",
+                "digitalSourceType": "http://cv.iptc.org/newscodes/digitalsourcetype/digitalCreation",
+            }]}}],
+        }
+        with Settings.from_json(json.dumps(settings_json)) as settings:
+            with Context(settings=settings, signer=signer) as context:
+                del signer
+                gc.collect()
+                # Repeated signing also exercises callback lifetime after transfer.
+                for _ in range(2):
+                    output = io.BytesIO()
+                    with Builder(manifest, context=context) as builder:
+                        with (Path(FIXTURES_DIR) / "C.jpg").open("rb") as source:
+                            builder.sign("image/jpeg", source, output)
+                    output.seek(0)
+                    with Reader("image/jpeg", output, context=context) as reader:
+                        report = json.loads(reader.json())
+                        active = report["manifests"][report["active_manifest"]]
+                        self.assertEqual(active["title"], manifest["title"])
+                        self.assertEqual(active["signature_info"]["alg"], "Ed25519")
+                        self.assertIn(report["validation_state"], ("Valid", "Trusted"))
+        self.assertEqual(len(claims), 2)
+        # CBOR maps identify full serialized claims, unlike COSE Sig_structure arrays.
+        self.assertTrue(all(claim and claim[0] >> 5 == 5 for claim in claims))
+
 class TestStream(unittest.TestCase):
     def setUp(self):
         self.temp_file = io.BytesIO()