NativePlugin

NativePlugin

NativePlugin(verifier)

Bases: BasePlugin

Native function interception plugin.

Patches ctypes.CDLL.init and optionally cffi.FFI.dlopen at the class level. Uses reference counting so nested sandboxes work correctly.

Each library:function pair has its own FIFO deque of NativeMockConfig objects.

Source code in src/tripwire/plugins/native_plugin.py

def __init__(self, verifier: StrictVerifier) -> None:
    super().__init__(verifier)
    self._queues: dict[str, deque[NativeMockConfig]] = {}
    self._registry_lock: threading.Lock = threading.Lock()

mock_call

mock_call(library, function, *, returns, raises=None, required=True)

Register a mock for a single native function call.

Args: library: The library name (e.g., "libm"). function: The function name (e.g., "sqrt"). returns: Value to return when this mock is consumed. raises: If provided, this exception is raised instead of returning. required: If False, the mock is not reported as unused at teardown.

Source code in src/tripwire/plugins/native_plugin.py

def mock_call(
    self,
    library: str,
    function: str,
    *,
    returns: Any,  # noqa: ANN401
    raises: BaseException | None = None,
    required: bool = True,
) -> None:
    """Register a mock for a single native function call.

    Args:
        library: The library name (e.g., "libm").
        function: The function name (e.g., "sqrt").
        returns: Value to return when this mock is consumed.
        raises: If provided, this exception is raised instead of returning.
        required: If False, the mock is not reported as unused at teardown.
    """
    config = NativeMockConfig(
        library=library,
        function=function,
        returns=returns,
        raises=raises,
        required=required,
    )
    queue_key = f"{library}:{function}"
    with self._registry_lock:
        if queue_key not in self._queues:
            self._queues[queue_key] = deque()
        self._queues[queue_key].append(config)

check_conflicts

check_conflicts()

Verify ctypes.CDLL.init has not been patched by a third party.

Source code in src/tripwire/plugins/native_plugin.py

def check_conflicts(self) -> None:
    """Verify ctypes.CDLL.__init__ has not been patched by a third party."""
    current_init = ctypes.CDLL.__init__
    if (
        current_init is not _CDLL_INIT_ORIGINAL
        and current_init is not _patched_cdll_init
    ):
        patcher = _identify_native_patcher(current_init)
        raise ConflictError(
            target="ctypes.CDLL.__init__",
            patcher=patcher,
        )

install_patches

install_patches()

Install ctypes.CDLL and optionally cffi.FFI patches.

Source code in src/tripwire/plugins/native_plugin.py

def install_patches(self) -> None:
    """Install ctypes.CDLL and optionally cffi.FFI patches."""
    NativePlugin._original_cdll_init = ctypes.CDLL.__init__
    setattr(ctypes.CDLL, "__init__", _patched_cdll_init)

    # Optionally patch cffi if available
    if _CFFI_AVAILABLE:
        NativePlugin._original_ffi_dlopen = cffi_lib.FFI.dlopen
        cffi_lib.FFI.dlopen = _patched_ffi_dlopen

restore_patches

restore_patches()

Restore original ctypes.CDLL and cffi.FFI functions.

Source code in src/tripwire/plugins/native_plugin.py

def restore_patches(self) -> None:
    """Restore original ctypes.CDLL and cffi.FFI functions."""
    if NativePlugin._original_cdll_init is not None:
        setattr(ctypes.CDLL, "__init__", NativePlugin._original_cdll_init)
        NativePlugin._original_cdll_init = None
    if NativePlugin._original_ffi_dlopen is not None and _CFFI_AVAILABLE:
        cffi_lib.FFI.dlopen = NativePlugin._original_ffi_dlopen
        NativePlugin._original_ffi_dlopen = None

matches

matches(interaction, expected)

Field-by-field comparison with dirty-equals support.

Source code in src/tripwire/plugins/native_plugin.py

def matches(self, interaction: Interaction, expected: dict[str, Any]) -> bool:
    """Field-by-field comparison with dirty-equals support."""
    try:
        for key, expected_val in expected.items():
            actual_val = interaction.details.get(key)
            if expected_val != actual_val:
                return False
        return True
    except Exception:
        return False

get_unused_mocks

get_unused_mocks()

Return all NativeMockConfig with required=True still in any queue.

Source code in src/tripwire/plugins/native_plugin.py

def get_unused_mocks(self) -> list[NativeMockConfig]:
    """Return all NativeMockConfig with required=True still in any queue."""
    unused: list[NativeMockConfig] = []
    with self._registry_lock:
        for queue in self._queues.values():
            for config in queue:
                if config.required:
                    unused.append(config)
    return unused

assert_call

assert_call(library, function, *, args=())

Typed helper: assert the next native function call interaction.

Wraps assert_interaction() for ergonomic use. All three fields (library, function, args) are required.

Source code in src/tripwire/plugins/native_plugin.py

def assert_call(
    self,
    library: str,
    function: str,
    *,
    args: tuple[Any, ...] = (),
) -> None:
    """Typed helper: assert the next native function call interaction.

    Wraps assert_interaction() for ergonomic use. All three fields
    (library, function, args) are required.
    """
    from tripwire._context import _get_test_verifier_or_raise  # noqa: PLC0415

    source_id = f"native:{library}:{function}"
    sentinel = _NativeSentinel(source_id)
    _get_test_verifier_or_raise().assert_interaction(
        sentinel,
        library=library,
        function=function,
        args=args,
    )