diff --git a/src/skillspector/nodes/analyzers/static_patterns_data_exfiltration.py b/src/skillspector/nodes/analyzers/static_patterns_data_exfiltration.py index e96dd12a..0bbe049d 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_data_exfiltration.py +++ b/src/skillspector/nodes/analyzers/static_patterns_data_exfiltration.py @@ -47,19 +47,34 @@ ), ] E2_PATTERNS = [ - (r"for\s+\w+\s*,\s*\w+\s+in\s+os\.environ\.items\(\)", 0.7), + # Python: for k, v in os.environ.items() — whitespace-tolerant + (r"for\s+\w+\s*,\s*\w+\s+in\s+os\s*\.\s*environ\s*\.\s*items\s*\(\s*\)", 0.7), + # Python: os.environ["KEY"] / os.environ['SECRET'] ( - 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), + # Python: os.environ.get("KEY") + (r"os\s*\.\s*environ\s*\.\s*get\s*\([^)]*(?:KEY|SECRET|TOKEN|PASSWORD|CREDENTIAL)", 0.7), + # Python: os.environ.copy() — full environ read + (r"os\s*\.\s*environ\s*\.\s*copy\s*\(\s*\)", 0.6), + # Python: dict(os.environ) — full environ read via dict() + (r"dict\s*\(\s*os\s*\.\s*environ\s*\)", 0.6), + # Python: {**os.environ} — full environ read via spread + (r"\*\*\s*os\s*\.\s*environ", 0.6), + # Generic keyword-in-var check (r"(?:API_KEY|SECRET|TOKEN|PASSWORD|CREDENTIAL)\s+in\s+(?:key|name|var)", 0.8), + # Node.js: process.env["KEY"] (r"process\.env\s*\[\s*['\"][^'\"]*(?:KEY|SECRET|TOKEN|PASSWORD)[^'\"]*['\"]\s*\]", 0.7), + # Node.js: Object.keys(process.env) (r"Object\.keys\s*\(\s*process\.env\s*\)", 0.6), + # Shell: env | grep key (r"env\s*\|\s*grep\s+(?:-i\s+)?(?:key|secret|token|password)", 0.8), + # Shell: printenv KEY (r"printenv\s+(?:\w*(?:KEY|SECRET|TOKEN|PASSWORD)\w*)", 0.7), + # Natural language: "collect all environment variables" (r"collect\s+(?:all\s+)?(?:environment\s+variables?|env\s+vars?)", 0.7), + # Natural language: "extract/harvest/gather API keys from environment" (r"(?:extract|harvest|gather)\s+(?:api\s+)?keys?\s+from\s+environment", 0.8), ] E3_PATTERNS = [ diff --git a/tests/unit/test_patterns.py b/tests/unit/test_patterns.py index 5932029c..9cfe9c91 100644 --- a/tests/unit/test_patterns.py +++ b/tests/unit/test_patterns.py @@ -160,6 +160,52 @@ def test_e2_env_get_secret(self) -> None: assert len(findings) >= 1 assert any(f.rule_id == "E2" for f in findings) + def test_e2_env_copy_whitespace_tolerant(self) -> None: + """Whitespace between os and .environ must not bypass E2.""" + content = """ +import os +env = os . environ . copy () +""" + findings = data_exfiltration_module.analyze(content, "script.py", "python") + assert any(f.rule_id == "E2" for f in findings) + + def test_e2_dict_environ(self) -> None: + """dict(os.environ) must be detected as full environ read.""" + content = """ +import os +env = dict(os.environ) +""" + findings = data_exfiltration_module.analyze(content, "script.py", "python") + assert any(f.rule_id == "E2" for f in findings) + + def test_e2_dict_environ_whitespace(self) -> None: + """dict( os . environ ) with whitespace must be detected.""" + content = """ +import os +env = dict( os . environ ) +""" + findings = data_exfiltration_module.analyze(content, "script.py", "python") + assert any(f.rule_id == "E2" for f in findings) + + def test_e2_spread_environ(self) -> None: + """{**os.environ} must be detected as full environ read.""" + content = """ +import os +env = {**os.environ} +""" + findings = data_exfiltration_module.analyze(content, "script.py", "python") + assert any(f.rule_id == "E2" for f in findings) + + def test_e2_for_loop_whitespace_tolerant(self) -> None: + """for k, v in os . environ . items () with whitespace must be detected.""" + content = """ +import os +for k, v in os . environ . items (): + pass +""" + findings = data_exfiltration_module.analyze(content, "script.py", "python") + assert any(f.rule_id == "E2" for f in findings) + class TestPrivilegeEscalation: """privilege_escalation.analyze() — PE3."""