process_utils

process_utils

Subprocess helpers with hard timeout and process-tree cleanup.

Functions:

run_with_timeout

run_with_timeout(
    args: str | Sequence[str],
    *,
    timeout: float,
    shell: bool = False,
    cwd: str | Path | None = None,
    env: Mapping[str, str] | None = None,
) -> CompletedProcess[str]

Run a command and kill its process group if the deadline expires.

subprocess.run(timeout=...) kills only the immediate child. Starting a new session on POSIX lets Diapason terminate the whole process group, so a timed-out shell command cannot leave grandchildren running in background.

Source code in src/diapason/security/process_utils.py

def run_with_timeout(
    args: str | Sequence[str],
    *,
    timeout: float,
    shell: bool = False,
    cwd: str | Path | None = None,
    env: Mapping[str, str] | None = None,
) -> subprocess.CompletedProcess[str]:
    """Run a command and kill its process group if the deadline expires.

    ``subprocess.run(timeout=...)`` kills only the immediate child.  Starting a
    new session on POSIX lets Diapason terminate the whole process group, so a
    timed-out shell command cannot leave grandchildren running in background.
    """
    popen_kwargs: dict[str, Any] = {
        "cwd": cwd,
        "env": env,
        "shell": shell,
        "stderr": subprocess.PIPE,
        "stdout": subprocess.PIPE,
        "text": True,
    }
    if os.name == "posix":
        popen_kwargs["start_new_session"] = True
    elif os.name == "nt":
        popen_kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP

    process = subprocess.Popen(args, **popen_kwargs)  # noqa: S603
    try:
        stdout, stderr = process.communicate(timeout=timeout)
    except subprocess.TimeoutExpired as exc:
        _terminate_process_tree(process)
        stdout, stderr = process.communicate()
        raise subprocess.TimeoutExpired(
            process.args,
            timeout,
            output=stdout,
            stderr=stderr,
        ) from exc

    return subprocess.CompletedProcess(
        process.args,
        process.returncode,
        stdout,
        stderr,
    )