From 1a34c1788b4e5cb716a7673f5a0517f8f721b9d9 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Sun, 8 Nov 2026 03:42:11 +0800 Subject: [PATCH 1/3] fix(benchmark): distinguish env command wrappers Signed-off-by: huangruiteng --- .../benchmark_toolkit/environment_access.py | 219 +++++++++++++++++- tests/capabilities/test_benchmark_toolkit.py | 101 ++++++++ 2 files changed, 313 insertions(+), 7 deletions(-) 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"), [ From 813ef6c60ad8595862519ceb01b5918d51eeb66e Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 8 Nov 2026 17:04:37 +0800 Subject: [PATCH 2/3] fix(benchmark): bind env assignment proofs at use sites Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../benchmark_toolkit/environment_access.py | 220 +++++++++++------- tests/capabilities/test_benchmark_toolkit.py | 205 ++++++++++++++++ 2 files changed, 342 insertions(+), 83 deletions(-) diff --git a/loopx/capabilities/benchmark_toolkit/environment_access.py b/loopx/capabilities/benchmark_toolkit/environment_access.py index 3e2aa0650c..e69df024b8 100644 --- a/loopx/capabilities/benchmark_toolkit/environment_access.py +++ b/loopx/capabilities/benchmark_toolkit/environment_access.py @@ -5,7 +5,7 @@ import ast import re import shlex -from collections.abc import Mapping +from collections.abc import Iterator, Mapping from typing import Any _SENSITIVE_ENV_NAME_PATTERN = re.compile( @@ -259,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: @@ -544,7 +545,7 @@ def _xargs_command(arguments: tuple[str, ...]) -> tuple[str, ...]: 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: + if any(character in value for character in "$`*?["): return False try: words = shlex.split(value) @@ -556,67 +557,114 @@ def _assignment_words(value: str) -> bool: ) -def _proven_assignment_expansions(command: str) -> frozenset[str]: - """Find bounded loop variables proven to expand only to assignments. +def _unwrapped_shell_command(tokens: tuple[str, ...]) -> tuple[str, tuple[str, ...]]: + executable, arguments = _shell_executable(tokens) + while executable in {"command", "builtin"} and arguments: + if executable == "command": + wrapped = _shell_wrapper_command(executable, arguments) + else: + wrapped = arguments[1:] if arguments[0] == "--" else arguments + executable, arguments = _shell_executable(wrapped) + return executable, arguments + - 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. +def _shell_segments_with_assignment_proofs( + segments: tuple[tuple[str, ...], ...], +) -> Iterator[tuple[tuple[str, ...], frozenset[str]]]: + """Expose proofs only after assignment inside one simple literal loop. + + This is deliberately not a shell interpreter. Branches, nested loops, + pipelines, background work and evaluation with side effects get no proof. + A loop's binding is established anew on every iteration; proofs cannot + enter from its prefix or escape past ``done``. Mutation invalidates both + source and derived facts before a subsequent use can rely on them. """ - assignment_variables: set[str] = set() - assignment_suffix_variables: set[str] = set() - segments = _shell_segments(command) - for segment_index, segment in enumerate(segments): + loop_starts = [i for i, part in enumerate(segments) if part[0] == "for"] + body_starts = [i for i, part in enumerate(segments) if part[0] == "do"] + body_ends = [i for i, part in enumerate(segments) if part[0] == "done"] + # ``do command`` executes the first body command in the same segment. + commands = tuple( + part[1:] if part[0] in {"do", "done"} else part for part in segments + ) + unsupported = any( + part[0] + in { + ".", + "eval", + "source", + "if", + "then", + "else", + "elif", + "fi", + "while", + "until", + "case", + "esac", + "select", + "function", + "{", + "}", + } + or part[0].startswith("(") + or any(token in {"()", "((", "))", "!"} for token in part) + or any("$(" in token or "`" in token for token in part) + or any(re.search(r"\$\{[^}]*=", token) for token in part) + or (part[0] and part[0][0] in ";&|\n" and any(c not in ";\n" for c in part[0])) + # Aliases and asynchronous traps can change an otherwise proven name. + or _unwrapped_shell_command(part)[0] in {"trap", "declare", "typeset", "local"} + for part in commands + if part + ) + body_start = body_end = -1 + loop_variable = "" + values: list[str] = [] + if not unsupported and len(loop_starts) == len(body_starts) == len(body_ends) == 1: + loop_start, body_start, body_end = ( + loop_starts[0], + body_starts[0], + body_ends[0], + ) + header = segments[loop_start] 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" + loop_start < body_start < body_end + and len(header) >= 3 + and re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", header[1]) + and header[2] == "in" ): + loop_variable = header[1] + values = list(header[3:]) + for part in segments[loop_start + 1 : body_start]: + if not all(c in ";\n" for c in part[0]): + values.extend(part) + if any(c in value for value in values for c in "$`*?[{}~()<>"): + values = [] + loop_variable = "" + + assignments: set[str] = set() + suffixes: set[str] = set() + for index, part in enumerate(segments): + if index == body_start and loop_variable and values: + if all(_assignment_words(value) for value in values): + assignments.add(loop_variable) + if all( + ":" in value and _assignment_words(value.split(":", 1)[1]) + for value in values + ): + suffixes.add(loop_variable) + if index == body_end: + assignments.clear() + suffixes.clear() + tokens = commands[index] + yield tokens, frozenset(assignments) + if not (loop_variable and body_start <= index < body_end): 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) + executable, arguments = _unwrapped_shell_command(tokens) if executable in { + ".", + "eval", + "source", "declare", "export", "local", @@ -624,35 +672,38 @@ def _proven_assignment_expansions(command: str) -> frozenset[str]: "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: + "getopts", + "mapfile", + "readarray", + "let", + } or (executable == "printf" and "-v" in arguments): + assignments.clear() + suffixes.clear() + for token in tokens: assignment = re.fullmatch( - r"([A-Za-z_][A-Za-z0-9_]*)=(.*)", token, flags=re.DOTALL + 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: + variable, subscript, append, value = assignment.groups() + if subscript or append: + assignments.clear() + suffixes.clear() 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) + proven_value = _assignment_words(value) or ( + suffix is not None and suffix.group(1) in suffixes + ) + assignments.discard(variable) + if variable in suffixes: + assignments.clear() + suffixes.discard(variable) + # Prefix assignments on a utility invocation do not persist, and + # argv expansion sees the old binding rather than this new value. + if not executable and proven_value: + assignments.add(variable) def _shell_value_has_credential_read(value: str) -> bool: @@ -678,6 +729,10 @@ def _env_command_has_credential_probe( ) -> 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] @@ -707,8 +762,6 @@ def _env_command_has_credential_probe( 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( @@ -787,12 +840,13 @@ 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, assignment_expansions=assignment_expansions ) - for segment in _shell_segments(command) + for segment, assignment_expansions in _shell_segments_with_assignment_proofs( + _shell_segments(command) + ) ) diff --git a/tests/capabilities/test_benchmark_toolkit.py b/tests/capabilities/test_benchmark_toolkit.py index a18ec0c58f..1818147efe 100644 --- a/tests/capabilities/test_benchmark_toolkit.py +++ b/tests/capabilities/test_benchmark_toolkit.py @@ -2148,6 +2148,24 @@ def test_bare_sensitive_filename_does_not_match_unrelated_path_basename() -> Non ("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), @@ -2175,6 +2193,49 @@ def test_bare_sensitive_filename_does_not_match_unrelated_path_basename() -> Non ), False, ), + ( + "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", ( @@ -2301,6 +2362,150 @@ def test_env_wrapper_classification_matches_isolated_process( assert receipt["evidence_counts"]["credential_probe"] == int(reads_environment) +@pytest.mark.parametrize( + ("command", "expected_probe"), + [ + ('env MODE=fast printf %s "$API_KEY"', True), + ('env MODE=fast printf %s "$(printenv API_KEY)"', True), + ( + "for settings in 'MODE=fast'; do trap 'settings=printenv' DEBUG; " + "settings='MODE=fast'; env $settings API_KEY; done", + True, + ), + ( + "for settings in 'MODE=fast'; do settings+=' printenv'; " + "env $settings API_KEY; done", + True, + ), + ( + "for spec in 'fast:MODE=fast'; do spec[0]='probe:printenv'; " + "settings=${spec#*:}; env $settings API_KEY; done", + True, + ), + ( + "for spec in 'fast:MODE=fast'; do command export spec='probe:printenv'; " + "settings=${spec#*:}; env $settings API_KEY; done", + True, + ), + ( + "for spec in {fast:MODE=fast,printenv}; do settings=${spec#*:}; " + "env $settings API_KEY; done", + True, + ), + ( + "env $settings API_KEY; for settings in 'MODE=fast'; do true; done", + True, + ), + ( + "for spec in 'fast:MODE=fast'; do spec='probe:printenv'; " + "settings=${spec#*:}; env $settings API_KEY; done", + True, + ), + ( + "for spec in 'fast:MODE=fast'; do settings=${spec#*:}; " + "spec='probe:printenv'; settings=${spec#*:}; env $settings API_KEY; done", + True, + ), + ( + "for settings in 'MODE=fast'; do env $settings printenv API_KEY; done", + True, + ), + ( + "for spec in 'fast:MODE=fast'; do false && settings=${spec#*:}; " + "env $settings API_KEY; done", + True, + ), + ( + "for settings in 'MODE=fast'; do env $settings printf ordinary-output; done", + False, + ), + ( + "for spec in 'fast:MODE=fast'; do settings=${spec#*:}; " + "env $settings printf ordinary-output; done", + False, + ), + ], +) +def test_env_assignment_proof_matches_shell_and_source_cli( + command: str, + 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 expected_probe 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"), [ From ffd707491d112b2ae6a64b275cc0abdce5bcfe11 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Tue, 10 Nov 2026 11:47:10 +0800 Subject: [PATCH 3/3] fix(benchmark): remove unsound dynamic shell assignment proofs Signed-off-by: huangruiteng --- .../benchmark_toolkit/environment_access.py | 191 +----------------- tests/capabilities/test_benchmark_toolkit.py | 44 ++-- 2 files changed, 30 insertions(+), 205 deletions(-) diff --git a/loopx/capabilities/benchmark_toolkit/environment_access.py b/loopx/capabilities/benchmark_toolkit/environment_access.py index e69df024b8..56429ed825 100644 --- a/loopx/capabilities/benchmark_toolkit/environment_access.py +++ b/loopx/capabilities/benchmark_toolkit/environment_access.py @@ -5,7 +5,7 @@ import ast import re import shlex -from collections.abc import Iterator, Mapping +from collections.abc import Mapping from typing import Any _SENSITIVE_ENV_NAME_PATTERN = re.compile( @@ -542,170 +542,6 @@ 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 any(character in value for character in "$`*?["): - 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 _unwrapped_shell_command(tokens: tuple[str, ...]) -> tuple[str, tuple[str, ...]]: - executable, arguments = _shell_executable(tokens) - while executable in {"command", "builtin"} and arguments: - if executable == "command": - wrapped = _shell_wrapper_command(executable, arguments) - else: - wrapped = arguments[1:] if arguments[0] == "--" else arguments - executable, arguments = _shell_executable(wrapped) - return executable, arguments - - -def _shell_segments_with_assignment_proofs( - segments: tuple[tuple[str, ...], ...], -) -> Iterator[tuple[tuple[str, ...], frozenset[str]]]: - """Expose proofs only after assignment inside one simple literal loop. - - This is deliberately not a shell interpreter. Branches, nested loops, - pipelines, background work and evaluation with side effects get no proof. - A loop's binding is established anew on every iteration; proofs cannot - enter from its prefix or escape past ``done``. Mutation invalidates both - source and derived facts before a subsequent use can rely on them. - """ - - loop_starts = [i for i, part in enumerate(segments) if part[0] == "for"] - body_starts = [i for i, part in enumerate(segments) if part[0] == "do"] - body_ends = [i for i, part in enumerate(segments) if part[0] == "done"] - # ``do command`` executes the first body command in the same segment. - commands = tuple( - part[1:] if part[0] in {"do", "done"} else part for part in segments - ) - unsupported = any( - part[0] - in { - ".", - "eval", - "source", - "if", - "then", - "else", - "elif", - "fi", - "while", - "until", - "case", - "esac", - "select", - "function", - "{", - "}", - } - or part[0].startswith("(") - or any(token in {"()", "((", "))", "!"} for token in part) - or any("$(" in token or "`" in token for token in part) - or any(re.search(r"\$\{[^}]*=", token) for token in part) - or (part[0] and part[0][0] in ";&|\n" and any(c not in ";\n" for c in part[0])) - # Aliases and asynchronous traps can change an otherwise proven name. - or _unwrapped_shell_command(part)[0] in {"trap", "declare", "typeset", "local"} - for part in commands - if part - ) - body_start = body_end = -1 - loop_variable = "" - values: list[str] = [] - if not unsupported and len(loop_starts) == len(body_starts) == len(body_ends) == 1: - loop_start, body_start, body_end = ( - loop_starts[0], - body_starts[0], - body_ends[0], - ) - header = segments[loop_start] - if ( - loop_start < body_start < body_end - and len(header) >= 3 - and re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", header[1]) - and header[2] == "in" - ): - loop_variable = header[1] - values = list(header[3:]) - for part in segments[loop_start + 1 : body_start]: - if not all(c in ";\n" for c in part[0]): - values.extend(part) - if any(c in value for value in values for c in "$`*?[{}~()<>"): - values = [] - loop_variable = "" - - assignments: set[str] = set() - suffixes: set[str] = set() - for index, part in enumerate(segments): - if index == body_start and loop_variable and values: - if all(_assignment_words(value) for value in values): - assignments.add(loop_variable) - if all( - ":" in value and _assignment_words(value.split(":", 1)[1]) - for value in values - ): - suffixes.add(loop_variable) - if index == body_end: - assignments.clear() - suffixes.clear() - tokens = commands[index] - yield tokens, frozenset(assignments) - if not (loop_variable and body_start <= index < body_end): - continue - executable, arguments = _unwrapped_shell_command(tokens) - if executable in { - ".", - "eval", - "source", - "declare", - "export", - "local", - "read", - "readonly", - "typeset", - "unset", - "getopts", - "mapfile", - "readarray", - "let", - } or (executable == "printf" and "-v" in arguments): - assignments.clear() - suffixes.clear() - for token in tokens: - assignment = re.fullmatch( - r"([A-Za-z_][A-Za-z0-9_]*)(\[[^]]*\])?(\+)?=(.*)", - token, - flags=re.DOTALL, - ) - if assignment is None: - break - variable, subscript, append, value = assignment.groups() - if subscript or append: - assignments.clear() - suffixes.clear() - continue - suffix = re.fullmatch(r"\$\{([A-Za-z_][A-Za-z0-9_]*)#\*:\}", value) - proven_value = _assignment_words(value) or ( - suffix is not None and suffix.group(1) in suffixes - ) - assignments.discard(variable) - if variable in suffixes: - assignments.clear() - suffixes.discard(variable) - # Prefix assignments on a utility invocation do not persist, and - # argv expansion sees the old binding rather than this new value. - if not executable and proven_value: - assignments.add(variable) - - def _shell_value_has_credential_read(value: str) -> bool: if "$(" in value or "`" in value or "${!" in value: return True @@ -725,7 +561,6 @@ 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.""" @@ -764,15 +599,6 @@ def _env_command_has_credential_probe( if assignment: 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): @@ -783,7 +609,6 @@ def _env_command_has_credential_probe( return _shell_segment_has_credential_probe( arguments[index:], depth=depth + 1, - assignment_expansions=assignment_expansions, ) @@ -791,7 +616,6 @@ def _shell_segment_has_credential_probe( tokens: tuple[str, ...], *, depth: int = 0, - assignment_expansions: frozenset[str] = frozenset(), ) -> bool: if depth >= 8: return True @@ -802,7 +626,6 @@ def _shell_segment_has_credential_probe( return _env_command_has_credential_probe( arguments, depth=depth, - assignment_expansions=assignment_expansions, ) if executable == "printenv": return True @@ -811,14 +634,12 @@ def _shell_segment_has_credential_probe( return bool(wrapped) and _shell_segment_has_credential_probe( 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, - assignment_expansions=assignment_expansions, ) if any(_PROC_ENVIRON_PATTERN.search(argument) for argument in arguments): return True @@ -840,13 +661,11 @@ 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, assignment_expansions=assignment_expansions - ) - for segment, assignment_expansions in _shell_segments_with_assignment_proofs( - _shell_segments(command) - ) + _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 1818147efe..aabd5c3216 100644 --- a/tests/capabilities/test_benchmark_toolkit.py +++ b/tests/capabilities/test_benchmark_toolkit.py @@ -2182,7 +2182,7 @@ def test_bare_sensitive_filename_does_not_match_unrelated_path_basename() -> Non "for settings in 'MODE=fast' 'MODE=slow TRIAL=2'; do " "env $settings pytest -q; done" ), - False, + True, ), ( "cmd", @@ -2191,7 +2191,7 @@ def test_bare_sensitive_filename_does_not_match_unrelated_path_basename() -> Non "settings=${spec#*:}; env APPROACH=trial $settings pytest -q; " "done" ), - False, + True, ), ( "cmd", @@ -2363,71 +2363,77 @@ def test_env_wrapper_classification_matches_isolated_process( @pytest.mark.parametrize( - ("command", "expected_probe"), + ("command", "reads_environment", "expected_probe"), [ - ('env MODE=fast printf %s "$API_KEY"', True), - ('env MODE=fast printf %s "$(printenv API_KEY)"', True), + 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, True, ), ( "for settings in 'MODE=fast'; do settings+=' printenv'; " "env $settings API_KEY; done", - True, + True, True, ), ( "for spec in 'fast:MODE=fast'; do spec[0]='probe:printenv'; " "settings=${spec#*:}; env $settings API_KEY; done", - True, + True, True, ), ( "for spec in 'fast:MODE=fast'; do command export spec='probe:printenv'; " "settings=${spec#*:}; env $settings API_KEY; done", - True, + True, True, ), ( "for spec in {fast:MODE=fast,printenv}; do settings=${spec#*:}; " "env $settings API_KEY; done", - True, + True, True, ), ( "env $settings API_KEY; for settings in 'MODE=fast'; do true; done", - True, + True, True, ), ( "for spec in 'fast:MODE=fast'; do spec='probe:printenv'; " "settings=${spec#*:}; env $settings API_KEY; done", - True, + True, True, ), ( "for spec in 'fast:MODE=fast'; do settings=${spec#*:}; " "spec='probe:printenv'; settings=${spec#*:}; env $settings API_KEY; done", - True, + True, True, ), ( "for settings in 'MODE=fast'; do env $settings printenv API_KEY; done", - True, + True, True, ), ( "for spec in 'fast:MODE=fast'; do false && settings=${spec#*:}; " "env $settings API_KEY; done", - True, + True, True, ), ( "for settings in 'MODE=fast'; do env $settings printf ordinary-output; done", - False, + False, True, ), ( "for spec in 'fast:MODE=fast'; do settings=${spec#*:}; " "env $settings printf ordinary-output; done", - False, + False, True, ), ], ) -def test_env_assignment_proof_matches_shell_and_source_cli( +def test_env_dynamic_wrappers_remain_conservative_in_shell_and_cli( command: str, + reads_environment: bool, expected_probe: bool, tmp_path: Path, ) -> None: @@ -2444,7 +2450,7 @@ def test_env_assignment_proof_matches_shell_and_source_cli( check=True, timeout=10, ) - assert result.stdout.strip() == (marker if expected_probe else "ordinary-output") + 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(