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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
84 changes: 49 additions & 35 deletions src/c2pa/c2pa.py
Original file line number Diff line number Diff line change
Expand Up @@ -333,16 +333,15 @@ def _live_teardown_lock(self):
return lock

def _ensure_not_borrowed(self):
"""Raise if a native call is in flight on this handle.
"""Raise if any native call, shared or mutating, is in flight on
this handle, ensuring exclusive mutating calls.

Raises:
C2paError: If a native call is in flight on this resource.
"""
if self._inflight > 0:
name = type(self).__name__
raise C2paError(
f"{name} is in use by another operation and "
f"cannot be consumed")
f"{type(self).__name__} is in use by another operation")

def _ensure_no_mutating_call(self):
"""Raise if a mutating native call is in flight on this handle.
Expand All @@ -355,7 +354,7 @@ def _ensure_no_mutating_call(self):
f"{type(self).__name__} is running a mutating operation")

@contextlib.contextmanager
def _guarded_op(self, *, refuse_mut=True):
def _guarded_op(self, *, refuse_mut=True, exclusive=False):
"""Hold this resource's operation lock its duration,
and mark this thread as inside a native-error section.

Expand All @@ -370,7 +369,9 @@ def _guarded_op(self, *, refuse_mut=True):
with _native_section():
try:
with self._live_op_lock():
if refuse_mut:
if exclusive:
self._ensure_not_borrowed()
elif refuse_mut:
self._ensure_no_mutating_call()
yield
finally:
Expand Down Expand Up @@ -429,9 +430,9 @@ def _native_call(self):
@contextlib.contextmanager
def _exclusive_native_call(self):
"""Reserve the handle for a mutating call:
no other mutating call may run alongside it."""
no other call, shared or mutating, may run alongside it."""
with self._reserve(mutating=True,
refuse=self._ensure_no_mutating_call):
refuse=self._ensure_not_borrowed):
yield

@staticmethod
Expand Down Expand Up @@ -513,7 +514,8 @@ def _teardown(self, free_handle: bool):
lock = self._live_op_lock()
if not lock.acquire(blocking=False):
self._close_lifecycle()
_register_for_section_flush(self)
if _in_native_section():
_register_for_section_flush(self)
return

try:
Expand Down Expand Up @@ -545,6 +547,8 @@ def _record_pending_intent(self, free_handle: bool):
free only leaks.
"""
with self._live_teardown_lock():
if self._released:
return
pending = self._pending_teardown
if pending is None:
self._pending_teardown = free_handle
Expand Down Expand Up @@ -582,12 +586,12 @@ def _finish_teardown(self, free_handle: bool):
self._detach_in_child()
return

if self._released:
# Already done by another caller (concurrent caller).
return

self._released = True
self._lifecycle_state = LifecycleState.CLOSED
with self._live_teardown_lock():
if self._released:
return
self._released = True
self._pending_teardown = None
self._lifecycle_state = LifecycleState.CLOSED
self._safe_release()

handle, self._handle = self._handle, None
Expand Down Expand Up @@ -2101,15 +2105,21 @@ def set(self, path: str, value: str) -> 'Settings':

Args:
path: Dot-notation path (e.g. "builder.thumbnail.enabled").
value: The value to set.
value: Value to set, as JSON string.

Returns:
self, for method chaining.

Raises:
C2paError: If path or value contains a null byte, or native
rejects the path or the parsed value.
"""
path_bytes = _to_utf8_bytes(path, "settings path")
value_bytes = _to_utf8_bytes(value, "settings value")
_check_cstr_arg("settings path", path_bytes)
_check_cstr_arg("settings value", value_bytes)

with self._guarded_op():
with self._guarded_op(exclusive=True):
self._ensure_valid_state()

