Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -58,9 +58,21 @@
),
]
E2_PYTHON_FALLBACK_PATTERNS = [
# 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'] — whitespace-tolerant
(
r"os\s*\.\s*environ\s*\[\s*['\"][^'\"]*(?:KEY|SECRET|TOKEN|PASSWORD|CREDENTIAL)[^'\"]*['\"]\s*\]",
0.8,
),
# Python: os.environ.get("KEY") — whitespace-tolerant
(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 dict-spread.
# Require braces so bare ``2 ** os.environ`` (exponentiation) is not flagged.
(r"\{\s*\*\*\s*os\s*\.\s*environ\s*\}", 0.6),
]
E2_OTHER_PATTERNS = [
Expand All @@ -79,6 +91,8 @@
"items": 0.7,
"keys": 0.6,
"values": 0.6,
"get": 0.7,
"setdefault": 0.6,
}
_ENVIRONMENT_COLLECTION_CALLS = frozenset({"dict", "list", "tuple", "set", "frozenset"})
_ENVIRONMENT_COPY_CALLS = frozenset({"copy.copy", "copy.deepcopy"})
Expand Down Expand Up @@ -182,11 +196,14 @@ def _analyze_python_environment_reads(
) -> 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
targeted single-key lookup or passing ``os.environ`` through to a child process.
Credential flows to network and execution sinks remain covered by the behavioral
taint analyzer. AST parsing makes this check insensitive to formatting and lets it
resolve ``os`` / ``environ`` import aliases.
Detects full mapping copies/enumerations (``items()``, ``keys()``,
``values()``, ``copy()``, ``dict(os.environ)``, ``{**os.environ}``),
single-key lookups (``os.environ['KEY']``, ``os.environ.get('SECRET')``),
and iteration over ``os.environ``. Single-key access is flagged at the same
severity because credential keys are the primary target of env harvesting.
Credential flows to network and execution sinks remain covered by the
behavioral taint analyzer. AST parsing makes this 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. Standalone callers parse through the
Expand Down Expand Up @@ -263,6 +280,10 @@ def emit(node: ast.AST, confidence: float) -> None:
if _is_os_environ_reference(ast_node.iter, aliases):
emit(ast_node.iter, 0.7)

elif isinstance(ast_node, ast.Subscript):
if _is_os_environ_reference(ast_node.value, aliases):
emit(ast_node, 0.7)

return findings


Expand Down
38 changes: 38 additions & 0 deletions tests/nodes/analyzers/test_static_patterns.py
Original file line number Diff line number Diff line change
Expand Up @@ -333,6 +333,44 @@ def test_e2_env_harvesting_produces_finding(self):
e2 = next(f for f in findings if f.rule_id == "E2")
assert e2.severity == "HIGH"

def test_e2_whitespace_tolerant_environ_access(self):
"""Whitespace-obfuscated os.environ access is still detected."""
state = {
"components": ["script.py"],
"file_cache": {
"script.py": "import os\nx = os . environ [ 'API_KEY' ]\ny = os.environ.get('SECRET')",
},
}
findings = static_runner.run_static_patterns(state, [data_exfiltration_module])
e2 = [f for f in findings if f.rule_id == "E2"]
assert len(e2) >= 2

def test_e2_exponentiation_not_flagged(self):
"""Bare ``2 ** os.environ`` (exponentiation) must not be flagged as E2."""
# Malformed Python (triggers regex fallback) with exponentiation
state = {
"components": ["script.py"],
"file_cache": {
"script.py": "import os\nresult = 2 ** os.environ\n def broken(",
},
}
findings = static_runner.run_static_patterns(state, [data_exfiltration_module])
e2 = [f for f in findings if f.rule_id == "E2"]
# Should NOT flag the exponentiation as env harvesting
assert not any("**" in f.matched_text for f in e2)

def test_e2_dict_spread_environ_flagged(self):
"""``{**os.environ}`` (dict spread) is flagged as full environ read."""
state = {
"components": ["script.py"],
"file_cache": {
"script.py": "import os\nenv_copy = {**os.environ}",
},
}
findings = static_runner.run_static_patterns(state, [data_exfiltration_module])
e2 = [f for f in findings if f.rule_id == "E2"]
assert len(e2) >= 1

def test_e5_boto3_put_object_produces_finding(self):
"""boto3 put_object yields E5, MEDIUM severity."""
state = {
Expand Down