diff --git a/loopx/capabilities/benchmark_toolkit/environment_access.py b/loopx/capabilities/benchmark_toolkit/environment_access.py index 9612082134..3e2aa0650c 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" ) @@ -543,27 +541,231 @@ def _xargs_command(arguments: tuple[str, ...]) -> tuple[str, ...]: return () +def _assignment_words(value: str) -> bool: + """Return whether a literal shell word list contains only assignments.""" + + if "$(" in value or "`" in value or "$" in value: + return False + try: + words = shlex.split(value) + except ValueError: + return False + return all( + re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*=.*", word, flags=re.DOTALL) is not None + for word in words + ) + + +def _proven_assignment_expansions(command: str) -> frozenset[str]: + """Find bounded loop variables proven to expand only to assignments. + + Benchmark task scripts commonly iterate over a literal list of environment + configurations, then pass that list to ``env`` before a fixed task command. + Only literal ``for`` values and the exact ``${name#*:}`` suffix transform + are accepted. Dynamic loop values and other shell transforms stay unknown. + """ + + assignment_variables: set[str] = set() + assignment_suffix_variables: set[str] = set() + segments = _shell_segments(command) + for segment_index, segment in enumerate(segments): + if ( + len(segment) < 3 + or segment[0] != "for" + or re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", segment[1]) is None + or segment[2] != "in" + ): + continue + variable = segment[1] + values = list(segment[3:]) + found_do = False + for continued in segments[segment_index + 1 :]: + if continued and continued[0] == "do": + found_do = True + break + values.extend(continued) + if not found_do: + continue + if values and all(_assignment_words(value) for value in values): + assignment_variables.add(variable) + if values and all( + ":" in value and _assignment_words(value.split(":", 1)[1]) + for value in values + ): + assignment_suffix_variables.add(variable) + + changed = True + while changed: + changed = False + for segment in segments: + for token in segment: + suffix = re.fullmatch( + r"([A-Za-z_][A-Za-z0-9_]*)=\$\{" + r"([A-Za-z_][A-Za-z0-9_]*)#\*:\}", + token, + ) + if suffix and suffix.group(2) in assignment_suffix_variables: + if suffix.group(1) not in assignment_variables: + assignment_variables.add(suffix.group(1)) + changed = True + + if any( + _shell_executable(segment)[0] in {".", "eval", "source"} for segment in segments + ): + return frozenset() + invalidated: set[str] = set() + for segment in segments: + executable, arguments = _shell_executable(segment) + if executable in { + "declare", + "export", + "local", + "read", + "readonly", + "typeset", + "unset", + }: + mutated_names = { + argument.split("=", 1)[0] + for argument in arguments + if not argument.startswith("-") + } + invalidated.update(assignment_variables.intersection(mutated_names)) + if executable == "printf" and "-v" in arguments: + variable_index = arguments.index("-v") + 1 + if variable_index < len(arguments): + invalidated.add(arguments[variable_index]) + if executable in {"getopts", "mapfile", "readarray"}: + invalidated.update(assignment_variables.intersection(arguments)) + for token in segment: + assignment = re.fullmatch( + r"([A-Za-z_][A-Za-z0-9_]*)=(.*)", token, flags=re.DOTALL + ) + if assignment is None: + break + variable, value = assignment.groups() + if variable not in assignment_variables: + continue + suffix = re.fullmatch(r"\$\{([A-Za-z_][A-Za-z0-9_]*)#\*:\}", value) + if suffix and suffix.group(1) in assignment_suffix_variables: + continue + if not _assignment_words(value): + invalidated.add(variable) + assignment_variables.difference_update(invalidated) + return frozenset(assignment_variables) + + +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, + assignment_expansions: frozenset[str], +) -> bool: + """Separate environment listing from a bounded literal command wrapper.""" + + 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: + if _shell_value_has_credential_read(assignment.group(1)): + return True + index += 1 + continue + expansion = re.fullmatch( + r"\$(?:([A-Za-z_][A-Za-z0-9_]*)|\{([A-Za-z_][A-Za-z0-9_]*)\})", + argument, + ) + if expansion: + name = expansion.group(1) or expansion.group(2) + if name in assignment_expansions: + 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, + assignment_expansions=assignment_expansions, + ) + + def _shell_segment_has_credential_probe( tokens: tuple[str, ...], *, depth: int = 0, + assignment_expansions: frozenset[str] = frozenset(), ) -> bool: if depth >= 8: return True 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, + assignment_expansions=assignment_expansions, + ) + 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, + assignment_expansions=assignment_expansions, ) if executable == "xargs": wrapped = _xargs_command(arguments) return bool(wrapped) and _shell_segment_has_credential_probe( - wrapped, depth=depth + 1 + wrapped, + depth=depth + 1, + assignment_expansions=assignment_expansions, ) if any(_PROC_ENVIRON_PATTERN.search(argument) for argument in arguments): return True @@ -585,8 +787,11 @@ def _shell_segment_has_credential_probe( def _shell_command_has_credential_probe(command: str) -> bool: + assignment_expansions = _proven_assignment_expansions(command) return any( - _shell_segment_has_credential_probe(segment) + _shell_segment_has_credential_probe( + segment, assignment_expansions=assignment_expansions + ) for segment in _shell_segments(command) ) diff --git a/tests/capabilities/test_benchmark_toolkit.py b/tests/capabilities/test_benchmark_toolkit.py index 247ace13da..a18ec0c58f 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,76 @@ 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", "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" + ), + False, + ), + ( + "cmd", + ( + "for spec in 'base:' 'fast:MODE=fast TRIAL=2'; do " + "settings=${spec#*:}; env APPROACH=trial $settings pytest -q; " + "done" + ), + False, + ), + ( + "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 +2273,34 @@ 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( ("orchestrator_input", "expected_probe"), [