SubprocessPlugin

SubprocessPlugin

SubprocessPlugin(verifier)

Bases: BasePlugin

Subprocess interception plugin.

Patches subprocess.run and shutil.which globally. Uses reference counting so nested sandboxes work correctly, following the HttpPlugin pattern exactly.

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

def __init__(self, verifier: "StrictVerifier") -> None:
    super().__init__(verifier)
    # FIFO queue for run mocks (per-plugin instance, per-verifier)
    self._run_queue: deque[RunMockConfig] = deque()
    # Dict keyed by binary name for which mocks
    self._which_mocks: dict[str, WhichMockConfig] = {}
    # Set of which() names that were actually called (for unused-mock tracking)
    self._which_called: set[str] = set()
    self._run_sentinel = SubprocessRunSentinel(self)
    self._which_sentinel = SubprocessWhichSentinel(self)

run property

run

Sentinel used as source argument in assert_interaction() for subprocess.run.

which property

which

Sentinel used as source argument in assert_interaction() for shutil.which.

install

install()

No-op. Called to ensure plugin is registered before sandbox entry.

Access to any attribute of subprocess triggers plugin creation via _SubprocessProxy.getattr. This method exists as a named no-op so tests that want the bouncer active without any mocks have an explicit API to call.

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

def install(self) -> None:
    """No-op. Called to ensure plugin is registered before sandbox entry.

    Access to any attribute of subprocess triggers plugin creation via
    _SubprocessProxy.__getattr__. This method exists as a named no-op so
    tests that want the bouncer active without any mocks have an explicit
    API to call.
    """

mock_run

mock_run(command, *, returncode=0, stdout='', stderr='', raises=None, required=True)

Register a FIFO subprocess.run mock.

Calls are matched in registration order. An unmocked or out-of-order call raises UnmockedInteractionError immediately (bouncer guarantee).

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

def mock_run(
    self,
    command: list[str],
    *,
    returncode: int = 0,
    stdout: str = "",
    stderr: str = "",
    raises: BaseException | None = None,
    required: bool = True,
) -> None:
    """Register a FIFO subprocess.run mock.

    Calls are matched in registration order. An unmocked or out-of-order
    call raises UnmockedInteractionError immediately (bouncer guarantee).
    """
    self._run_queue.append(
        RunMockConfig(
            command=command,
            returncode=returncode,
            stdout=stdout,
            stderr=stderr,
            raises=raises,
            required=required,
        )
    )

mock_which

mock_which(name, returns, *, required=False)

Register a shutil.which mock keyed by binary name.

Always-on: unregistered names are swallowed (return None) and recorded on the timeline, requiring assertion at teardown. Registered names return the configured value. required=False by default because tests often register more alternatives than will be hit in a given path.

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

def mock_which(
    self,
    name: str,
    returns: str | None,
    *,
    required: bool = False,
) -> None:
    """Register a shutil.which mock keyed by binary name.

    Always-on: unregistered names are swallowed (return None) and recorded
    on the timeline, requiring assertion at teardown. Registered names
    return the configured value. required=False by default because
    tests often register more alternatives than will be hit in a given path.
    """
    self._which_mocks[name] = WhichMockConfig(
        name=name,
        returns=returns,
        required=required,
    )

assert_run

assert_run(command, returncode, stdout, stderr)

Assert the next subprocess.run interaction with all 4 fields.

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

def assert_run(
    self,
    command: list[str],
    returncode: int,
    stdout: str,
    stderr: str,
) -> None:
    """Assert the next subprocess.run interaction with all 4 fields."""
    self.verifier.assert_interaction(
        self._run_sentinel,
        command=command,
        returncode=returncode,
        stdout=stdout,
        stderr=stderr,
    )

assert_which

assert_which(name, returns)

Assert the next shutil.which interaction with all 2 fields.

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

def assert_which(
    self,
    name: str,
    returns: str | None,
) -> None:
    """Assert the next shutil.which interaction with all 2 fields."""
    self.verifier.assert_interaction(
        self._which_sentinel,
        name=name,
        returns=returns,
    )

check_conflicts

check_conflicts()

Verify subprocess.run and shutil.which have not been patched by a third party.

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

def check_conflicts(self) -> None:
    """Verify subprocess.run and shutil.which have not been patched by a third party."""
    current_run = subprocess.run
    if (
        current_run is not _SUBPROCESS_RUN_ORIGINAL
        and current_run is not _tripwire_subprocess_run
    ):
        patcher = _identify_subprocess_patcher(current_run)
        raise ConflictError(
            target="subprocess.run",
            patcher=patcher,
        )

    current_which = shutil.which
    if (
        current_which is not _SHUTIL_WHICH_ORIGINAL
        and current_which is not _tripwire_shutil_which
    ):
        patcher = _identify_subprocess_patcher(current_which)
        raise ConflictError(
            target="shutil.which",
            patcher=patcher,
        )