_check_ffi_operation_result(
Expand All @@ -2134,8 +2144,9 @@ def update(
self, for method chaining.
"""
data_bytes = _to_utf8_bytes(data, "settings data")
_check_cstr_arg("settings data", data_bytes)

with self._guarded_op():
with self._guarded_op(exclusive=True):
self._ensure_valid_state()

_check_ffi_operation_result(
Expand Down Expand Up @@ -2211,10 +2222,11 @@ class Context(ManagedResource, ContextProvider):
used directly again after that.
"""

class _NativeBuilder(ManagedResource):
"""Short-lived wrapper so the native context builder rides the normal
lifecycle: any failure inside its `with` block frees it via close()
unless a consuming call already took it.
class _NativeContextBuilder(ManagedResource):
"""Wrapper so the context builder gets a lifecycle:
a failure inside the `with` block frees via close()
unless a consuming call already did.
Wrapper should be short-lived.
"""

def __init__(self):
Expand Down Expand Up @@ -2250,27 +2262,28 @@ def __init__(
else:
# Any failure inside the with frees the builder via close();
# a successful build consumes it, so close() is then a no-op.
with self._NativeBuilder() as nb:
with self._NativeContextBuilder() as context_builder:
if settings is not None:
# Count in-progress reads.
with nb._guarded_op(), settings._native_call():
with context_builder._guarded_op(exclusive=True), \
settings._native_call():
_check_ffi_operation_result(
_lib.c2pa_context_builder_set_settings(
nb._handle, settings._c_settings),
context_builder._handle,
settings._c_settings),
"Failed to set settings on Context",
check=lambda r: r != 0)

if signer is not None:
# Retain a rejected signer for later teardown.
self._signer_callback_cb = signer._callback_cb
_check_handle_arg('builder', nb._handle)
_check_handle_arg('builder', context_builder._handle)
signer._consume_no_replacement(
lambda h: _lib.c2pa_context_builder_set_signer(
nb._handle, h),
context_builder._handle, h),
"Failed to set signer on Context: {}")
self._has_signer = True

context_ptr = nb._consume_into(
context_ptr = context_builder._consume_into(
lambda h: _lib.c2pa_context_builder_build(h),
"Failed to build Context: {}")

Expand Down Expand Up @@ -2630,6 +2643,7 @@ def close(self):
even if errors occur during cleanup.
Errors during cleanup are logged but not raised to ensure cleanup.
Multiple calls to close() are handled gracefully.
Only the Stream's owner closes the stream.
"""
# Checked before the lock, as _live_op_lock() and __del__ do:
# a child inherits _close_lock in whatever state it had at fork(),
Expand Down Expand Up @@ -3353,8 +3367,8 @@ def with_fragment(self, format: Optional[str], stream,
Reader. The Reader is untouched, and the call can be retried
after the other one returns. One thread feeds fragments to a
Reader, and the caller serializes those calls.
C2paError: "Reader is in use by another operation and cannot be
consumed" when another native call is in flight on it.
C2paError: "Reader is in use by another operation" when another
native call is in flight on it.

While this call runs, read methods on other threads raise
C2paError("Reader is running a mutating operation") and is_valid is
Expand Down Expand Up @@ -4103,7 +4117,7 @@ def set_no_embed(self):
into the asset when signing.
This is useful when creating cloud or sidecar manifests.
"""
with self._guarded_op():
with self._guarded_op(exclusive=True):
self._ensure_valid_state()
_lib.c2pa_builder_set_no_embed(self._handle)

Expand All @@ -4121,7 +4135,7 @@ def set_remote_url(self, remote_url: str):
"""
url_bytes = _to_utf8_bytes(remote_url, "remote URL")

with self._guarded_op():
with self._guarded_op(exclusive=True):
self._ensure_valid_state()

result = _lib.c2pa_builder_set_remote_url(self._handle, url_bytes)
Expand Down Expand Up @@ -4157,7 +4171,7 @@ def set_intent(
Raises:
C2paError: If there was an error setting the intent
"""
with self._guarded_op():
with self._guarded_op(exclusive=True):
self._ensure_valid_state()

result = _lib.c2pa_builder_set_intent(
Expand Down Expand Up @@ -4273,7 +4287,7 @@ def add_action(self, action_json: Union[str, dict]) -> None:
"""
action_str = _to_utf8_bytes(action_json, "action JSON")

with self._guarded_op():
with self._guarded_op(exclusive=True):
self._ensure_valid_state()

result = _lib.c2pa_builder_add_action(self._handle, action_str)
Expand Down
90 changes: 70 additions & 20 deletions tests/test_unit_tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
import ctypes
import threading
import concurrent.futures
from unittest.mock import patch

# Suppress deprecation warnings
warnings.simplefilter("ignore", category=DeprecationWarning)
Expand All @@ -52,6 +53,14 @@
ALTERNATIVE_INGREDIENT_TEST_FILE = os.path.join(FIXTURES_DIR, "cloud.jpg")


def _patch_free(test, fn):
"""Route ManagedResource._free_native_ptr to `fn` until `test` ends."""
patcher = patch.object(
ManagedResource, '_free_native_ptr', staticmethod(fn))
patcher.start()
test.addCleanup(patcher.stop)


def _fail_with_native_error(tag_bytes):
"""Build a mock FFI callable that sets a native error and returns None.
"""
Expand Down Expand Up @@ -6675,6 +6684,53 @@ def test_settings_update_dict(self):
self.assertIs(result, settings)
settings.close()

def test_settings_set_rejects_nul_in_path(self):
settings = Settings()
try:
with self.assertRaises(Error) as caught:
settings.set(
"builder.thumbnail.enabled\x00.tail", "false")
self.assertIn("null byte", str(caught.exception))
# Instance untouched by the refused call.
settings.set("builder.thumbnail.enabled", "false")
finally:
settings.close()

def test_settings_set_rejects_nul_in_value(self):
settings = Settings()
try:
with self.assertRaises(Error) as caught:
settings.set(
"builder.thumbnail.enabled", "false\x00true")
self.assertIn("null byte", str(caught.exception))
settings.set("builder.thumbnail.enabled", "false")
finally:
settings.close()

def test_settings_update_rejects_nul_in_json_string(self):
settings = Settings.from_dict({
"builder": {"thumbnail": {"enabled": True}},
})
try:
with self.assertRaises(Error) as caught:
settings.update(
'{"verify": {"verify_after_sign": true}}\x00'
'{"builder": {"thumbnail": {"enabled": false}}}')
self.assertIn("null byte", str(caught.exception))
finally:
settings.close()

def test_settings_set_string_value_needs_json_quotes(self):
settings = Settings()
try:
with self.assertRaises(Error):
settings.set(
"builder.claim_generator_info.name", "MyApp")
settings.set(
"builder.claim_generator_info.name", '"MyApp"')
finally:
settings.close()

def test_settings_is_valid_after_close(self):
settings = Settings()
settings.close()
Expand Down Expand Up @@ -7982,10 +8038,14 @@ 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)
# Registered first so it runs last, after every patch has unwound.
self.addCleanup(self._assert_free_hook_restored)
_patch_free(self, self.freed.append)

def tearDown(self):
ManagedResource._free_native_ptr = self._real_free
def _assert_free_hook_restored(self):
self.assertIs(
ManagedResource._free_native_ptr, self._real_free,
"{} leaked a _free_native_ptr patch".format(self.id()))

def _free_counts(self):
counts = {}
Expand All @@ -7995,7 +8055,7 @@ def _free_counts(self):

def _use_real_frees(self):
"""Undo free recorder, so native handles are really freed."""
ManagedResource._free_native_ptr = self._real_free
_patch_free(self, self._real_free)

def _make_signer(self):
with open(os.path.join(self.data_dir, "es256_certs.pem"), "rb") as f:
Expand Down Expand Up @@ -8351,12 +8411,9 @@ def test_context_build_failure_consumes_signer(self):

# 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:
with patch.object(ManagedResource, '_free_native_ptr',
staticmethod(freed.append)):
signer.close()
finally:
ManagedResource._free_native_ptr = real_free
self.assertEqual(freed, [])

def test_context_with_signer_consumes_it_on_success(self):
Expand Down Expand Up @@ -8501,7 +8558,7 @@ def polluting_free(ptr):
c2pa_module._lib.c2pa_error_set_last(
"Other: UntrackedPointer: {:#x}".format(ptr).encode())
return -1
ManagedResource._free_native_ptr = staticmethod(polluting_free)
_patch_free(self, polluting_free)

def ffi_call(handle):
nonlocal bystander
Expand Down Expand Up @@ -8590,7 +8647,7 @@ def flaky_free(ptr):
raise RuntimeError("simulated free failure")
self.freed.append(ptr)
return 0
ManagedResource._free_native_ptr = staticmethod(flaky_free)
_patch_free(self, flaky_free)

with self.assertLogs('c2pa', level='ERROR') as captured:
with c2pa_module._native_section():
Expand Down Expand Up @@ -8648,11 +8705,7 @@ def _instrument_frees(self):
"""Record frees instead of performing them, and restore on teardown.
"""
freed = []
real_free = ManagedResource._free_native_ptr
ManagedResource._free_native_ptr = staticmethod(freed.append)
self.addCleanup(
lambda: setattr(
ManagedResource, '_free_native_ptr', real_free))
_patch_free(self, freed.append)
return freed

def _free_count(self, freed, handle):
Expand Down Expand Up @@ -10361,10 +10414,7 @@ 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)
_patch_free(self, counting_free)

def test_generic_exception_frees_the_reserved_handle(self):
"""A reserved consume that raises must free, not drop, the handle.
Expand Down
Loading
Loading