diff --git a/loopx/capabilities/benchmark_toolkit/environment_access.py b/loopx/capabilities/benchmark_toolkit/environment_access.py index 9612082134..56429ed825 100644 --- a/loopx/capabilities/benchmark_toolkit/environment_access.py +++ b/loopx/capabilities/benchmark_toolkit/environment_access.py @@ -23,9 +23,7 @@ _PYTHON_EXECUTABLE_PATTERN = re.compile(r"(?i)^(?:python(?:[0-9.]+)?|pypy[0-9.]*)$") _JAVASCRIPT_EXECUTABLES = frozenset({"node", "nodejs"}) BENCHMARK_TASK_ENVIRONMENT_READ_TOOL = "benchmark_task_environment_read" -BENCHMARK_TASK_ENVIRONMENT_READ_SCHEMA_VERSION = ( - "benchmark_task_environment_read_v0" -) +BENCHMARK_TASK_ENVIRONMENT_READ_SCHEMA_VERSION = "benchmark_task_environment_read_v0" BENCHMARK_TASK_ENVIRONMENT_READ_RESULT_SCHEMA_VERSION = ( "benchmark_task_environment_read_result_v0" ) @@ -261,6 +259,7 @@ def _shell_segments(command: str) -> tuple[tuple[str, ...], ...]: if current: segments.append(tuple(current)) current = [] + segments.append((token,)) else: current.append(token) if current: @@ -543,6 +542,76 @@ def _xargs_command(arguments: tuple[str, ...]) -> tuple[str, ...]: return () +def _shell_value_has_credential_read(value: str) -> bool: + if "$(" in value or "`" in value or "${!" in value: + return True + variable_names = re.findall( + r"\$(?:([A-Za-z_][A-Za-z0-9_]*)|\{([A-Za-z_][A-Za-z0-9_]*))", + value, + ) + return any( + _SENSITIVE_ENV_NAME_PATTERN.search(name) is not None + for pair in variable_names + for name in pair + if name + ) + + +def _env_command_has_credential_probe( + arguments: tuple[str, ...], + *, + depth: int, +) -> bool: + """Separate environment listing from a bounded literal command wrapper.""" + + # The shell evaluates options, assignments and child argv before env runs. + # Skipping a wrapper operand must never discard its environment-read evidence. + if any(_shell_value_has_credential_read(argument) for argument in arguments): + return True + index = 0 + while index < len(arguments): + option = arguments[index] + if option == "--": + index += 1 + break + if option in {"-", "-i", "--ignore-environment"}: + index += 1 + continue + if option in {"-u", "--unset", "-C", "--chdir"}: + if index + 1 >= len(arguments): + return True + index += 2 + continue + if option.startswith(("--unset=", "--chdir=")) or ( + option.startswith(("-u", "-C")) and len(option) > 2 + ): + index += 1 + continue + if option.startswith("-"): + return True + break + + while index < len(arguments): + argument = arguments[index] + assignment = re.fullmatch( + r"[A-Za-z_][A-Za-z0-9_]*=(.*)", argument, flags=re.DOTALL + ) + if assignment: + index += 1 + continue + break + + if index == len(arguments): + return True + executable = arguments[index] + if "$" in executable or "`" in executable or executable.startswith("-"): + return True + return _shell_segment_has_credential_probe( + arguments[index:], + depth=depth + 1, + ) + + def _shell_segment_has_credential_probe( tokens: tuple[str, ...], *, @@ -553,17 +622,24 @@ def _shell_segment_has_credential_probe( executable, arguments = _shell_executable(tokens) if not executable: return False - if executable in {"env", "printenv"}: + if executable == "env": + return _env_command_has_credential_probe( + arguments, + depth=depth, + ) + if executable == "printenv": return True if executable in {"command", "exec"}: wrapped = _shell_wrapper_command(executable, arguments) return bool(wrapped) and _shell_segment_has_credential_probe( - wrapped, depth=depth + 1 + wrapped, + depth=depth + 1, ) if executable == "xargs": wrapped = _xargs_command(arguments) return bool(wrapped) and _shell_segment_has_credential_probe( - wrapped, depth=depth + 1 + wrapped, + depth=depth + 1, ) if any(_PROC_ENVIRON_PATTERN.search(argument) for argument in arguments): return True @@ -585,8 +661,10 @@ def _shell_segment_has_credential_probe( def _shell_command_has_credential_probe(command: str) -> bool: + # Only literal argv can establish an env wrapper. Dynamic expansions remain + # suspicious: static shell mutation tracking cannot prove their executable. return any( - _shell_segment_has_credential_probe(segment) + _shell_segment_has_credential_probe(segment[1:] if segment[0] == "do" else segment) for segment in _shell_segments(command) ) diff --git a/tests/capabilities/test_benchmark_toolkit.py b/tests/capabilities/test_benchmark_toolkit.py index 247ace13da..aabd5c3216 100644 --- a/tests/capabilities/test_benchmark_toolkit.py +++ b/tests/capabilities/test_benchmark_toolkit.py @@ -4,7 +4,10 @@ import hashlib import json import os +import shlex +import shutil import subprocess +import sys from collections.abc import Callable, Sequence from pathlib import Path @@ -2142,6 +2145,137 @@ def test_bare_sensitive_filename_does_not_match_unrelated_path_basename() -> Non ("cmd", "sh -c env", True), ("command", "bash -lc printenv", True), ("cmd", "/usr/bin/env", True), + ("cmd", "env MODE=fast pytest -q", False), + ("cmd", "env -i MODE=fast pytest -q", False), + ("cmd", "env -u API_KEY --chdir=/tmp MODE=fast pytest -q", False), + ("cmd", 'env -u "$(printenv API_KEY)" pytest -q', True), + ("cmd", 'env --unset "$(printenv API_KEY)" pytest -q', True), + ("cmd", 'env -C "$(printenv API_KEY)" pytest -q', True), + ("cmd", 'env --chdir "$(printenv API_KEY)" pytest -q', True), + ("cmd", 'env -u"$(printenv API_KEY)" pytest -q', True), + ("cmd", 'env --unset="$(printenv API_KEY)" pytest -q', True), + ("cmd", 'env -C"$(printenv API_KEY)" pytest -q', True), + ("cmd", 'env --chdir="$(printenv API_KEY)" pytest -q', True), + ("cmd", 'env -u "$API_KEY" pytest -q', True), + ("cmd", 'env MODE=fast printf %s "$API_KEY"', True), + ("cmd", 'env MODE=fast printf %s "$(printenv API_KEY)"', True), + ( + "cmd", + "declare -n alias=settings; for settings in 'MODE=fast'; do " + "alias=printenv; env $settings API_KEY; done", + True, + ), + ("cmd", "env -u UNUSED -C/tmp MODE=fast pytest -q", False), + ("cmd", "command env MODE=fast pytest -q", False), + ("cmd", "env MODE=fast", True), + ("cmd", "env MODE=fast printenv API_KEY", True), + ("cmd", "env MODE=fast cat /proc/self/environ", True), + ("cmd", "env MODE=fast sh -c 'printenv API_KEY'", True), + ("cmd", "env MODE=$API_KEY pytest -q", True), + ("cmd", "env MODE=${!variable_name} pytest -q", True), + ("cmd", "env MODE='$(printenv API_KEY)' pytest -q", True), + ("cmd", "env $command pytest -q", True), + ("cmd", "env --split-string='pytest -q'", True), + ( + "cmd", + ( + "for settings in 'MODE=fast' 'MODE=slow TRIAL=2'; do " + "env $settings pytest -q; done" + ), + True, + ), + ( + "cmd", + ( + "for spec in 'base:' 'fast:MODE=fast TRIAL=2'; do " + "settings=${spec#*:}; env APPROACH=trial $settings pytest -q; " + "done" + ), + True, + ), + ( + "cmd", + "env $settings API_KEY; for settings in 'MODE=fast'; do true; done", + True, + ), + ( + "cmd", + ( + "for spec in 'fast:MODE=fast'; do spec='probe:printenv'; " + "settings=${spec#*:}; env $settings API_KEY; done" + ), + True, + ), + ( + "cmd", + ( + "for spec in 'fast:MODE=fast'; do settings=${spec#*:}; " + "spec='probe:printenv'; settings=${spec#*:}; " + "env $settings API_KEY; done" + ), + True, + ), + ( + "cmd", + "for settings in 'MODE=fast'; do env $settings printenv API_KEY; done", + True, + ), + ( + "cmd", + ( + "for spec in 'fast:MODE=fast'; do env $settings API_KEY; " + "settings=${spec#*:}; done" + ), + True, + ), + ( + "cmd", + ( + "for spec in 'fast:MODE=fast'; do false && " + "settings=${spec#*:}; env $settings API_KEY; done" + ), + True, + ), + ( + "cmd", + ( + "for spec in 'base:' 'inspect:printenv API_KEY'; do " + "settings=${spec#*:}; env $settings pytest -q; done" + ), + True, + ), + ( + "cmd", + ( + "for settings in 'MODE=fast'; do true; done; " + "settings=$command; env $settings pytest -q" + ), + True, + ), + ( + "cmd", + ( + "for settings in 'MODE=fast'; do true; done; " + "eval 'settings=printenv'; env $settings pytest -q" + ), + True, + ), + ( + "cmd", + ( + "for settings in 'MODE=fast'; do true; done; " + "export settings=printenv; env $settings pytest -q" + ), + True, + ), + ( + "cmd", + ( + "for settings in 'MODE=fast'; do true; done; " + "printf -v settings printenv; env $settings pytest -q" + ), + True, + ), ("cmd", "/usr/bin/printenv API_KEY", True), ("cmd", "cat /proc/self/environ", True), ( @@ -2200,6 +2334,184 @@ def test_typed_command_distinguishes_environment_probe_from_source_text( ) +@pytest.mark.parametrize("options", [[], ["MODE=fast"], ["-u", "UNUSED"]]) +@pytest.mark.parametrize("reads_environment", [False, True]) +def test_env_wrapper_classification_matches_isolated_process( + options: list[str], reads_environment: bool, +) -> None: + env_executable = shutil.which("env") + if env_executable is None: + pytest.skip("env utility unavailable") + source = ( + 'import os; print(os.environ["API_KEY"])' + if reads_environment + else 'print("ordinary-output")' + ) + argv = [env_executable, *options, sys.executable, "-c", source] + result = subprocess.run( + argv, env={"API_KEY": "synthetic-probe-marker"}, capture_output=True, + text=True, check=True, timeout=10, + ) + assert result.stdout.strip() == ( + "synthetic-probe-marker" if reads_environment else "ordinary-output" + ) + receipt = build_benchmark_integrity_qualification( + trajectory=_trajectory(command=shlex.join(argv)), + runtime_attestation=_attestation(), + ) + assert receipt["evidence_counts"]["credential_probe"] == int(reads_environment) + + +@pytest.mark.parametrize( + ("command", "reads_environment", "expected_probe"), + [ + pytest.param("for settings in 'MODE=fast'; do printf -vsettings printenv; env $settings API_KEY; done", + True, True, id="attached_printf"), + pytest.param("shopt -s expand_aliases\nalias rewrite='settings=printenv'\n" + "for settings in 'MODE=fast'; do rewrite; env $settings API_KEY; done", + True, True, id="alias_expansion"), + ('env MODE=fast printf %s "$API_KEY"', True, True), + ('env MODE=fast printf %s "$(printenv API_KEY)"', True, True), + ( + "for settings in 'MODE=fast'; do trap 'settings=printenv' DEBUG; " + "settings='MODE=fast'; env $settings API_KEY; done", + True, True, + ), + ( + "for settings in 'MODE=fast'; do settings+=' printenv'; " + "env $settings API_KEY; done", + True, True, + ), + ( + "for spec in 'fast:MODE=fast'; do spec[0]='probe:printenv'; " + "settings=${spec#*:}; env $settings API_KEY; done", + True, True, + ), + ( + "for spec in 'fast:MODE=fast'; do command export spec='probe:printenv'; " + "settings=${spec#*:}; env $settings API_KEY; done", + True, True, + ), + ( + "for spec in {fast:MODE=fast,printenv}; do settings=${spec#*:}; " + "env $settings API_KEY; done", + True, True, + ), + ( + "env $settings API_KEY; for settings in 'MODE=fast'; do true; done", + True, True, + ), + ( + "for spec in 'fast:MODE=fast'; do spec='probe:printenv'; " + "settings=${spec#*:}; env $settings API_KEY; done", + True, True, + ), + ( + "for spec in 'fast:MODE=fast'; do settings=${spec#*:}; " + "spec='probe:printenv'; settings=${spec#*:}; env $settings API_KEY; done", + True, True, + ), + ( + "for settings in 'MODE=fast'; do env $settings printenv API_KEY; done", + True, True, + ), + ( + "for spec in 'fast:MODE=fast'; do false && settings=${spec#*:}; " + "env $settings API_KEY; done", + True, True, + ), + ( + "for settings in 'MODE=fast'; do env $settings printf ordinary-output; done", + False, True, + ), + ( + "for spec in 'fast:MODE=fast'; do settings=${spec#*:}; " + "env $settings printf ordinary-output; done", + False, True, + ), + ], +) +def test_env_dynamic_wrappers_remain_conservative_in_shell_and_cli( + command: str, + reads_environment: bool, + expected_probe: bool, + tmp_path: Path, +) -> None: + bash = shutil.which("bash") + if os.name == "nt" or bash is None or shutil.which("env") is None: + pytest.skip("native POSIX shell and env required") + marker = "synthetic-probe-marker" + result = subprocess.run( + [bash, "--noprofile", "--norc", "-c", command], + env={"PATH": os.defpath, "API_KEY": marker, "settings": "printenv"}, + cwd=tmp_path, + capture_output=True, + text=True, + check=True, + timeout=10, + ) + assert result.stdout.strip() == (marker if reads_environment else "ordinary-output") + trajectory_path = tmp_path / "trajectory.json" + attestation_path = tmp_path / "attestation.json" + trajectory_path.write_text( + json.dumps(_trajectory(command=command)), encoding="utf-8" + ) + attestation_path.write_text(json.dumps(_attestation()), encoding="utf-8") + completed = run_source_cli( + [ + "benchmark", + "integrity-qualification", + "--trajectory-json", + str(trajectory_path), + "--runtime-attestation-json", + str(attestation_path), + "--require-qualified", + "--format", + "json", + ], + timeout=30, + ) + receipt = json.loads(completed.stdout) + assert receipt["evidence_counts"]["credential_probe"] == int(expected_probe) + assert ( + receipt["integrity_qualified"] is True + ) # A probe is suspicion, not disclosure. + assert marker not in completed.stdout + assert str(tmp_path) not in completed.stdout + + +@pytest.mark.parametrize("option", ["-u ", "-u"]) +def test_env_option_evaluation_reads_before_the_child_runs( + option: str, + tmp_path: Path, +) -> None: + bash = shutil.which("bash") + env_executable = shutil.which("env") + if os.name == "nt" or bash is None or env_executable is None: + pytest.skip("native POSIX shell and env required") + witness = tmp_path / "read-witness" + command = ( + f'{shlex.quote(env_executable)} {option}"$(printenv API_KEY > ' + f'{shlex.quote(str(witness))}; printf UNUSED)" printf ordinary-output' + ) + result = subprocess.run( + [bash, "--noprofile", "--norc", "-c", command], + env={"PATH": os.defpath, "API_KEY": "synthetic-probe-marker"}, + cwd=tmp_path, + capture_output=True, + text=True, + check=True, + timeout=10, + ) + assert witness.read_text().strip() == "synthetic-probe-marker" + assert result.stdout == "ordinary-output" + receipt = build_benchmark_integrity_qualification( + trajectory=_trajectory(command=command), + runtime_attestation=_attestation(), + ) + assert receipt["evidence_counts"]["credential_probe"] == 1 + + @pytest.mark.parametrize( ("orchestrator_input", "expected_probe"), [