diff --git a/src/skillspector/nodes/analyzers/static_patterns_data_exfiltration.py b/src/skillspector/nodes/analyzers/static_patterns_data_exfiltration.py index e96dd12a..ad89a203 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_data_exfiltration.py +++ b/src/skillspector/nodes/analyzers/static_patterns_data_exfiltration.py @@ -17,6 +17,7 @@ from __future__ import annotations +import ast import re import sys @@ -25,7 +26,16 @@ from skillspector.state import AnalyzerNodeResponse, SkillspectorState from . import static_runner -from .common import get_context, get_line_number, is_code_example +from .common import ( + apply_import_aliases, + build_import_aliases, + get_context, + get_context_from_lines, + get_line_number, + is_code_example, + resolve_call_name, + resolve_dotted_name, +) from .pattern_defaults import PatternCategory logger = get_logger(__name__) @@ -46,14 +56,18 @@ 0.7, ), ] -E2_PATTERNS = [ - (r"for\s+\w+\s*,\s*\w+\s+in\s+os\.environ\.items\(\)", 0.7), +E2_PYTHON_FALLBACK_PATTERNS = [ + (r"for\s+\w+\s*,\s*\w+\s+in\s+os\s*\.\s*environ\s*\.\s*items\s*\(\s*\)", 0.7), ( - r"os\.environ\s*\[\s*['\"][^'\"]*(?:KEY|SECRET|TOKEN|PASSWORD|CREDENTIAL)[^'\"]*['\"]\s*\]", + r"os\s*\.\s*environ\s*\[\s*['\"][^'\"]*(?:KEY|SECRET|TOKEN|PASSWORD|CREDENTIAL)[^'\"]*['\"]\s*\]", 0.8, ), - (r"os\.environ\.get\s*\([^)]*(?:KEY|SECRET|TOKEN|PASSWORD|CREDENTIAL)", 0.7), - (r"os\.environ\s*\.\s*copy\s*\(\)", 0.6), + (r"os\s*\.\s*environ\s*\.\s*get\s*\([^)]*(?:KEY|SECRET|TOKEN|PASSWORD|CREDENTIAL)", 0.7), + (r"os\s*\.\s*environ\s*\.\s*copy\s*\(\s*\)", 0.6), + (r"dict\s*\(\s*os\s*\.\s*environ\s*\)", 0.6), + (r"\{\s*\*\*\s*os\s*\.\s*environ\s*\}", 0.6), +] +E2_OTHER_PATTERNS = [ (r"(?:API_KEY|SECRET|TOKEN|PASSWORD|CREDENTIAL)\s+in\s+(?:key|name|var)", 0.8), (r"process\.env\s*\[\s*['\"][^'\"]*(?:KEY|SECRET|TOKEN|PASSWORD)[^'\"]*['\"]\s*\]", 0.7), (r"Object\.keys\s*\(\s*process\.env\s*\)", 0.6), @@ -62,6 +76,17 @@ (r"collect\s+(?:all\s+)?(?:environment\s+variables?|env\s+vars?)", 0.7), (r"(?:extract|harvest|gather)\s+(?:api\s+)?keys?\s+from\s+environment", 0.8), ] +E2_PATTERNS = E2_PYTHON_FALLBACK_PATTERNS + E2_OTHER_PATTERNS + +_ENVIRONMENT_MAPPING_METHOD_CONFIDENCE = { + "copy": 0.6, + "items": 0.7, + "keys": 0.6, + "values": 0.6, +} +_ENVIRONMENT_COLLECTION_CALLS = frozenset({"dict", "list", "tuple", "set", "frozenset"}) +_ENVIRONMENT_COPY_CALLS = frozenset({"copy.copy", "copy.deepcopy"}) +_SENSITIVE_ENV_KEY_PATTERN = re.compile(r"(?:KEY|SECRET|TOKEN|PASSWORD|CREDENTIAL)", re.IGNORECASE) 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), @@ -119,6 +144,146 @@ ] +def _resolve_expression_name(node: ast.expr, aliases: dict[str, str]) -> str | None: + """Resolve a Python expression to its import-normalized dotted name.""" + name = resolve_dotted_name(node) + return apply_import_aliases(name, aliases) if name is not None else None + + +def _is_os_environ_reference(node: ast.expr, aliases: dict[str, str]) -> bool: + """Return whether *node* is ``os.environ``, including imported aliases.""" + return _resolve_expression_name(node, aliases) == "os.environ" + + +def _is_sensitive_environment_key(node: ast.expr) -> bool: + """Return whether a literal environment key looks credential-like.""" + return ( + isinstance(node, ast.Constant) + and isinstance(node.value, str) + and _SENSITIVE_ENV_KEY_PATTERN.search(node.value) is not None + ) + + +def _has_direct_environ_argument(call: ast.Call, aliases: dict[str, str]) -> bool: + """Return whether a call receives ``os.environ`` directly, not via a lookup.""" + return any(_is_os_environ_reference(arg, aliases) for arg in call.args) or any( + keyword.arg is None and _is_os_environ_reference(keyword.value, aliases) + for keyword in call.keywords + ) + + +def _is_dynamic_copy_call(call: ast.Call, aliases: dict[str, str]) -> bool: + """Recognize ``__import__('copy').copy(...)`` without broad call matching.""" + func = call.func + if not isinstance(func, ast.Attribute) or func.attr not in {"copy", "deepcopy"}: + return False + if ( + not isinstance(func.value, ast.Call) + or resolve_call_name(func.value, aliases) != "__import__" + ): + return False + return ( + bool(func.value.args) + and isinstance(func.value.args[0], ast.Constant) + and func.value.args[0].value == "copy" + ) + + +def _analyze_python_environment_reads(content: str, file_path: str) -> list[AnalyzerFinding] | None: + """Detect materializing or enumerating the complete ``os.environ`` mapping. + + A full mapping copy or enumeration is an environment-harvesting signal, unlike a + single-key lookup or passing ``os.environ`` through to a child process. AST parsing + makes the check insensitive to formatting and lets it resolve ``os`` / ``environ`` + import aliases. + + ``None`` means the source could not be parsed, so callers can retain the regex + fallback for malformed Python files. + """ + try: + tree = ast.parse(content, filename=file_path) + except (SyntaxError, ValueError): + return None + + aliases = build_import_aliases(tree) + lines = content.splitlines() + findings: list[AnalyzerFinding] = [] + emitted: set[int] = set() + tag = [PatternCategory.DATA_EXFILTRATION.value] + + def emit(node: ast.AST, confidence: float) -> None: + node_id = id(node) + if node_id in emitted: + return + emitted.add(node_id) + lineno = getattr(node, "lineno", 1) + end_lineno = getattr(node, "end_lineno", None) + matched_text = ast.get_source_segment(content, node) + findings.append( + AnalyzerFinding( + rule_id="E2", + message="Env Variable Harvesting", + severity=Severity.HIGH, + location=Location(file=file_path, start_line=lineno, end_line=end_lineno), + confidence=confidence, + tags=tag, + context=get_context_from_lines(lines, lineno), + matched_text=(matched_text or "os.environ")[:200], + ) + ) + + for ast_node in ast.walk(tree): + if isinstance(ast_node, ast.Call): + call_name = resolve_call_name(ast_node, aliases) + if call_name == "os.environ.get" and ast_node.args: + if _is_sensitive_environment_key(ast_node.args[0]): + emit(ast_node, 0.7) + continue + + if call_name is not None: + method = call_name.rpartition(".")[2] + if ( + call_name.startswith("os.environ.") + and method in _ENVIRONMENT_MAPPING_METHOD_CONFIDENCE + ): + emit(ast_node, _ENVIRONMENT_MAPPING_METHOD_CONFIDENCE[method]) + continue + if call_name in _ENVIRONMENT_COLLECTION_CALLS and _has_direct_environ_argument( + ast_node, aliases + ): + emit(ast_node, 0.6) + continue + if call_name in _ENVIRONMENT_COPY_CALLS and _has_direct_environ_argument( + ast_node, aliases + ): + emit(ast_node, 0.6) + continue + + if _is_dynamic_copy_call(ast_node, aliases) and _has_direct_environ_argument( + ast_node, aliases + ): + emit(ast_node, 0.6) + + elif isinstance(ast_node, ast.Subscript): + if _is_os_environ_reference(ast_node.value, aliases) and _is_sensitive_environment_key( + ast_node.slice + ): + emit(ast_node, 0.8) + + elif isinstance(ast_node, ast.Dict): + if any( + key is None and _is_os_environ_reference(value, aliases) + for key, value in zip(ast_node.keys, ast_node.values, strict=True) + ): + emit(ast_node, 0.6) + + elif isinstance(ast_node, (ast.For, ast.AsyncFor, ast.comprehension)): + if _is_os_environ_reference(ast_node.iter, aliases): + emit(ast_node.iter, 0.7) + + return findings + + def analyze(content: str, file_path: str, file_type: str) -> list[AnalyzerFinding]: """Analyze content for data exfiltration patterns (E1–E5).""" findings: list[AnalyzerFinding] = [] @@ -151,7 +316,16 @@ def ctx(start: int) -> str: matched_text=match.group(0)[:200], ) ) - for pattern, confidence in E2_PATTERNS: + e2_patterns = E2_PATTERNS + if file_type == "python": + python_e2_findings = _analyze_python_environment_reads(content, file_path) + if python_e2_findings is None: + logger.debug("Using E2 regex fallback for unparsable Python file: %s", file_path) + else: + findings.extend(python_e2_findings) + e2_patterns = E2_OTHER_PATTERNS + + for pattern, confidence in e2_patterns: for match in re.finditer(pattern, content, re.IGNORECASE | re.MULTILINE): line_num = get_line_number(content, match.start()) findings.append( diff --git a/tests/unit/test_patterns.py b/tests/unit/test_patterns.py index 5932029c..7502c8dc 100644 --- a/tests/unit/test_patterns.py +++ b/tests/unit/test_patterns.py @@ -160,6 +160,65 @@ def test_e2_env_get_secret(self) -> None: assert len(findings) >= 1 assert any(f.rule_id == "E2" for f in findings) + @pytest.mark.parametrize( + "expression", + [ + "os.environ.copy()", + "dict(os.environ)", + "{**os.environ}", + "dict(os.environ.items())", + '__import__("copy").copy(os.environ)', + "os . environ . copy ()", + ], + ) + def test_e2_full_environment_read_forms(self, expression: str) -> None: + """Materializing the whole environment is detected independently of spelling.""" + content = f"import os\nresult = {expression}\n" + + findings = data_exfiltration_module.analyze(content, "script.py", "python") + e2 = [finding for finding in findings if finding.rule_id == "E2"] + + assert len(e2) == 1 + assert e2[0].location.start_line == 2 + + @pytest.mark.parametrize( + ("imports", "expression", "expected_line"), + [ + ("import os as operating_system", "operating_system.environ.copy()", 2), + ("from os import environ as environment", "dict(environment)", 2), + ("import copy as copier\nimport os", "copier.copy(os.environ)", 3), + ], + ) + def test_e2_full_environment_read_import_aliases( + self, imports: str, expression: str, expected_line: int + ) -> None: + """Import aliases cannot hide a full environment copy or enumeration.""" + content = f"{imports}\nresult = {expression}\n" + + findings = data_exfiltration_module.analyze(content, "script.py", "python") + e2 = [finding for finding in findings if finding.rule_id == "E2"] + + assert len(e2) == 1 + assert e2[0].location.start_line == expected_line + + @pytest.mark.parametrize( + "expression", + [ + 'os.environ["PATH"]', + 'os.environ.get("PATH")', + "os.environ.copy", + "2 ** os.environ", + "subprocess.run(command, env=os.environ, check=False)", + ], + ) + def test_e2_does_not_flag_non_harvesting_environment_use(self, expression: str) -> None: + """Single-key access and process environment plumbing are not harvesting.""" + content = f"import os\nresult = {expression}\n" + + findings = data_exfiltration_module.analyze(content, "script.py", "python") + + assert not any(finding.rule_id == "E2" for finding in findings) + class TestPrivilegeEscalation: """privilege_escalation.analyze() — PE3."""