Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions docs/class-diagram.md
Original file line number Diff line number Diff line change
Expand Up @@ -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()
}

Expand Down
38 changes: 38 additions & 0 deletions src/c2pa/c2pa.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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: {}"
}

Expand Down Expand Up @@ -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."""
Expand Down
17 changes: 17 additions & 0 deletions tests/test_unit_tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down