# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/against-what-environment-and-file-descriptor-inheritance-leaks-must-an-instance-sanitize-child-subprocess-contexts-to-prevent-credential-bleed-and-port-locking.md
# Remnant Warrant:    https://corpusfetch.com/c/against-what-environment-and-file-descriptor-inheritance-leaks-must-an-instance-sanitize-child-subprocess-contexts-to-prevent-credential-bleed-and-port-locking.warrant.json
# Markdown SHA-256:   594cc3c5e0f9d4beaa493471b43daad4a9c2d4a4b1995db8c1a5cb523e45439c
# Code AST SHA-256:   e680ef86eb55a9d06a805cb84b1270636ca80ded4d52de22e6c344a46531324b
# Verified Classes:   SubprocessIsolationGuard
# Defensive Guards:   3 fail-closed invariants (zero external deps)

import os
import signal
import subprocess
from pathlib import Path
from typing import Iterable, Mapping, Sequence

# Baseline safe environment keys that never contain credentials
_DEFAULT_SAFE_ENV_KEYS = ("HOME", "USER", "LOGNAME", "TMPDIR", "TEMP", "TMP")


def build_scrubbed_env(
    *,
    passthrough_keys: Iterable[str] = (),
    extra_env: Mapping[str, str] | None = None,
) -> dict[str, str]:
    '''
    Constructs a minimal, deterministic child environment dictionary from scratch.
    Prevents parent secrets (AWS_*, OPENAI_*, DATABASE_URL, *TOKEN*, *KEY*)
    from leaking into child build scripts, test runners, or CLI tools.
    '''
    clean_env: dict[str, str] = {
        "PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin",
        "LC_ALL": "C.UTF-8",
        "LANG": "C.UTF-8",
        "TZ": "UTC",
        # Force non-interactive fail-fast behavior across common CLI tools
        "CI": "true",
        "GIT_TERMINAL_PROMPT": "0",
        "DEBIAN_FRONTEND": "noninteractive",
        "PIP_NO_INPUT": "1",
        "NPM_CONFIG_YES": "true",
    }

    for key in (*_DEFAULT_SAFE_ENV_KEYS, *passthrough_keys):
        if key in os.environ:
            clean_env[key] = os.environ[key]

    if extra_env:
        for k, v in extra_env.items():
            clean_env[str(k)] = str(v)

    return clean_env


def run_isolated_subprocess(
    argv: Sequence[str | Path],
    *,
    cwd: str | Path | None = None,
    passthrough_env_keys: Iterable[str] = (),
    extra_env: Mapping[str, str] | None = None,
    stdin_bytes: bytes | None = None,
    timeout_sec: float = 30.0,
) -> subprocess.CompletedProcess[bytes]:
    '''
    Spawns a child process with a scrubbed allowlist environment, closed extra file
    descriptors, non-interactive stdin, and full process-group tree reaping on timeout.
    '''
    if not argv:
        raise ValueError("argv must not be empty")

    str_argv = [str(arg) for arg in argv]
    child_env = build_scrubbed_env(
        passthrough_keys=passthrough_env_keys,
        extra_env=extra_env,
    )

    proc = subprocess.Popen(
        str_argv,
        cwd=cwd,
        env=child_env,
        stdin=subprocess.DEVNULL if stdin_bytes is None else subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        shell=False,
        close_fds=True,          # Gate 1: Never leak parent sockets or lock fds
        start_new_session=True,  # Gate 2: Isolate PGID so whole tree can be reaped
    )

    try:
        stdout_data, stderr_data = proc.communicate(
            input=stdin_bytes, timeout=timeout_sec
        )
    except subprocess.TimeoutExpired:
        # Gate 3: Kill the entire child process group (including grandchildren)
        try:
            os.killpg(proc.pid, signal.SIGKILL)
        except ProcessLookupError:
            pass
        proc.communicate()  # Reap zombie PID from kernel process table
        raise TimeoutError(
            f"Subprocess {str_argv[0]!r} timed out after {timeout_sec}s; "
            f"reaped entire process group {proc.pid}."
        )

    if proc.returncode != 0:
        err_preview = stderr_data.decode("utf-8", errors="replace")[:500]
        raise subprocess.CalledProcessError(
            proc.returncode, str_argv, output=stdout_data, stderr=err_preview
        )

    return subprocess.CompletedProcess(
        args=str_argv,
        returncode=proc.returncode,
        stdout=stdout_data,
        stderr=stderr_data,
    )


class SubprocessIsolationGuard:
    """Operational boundary reference implementation."""
    build_scrubbed_env = staticmethod(build_scrubbed_env)
    run_isolated_subprocess = staticmethod(run_isolated_subprocess)