diff --git a/docs/context-settings.md b/docs/context-settings.md index 35692a82..bd950fb9 100644 --- a/docs/context-settings.md +++ b/docs/context-settings.md @@ -752,6 +752,8 @@ ctx = Context() assert isinstance(ctx, ContextProvider) # True ``` +A provider that does not derive from `ManagedResource` runs without in-flight teardown protection. `Reader` and `Builder` check `is_valid` before use, but nothing defers a teardown that arrives mid-construction, so closing such a provider on another thread while a `Reader` or `Builder` is being built from it can free the native context while that construction is still using it. The built-in `Context` carries that protection. Custom providers that share a context across threads should keep it alive for the duration of any construction that uses it. + ## Migrating from load_settings The `load_settings()` function is deprecated. Replace it with `Settings` and `Context` APIs: diff --git a/docs/native-resources-management.md b/docs/native-resources-management.md index 17b01d92..9d4de693 100644 --- a/docs/native-resources-management.md +++ b/docs/native-resources-management.md @@ -142,7 +142,9 @@ Each callback checks `_initialized` and `_closed` before touching the underlying ### `Stream` cleanup -`Stream` holds `_close_lock`, a plain `Lock` rather than an `RLock`. It serializes the three cleanup paths: `close()`, `__del__`, and a `close()` on another thread. Without it, two of them reach the same stream and call `c2pa_release_stream` twice on one native handle. `Stream` needs its own lock since it does not inherit the `_op_lock` machinery. +`Stream` holds `_close_lock`, an `RLock`. It serializes the three cleanup paths: `close()`, `__del__`, and a `close()` on another thread. Without it, two of them reach the same stream and call `c2pa_release_stream` twice on one native handle. `Stream` needs its own lock since it does not inherit the `_op_lock` machinery. + +The lock is reentrant for the same reason `_op_lock` is. `close()` sets the four callback attributes to `None` inside the locked region, which can drop the last reference to an object whose finalizer runs at that bytecode boundary. `__del__` takes the same lock. A plain `Lock` deadlocks against itself when that finalizer belongs to the stream being closed. Cleanup runs in the direction of dependency: whatever can still invoke or reach the other is torn down first. Because callbacks run the opposite way for a `Stream`, its close order is the reverse of `ManagedResource`'s: @@ -159,6 +161,8 @@ Cleanup runs in the direction of dependency: whatever can still invoke or reach Both `close()` and `__del__` take the foreign-process branch, marking the stream closed without calling into the native library. [Fork safety](#fork-safety) covers why. +Both check `is_foreign_process()` before acquiring `_close_lock`, as `_teardown()` and `_lock()` do. A child inherits the lock in whatever state it had at `fork()`, and the thread holding it does not exist there to release it, so a child that acquired first would wait on it forever. + ### Reference cycles in the callbacks Each ctypes callback closes over the `Stream` it belongs to. Captured directly, that forms a cycle: the `Stream` holds the callback, the callback's closure holds the `Stream`. Nothing in that loop reaches a refcount of zero, so cleanup falls to the cycle collector. [Why `__del__` is not reliable enough](#why-__del__-is-not-reliable-enough) covers why that timing cannot be relied on. @@ -271,6 +275,8 @@ The check catches a borrow already in flight. The `CLOSED` mark catches one arri The mark is provisional. `_abort_consume()` restores the previous state when the native call turns out not to have taken the handle, which keeps the retained branch of the [ownership-taken triage](#why-an-ownership-taken-failure-does-not-free) handing back a usable object. +`_raise_consume_failure()` performs that restore, on the pre-consume branch only. The reservation is held until the branch is known. `_read_native_error()` is itself a native call and releases the GIL, so a resource restored to `ACTIVE` before the error is classified is visible as usable to another thread while the native side may already own its handle. + `_consume_and_swap()` is excluded. `_swap_handle()` requires the resource to stay `ACTIVE` and the object remains usable with its replacement pointer, so there is no `CLOSED` mark to make and no check. Its callers (`Reader.with_fragment`, `Builder.with_archive`) pass streams whose callbacks re-enter this API, so they hold their own `_native_call()`. `Reader.with_fragment()` additionally serializes itself with a lock of its own, described in [`Reader.with_fragment()`](#readerwith_fragment). ### Context lifetime during a context-sign @@ -350,7 +356,7 @@ Each transition has one method that performs it, and subclasses must go through | `_activate(handle)` | UNINITIALIZED to ACTIVE | Rejects a null handle, and refuses to run on an already-activated resource. A rejected activation leaves the object exactly as it was. | | `_swap_handle(new_handle)` | ACTIVE to ACTIVE | Requires the resource to already be active and the replacement to be non-null. Used when an FFI call consumed the old handle and returned a new one. | | `_teardown(free_handle=False)` | ACTIVE to CLOSED | Drops the handle without freeing it, for when ownership passed to the native side (e.g. `Signer` into `Context`). Runs `_release()` first, so subclass cleanup still happens. Unlike the other two it enforces no precondition on the current state: it closes whatever it is given. | -| `_release_handle()` | ACTIVE to CLOSED | Frees the handle (guarded, via `_teardown(free_handle=True)`) and closes the object. Same post-state as the consumed teardown. | +| `_release_handle()` | ACTIVE to CLOSED | Frees the handle (guarded, via `_teardown(free_handle=True)`) and closes the object. Same post-state as the consumed teardown. A resource that is already non-ACTIVE takes the other branch, which clears the handle without freeing it; the reserved consume paths call `_teardown()` directly for that reason. | Because activation is the only way in, no code path can leave an object ACTIVE while holding a null handle. @@ -522,16 +528,16 @@ sequenceDiagram S->>S: _teardown(free_handle=False) Note right of S: Consumed: native took the signer else non-zero status - S->>S: _abort_consume(): restore the previous state S->>S: _raise_consume_failure() reads the native error - Note right of S: The error is read before any free,
so a free's own error cannot overwrite it + Note right of S: The error is read before any free,
so a free's own error cannot overwrite it.
The reservation is held until the branch is known alt error carries a pre-consume tag + S->>S: _abort_consume(): restore the previous state Note right of S: Rejected before ownership moved:
Signer stays ACTIVE, typed error raised else any other error S->>S: _teardown(free_handle=False) Note right of S: Native took it, then failed and
dropped the value itself: free nothing else error slot empty - S->>S: _release_handle() guarded free + S->>S: _teardown(free_handle=True) guarded free Note right of S: Ownership unknown: a real free if still
ours, a -1 no-op if native took it end end @@ -595,7 +601,11 @@ Those two steps have to run as one unit. Two threads interleaving them can close `Reader._fragment_lock` covers both steps. It is an `RLock`, and `with_fragment()` is the only method that takes it. -Unlike `_op_lock`, this lock *is* held across the native call. The rule against that exists to stop a re-entering callback from blocking on a lock its own thread holds. Nothing on the callback path takes `_fragment_lock`, so a re-entering callback cannot block on it. +The guard spans the native call, which drives caller-supplied stream callbacks. `with_fragment()` therefore takes it with `acquire(blocking=False)` and releases it in a `finally`. A second thread finding it held is refused with `C2paError` rather than parked behind a native call that is waiting on a callback to return. A callback that starts a thread of its own and waits for it would otherwise deadlock: the new thread waits for the guard, and the call holding the guard waits for the callback. + +A refusal leaves the Reader untouched. No stream is built and no handle is consumed, so the call succeeds once the other thread returns. + +Reentrancy applies to the owning thread only. A callback that calls `with_fragment()` synchronously passes the guard and reaches the native call, which rejects the handle it has already consumed. The two locks nest in a [fixed order](#lock-ordering): `_fragment_lock` outside, then `_native_call()` and `_op_lock` inside it. @@ -650,7 +660,7 @@ The two failure paths are indistinguishable from the return value alone. Only th | --- | --- | --- | | One of `_PRE_CONSUME_ERROR_TAGS` | Still ours: rejected before ownership moved | Handle kept, resource stays `ACTIVE`, typed error raised. Normal cleanup frees it later. | | Any other error | Taken, then the operation failed | `_teardown(free_handle=False)`: the native side already dropped the value, so nothing is freed here. Resource goes `CLOSED`, error typed from the native message. | -| No error at all | Unknown | `_release_handle()` guarded free, the caller's message is raised with `"Unknown error"` filled in. | +| No error at all | Unknown | Guarded free, the caller's message is raised with `"Unknown error"` filled in. A reserved consume frees through `_teardown(free_handle=True)`, because `_release_handle()` treats a reserved resource as one it does not own. | This error and ownership triage relies on the native error still being readable (and correctly being the last error encountered) after the call returns. Reading an error copies the message out and frees the copy, but leaves the native slot set until the next error overwrites it. diff --git a/src/c2pa/c2pa.py b/src/c2pa/c2pa.py index 2a9b30ca..1f66226f 100644 --- a/src/c2pa/c2pa.py +++ b/src/c2pa/c2pa.py @@ -462,7 +462,8 @@ def _teardown(self, free_handle: bool): handle, self._handle = self._handle, None if free_handle and handle: try: - ManagedResource._free_native_ptr(handle) + # Subclasses may override the deallocator. + type(self)._free_native_ptr(handle) except Exception: logger.error("Failed to free native %s resources", type(self).__name__, exc_info=True) @@ -559,7 +560,7 @@ def _swap_handle(self, new_handle): "InvalidBufferSize:", ) - def _invoke_consume(self, ffi_call, error_message): + def _invoke_consume(self, ffi_call, error_message, *, reserved=False): """Run an FFI call that consumes this handle, returning its raw result. A marshalling ArgumentError is re-raised untouched (call never reached @@ -573,6 +574,9 @@ def _invoke_consume(self, ffi_call, error_message): result (a replacement pointer, a status code, ...). error_message: Format string with one placeholder, used to wrap a callback exception. + reserved: True when the caller reserved the handle with + _begin_consume(), which frees here rather than through + _release_handle(). Raises: ctypes.ArgumentError: If marshalling failed; handle untouched. @@ -585,10 +589,15 @@ def _invoke_consume(self, ffi_call, error_message): # is untouched and still ours. Re-raise as-is. raise except Exception as e: - self._release_handle() + if reserved: + # A reservation leaves the resource CLOSED with the handle set, + # which _release_handle() nulls without freeing. + self._teardown(free_handle=True) + else: + self._release_handle() raise C2paError(error_message.format(e)) from e - def _raise_consume_failure(self, error_message): + def _raise_consume_failure(self, error_message, previous_state=None): """Raise the error from an FFI handler consuming call. The native error is read before any free so a free's own @@ -603,9 +612,19 @@ def _raise_consume_failure(self, error_message): with another one and, because that substitute carries a pre-consume tag, invert the retain/consume decision made below. + A caller that reserved the handle with _begin_consume() passes + previous_state and stays reserved until this classification finishes. + _read_native_error() is a native call and releases the GIL, so a + resource restored to ACTIVE before the tags are examined is visible as + usable to another thread while native may already own its handle. Only + the pre-consume branch hands the resource back. + Args: error_message: Format string with one placeholder, used when the native layer offers no error of its own. + previous_state: Lifecycle state to restore if the handle turns out + to have been rejected before native took ownership. None when + the caller holds no reservation. Raises: C2paError: Always; typed by the native error when there is one. @@ -619,6 +638,8 @@ def _raise_consume_failure(self, error_message): "ownership (%s); handle retained", type(self).__name__, error) + if previous_state is not None: + self._abort_consume(previous_state) _raise_typed_c2pa_error(error) # A non-tag error means the native side took ownership then failed, @@ -629,7 +650,12 @@ def _raise_consume_failure(self, error_message): _raise_typed_c2pa_error(error) # No error in the slot: ownership is unknown, so free defensively. - self._release_handle() + # A reservation leaves the resource CLOSED with the handle set, + # which _release_handle() nulls without freeing. + if previous_state is not None: + self._teardown(free_handle=True) + else: + self._release_handle() raise C2paError(error_message.format("Unknown error")) def _begin_consume(self): @@ -673,7 +699,19 @@ def _consume_and_swap(self, ffi_call, error_message): """ new_ptr = self._invoke_consume(ffi_call, error_message) if new_ptr: - self._swap_handle(new_ptr) + try: + self._swap_handle(new_ptr) + except Exception: + # _swap_handle refuses a resource a concurrent close() left + # CLOSED. Native consumed the old pointer and returned this + # one, so nothing else holds it. + try: + ManagedResource._free_native_ptr(new_ptr) + except Exception: + logger.error( + "Failed to free the replacement %s handle", + type(self).__name__, exc_info=True) + raise return self._raise_consume_failure(error_message) @@ -685,15 +723,15 @@ def _consume_no_replacement(self, ffi_call, error_message): """ previous_state = self._begin_consume() try: - result = self._invoke_consume(ffi_call, error_message) + result = self._invoke_consume( + ffi_call, error_message, reserved=True) except Exception: self._abort_consume(previous_state) raise if result == 0: self._teardown(free_handle=False) return - self._abort_consume(previous_state) - self._raise_consume_failure(error_message) + self._raise_consume_failure(error_message, previous_state) def _consume_into(self, ffi_call, error_message): """Run an FFI call that consumes this handle and returns a *different* @@ -703,15 +741,15 @@ def _consume_into(self, ffi_call, error_message): """ previous_state = self._begin_consume() try: - result = self._invoke_consume(ffi_call, error_message) + result = self._invoke_consume( + ffi_call, error_message, reserved=True) except Exception: self._abort_consume(previous_state) raise if result: self._teardown(free_handle=False) return result - self._abort_consume(previous_state) - self._raise_consume_failure(error_message) + self._raise_consume_failure(error_message, previous_state) @classmethod def _wrap_native_handle(cls, handle): @@ -1644,11 +1682,35 @@ def load_settings(settings: Union[str, dict], format: str = "json") -> None: check=lambda r: r != 0) +@contextlib.contextmanager +def _context_guard(context): + """Hold a caller-supplied context valid across a native call. + + ContextProvider requires only is_valid and execution_context. + A provider that also manages a native handle, + such as the built-in Context, offers _native_call, + which counts the call in flight so a concurrent close() records + its intent and defers the free until the call returns. A provider + implementing just the two required properties runs without that guard. + """ + native_call = getattr(context, "_native_call", None) + if native_call is None: + yield + return + with native_call(): + yield + + class ContextProvider(ABC): """Abstract base class for types that provide a C2PA context. Subclass to implement a custom context provider. The built-in Context class is the standard implementation. + + A provider that does not derive from ManagedResource is used without + in-flight teardown protection: closing it on another thread while a Reader + or Builder is being constructed from it can free the native context while + that construction is still using it. """ @property @@ -2004,7 +2066,7 @@ def __init__(self, file_like_stream): self._initialized = False self._stream = None # Serializes close() and __del__ against a concurrent double-free. - self._close_lock = threading.Lock() + self._close_lock = threading.RLock() # Generate unique stream ID using object ID and counter stream_counter = next(Stream._stream_id_counter) @@ -2254,14 +2316,18 @@ def close(self): Errors during cleanup are logged but not raised to ensure cleanup. Multiple calls to close() are handled gracefully. """ + # Checked before the lock, as _lock() and __del__ do: + # a child inherits _close_lock in whatever state it had at fork(), + # and the thread holding it does not exist there to release it. + if is_foreign_process(self): + self._closed = True + self._initialized = False + return + # Serializes against __del__ / a concurrent close(). with self._close_lock: if self._closed: return - if is_foreign_process(self): - self._closed = True - self._initialized = False - return try: # Clean up stream first as it depends on callbacks @@ -2818,7 +2884,7 @@ def _init_from_context(self, context, format_or_path, try: # The Context is caller-supplied and may be shared, so its handle # needs its own in-flight guard across the native call. - with context._native_call(): + with _context_guard(context): # Adopt before the consuming call: _consume_and_swap needs an # active resource, and cleanup then owns the pointer either # way. @@ -2965,6 +3031,9 @@ def with_fragment(self, format: Optional[str], stream, underlying object, in which case this Reader is closed and cannot be retried: create a new one instead of reusing this instance. + C2paError: If another thread is inside this method on the same + Reader. This one leaves the Reader untouched, so the call can + be retried once that thread returns. """ format_arg = _format_ffi_arg(_encode_format(format, "Reader")) @@ -2974,7 +3043,18 @@ def with_fragment(self, format: Optional[str], stream, raise C2paError(f"{type(self).__name__} is closed") # The native call and the ownership transfer are one unit. - with self._fragment_lock: + # Taken without blocking because the call drives caller-supplied stream + # callbacks: a second thread, including one a callback starts, would + # otherwise wait here for a native call that is itself waiting on that + # callback to return. + # + # Reentrant, so the thread already inside this region passes through + # and re-enters the native call, which rejects the handle it consumed. + if not self._fragment_lock.acquire(blocking=False): + raise C2paError( + f"{type(self).__name__} is already processing a fragment " + f"on another thread") + try: # The native reader keeps reading through both streams. main_obj = Stream(stream) frag_obj = Stream(fragment_stream) @@ -3028,6 +3108,8 @@ def with_fragment(self, format: Optional[str], stream, # and a reader must never be served them. self._manifest_json_str_cache = None self._manifest_data_cache = None + finally: + self._fragment_lock.release() return self @@ -3670,7 +3752,7 @@ def _init_from_context(self, context, json_str): # The Context is caller-supplied and may be shared, # so its handle needs its own in-flight guard across # the native call, especially for state checks. - with context._native_call(): + with _context_guard(context): # Adopt before the consuming call. self._create_and_activate( lambda: _lib.c2pa_builder_from_context( @@ -4039,7 +4121,7 @@ def _sign_internal( # Entered inside self's guard, matching the Builder to # Signer order, so the two acquisitions are always # taken in one direction. - with self._context._native_call(): + with _context_guard(self._context): result = _lib.c2pa_builder_sign_context( self._handle, format_arg, diff --git a/tests/test_unit_tests.py b/tests/test_unit_tests.py index e7af0c05..2c1fb42e 100644 --- a/tests/test_unit_tests.py +++ b/tests/test_unit_tests.py @@ -9677,6 +9677,200 @@ def test_ed25519_sign_with_empty_data_raises(self): c2pa_module.ed25519_sign(b"", "not a key") +class TestConsumeOwnership(unittest.TestCase): + """Ownership of the native handle across the consuming call paths.""" + + def setUp(self): + self.freed = [] + self._real_free = ManagedResource._free_native_ptr + + def counting_free(ptr): + self.freed.append(ptr) + return self._real_free(ptr) + + ManagedResource._free_native_ptr = staticmethod(counting_free) + + def tearDown(self): + ManagedResource._free_native_ptr = staticmethod(self._real_free) + + def test_generic_exception_frees_the_reserved_handle(self): + """A reserved consume that raises must free, not drop, the handle. + + _begin_consume() leaves the resource CLOSED with the handle still set, + which _release_handle() reads as "not ours" and nulls without freeing, + while _abort_consume() can no longer restore it. + """ + def boom(handle): + raise RuntimeError("callback failed after the reservation") + + for name in ("_consume_no_replacement", "_consume_into"): + with self.subTest(helper=name): + resource = Settings() + self.freed.clear() + + with self.assertRaises(Error): + getattr(resource, name)(boom, "consume failed: {}") + + self.assertEqual( + len(self.freed), 1, + "{} dropped the handle without freeing it".format(name)) + self.assertIsNone(resource._handle) + + def test_marshalling_error_retains_the_handle(self): + """Positive control for the free counter. + + An ArgumentError means the call never reached native, so the handle is + untouched and must NOT be freed. Without this, a zero-free assertion + could pass simply because the counter never fires. + """ + def bad_marshal(handle): + raise ctypes.ArgumentError("marshalling failed") + + resource = Settings() + self.freed.clear() + + with self.assertRaises(ctypes.ArgumentError): + resource._consume_no_replacement(bad_marshal, "consume: {}") + + self.assertEqual(self.freed, []) + self.assertIsNotNone(resource._handle) + self.assertEqual(resource._lifecycle_state, LifecycleState.ACTIVE) + + def test_pre_consume_rejection_restores_the_resource(self): + """A handle native rejected before taking ownership stays usable. + + The reservation is held until _raise_consume_failure classifies the + error, so no other thread sees the resource as ACTIVE while its + ownership is still undetermined. + """ + resource = Settings() + self.freed.clear() + real_read = c2pa_module._read_native_error + c2pa_module._read_native_error = ( + lambda: "Other: UntrackedPointer: 0x1234") + try: + with self.assertRaises(Error): + resource._consume_no_replacement(lambda h: 1, "consume: {}") + finally: + c2pa_module._read_native_error = real_read + + self.assertEqual(resource._lifecycle_state, LifecycleState.ACTIVE) + self.assertIsNotNone(resource._handle) + self.assertEqual(self.freed, []) + + def test_post_consume_failure_keeps_the_resource_closed(self): + """An error without a pre-consume tag means native took ownership. + + The value is native's to drop, so the resource stays closed and frees + nothing. + """ + resource = Settings() + self.freed.clear() + real_read = c2pa_module._read_native_error + c2pa_module._read_native_error = lambda: "Other: operation failed" + try: + with self.assertRaises(Error): + resource._consume_no_replacement(lambda h: 1, "consume: {}") + finally: + c2pa_module._read_native_error = real_read + + self.assertEqual(resource._lifecycle_state, LifecycleState.CLOSED) + self.assertEqual(self.freed, []) + + def test_failure_without_a_native_error_frees_the_handle(self): + """An empty error slot leaves ownership unknown, so the handle is + freed defensively rather than dropped. + """ + resource = Settings() + self.freed.clear() + real_read = c2pa_module._read_native_error + c2pa_module._read_native_error = lambda: None + try: + with self.assertRaises(Error): + resource._consume_no_replacement(lambda h: 1, "consume: {}") + finally: + c2pa_module._read_native_error = real_read + + self.assertEqual( + len(self.freed), 1, + "an unknown-ownership failure dropped the handle without freeing") + self.assertIsNone(resource._handle) + + def test_rejected_replacement_is_freed(self): + """A replacement _swap_handle refuses must not be left unowned. + + Native consumed the old pointer and returned this one, so nothing else + holds it. + """ + resource = Settings() + spare = Settings() + replacement = spare._handle + # Detach so only the code under test can free it. + spare._handle = None + spare._lifecycle_state = LifecycleState.CLOSED + self.freed.clear() + + # A close() arriving mid-call leaves the resource CLOSED. + resource._lifecycle_state = LifecycleState.CLOSED + + with self.assertRaises(Error): + resource._consume_and_swap(lambda h: replacement, "swap: {}") + + self.assertIn(replacement, self.freed) + + +class TestContextProviderContract(unittest.TestCase): + """The published ContextProvider contract is is_valid plus + execution_context, and nothing more. + """ + + class _MinimalProvider(ContextProvider): + """Implements exactly what the abstract base class declares.""" + + def __init__(self): + self._inner = Context(Settings()) + + @property + def is_valid(self): + return self._inner.is_valid + + @property + def execution_context(self): + return self._inner.execution_context + + def test_reader_accepts_a_minimal_provider(self): + provider = self._MinimalProvider() + try: + Reader("image/jpeg", io.BytesIO(b"not a real jpeg"), + context=provider) + except AttributeError as e: + self.fail("Reader requires more than the documented " + "ContextProvider contract: {}".format(e)) + except Error: + # Rejecting the bytes is the native library doing its job. + pass + + def test_builder_accepts_a_minimal_provider(self): + provider = self._MinimalProvider() + try: + Builder({"claim_generator": "test"}, context=provider) + except AttributeError as e: + self.fail("Builder requires more than the documented " + "ContextProvider contract: {}".format(e)) + + def test_built_in_context_still_gets_in_flight_protection(self): + """The compatibility shim must not silently drop the guard for the + provider that does implement it. + """ + context = Context(Settings()) + self.assertEqual(context._inflight, 0) + with c2pa_module._context_guard(context): + self.assertGreater( + context._inflight, 0, + "built-in Context lost its in-flight guard") + self.assertEqual(context._inflight, 0) + + class TestLockOrderStaticAnalysis(unittest.TestCase): """Static analysis over the source, not runtime behavior: no threads are spawned here. @@ -9735,6 +9929,33 @@ def lock_name_for_with(item): return "_op_lock" return None + def lock_name_for_acquire(node): + """A lock taken with acquire() and released in a finally nests just + as a `with` does, so the scan has to follow it or it silently stops + seeing whole regions. + """ + call = node.value if isinstance(node, ast.Expr) else node + if isinstance(call, ast.UnaryOp) and isinstance(call.op, ast.Not): + call = call.operand + if not (isinstance(call, ast.Call) + and isinstance(call.func, ast.Attribute) + and call.func.attr == "acquire"): + return None + owner = call.func.value + if (isinstance(owner, ast.Attribute) + and isinstance(owner.value, ast.Name) + and owner.value.id == "self" + and any(owner.attr in attrs + for attrs in lock_attrs_by_class.values())): + return owner.attr + return None + + def acquires_in_test(node): + """Lock taken by `if not self._X.acquire(...)`-style guards.""" + if isinstance(node, ast.If): + return lock_name_for_acquire(node.test) + return None + def orders_in(node, stack, pairs): """Record (outer, inner) for every nesting this node contains.""" if isinstance(node, (ast.With, ast.AsyncWith)): @@ -9749,8 +9970,29 @@ def orders_in(node, stack, pairs): for _ in names: stack.pop() return + # A statement list can open a lock partway through via acquire(); + # everything after it in that list is nested inside. + for field, value in ast.iter_fields(node): + if not isinstance(value, list): + continue + held = [] + for child in value: + if not isinstance(child, ast.stmt): + continue + name = (lock_name_for_acquire(child) + or acquires_in_test(child)) + if name: + if stack: + pairs.add((stack[-1], name)) + stack.append(name) + held.append(name) + continue + orders_in(child, stack, pairs) + for _ in held: + stack.pop() for child in ast.iter_child_nodes(node): - orders_in(child, stack, pairs) + if not isinstance(child, ast.stmt): + orders_in(child, stack, pairs) pairs_by_method = {} for cls in ast.walk(tree): diff --git a/tests/test_unit_tests_threaded.py b/tests/test_unit_tests_threaded.py index ffff10c5..20537e46 100644 --- a/tests/test_unit_tests_threaded.py +++ b/tests/test_unit_tests_threaded.py @@ -27,6 +27,7 @@ import threading import concurrent.futures import time +import signal import asyncio import random from unittest.mock import MagicMock, patch @@ -34,6 +35,7 @@ from c2pa import Builder, C2paError as Error, Reader, C2paSigningAlg as SigningAlg, C2paSignerInfo, Signer, sdk_version # noqa: E501 from c2pa import Context, Settings from c2pa.c2pa import ManagedResource, Stream, LifecycleState +import c2pa.c2pa as c2pa_module from c2pa.lib import is_foreign_process, record_owner_pid PROJECT_PATH = os.getcwd() @@ -573,19 +575,37 @@ def gated_native_call(): reader._native_call = gated_native_call class ContentionReportingLock: - """Flags when a caller has to wait for the lock it wraps.""" + """Flags when a caller finds the lock it wraps already held. + + with_fragment takes this lock with acquire(blocking=False) and + releases it in a finally, so those are the methods wrapped here. + """ def __init__(self, inner): self._inner = inner - def __enter__(self): + def acquire(self, blocking=True, timeout=-1): + if not blocking: + acquired = self._inner.acquire(blocking=False) + if not acquired: + # The second caller is refused rather than parked, + # which is the mutual exclusion this test checks for. + contended.set() + return acquired if not self._inner.acquire(blocking=False): contended.set() - self._inner.acquire() + return self._inner.acquire(blocking, timeout) + return True + + def release(self): + self._inner.release() + + def __enter__(self): + self.acquire() return self def __exit__(self, exc_type, exc_val, exc_tb): - self._inner.release() + self.release() return False real_fragment_lock = reader._fragment_lock @@ -3393,6 +3413,296 @@ def thread_work(thread_id): self.assertNotEqual(current_manifest["active_manifest"], thread_manifest_data[other_thread_id]["active_manifest"]) +class TestWithFragmentReentrancy(unittest.TestCase): + """with_fragment drives caller-supplied stream callbacks, so it must not + hold a lock a callback-spawned thread would wait on. + """ + + def test_reentrant_call_is_refused_rather_than_blocked(self): + init_path = os.path.join(FIXTURES_FOLDER, "dashinit.mp4") + fragment_path = os.path.join(FIXTURES_FOLDER, "dash1.m4s") + with open(init_path, "rb") as handle: + init_bytes = handle.read() + with open(fragment_path, "rb") as handle: + fragment_bytes = handle.read() + + reader = Reader("video/mp4", io.BytesIO(init_bytes)) + state = {"fired": False, "result": None, "hung": None} + + class ReentrantStream(io.BytesIO): + """Re-enters the API from another thread, from inside a callback, + and waits for it: the shape that deadlocks a lock held across the + native call. + """ + + def _reenter_once(self): + if state["fired"]: + return + state["fired"] = True + + def second_call(): + try: + reader.with_fragment( + "video/mp4", + io.BytesIO(init_bytes), + io.BytesIO(fragment_bytes)) + state["result"] = "completed" + except Error as e: + state["result"] = e + + thread = threading.Thread(target=second_call, daemon=True) + thread.start() + thread.join(10) + state["hung"] = thread.is_alive() + + def read(self, size=-1): + self._reenter_once() + return super().read(size) + + def seek(self, offset, whence=0): + self._reenter_once() + return super().seek(offset, whence) + + reader.with_fragment("video/mp4", + ReentrantStream(init_bytes), + io.BytesIO(fragment_bytes)) + + self.assertTrue(state["fired"], "the callback never re-entered") + self.assertFalse( + state["hung"], + "a with_fragment call started from a stream callback blocked on " + "the lock the running call holds") + self.assertIsInstance( + state["result"], Error, + "the re-entrant call must be refused, not silently interleaved") + + def test_same_thread_reentry_does_not_corrupt_the_reader(self): + """_fragment_lock is reentrant, so a callback calling with_fragment + synchronously passes the guard. The native layer rejects the handle it + already consumed, and the Reader survives. + """ + init_path = os.path.join(FIXTURES_FOLDER, "dashinit.mp4") + fragment_path = os.path.join(FIXTURES_FOLDER, "dash1.m4s") + with open(init_path, "rb") as handle: + init_bytes = handle.read() + with open(fragment_path, "rb") as handle: + fragment_bytes = handle.read() + + reader = Reader("video/mp4", io.BytesIO(init_bytes)) + state = {"fired": False, "inner": None} + + class SelfReentrantStream(io.BytesIO): + def _reenter_once(self): + if state["fired"]: + return + state["fired"] = True + try: + reader.with_fragment("video/mp4", + io.BytesIO(init_bytes), + io.BytesIO(fragment_bytes)) + state["inner"] = "completed" + except Error as e: + state["inner"] = e + + def read(self, size=-1): + self._reenter_once() + return super().read(size) + + def seek(self, offset, whence=0): + self._reenter_once() + return super().seek(offset, whence) + + reader.with_fragment("video/mp4", + SelfReentrantStream(init_bytes), + io.BytesIO(fragment_bytes)) + + self.assertTrue(state["fired"], "the callback never re-entered") + self.assertIsInstance( + state["inner"], Error, + "a nested consume on the same handle must be rejected") + # The outer call still owns a live handle. + self.assertTrue(reader.is_valid) + self.assertIsInstance(reader.json(), str) + + def test_refused_call_leaves_the_reader_usable(self): + """The refusal reports contention without touching the Reader, so the + caller can retry once the other thread returns. + """ + init_path = os.path.join(FIXTURES_FOLDER, "dashinit.mp4") + fragment_path = os.path.join(FIXTURES_FOLDER, "dash1.m4s") + with open(init_path, "rb") as handle: + init_bytes = handle.read() + with open(fragment_path, "rb") as handle: + fragment_bytes = handle.read() + + reader = Reader("video/mp4", io.BytesIO(init_bytes)) + + holding = threading.Event() + release = threading.Event() + + def hold_the_guard(): + reader._fragment_lock.acquire() + holding.set() + release.wait(10) + reader._fragment_lock.release() + + holder = threading.Thread(target=hold_the_guard, daemon=True) + holder.start() + self.assertTrue(holding.wait(5), "the guard was never taken") + + with self.assertRaises(Error): + reader.with_fragment("video/mp4", + io.BytesIO(init_bytes), + io.BytesIO(fragment_bytes)) + + # Refused before any stream was built or handle consumed. + self.assertTrue(reader.is_valid) + + release.set() + holder.join(5) + + # The same call succeeds once the other thread is out. + reader.with_fragment("video/mp4", + io.BytesIO(init_bytes), + io.BytesIO(fragment_bytes)) + self.assertTrue(reader.is_valid) + + +class TestStreamCloseReentrancy(unittest.TestCase): + """close() clears the callback references inside _close_lock, which can run + a finalizer at that bytecode boundary, and __del__ takes the same lock. + """ + + def test_close_can_be_reentered_on_the_same_thread(self): + stream = Stream(io.BytesIO(b"payload")) + finished = threading.Event() + + def hold_then_reenter(): + with stream._close_lock: + # A finalizer running here re-takes the lock this thread holds. + stream.close() + finished.set() + + worker = threading.Thread(target=hold_then_reenter, daemon=True) + worker.start() + + self.assertTrue( + finished.wait(10), + "close() blocked re-entering _close_lock from the thread that " + "already holds it") + self.assertTrue(stream._closed) + + +@unittest.skipUnless(hasattr(os, "fork"), "requires fork()") +class TestStreamCloseAfterFork(unittest.TestCase): + """A forked child must not wait on a lock no surviving thread will + release. + """ + + def test_close_in_child_does_not_block_on_an_inherited_lock(self): + stream = Stream(io.BytesIO(b"payload")) + + holding = threading.Event() + release = threading.Event() + + def hold_the_lock(): + with stream._close_lock: + holding.set() + release.wait(30) + + holder = threading.Thread(target=hold_the_lock, daemon=True) + holder.start() + self.assertTrue(holding.wait(5), "lock was never taken") + + # The child inherits _close_lock held by a thread that does not exist + # there, so close() has to take the foreign-process path without + # acquiring it. + pid = os.fork() + if pid == 0: + try: + stream.close() + # Exit 3 rather than 0 if close() returned without marking the + # stream closed, so a silent no-op cannot pass as success. + marked = stream._closed and not stream._initialized + os._exit(0 if marked else 3) + except BaseException: + os._exit(2) + + deadline = time.time() + 15 + status = None + while time.time() < deadline: + done, wait_status = os.waitpid(pid, os.WNOHANG) + if done: + status = wait_status + break + time.sleep(0.05) + + if status is None: + os.kill(pid, signal.SIGKILL) + os.waitpid(pid, 0) + release.set() + holder.join(5) + self.fail("close() in the forked child blocked on the inherited " + "lock instead of taking the foreign-process path") + + release.set() + holder.join(5) + self.assertEqual( + os.WEXITSTATUS(status), 0, + "close() in the forked child raised (2) or returned without " + "closing the stream (3)") + + +class TestConsumeReservationWindow(unittest.TestCase): + """The consume reservation must outlast ownership classification. + + _read_native_error() is a native call that releases the GIL, so a resource + restored to ACTIVE before the error is classified is visible as usable to + another thread while native may already own its handle. + """ + + def test_no_thread_sees_a_consumed_handle_as_valid(self): + resource = Settings() + + reading = threading.Event() + may_finish = threading.Event() + seen_valid = [] + + real_read = c2pa_module._read_native_error + + def gated_read(): + # Stand in for the GIL release inside the real native call. + reading.set() + may_finish.wait(10) + # No pre-consume tag: native took ownership and then failed. + return "Other: operation failed after taking ownership" + + def observer(): + if not reading.wait(10): + return + # The consuming call is mid-classification right now. + seen_valid.append(resource.is_valid) + may_finish.set() + + watcher = threading.Thread(target=observer, daemon=True) + watcher.start() + + c2pa_module._read_native_error = gated_read + try: + with self.assertRaises(Error): + resource._consume_no_replacement(lambda h: 1, "consume: {}") + finally: + c2pa_module._read_native_error = real_read + may_finish.set() + watcher.join(10) + + self.assertTrue(seen_valid, "observer never sampled the resource") + self.assertFalse( + seen_valid[0], + "another thread saw a resource whose handle native may already " + "own as valid") + + class TestLocking(unittest.TestCase): """Tests for the locks that guard native resources: - the per-object operation lock that serializes native calls against teardown, @@ -4489,15 +4799,25 @@ def test_every_borrowed_handle_is_guarded(self): handle_attrs = {"_handle", "execution_context"} def guarded_names(node): - """Names X with an active `with X._native_call():` at this node.""" + """Names X guarded at this node, by either form: + `with X._native_call():`, or `with _context_guard(X):` for a + caller-supplied ContextProvider, which enters X._native_call() + when X offers it. + """ found = set() for item in getattr(node, "items", []): call = item.context_expr - if (isinstance(call, ast.Call) - and isinstance(call.func, ast.Attribute) + if not isinstance(call, ast.Call): + continue + if (isinstance(call.func, ast.Attribute) and call.func.attr == "_native_call" and isinstance(call.func.value, ast.Name)): found.add(call.func.value.id) + elif (isinstance(call.func, ast.Name) + and call.func.id == "_context_guard" + and call.args + and isinstance(call.args[0], ast.Name)): + found.add(call.args[0].id) return found def borrowed_in_call(call):