diff --git a/docs/class-diagram.md b/docs/class-diagram.md index 33cfd308..5cf3fc6c 100644 --- a/docs/class-diagram.md +++ b/docs/class-diagram.md @@ -66,6 +66,7 @@ classDiagram +from_info(signer_info) Signer$ +from_callback(callback, alg, certs, tsa_url) Signer$ +reserve_size() int + +with_ocsp_response(ocsp_response) Signer +close() } diff --git a/src/c2pa/c2pa.py b/src/c2pa/c2pa.py index 69b0d4eb..f804ea26 100644 --- a/src/c2pa/c2pa.py +++ b/src/c2pa/c2pa.py @@ -81,6 +81,7 @@ 'c2pa_signer_create', 'c2pa_signer_from_info', 'c2pa_signer_reserve_size', + 'c2pa_signer_with_ocsp_response', 'c2pa_ed25519_sign', 'c2pa_signature_free', # Settings bindings @@ -953,6 +954,11 @@ def _setup_function(func, argtypes, restype=None): _setup_function( _lib.c2pa_signer_reserve_size, [ ctypes.POINTER(C2paSigner)], ctypes.c_int64) +_setup_function( + _lib.c2pa_signer_with_ocsp_response, [ + ctypes.POINTER(C2paSigner), + ctypes.POINTER(ctypes.c_ubyte), ctypes.c_size_t], + ctypes.POINTER(C2paSigner)) _setup_function( _lib.c2pa_ed25519_sign, [ ctypes.POINTER( @@ -3003,6 +3009,7 @@ class Signer(ManagedResource): 'callback_error': "Error in signer callback: {}", 'invalid_certs': "Invalid certificate data: {}", 'invalid_tsa': "Invalid TSA URL: {}", + 'ocsp_error': "Error stapling OCSP response: {}", 'encoding_error': "Invalid UTF-8 characters in input: {}" } @@ -3220,6 +3227,37 @@ def reserve_size(self) -> int: return result + def with_ocsp_response(self, ocsp_response: bytes) -> 'Signer': + """Staple a DER-encoded OCSP response into every signature this + signer produces. + + Calls stack: call once per certificate along the chain, signing + certificate first. The caller fetches the responses and keeps them + fresh: OCSP responses expire, so create a new signer when they are + refreshed. + + Args: + ocsp_response: The DER-encoded OCSP response + + Returns: + This signer instance, for method chaining. + + Raises: + C2paError: If the response is empty or could not be stapled + """ + self._ensure_valid_state() + + ocsp_array = ( + ctypes.c_ubyte * + len(ocsp_response)).from_buffer_copy(ocsp_response) + # This native call retains the original signer on failure. + result = _lib.c2pa_signer_with_ocsp_response( + self._handle, ocsp_array, len(ocsp_response)) + self._swap_handle(_check_ffi_operation_result( + result, Signer._ERROR_MESSAGES['ocsp_error'])) + + return self + class Builder(ManagedResource): """High-level wrapper for C2PA Builder operations.""" diff --git a/tests/test_unit_tests.py b/tests/test_unit_tests.py index bc925aad..9bd3e93b 100644 --- a/tests/test_unit_tests.py +++ b/tests/test_unit_tests.py @@ -1639,6 +1639,23 @@ def _local_signer(self): self.addCleanup(signer.close) return signer + def test_with_ocsp_response(self): + signer = self._local_signer() + reserve_size = signer.reserve_size() + ocsp_response = bytes(range(256)) * 4 + with self.assertRaises(Error): + signer.with_ocsp_response(b"") + self.assertEqual(signer.reserve_size(), reserve_size) + self.assertIs(signer.with_ocsp_response(ocsp_response), signer) + self.assertEqual( + signer.reserve_size(), reserve_size + len(ocsp_response)) + + builder = Builder(self.manifestDefinitionV2) + with open(self.testPath, "rb") as source: + manifest_bytes = builder.sign( + signer, "image/jpeg", source, io.BytesIO()) + self.assertIn(ocsp_response, manifest_bytes) + def _active_signature_info(self, signed_bytes): """signature_info of the active manifest in a signed asset.""" signed_bytes.seek(0)