Skip to content

Commit f74a5f8

Browse files
Jean-Michel Rogeroclaude
andcommitted
Fix persistence scanner false positives with DRY refactor
Problem: _check_tmp_scripts flagged every .py in /tmp (30+ pytest/ Claude temp files). _check_xdg_autostart flagged every .desktop (Twake, etc.). Neither checked relevance to the threat. Fix 1 — /tmp scripts: AST-verified imports via scan_python_imports(). Only flags .py files that actually import the package. String mentions and unrelated scripts are ignored. Shell scripts checked for active (non-comment) references. Fix 2 — XDG/systemd/LaunchAgents: DRY _check_config_dir() helper replaces three near-identical functions. All now filter by package content. Twake-Desktop.desktop no longer flagged. 355 tests pass (16 persistence tests: 7 updated, 9 unchanged). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
1 parent 5152dd4 commit f74a5f8

2 files changed

Lines changed: 224 additions & 78 deletions

File tree

scan_supply_chain/persistence_scanner.py

Lines changed: 94 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22
33
Checks common persistence mechanisms that any supply chain attack
44
might abuse, independent of the specific threat profile.
5+
Every checker filters by the target package name — no generic noise.
56
"""
67

78
from __future__ import annotations
@@ -25,18 +26,39 @@ def scan_persistence(results: ScanResults, package: str) -> None:
2526

2627
_check_crontab(results, package)
2728
_check_shell_rc(results, package)
28-
_check_tmp_scripts(results)
29+
_check_tmp_scripts(results, package)
2930

3031
if sys.platform == "linux":
31-
_check_systemd_user(results, package)
32-
_check_xdg_autostart(results)
32+
_check_config_dir(
33+
results,
34+
Path.home() / ".config" / "systemd" / "user",
35+
"*.service",
36+
"systemd user service",
37+
package,
38+
)
39+
_check_config_dir(
40+
results,
41+
Path.home() / ".config" / "autostart",
42+
"*.desktop",
43+
"XDG autostart",
44+
package,
45+
)
3346
elif sys.platform == "darwin":
34-
_check_launch_agents(results, package)
47+
_check_config_dir(
48+
results,
49+
Path.home() / "Library" / "LaunchAgents",
50+
"*.plist",
51+
"LaunchAgent",
52+
package,
53+
)
3554

3655
if len(results.findings) == count_before:
3756
print_clean("No suspicious persistence found")
3857

3958

59+
# ── Helpers ─────────────────────────────────────────────────────────────
60+
61+
4062
def _add_persistence(results: ScanResults, description: str, evidence: str) -> None:
4163
print_ioc_found(description)
4264
results.iocs.append(f"persistence:{description}")
@@ -50,6 +72,32 @@ def _add_persistence(results: ScanResults, description: str, evidence: str) -> N
5072
)
5173

5274

75+
def _check_config_dir(
76+
results: ScanResults,
77+
directory: Path,
78+
glob_pattern: str,
79+
label: str,
80+
package: str,
81+
) -> None:
82+
"""Glob a config directory for files mentioning the package."""
83+
if not directory.is_dir():
84+
return
85+
try:
86+
for config_file in directory.glob(glob_pattern):
87+
text = config_file.read_text(errors="ignore")
88+
if package in text:
89+
_add_persistence(
90+
results,
91+
f"{label}: {config_file.name}",
92+
str(config_file),
93+
)
94+
except (PermissionError, OSError):
95+
logger.debug("Cannot read %s", directory)
96+
97+
98+
# ── Individual checkers ─────────────────────────────────────────────────
99+
100+
53101
def _check_crontab(results: ScanResults, package: str) -> None:
54102
if not shutil.which("crontab"):
55103
return
@@ -83,66 +131,62 @@ def _check_shell_rc(results: ScanResults, package: str) -> None:
83131
logger.debug("Cannot read %s", rc_path)
84132

85133

86-
def _check_tmp_scripts(results: ScanResults) -> None:
134+
def _check_tmp_scripts(results: ScanResults, package: str) -> None:
135+
"""Check /tmp for scripts that actually import the package."""
87136
tmp = Path("/tmp") if sys.platform != "win32" else None
88137
if tmp is None or not tmp.is_dir():
89138
return
90139
try:
91140
for f in tmp.iterdir():
92-
if f.is_file() and f.suffix in (".py", ".sh", ".bash"):
93-
_add_persistence(
94-
results,
95-
f"/tmp script: {f.name}",
96-
str(f),
97-
)
141+
if not f.is_file():
142+
continue
143+
if f.suffix == ".py":
144+
_check_tmp_python_file(results, f, package)
145+
elif f.suffix in (".sh", ".bash"):
146+
_check_tmp_shell_file(results, f, package)
98147
except (PermissionError, OSError):
99148
logger.debug("Cannot read /tmp")
100149

101150

102-
def _check_systemd_user(results: ScanResults, package: str) -> None:
103-
systemd_dir = Path.home() / ".config" / "systemd" / "user"
104-
if not systemd_dir.is_dir():
105-
return
151+
def _check_tmp_python_file(results: ScanResults, path: Path, package: str) -> None:
152+
"""Flag a /tmp .py file only if it actually imports the package."""
106153
try:
107-
for service_file in systemd_dir.glob("*.service"):
108-
text = service_file.read_text(errors="ignore")
109-
if package in text:
110-
_add_persistence(
111-
results,
112-
f"systemd user service: {service_file.name}",
113-
str(service_file),
114-
)
154+
text = path.read_text(errors="ignore")
115155
except (PermissionError, OSError):
116-
logger.debug("Cannot read systemd user dir")
117-
156+
return
118157

119-
def _check_xdg_autostart(results: ScanResults) -> None:
120-
autostart = Path.home() / ".config" / "autostart"
121-
if not autostart.is_dir():
158+
if package not in text:
122159
return
123-
try:
124-
for desktop_file in autostart.glob("*.desktop"):
125-
_add_persistence(
126-
results,
127-
f"XDG autostart: {desktop_file.name}",
128-
str(desktop_file),
129-
)
130-
except (PermissionError, OSError):
131-
logger.debug("Cannot read autostart dir")
132160

161+
from .ast_scanner import scan_python_imports
133162

134-
def _check_launch_agents(results: ScanResults, package: str) -> None:
135-
agents_dir = Path.home() / "Library" / "LaunchAgents"
136-
if not agents_dir.is_dir():
137-
return
163+
lines = text.splitlines()
164+
ast_refs = scan_python_imports(text, lines, package, str(path))
165+
166+
if ast_refs is not None:
167+
# AST parsed successfully — trust its result
168+
if ast_refs:
169+
_add_persistence(results, f"/tmp script: {path.name}", str(path))
170+
else:
171+
# SyntaxError fallback — check non-comment lines
172+
if _has_active_reference(text, package):
173+
_add_persistence(results, f"/tmp script: {path.name}", str(path))
174+
175+
176+
def _check_tmp_shell_file(results: ScanResults, path: Path, package: str) -> None:
177+
"""Flag a /tmp shell script only if it references the package."""
138178
try:
139-
for plist in agents_dir.glob("*.plist"):
140-
text = plist.read_text(errors="ignore")
141-
if package in text:
142-
_add_persistence(
143-
results,
144-
f"LaunchAgent: {plist.name}",
145-
str(plist),
146-
)
179+
text = path.read_text(errors="ignore")
147180
except (PermissionError, OSError):
148-
logger.debug("Cannot read LaunchAgents")
181+
return
182+
183+
if _has_active_reference(text, package):
184+
_add_persistence(results, f"/tmp script: {path.name}", str(path))
185+
186+
187+
def _has_active_reference(text: str, package: str) -> bool:
188+
"""Check if any non-comment line contains the package name."""
189+
return any(
190+
package in line and not line.strip().startswith("#")
191+
for line in text.splitlines()
192+
)

0 commit comments

Comments
 (0)