diff --git a/examples/read.py b/examples/read.py index e4b718a9..b2ef9fc6 100644 --- a/examples/read.py +++ b/examples/read.py @@ -36,7 +36,7 @@ def read_c2pa_data(media_path: str): # All objects using this context will have trust configured. with c2pa.Context(settings) as context: with c2pa.Reader(media_path, context=context) as reader: - print(reader.detailed_json()) + print(reader.crjson()) except Exception as e: print(f"Error reading C2PA data from {media_path}: {e}") sys.exit(1) diff --git a/examples/sign.py b/examples/sign.py index e6c14859..09133b35 100644 --- a/examples/sign.py +++ b/examples/sign.py @@ -110,6 +110,6 @@ def callback_signer_es256(data: bytes) -> bytes: # The validation state will depend on loaded trust settings. # Without loaded trust settings, # the manifest validation_state will be "Invalid". - print(reader.json()) + print(reader.crjson()) print("\nExample completed successfully!") diff --git a/examples/sign_info.py b/examples/sign_info.py index 6b256647..3735ad0b 100644 --- a/examples/sign_info.py +++ b/examples/sign_info.py @@ -40,7 +40,7 @@ print("\nReading existing C2PA metadata:") with open(fixtures_dir + "C.jpg", "rb") as file: with c2pa.Reader("image/jpeg", file) as reader: - print(reader.json()) + print(reader.crjson()) # Create a signer from certificate and key files with open(fixtures_dir + "es256_certs.pem", "rb") as cert_file: @@ -103,7 +103,7 @@ # The validation state will depend on loaded trust settings. # Without loaded trust settings, # the manifest validation_state will be "Invalid". - print(reader.json()) + print(reader.crjson()) print("\nExample completed successfully!") diff --git a/src/c2pa/c2pa.py b/src/c2pa/c2pa.py index 92d24eff..02447e67 100644 --- a/src/c2pa/c2pa.py +++ b/src/c2pa/c2pa.py @@ -230,19 +230,35 @@ class ManagedResource: for native resources (e.g. pointers). Subclasses must: - - Set `self._handle` to the native pointer after creation. - - Set `self._lifecycle_state = LifecycleState.ACTIVE` once initialized. + - Call `_activate(handle)` once the native pointer is created and + validated, which takes ownership of it and marks the resource active. + Never assign `self._handle` or `self._lifecycle_state` directly. + - Call `_swap_handle(new_handle)` instead when an FFI call consumed the + current handle and returned a replacement. + - Call `_mark_consumed()` when an FFI call took ownership of the handle + without returning a replacement. - Override `_release()` to free class-specific resources (streams, caches, callbacks, etc.), called before the native pointer is freed. + - Override `_init_attrs()` to set the class's own attributes to their + defaults, and call it from `__init__`. `_wrap_native_handle` calls it + too, so an instance built around an existing handle is never missing + the attributes the rest of the class reads. The native pointer is freed automatically via `_free_native_ptr`. """ + def _init_attrs(self): + """Set this class's own attributes to their defaults. + + Called by __init__ and by _wrap_native_handle, which bypasses + __init__. Keeping the defaults here means an instance built around an + existing handle is never missing what the rest of the class reads. + """ + def __init__(self): self._lifecycle_state = LifecycleState.UNINITIALIZED self._handle = None - _clear_error_state() record_owner_pid(self) @staticmethod @@ -265,7 +281,6 @@ def _ensure_valid_state(self): if not self._handle: raise C2paError( f"{name} has an invalid internal state (active but no handle)") - _clear_error_state() def _release(self): """Override to free class-specific resources (streams, caches, etc.). @@ -274,26 +289,188 @@ def _release(self): The default implementation does nothing. """ + def _safe_release(self): + """Run _release(), logging (never raising) if it fails. + """ + try: + self._release() + except Exception: + logger.error( + "Failed to release %s resources", + type(self).__name__, + exc_info=True, + ) + def _mark_consumed(self): """Mark as consumed by an FFI call that took ownership of native resources e.g. pointers. This means we should not call clean-up here anymore, and leave it to the new owner. """ + if is_foreign_process(self): + self._handle = None + self._lifecycle_state = LifecycleState.CLOSED + return + + # Callers raise straight after consuming, so a failing _release() + # here would mask the error they are reporting. + self._safe_release() + self._handle = None self._lifecycle_state = LifecycleState.CLOSED + def _activate(self, handle): + """Attach a native handle to self and mark it active. + Ownership of `handle` transfers here. + + Args: + handle: Non-null native pointer to take ownership of + + Raises: + C2paError: If the handle is null, + or the resource is not uninitialized + """ + name = type(self).__name__ + # A rejected activation must leave the object as it was. + if not handle: + raise C2paError(f"{name}: cannot activate a null handle") + if self._lifecycle_state != LifecycleState.UNINITIALIZED: + raise C2paError( + f"{name}: already activated " + f"({self._lifecycle_state.name})") + + self._handle = handle + self._lifecycle_state = LifecycleState.ACTIVE + + def _swap_handle(self, new_handle): + """Replace the handle after an FFI call consumed the old one and + returned a replacement. + A null return from such a call is ambiguous (the callee may have + failed validation before taking ownership, or failed the operation + after), so callers must not call this with a null replacement. + Requires the resource to be active. + + Args: + new_handle: Non-null native pointer returned by the FFI call + + Raises: + C2paError: If the resource is not ACTIVE or new_handle is null + """ + name = type(self).__name__ + if self._lifecycle_state != LifecycleState.ACTIVE: + raise C2paError( + f"{name}: cannot swap the handle of a resource that is not " + f"active ({self._lifecycle_state.name})") + if not new_handle: + raise C2paError(f"{name}: cannot swap in a null handle") + + self._handle = new_handle + + # Errors set by native lib, hinting at the cause of the error + # These errors here means the pointer got somehow rejected by the lib, + # so it is still ours to deal with. + _PRE_CONSUME_ERROR_TAGS = ("UntrackedPointer:", "WrongPointerType:") + + def _consume_and_swap(self, ffi_call, error_message): + """Run an FFI call that consumes this handle and returns a replacement. + + The native lib may take ownership partway through the call. + The native error tells if ownership was transferred: + - a pointer rejection (`_PRE_CONSUME_ERROR_TAGS`) precedes the + transfer, so the handle is still ours and is kept; + - any other error means it was taken, then the operation failed. + + The error is read without clearing it first. + The slot is sticky: c2pa_error() peeks and nothing empties it. + + Args: + ffi_call: Callable taking the current handle, returning the + replacement or null. + error_message: Format string with one placeholder, used when the + native layer offers no error of its own. + + Raises: + C2paError: If the call fails, typed by native error when one. + """ + try: + new_ptr = ffi_call(self._handle) + except ctypes.ArgumentError: + # Marshalling failed, so the call never reached the native side + # and the handle is untouched. + raise + except BaseException as e: + self._mark_consumed() + raise C2paError(error_message.format(e)) from e + + if new_ptr: + self._swap_handle(new_ptr) + return + + error = _read_native_error() + if error: + if any(tag in error + for tag in ManagedResource._PRE_CONSUME_ERROR_TAGS): + # Rejected before ownership transferred, + # so the handle is still ours. + logger.warning( + "%s: native call rejected the handle before taking " + "ownership (%s); handle retained", + type(self).__name__, + error) + _raise_typed_c2pa_error(error) + + # Ownership transferred and then an operation failed. + self._mark_consumed() + _raise_typed_c2pa_error(error) + + self._mark_consumed() + raise C2paError(error_message.format("Unknown error")) + + @classmethod + def _wrap_native_handle(cls, handle): + """Build a brand-new instance around an already-valid, + already-owned native handle (bypassing __init__). + + Everything an instance needs besides the native handle must be set in + `_init_attrs()`. `_wrap_native_handle` never runs `__init__`. + An attribute a subclass sets only in `__init__` will be + missing (or left at its `_init_attrs` default) on a wrapped instance. + + Ownership of `handle` transfers only on successful return. If this + raises, the caller still owns the pointer and must free it. + + Args: + handle: Non-null native pointer to take ownership of + + Raises: + C2paError: If the handle is null + """ + obj = object.__new__(cls) + ManagedResource.__init__(obj) + obj._init_attrs() + obj._activate(handle) + return obj + def _cleanup_resources(self): """Release native resources idempotently.""" try: if is_foreign_process(self): + # A forked child holds a separate copy of this object and the + # parent still owns the real handle and frees it. Mark this + # copy closed and null its handle so the child cannot mistake + # it for usable or free it, but do not free here. + # Mutating this copy does not touch the parent's. + if hasattr(self, '_handle'): + self._handle = None + if hasattr(self, '_lifecycle_state'): + self._lifecycle_state = LifecycleState.CLOSED return if ( hasattr(self, '_lifecycle_state') and self._lifecycle_state != LifecycleState.CLOSED ): self._lifecycle_state = LifecycleState.CLOSED - self._release() + self._safe_release() if hasattr(self, '_handle') and self._handle: try: ManagedResource._free_native_ptr(self._handle) @@ -420,18 +597,25 @@ class C2paStream(ctypes.Structure): ] -def _clear_error_state(): - """Clear any existing error state from the C library. +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. - This function should be called at the beginning of object initialization - and before any operations that could potentially raise an error, - to ensure that stale error states from previous operations don't interfere - with new objects being created, or independent function calls. + 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. """ error = _lib.c2pa_error() - if error: - # Free the error to clear the state + if not error: + return None + try: + message = ctypes.string_at(error).decode('utf-8') + finally: _lib.c2pa_string_free(error) + return message or None class C2paSignerInfo(ctypes.Structure): @@ -454,7 +638,6 @@ def __init__(self, alg, sign_cert, private_key, ta_url): private_key: The private key as a string ta_url: The timestamp authority URL as bytes """ - _clear_error_state() if sign_cert is None: raise ValueError("sign_cert must be set") @@ -864,6 +1047,24 @@ class _C2paVerify(C2paError): pass +class _C2paUntrackedPointer(C2paError): + """Exception raised when the native layer does not recognize a pointer. + + Raised when a consume-and-return call rejects the handle it was given + before taking ownership of it, so the caller still owns that handle. + """ + pass + + +class _C2paWrongPointerType(C2paError): + """Exception raised when a pointer is tracked under a different type. + + Like _C2paUntrackedPointer, this is rejected before ownership transfer, + so the caller still owns the handle it passed in. + """ + pass + + # Attach exception subclasses to C2paError for backward compatibility # Preserves behavior for exception catching like except C2paError.ManifestNotFound, # also reduces imports (think of it as an alias of sorts) @@ -882,6 +1083,8 @@ class _C2paVerify(C2paError): C2paError.ResourceNotFound = _C2paResourceNotFound C2paError.Signature = _C2paSignature C2paError.Verify = _C2paVerify +C2paError.UntrackedPointer = _C2paUntrackedPointer +C2paError.WrongPointerType = _C2paWrongPointerType class _StringContainer: @@ -1006,6 +1209,10 @@ def _raise_typed_c2pa_error(error_str: str) -> None: raise C2paError.Signature(error_str) elif error_type == "Verify": raise C2paError.Verify(error_str) + elif error_type == "UntrackedPointer": + raise C2paError.UntrackedPointer(error_str) + elif error_type == "WrongPointerType": + raise C2paError.WrongPointerType(error_str) # If no recognized error type, raise base C2paError raise C2paError(error_str) @@ -1033,10 +1240,8 @@ def _parse_operation_result_for_error( """ if not result: # pragma: no cover if check_error: - error = _lib.c2pa_error() - if error: - error_str = ctypes.string_at(error).decode('utf-8') - _lib.c2pa_string_free(error) + error_str = _read_native_error() + if error_str: _raise_typed_c2pa_error(error_str) return None @@ -1056,7 +1261,11 @@ def _check_ffi_operation_result( Args: result: The return value from the FFI call - fallback_msg: Error message if the native library has no error details + fallback_msg: Error message if the native library has no error details. + An error message template ending in `: {}` may be passed + unformatted. An "Unknown error" fallback is filled in here, since + reaching this point means the native layer offered nothing better. + Plain messages with no placeholder are used as-is. check: Predicate that returns True when the result indicates failure. Defaults to `not r` (for pointer-returning calls). Use `lambda r: r != 0` for status-code-returning calls. @@ -1072,7 +1281,7 @@ def _check_ffi_operation_result( error = _parse_operation_result_for_error(_lib.c2pa_error()) if error: raise C2paError(error) - raise C2paError(fallback_msg) + raise C2paError(fallback_msg.format("Unknown error")) return result @@ -1168,7 +1377,6 @@ def load_settings(settings: Union[str, dict], format: str = "json") -> None: DeprecationWarning, stacklevel=2, ) - _clear_error_state() # Convert to JSON string as necessary try: @@ -1267,16 +1475,16 @@ def __init__(self): """Create new Settings with default values.""" super().__init__() - ptr = _lib.c2pa_settings_new() + settings_ptr = _lib.c2pa_settings_new() try: - _check_ffi_operation_result(ptr, "Failed to create Settings") - except Exception: - if ptr: - ManagedResource._free_native_ptr(ptr) + _check_ffi_operation_result( + settings_ptr, "Failed to create Settings") + except BaseException: + if settings_ptr: + ManagedResource._free_native_ptr(settings_ptr) raise - self._handle = ptr - self._lifecycle_state = LifecycleState.ACTIVE + self._activate(settings_ptr) @classmethod def from_json(cls, json_str: str) -> 'Settings': @@ -1433,16 +1641,15 @@ def __init__( C2paError: If creation fails """ super().__init__() - self._has_signer = False - self._signer_callback_cb = None + self._init_attrs() if settings is None and signer is None: # Simple default context - ptr = _lib.c2pa_context_new() + context_ptr = _lib.c2pa_context_new() _check_ffi_operation_result( - ptr, "Failed to create Context" + context_ptr, "Failed to create Context" ) - self._handle = ptr + self._activate(context_ptr) else: # Use ContextBuilder for settings/signer builder_ptr = _lib.c2pa_context_builder_new() @@ -1462,34 +1669,36 @@ def __init__( if signer is not None: signer._ensure_valid_state() - result = ( - _lib.c2pa_context_builder_set_signer( - builder_ptr, signer._handle, + # c2pa_context_builder_set_signer takes ownership of the + # signer pointer , on its error path as well as on success. + self._signer_callback_cb = signer._callback_cb + try: + result = ( + _lib.c2pa_context_builder_set_signer( + builder_ptr, signer._handle, + ) ) - ) + except ctypes.ArgumentError: + # Marshalling failed, so the call never reached the + # native side and the signer was never taken. + raise + signer._mark_consumed() if result != 0: _parse_operation_result_for_error(None) + self._has_signer = True # Build consumes builder_ptr - ptr = ( + context_ptr = ( _lib.c2pa_context_builder_build(builder_ptr) ) builder_ptr = None - self._handle = ptr _check_ffi_operation_result( - ptr, "Failed to build Context" + context_ptr, "Failed to build Context" ) - # Build succeeded, consume the Signer. - # Keep its callback ref alive on this Context, - # then mark it so it won't double-free the - # pointer the Context now owns. - if signer is not None: - self._signer_callback_cb = signer._callback_cb - signer._mark_consumed() - self._has_signer = True - except Exception: + self._activate(context_ptr) + except BaseException: # Free builder if build was not reached if builder_ptr is not None: try: @@ -1498,7 +1707,10 @@ def __init__( pass raise - self._lifecycle_state = LifecycleState.ACTIVE + def _init_attrs(self): + super()._init_attrs() + self._has_signer = False + self._signer_callback_cb = None def _release(self): """Release Context-specific resources.""" @@ -1572,15 +1784,6 @@ class Stream: 'stream_error': "Error cleaning up stream: {}", 'callback_error': "Error cleaning up callback {}: {}", 'cleanup_error': "Error during cleanup: {}", - 'read': "Stream is closed or not initialized during read operation", - 'memory_error': "Memory error during stream operation: {}", - 'read_error': "Error during read operation: {}", - 'seek': "Stream is closed or not initialized during seek operation", - 'seek_error': "Error during seek operation: {}", - 'write': "Stream is closed or not initialized during write operation", - 'write_error': "Error during write operation: {}", - 'flush': "Stream is closed or not initialized during flush operation", - 'flush_error': "Error during flush operation: {}" } def __init__(self, file_like_stream): @@ -1788,7 +1991,6 @@ def flush_callback(ctx): self._flush_cb = FlushCallback(flush_callback) # Create the stream - _clear_error_state() self._stream = _lib.c2pa_create_stream( None, self._read_cb, @@ -1797,8 +1999,9 @@ def flush_callback(ctx): self._flush_cb ) if not self._stream: - error = _parse_operation_result_for_error(_lib.c2pa_error()) - raise Exception("Failed to create stream: {}".format(error)) + error = _read_native_error() + raise C2paError( + "Failed to create stream: {}".format(error or "Unknown error")) self._initialized = True record_owner_pid(self) @@ -1926,15 +2129,14 @@ def _get_supported_mime_types(ffi_func, cache): if cache is not None: return list(cache), cache - _clear_error_state() count = ctypes.c_size_t() arr = ffi_func(ctypes.byref(count)) if not arr: - error = _parse_operation_result_for_error(_lib.c2pa_error()) - if error: - raise C2paError(f"Failed to get supported MIME types: {error}") - return [], cache + error = _read_native_error() + raise C2paError( + "Failed to get supported MIME types: " + f"{error or 'Unknown error'}") if count.value <= 0: try: @@ -2011,16 +2213,12 @@ class Reader(ManagedResource): # Class-level error messages to avoid multiple creation _ERROR_MESSAGES = { - 'unsupported': "Unsupported format", 'io_error': "IO error: {}", 'manifest_error': "Invalid manifest data: must be bytes", 'reader_error': "Failed to create reader: {}", 'cleanup_error': "Error during cleanup: {}", 'stream_error': "Error cleaning up stream: {}", - 'file_error': "Error cleaning up file: {}", - 'reader_cleanup_error': "Error cleaning up reader: {}", 'encoding_error': "Invalid UTF-8 characters in input: {}", - 'closed_error': "Reader is closed", 'fragment_error': "Failed to process fragment: {}" } @@ -2148,20 +2346,7 @@ def __init__( contain invalid UTF-8 characters """ super().__init__() - - self._own_stream = None - - # This is used to keep track of a file - # we may have opened ourselves, and that we need to close later - self._backing_file = None - - # Caches for manifest JSON string and parsed data. - # These are invalidated when with_fragment() is called, because each - # new BMFF fragment can refine or update the manifest content as the - # reader progressively builds its understanding of the fragmented stream. - # They are also cleared on close() to release memory. - self._manifest_json_str_cache = None - self._manifest_data_cache = None + self._init_attrs() self._context = context @@ -2202,11 +2387,11 @@ def __init__( with Stream(stream) as stream_obj: self._create_reader( format_bytes, stream_obj, manifest_data) - self._lifecycle_state = LifecycleState.ACTIVE def _create_reader(self, format_bytes, stream_obj, manifest_data=None): - """Create a Reader from a Stream. + """Create a native reader from a Stream + and activate this Reader around it. Args: format_bytes: UTF-8 encoded format/MIME type @@ -2214,7 +2399,7 @@ def _create_reader(self, format_bytes, stream_obj, manifest_data: Optional manifest bytes """ if manifest_data is None: - self._handle = _lib.c2pa_reader_from_stream( + reader_ptr = _lib.c2pa_reader_from_stream( format_bytes, stream_obj._stream) else: if not isinstance(manifest_data, bytes): @@ -2222,7 +2407,7 @@ def _create_reader(self, format_bytes, stream_obj, manifest_array = ( ctypes.c_ubyte * len(manifest_data)).from_buffer_copy(manifest_data) - self._handle = ( + reader_ptr = ( _lib.c2pa_reader_from_manifest_data_and_stream( format_bytes, stream_obj._stream, @@ -2232,8 +2417,10 @@ def _create_reader(self, format_bytes, stream_obj, ) _check_ffi_operation_result( - self._handle, - Reader._ERROR_MESSAGES['reader_error'].format("Unknown error")) + reader_ptr, + Reader._ERROR_MESSAGES['reader_error']) + + self._activate(reader_ptr) def _init_from_file(self, path, format_bytes, manifest_data=None): @@ -2248,7 +2435,6 @@ def _init_from_file(self, path, format_bytes, self._backing_file = open(path, 'rb') self._own_stream = Stream(self._backing_file) self._create_reader(format_bytes, self._own_stream, manifest_data) - self._lifecycle_state = LifecycleState.ACTIVE except C2paError: self._close_streams() raise @@ -2300,13 +2486,18 @@ def _init_from_context(self, context, format_or_path, _check_ffi_operation_result(reader_ptr, Reader._ERROR_MESSAGES[ 'reader_error' - ].format("Unknown error") + ] ) - except Exception: + except BaseException: if reader_ptr: ManagedResource._free_native_ptr(reader_ptr) raise + # Adopt the handle before the consuming call: _consume_and_swap + # needs an active resource, and from here on normal cleanup owns + # the pointer whichever way the call goes. + self._activate(reader_ptr) + if manifest_data is not None: manifest_array = ( ctypes.c_ubyte * @@ -2314,48 +2505,53 @@ 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) - new_ptr = ( - _lib.c2pa_reader_with_manifest_data_and_stream( - reader_ptr, - format_bytes, - self._own_stream._stream, - manifest_array, - len(manifest_data), - ) - ) + self._consume_and_swap( + lambda handle: ( + _lib.c2pa_reader_with_manifest_data_and_stream( + handle, + format_bytes, + self._own_stream._stream, + manifest_array, + len(manifest_data), + ) + ), + Reader._ERROR_MESSAGES['reader_error']) else: # Consume reader with stream - new_ptr = _lib.c2pa_reader_with_stream( - reader_ptr, format_bytes, - self._own_stream._stream, - ) + self._consume_and_swap( + lambda handle: _lib.c2pa_reader_with_stream( + handle, format_bytes, + self._own_stream._stream, + ), + Reader._ERROR_MESSAGES['reader_error']) + except BaseException: + self._close_streams() + raise - # reader_ptr has been consumed by the FFI call. - reader_ptr = None + def _init_attrs(self): + super()._init_attrs() + self._own_stream = None - self._handle = new_ptr + # Tracks a file we opened ourselves and must close later. + self._backing_file = None - _check_ffi_operation_result(new_ptr, - Reader._ERROR_MESSAGES[ - 'reader_error' - ].format("Unknown error") - ) + # Caches for manifest JSON string and parsed data. + # These are invalidated when with_fragment() is called. + self._manifest_json_str_cache = None + self._manifest_data_cache = None - self._lifecycle_state = LifecycleState.ACTIVE - except Exception: - self._close_streams() - raise + self._context = None def _close_streams(self): """Close owned stream and backing file if present.""" - if getattr(self, '_own_stream', None): + if self._own_stream: try: self._own_stream.close() except Exception: logger.error("Failed to close Reader stream") finally: self._own_stream = None - if getattr(self, '_backing_file', None): + if self._backing_file: try: self._backing_file.close() except Exception: @@ -2364,8 +2560,14 @@ def _close_streams(self): self._backing_file = None def _release(self): - """Release Reader-specific resources (stream, backing file).""" + """Release Reader-specific resources (caches, stream, backing file). + """ + + self._manifest_json_str_cache = None + self._manifest_data_cache = None self._close_streams() + # The Context is not ours to close, only to stop pinning. + self._context = None def _get_cached_manifest_data(self) -> Optional[dict]: """Get the cached manifest data, fetching and parsing if not cached. @@ -2420,20 +2622,14 @@ def with_fragment(self, format: str, stream, ) with Stream(stream) as main_obj, Stream(fragment_stream) as frag_obj: - new_ptr = _lib.c2pa_reader_with_fragment( - self._handle, - format_bytes, - main_obj._stream, - frag_obj._stream, - ) - - if not new_ptr: - self._mark_consumed() - _check_ffi_operation_result(new_ptr, - Reader._ERROR_MESSAGES[ - 'fragment_error' - ].format("Unknown error")) - self._handle = new_ptr + self._consume_and_swap( + lambda handle: _lib.c2pa_reader_with_fragment( + handle, + format_bytes, + main_obj._stream, + frag_obj._stream, + ), + Reader._ERROR_MESSAGES['fragment_error']) # Invalidate caches: processing a new BMFF fragment updates the native # reader's state, which can change the manifest data it returns. @@ -2444,12 +2640,6 @@ def with_fragment(self, format: str, stream, return self - def close(self): - """Release the reader resources.""" - self._manifest_json_str_cache = None - self._manifest_data_cache = None - super().close() - def json(self) -> str: """Get the manifest store as a JSON string. @@ -2692,12 +2882,8 @@ class Signer(ManagedResource): # Class-level error messages to avoid multiple creation _ERROR_MESSAGES = { - 'closed_error': "Signer is closed", 'cleanup_error': "Error during cleanup: {}", - 'signer_cleanup': "Error cleaning up signer: {}", '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: {}" @@ -2716,10 +2902,6 @@ def from_info(cls, signer_info: C2paSignerInfo) -> 'Signer': Raises: C2paError: If there was an error creating the signer """ - # Native libs plumbing: - # Clear any stale error state from previous operations - _clear_error_state() - signer_ptr = _lib.c2pa_signer_from_info(ctypes.byref(signer_info)) _check_ffi_operation_result( @@ -2837,10 +3019,6 @@ def wrapped_callback( cls._ERROR_MESSAGES['encoding_error'].format( str(e))) - # Native libs plumbing: - # Clear any stale error state from previous operations - _clear_error_state() - # Create the callback object using the callback function callback_cb = SignerCallback(wrapped_callback) @@ -2877,14 +3055,18 @@ def __init__(self, signer_ptr: ctypes.POINTER(C2paSigner)): C2paError: If the signer pointer is invalid """ super().__init__() - - self._callback_cb = None + self._init_attrs() if not signer_ptr: raise C2paError("Invalid signer pointer: pointer is null") - self._handle = signer_ptr - self._lifecycle_state = LifecycleState.ACTIVE + self._activate(signer_ptr) + + def _init_attrs(self): + super()._init_attrs() + # from_callback() replaces this with the real callback, which has to + # outlive the signer that calls it. + self._callback_cb = None def _release(self): """Release Signer-specific resources (callback reference).""" @@ -2922,18 +3104,14 @@ class Builder(ManagedResource): _ERROR_MESSAGES = { 'builder_error': "Failed to create builder: {}", 'cleanup_error': "Error during cleanup: {}", - 'builder_cleanup': "Error cleaning up builder: {}", - 'closed_error': "Builder is closed", - 'manifest_error': "Invalid manifest data: must be string or dict", 'url_error': "Error setting remote URL: {}", + 'intent_error': "Error setting intent for Builder: {}", 'resource_error': "Error adding resource: {}", 'ingredient_error': "Error adding ingredient: {}", 'archive_read_error': "Error loading ingredient from archive: {}", 'action_error': "Error adding action: {}", 'archive_error': "Error writing archive: {}", - 'sign_error': "Error during signing: {}", - 'encoding_error': "Invalid UTF-8 characters in manifest: {}", - 'json_error': "Failed to serialize manifest JSON: {}" + 'archive_load_error': "Failed to load archive into builder: {}", } @classmethod @@ -3007,25 +3185,22 @@ def from_archive( C2paError: If there was an error creating the builder from archive """ - # Handle builder._handle lifecycle somewhat manually - builder = object.__new__(cls) - ManagedResource.__init__(builder) - builder._context = None - builder._has_context_signer = False - stream_obj = Stream(stream) try: - builder._handle = ( - _lib.c2pa_builder_from_archive(stream_obj._stream) - ) + handle = _lib.c2pa_builder_from_archive(stream_obj._stream) - _check_ffi_operation_result(builder._handle, + _check_ffi_operation_result(handle, "Failed to create builder from archive" ) - builder._lifecycle_state = LifecycleState.ACTIVE - return builder + try: + # A builder from an archive here carries no context. + return cls._wrap_native_handle(handle) + except BaseException: + # No instance took ownership, so the handle is still ours. + ManagedResource._free_native_ptr(handle) + raise finally: stream_obj.close() @@ -3059,6 +3234,7 @@ def __init__( C2paError.Json: If the manifest JSON cannot be serialized """ super().__init__() + self._init_attrs() self._context = context self._has_context_signer = ( @@ -3072,13 +3248,13 @@ def __init__( if context is not None: self._init_from_context(context, json_str) else: - self._handle = _lib.c2pa_builder_from_json(json_str) + builder_ptr = _lib.c2pa_builder_from_json(json_str) _check_ffi_operation_result( - self._handle, - Builder._ERROR_MESSAGES['builder_error'].format("Unknown error")) + builder_ptr, + Builder._ERROR_MESSAGES['builder_error']) - self._lifecycle_state = LifecycleState.ACTIVE + self._activate(builder_ptr) def _init_from_context(self, context, json_str): """Initialize Builder from a ContextProvider. @@ -3096,23 +3272,32 @@ def _init_from_context(self, context, json_str): _check_ffi_operation_result(builder_ptr, Builder._ERROR_MESSAGES[ 'builder_error' - ].format("Unknown error") + ] ) - except Exception: + except BaseException: if builder_ptr: ManagedResource._free_native_ptr(builder_ptr) raise - # Consume-and-return: builder_ptr is consumed, - # new_ptr is the valid pointer going forward - new_ptr = _lib.c2pa_builder_with_definition(builder_ptr, json_str) - self._handle = new_ptr + # Adopt the handle before the consuming call: + # _consume_and_swap needs an active resource, + # and from here on normal cleanup owns the pointer. + self._activate(builder_ptr) + + self._consume_and_swap( + lambda handle: _lib.c2pa_builder_with_definition( + handle, json_str), + Builder._ERROR_MESSAGES['builder_error']) - _check_ffi_operation_result(new_ptr, - Builder._ERROR_MESSAGES[ - 'builder_error' - ].format("Unknown error") - ) + def _init_attrs(self): + super()._init_attrs() + self._context = None + self._has_context_signer = False + + def _release(self): + """Release the Builder's reference to its Context.""" + # The Context is not ours to close, only to stop pinning. + self._context = None def set_no_embed(self): """Set the no-embed flag. @@ -3143,7 +3328,7 @@ def set_remote_url(self, remote_url: str): _check_ffi_operation_result( result, - Builder._ERROR_MESSAGES['url_error'].format("Unknown error"), + Builder._ERROR_MESSAGES['url_error'], check=lambda r: r != 0) def set_intent( @@ -3182,7 +3367,7 @@ def set_intent( _check_ffi_operation_result( result, - "Error setting intent for Builder: Unknown error", + Builder._ERROR_MESSAGES['intent_error'], check=lambda r: r != 0) def add_resource(self, uri: str, stream: Any): @@ -3205,7 +3390,7 @@ def add_resource(self, uri: str, stream: Any): _check_ffi_operation_result( result, - Builder._ERROR_MESSAGES['resource_error'].format("Unknown error"), + Builder._ERROR_MESSAGES['resource_error'], check=lambda r: r != 0) def add_ingredient( @@ -3272,7 +3457,7 @@ def add_ingredient_from_stream( _check_ffi_operation_result( result, - Builder._ERROR_MESSAGES['ingredient_error'].format("Unknown error"), + Builder._ERROR_MESSAGES['ingredient_error'], check=lambda r: r != 0) def add_action(self, action_json: Union[str, dict]) -> None: @@ -3294,7 +3479,7 @@ def add_action(self, action_json: Union[str, dict]) -> None: _check_ffi_operation_result( result, - Builder._ERROR_MESSAGES['action_error'].format("Unknown error"), + Builder._ERROR_MESSAGES['action_error'], check=lambda r: r != 0) def to_archive(self, stream: Any) -> None: @@ -3315,7 +3500,7 @@ def to_archive(self, stream: Any) -> None: _check_ffi_operation_result( result, - Builder._ERROR_MESSAGES["archive_error"].format("Unknown error"), + Builder._ERROR_MESSAGES["archive_error"], check=lambda r: r != 0) def write_ingredient_archive(self, ingredient_id: str, stream: Any) -> None: @@ -3338,10 +3523,9 @@ def write_ingredient_archive(self, ingredient_id: str, stream: Any) -> None: result = _lib.c2pa_builder_write_ingredient_archive( self._handle, ingredient_id_str, stream_obj._stream) - _check_ffi_operation_result(result, - Builder._ERROR_MESSAGES["archive_error"].format( - "Unknown error" - ), + _check_ffi_operation_result( + result, + Builder._ERROR_MESSAGES["archive_error"], check=lambda r: r != 0) def add_ingredient_from_archive(self, stream: Any) -> None: @@ -3361,10 +3545,9 @@ def add_ingredient_from_archive(self, stream: Any) -> None: result = _lib.c2pa_builder_add_ingredient_from_archive( self._handle, stream_obj._stream) - _check_ffi_operation_result(result, - Builder._ERROR_MESSAGES["archive_read_error"].format( - "Unknown error" - ), + _check_ffi_operation_result( + result, + Builder._ERROR_MESSAGES["archive_read_error"], check=lambda r: r != 0) def with_archive(self, stream: Any) -> 'Builder': @@ -3385,18 +3568,10 @@ def with_archive(self, stream: Any) -> 'Builder': self._ensure_valid_state() with Stream(stream) as stream_obj: - try: - new_ptr = _lib.c2pa_builder_with_archive( - self._handle, stream_obj._stream) - except Exception as e: - self._mark_consumed() - raise C2paError( - f"Error loading archive: {e}" - ) - # Old handle consumed by FFI - self._handle = new_ptr - _check_ffi_operation_result( - new_ptr, "Failed to load archive into builder") + self._consume_and_swap( + lambda handle: _lib.c2pa_builder_with_archive( + handle, stream_obj._stream), + Builder._ERROR_MESSAGES['archive_load_error']) return self @@ -3459,9 +3634,9 @@ def _sign_internal( # Closing here ensures resources clean up, # and single use/single sign done by a Builder. self.close() - except Exception as e: + except BaseException as e: self.close() - raise C2paError(f"Error during signing: {e}") + raise C2paError(f"Error during signing: {e}") from e _check_ffi_operation_result( result, @@ -3677,8 +3852,6 @@ def format_embeddable(format: str, manifest_bytes: bytes) -> tuple[int, bytes]: Raises: C2paError: If there was an error converting the manifest """ - _clear_error_state() - format_str = format.encode('utf-8') manifest_array = (ctypes.c_ubyte * len(manifest_bytes)).from_buffer_copy( manifest_bytes @@ -3794,8 +3967,6 @@ def ed25519_sign(data: bytes, private_key: str) -> bytes: C2paError: If there was an error signing the data C2paError.Encoding: If the private key contains invalid UTF-8 chars """ - _clear_error_state() - if not data: raise C2paError("Data to sign cannot be empty") diff --git a/tests/perf/baseline.json b/tests/perf/baseline.json index 7af9ffb5..c151efe5 100644 --- a/tests/perf/baseline.json +++ b/tests/perf/baseline.json @@ -2,224 +2,299 @@ "_meta": { "memray_version": "1.19.3", "python_version": "3.12.13", - "c2pa_native_version": "c2pa-v0.89.0", - "iterations": 100, + "c2pa_native_version": "c2pa-v0.90.0", + "iterations": 200, "perf_env": "python-3.12-slim", "arch": "aarch64" }, "reader_jpeg_legacy": { - "peak_bytes": 3791301, - "leaked_bytes": 3292672, - "total_allocations": 722823 + "peak_bytes": 3851610, + "leaked_bytes": 3351823, + "total_allocations": 1362322 }, "reader_jpeg_with_context": { - "peak_bytes": 3785583, - "leaked_bytes": 3284873, - "total_allocations": 717272 + "peak_bytes": 3845367, + "leaked_bytes": 3345097, + "total_allocations": 1349879 }, "reader_manifest_data_context": { - "peak_bytes": 7576945, - "leaked_bytes": 3407648, - "total_allocations": 619356 + "peak_bytes": 7636730, + "leaked_bytes": 3468040, + "total_allocations": 1147359 }, "reader_mp4": { - "peak_bytes": 4159582, - "leaked_bytes": 3283805, - "total_allocations": 2089508 + "peak_bytes": 4222601, + "leaked_bytes": 3345724, + "total_allocations": 4095915 }, "reader_wav": { - "peak_bytes": 4464615, - "leaked_bytes": 3293702, - "total_allocations": 413484 + "peak_bytes": 4523095, + "leaked_bytes": 3355666, + "total_allocations": 742391 }, "builder_sign_jpeg_legacy": { - "peak_bytes": 7724599, - "leaked_bytes": 3408513, - "total_allocations": 560691 + "peak_bytes": 7785129, + "leaked_bytes": 3468507, + "total_allocations": 1041412 }, "builder_sign_jpeg_with_context": { - "peak_bytes": 7716613, - "leaked_bytes": 3400514, - "total_allocations": 554788 + "peak_bytes": 7779538, + "leaked_bytes": 3463042, + "total_allocations": 1027485 }, "builder_sign_png_legacy": { - "peak_bytes": 7961139, - "leaked_bytes": 3406984, - "total_allocations": 1981843 + "peak_bytes": 8023081, + "leaked_bytes": 3468300, + "total_allocations": 3883115 }, "builder_sign_png_with_context": { - "peak_bytes": 7955799, - "leaked_bytes": 3401929, - "total_allocations": 1975867 + "peak_bytes": 8017008, + "leaked_bytes": 3462829, + "total_allocations": 3869515 }, "builder_sign_jpeg_parallel_split_pool": { - "peak_bytes": 44002804, - "leaked_bytes": 3801899, - "total_allocations": 566497 + "peak_bytes": 45854797, + "leaked_bytes": 3840928, + "total_allocations": 1035646 }, "builder_sign_jpeg_parallel_split_barrier": { - "peak_bytes": 45784452, - "leaked_bytes": 3800550, - "total_allocations": 565361 + "peak_bytes": 45844809, + "leaked_bytes": 3861014, + "total_allocations": 1037741 }, "builder_sign_png_parallel_split_pool": { - "peak_bytes": 46054163, - "leaked_bytes": 3801867, - "total_allocations": 1987868 + "peak_bytes": 46586728, + "leaked_bytes": 3868054, + "total_allocations": 3877696 }, "builder_sign_png_parallel_split_barrier": { - "peak_bytes": 44206900, - "leaked_bytes": 3800490, - "total_allocations": 1986526 + "peak_bytes": 46082548, + "leaked_bytes": 3879161, + "total_allocations": 3879780 }, "builder_sign_gif": { - "peak_bytes": 14574556, - "leaked_bytes": 3400735, - "total_allocations": 8550061 + "peak_bytes": 14635465, + "leaked_bytes": 3461270, + "total_allocations": 17017654 }, "builder_sign_heic": { - "peak_bytes": 4644811, - "leaked_bytes": 3407275, - "total_allocations": 831824 + "peak_bytes": 4698434, + "leaked_bytes": 3469086, + "total_allocations": 1563419 }, "builder_sign_m4a": { - "peak_bytes": 18885052, - "leaked_bytes": 3407378, - "total_allocations": 2647270 + "peak_bytes": 18833496, + "leaked_bytes": 3469085, + "total_allocations": 5194205 }, "builder_sign_webp": { - "peak_bytes": 8928848, - "leaked_bytes": 3399564, - "total_allocations": 499272 + "peak_bytes": 8991237, + "leaked_bytes": 3461271, + "total_allocations": 916145 }, "builder_sign_avi": { - "peak_bytes": 7068483, - "leaked_bytes": 3399340, - "total_allocations": 45032279 + "peak_bytes": 7130933, + "leaked_bytes": 3461270, + "total_allocations": 89982012 }, "builder_sign_mp4": { - "peak_bytes": 6198764, - "leaked_bytes": 3407319, - "total_allocations": 1944326 + "peak_bytes": 6245379, + "leaked_bytes": 3469085, + "total_allocations": 3788717 }, "builder_sign_tiff": { - "peak_bytes": 13151779, - "leaked_bytes": 3400355, - "total_allocations": 5472611 + "peak_bytes": 13213169, + "leaked_bytes": 3461271, + "total_allocations": 10862700 }, "builder_sign_jpeg_parent_of": { - "peak_bytes": 14204315, - "leaked_bytes": 3401481, - "total_allocations": 1292637 + "peak_bytes": 14265295, + "leaked_bytes": 3461665, + "total_allocations": 2506107 }, "builder_sign_jpeg_component_of": { - "peak_bytes": 14204923, - "leaked_bytes": 3400730, - "total_allocations": 1315125 + "peak_bytes": 14266996, + "leaked_bytes": 3462012, + "total_allocations": 2551180 }, "builder_sign_jpeg_parent_and_component": { - "peak_bytes": 14588185, - "leaked_bytes": 3549911, - "total_allocations": 2299453 + "peak_bytes": 14665241, + "leaked_bytes": 3614613, + "total_allocations": 4523960 }, "builder_sign_jpeg_parent_and_component_mixed_mime": { - "peak_bytes": 14506586, - "leaked_bytes": 3401364, - "total_allocations": 2796028 + "peak_bytes": 14568780, + "leaked_bytes": 3462718, + "total_allocations": 5517180 }, "builder_sign_jpeg_two_components_same_mime": { - "peak_bytes": 14601941, - "leaked_bytes": 3558116, - "total_allocations": 2289245 + "peak_bytes": 14559274, + "leaked_bytes": 3564233, + "total_allocations": 4497379 }, "builder_sign_jpeg_two_components_mixed_mime": { - "peak_bytes": 14503695, - "leaked_bytes": 3401276, - "total_allocations": 2785702 + "peak_bytes": 14564839, + "leaked_bytes": 3461873, + "total_allocations": 5490592 }, "builder_sign_jpeg_archive_roundtrip": { - "peak_bytes": 14236247, - "leaked_bytes": 3420354, - "total_allocations": 1777705 + "peak_bytes": 14297571, + "leaked_bytes": 3481212, + "total_allocations": 3467149 + }, + "builder_from_archive_roundtrip": { + "peak_bytes": 14297349, + "leaked_bytes": 3480475, + "total_allocations": 3101030 + }, + "builder_with_archive_swap": { + "peak_bytes": 3681081, + "leaked_bytes": 3350198, + "total_allocations": 704373 + }, + "reader_with_fragment_swap": { + "peak_bytes": 3778159, + "leaked_bytes": 3353205, + "total_allocations": 3787587 + }, + "with_fragment_pre_consume_rejection": { + "peak_bytes": 3778057, + "leaked_bytes": 3354795, + "total_allocations": 2094004 + }, + "with_archive_post_consume_failure": { + "peak_bytes": 3350600, + "leaked_bytes": 3308056, + "total_allocations": 175290 + }, + "with_fragment_marshalling_error": { + "peak_bytes": 3708068, + "leaked_bytes": 3352335, + "total_allocations": 2077090 + }, + "with_fragment_mixed_outcomes": { + "peak_bytes": 3779175, + "leaked_bytes": 3356294, + "total_allocations": 2656787 }, "builder_to_archive_with_ingredient": { - "peak_bytes": 14010137, - "leaked_bytes": 3278101, - "total_allocations": 957452 + "peak_bytes": 14069232, + "leaked_bytes": 3337316, + "total_allocations": 1830896 }, "builder_sign_jpeg_archive_roundtrip_ingredient_in_archive": { - "peak_bytes": 14225579, - "leaked_bytes": 3420822, - "total_allocations": 2981495 + "peak_bytes": 14287046, + "leaked_bytes": 3481977, + "total_allocations": 5879957 }, "builder_write_ingredient_archive": { - "peak_bytes": 14010131, - "leaked_bytes": 3278099, - "total_allocations": 944654 + "peak_bytes": 14069289, + "leaked_bytes": 3337377, + "total_allocations": 1805304 }, "builder_sign_jpeg_add_ingredient_from_archive": { - "peak_bytes": 14075511, - "leaked_bytes": 3421125, - "total_allocations": 1753642 + "peak_bytes": 14133742, + "leaked_bytes": 3480831, + "total_allocations": 3415920 }, "builder_ingredient_archive_roundtrip": { - "peak_bytes": 14222976, - "leaked_bytes": 3419885, - "total_allocations": 2609017 + "peak_bytes": 14284443, + "leaked_bytes": 3480809, + "total_allocations": 5132060 }, "builder_sign_jpeg_two_ingredient_archives": { - "peak_bytes": 14075190, - "leaked_bytes": 3421103, - "total_allocations": 2161232 + "peak_bytes": 14134560, + "leaked_bytes": 3481604, + "total_allocations": 4215728 }, "reader_error_no_manifest": { - "peak_bytes": 3504260, - "leaked_bytes": 3263283, - "total_allocations": 178873 + "peak_bytes": 3564471, + "leaked_bytes": 3323629, + "total_allocations": 276175 }, "builder_error_invalid_manifest": { - "peak_bytes": 3292723, - "leaked_bytes": 3235293, - "total_allocations": 96622 + "peak_bytes": 3352053, + "leaked_bytes": 3297079, + "total_allocations": 113926 }, "reader_string_apis": { - "peak_bytes": 3915891, - "leaked_bytes": 3284213, - "total_allocations": 1185492 + "peak_bytes": 3978113, + "leaked_bytes": 3346111, + "total_allocations": 2287335 + }, + "signer_construction": { + "peak_bytes": 3350893, + "leaked_bytes": 3288137, + "total_allocations": 153245 + }, + "builder_from_context_construction": { + "peak_bytes": 3350600, + "leaked_bytes": 3288582, + "total_allocations": 112688 }, "fork_reader_collect": { - "peak_bytes": 3789318, - "leaked_bytes": 3291267, - "total_allocations": 705123 + "peak_bytes": 3850530, + "leaked_bytes": 3353063, + "total_allocations": 1328122 }, "fork_contended_mutex": { - "peak_bytes": 7617864, - "leaked_bytes": 3421711, - "total_allocations": 33591148 + "peak_bytes": 7679019, + "leaked_bytes": 3482128, + "total_allocations": 67472694 }, "fork_thread_local_orphan": { - "peak_bytes": 3876269, - "leaked_bytes": 3379616, - "total_allocations": 731511 + "peak_bytes": 3936170, + "leaked_bytes": 3439733, + "total_allocations": 1381055 }, "fork_gc_cycle": { - "peak_bytes": 3790340, - "leaked_bytes": 3291400, - "total_allocations": 706802 + "peak_bytes": 3850434, + "leaked_bytes": 3353160, + "total_allocations": 1332098 }, "fork_parent_frees_after_fork": { - "peak_bytes": 5989167, - "leaked_bytes": 3289951, - "total_allocations": 12461253 + "peak_bytes": 5447584, + "leaked_bytes": 3350400, + "total_allocations": 24829257 + }, + "fork_child_closes_then_parent_frees": { + "peak_bytes": 5446620, + "leaked_bytes": 3350407, + "total_allocations": 24829254 }, "fork_child_sys_exit": { - "peak_bytes": 3789334, - "leaked_bytes": 3291456, - "total_allocations": 708625 + "peak_bytes": 3850546, + "leaked_bytes": 3353234, + "total_allocations": 1335925 }, "fork_stream_cleanup": { - "peak_bytes": 3402893, - "leaked_bytes": 3230663, - "total_allocations": 93141 + "peak_bytes": 3464063, + "leaked_bytes": 3291969, + "total_allocations": 105340 + }, + "fork_swap_cleanup": { + "peak_bytes": 3681171, + "leaked_bytes": 3350696, + "total_allocations": 714376 + }, + "fork_contended_mutex_swap": { + "peak_bytes": 7302379, + "leaked_bytes": 3475147, + "total_allocations": 35948516 + }, + "fork_contended_mutex_wrap": { + "peak_bytes": 7288748, + "leaked_bytes": 3463411, + "total_allocations": 34847186 + }, + "fork_consumed_signer": { + "peak_bytes": 3350894, + "leaked_bytes": 3288906, + "total_allocations": 175055 + }, + "swap_chain_churn": { + "peak_bytes": 3681161, + "leaked_bytes": 3350287, + "total_allocations": 672537 } } \ No newline at end of file diff --git a/tests/perf/scenarios.py b/tests/perf/scenarios.py index 0e367fb0..23300aed 100644 --- a/tests/perf/scenarios.py +++ b/tests/perf/scenarios.py @@ -9,6 +9,7 @@ Each function is called N times by run_profile.py. """ +import ctypes import gc import io import json @@ -27,6 +28,7 @@ Signer, Stream, ) +import c2pa.c2pa as c2pa_module FIXTURES_DIR = Path(__file__).parent.parent / "fixtures" READING_FIXTURES_DIR = FIXTURES_DIR / "files-for-reading-tests" @@ -36,6 +38,8 @@ CLOUD_JPEG = FIXTURES_DIR / "cloud.jpg" SOURCE_JPEG = FIXTURES_DIR / "A.jpg" SIGNING_PNG = SIGNING_FIXTURES_DIR / "sample1.png" +DASH_INIT_MP4 = FIXTURES_DIR / "dashinit.mp4" +DASH_FRAGMENT = FIXTURES_DIR / "dash1.m4s" _DST_COMPOSITE = "http://cv.iptc.org/newscodes/digitalsourcetype/compositeWithTrainedAlgorithmicMedia" @@ -477,6 +481,213 @@ def scenario_builder_sign_jpeg_archive_roundtrip(iterations: int = 100) -> None: builder.sign("image/jpeg", io.BytesIO(source_bytes), io.BytesIO()) +def scenario_builder_with_archive_swap(iterations: int = 100) -> None: + """Loop Builder.with_archive(), the consume-and-return FFI path. + + c2pa_builder_with_archive consumes the old native handle and returns a + replacement, so the Python side swaps the pointer without freeing the + consumed one. Freeing it would be a double-free, and failing to adopt the + replacement would leak. The other builder scenarios never swap a live + handle, so neither mistake would show up there. + """ + context = Context(signer=_make_signer()) + archive = io.BytesIO() + Builder(MANIFEST_BASE).to_archive(archive) + archive_bytes = archive.getvalue() + for _ in _iterate(iterations): + builder = Builder(MANIFEST_BASE, context=context) + builder.with_archive(io.BytesIO(archive_bytes)) + builder.close() + + +def scenario_reader_with_fragment_swap(iterations: int = 100) -> None: + """Loop Reader.with_fragment(), the other consume-and-return FFI path. + + Same ownership hand-off as with_archive: c2pa_reader_with_fragment eats + the old reader handle and returns a new one. + """ + init_bytes = DASH_INIT_MP4.read_bytes() + fragment_bytes = DASH_FRAGMENT.read_bytes() + for _ in _iterate(iterations): + reader = Reader("video/mp4", io.BytesIO(init_bytes)) + try: + reader.with_fragment( + "video/mp4", + io.BytesIO(init_bytes), + io.BytesIO(fragment_bytes), + ) + except C2paError: + # A failed call consumed the old handle just as a successful one + # would, so the scenario measures both outcomes. + pass + finally: + reader.close() + + +def scenario_builder_from_archive_roundtrip(iterations: int = 100) -> None: + """Loop Builder.from_archive() itself (context-less alternate constructor), + then sign. Regression guard for the classmethod's native-handle wrapping. + """ + signer = _make_signer() + source_bytes = SOURCE_JPEG.read_bytes() + ingredient_bytes = SIGNED_JPEG.read_bytes() + archive_bytes = io.BytesIO() + Builder(MANIFEST_BASE).to_archive(archive_bytes) + archive_bytes = archive_bytes.getvalue() + for _ in _iterate(iterations): + # from_archive() yields a context-less Builder, so sign() needs an + # explicit signer (no Context to pull one from). + builder = Builder.from_archive(io.BytesIO(archive_bytes)) + with io.BytesIO(ingredient_bytes) as ing: + builder.add_ingredient( + {"relationship": "parentOf", "instance_id": _PARENT_ID}, + "image/jpeg", ing, + ) + builder.sign(signer, "image/jpeg", io.BytesIO(source_bytes), io.BytesIO()) + + +# Consume-and-return failure paths. +# +# These calls take ownership partway through their body, so a null return is +# ambiguous and getting it wrong leaks one handle per call. The success paths +# are covered above; these loop the failure paths, where the leak would be. + +def _untracked_reader_handle(): + """A pointer the native registry does not know about. + + A never-allocated buffer, so it is rejected like a stale handle without + allocating a real Reader per call, which would swamp the measurement. + Freed handles are unusable here: recycled addresses become tracked again. + """ + buf = ctypes.create_string_buffer(64) + return ctypes.cast(buf, ctypes.POINTER(c2pa_module.C2paReader)), buf + + +def scenario_reader_with_fragment_pre_consume_rejection( + iterations: int = 100) -> None: + """Loop the rejection that precedes the ownership transfer. + + The handle is still ours, so treating this as consumed drops a pointer + the registry still holds and leaked_bytes climbs with iterations. + """ + init_bytes = DASH_INIT_MP4.read_bytes() + fragment_bytes = DASH_FRAGMENT.read_bytes() + for _ in _iterate(iterations): + reader = Reader("video/mp4", io.BytesIO(init_bytes)) + real_handle = reader._handle + # Keep the buffer alive: the cast pointer does not own it, and an + # early collection would hand the FFI a dangling address. + bogus, _buf = _untracked_reader_handle() + reader._handle = bogus + try: + reader.with_fragment("video/mp4", io.BytesIO(init_bytes), + io.BytesIO(fragment_bytes)) + raise AssertionError("pre-consume rejection did not raise") + except C2paError as e: + # Fail loudly: without these the scenario still runs when the + # ownership logic regresses, and a rejection that stops being + # recognised looks identical to a pass. + if not any(tag in str(e) for tag in + c2pa_module.ManagedResource._PRE_CONSUME_ERROR_TAGS): + raise AssertionError( + f"expected a pre-consume rejection, got: {e}") from e + if reader._handle is None: + raise AssertionError( + "handle was dropped on a pre-consume rejection; the " + "native side never took ownership, so this leaks") from e + finally: + # Restore before close() so the real handle is freed exactly once. + reader._handle = real_handle + reader.close() + + +def scenario_builder_with_archive_post_consume_failure( + iterations: int = 100) -> None: + """Loop a failure after the ownership transfer. + + The control: if the fix over-corrected into retaining handles the native + side already dropped, this scenario double-frees or leaks. + """ + for _ in _iterate(iterations): + builder = Builder(MANIFEST_BASE) + try: + builder.with_archive(io.BytesIO(b"not a valid archive")) + raise AssertionError("post-consume failure did not raise") + except C2paError: + pass + finally: + builder.close() + + +def scenario_with_fragment_marshalling_error(iterations: int = 100) -> None: + """Loop a failure that never reaches native code. + + Nothing was consumed, so the reader must stay usable. The old blanket + except marked it consumed here, leaking on what is only a type error. + """ + init_bytes = DASH_INIT_MP4.read_bytes() + for _ in _iterate(iterations): + reader = Reader("video/mp4", io.BytesIO(init_bytes)) + try: + reader.with_fragment("video/mp4", object(), object()) + raise AssertionError("marshalling error did not raise") + except (C2paError, TypeError, ctypes.ArgumentError): + pass + finally: + # Must still be usable: nothing was handed over. + reader.json() + reader.close() + + +def scenario_with_fragment_mixed_outcomes(iterations: int = 100) -> None: + """Interleave success, pre-consume rejection and post-consume failure. + + Each path leaves a different state behind, so running them in sequence + catches a stale error being read as the current call's. + """ + init_bytes = DASH_INIT_MP4.read_bytes() + fragment_bytes = DASH_FRAGMENT.read_bytes() + for i in _iterate(iterations): + phase = i % 3 + reader = Reader("video/mp4", io.BytesIO(init_bytes)) + try: + if phase == 0: + reader.with_fragment("video/mp4", io.BytesIO(init_bytes), + io.BytesIO(fragment_bytes)) + elif phase == 1: + real_handle = reader._handle + bogus, _buf = _untracked_reader_handle() + reader._handle = bogus + try: + reader.with_fragment("video/mp4", io.BytesIO(init_bytes), + io.BytesIO(fragment_bytes)) + raise AssertionError( + "pre-consume rejection did not raise") + except C2paError as e: + # A stale error from the phase before must not be read as + # this call's; the rejection would stop being recognised. + if not any( + tag in str(e) for tag in c2pa_module + .ManagedResource._PRE_CONSUME_ERROR_TAGS): + raise AssertionError( + f"expected a pre-consume rejection, got: {e}" + ) from e + if reader._handle is None: + raise AssertionError( + "handle dropped on a pre-consume rejection" + ) from e + finally: + reader._handle = real_handle + else: + try: + reader.with_fragment("video/mp4", io.BytesIO(b"garbage"), + io.BytesIO(b"garbage")) + except C2paError: + pass + finally: + reader.close() + + # Archive scenarios: builder as working store (to_archive/with_archive) and # per-ingredient archives (write_ingredient_archive/add_ingredient_from_archive). @@ -678,6 +889,37 @@ def scenario_reader_string_apis(iterations: int = 100) -> None: reader.close() +def scenario_builder_from_context_construction(iterations: int = 100) -> None: + """Loop Builder(context=...) construction, the consume-and-swap path. + + c2pa_builder_from_context hands back a handle that + c2pa_builder_with_definition then consumes and replaces. The Builder + adopts the first handle before that call so _consume_and_swap can own the + swap, which means a mis-sequenced swap leaks the replacement or frees the + consumed pointer twice. The other context builder scenarios sign a full + asset per iteration, so a one-handle regression here would sit under their + noise. This one only constructs and closes. + + scenario_reader_manifest_data_context is the Reader-side equivalent. + """ + context = Context() + for _ in _iterate(iterations): + builder = Builder(MANIFEST_BASE, context=context) + builder.close() + + +def scenario_signer_construction(iterations: int = 100) -> None: + """Loop Signer.from_info()/__init__ construction and teardown. + + Every other scenario calls _make_signer() once outside its loop, so + repeated Signer construction/destruction has no coverage elsewhere. + Regression guard for Signer.__init__'s native-handle activation. + """ + for _ in _iterate(iterations): + signer = _make_signer() + signer.close() + + # jpeg + png context variants, paired with the `_legacy` scenarios above for # side-by-side comparison. @@ -878,6 +1120,37 @@ def scenario_fork_parent_frees_after_fork(iterations: int = 100) -> None: r.close() +def scenario_fork_child_closes_then_parent_frees(iterations: int = 100) -> None: + """Fork safety benchmark scenario: + 20 Readers created, fork, the CHILD closes all 20 inherited Readers (and + runs GC so any __del__ fires) before exiting, then the parent closes its + own 20 copies. Exercises the child-side path where _cleanup_resources marks + the child's copy CLOSED and nulls the handle while skipping the native free + — the branch scenario_fork_parent_frees_after_fork never hits (its child + does nothing). Two invariants: the child must exit cleanly (no deadlock via + the 5 s alarm, no crash from a child-side double-free), and the parent must + still free all 20 (leaked_bytes stays at baseline — the child's state + mutation does not suppress the parent's frees, since the copies are + independent post-fork). + """ + if not hasattr(os, "fork"): + return + for _ in _iterate(iterations): + readers = [] + for _ in range(20): + with open(SIGNED_JPEG, "rb") as f: + readers.append(Reader("image/jpeg", f)) + + def _child(): + for r in readers: + r.close() # foreign teardown: mark closed, skip native free + gc.collect() + + _fork_wait(_child) + for r in readers: + r.close() # parent's own copies: real free + + def scenario_fork_child_sys_exit(iterations: int = 100) -> None: """Fork safety benchmark scenario: Child calls sys.exit(0), full Python shutdown: atexit, finalizers, GC. @@ -900,6 +1173,157 @@ def _child(): context.close() +def _fork_contended_over(make_object, iterations): + """Fork over an object built by make_object() while 8 threads churn + Readers, so the registry Mutex is likely held at the instant of fork(). + + The child closes the inherited object. Without the PID guard that close + calls into the native library and can block forever on a mutex left + locked by a thread that fork() did not clone, which _fork_wait's alarm + reports as a timeout. The parent closes afterwards for the real free. + """ + if not hasattr(os, "fork"): + return + stop = threading.Event() + + def _worker(): + while not stop.is_set(): + with open(SIGNED_JPEG, "rb") as f: + r = Reader("image/jpeg", f) + r.close() + + threads = [threading.Thread(target=_worker, daemon=True) + for _ in range(8)] + for t in threads: + t.start() + try: + for _ in _iterate(iterations): + for _ in range(5): + obj = make_object() + + def _child(o=obj): + o.close() + gc.collect() + + _fork_wait(_child) + obj.close() + finally: + stop.set() + for t in threads: + t.join(timeout=5) + + +def scenario_fork_contended_mutex_swap(iterations: int = 100) -> None: + """Fork safety benchmark scenario: + fork over a Builder whose handle came from with_archive(), under the same + thread contention as fork_contended_mutex. That scenario only ever forks + over handles that came straight from a constructor, so a swapped-in + handle losing its stamp would go unnoticed there. + """ + if not hasattr(os, "fork"): + return + context = Context(signer=_make_signer()) + archive = io.BytesIO() + Builder(MANIFEST_BASE).to_archive(archive) + archive_bytes = archive.getvalue() + + def _make(): + builder = Builder(MANIFEST_BASE, context=context) + builder.with_archive(io.BytesIO(archive_bytes)) + return builder + + _fork_contended_over(_make, iterations) + + +def scenario_fork_contended_mutex_wrap(iterations: int = 100) -> None: + """Fork safety benchmark scenario: + fork over a Builder built by from_archive(), under thread contention. + from_archive is the only path that bypasses __init__, so it is the one + most likely to be missing the PID stamp the child's close() depends on. + """ + if not hasattr(os, "fork"): + return + archive = io.BytesIO() + Builder(MANIFEST_BASE).to_archive(archive) + archive_bytes = archive.getvalue() + + _fork_contended_over( + lambda: Builder.from_archive(io.BytesIO(archive_bytes)), iterations) + + +def scenario_fork_consumed_signer(iterations: int = 100) -> None: + """Fork safety benchmark scenario: + the parent builds a Context that consumed a Signer, then forks. The child + closes both. The consumed Signer holds no handle, so it must be inert in + either process, and the Context must be skipped by the PID guard. + """ + if not hasattr(os, "fork"): + return + for _ in _iterate(iterations): + signer = _make_signer() + context = Context(signer=signer) + + def _child(c=context, s=signer): + s.close() + c.close() + gc.collect() + + _fork_wait(_child) + signer.close() + context.close() + + +def scenario_swap_chain_churn(iterations: int = 100) -> None: + """Loop with_archive() repeatedly on one Builder, so a chain of handles + is consumed and replaced on a single live object. Every other scenario + swaps a given object at most once. + + This one is a crash and allocation-churn guard rather than a leak gate. + Only one Builder is closed however many times the loop runs, so a + close-path leak here is O(1) and invisible against the interpreter's + allocation floor. What a broken swap does instead is fail loudly: keeping + the consumed pointer makes the next call raise UntrackedPointer from the + native registry, and freeing it makes the free itself fail. total_allocations + still tracks the churn. + """ + context = Context(signer=_make_signer()) + archive = io.BytesIO() + Builder(MANIFEST_BASE).to_archive(archive) + archive_bytes = archive.getvalue() + builder = Builder(MANIFEST_BASE, context=context) + for _ in _iterate(iterations): + builder.with_archive(io.BytesIO(archive_bytes)) + builder.close() + context.close() + + +def scenario_fork_swap_cleanup(iterations: int = 100) -> None: + """Fork safety benchmark scenario: + the handle a Builder owns at fork time came from with_archive(), which + consumed the original and returned a replacement. The child must skip the + free on the swapped-in handle just as it would on an original one, and the + parent must still free it exactly once afterwards. The other fork + scenarios only ever fork over handles that came straight from a + constructor. + """ + if not hasattr(os, "fork"): + return + context = Context(signer=_make_signer()) + archive = io.BytesIO() + Builder(MANIFEST_BASE).to_archive(archive) + archive_bytes = archive.getvalue() + for _ in _iterate(iterations): + builder = Builder(MANIFEST_BASE, context=context) + builder.with_archive(io.BytesIO(archive_bytes)) + + def _child(b=builder): + b.close() + gc.collect() + + _fork_wait(_child) + builder.close() + + def scenario_fork_stream_cleanup(iterations: int = 100) -> None: """Fork safety benchmark scenario: Stream wraps a BytesIO with ctypes callbacks stored as instance attributes. @@ -943,6 +1367,16 @@ def scenario_fork_stream_cleanup(iterations: int = 100) -> None: "builder_sign_jpeg_two_components_same_mime": scenario_builder_sign_jpeg_two_components_same_mime, "builder_sign_jpeg_two_components_mixed_mime": scenario_builder_sign_jpeg_two_components_mixed_mime, "builder_sign_jpeg_archive_roundtrip": scenario_builder_sign_jpeg_archive_roundtrip, + "builder_from_archive_roundtrip": scenario_builder_from_archive_roundtrip, + "builder_with_archive_swap": scenario_builder_with_archive_swap, + "reader_with_fragment_swap": scenario_reader_with_fragment_swap, + "with_fragment_pre_consume_rejection": + scenario_reader_with_fragment_pre_consume_rejection, + "with_archive_post_consume_failure": + scenario_builder_with_archive_post_consume_failure, + "with_fragment_marshalling_error": + scenario_with_fragment_marshalling_error, + "with_fragment_mixed_outcomes": scenario_with_fragment_mixed_outcomes, "builder_to_archive_with_ingredient": scenario_builder_to_archive_with_ingredient, "builder_sign_jpeg_archive_roundtrip_ingredient_in_archive": scenario_builder_sign_jpeg_archive_roundtrip_ingredient_in_archive, "builder_write_ingredient_archive": scenario_builder_write_ingredient_archive, @@ -952,13 +1386,23 @@ def scenario_fork_stream_cleanup(iterations: int = 100) -> None: "reader_error_no_manifest": scenario_reader_error_no_manifest, "builder_error_invalid_manifest": scenario_builder_error_invalid_manifest, "reader_string_apis": scenario_reader_string_apis, + "signer_construction": scenario_signer_construction, + "builder_from_context_construction": + scenario_builder_from_context_construction, "fork_reader_collect": scenario_fork_reader_collect, "fork_contended_mutex": scenario_fork_contended_mutex, "fork_thread_local_orphan": scenario_fork_thread_local_orphan, "fork_gc_cycle": scenario_fork_gc_cycle, "fork_parent_frees_after_fork": scenario_fork_parent_frees_after_fork, + "fork_child_closes_then_parent_frees": + scenario_fork_child_closes_then_parent_frees, "fork_child_sys_exit": scenario_fork_child_sys_exit, "fork_stream_cleanup": scenario_fork_stream_cleanup, + "fork_swap_cleanup": scenario_fork_swap_cleanup, + "fork_contended_mutex_swap": scenario_fork_contended_mutex_swap, + "fork_contended_mutex_wrap": scenario_fork_contended_mutex_wrap, + "fork_consumed_signer": scenario_fork_consumed_signer, + "swap_chain_churn": scenario_swap_chain_churn, } diff --git a/tests/test_unit_tests.py b/tests/test_unit_tests.py index e0580ccb..f440d115 100644 --- a/tests/test_unit_tests.py +++ b/tests/test_unit_tests.py @@ -11,9 +11,12 @@ # specific language governing permissions and limitations under # each license. +import gc +import inspect import os import io import json +import re import unittest import ctypes import warnings @@ -30,7 +33,8 @@ from c2pa import Builder, C2paError as Error, Reader, C2paSigningAlg as SigningAlg, C2paSignerInfo, Signer, sdk_version, C2paBuilderIntent, C2paDigitalSourceType from c2pa import Settings, Context, ContextBuilder, ContextProvider -from c2pa.c2pa import Stream, LifecycleState, load_settings, create_signer, create_signer_from_info, ed25519_sign, format_embeddable +from c2pa.c2pa import Stream, LifecycleState, ManagedResource, load_settings, create_signer, create_signer_from_info, ed25519_sign, format_embeddable +import c2pa.c2pa as c2pa_module PROJECT_PATH = os.getcwd() @@ -6925,5 +6929,1633 @@ def test_callbacks_return_minus_one_after_stream_collected(self): self.assertEqual(flush_cb(None), -1) +class TestManagedResourceLifecycle(unittest.TestCase): + """Lifecycle primitives (_activate, _swap_handle, _wrap_native_handle), + the _owner_pid stamp that governs which process may free a handle, and + the ownership hand-offs between Python and the native library. + + setUp records frees instead of performing them, so a miscount reads as a + leak or a double-free rather than a crash. + Tests holding real handles call _use_real_frees() first. + """ + + class _FakeHandleResource(ManagedResource): + """Concrete subclass with no resources of its own.""" + + class _CallbackHoldingResource(ManagedResource): + """Mimics Signer: its _release() reads an attribute that _init_attrs() + is responsible for defaulting.""" + + def _release(self): + if self._callback_cb: + self._callback_cb = None + + class _ReleaseRecordingResource(ManagedResource): + """Records _release() calls for test asserts.""" + + def __init__(self): + super().__init__() + self.release_calls = 0 + + def _release(self): + self.release_calls += 1 + + class _ExtenderResource(ManagedResource): + """Am extender that owns a raw handle and wraps it via + _wrap_native_handle. It carries several attributes of its own, all + defaulted in _init_attrs (not __init__), and _release reads them, so a + missing attribute would surface as an AttributeError on teardown. + """ + + def _init_attrs(self): + super()._init_attrs() + self.label = "extender" + self.buffer = [] + self.released = False + + def _release(self): + # Reads attributes _init_attrs is responsible for defaulting. + self.buffer.append(self.label) + self.released = True + + def setUp(self): + self.data_dir = FIXTURES_DIR + self.freed = [] + self._real_free = ManagedResource._free_native_ptr + ManagedResource._free_native_ptr = staticmethod(self.freed.append) + + def tearDown(self): + ManagedResource._free_native_ptr = self._real_free + + def _free_counts(self): + counts = {} + for handle in self.freed: + counts[handle] = counts.get(handle, 0) + 1 + return counts + + def _use_real_frees(self): + """Undo free recorder, so native handles are really freed.""" + ManagedResource._free_native_ptr = self._real_free + + def _make_signer(self): + with open(os.path.join(self.data_dir, "es256_certs.pem"), "rb") as f: + certs = f.read() + with open(os.path.join(self.data_dir, "es256_private.key"), "rb") as f: + key = f.read() + return Signer.from_info(C2paSignerInfo( + b"es256", certs, key, b"http://timestamp.digicert.com")) + + def test_release_failure_still_frees_handle(self): + res = self._CallbackHoldingResource() + # _callback_cb is never set, so _release() raises AttributeError. + res._activate(0xBBBB) + + res.close() + + self.assertEqual(self.freed, [0xBBBB], + "handle leaked when _release() raised") + self.assertIsNone(res._handle) + + def test_release_failure_is_logged(self): + res = self._CallbackHoldingResource() + res._activate(0xBBBB) + + with self.assertLogs('c2pa', level='ERROR') as captured: + res.close() + + self.assertTrue( + any('Failed to release' in line for line in captured.output), + f"_release() failure was not logged: {captured.output}") + + def test_activate_rejects_null_handle(self): + res = self._FakeHandleResource() + + with self.assertRaises(Error) as ctx: + res._activate(None) + + self.assertIn("null handle", str(ctx.exception)) + self.assertEqual(res._lifecycle_state, LifecycleState.UNINITIALIZED) + + def test_activate_rejects_double_activation(self): + res = self._FakeHandleResource() + res._activate(0x1111) + + with self.assertRaises(Error) as ctx: + res._activate(0x2222) + + self.assertIn("already activated", str(ctx.exception)) + # The first handle is still owned, and is freed exactly once. + self.assertEqual(res._handle, 0x1111) + res.close() + self.assertEqual(self.freed, [0x1111]) + + def test_activate_rejects_reactivation_after_close(self): + res = self._FakeHandleResource() + res._activate(0x3333) + res.close() + self.freed.clear() + + with self.assertRaises(Error): + res._activate(0x4444) + + # Staying CLOSED is what prevents a second free of the handle. + self.assertEqual(res._lifecycle_state, LifecycleState.CLOSED) + self.assertIsNone(res._handle) + self.assertEqual(self.freed, []) + + def test_activate_does_not_mutate_on_rejection(self): + res = self._FakeHandleResource() + res._activate(0x5555) + + with self.assertRaises(Error): + res._activate(0x6666) + + self.assertEqual(res._handle, 0x5555, + "rejected activation replaced the handle") + self.assertEqual(res._lifecycle_state, LifecycleState.ACTIVE) + + def test_swap_handle_does_not_free_consumed_handle(self): + res = self._FakeHandleResource() + res._activate(0xAAA1) + + res._swap_handle(0xAAA2) + + # The FFI already owns and frees the old pointer, + # so freeing it here would be a double-free. + self.assertEqual(self.freed, []) + self.assertEqual(res._handle, 0xAAA2) + + res.close() + self.assertEqual(self.freed, [0xAAA2]) + + def test_swap_handle_requires_active_resource(self): + uninitialized = self._FakeHandleResource() + with self.assertRaises(Error) as ctx: + uninitialized._swap_handle(0x1) + self.assertIn("not active", str(ctx.exception)) + + closed = self._FakeHandleResource() + closed._activate(0x2) + closed.close() + with self.assertRaises(Error): + closed._swap_handle(0x3) + + def test_swap_handle_rejects_null_replacement(self): + res = self._FakeHandleResource() + res._activate(0x7777) + + with self.assertRaises(Error) as ctx: + res._swap_handle(None) + + self.assertIn("null handle", str(ctx.exception)) + self.assertEqual(res._handle, 0x7777) + self.assertEqual(res._lifecycle_state, LifecycleState.ACTIVE) + + def test_wrap_native_handle_bypasses_init(self): + seen = [] + + class Probe(ManagedResource): + def __init__(self): + raise AssertionError("__init__ must be bypassed") + + def _init_attrs(self): + super()._init_attrs() + self._tag = 'from _init_attrs' + + def _release(self): + seen.append(self._tag) + + obj = Probe._wrap_native_handle(0xC0DE) + + self.assertEqual(obj._tag, 'from _init_attrs') + self.assertEqual(obj._lifecycle_state, LifecycleState.ACTIVE) + self.assertTrue(obj.is_valid) + self.assertTrue(hasattr(obj, '_owner_pid')) + + obj.close() + self.assertEqual(seen, ['from _init_attrs'], + "_release() could not see the class's own attrs") + + def test_wrap_native_handle_rejects_null(self): + with self.assertRaises(Error): + self._FakeHandleResource._wrap_native_handle(None) + + def test_close_after_wrap_is_idempotent(self): + obj = self._FakeHandleResource._wrap_native_handle(0xD00D) + + obj.close() + obj.close() + + self.assertEqual(self.freed, [0xD00D], "handle freed more than once") + + def test_every_construction_path_records_owner_pid(self): + pid = os.getpid() + + plain = self._FakeHandleResource() + self.assertEqual(plain._owner_pid, pid) + + activated = self._FakeHandleResource() + activated._activate(0xA1) + self.assertEqual(activated._owner_pid, pid) + + wrapped = self._FakeHandleResource._wrap_native_handle(0xA2) + self.assertEqual(wrapped._owner_pid, pid) + + # A swap keeps the original stamp: + # the replacement handle was allocated by the same process + # that created the object. + wrapped._swap_handle(0xA3) + self.assertEqual(wrapped._owner_pid, pid) + + def test_foreign_child_skips_free_for_wrapped_and_swapped(self): + wrapped = self._FakeHandleResource._wrap_native_handle(0xC1) + wrapped._owner_pid = os.getpid() + 1 + wrapped.close() + + swapped = self._FakeHandleResource() + swapped._activate(0xC2) + swapped._swap_handle(0xC3) + swapped._owner_pid = os.getpid() + 1 + swapped.close() + + self.assertEqual(self.freed, [], + "forked child freed a pointer its parent still owns") + # The child must not free a pointer the parent still owns. + # The child does mark its own copies closed and nulls their handles, + # which stops the child from reusing a parent-owned handle. + self.assertEqual(wrapped._lifecycle_state, LifecycleState.CLOSED) + self.assertIsNone(wrapped._handle) + self.assertEqual(swapped._lifecycle_state, LifecycleState.CLOSED) + self.assertIsNone(swapped._handle) + + # A second foreign teardown is a no-op: still nothing freed. + wrapped.close() + swapped.close() + self.assertEqual(self.freed, []) + + def test_owning_process_frees_wrapped_and_swapped_exactly_once(self): + wrapped = self._FakeHandleResource._wrap_native_handle(0xC4) + wrapped.close() + wrapped.close() + + swapped = self._FakeHandleResource() + swapped._activate(0xC5) + swapped._swap_handle(0xC6) + swapped.close() + + # 0xC5 was consumed by the (simulated) FFI swap, + # so only the replacement is ours to free. + self.assertEqual(self._free_counts(), {0xC4: 1, 0xC6: 1}) + + def test_foreign_child_skips_release(self): + foreign = self._ReleaseRecordingResource() + foreign._activate(0xD1) + foreign._owner_pid = os.getpid() + 1 + foreign.close() + self.assertEqual(foreign.release_calls, 0) + + owned = self._ReleaseRecordingResource() + owned._activate(0xD2) + owned.close() + self.assertEqual(owned.release_calls, 1) + + def test_consumed_resource_frees_nothing_in_either_process(self): + owned = self._FakeHandleResource() + owned._activate(0xE1) + owned._mark_consumed() + owned.close() + + foreign = self._FakeHandleResource() + foreign._activate(0xE2) + foreign._mark_consumed() + foreign._owner_pid = os.getpid() + 1 + foreign.close() + + self.assertEqual(self.freed, []) + + # Consuming a handle hands the native pointer to a new owner, + # but the Python-side resources are still ours and we need to free. + def test_mark_consumed_releases_python_resources(self): + res = self._ReleaseRecordingResource() + res._activate(0xF1) + + res._mark_consumed() + + self.assertEqual(res.release_calls, 1) + self.assertEqual(res._lifecycle_state, LifecycleState.CLOSED) + self.assertIsNone(res._handle) + self.assertEqual(self.freed, []) + + def test_mark_consumed_swallows_failing_release(self): + res = self._CallbackHoldingResource() + res._activate(0xF2) + + with self.assertLogs("c2pa", level="ERROR"): + res._mark_consumed() + + self.assertEqual(res._lifecycle_state, LifecycleState.CLOSED) + self.assertIsNone(res._handle) + + def test_mark_consumed_in_foreign_process_skips_release(self): + res = self._ReleaseRecordingResource() + res._activate(0xF3) + res._owner_pid = os.getpid() + 1 + + res._mark_consumed() + + self.assertEqual(res.release_calls, 0) + self.assertEqual(res._lifecycle_state, LifecycleState.CLOSED) + self.assertIsNone(res._handle) + + def test_extender_wraps_handle_fully_built(self): + obj = self._ExtenderResource._wrap_native_handle(0xE0) + + # Every attribute _init_attrs defaults is present, + # even though __init__ never ran. + self.assertEqual(obj.label, "extender") + self.assertEqual(obj.buffer, []) + self.assertFalse(obj.released) + self.assertTrue(obj.is_valid) + self.assertEqual(obj._owner_pid, os.getpid()) + + # _release reads those attributes, so a missing one would raise here. + obj.close() + obj.close() + + self.assertTrue(obj.released) + self.assertEqual(obj.buffer, ["extender"]) + self.assertEqual(self.freed, [0xE0], "wrapped handle freed once") + + def test_extender_foreign_teardown_skips_native_free(self): + obj = self._ExtenderResource._wrap_native_handle(0xE1) + # Stamp a foreign owner: + # teardown runs in a process that did not create the handle, + # so it must not free the pointer or run _release. + obj._owner_pid = os.getpid() + 1 + + obj.close() + + self.assertEqual(self.freed, [], + "forked child freed a handle its parent still owns") + self.assertFalse(obj.released, "foreign teardown ran _release") + # The child marks its own copy closed and nulls the handle: safe (the + # parent holds a separate copy) and it stops the child reusing a + # parent-owned handle. + self.assertEqual(obj._lifecycle_state, LifecycleState.CLOSED) + self.assertIsNone(obj._handle) + + # A second foreign teardown stays a no-op, and any operation on the + # now-closed child copy fails loudly instead of reaching native code. + obj.close() + self.assertEqual(self.freed, []) + with self.assertRaises(Error): + obj._ensure_valid_state() + + def test_signer_init_rejects_null_pointer(self): + with self.assertRaises(Error): + Signer(None) + + def test_builder_from_archive_wraps_handle(self): + self._use_real_frees() + archive = io.BytesIO() + Builder({"claim_generator": "test", "format": "image/jpeg"}).to_archive( + archive) + + builder = Builder.from_archive(io.BytesIO(archive.getvalue())) + + self.assertTrue(builder.is_valid) + self.assertIsNone(builder._context) + self.assertFalse(builder._has_context_signer) + builder.close() + self.assertEqual(builder._lifecycle_state, LifecycleState.CLOSED) + + def test_context_build_failure_consumes_signer(self): + self._use_real_frees() + signer = self._make_signer() + real_build = c2pa_module._lib.c2pa_context_builder_build + c2pa_module._lib.c2pa_context_builder_build = lambda ptr: None + try: + with self.assertRaises(Error): + Context(signer=signer) + finally: + c2pa_module._lib.c2pa_context_builder_build = real_build + + self.assertIsNone(signer._handle, + "Signer still holds a pointer the native side freed") + self.assertEqual(signer._lifecycle_state, LifecycleState.CLOSED) + + # Nothing left to free, so close() must be a no-op. + freed = [] + real_free = ManagedResource._free_native_ptr + ManagedResource._free_native_ptr = staticmethod(freed.append) + try: + signer.close() + finally: + ManagedResource._free_native_ptr = real_free + self.assertEqual(freed, []) + + def test_context_with_signer_consumes_it_on_success(self): + self._use_real_frees() + signer = self._make_signer() + + context = Context(signer=signer) + + self.assertTrue(context.is_valid) + self.assertIsNone(signer._handle) + self.assertEqual(signer._lifecycle_state, LifecycleState.CLOSED) + self.assertTrue(context.has_signer) + context.close() + + def test_construction_failure_leaves_nothing_to_free(self): + # Activation happens after the null check, so a failed construction + # leaves no handle on the object for __del__ to find. + real_new = c2pa_module._lib.c2pa_context_new + c2pa_module._lib.c2pa_context_new = lambda: None + try: + with self.assertRaises(Error): + Context() + finally: + c2pa_module._lib.c2pa_context_new = real_new + + real_json = c2pa_module._lib.c2pa_builder_from_json + c2pa_module._lib.c2pa_builder_from_json = lambda j: None + try: + with self.assertRaises(Error): + Builder({"claim_generator": "test"}) + finally: + c2pa_module._lib.c2pa_builder_from_json = real_json + + +class TestManagedResourceObjects(TestContextAPIs): + """Tests native resource handling management when managed manually. + """ + + @staticmethod + def _ptr_addr(ptr): + """Address a ctypes pointer points at, or None for a null pointer. + + ctypes pointers compare by identity, not by value: two pointer objects + for the same address are unequal. Compare addresses instead. + """ + if not ptr: + return None + return ctypes.cast(ptr, ctypes.c_void_p).value + + def _instrument_frees(self): + """Record frees instead of performing them, and restore on teardown. + + This patches the base class, so every ManagedResource freed while the + patch is installed lands in the list, including objects the garbage + collector reclaims mid-test. Ask _free_count() about one handle rather + than asserting on the length of the list. + """ + freed = [] + real_free = ManagedResource._free_native_ptr + ManagedResource._free_native_ptr = staticmethod(freed.append) + self.addCleanup( + lambda: setattr( + ManagedResource, '_free_native_ptr', real_free)) + return freed + + def _free_count(self, freed, handle): + """How many times `handle` was freed, ignoring unrelated frees.""" + target = self._ptr_addr(handle) + self.assertIsNotNone(target, "cannot count frees of a null handle") + return sum(1 for ptr in freed if self._ptr_addr(ptr) == target) + + def _make_archive(self, manifest=None): + archive = io.BytesIO() + builder = Builder(manifest or self.test_manifest) + try: + builder.to_archive(archive) + finally: + builder.close() + archive.seek(0) + return archive + + def test_settings_activation_paths(self): + pid = os.getpid() + for label, factory in ( + ("Settings()", lambda: Settings()), + ("from_json", lambda: Settings.from_json('{"version_major": 1}')), + ("from_dict", lambda: Settings.from_dict({"version_major": 1})), + ): + with self.subTest(path=label): + settings = factory() + try: + self.assertTrue(settings.is_valid) + self.assertEqual(settings._owner_pid, pid) + finally: + settings.close() + + def test_context_activation_paths(self): + pid = os.getpid() + settings = Settings.from_dict({"version_major": 1}) + try: + for label, factory in ( + # No settings and no signer takes the c2pa_context_new path. + ("Context()", lambda: Context()), + # Anything else goes through the ContextBuilder path. + ("Context(settings)", lambda: Context(settings)), + ("from_dict", lambda: Context.from_dict({"version_major": 1})), + ("builder()", + lambda: Context.builder().with_settings(settings).build()), + ): + with self.subTest(path=label): + context = factory() + try: + self.assertTrue(context.is_valid) + self.assertEqual(context._owner_pid, pid) + finally: + context.close() + finally: + settings.close() + + def test_reader_activation_paths(self): + pid = os.getpid() + context = Context() + try: + with open(DEFAULT_TEST_FILE, "rb") as f: + from_stream = Reader("image/jpeg", f) + self.addCleanup(from_stream.close) + self.assertEqual(from_stream._owner_pid, pid) + + from_path = Reader(DEFAULT_TEST_FILE) + self.addCleanup(from_path.close) + self.assertEqual(from_path._owner_pid, pid) + + with open(DEFAULT_TEST_FILE, "rb") as f: + with_context = Reader(DEFAULT_TEST_FILE, context=context) + self.addCleanup(with_context.close) + self.assertEqual(with_context._owner_pid, pid) + + with open(DEFAULT_TEST_FILE, "rb") as f: + created = Reader.try_create("image/jpeg", f) + self.addCleanup(created.close) + self.assertEqual(created._owner_pid, pid) + finally: + context.close() + + def test_builder_activation_paths(self): + pid = os.getpid() + context = Context() + try: + plain = Builder(self.test_manifest) + self.addCleanup(plain.close) + self.assertEqual(plain._owner_pid, pid) + + from_json = Builder.from_json(self.test_manifest) + self.addCleanup(from_json.close) + self.assertEqual(from_json._owner_pid, pid) + + with_context = Builder(self.test_manifest, context=context) + self.addCleanup(with_context.close) + self.assertEqual(with_context._owner_pid, pid) + + # from_archive is the only caller of _wrap_native_handle, + # which bypasses __init__ entirely + wrapped = Builder.from_archive(self._make_archive()) + self.addCleanup(wrapped.close) + self.assertEqual(wrapped._owner_pid, pid) + self.assertIsNone(wrapped._context) + self.assertFalse(wrapped._has_context_signer) + finally: + context.close() + + def test_signer_activation_paths(self): + signer = self._ctx_make_signer() + self.addCleanup(signer.close) + self.assertTrue(signer.is_valid) + self.assertEqual(signer._owner_pid, os.getpid()) + + callback_signer = self._ctx_make_callback_signer() + self.addCleanup(callback_signer.close) + self.assertEqual(callback_signer._owner_pid, os.getpid()) + + def test_builder_with_archive_swaps_the_handle(self): + context = Context() + self.addCleanup(context.close) + builder = Builder(self.test_manifest, context=context) + original_handle = builder._handle + original_stamp = builder._owner_pid + + result = builder.with_archive(self._make_archive()) + + self.assertIs(result, builder, "with_archive should return self") + self.assertNotEqual(builder._handle, original_handle, + "the native handle was not replaced") + self.assertEqual(builder._lifecycle_state, LifecycleState.ACTIVE) + # The replacement came from this process, so the stamp still applies. + self.assertEqual(builder._owner_pid, original_stamp) + self.assertEqual(builder._owner_pid, os.getpid()) + builder.close() + + def test_reader_with_fragment_swaps_the_handle(self): + init_path = os.path.join(FIXTURES_DIR, "dashinit.mp4") + fragment_path = os.path.join(FIXTURES_DIR, "dash1.m4s") + context = Context() + self.addCleanup(context.close) + + with open(init_path, "rb") as init: + reader = Reader("video/mp4", init, context=context) + self.addCleanup(reader.close) + original_handle = reader._handle + + # The Reader consumed the first handle, so the init stream is reopened. + with open(init_path, "rb") as init, open(fragment_path, "rb") as frag: + result = reader.with_fragment("video/mp4", init, frag) + + self.assertIs(result, reader, "with_fragment should return self") + self.assertNotEqual(reader._handle, original_handle, + "the native handle was not replaced") + self.assertEqual(reader._lifecycle_state, LifecycleState.ACTIVE) + self.assertEqual(reader._owner_pid, os.getpid()) + + def test_swapped_builder_is_freed_exactly_once(self): + context = Context() + self.addCleanup(context.close) + builder = Builder(self.test_manifest, context=context) + original_handle = builder._handle + archive = self._make_archive() + + # Instrument across the swap so a free of the consumed pointer is recorded. + freed = self._instrument_frees() + builder.with_archive(archive) + swapped_handle = builder._handle + + self.assertEqual(self._free_count(freed, original_handle), 0, + "the swap freed the consumed pointer") + + builder.close() + builder.close() + + # Only the replacement is ours to free: the original was consumed by + # the FFI call that returned it. + self.assertEqual(self._free_count(freed, swapped_handle), 1) + self.assertEqual(self._free_count(freed, original_handle), 0) + + def test_repeated_swaps_on_one_builder(self): + # Each with_archive consumes the handle the previous one returned, so + # a chain of swaps is where a wrong swap surfaces: keeping the + # consumed pointer makes the next call raise UntrackedPointer from + # the native registry. + context = Context() + self.addCleanup(context.close) + builder = Builder(self.test_manifest, context=context) + self.addCleanup(builder.close) + + seen = [builder._handle] + for _ in range(5): + builder.with_archive(self._make_archive()) + self.assertEqual(builder._lifecycle_state, LifecycleState.ACTIVE) + seen.append(builder._handle) + + self.assertTrue(builder.is_valid) + self.assertEqual(builder._owner_pid, os.getpid()) + + def test_context_consumes_signer_but_not_settings(self): + settings = Settings.from_dict({"version_major": 1}) + signer = self._ctx_make_signer() + + context = Context(settings=settings, signer=signer) + self.addCleanup(context.close) + + # The signer pointer moved to the native context builder. + self.assertIsNone(signer._handle) + self.assertEqual(signer._lifecycle_state, LifecycleState.CLOSED) + self.assertTrue(context.has_signer) + + # Settings are copied, not consumed, so the caller still owns them. + self.assertTrue(settings.is_valid) + self.assertEqual(settings._owner_pid, os.getpid()) + settings.close() + + def test_consumed_signer_close_frees_nothing(self): + signer = self._ctx_make_signer() + # Captured before the context consumes it: close() nulls the handle, + # so afterwards there is no pointer left to identify the free by. + signer_handle = signer._handle + context = Context(signer=signer) + self.addCleanup(context.close) + + freed = self._instrument_frees() + signer.close() + + self.assertEqual(self._free_count(freed, signer_handle), 0, + "closing a consumed Signer freed a pointer the " + "context now owns") + + def test_builder_with_archive_null_return_consumes_self(self): + builder = Builder(self.test_manifest) + consumed_handle = builder._handle + real_call = c2pa_module._lib.c2pa_builder_with_archive + c2pa_module._lib.c2pa_builder_with_archive = lambda b, s: None + try: + with self.assertRaises(Error): + builder.with_archive(self._make_archive()) + finally: + c2pa_module._lib.c2pa_builder_with_archive = real_call + + # The FFI consumed the old handle and returned no replacement, + # so there is nothing left for this object to own... + self.assertIsNone(builder._handle) + self.assertEqual(builder._lifecycle_state, LifecycleState.CLOSED) + + freed = self._instrument_frees() + builder.close() + self.assertEqual(self._free_count(freed, consumed_handle), 0, + "close() freed a handle the FFI already consumed") + + def test_reader_with_fragment_null_return_consumes_self(self): + init_path = os.path.join(FIXTURES_DIR, "dashinit.mp4") + fragment_path = os.path.join(FIXTURES_DIR, "dash1.m4s") + with open(init_path, "rb") as init: + reader = Reader("video/mp4", init) + consumed_handle = reader._handle + + # The mock sets no error, which no native path does. Plant an + # operation-style one so a stale tag from another test is not read. + c2pa_module._lib.c2pa_error_set_last(b"Other: mocked null return") + + real_call = c2pa_module._lib.c2pa_reader_with_fragment + c2pa_module._lib.c2pa_reader_with_fragment = ( + lambda r, f, s, frag: None) + try: + with open(init_path, "rb") as init, \ + open(fragment_path, "rb") as frag: + with self.assertRaises(Error): + reader.with_fragment("video/mp4", init, frag) + finally: + c2pa_module._lib.c2pa_reader_with_fragment = real_call + + self.assertIsNone(reader._handle) + self.assertEqual(reader._lifecycle_state, LifecycleState.CLOSED) + + freed = self._instrument_frees() + reader.close() + self.assertEqual(self._free_count(freed, consumed_handle), 0, + "close() freed a handle the FFI already consumed") + + def test_reader_with_fragment_ffi_raise_consumes_self(self): + # If the ctypes call itself raises (not a null return), the callee has + # already consumed the old handle, so with_fragment must mark self + # consumed rather than leave a dangling pointer that close() would + # double-free. + init_path = os.path.join(FIXTURES_DIR, "dashinit.mp4") + fragment_path = os.path.join(FIXTURES_DIR, "dash1.m4s") + with open(init_path, "rb") as init: + reader = Reader("video/mp4", init) + consumed_handle = reader._handle + + def _raise(*_args): + raise RuntimeError("boom") + + real_call = c2pa_module._lib.c2pa_reader_with_fragment + c2pa_module._lib.c2pa_reader_with_fragment = _raise + try: + with open(init_path, "rb") as init, \ + open(fragment_path, "rb") as frag: + with self.assertRaises(Error): + reader.with_fragment("video/mp4", init, frag) + finally: + c2pa_module._lib.c2pa_reader_with_fragment = real_call + + self.assertIsNone(reader._handle) + self.assertEqual(reader._lifecycle_state, LifecycleState.CLOSED) + + freed = self._instrument_frees() + reader.close() + self.assertEqual(self._free_count(freed, consumed_handle), 0, + "close() freed a handle the FFI already consumed") + + # Consume-and-return ownership: the native call takes the handle partway + # through its body, so a null return does not say on its own whether the + # handle was consumed. These pin the classification down. + + @staticmethod + def _is_pre_consume_rejection(error_message): + """True if this native error means ownership never transferred.""" + if not error_message: + return False + return any(tag in error_message + for tag in ManagedResource._PRE_CONSUME_ERROR_TAGS) + + def _stale_reader_handle(self): + """A freed, untracked pointer, captured before close() nulls it. + + Take a fresh one per call: recycled addresses become tracked again. + """ + init_path = os.path.join(FIXTURES_DIR, "dashinit.mp4") + with open(init_path, "rb") as init: + victim = Reader("video/mp4", init) + stale = ctypes.cast(victim._handle, + ctypes.POINTER(c2pa_module.C2paReader)) + victim.close() + return stale + + @staticmethod + def _untracked_reader_handle(): + """A pointer the native registry never handed out. + + Rejected like a stale handle, but not a freed address, so it cannot + be recycled and start passing the registry lookup. + """ + buf = ctypes.create_string_buffer(64) + return (ctypes.cast(buf, ctypes.POINTER(c2pa_module.C2paReader)), + buf) + + def test_with_fragment_pre_consume_rejection_keeps_handle(self): + # Rejected before Box::from_raw, so nothing was consumed and the + # handle is still ours. Dropping it here would leak it. + init_path = os.path.join(FIXTURES_DIR, "dashinit.mp4") + fragment_path = os.path.join(FIXTURES_DIR, "dash1.m4s") + with open(init_path, "rb") as init: + reader = Reader("video/mp4", init) + real_handle = reader._handle + + reader._handle = self._stale_reader_handle() + try: + with open(init_path, "rb") as init, \ + open(fragment_path, "rb") as frag: + with self.assertRaises(Error) as caught: + reader.with_fragment("video/mp4", init, frag) + finally: + reader._handle = real_handle + + self.assertIn("UntrackedPointer", str(caught.exception)) + # Ownership never transferred, so the resource stays usable. + self.assertIsNotNone(reader._handle) + self.assertEqual(reader._lifecycle_state, LifecycleState.ACTIVE) + self.assertTrue(reader.json()) + reader.close() + + def test_with_fragment_pre_consume_rejection_does_not_leak(self): + # A handle dropped on this path leaks one reader per call. + init_path = os.path.join(FIXTURES_DIR, "dashinit.mp4") + fragment_path = os.path.join(FIXTURES_DIR, "dash1.m4s") + + for _ in range(25): + with open(init_path, "rb") as init: + reader = Reader("video/mp4", init) + real_handle = reader._handle + reader._handle = self._stale_reader_handle() + try: + with open(init_path, "rb") as init, \ + open(fragment_path, "rb") as frag: + with self.assertRaises(Error): + reader.with_fragment("video/mp4", init, frag) + finally: + reader._handle = real_handle + # Still owns a working handle every time round, so nothing + # leaked: a dropped handle would leave the reader closed. + self.assertEqual(reader._lifecycle_state, LifecycleState.ACTIVE) + self.assertTrue(reader.json()) + reader.close() + + def test_with_archive_post_consume_failure_consumes_handle(self): + # Ownership taken, then the operation failed: the handle is gone, + # so close() must not free it again. + builder = Builder(json.dumps( + {"claim_generator_info": [{"name": "test", "version": "0.1"}], + "assertions": []})) + consumed_handle = builder._handle + + with self.assertRaises(Error): + builder.with_archive(io.BytesIO(b"not a valid archive")) + + self.assertIsNone(builder._handle) + self.assertEqual(builder._lifecycle_state, LifecycleState.CLOSED) + + freed = self._instrument_frees() + builder.close() + self.assertEqual(self._free_count(freed, consumed_handle), 0, + "close() freed a handle the FFI already consumed") + + def test_with_fragment_marshalling_error_keeps_handle(self): + # Never reaches native code, so nothing was consumed. The old + # blanket except marked it consumed here and leaked the handle. + init_path = os.path.join(FIXTURES_DIR, "dashinit.mp4") + with open(init_path, "rb") as init: + reader = Reader("video/mp4", init) + real_handle = reader._handle + + def _bad_marshalling(*_args): + raise ctypes.ArgumentError("wrong argument type") + + real_call = c2pa_module._lib.c2pa_reader_with_fragment + c2pa_module._lib.c2pa_reader_with_fragment = _bad_marshalling + try: + with open(init_path, "rb") as init, \ + open(init_path, "rb") as frag: + with self.assertRaises(ctypes.ArgumentError): + reader.with_fragment("video/mp4", init, frag) + finally: + c2pa_module._lib.c2pa_reader_with_fragment = real_call + + self.assertIs(reader._handle, real_handle) + self.assertEqual(reader._lifecycle_state, LifecycleState.ACTIVE) + self.assertTrue(reader.json(), + "a marshalling failure never reached the FFI, so " + "the reader must still be usable") + + reader.close() + + def test_unknown_failure_drops_handle_without_freeing(self): + # Ownership unknowable, so the handle is let go rather than freed. + # Needs a mock: every real null return sets an error. + init_path = os.path.join(FIXTURES_DIR, "dashinit.mp4") + fragment_path = os.path.join(FIXTURES_DIR, "dash1.m4s") + with open(init_path, "rb") as init: + reader = Reader("video/mp4", init) + + consumed_handle = reader._handle + # Not a pre-consume tag, so the mocked failure reads as unplaceable. + c2pa_module._lib.c2pa_error_set_last(b"Other: mocked null return") + real_call = c2pa_module._lib.c2pa_reader_with_fragment + c2pa_module._lib.c2pa_reader_with_fragment = ( + lambda r, f, s, frag: None) + try: + with open(init_path, "rb") as init, \ + open(fragment_path, "rb") as frag: + with self.assertRaises(Error): + reader.with_fragment("video/mp4", init, frag) + finally: + c2pa_module._lib.c2pa_reader_with_fragment = real_call + + # Unplaceable, so the handle is let go rather than freed twice. + self.assertIsNone(reader._handle) + self.assertEqual(reader._lifecycle_state, LifecycleState.CLOSED) + + freed = self._instrument_frees() + reader.close() + self.assertEqual(self._free_count(freed, consumed_handle), 0, + "close() freed a handle of unknown ownership") + + def test_pre_consume_rejection_is_typed(self): + # Rejections arrive wrapped as "Other: UntrackedPointer: 0x...". + self.assertTrue(self._is_pre_consume_rejection( + "Other: UntrackedPointer: 0x600001234567")) + self.assertTrue(self._is_pre_consume_rejection( + "Other: WrongPointerType: 0x600001234567")) + self.assertFalse(self._is_pre_consume_rejection( + "Verify: invalid JUMBF header")) + self.assertFalse(self._is_pre_consume_rejection(None)) + self.assertFalse(self._is_pre_consume_rejection("")) + + def test_pre_consume_tags_still_match_the_native_wording(self): + # Classification keys on error text (the numeric code is not + # exported), so a native rename would silently misjudge ownership. + init_path = os.path.join(FIXTURES_DIR, "dashinit.mp4") + fragment_path = os.path.join(FIXTURES_DIR, "dash1.m4s") + with open(init_path, "rb") as init: + reader = Reader("video/mp4", init) + real_handle = reader._handle + + reader._handle = self._stale_reader_handle() + try: + with open(init_path, "rb") as init, \ + open(fragment_path, "rb") as frag: + with self.assertRaises(Error) as caught: + reader.with_fragment("video/mp4", init, frag) + finally: + reader._handle = real_handle + + message = str(caught.exception) + self.assertTrue( + self._is_pre_consume_rejection(message), + f"the native rejection wording changed and no longer matches " + f"_PRE_CONSUME_ERROR_TAGS; ownership will be misjudged: " + f"{message!r}") + reader.close() + + def test_stale_handle_is_actually_rejected_every_time(self): + # A handle that stopped being rejected would quietly measure the + # success path. Uses the never-allocated buffer: freed addresses get + # recycled and start passing the registry lookup. + init_path = os.path.join(FIXTURES_DIR, "dashinit.mp4") + fragment_path = os.path.join(FIXTURES_DIR, "dash1.m4s") + + for _ in range(25): + with open(init_path, "rb") as init: + reader = Reader("video/mp4", init) + real_handle = reader._handle + bogus, _buf = self._untracked_reader_handle() + reader._handle = bogus + try: + with open(init_path, "rb") as init, \ + open(fragment_path, "rb") as frag: + with self.assertRaises(Error) as caught: + reader.with_fragment("video/mp4", init, frag) + self.assertTrue( + self._is_pre_consume_rejection( + str(caught.exception)), + "the bogus handle was not rejected, so this stopped " + "exercising the pre-consume path") + finally: + reader._handle = real_handle + reader.close() + + def test_perf_scenario_bogus_handle_is_rejected(self): + # The perf scenarios use a plain buffer so looping does not swamp + # the measurement. It still has to produce a real rejection. + init_path = os.path.join(FIXTURES_DIR, "dashinit.mp4") + fragment_path = os.path.join(FIXTURES_DIR, "dash1.m4s") + with open(init_path, "rb") as init: + reader = Reader("video/mp4", init) + real_handle = reader._handle + + bogus, _buf = self._untracked_reader_handle() + reader._handle = bogus + try: + with open(init_path, "rb") as init, \ + open(fragment_path, "rb") as frag: + with self.assertRaises(Error) as caught: + reader.with_fragment("video/mp4", init, frag) + finally: + reader._handle = real_handle + + self.assertTrue( + self._is_pre_consume_rejection(str(caught.exception)), + "the perf scenarios' bogus handle is no longer rejected, so " + "with_fragment_pre_consume_rejection measures nothing") + # Handle kept, so the reader still works and frees normally. + self.assertEqual(reader._lifecycle_state, LifecycleState.ACTIVE) + self.assertTrue(reader.json()) + 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. + 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 "") + + # Pre-consume rejection: reports UntrackedPointer, not NotSupported. + with open(init_path, "rb") as init: + reader = Reader("video/mp4", init) + real_handle = reader._handle + reader._handle = self._stale_reader_handle() + try: + with open(init_path, "rb") as init, \ + open(fragment_path, "rb") as frag: + with self.assertRaises(Error) as caught: + reader.with_fragment("video/mp4", init, frag) + finally: + reader._handle = real_handle + self.assertIn("UntrackedPointer", str(caught.exception)) + self.assertNotIn("NotSupported", str(caught.exception)) + reader.close() + + # Post-consume failure: reports the operation error, not the + # UntrackedPointer left by the step above. + builder = Builder(json.dumps( + {"claim_generator_info": [{"name": "test", "version": "0.1"}], + "assertions": []})) + with self.assertRaises(Error) as caught: + builder.with_archive(io.BytesIO(b"not a valid archive")) + self.assertNotIn("UntrackedPointer", str(caught.exception)) + builder.close() + + def test_pre_consume_classification_holds_across_threads(self): + # The error slot is thread-local, so concurrent calls do not mask + # each other's errors and misjudge ownership. + init_path = os.path.join(FIXTURES_DIR, "dashinit.mp4") + fragment_path = os.path.join(FIXTURES_DIR, "dash1.m4s") + init_bytes = open(init_path, "rb").read() + frag_bytes = open(fragment_path, "rb").read() + problems = [] + + def worker(): + for _ in range(10): + reader = Reader("video/mp4", io.BytesIO(init_bytes)) + real_handle = reader._handle + # A never-allocated buffer, not a freed pointer: with several + # threads churning the allocator, a freed address gets reused + # and starts passing the registry lookup, which would end the + # iteration in a real consume instead of a rejection. + bogus, _buf = self._untracked_reader_handle() + reader._handle = bogus + try: + reader.with_fragment("video/mp4", + io.BytesIO(init_bytes), + io.BytesIO(frag_bytes)) + problems.append("rejection did not raise") + except Error as e: + if not self._is_pre_consume_rejection(str(e)): + problems.append(f"misclassified: {str(e)[:60]}") + finally: + reader._handle = real_handle + if reader._lifecycle_state != LifecycleState.ACTIVE: + problems.append("handle dropped on a rejection") + reader.close() + + threads = [threading.Thread(target=worker) for _ in range(6)] + for t in threads: + t.start() + for t in threads: + t.join() + + 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 + + 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") + + def test_read_native_error_returns_none_for_an_empty_message(self): + # c2pa_error() returns an owned pointer to "" when no error is set, + # never NULL, so the pointer cannot be the "is there an error" test. + original = c2pa_module._lib.c2pa_error + empty = ctypes.create_string_buffer(b"") + + try: + c2pa_module._lib.c2pa_error = lambda: ctypes.cast( + empty, ctypes.c_void_p).value + self.assertIsNone( + c2pa_module._read_native_error(), + "an empty native message must read as None, otherwise " + "callers' 'if error:' checks are only accidentally right") + 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. + init_path = os.path.join(FIXTURES_DIR, "dashinit.mp4") + fragment_path = os.path.join(FIXTURES_DIR, "dash1.m4s") + + c2pa_module._lib.c2pa_error_set_last( + b"UntrackedPointer: 0xdeadbeef") + + with open(init_path, "rb") as init: + reader = Reader("video/mp4", init) + + real_call = c2pa_module._lib.c2pa_reader_with_fragment + c2pa_module._lib.c2pa_reader_with_fragment = ( + lambda r, f, s, frag: None) + try: + with open(init_path, "rb") as init, \ + open(fragment_path, "rb") as frag: + with self.assertRaises(Error): + 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() + + # 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 + # only tests that drive the primitive as the generic entry point it is. + + def _raw_builder_handle(self): + manifest = json.dumps( + {"claim_generator": "raw_ffi_test", "format": "image/jpeg"} + ).encode("utf-8") + handle = c2pa_module._lib.c2pa_builder_from_json(manifest) + self.assertTrue(handle, "the FFI did not return a builder pointer") + return handle + + def test_wrap_raw_ffi_builder_pointer(self): + builder = Builder._wrap_native_handle(self._raw_builder_handle()) + + self.assertTrue(builder.is_valid) + self.assertEqual(builder._lifecycle_state, LifecycleState.ACTIVE) + self.assertEqual(builder._owner_pid, os.getpid()) + + archive = io.BytesIO() + builder.to_archive(archive) + self.assertTrue(archive.getvalue()) + + builder.close() + self.assertEqual(builder._lifecycle_state, LifecycleState.CLOSED) + self.assertIsNone(builder._handle) + + def test_wrap_raw_ffi_settings_pointer(self): + handle = c2pa_module._lib.c2pa_settings_new() + self.assertTrue(handle) + + settings = Settings._wrap_native_handle(handle) + try: + self.assertTrue(settings.is_valid) + self.assertEqual(settings._owner_pid, os.getpid()) + settings.set("version_major", "1") + finally: + settings.close() + + def test_wrap_raw_ffi_context_pointer(self): + handle = c2pa_module._lib.c2pa_context_new() + self.assertTrue(handle) + + context = Context._wrap_native_handle(handle) + try: + self.assertTrue(context.is_valid) + self.assertEqual(context._owner_pid, os.getpid()) + # Handing the wrapped pointer back to the FFI proves it is live. + builder = Builder(self.test_manifest, context=context) + self.assertTrue(builder.is_valid) + builder.close() + finally: + context.close() + + def test_wrapped_raw_pointer_freed_exactly_once(self): + handle = self._raw_builder_handle() + freed = self._instrument_frees() + + builder = Builder._wrap_native_handle(handle) + builder.close() + builder.close() + del builder + gc.collect() + + self.assertEqual(self._free_count(freed, handle), 1, + "wrapped handle not freed exactly once") + + def test_wrap_supplies_defaults(self): + # _init_attrs() runs on the wrap path, so an instance built around a + # raw handle still has everything the rest of the class reads. + builder = Builder._wrap_native_handle(self._raw_builder_handle()) + self.addCleanup(builder.close) + + self.assertIsNone(builder._context) + self.assertFalse(builder._has_context_signer) + + def test_init_attrs_covers_what_init_sets(self): + # Anything __init__ sets but _init_attrs() misses is absent on a + # wrapped instance, which is the trap _init_attrs() exists to close. + for cls in (Builder, Context, Reader, Signer): + with self.subTest(cls=cls.__name__): + defaulted = set(re.findall( + r"self\.(_[a-z][a-z0-9_]*)\s*=", + inspect.getsource(cls._init_attrs))) + assigned = set(re.findall( + r"self\.(_[a-z][a-z0-9_]*)\s*=", + inspect.getsource(cls.__init__))) + self.assertEqual( + assigned - defaulted, set(), + f"{cls.__name__}.__init__ sets attributes that " + f"_init_attrs() does not default") + + def test_init_attrs_overrides_chain_to_super(self): + # A subclass of these would silently lose the parent's defaults if + # the chain were broken. + for cls in (Builder, Context, Reader, Signer): + with self.subTest(cls=cls.__name__): + self.assertIn( + "super()._init_attrs()", + inspect.getsource(cls._init_attrs), + f"{cls.__name__}._init_attrs() does not chain to super()") + + def test_wrap_raw_ffi_signer_pointer(self): + # Signer._release() reads _callback_cb, so a wrap that skipped the + # defaults would fail during cleanup rather than at the wrap. + freed = self._instrument_frees() + + signer = Signer._wrap_native_handle(0xABCD) + self.assertIsNone(signer._callback_cb) + self.assertEqual(signer._owner_pid, os.getpid()) + + # Cleanup swallows a failing _release(), so the error log is the only + # way to see one. + with self.assertNoLogs("c2pa", level="ERROR"): + signer.close() + + self.assertEqual(freed, [0xABCD]) + + def test_signer_release_clears_callback(self): + signer = self._ctx_make_callback_signer() + self.assertIsNotNone(signer._callback_cb) + + signer.close() + + self.assertIsNone(signer._callback_cb) + + def test_builder_release_clears_context(self): + context = Context() + self.addCleanup(context.close) + builder = Builder(self.test_manifest, context=context) + + builder.close() + + self.assertIsNone(builder._context, + "closed Builder still pins its Context") + # The Builder does not own the Context, so it must not close it. + self.assertTrue(context.is_valid) + + def test_consumed_reader_closes_backing_file(self): + # A failed with_fragment consumes the reader. + # # Reader(path) opened the backing file itself, + # so nothing else will ever close it. + reader = Reader(DEFAULT_TEST_FILE) + backing_file = reader._backing_file + self.assertFalse(backing_file.closed) + + real_call = c2pa_module._lib.c2pa_reader_with_fragment + c2pa_module._lib.c2pa_reader_with_fragment = ( + lambda r, f, s, frag: None) + try: + with open(DEFAULT_TEST_FILE, "rb") as main, \ + open(DEFAULT_TEST_FILE, "rb") as frag: + with self.assertRaises(Error): + reader.with_fragment("image/jpeg", main, frag) + finally: + c2pa_module._lib.c2pa_reader_with_fragment = real_call + + self.assertTrue(backing_file.closed, + "consumed Reader leaked its backing file") + + def test_consumed_builder_releases_context(self): + context = Context() + self.addCleanup(context.close) + builder = Builder(self.test_manifest, context=context) + archive = self._make_archive() + + real_call = c2pa_module._lib.c2pa_builder_with_archive + c2pa_module._lib.c2pa_builder_with_archive = lambda b, s: None + try: + with self.assertRaises(Error): + builder.with_archive(archive) + finally: + c2pa_module._lib.c2pa_builder_with_archive = real_call + + self.assertIsNone(builder._context, + "consumed Builder still pins its Context") + self.assertTrue(context.is_valid) + + def test_context_takes_callback_before_consuming_signer(self): + # Consuming the signer releases its callback reference, + # so the Context has to take it first or the callback dies with the signer. + signer = self._ctx_make_callback_signer() + callback = signer._callback_cb + self.assertIsNotNone(callback) + + context = Context(signer=signer) + self.addCleanup(context.close) + + self.assertIs(context._signer_callback_cb, callback) + self.assertIsNone(signer._callback_cb) + + def test_reader_close_closes_backing_file(self): + # _close_streams reads the attrs _init_attrs() defaults, so this is + # the regression guard for reading them directly. + reader = Reader(DEFAULT_TEST_FILE) + reader.json() + backing_file = reader._backing_file + self.assertIsNotNone(backing_file) + + reader.close() + + self.assertTrue(backing_file.closed, "Reader left its file open") + self.assertIsNone(reader._backing_file) + self.assertIsNone(reader._manifest_json_str_cache) + self.assertIsNone(reader._manifest_data_cache) + + def test_consumed_reader_clears_caches(self): + # Consuming marks the reader closed, so close() will not run later. + # Anything cleanup owes the object has to happen at consume time. + reader = Reader(DEFAULT_TEST_FILE) + reader.json() + self.assertIsNotNone(reader._manifest_json_str_cache) + + real_call = c2pa_module._lib.c2pa_reader_with_fragment + c2pa_module._lib.c2pa_reader_with_fragment = ( + lambda r, f, s, frag: None) + try: + with open(DEFAULT_TEST_FILE, "rb") as main, \ + open(DEFAULT_TEST_FILE, "rb") as frag: + with self.assertRaises(Error): + reader.with_fragment("image/jpeg", main, frag) + finally: + c2pa_module._lib.c2pa_reader_with_fragment = real_call + + self.assertIsNone(reader._manifest_json_str_cache, + "consumed Reader kept its manifest cache") + self.assertIsNone(reader._manifest_data_cache) + + def test_reader_del_clears_caches(self): + # __del__ goes through _cleanup_resources, not close(), so cache + # clearing has to live somewhere both paths reach. + reader = Reader(DEFAULT_TEST_FILE) + reader.json() + self.assertIsNotNone(reader._manifest_json_str_cache) + + # __del__ runs _cleanup_resources directly, so drive that rather than + # dropping the reference: the assertions need the object afterwards. + reader._cleanup_resources() + + self.assertIsNone(reader._manifest_json_str_cache, + "cleanup left the manifest cache alive") + self.assertIsNone(reader._manifest_data_cache) + + def test_from_archive_frees_handle_when_wrap_fails(self): + # The wrap raising means no Python object took ownership, so + # from_archive still holds the handle and has to free it. + archive = self._make_archive() # closes a Builder; keep it off the count + freed = self._instrument_frees() + real_wrap = Builder._wrap_native_handle + + def _boom(*args, **kwargs): + raise Error("wrap failed") + + Builder._wrap_native_handle = _boom + try: + with self.assertRaises(Error): + Builder.from_archive(archive) + finally: + Builder._wrap_native_handle = real_wrap + + self.assertEqual(len(freed), 1, + "from_archive leaked the handle when the wrap failed") + + def test_sign_failure_chains_the_original_exception(self): + # The wrapper re-raises as C2paError; losing __cause__ hides which + # call actually failed. + builder = Builder(self.test_manifest) + signer = self._ctx_make_signer() + self.addCleanup(signer.close) + + sentinel = RuntimeError("native call blew up") + real_sign = c2pa_module._lib.c2pa_builder_sign + + def _boom(*args): + raise sentinel + + c2pa_module._lib.c2pa_builder_sign = _boom + 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.assertIs(ctx.exception.__cause__, sentinel, + "signing error dropped the original exception") + + +class TestErrorPlumbing(unittest.TestCase): + """Covers the error helpers themselves, which had no direct tests.""" + + def _set_native_error(self, text): + c2pa_module._lib.c2pa_error_set_last(text.encode('utf-8')) + + def test_every_error_tag_maps_to_its_typed_subclass(self): + # The wire text is "Tag: message"; each tag gets its own subclass so + # callers can catch precisely. + tags = [ + "Assertion", "AssertionNotFound", "Decoding", "Encoding", + "FileNotFound", "Io", "Json", "Manifest", "ManifestNotFound", + "NotSupported", "Other", "RemoteManifest", "ResourceNotFound", + "Signature", "Verify", "UntrackedPointer", "WrongPointerType", + ] + for tag in tags: + with self.subTest(tag=tag): + expected = getattr(Error, tag) + with self.assertRaises(expected): + c2pa_module._raise_typed_c2pa_error(f"{tag}: detail") + + def test_unmapped_tag_falls_back_to_base_error(self): + with self.assertRaises(Error) as ctx: + c2pa_module._raise_typed_c2pa_error("Nonsense: detail") + # Base class only: no subclass should claim an unknown tag. + self.assertIs(type(ctx.exception), Error) + + def test_check_ffi_operation_result_raises_with_native_message(self): + self._set_native_error("Io: disk exploded") + with self.assertRaises(Error) as ctx: + c2pa_module._check_ffi_operation_result(None, "fallback text") + self.assertIn("disk exploded", str(ctx.exception)) + + def test_check_ffi_operation_result_uses_fallback_when_slot_empty(self): + # The slot is sticky and thread-local, so a fresh thread is the only + # way to observe it unset. + captured = [] + + def run(): + try: + c2pa_module._check_ffi_operation_result(None, "fallback text") + except BaseException as e: # noqa: BLE001 - reported below + captured.append(e) + + t = threading.Thread(target=run) + t.start() + t.join() + + self.assertEqual(len(captured), 1, "expected a raise on failure") + self.assertIsInstance(captured[0], Error) + self.assertIn("fallback text", str(captured[0])) + + def test_check_ffi_operation_result_passes_success_through(self): + self.assertEqual( + c2pa_module._check_ffi_operation_result(42, "unused"), 42) + + def test_stream_creation_failure_reports_a_real_message(self): + # Regression: used to raise bare Exception("...: None"), because + # _parse_operation_result_for_error never returns a message. + real = c2pa_module._lib.c2pa_create_stream + c2pa_module._lib.c2pa_create_stream = lambda *a: None + try: + # With an error set, the message must carry it. + self._set_native_error("Io: stream refused") + with self.assertRaises(Error) as ctx: + c2pa_module.Stream(io.BytesIO(b"x")) + self.assertIn("stream refused", str(ctx.exception)) + + # With the slot unset, the old code raised bare Exception. + captured = [] + + def run(): + try: + c2pa_module.Stream(io.BytesIO(b"x")) + except BaseException as e: # noqa: BLE001 - reported below + captured.append(e) + + t = threading.Thread(target=run) + t.start() + t.join() + + self.assertEqual(len(captured), 1, "expected a raise on failure") + self.assertIsInstance( + captured[0], Error, + "stream failure must raise C2paError, not bare Exception") + self.assertNotIn("None", str(captured[0])) + finally: + c2pa_module._lib.c2pa_create_stream = real + + def test_supported_mime_types_raises_instead_of_returning_empty(self): + # Regression: `if error:` was always False, so a native failure + # returned an empty list instead of raising. Only visible with the + # slot unset, hence the fresh thread. + captured = [] + + def run(): + try: + captured.append( + c2pa_module._get_supported_mime_types( + lambda count: None, None)) + except BaseException as e: # noqa: BLE001 - reported below + captured.append(e) + + t = threading.Thread(target=run) + t.start() + t.join() + + self.assertIsInstance( + captured[0], Error, + "a failed MIME lookup returned data instead of raising") + + def test_supported_mime_types_reports_the_native_message(self): + self._set_native_error("Io: mime lookup failed") + with self.assertRaises(Error) as ctx: + c2pa_module._get_supported_mime_types(lambda count: None, None) + self.assertIn("mime lookup failed", str(ctx.exception)) + + +class TestErrorsStillRaiseAfterCleanup(unittest.TestCase): + """Each surface that lost a _clear_error_state() call still reports.""" + + def test_reader_on_garbage_raises(self): + with self.assertRaises(Error): + Reader("image/jpeg", io.BytesIO(b"not an image")).json() + + def test_signer_from_info_with_bad_certs_raises(self): + with self.assertRaises(Error): + c2pa_module.create_signer_from_info(C2paSignerInfo( + alg=b"es256", + sign_cert=b"not a certificate", + private_key=b"not a key", + ta_url=b"", + )) + + def test_ed25519_sign_with_empty_data_raises(self): + with self.assertRaises(Error): + c2pa_module.ed25519_sign(b"", "not a key") + + if __name__ == '__main__': unittest.main(warnings='ignore') diff --git a/tests/test_unit_tests_threaded.py b/tests/test_unit_tests_threaded.py index efb7ba27..e7d49b16 100644 --- a/tests/test_unit_tests_threaded.py +++ b/tests/test_unit_tests_threaded.py @@ -96,12 +96,17 @@ def test_no_stamp_calls_free(self): obj._cleanup_resources() mock_free.assert_called_once() - def test_foreign_pid_leaves_state_unchanged(self): - """Guard returns early; lifecycle state stays ACTIVE (not CLOSED).""" + def test_foreign_pid_marks_closed_without_free(self): + """A foreign child skips the native free but marks its own copy closed + and nulls the handle, so the child cannot reuse a parent-owned handle. + The parent holds a separate copy and frees it independently. + """ obj = _make_resource(pid_offset=1) - with patch('c2pa.c2pa._lib'): + with patch.object(ManagedResource, '_free_native_ptr') as mock_free: obj._cleanup_resources() - self.assertEqual(obj._lifecycle_state, LifecycleState.ACTIVE) + mock_free.assert_not_called() + self.assertEqual(obj._lifecycle_state, LifecycleState.CLOSED) + self.assertIsNone(obj._handle) def test_double_cleanup_is_idempotent(self): """Second call is a no-op after successful first cleanup.""" @@ -2911,5 +2916,123 @@ def thread_work(thread_id): self.assertNotEqual(current_manifest["active_manifest"], thread_manifest_data[other_thread_id]["active_manifest"]) +class TestManagedResourceCrossThread(unittest.TestCase): + """Tests cross-thread resources handling, especially closing/releasind. + """ + + def setUp(self): + self.freed = [] + self._real_free = ManagedResource._free_native_ptr + ManagedResource._free_native_ptr = staticmethod(self.freed.append) + + def tearDown(self): + ManagedResource._free_native_ptr = self._real_free + + def _free_counts(self): + counts = {} + for handle in self.freed: + counts[handle] = counts.get(handle, 0) + 1 + return counts + + def test_cross_thread_create_and_close_frees_exactly_once(self): + count = 300 + pid = os.getpid() + + def create(index): + res = _ConcreteResource() + res._activate(0x10000 + index) + return res + + with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool: + created = list(pool.map(create, range(count))) + + # Created on worker threads, closed on the main thread. + for res in created: + self.assertEqual(res._owner_pid, pid) + self.assertFalse(is_foreign_process(res)) + res.close() + + with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool: + made_on_main = [] + for index in range(count): + res = _ConcreteResource() + res._activate(0x20000 + index) + made_on_main.append(res) + # Created on the main thread, closed on worker threads. + list(pool.map(lambda r: r.close(), made_on_main)) + + expected = {0x10000 + i: 1 for i in range(count)} + expected.update({0x20000 + i: 1 for i in range(count)}) + # Restrict to this test's handles: resources dropped by other tests in + # the class can be collected at any point and land in self.freed. + counts = {handle: value + for handle, value in self._free_counts().items() + if handle in expected} + self.assertEqual(counts, expected) + + def test_third_thread_gc_of_dropped_reference_frees_exactly_once(self): + import gc + + def make_and_drop(index): + res = _ConcreteResource() + res._activate(0x30000 + index) + # Reference dies here; __del__ may run on this thread or later. + return index + + with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool: + list(pool.map(make_and_drop, range(200))) + + gc.collect() + + # Count only this test's handles + counts = {handle: count + for handle, count in self._free_counts().items() + if 0x30000 <= handle < 0x30000 + 200} + self.assertEqual(len(counts), 200, + "dropped resources were not all freed") + self.assertEqual(set(counts.values()), {1}, + "a dropped resource was freed more than once") + + def test_settings_relayed_across_threads_stays_usable(self): + ManagedResource._free_native_ptr = self._real_free + + manifest = { + "claim_generator": "threaded_stamp_test", + "format": "image/jpeg", + "assertions": [], + } + settings = Settings() + pid = os.getpid() + results = [] + errors = [] + + def build_context_and_builder(): + try: + context = Context(settings=settings) + builder = Builder(manifest, context=context) + results.append(( + builder._owner_pid, context._owner_pid, builder.is_valid)) + builder.close() + context.close() + except Exception as exc: + errors.append(exc) + + # Each thread owns the Settings for its turn. + for _ in range(8): + thread = threading.Thread(target=build_context_and_builder) + thread.start() + thread.join() + + settings.close() + + self.assertEqual(errors, []) + self.assertEqual(len(results), 8) + for builder_pid, context_pid, valid in results: + self.assertEqual(builder_pid, pid) + self.assertEqual(context_pid, pid) + self.assertTrue(valid) + self.assertEqual(settings._owner_pid, pid) + + if __name__ == '__main__': unittest.main()