diff --git a/src/c2pa/c2pa.py b/src/c2pa/c2pa.py index 1ad8dce1..b6bc1931 100644 --- a/src/c2pa/c2pa.py +++ b/src/c2pa/c2pa.py @@ -271,8 +271,9 @@ def __init__(self): self._pending_teardown = None record_owner_pid(self) - def _lock(self): - """Return this resource's operation lock. + def _state_lock(self): + """Return this resource's raw operation lock, with no side effects + beyond mutual exclusion. Reentrant because it is possible to run a finalizer at any bytecode boundary, including inside a region this thread has already locked, @@ -280,6 +281,7 @@ def _lock(self): locked region. Falls back to a fresh lock when the attribute is missing. + Unlike _lock(), this does not open a native-error section. Never hold this across a native call that drives stream callbacks (construction, resource_to_stream, the Builder stream methods, @@ -306,6 +308,19 @@ def _lock(self): pass return lock + @contextlib.contextmanager + def _lock(self): + """Hold this resource's operation lock its duration, + and mark this thread as inside a native-error section. + + Never hold this across a native call that drives stream callbacks. + Those calls release the GIL and re-enter caller-supplied + code, which may call back into this API on another thread. + Only calls that touch no callbacks are serialized here. + """ + with self._state_lock(), _native_section(): + yield + @contextlib.contextmanager def _native_call(self): """Hold the handle valid across a native call that goes back @@ -319,25 +334,22 @@ def _native_call(self): The resource is marked closed as soon as the teardown is recorded, so a caller that closed it cannot keep using it while the free is pending. + + Also opens a native-error section around the yielded body (see + _lock()): the in-flight guard alone only protects this resource's + own handle, not the shared thread-local error slot a caller inside + the block is about to read. """ - with self._lock(): + with self._state_lock(): self._ensure_valid_state() self._inflight = getattr(self, '_inflight', 0) + 1 try: - yield + with _native_section(): + yield finally: - with self._lock(): + with self._state_lock(): self._inflight -= 1 - pending = (self._pending_teardown - if self._inflight == 0 else None) - if pending is not None: - self._pending_teardown = None - # Released the lock before the free: - # _teardown takes it again, and keeping the two acquisitions - # separate means the counter update is never held across - # the release work. - if pending is not None: - self._teardown(pending) + self._maybe_flush_pending() @staticmethod def _free_native_ptr(ptr): @@ -398,6 +410,11 @@ def _teardown(self, free_handle: bool): Holds the operation lock so the free cannot happen between another thread's state check and its use of the handle in a native call. + Deferred (instead of run now) when either gate is blocking: + - this resource's own handle is in flight in a native call + - this thread is inside a native-error section for some call, + that may access the native error slot. + The forked-child case is handled before the lock is taken, because _lock() refuses in a child: this path has to finish rather than report an error, so it cannot rely on acquiring. @@ -409,28 +426,56 @@ def _teardown(self, free_handle: bool): self._lifecycle_state = LifecycleState.CLOSED return - with self._lock(): - if getattr(self, '_inflight', 0) > 0: - # A native call is running that re-enters calling non-native code - # and is still using this handle. - # Record the intent and whichever caller leaves - # _native_call last performs the free. - # Mark the resource closed now so it cannot be used - # while the free is pending. + with self._state_lock(): + if getattr(self, '_inflight', 0) > 0 or _in_native_section(): + # Mark the resource closed now so it cannot be used while + # the free is pending, but record the intent: + # whichever check is blocking will call + # _maybe_flush_pending() once it clears. self._pending_teardown = free_handle self._lifecycle_state = LifecycleState.CLOSED + if _in_native_section(): + _register_for_section_flush(self) return + self._finish_teardown(free_handle) + + def _finish_teardown(self, free_handle: bool): + """The part of _teardown that only runs once nothing is blocking + teardown. Steps: release, null the handle, free if requested. + + Not called directly outside _teardown/_maybe_flush_pending: + callers that want to close a resource still go through _teardown, + which decides whether this can run now or must be deferred. + """ + if is_foreign_process(self): + self._handle = None self._lifecycle_state = LifecycleState.CLOSED - self._safe_release() + return - handle, self._handle = self._handle, None - if free_handle and handle: - try: - ManagedResource._free_native_ptr(handle) - except Exception: - logger.error("Failed to free native %s resources", - type(self).__name__, exc_info=True) + self._lifecycle_state = LifecycleState.CLOSED + self._safe_release() + + handle, self._handle = self._handle, None + if free_handle and handle: + try: + ManagedResource._free_native_ptr(handle) + except Exception: + logger.error("Failed to free native %s resources", + type(self).__name__, exc_info=True) + + def _maybe_flush_pending(self): + """Called when a gate that may have been blocking a deferred + teardown clears (this resource's own _inflight dropping to 0, or + this thread's native-error section closing). + """ + with self._state_lock(): + if self._pending_teardown is None: + return + if getattr(self, '_inflight', 0) > 0 or _in_native_section(): + return + free_handle, self._pending_teardown = self._pending_teardown, None + self._finish_teardown(free_handle) def _release_handle(self): """Free this handle, then close the object. Used only where ownership is @@ -480,9 +525,11 @@ def _create_and_activate(self, ffi_call, error_message, *, Raises: C2paError: If the pointer fails validation; it is freed first. """ - ptr = ffi_call() + ptr = None try: - _check_ffi_operation_result(ptr, error_message, check=check) + with self._lock(): + ptr = ffi_call() + _check_ffi_operation_result(ptr, error_message, check=check) self._activate(ptr) except Exception: if ptr: @@ -543,6 +590,8 @@ def _invoke_consume(self, ffi_call, error_message): ctypes.ArgumentError: If marshalling failed; handle untouched. C2paError: If the call raised any other exception. """ + # Same thread that makes the call, same thread-local slot. + _mark_sentinel_no_native_error() try: return ffi_call(self._handle) except ctypes.ArgumentError: @@ -557,16 +606,12 @@ def _raise_consume_failure(self, error_message): """Raise the error from an FFI handler consuming call. The native error is read before any free so a free's own - pointer-tracking error cannot overwrite it: the native error slot is - sticky and thread-local and the SDK does not clear it before the call, - so this trusts that the failing native path set its own error. - - That ordering is required: - c2pa_free on a handle the registry no longer tracks returns -1 and - overwrites the slot with its own "Other: UntrackedPointer: 0x..." - message. Freeing first would therefore replace the real failure - with another one and, because that substitute carries a pre-consume - tag, invert the retain/consume decision made below. + pointer-tracking error cannot overwrite it. + The native error slot is sticky and thread-local, + and the native SDK does not clear it before the call. + _invoke_consume marks the slot as carrying no error right before a + consuming call, so a failure that sets no error of its own reads back + as no error rather than as a stale one left by an earlier call. Args: error_message: Format string with one placeholder, used when the @@ -593,8 +638,15 @@ def _raise_consume_failure(self, error_message): self._teardown(free_handle=False) _raise_typed_c2pa_error(error) - # No error in the slot: ownership is unknown, so free defensively. - self._release_handle() + # The call failed without setting an error of its own, + # so ownership is unknown. + # Treat as consumed: a free here can race a recycled address + # in other threads. + logger.debug( + "%s: consuming call failed without setting its own " + "native error; treating as consumed", + type(self).__name__) + self._teardown(free_handle=False) raise C2paError(error_message.format("Unknown error")) def _consume_and_swap(self, ffi_call, error_message): @@ -792,16 +844,47 @@ class C2paStream(ctypes.Structure): ] +# Unaligned address passed to c2pa_free to plant a marker +# in the native error slot. +# Never a real handle: allocations are aligned, and the Python +# layer only passes real handles or this constant to c2pa_free. +_MARKER_ADDR = 1 + +# Exact text the native lib writes for a failed free of _MARKER_ADDR. +# Learned at import by _learn_sentinel_no_native_error_text(). +# The format is a native implementation detail, +# so it is read back rather than hardcoded. +_NO_NATIVE_ERROR_TEXT = None + + +def _mark_sentinel_no_native_error(): + """Write the no-error marker into this thread's native error slot. + + A c2pa_free of an address the registry does not track writes + an expected error message learned at import into the + thread-local error slot and returns -1. + + This marker mechanism exists to distinguish a consuming call that + failed without setting its own error from a stale message left + by an earlier call on the same thread. + """ + _lib.c2pa_free(_MARKER_ADDR) + + +def _is_no_native_error(message: str) -> bool: + """True for the sentinel marker meaning "no current error of our own".""" + return message == _NO_NATIVE_ERROR_TEXT + + def _read_native_error() -> Optional[str]: """Read the last error from the native library, or None if unset. - Peeks: the error stays in the native slot, - until the next error overwrites it. - - With no error set the native side still returns an owned pointer to an - empty string, so the pointer alone does not tell us whether there is an - error. Only a non-empty message counts as one; the empty string still - has to be freed. + The slot is marked as carrying no error before returning, so a + given error is reported once, by the caller that observes it. The native + slot is thread-local and sticky, so a message left in place stays readable + indefinitely and is available to be reported again by a later, + unrelated call that failed without setting an error of its own + (or a missing clear of an error slot). """ error = _lib.c2pa_error() if not error: @@ -810,7 +893,56 @@ def _read_native_error() -> Optional[str]: message = ctypes.string_at(error).decode('utf-8') finally: _lib.c2pa_string_free(error) - return message or None + if not message: + return None + + _mark_sentinel_no_native_error() + if _is_no_native_error(message): + return None + return message + + +_native_section_state = threading.local() + + +def _in_native_section() -> bool: + """True while this thread is between an FFI call and reading back the + native error it may have set (see _native_section()).""" + return getattr(_native_section_state, 'depth', 0) > 0 + + +def _register_for_section_flush(resource): + """Record that `resource`'s teardown was deferred only because this + thread's native-error section was open.""" + pending = getattr(_native_section_state, 'pending_resources', None) + if pending is not None: + pending.append(resource) + + +@contextlib.contextmanager +def _native_section(): + """Mark this thread as inside a section where a native call's result is + about to be read back: an error-slot check, or a consuming call's + success/failure classification. + + Reentrant: a call whose own native call triggers another one + recursively (same thread) nests correctly here. Only the outermost + span flushes, so nothing is freed before an inner, still-open span is + done reading its own error. + """ + state = _native_section_state + depth = getattr(state, 'depth', 0) + state.depth = depth + 1 + if depth == 0: + state.pending_resources = [] + try: + yield + finally: + state.depth -= 1 + if state.depth == 0: + pending, state.pending_resources = state.pending_resources, [] + for resource in pending: + resource._maybe_flush_pending() class C2paSignerInfo(ctypes.Structure): @@ -1135,6 +1267,38 @@ def _setup_function(func, argtypes, restype=None): ) _setup_function(_lib.c2pa_free, [ctypes.c_void_p], ctypes.c_int) + +def _learn_sentinel_no_native_error_text(): + """Plant the marker once and read back the exact text the native lib + produces for it, so equality checks match this build of the lib. + + Runs on the importing thread; the text is a format constant, so the + learned value holds for every thread. Raises at import when the read + back text is empty, because the marker mechanism cannot work then. + """ + _mark_sentinel_no_native_error() + raw = _lib.c2pa_error() + if not raw: + raise ImportError( + "c2pa native library did not report an error for a free of " + "an untracked pointer; the error-slot marker cannot work") + try: + text = ctypes.string_at(raw).decode('utf-8') + finally: + _lib.c2pa_string_free(raw) + if not text: + raise ImportError( + "c2pa native library reported an empty error for a free of " + "an untracked pointer; the error-slot marker cannot work") + return text + + +_NO_NATIVE_ERROR_TEXT = _learn_sentinel_no_native_error_text() +assert "0x1" in _NO_NATIVE_ERROR_TEXT, ( + "c2pa native library's untracked-pointer error text no longer " + "includes the planted address; the error-slot marker assumption " + "no longer holds") + _setup_function( _lib.c2pa_context_builder_set_signer, [ctypes.POINTER(C2paContextBuilder), ctypes.POINTER(C2paSigner)], @@ -1556,11 +1720,12 @@ def load_settings(settings: Union[str, dict], format: str = "json") -> None: except (AttributeError, UnicodeEncodeError) as e: raise C2paError(f"Failed to encode settings to UTF-8: {e}") - result = _lib.c2pa_load_settings(settings_bytes, format_bytes) - _check_ffi_operation_result( - result, - "Error loading settings", - check=lambda r: r != 0) + with _native_section(): + result = _lib.c2pa_load_settings(settings_bytes, format_bytes) + _check_ffi_operation_result( + result, + "Error loading settings", + check=lambda r: r != 0) class ContextProvider(ABC): @@ -1793,11 +1958,12 @@ def __init__( # a successful build consumes it, so close() is then a no-op. with self._NativeBuilder() as nb: if settings is not None: - _check_ffi_operation_result( - _lib.c2pa_context_builder_set_settings( - nb._handle, settings._c_settings), - "Failed to set settings on Context", - check=lambda r: r != 0) + with nb._lock(): + _check_ffi_operation_result( + _lib.c2pa_context_builder_set_settings( + nb._handle, settings._c_settings), + "Failed to set settings on Context", + check=lambda r: r != 0) if signer is not None: # The signer's in-flight guard: @@ -1818,9 +1984,10 @@ def __init__( "Failed to set signer on Context: {}") self._has_signer = True - context_ptr = nb._consume_into( - lambda h: _lib.c2pa_context_builder_build(h), - "Failed to build Context: {}") + with nb._native_call(): + context_ptr = nb._consume_into( + lambda h: _lib.c2pa_context_builder_build(h), + "Failed to build Context: {}") self._activate(context_ptr) @@ -2754,25 +2921,27 @@ def _init_from_context(self, context, format_or_path, # Consume current reader, # with manifest data and stream (C FFI pattern), # to create a new one (switch out) - self._consume_and_swap( - lambda handle: ( - _lib.c2pa_reader_with_manifest_data_and_stream( - handle, - format_arg, - self._own_stream._stream, - manifest_array, - len(manifest_data), - ) - ), - Reader._ERROR_MESSAGES['reader_error']) + with self._native_call(): + self._consume_and_swap( + lambda handle: ( + _lib.c2pa_reader_with_manifest_data_and_stream( + handle, + format_arg, + self._own_stream._stream, + manifest_array, + len(manifest_data), + ) + ), + Reader._ERROR_MESSAGES['reader_error']) else: # Consume reader with stream - self._consume_and_swap( - lambda handle: _lib.c2pa_reader_with_stream( - handle, format_arg, - self._own_stream._stream, - ), - Reader._ERROR_MESSAGES['reader_error']) + with self._native_call(): + self._consume_and_swap( + lambda handle: _lib.c2pa_reader_with_stream( + handle, format_arg, + self._own_stream._stream, + ), + Reader._ERROR_MESSAGES['reader_error']) except Exception: self._close_streams() raise @@ -3205,10 +3374,12 @@ def from_info(cls, signer_info: C2paSignerInfo) -> 'Signer': Raises: C2paError: If there was an error creating the signer """ - signer_ptr = _lib.c2pa_signer_from_info(ctypes.byref(signer_info)) + with _native_section(): + signer_ptr = _lib.c2pa_signer_from_info(ctypes.byref(signer_info)) - _check_ffi_operation_result( - signer_ptr, "Failed to create signer from configured signer_info") + _check_ffi_operation_result( + signer_ptr, + "Failed to create signer from configured signer_info") try: return cls(signer_ptr) @@ -3331,16 +3502,17 @@ def wrapped_callback( callback_cb = SignerCallback(wrapped_callback) # Create the signer with the wrapped callback - signer_ptr = _lib.c2pa_signer_create( - None, - callback_cb, - alg, - certs_bytes, - tsa_url_bytes - ) + with _native_section(): + signer_ptr = _lib.c2pa_signer_create( + None, + callback_cb, + alg, + certs_bytes, + tsa_url_bytes + ) - _check_ffi_operation_result(signer_ptr, - "Failed to create signer") + _check_ffi_operation_result(signer_ptr, + "Failed to create signer") try: # Create and return the signer instance with the callback @@ -3502,11 +3674,11 @@ def from_archive( stream_obj = Stream(stream) try: - handle = _lib.c2pa_builder_from_archive(stream_obj._stream) + with _native_section(): + handle = _lib.c2pa_builder_from_archive(stream_obj._stream) - _check_ffi_operation_result(handle, - "Failed to create builder from archive" - ) + _check_ffi_operation_result( + handle, "Failed to create builder from archive") try: # A builder from an archive here carries no context. @@ -3585,10 +3757,11 @@ def _init_from_context(self, context, json_str): context.execution_context), Builder._ERROR_MESSAGES['builder_error']) - self._consume_and_swap( - lambda handle: _lib.c2pa_builder_with_definition( - handle, json_str), - Builder._ERROR_MESSAGES['builder_error']) + with self._native_call(): + self._consume_and_swap( + lambda handle: _lib.c2pa_builder_with_definition( + handle, json_str), + Builder._ERROR_MESSAGES['builder_error']) def _init_attrs(self): super()._init_attrs() @@ -3920,9 +4093,10 @@ def _sign_internal( try: # _native_call covers the signing call only. - # The close() below is deliberately outside it, - # so the deferred teardown it records is performed - # on the way out rather than being deferred forever. + # The result check and the close() are deliberately + # outside of it: the check needs its own, later section, + # and close() runs only once that check has read whatever + # error this call set. with self._native_call(): if signer is not None: # Signer needs its own in-flight guard. @@ -3945,18 +4119,25 @@ def _sign_internal( dest_stream._stream, ctypes.byref(manifest_bytes_ptr), ) - # Sign borrows the Builder without taking ownership. - # Closing here ensures resources clean up, - # and single use/single sign done by a Builder. - self.close() except Exception as e: self.close() raise C2paError(f"Error during signing: {e}") from e - _check_ffi_operation_result( - result, - "Error during signing", - check=lambda r: r < 0) + try: + # Own section (the native_call already closed, so its + # own reads are done): close() can free this Builder, + # and freeing can write to the same thread-local error slot + # this check reads. + with _native_section(): + _check_ffi_operation_result( + result, + "Error during signing", + check=lambda r: r < 0) + finally: + # Sign borrows the Builder without taking ownership. + # Closing here ensures resources clean up, and single + # use/single sign done by a Builder. + self.close() # Capture the manifest bytes if available manifest_bytes = b"" @@ -4188,17 +4369,18 @@ def format_embeddable(format: str, manifest_bytes: bytes) -> tuple[int, bytes]: ) result_bytes_ptr = ctypes.POINTER(ctypes.c_ubyte)() - result = _lib.c2pa_format_embeddable( - format_str, - manifest_array, - len(manifest_bytes), - ctypes.byref(result_bytes_ptr) - ) + with _native_section(): + result = _lib.c2pa_format_embeddable( + format_str, + manifest_array, + len(manifest_bytes), + ctypes.byref(result_bytes_ptr) + ) - _check_ffi_operation_result( - result, - "Failed to format embeddable manifest", - check=lambda r: r < 0) + _check_ffi_operation_result( + result, + "Failed to format embeddable manifest", + check=lambda r: r < 0) size = result try: @@ -4320,14 +4502,15 @@ def ed25519_sign(data: bytes, private_key: str) -> bytes: f"Invalid UTF-8 characters in private key: {str(e)}") # Perform the signing operation - signature_ptr = _lib.c2pa_ed25519_sign( - data_array, - data_size, - key_bytes - ) + with _native_section(): + signature_ptr = _lib.c2pa_ed25519_sign( + data_array, + data_size, + key_bytes + ) - _check_ffi_operation_result(signature_ptr, - "Failed to sign data with Ed25519") + _check_ffi_operation_result(signature_ptr, + "Failed to sign data with Ed25519") try: # Ed25519 signatures are always 64 bytes diff --git a/tests/test_unit_tests.py b/tests/test_unit_tests.py index d8c09bdd..59e679b3 100644 --- a/tests/test_unit_tests.py +++ b/tests/test_unit_tests.py @@ -30,6 +30,7 @@ import shutil import ctypes import threading +import concurrent.futures # Suppress deprecation warnings warnings.simplefilter("ignore", category=DeprecationWarning) @@ -50,6 +51,15 @@ ALTERNATIVE_INGREDIENT_TEST_FILE = os.path.join(FIXTURES_DIR, "cloud.jpg") +def _fail_with_native_error(tag_bytes): + """Build a mock FFI callable that sets a native error and returns None. + """ + def _mock(*args): + c2pa_module._lib.c2pa_error_set_last(tag_bytes) + return None + return _mock + + def load_test_settings_json(): """ Load default (legacy) trust configuration test settings from a @@ -8276,12 +8286,11 @@ def test_construction_failure_leaves_nothing_to_free(self): c2pa_module._lib.c2pa_builder_from_json = real_json def test_context_build_null_return_frees_builder(self): - # Set a pre-consume tag in the error slot to mock a pointer rejection. + # Mock a pointer rejection. settings = Settings() - c2pa_module._lib.c2pa_error_set_last( - b"UntrackedPointer: mocked pre-consume rejection") real_build = c2pa_module._lib.c2pa_context_builder_build - c2pa_module._lib.c2pa_context_builder_build = lambda ptr: None + c2pa_module._lib.c2pa_context_builder_build = _fail_with_native_error( + b"UntrackedPointer: mocked pre-consume rejection") try: with self.assertRaises(Error): Context(settings=settings) @@ -8340,6 +8349,172 @@ def test_consume_no_replacement_marks_consumed_on_other_error(self): self.assertIsNone(res._handle) self.assertEqual(res._lifecycle_state, LifecycleState.CLOSED) + def test_invoke_consume_success_does_not_consult_error_slot(self): + """A successful consuming call must not read the error slot at all: + only a failure inspects it.""" + res = self._FakeHandleResource() + res._activate(0xCAFE) + + res._consume_no_replacement(lambda h: 0, "set failed: {}") + + self.assertIsNone(c2pa_module._read_native_error()) + + def test_consume_no_replacement_retains_on_tag_set_by_the_call_itself(self): + """Only a *stale* tag left over from before the call is the + thing being defended against.""" + res = self._FakeHandleResource() + res._activate(0xCAFE) + + def fake_call(handle): + c2pa_module._lib.c2pa_error_set_last( + b"UntrackedPointer: rejected by the call itself") + return -1 + + with self.assertRaises(Error): + res._consume_no_replacement(fake_call, "set failed: {}") + + # Rejected before ownership transferred: handle retained. + self.assertEqual(res._handle, 0xCAFE) + self.assertEqual(res._lifecycle_state, LifecycleState.ACTIVE) + self.assertEqual(self.freed, []) + res.close() + self.assertEqual(self.freed, [0xCAFE]) + + def test_native_section_defers_unrelated_finalizer_free(self): + """A finalizer for a completely unrelated resource firing mid + native-call must not free immediately. + """ + victim = self._FakeHandleResource() + victim._activate(0xCAFE) + bystander = self._FakeHandleResource() + bystander._activate(0xB00B) + + def polluting_free(ptr): + self.freed.append(ptr) + # Freeing and untracked/ pointer writes its own error into the + # same thread-local slot. + c2pa_module._lib.c2pa_error_set_last( + "Other: UntrackedPointer: {:#x}".format(ptr).encode()) + return -1 + ManagedResource._free_native_ptr = staticmethod(polluting_free) + + def ffi_call(handle): + nonlocal bystander + del bystander # last reference dropped: __del__ fires right here + return None # the real call failed but set no error of its own + + with victim._native_call(): + with self.assertRaises(Error): + victim._consume_no_replacement(ffi_call, "op failed: {}") + + self.assertIsNone( + victim._handle, + "victim was wrongly retained") + self.assertEqual(victim._lifecycle_state, LifecycleState.CLOSED) + self.assertEqual(self.freed, [0xB00B], + "bystander's deferred free did not run exactly once") + + def test_teardown_deferred_by_own_inflight_and_section_together(self): + """A resource blocked by its own handle being in-flight, + and a wholly separate native-error section is also open on this thread + must not free until both clear, and must free exactly once.""" + res = self._FakeHandleResource() + res._activate(0xCAFE) + + call_cm = res._native_call() + call_cm.__enter__() + try: + section_cm = c2pa_module._native_section() + section_cm.__enter__() + try: + res.close() + self.assertEqual(res._lifecycle_state, LifecycleState.CLOSED) + self.assertEqual(self.freed, [], + "freed while still in flight") + finally: + section_cm.__exit__(None, None, None) + # The independent section closed, but res's own in-flight + # guard is still up: still not freed. + self.assertEqual(self.freed, [], + "flushed while the in-flight guard still held") + finally: + call_cm.__exit__(None, None, None) + # Both gates clear only once native_call's own exit drops inflight + # to 0 -- that is what should finally trigger the free. + self.assertEqual(self.freed, [0xCAFE]) + + def test_nested_native_sections_flush_only_at_outermost_close(self): + """A native-error section opened inside another, already-open one + on the same thread must not flush anything until the outermost + one closes.""" + res = self._FakeHandleResource() + res._activate(0xCAFE) + + outer = c2pa_module._native_section() + outer.__enter__() + try: + inner = c2pa_module._native_section() + inner.__enter__() + try: + res.close() + self.assertEqual(self.freed, []) + finally: + inner.__exit__(None, None, None) + # Inner closed, outer is still open: still deferred. + self.assertEqual(self.freed, [], + "inner section flushed before the outer closed") + finally: + outer.__exit__(None, None, None) + self.assertEqual(self.freed, [0xCAFE]) + + def test_native_section_flush_isolates_exceptions(self): + """One deferred free raising during a section's flush must not + stop the rest of that flush from running.""" + good = self._FakeHandleResource() + good._activate(0xC0FFEE) + bad = self._FakeHandleResource() + bad._activate(0xBAD) + + def flaky_free(ptr): + if ptr == 0xBAD: + raise RuntimeError("simulated free failure") + self.freed.append(ptr) + return 0 + ManagedResource._free_native_ptr = staticmethod(flaky_free) + + with self.assertLogs('c2pa', level='ERROR') as captured: + with c2pa_module._native_section(): + bad.close() + good.close() + + self.assertEqual(self.freed, [0xC0FFEE], + "a failing deferred free stopped the rest") + self.assertTrue( + any('Failed to free native' in line + for line in captured.output), + "the failing deferred free was not logged: " + "{}".format(captured.output)) + + def test_stale_error_not_misattributed_after_preset_error(self): + """A stale tag left by an earlier, unrelated call on this thread + must not be read as this call's own error.""" + # A stale tag from an earlier, unrelated call. + c2pa_module._lib.c2pa_error_set_last( + b"Other: UntrackedPointer: 0xdeadbeef") + + res = self._FakeHandleResource() + res._activate(0xCAFE) + + # Fails without setting any error of its own. + # The sentinel inside _invoke_consume must have cleared + # the stale tag, so this routes to the "no error of our own" branch. + with self.assertRaises(Error): + res._consume_no_replacement(lambda h: -1, "op failed: {}") + + self.assertIsNone(res._handle) + self.assertEqual(res._lifecycle_state, LifecycleState.CLOSED) + self.assertEqual(self.freed, [], "consumed branch must not free") + class TestManagedResourceObjects(TestContextAPIs): """Tests native resource handling management when managed manually. @@ -8601,9 +8776,9 @@ def test_builder_with_archive_null_return_marks_consumed(self): # Mimic a non-tag error: native took ownership then failed and dropped # the value itself, so the handle is marked consumed, not freed. - c2pa_module._lib.c2pa_error_set_last(b"Other: mocked test error") real_call = c2pa_module._lib.c2pa_builder_with_archive - c2pa_module._lib.c2pa_builder_with_archive = lambda b, s: None + c2pa_module._lib.c2pa_builder_with_archive = _fail_with_native_error( + b"Other: mocked test error") # Instrument before the failure... freed = self._instrument_frees() @@ -8637,11 +8812,9 @@ def test_reader_with_fragment_null_return_marks_consumed(self): # Mimic a non-tag error: native took ownership then failed and dropped # the value itself, so the handle is marked consumed, not freed. - c2pa_module._lib.c2pa_error_set_last(b"Other: mocked test error") - real_call = c2pa_module._lib.c2pa_reader_with_fragment - c2pa_module._lib.c2pa_reader_with_fragment = ( - lambda r, f, s, frag: None) + c2pa_module._lib.c2pa_reader_with_fragment = _fail_with_native_error( + b"Other: mocked test error") # Instrument before failure so any free would be counted. freed = self._instrument_frees() @@ -8970,10 +9143,9 @@ def test_unknown_failure_drops_handle_without_freeing(self): consumed_handle = reader._handle # Simulate an error being set - c2pa_module._lib.c2pa_error_set_last(b"Other: mocked test error") real_call = c2pa_module._lib.c2pa_reader_with_fragment - c2pa_module._lib.c2pa_reader_with_fragment = ( - lambda r, f, s, frag: None) + c2pa_module._lib.c2pa_reader_with_fragment = _fail_with_native_error( + b"Other: mocked test error") try: with open(init_path, "rb") as init, \ open(fragment_path, "rb") as frag: @@ -9084,17 +9256,14 @@ def test_perf_scenario_bogus_handle_is_rejected(self): reader.close() def test_every_null_return_sets_its_own_error(self): - # Reading the slot without clearing it is only sound because every - # null return sets an error. Check each path reports its own. + # Each null-returning path must report the error it set itself, never + # one left behind by an earlier call. init_path = os.path.join(FIXTURES_DIR, "dashinit.mp4") fragment_path = os.path.join(FIXTURES_DIR, "dash1.m4s") - # Leave a recognisable error behind, so anything stale shows up. - try: - Reader("image/jpeg", io.BytesIO(b"not an image")).json() - except Error: - pass - self.assertIn("NotSupported", c2pa_module._read_native_error() or "") + # Set a recognizable error, so anything stale shows up below. + c2pa_module._lib.c2pa_error_set_last( + b"NotSupported: planted by the test") # Pre-consume rejection: reports UntrackedPointer, not NotSupported. with open(init_path, "rb") as init: @@ -9164,21 +9333,19 @@ def worker(): self.assertEqual(problems, [], "ownership was misjudged under concurrency") - def test_reading_the_native_error_does_not_empty_the_slot(self): - # c2pa_error() peeks, so nothing Python can call empties the slot. - # _consume_and_swap depends on this. - try: - Reader("image/jpeg", io.BytesIO(b"not an image")).json() - except Error: - pass + def test_reading_the_native_error_consumes_it(self): + # c2pa_error() itself peeks, so _read_native_error marks the slot as + # carrying no error once it has read one. + # An error belongs to the caller that observes it; + # leaving it readable lets a later, unrelated failure report it as its own. + c2pa_module._lib.c2pa_error_set_last(b"Io: read me exactly once") first = c2pa_module._read_native_error() self.assertTrue(first, "expected a native error to have been set") - self.assertEqual( - c2pa_module._read_native_error(), first, - "reading emptied the native slot; the comments in " - "_consume_and_swap about a persistent error are now wrong") + self.assertIsNone( + c2pa_module._read_native_error(), + "the native error stayed readable after being reported once") def test_read_native_error_returns_none_for_an_empty_message(self): # c2pa_error() returns an owned pointer to "" when no error is set, @@ -9196,22 +9363,29 @@ def test_read_native_error_returns_none_for_an_empty_message(self): finally: c2pa_module._lib.c2pa_error = original - def test_mocked_null_without_error_is_a_known_limitation(self): - # A null with no error of its own is the case that breaks: the slot - # still holds whatever came before. No native path does this, so it - # is pinned here rather than defended in _consume_and_swap. + def test_null_return_with_no_native_error_is_treated_as_consumed(self): + # A null with no error of its own used to be the case that broke: + # the slot still held whatever an unrelated, earlier call on this same + # (pooled) thread left behind, and a stale UntrackedPointer/ + # WrongPointerType tag would make this call believe it still owned a + # handle the native side already dropped. init_path = os.path.join(FIXTURES_DIR, "dashinit.mp4") fragment_path = os.path.join(FIXTURES_DIR, "dash1.m4s") + # A stale, unrelated tag left by a prior call on this thread. c2pa_module._lib.c2pa_error_set_last( b"UntrackedPointer: 0xdeadbeef") with open(init_path, "rb") as init: reader = Reader("video/mp4", init) + consumed_handle = reader._handle real_call = c2pa_module._lib.c2pa_reader_with_fragment + # The fake native call sets no error of its own, + # the planted sentinel _invoke_consume is left in the slot. c2pa_module._lib.c2pa_reader_with_fragment = ( lambda r, f, s, frag: None) + freed = self._instrument_frees() try: with open(init_path, "rb") as init, \ open(fragment_path, "rb") as frag: @@ -9219,16 +9393,12 @@ def test_mocked_null_without_error_is_a_known_limitation(self): reader.with_fragment("video/mp4", init, frag) finally: c2pa_module._lib.c2pa_reader_with_fragment = real_call - # Nothing clears the slot, so a planted tag would follow other - # tests around and change how their failures are classified. - c2pa_module._lib.c2pa_error_set_last( - b"Other: cleared by test teardown") - # The stale tag wins, so the handle is kept. Safe here (the mock - # consumed nothing), and the reader is still usable. - self.assertIsNotNone(reader._handle) - self.assertEqual(reader._lifecycle_state, LifecycleState.ACTIVE) - reader.close() + # The sentinel survived, not the stale tag. + self.assertIsNone(reader._handle) + self.assertEqual(reader._lifecycle_state, LifecycleState.CLOSED) + self.assertEqual(self._free_count(freed, consumed_handle), 0, + "consumed handle was freed instead of marked consumed") # Backfilling a pointer minted by a direct FFI call. Builder.from_archive # is the only production caller of _wrap_native_handle, so these are the @@ -9377,10 +9547,9 @@ def test_consumed_reader_closes_backing_file(self): self.assertFalse(backing_file.closed) # Simulate an error being set - c2pa_module._lib.c2pa_error_set_last(b"Other: mocked test error") real_call = c2pa_module._lib.c2pa_reader_with_fragment - c2pa_module._lib.c2pa_reader_with_fragment = ( - lambda r, f, s, frag: None) + c2pa_module._lib.c2pa_reader_with_fragment = _fail_with_native_error( + b"Other: mocked test error") try: with open(DEFAULT_TEST_FILE, "rb") as main, \ open(DEFAULT_TEST_FILE, "rb") as frag: @@ -9399,9 +9568,9 @@ def test_consumed_builder_releases_context(self): archive = self._make_archive() # Simulate an error being set - c2pa_module._lib.c2pa_error_set_last(b"Other: mocked test error") real_call = c2pa_module._lib.c2pa_builder_with_archive - c2pa_module._lib.c2pa_builder_with_archive = lambda b, s: None + c2pa_module._lib.c2pa_builder_with_archive = _fail_with_native_error( + b"Other: mocked test error") try: with self.assertRaises(Error): builder.with_archive(archive) @@ -9448,10 +9617,9 @@ def test_consumed_reader_clears_caches(self): self.assertIsNotNone(reader._manifest_json_str_cache) # Simulate an error being set - c2pa_module._lib.c2pa_error_set_last(b"Other: mocked test error") real_call = c2pa_module._lib.c2pa_reader_with_fragment - c2pa_module._lib.c2pa_reader_with_fragment = ( - lambda r, f, s, frag: None) + c2pa_module._lib.c2pa_reader_with_fragment = _fail_with_native_error( + b"Other: mocked test error") try: with open(DEFAULT_TEST_FILE, "rb") as main, \ open(DEFAULT_TEST_FILE, "rb") as frag: @@ -9523,6 +9691,36 @@ def _boom(*args): self.assertIs(ctx.exception.__cause__, sentinel, "signing error dropped the original exception") + def test_sign_reports_the_native_error_it_set(self): + """sign() reads its error in a later section than the call itself. + The signing call runs inside one _native_call() block and the result + check runs in a separate _native_section() afterwards, so anything + that marks the slot as carrying no error on section exit would discard + the real message between the two. + """ + builder = Builder(self.test_manifest) + signer = self._ctx_make_signer() + self.addCleanup(signer.close) + + real_sign = c2pa_module._lib.c2pa_builder_sign + + def _fail(*args): + c2pa_module._lib.c2pa_error_set_last( + b"Signature: native signing refused") + return -1 + + c2pa_module._lib.c2pa_builder_sign = _fail + try: + with self.assertRaises(Error) as ctx: + builder.sign(signer, "image/jpeg", + io.BytesIO(b"x"), io.BytesIO()) + finally: + c2pa_module._lib.c2pa_builder_sign = real_sign + + self.assertIn("native signing refused", str(ctx.exception), + "the native signing error was lost before it was read") + self.assertIsInstance(ctx.exception, Error.Signature) + class TestErrorPlumbing(unittest.TestCase): """Covers the error helpers themselves, which had no direct tests.""" @@ -9654,6 +9852,184 @@ def test_supported_mime_types_reports_the_native_message(self): c2pa_module._get_supported_mime_types(lambda count: None, None) self.assertIn("mime lookup failed", str(ctx.exception)) + def test_reading_an_error_does_not_leave_it_readable(self): + """An error is reportable once, by the reader that observes it. + """ + self._set_native_error("Io: read me once") + + self.assertEqual( + c2pa_module._read_native_error(), "Io: read me once") + self.assertIsNone( + c2pa_module._read_native_error(), + "the same native error was reported a second time") + + def test_handled_error_does_not_survive_later_operations(self): + """A caught failure must not leave its error in-place + (tests the slot is cleaned up). + """ + with self.assertRaises(Error): + Reader("image/jpeg", io.BytesIO(b"not an image")) + + for _ in range(20): + c2pa_module.Stream(io.BytesIO(b"x")) + + self.assertIsNone( + c2pa_module._read_native_error(), + "a handled error was still resident after 20 successful calls") + + def test_later_failure_does_not_inherit_a_handled_errors_type(self): + """A failure with no error of its own must not see an older one. + """ + with self.assertRaises(Error) as first: + Reader("image/jpeg", io.BytesIO(b"not an image")) + self.assertIsInstance(first.exception, Error.NotSupported) + + with self.assertRaises(Error) as second: + c2pa_module._check_ffi_operation_result( + None, "Later unrelated failure: {}") + + self.assertNotIsInstance( + second.exception, Error.NotSupported, + "the later failure inherited the handled error's type") + self.assertIn("Unknown error", str(second.exception)) + self.assertNotIn( + "type is unsupported", str(second.exception), + "the later failure reported the handled error's message") + + def test_the_no_native_error_sentinel_never_reaches_a_caller(self): + """The sentinel is an internal marker, not a message for users.""" + sentinel = c2pa_module._NO_NATIVE_ERROR_TEXT + + c2pa_module._mark_sentinel_no_native_error() + self.assertIsNone( + c2pa_module._read_native_error(), + "the sentinel was reported as if it were a native error") + + c2pa_module._mark_sentinel_no_native_error() + with self.assertRaises(Error) as ctx: + c2pa_module._check_ffi_operation_result(None, "fallback: {}") + self.assertNotIn(sentinel, str(ctx.exception)) + self.assertIn("Unknown error", str(ctx.exception)) + + def test_mark_sentinel_writes_the_learned_text(self): + c2pa_module._mark_sentinel_no_native_error() + raw = c2pa_module._lib.c2pa_error() + try: + text = ctypes.string_at(raw).decode('utf-8') + finally: + c2pa_module._lib.c2pa_string_free(raw) + self.assertEqual(text, c2pa_module._NO_NATIVE_ERROR_TEXT) + + def test_read_native_error_maps_sentinel_to_none(self): + c2pa_module._mark_sentinel_no_native_error() + self.assertIsNone(c2pa_module._read_native_error()) + + def test_runtime_does_not_call_error_set_last(self): + """The marker mechanism must not depend on c2pa_error_set_last, + so this module loads against native builds that lack it.""" + for fn in (c2pa_module.ManagedResource._invoke_consume, + c2pa_module._read_native_error, + c2pa_module._mark_sentinel_no_native_error): + self.assertNotIn( + 'c2pa_error_set_last', inspect.getsource(fn)) + + +class TestMarkerOutlivesPointerConsumptionSemantics(unittest.TestCase): + """The marker is needed for reasons independent of pointer ownership. + + The native error slot is sticky and thread-local, so failure paths + that carry no still need to tell an error this call set from an + earlier, unrelated call left behind. + """ + + def setUp(self): + # Leave no message from an earlier test in this thread's slot. + c2pa_module._mark_sentinel_no_native_error() + + def test_non_consuming_failure_does_not_inherit_a_read_error(self): + c2pa_module._lib.c2pa_error_set_last(b"Signature: earlier task") + # The rightful owner reports it, which re-marks the slot. + self.assertEqual( + c2pa_module._read_native_error(), "Signature: earlier task") + + # A later, unrelated failure that sets no error of its own must + # report its own fallback, not the message above. + with self.assertRaises(Error) as ctx: + c2pa_module._check_ffi_operation_result( + 0, "later op failed: {}", check=lambda r: r == 0) + + self.assertNotIn("earlier task", str(ctx.exception)) + self.assertIn("Unknown error", str(ctx.exception)) + self.assertNotIsInstance(ctx.exception, Error.Signature) + + def test_settings_set_failure_reports_its_own_error(self): + settings = Settings() + self.addCleanup(settings.close) + + c2pa_module._lib.c2pa_error_set_last(b"Signature: earlier task") + self.assertEqual( + c2pa_module._read_native_error(), "Signature: earlier task") + + with self.assertRaises(Error) as ctx: + settings.set("builder.thumbnail.enabled", "not-a-json-value") + + self.assertNotIn("earlier task", str(ctx.exception)) + + def test_marker_is_per_thread_across_pooled_reuse(self): + """The slot is thread-local, so a pooled worker must not hand one + task's error to the next task that runs on it.""" + def failing_task(): + c2pa_module._lib.c2pa_error_set_last(b"Io: first task") + return c2pa_module._read_native_error() + + def quiet_task(): + # Sets no error; must not see the previous task's message. + return c2pa_module._read_native_error() + + # One worker guarantees both tasks run on the same OS thread. + with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool: + self.assertEqual(pool.submit(failing_task).result(), + "Io: first task") + self.assertIsNone( + pool.submit(quiet_task).result(), + "a pooled thread carried an error across unrelated tasks") + + def test_one_thread_marker_does_not_clear_another_threads_error(self): + """Marking on one thread must leave another thread's pending error + readable: the slot is per thread, and so is the marker.""" + set_on_worker = threading.Event() + marked_on_main = threading.Event() + seen = {} + + def worker(): + c2pa_module._lib.c2pa_error_set_last(b"Io: worker error") + set_on_worker.set() + self.assertTrue(marked_on_main.wait(5)) + seen["worker"] = c2pa_module._read_native_error() + + thread = threading.Thread(target=worker, daemon=True) + thread.start() + self.assertTrue(set_on_worker.wait(5)) + + c2pa_module._mark_sentinel_no_native_error() + marked_on_main.set() + thread.join(5) + + self.assertEqual(seen.get("worker"), "Io: worker error") + + def test_marker_path_is_reached_without_any_consuming_call(self): + """The non-consuming path reaches the marker through _read_native_error, + never through _invoke_consume.""" + self.assertIn("_read_native_error", + inspect.getsource( + c2pa_module._check_ffi_operation_result)) + self.assertNotIn("_invoke_consume", + inspect.getsource( + c2pa_module._check_ffi_operation_result)) + # _read_native_error is what re-marks the slot after every read. + self.assertIn("_mark_sentinel_no_native_error", + inspect.getsource(c2pa_module._read_native_error)) + class TestErrorsStillRaiseAfterCleanup(unittest.TestCase): """Each surface that lost a _clear_error_state() call still reports.""" diff --git a/tests/test_unit_tests_threaded.py b/tests/test_unit_tests_threaded.py index 001ab50e..bf941375 100644 --- a/tests/test_unit_tests_threaded.py +++ b/tests/test_unit_tests_threaded.py @@ -33,7 +33,7 @@ from c2pa import Builder, C2paError as Error, Reader, C2paSigningAlg as SigningAlg, C2paSignerInfo, Signer, sdk_version # noqa: E501 from c2pa import Context, Settings -from c2pa.c2pa import ManagedResource, Stream, LifecycleState +from c2pa.c2pa import ManagedResource, Stream, LifecycleState, _native_section from c2pa.lib import is_foreign_process, record_owner_pid PROJECT_PATH = os.getcwd() @@ -3568,31 +3568,35 @@ def test_no_nested_op_locks(self): held = threading.local() violations = [] real_lock = ManagedResource._lock - - def tracking_lock(resource): - lock = real_lock(resource) - depth = getattr(held, 'stack', None) - if depth is None: - depth = held.stack = [] - - class Tracked: - def __enter__(self): - others = [r for r in depth if r is not resource] - if others: - violations.append( - "{} while holding {}".format( - type(resource).__name__, - [type(o).__name__ for o in others])) - depth.append(resource) - return lock.__enter__() - - def __exit__(self, *exc): - depth.pop() - return lock.__exit__(*exc) - - return Tracked() - - ManagedResource._lock = tracking_lock + real_state_lock = ManagedResource._state_lock + + def make_tracking(real): + def tracking(resource): + lock = real(resource) + depth = getattr(held, 'stack', None) + if depth is None: + depth = held.stack = [] + + class Tracked: + def __enter__(self): + others = [r for r in depth if r is not resource] + if others: + violations.append( + "{} while holding {}".format( + type(resource).__name__, + [type(o).__name__ for o in others])) + depth.append(resource) + return lock.__enter__() + + def __exit__(self, *exc): + depth.pop() + return lock.__exit__(*exc) + + return Tracked() + return tracking + + ManagedResource._lock = make_tracking(real_lock) + ManagedResource._state_lock = make_tracking(real_state_lock) try: reader = Reader("image/jpeg", io.BytesIO(data)) reader.json() @@ -3602,6 +3606,7 @@ def __exit__(self, *exc): reader.close() finally: ManagedResource._lock = real_lock + ManagedResource._state_lock = real_state_lock self.assertEqual(violations, [], "a thread held two operation locks at once") @@ -3643,6 +3648,49 @@ def closer_worker(): self._join_all(threads, "concurrent storm") self.assertEqual(errors, []) + def test_native_section_deferred_free_is_thread_local(self): + """Two threads each with their own open native-error section: one + thread's section closing must not flush a free deferred inside + the other thread's still-open section. + """ + freed = self._counted_free() + resource = _ConcreteResource() + resource._activate(0x1001) + + thread_ready = threading.Event() + release_thread = threading.Event() + + def worker(): + with _native_section(): + resource.close() + thread_ready.set() + release_thread.wait(self.JOIN_TIMEOUT) + # Flush happens here, on the worker thread, once its own + # section closes. + + thread = threading.Thread(target=worker) + thread.start() + try: + self.assertTrue( + thread_ready.wait(self.JOIN_TIMEOUT), + "worker thread did not reach its open section in time") + + # A section opened and closed entirely on this (main) thread, + # while the worker's section is still open on its own thread. + with _native_section(): + pass + + self.assertEqual( + freed, [], + "a different thread's section flushed this thread's " + "pending resource") + finally: + release_thread.set() + self._join_all([thread], "native-section worker") + + self.assertEqual(freed, [0x1001], + "worker thread's own section never flushed") + def _counted_free(self): """Patch _free_native_ptr to count frees; returns the list.""" freed = []