diff --git a/src/c2pa/c2pa.py b/src/c2pa/c2pa.py index 0255e6db..6e0fe836 100644 --- a/src/c2pa/c2pa.py +++ b/src/c2pa/c2pa.py @@ -13,6 +13,7 @@ # Version: 0.37.8 +import contextlib import ctypes import enum import json @@ -266,6 +267,8 @@ def __init__(self): self._lifecycle_state = LifecycleState.UNINITIALIZED self._handle = None self._op_lock = threading.RLock() + self._inflight = 0 + self._pending_teardown = None record_owner_pid(self) def _lock(self): @@ -296,6 +299,39 @@ def _lock(self): pass return lock + @contextlib.contextmanager + def _native_call(self): + """Hold the handle valid across a native call that goes back + and forth to native layers. + + Calls that pass a Stream to the native library run caller-supplied + callbacks, so the lock cannot be held across them: the callback may + re-enter this API on another thread and deadlock. Instead the call is + counted as in flight, and a teardown arriving meanwhile records its + intent rather than freeing. The last caller out performs the free. + + The resource is marked closed as soon as the teardown is recorded, so + a caller that closed it cannot keep using it while the free is + pending. + """ + with self._lock(): + self._ensure_valid_state() + self._inflight = getattr(self, '_inflight', 0) + 1 + try: + yield + finally: + with self._lock(): + self._inflight -= 1 + pending = (self._pending_teardown + if self._inflight == 0 else None) + if pending is not None: + self._pending_teardown = None + # Released the lock before the free: _teardown takes it again, + # and keeping the two acquisitions separate means the counter + # update is never held across the release work. + if pending is not None: + self._teardown(pending) + @staticmethod def _free_native_ptr(ptr): """Free a native pointer by passing it to c2pa_free. @@ -356,6 +392,16 @@ def _teardown(self, free_handle: bool): thread's state check and its use of the handle in a native call. """ with self._lock(): + if getattr(self, '_inflight', 0) > 0: + # A native call is running that re-enters caller Python and + # is still using this handle. Record the intent; whichever + # caller leaves _native_call last performs the free. Mark the + # resource closed now so it cannot be used while the free is + # pending. + self._pending_teardown = free_handle + self._lifecycle_state = LifecycleState.CLOSED + return + if is_foreign_process(self): self._handle = None self._lifecycle_state = LifecycleState.CLOSED @@ -1735,13 +1781,22 @@ def __init__( check=lambda r: r != 0) if signer is not None: - signer._ensure_valid_state() - # A rejected signer is retained, not closed and leaked. - self._signer_callback_cb = signer._callback_cb - signer._consume_no_replacement( - lambda h: _lib.c2pa_context_builder_set_signer( - nb._handle, h), - "Failed to set signer on Context: {}") + # The signer's own in-flight guard: this hands its handle + # to native, so a signer.close() on another thread must + # not free it between the state check and the call. The + # guard also makes the check and the consume atomic. + # + # _consume_no_replacement tears the signer down from + # inside this region. A teardown recorded while the guard + # is held is deferred and performed as the guard unwinds, + # which is still before __init__ returns. + with signer._native_call(): + # A rejected signer is retained, not closed and leaked. + self._signer_callback_cb = signer._callback_cb + signer._consume_no_replacement( + lambda h: _lib.c2pa_context_builder_set_signer( + nb._handle, h), + "Failed to set signer on Context: {}") self._has_signer = True context_ptr = nb._consume_into( @@ -2658,12 +2713,19 @@ def _init_from_context(self, context, format_or_path, self._own_stream = Stream(stream) try: - # Adopt before the consuming call: _consume_and_swap needs an - # active resource, and cleanup then owns the pointer either way. - self._create_and_activate( - lambda: _lib.c2pa_reader_from_context( - context.execution_context), - Reader._ERROR_MESSAGES['reader_error']) + # The Context is caller-supplied and may be shared, so its handle + # needs its own in-flight guard across the native call: the + # execution_context property validates and returns the handle, and + # without the guard a context.close() on another thread could free + # it before c2pa_reader_from_context reads it. + with context._native_call(): + # Adopt before the consuming call: _consume_and_swap needs an + # active resource, and cleanup then owns the pointer either + # way. + self._create_and_activate( + lambda: _lib.c2pa_reader_from_context( + context.execution_context), + Reader._ERROR_MESSAGES['reader_error']) if manifest_data is not None: manifest_array = ( @@ -2702,6 +2764,10 @@ def _init_attrs(self): # Tracks a file we opened ourselves and must close later. self._backing_file = None + # Fragment streams handed to the native reader by with_fragment, + # which it keeps reading from for the rest of its life. + self._fragment_streams = [] + # Caches for manifest JSON string and parsed data. # These are invalidated when with_fragment() is called. self._manifest_json_str_cache = None @@ -2725,6 +2791,12 @@ def _close_streams(self): logger.warning("Failed to close Reader backing file") finally: self._backing_file = None + for fragment in getattr(self, '_fragment_streams', []): + try: + fragment.close() + except Exception: + logger.warning("Failed to close Reader fragment stream") + self._fragment_streams = [] def _release(self): """Release Reader-specific resources (caches, stream, backing file). @@ -2787,19 +2859,38 @@ def with_fragment(self, format: Optional[str], stream, cannot be retried: create a new one instead of reusing this instance. """ - self._ensure_valid_state() - format_arg = _format_ffi_arg(_encode_format(format, "Reader")) - with Stream(stream) as main_obj, Stream(fragment_stream) as frag_obj: - self._consume_and_swap( - lambda handle: _lib.c2pa_reader_with_fragment( - handle, - format_arg, - main_obj._stream, - frag_obj._stream, - ), - Reader._ERROR_MESSAGES['fragment_error']) + # The native reader keeps reading through both streams after this + # returns, so they are owned here and released by _release() rather + # than at the end of a with block. + main_obj = Stream(stream) + frag_obj = Stream(fragment_stream) + try: + with self._native_call(): + self._consume_and_swap( + lambda handle: _lib.c2pa_reader_with_fragment( + handle, + format_arg, + main_obj._stream, + frag_obj._stream, + ), + Reader._ERROR_MESSAGES['fragment_error']) + except Exception: + main_obj.close() + frag_obj.close() + raise + + # Replace the streams this reader owned, closing the previous ones so + # repeated calls do not accumulate them. + previous = self._own_stream + self._own_stream = main_obj + self._fragment_streams.append(frag_obj) + if previous is not None and previous is not main_obj: + try: + previous.close() + except Exception: + logger.warning("Failed to close previous Reader stream") # Invalidate caches: processing a new BMFF fragment updates the native # reader's state, which can change the manifest data it returns. @@ -3000,10 +3091,8 @@ def resource_to_stream(self, uri: str, stream: Any) -> int: Raises: C2paError: If there was an error writing the resource to stream """ - self._ensure_valid_state() - uri_str = uri.encode('utf-8') - with Stream(stream) as stream_obj: + with self._native_call(), Stream(stream) as stream_obj: result = _lib.c2pa_reader_resource_to_stream( self._handle, uri_str, stream_obj._stream) @@ -3451,11 +3540,17 @@ def _init_from_context(self, context, json_str): if not context.is_valid: raise C2paError("Context is not valid") - # Adopt before the consuming call: _consume_and_swap needs an - # active resource, and cleanup then owns the pointer either way. - self._create_and_activate( - lambda: _lib.c2pa_builder_from_context(context.execution_context), - Builder._ERROR_MESSAGES['builder_error']) + # The Context is caller-supplied and may be shared, so its handle + # needs its own in-flight guard across the native call: without it a + # context.close() on another thread frees the handle between the + # is_valid check and c2pa_builder_from_context reading it. + with context._native_call(): + # Adopt before the consuming call: _consume_and_swap needs an + # active resource, and cleanup then owns the pointer either way. + self._create_and_activate( + lambda: _lib.c2pa_builder_from_context( + context.execution_context), + Builder._ERROR_MESSAGES['builder_error']) self._consume_and_swap( lambda handle: _lib.c2pa_builder_with_definition( @@ -3558,10 +3653,8 @@ def add_resource(self, uri: str, stream: Any): Raises: C2paError: If there was an error adding the resource """ - self._ensure_valid_state() - uri_bytes = _to_utf8_bytes(uri, "resource URI") - with Stream(stream) as stream_obj: + with self._native_call(), Stream(stream) as stream_obj: result = _lib.c2pa_builder_add_resource( self._handle, uri_bytes, stream_obj._stream) @@ -3622,7 +3715,7 @@ def add_ingredient_from_stream( ingredient_str = _to_utf8_bytes(ingredient_json, "ingredient JSON") format_str = _to_utf8_bytes(format, "ingredient format") - with Stream(source) as source_stream: + with self._native_call(), Stream(source) as source_stream: result = ( _lib.c2pa_builder_add_ingredient_from_stream( self._handle, @@ -3671,9 +3764,7 @@ def to_archive(self, stream: Any) -> None: Raises: C2paError: If there was an error writing the archive """ - self._ensure_valid_state() - - with Stream(stream) as stream_obj: + with self._native_call(), Stream(stream) as stream_obj: result = _lib.c2pa_builder_to_archive( self._handle, stream_obj._stream) @@ -3698,7 +3789,7 @@ def write_ingredient_archive(self, ingredient_id: str, stream: Any) -> None: ingredient_id_str = _to_utf8_bytes(ingredient_id, "ingredient_id") - with Stream(stream) as stream_obj: + with self._native_call(), Stream(stream) as stream_obj: result = _lib.c2pa_builder_write_ingredient_archive( self._handle, ingredient_id_str, stream_obj._stream) @@ -3718,9 +3809,7 @@ def add_ingredient_from_archive(self, stream: Any) -> None: Raises: C2paError: If there was an error reading the archive """ - self._ensure_valid_state() - - with Stream(stream) as stream_obj: + with self._native_call(), Stream(stream) as stream_obj: result = _lib.c2pa_builder_add_ingredient_from_archive( self._handle, stream_obj._stream) @@ -3749,7 +3838,7 @@ def with_archive(self, stream: Any) -> 'Builder': """ self._ensure_valid_state() - with Stream(stream) as stream_obj: + with self._native_call(), Stream(stream) as stream_obj: self._consume_and_swap( lambda handle: _lib.c2pa_builder_with_archive( handle, stream_obj._stream), @@ -3797,23 +3886,36 @@ def _sign_internal( manifest_bytes_ptr = ctypes.POINTER(ctypes.c_ubyte)() try: - if signer is not None: - result = _lib.c2pa_builder_sign( - self._handle, - format_arg, - source_stream._stream, - dest_stream._stream, - signer._handle, - ctypes.byref(manifest_bytes_ptr) - ) - else: - result = _lib.c2pa_builder_sign_context( - self._handle, - format_arg, - source_stream._stream, - dest_stream._stream, - ctypes.byref(manifest_bytes_ptr), - ) + # _native_call covers the signing call only. The close() below is + # deliberately outside it, so the deferred teardown it records is + # performed on the way out rather than being deferred forever. + with self._native_call(): + if signer is not None: + # c2pa_builder_sign borrows the signer's handle, so the + # signer needs its own in-flight guard: the Builder's + # guard holds only the Builder's handle valid, and a + # signer.close() on another thread would otherwise free + # this handle mid-call. Entered inside self's guard so + # concurrent signs sharing objects acquire in one order. + # The check above is a fast-fail; this re-check inside + # the guard is the one that makes check-then-use atomic. + with signer._native_call(): + result = _lib.c2pa_builder_sign( + self._handle, + format_arg, + source_stream._stream, + dest_stream._stream, + signer._handle, + ctypes.byref(manifest_bytes_ptr) + ) + else: + result = _lib.c2pa_builder_sign_context( + self._handle, + format_arg, + source_stream._stream, + dest_stream._stream, + ctypes.byref(manifest_bytes_ptr), + ) # Sign borrows the Builder without taking ownership. # Closing here ensures resources clean up, # and single use/single sign done by a Builder. diff --git a/tests/test_unit_tests_threaded.py b/tests/test_unit_tests_threaded.py index f2b2a643..0b9b4f95 100644 --- a/tests/test_unit_tests_threaded.py +++ b/tests/test_unit_tests_threaded.py @@ -11,9 +11,12 @@ # specific language governing permissions and limitations under # each license. +import ast import ctypes import gc import os +import re +import inspect import io import json import subprocess @@ -3048,6 +3051,17 @@ class TestManagedResourceLockDeadlock(unittest.TestCase): JOIN_TIMEOUT = 30 + @classmethod + def setUpClass(cls): + with open(DEFAULT_TEST_FILE, 'rb') as handle: + cls.image_bytes = handle.read() + with open(os.path.join(FIXTURES_FOLDER, + "es256_certs.pem"), 'rb') as handle: + cls.certs = handle.read() + with open(os.path.join(FIXTURES_FOLDER, + "es256_private.key"), 'rb') as handle: + cls.private_key = handle.read() + def _join_all(self, threads, what): for thread in threads: thread.join(self.JOIN_TIMEOUT) @@ -3160,7 +3174,7 @@ def body(): def test_close_racing_json_does_not_deadlock(self): """close() on one thread against json() on another.""" - data = open(DEFAULT_TEST_FILE, 'rb').read() + data = self.image_bytes errors = [] def rounds(): @@ -3189,7 +3203,7 @@ def rounds(): def test_context_manager_exit_racing_json_does_not_deadlock(self): """__exit__ closes while another thread is calling json().""" - data = open(DEFAULT_TEST_FILE, 'rb').read() + data = self.image_bytes errors = [] def body(): @@ -3228,7 +3242,7 @@ def test_consume_failure_teardown_does_not_deadlock(self): with_fragment on a JPEG returns NotSupported, which routes through _raise_consume_failure. """ - data = open(DEFAULT_TEST_FILE, 'rb').read() + data = self.image_bytes errors = [] def body(): @@ -3253,11 +3267,9 @@ def test_close_during_sign_does_not_deadlock(self): """_sign_internal calls self.close() inside its own try block, so signing re-enters the lock on the signing thread. """ - certs = open(os.path.join(FIXTURES_FOLDER, - "es256_certs.pem"), 'rb').read() - key = open(os.path.join(FIXTURES_FOLDER, - "es256_private.key"), 'rb').read() - data = open(DEFAULT_TEST_FILE, 'rb').read() + certs = self.certs + key = self.private_key + data = self.image_bytes signer_info = C2paSignerInfo( alg=b"es256", sign_cert=certs, @@ -3295,7 +3307,7 @@ def test_stream_callback_reentering_api_does_not_deadlock(self): This passes only because construction does not hold the lock. """ - data = open(DEFAULT_TEST_FILE, 'rb').read() + data = self.image_bytes other = Reader("image/jpeg", io.BytesIO(data)) errors = [] @@ -3327,7 +3339,7 @@ def test_stream_callback_blocking_on_other_thread_does_not_deadlock(self): A lock held across construction deadlocks here, whether it is global or per-object. This is the test that pins the scoping decision. """ - data = open(DEFAULT_TEST_FILE, 'rb').read() + data = self.image_bytes target = Reader("image/jpeg", io.BytesIO(data)) errors = [] @@ -3366,7 +3378,7 @@ def test_no_nested_op_locks(self): That property, not the tests above, is what makes the design deadlock-free: with only one lock ever held, no cycle can form. """ - data = open(DEFAULT_TEST_FILE, 'rb').read() + data = self.image_bytes held = threading.local() violations = [] real_lock = ManagedResource._lock @@ -3410,7 +3422,7 @@ def __exit__(self, *exc): def test_concurrent_storm_terminates(self): """Readers, closers and collection running together must all finish.""" - data = open(DEFAULT_TEST_FILE, 'rb').read() + data = self.image_bytes stop = threading.Event() shared = [Reader("image/jpeg", io.BytesIO(data))] errors = [] @@ -3445,6 +3457,551 @@ def closer_worker(): self._join_all(threads, "concurrent storm") self.assertEqual(errors, []) + def _counted_free(self): + """Patch _free_native_ptr to count frees; returns the list.""" + freed = [] + real = ManagedResource._free_native_ptr + + def counting(ptr): + freed.append(ptr) + return real(ptr) + + ManagedResource._free_native_ptr = staticmethod(counting) + self.addCleanup( + lambda: setattr(ManagedResource, '_free_native_ptr', real)) + return freed + + def _thumbnail_uri(self, reader): + manifests = json.loads(reader.json()).get("manifests", {}) + for manifest in manifests.values(): + thumbnail = manifest.get("thumbnail") + if thumbnail and thumbnail.get("identifier"): + return thumbnail["identifier"] + self.skipTest("fixture has no thumbnail resource to stream") + + def test_close_inside_callback_defers_free(self): + """A close() from inside a stream callback must not free the handle + the native call is still using.""" + freed = self._counted_free() + reader = Reader("image/jpeg", io.BytesIO(self.image_bytes)) + uri = self._thumbnail_uri(reader) + during = [] + + class Closer(io.BytesIO): + def write(self, buffer): + reader.close() + during.append(len(freed)) + return super().write(buffer) + + try: + reader.resource_to_stream(uri, Closer()) + except Error: + pass + + self.assertEqual(during, [0], "handle was freed mid-call") + self.assertEqual(len(freed), 1, "deferred free did not run once") + self.assertEqual(reader._inflight, 0) + self.assertIsNone(reader._pending_teardown) + self.assertEqual(reader._lifecycle_state, LifecycleState.CLOSED) + + def test_cross_thread_close_during_callback_defers_free(self): + """Same race, with the close arriving from another thread.""" + freed = self._counted_free() + reader = Reader("image/jpeg", io.BytesIO(self.image_bytes)) + uri = self._thumbnail_uri(reader) + during = [] + started = threading.Event() + + class Slow(io.BytesIO): + def write(self, buffer): + started.set() + time.sleep(0.3) + during.append(len(freed)) + return super().write(buffer) + + def closer(): + started.wait(self.JOIN_TIMEOUT) + reader.close() + + thread = threading.Thread(target=closer) + thread.start() + try: + reader.resource_to_stream(uri, Slow()) + except Error: + pass + self._join_all([thread], "cross-thread closer") + + self.assertEqual(during, [0], "handle was freed mid-call") + self.assertEqual(len(freed), 1) + self.assertEqual(reader._inflight, 0) + + def test_deferred_teardown_still_closes(self): + """After a deferred free the resource is closed and a later close() + is a no-op rather than a second free.""" + freed = self._counted_free() + reader = Reader("image/jpeg", io.BytesIO(self.image_bytes)) + uri = self._thumbnail_uri(reader) + + class Closer(io.BytesIO): + def write(self, buffer): + reader.close() + return super().write(buffer) + + try: + reader.resource_to_stream(uri, Closer()) + except Error: + pass + + self.assertEqual(len(freed), 1) + reader.close() + self.assertEqual(len(freed), 1, "second close() freed again") + self.assertIsNone(reader._handle) + + def test_use_after_deferred_close_is_rejected(self): + """Deferring must not leave the resource usable: the free is pending, + so the handle is about to go away.""" + reader = Reader("image/jpeg", io.BytesIO(self.image_bytes)) + uri = self._thumbnail_uri(reader) + states = [] + + class Closer(io.BytesIO): + def write(self, buffer): + reader.close() + states.append(reader._lifecycle_state) + try: + reader.json() + states.append("json succeeded") + except Error: + states.append("json rejected") + return super().write(buffer) + + try: + reader.resource_to_stream(uri, Closer()) + except Error: + pass + + self.assertEqual(states[0], LifecycleState.CLOSED) + self.assertEqual(states[1], "json rejected") + + def test_exception_from_callback_still_frees(self): + """An exception unwinding through the native call must not strand the + in-flight counter, or the handle is never freed.""" + freed = self._counted_free() + reader = Reader("image/jpeg", io.BytesIO(self.image_bytes)) + uri = self._thumbnail_uri(reader) + + class Exploding(io.BytesIO): + def write(self, buffer): + reader.close() + raise RuntimeError("callback failure") + + try: + reader.resource_to_stream(uri, Exploding()) + except Exception: + pass + + self.assertEqual(reader._inflight, 0, "in-flight counter stranded") + self.assertEqual(len(freed), 1, "deferred free did not run") + + def test_inflight_cleared_before_deferred_free(self): + """The counter must reach zero before the deferred free runs. + + _teardown defers whenever _inflight is above zero, so performing the + free while the counter is still raised would defer it a second time + and the handle would never be released. + """ + seen = [] + reader = Reader("image/jpeg", io.BytesIO(self.image_bytes)) + uri = self._thumbnail_uri(reader) + real_release = Reader._release + + def probing_release(self): + seen.append(self._inflight) + return real_release(self) + + class Closer(io.BytesIO): + def write(self, buffer): + reader.close() + return super().write(buffer) + + with patch.object(Reader, '_release', probing_release): + try: + reader.resource_to_stream(uri, Closer()) + except Error: + pass + + self.assertEqual(seen, [0], + "deferred free ran while still counted in flight") + self.assertIsNone(reader._handle) + + def test_release_raising_during_deferred_teardown_does_not_leak(self): + """The deferred free survives a failing _release: the handle must + still be freed.""" + freed = self._counted_free() + reader = Reader("image/jpeg", io.BytesIO(self.image_bytes)) + uri = self._thumbnail_uri(reader) + + def boom(self): + raise RuntimeError("release failure") + + class Closer(io.BytesIO): + def write(self, buffer): + reader.close() + return super().write(buffer) + + with patch.object(Reader, '_release', boom): + try: + reader.resource_to_stream(uri, Closer()) + except Error: + pass + + self.assertEqual(reader._inflight, 0) + self.assertEqual(len(freed), 1, "handle leaked when _release raised") + + def test_concurrent_closes_during_callback_free_once(self): + """Many threads closing while one native call is in flight must + produce exactly one free.""" + freed = self._counted_free() + reader = Reader("image/jpeg", io.BytesIO(self.image_bytes)) + uri = self._thumbnail_uri(reader) + started = threading.Event() + closers = [] + + class Slow(io.BytesIO): + def write(self, buffer): + started.set() + time.sleep(0.3) + return super().write(buffer) + + def closer(): + started.wait(self.JOIN_TIMEOUT) + reader.close() + + for _ in range(8): + thread = threading.Thread(target=closer) + closers.append(thread) + thread.start() + try: + reader.resource_to_stream(uri, Slow()) + except Error: + pass + self._join_all(closers, "concurrent closers") + + self.assertEqual(len(freed), 1, + "racing closers freed {} times".format(len(freed))) + self.assertEqual(reader._inflight, 0) + + def test_sign_with_internal_close_frees_once(self): + """_sign_internal closes the Builder inside its own try, so the close + defers and the free happens on the way out.""" + freed = self._counted_free() + signer_info = C2paSignerInfo( + alg=b"es256", + sign_cert=self.certs, + private_key=self.private_key, + ta_url=b"http://timestamp.digicert.com", + ) + manifest = { + "claim_generator": "python_test", + "claim_generator_info": [ + {"name": "python_test", "version": "0.0.1"}], + "format": "image/jpeg", + "assertions": [], + } + signer = Signer.from_info(signer_info) + builder = Builder(manifest) + builder.sign(signer, "image/jpeg", + io.BytesIO(self.image_bytes), io.BytesIO()) + + self.assertEqual(builder._lifecycle_state, LifecycleState.CLOSED) + self.assertEqual(builder._inflight, 0) + builder_frees = [f for f in freed if f is not None] + self.assertGreaterEqual(len(builder_frees), 1) + with self.assertRaises(Error): + builder.sign(signer, "image/jpeg", + io.BytesIO(self.image_bytes), io.BytesIO()) + + def test_class_a_construction_is_not_guarded(self): + """Construction is deliberately unguarded: no external caller holds a + reference yet, and guarding it would reintroduce the deadlock where a + stream callback re-enters the API.""" + entered = [] + real = ManagedResource._native_call + + def recording(resource): + entered.append(type(resource).__name__) + return real(resource) + + ManagedResource._native_call = recording + try: + Reader("image/jpeg", io.BytesIO(self.image_bytes)) + finally: + ManagedResource._native_call = real + + self.assertEqual(entered, [], + "construction entered _native_call: guarding it " + "reintroduces the callback deadlock") + + def test_every_callback_running_method_is_guarded(self): + """Coverage check: every method that hands a Stream to the native + library must be guarded, except the three construction paths. + + A method missed here keeps the use-after-free, and the symptom is a + rare segfault rather than a failing test, so this is checked + mechanically rather than by eye. + """ + source = inspect.getsource(sys.modules[Reader.__module__]) + lines = source.split("\n") + class_a = { + ("Reader", "_create_reader"), + ("Reader", "_init_from_context"), + ("Builder", "from_archive"), + } + stream_use = re.compile( + r"(_stream|stream_obj|source_stream|dest_stream|main_obj" + r"|frag_obj)\._stream") + + bodies = {} + current_class = current_method = None + start = None + for index, line in enumerate(lines): + if re.match(r"^class ", line): + current_class = line.split("(")[0].replace( + "class ", "").strip(":") + if re.match(r"^def ", line): + current_class = None + match = re.match(r"^ def (\w+)", line) + if match: + if current_class and current_method and start is not None: + bodies[(current_class, current_method)] = "\n".join( + lines[start:index]) + current_method = match.group(1) + start = index + if current_class and current_method and start is not None: + bodies[(current_class, current_method)] = "\n".join(lines[start:]) + + unguarded = [] + checked = 0 + for key, body in bodies.items(): + if not stream_use.search(body): + continue + checked += 1 + if key in class_a: + continue + if "_native_call()" not in body: + unguarded.append("{}.{}".format(*key)) + + self.assertGreater(checked, 0, "coverage scan found no methods") + self.assertEqual( + unguarded, [], + "these hand a Stream to native without _native_call(): {}".format( + unguarded)) + + def test_every_borrowed_handle_is_guarded(self): + """Coverage check: when a method hands a *second* object's handle to + the native library, that object needs its own _native_call() guard. + + test_every_callback_running_method_is_guarded only asks whether the + string "_native_call()" appears in the method body, which cannot + express *whose* handle is guarded. A method that guards self while + passing signer._handle to native passes that check and still has the + use-after-free, so the ownership is checked structurally here. + """ + module = sys.modules[Reader.__module__] + tree = ast.parse(inspect.getsource(module)) + + # Attributes that carry a native handle out of an object. + handle_attrs = {"_handle", "execution_context"} + + def guarded_names(node): + """Names X with an active `with X._native_call():` at this node.""" + found = set() + for item in getattr(node, "items", []): + call = item.context_expr + if (isinstance(call, ast.Call) + and isinstance(call.func, ast.Attribute) + and call.func.attr == "_native_call" + and isinstance(call.func.value, ast.Name)): + found.add(call.func.value.id) + return found + + def borrowed_in_call(call): + """Names X whose handle this _lib.* call receives, X not self.""" + if not (isinstance(call.func, ast.Attribute) + and isinstance(call.func.value, ast.Name) + and call.func.value.id == "_lib"): + return set() + names = set() + for arg in ast.walk(call): + if (isinstance(arg, ast.Attribute) + and arg.attr in handle_attrs + and isinstance(arg.value, ast.Name) + and arg.value.id != "self"): + names.add(arg.value.id) + return names + + def locally_owned(method): + """Names bound to an object this method itself constructed. + + A resource created inside the method never escapes to another + thread, so nothing can close it mid-call and it needs no guard. + Only handles reaching the method from outside (parameters, + attributes) are exposed to a concurrent teardown. + """ + owned = set() + for node in ast.walk(method): + # `with self._NativeBuilder() as nb:` / `x = Foo()` + if isinstance(node, (ast.With, ast.AsyncWith)): + for item in node.items: + if (isinstance(item.context_expr, ast.Call) + and isinstance(item.optional_vars, ast.Name)): + owned.add(item.optional_vars.id) + elif isinstance(node, ast.Assign): + if isinstance(node.value, ast.Call): + for target in node.targets: + if isinstance(target, ast.Name): + owned.add(target.id) + return owned + + unguarded = [] + checked = 0 + + for cls in ast.walk(tree): + if not isinstance(cls, ast.ClassDef): + continue + for method in cls.body: + if not isinstance(method, (ast.FunctionDef, + ast.AsyncFunctionDef)): + continue + owned = locally_owned(method) + + # Walk the body tracking which guards are open, so a borrowed + # handle is only accepted when its own guard encloses the use. + def visit(node, active): + nonlocal checked + if isinstance(node, (ast.With, ast.AsyncWith)): + active = active | guarded_names(node) + if isinstance(node, ast.Call): + for name in borrowed_in_call(node) - owned: + checked += 1 + if name not in active: + unguarded.append( + "{}.{} passes {}._handle to native " + "without {}._native_call()".format( + cls.name, method.name, name, name)) + for child in ast.iter_child_nodes(node): + visit(child, active) + + visit(method, frozenset()) + + self.assertGreater( + checked, 0, + "ownership scan found no borrowed handles: the scan is broken") + self.assertEqual( + unguarded, [], + "borrowed handles used without their own guard:\n " + + "\n ".join(unguarded)) + + +class TestSharedSignerTeardownRace(unittest.TestCase): + """A Signer shared across threads must not be freed mid-sign. + + Builder.sign borrows the signer's handle for the duration of the native + call. Without a guard on the signer itself, a close() on another thread + frees that handle while c2pa_builder_sign is using it, and the process + dies with SIGSEGV instead of raising. + """ + + def setUp(self): + self.data_dir = os.path.join(os.path.dirname(__file__), "fixtures") + with open(os.path.join(self.data_dir, "C.jpg"), "rb") as f: + self.image_bytes = f.read() + with open(os.path.join(self.data_dir, "es256_certs.pem"), "rb") as f: + self.certs = f.read() + with open(os.path.join(self.data_dir, "es256_private.key"), "rb") as f: + self.key = f.read() + self.manifest = { + "claim_generator_info": [{"name": "test", "version": "0.1"}], + "assertions": [], + } + + def _make_signer(self): + return Signer.from_info(C2paSignerInfo( + SigningAlg.ES256, self.certs, self.key, None)) + + def test_close_during_concurrent_sign_does_not_crash(self): + """Rotate a shared signer while other threads sign with it. + + Runs in a subprocess: the failure mode is a segfault, which would + take the test runner down with it rather than reporting a failure. + """ + source = textwrap.dedent(""" + import io, os, sys, threading + from c2pa import (Builder, Signer, C2paSignerInfo, + C2paSigningAlg as SigningAlg) + + data_dir = sys.argv[1] + certs = open(os.path.join(data_dir, "es256_certs.pem"), "rb").read() + key = open(os.path.join(data_dir, "es256_private.key"), "rb").read() + img = open(os.path.join(data_dir, "C.jpg"), "rb").read() + manifest = {"claim_generator_info": + [{"name": "test", "version": "0.1"}], + "assertions": []} + + def make(): + return Signer.from_info(C2paSignerInfo( + SigningAlg.ES256, certs, key, None)) + + box = {"signer": make(), "stop": False} + + def rotate(): + while not box["stop"]: + old = box["signer"] + try: + box["signer"] = make() + old.close() + except Exception: + pass + + def sign(): + for _ in range(120): + if box["stop"]: + return + try: + b = Builder(manifest) + b.sign(box["signer"], "image/jpeg", + io.BytesIO(img), io.BytesIO()) + b.close() + except Exception: + # A closed signer may legitimately be rejected; + # only a crash is a failure here. + pass + + rot = threading.Thread(target=rotate, daemon=True) + rot.start() + threads = [threading.Thread(target=sign) for _ in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + box["stop"] = True + rot.join(timeout=5) + print("OK") + """) + + result = subprocess.run( + [sys.executable, "-c", source, self.data_dir], + capture_output=True, text=True, timeout=300) + + self.assertNotEqual( + result.returncode, -11, + "SIGSEGV: a signer was freed while a sign was using its handle") + self.assertEqual( + result.returncode, 0, + "shared-signer teardown race failed (rc={}):\n{}".format( + result.returncode, result.stderr[-2000:])) + self.assertIn("OK", result.stdout) + if __name__ == '__main__': unittest.main()