DRY: Clean Code refactoring — eliminate duplication across scanners, platforms, and tests

- Extract BasePlatformPolicy with shared home_conda_dirs() and _first_existing_dir()
- Unify 3 cache walkers into _scan_cache_dir() with configurable search targets
- Add subprocess_utils.run_safe() eliminating 4 identical try/except blocks
- Add config.read_if_contains() replacing repeated read-then-check patterns
- Add scanner_check() context manager combining header + findings tracking
- Inline trivial _add_persistence() and _add_cache_finding() wrappers
- Name _SEPARATOR_WIDTH constant in formatting.py
- Add shared test helpers: mock_run_safe, mock_subprocess_run, mock_tool_available
- Bump version to 0.8.1

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: Jean-Michel Rogero

CHANGELOG.md

@@ -1,5 +1,35 @@
 # Changelog
 
+## 0.8.1 — 2026-04-02
+
+### Changed
+- **DRY: Platform base class** — `BasePlatformPolicy` provides default `home_conda_dirs()` and `_first_existing_dir()` helper. Linux and Darwin inherit defaults; Windows overrides casing.
+- **DRY: Cache scanner** — Three near-identical cache walkers (`pip`, `npm`, `pnpm`) unified into `_scan_cache_dir()` with configurable search targets.
+- **DRY: Subprocess helper** — New `subprocess_utils.run_safe()` replaces 4 identical `subprocess.run` try/except blocks across `persistence_scanner`, `ioc_windows`.
+- **DRY: File read helper** — New `config.read_if_contains()` replaces repeated read-then-check patterns in `persistence_scanner`, `history_scanner`.
+- **DRY: Scanner boilerplate** — New `scanner_check()` context manager combines `print_check_header()` + `track_findings()` into a single call.
+- **DRY: Inlined wrappers** — Removed trivial `_add_persistence()` and `_add_cache_finding()` one-liner delegates.
+- **DRY: Test helpers** — Shared `mock_run_safe`, `mock_subprocess_run`, `mock_tool_available`, `scan_results` fixture in `conftest.py`.
+- Named magic constant `_SEPARATOR_WIDTH` in `formatting.py`.
+- 355 tests, all passing.
+
+## 0.8.0 — 2026-04-01
+
+### Added
+- **Evidence scoring** — 4-tier confidence (LOW/MEDIUM/HIGH/CRITICAL) replaces binary clean/compromised verdict. New `Confidence` enum, `FindingCategory` enum, `Finding` dataclass, and `scoring.py` module.
+- **AST-based Python detection** — `ast_scanner.py` uses `ast.parse()` to find real imports and attribute access. Eliminates false positives from string literals, regex patterns, and comments. Falls back to regex on `SyntaxError`.
+- **Structured socket parsing** — `network_scanner.py` parses `ss`/`lsof` output into typed `ConnectionRecord` structs. C2 detection reports process name, PID, and executable path (Linux `/proc` enrichment).
+- **Persistence scanner** — `persistence_scanner.py` checks crontab, shell rc files, systemd user services, XDG autostart, `/tmp` scripts, and macOS LaunchAgents.
+- **Cache scanner** — `cache_scanner.py` checks pip, npm, and pnpm caches for traces of compromised packages. Ecosystem-gated.
+- **History scanner** — `history_scanner.py` searches `.bash_history` and `.zsh_history` for `pip install`/`npm install`/`yarn add`/`pnpm add` commands.
+- **Lockfile version extraction** — `yarn.lock` and `pnpm-lock.yaml` phantom dep reports now include the resolved version (e.g., `phantom:plain-crypto-js@4.2.1`).
+- 62 new tests. **348 tests total.**
+
+### Changed
+- C2 connection check uses structured parsing instead of IP substring matching.
+- Python source scanning uses AST first, regex as fallback (no change for JS/TS).
+- `ScanResults` gains a `findings` list alongside existing `iocs` (backward compatible).
+
 ## 0.7.0 — 2026-04-01
 
 ### Performance

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "scan-supply-chain"
-version = "0.7.0"
+version = "0.8.1"
 description = "Supply chain compromise scanner — detects known PyPI and npm attacks via data-driven threat profiles"
 readme = "README.md"
 license = "MIT"

scan_supply_chain/__init__.py

@@ -4,4 +4,4 @@
 Threat profiles are defined in threats/*.toml and are user-extensible.
 """
 
-__version__ = "0.7.0"
+__version__ = "0.8.1"

scan_supply_chain/cache_scanner.py

@@ -7,25 +7,22 @@
 import sys
 from pathlib import Path
 
-from .formatting import print_check_header
-from .models import FindingCategory, ScanResults, track_findings
+from .models import FindingCategory, ScanResults, scanner_check
 
 logger = logging.getLogger(__name__)
 
 
 def scan_caches(results: ScanResults, package: str, ecosystem: str) -> None:
     """Check package manager caches for traces of the compromised package."""
-    print_check_header("package manager caches")
-    with track_findings(results, "No cache traces found"):
+    with scanner_check(results, "package manager caches", "No cache traces found"):
         if ecosystem == "pypi":
             _scan_pip_cache(results, package)
         elif ecosystem == "npm":
             _scan_npm_cache(results, package)
             _scan_pnpm_store(results, package)
 
 
-def _add_cache_finding(results: ScanResults, description: str, evidence: str) -> None:
-    results.add_finding(FindingCategory.CACHE_TRACE, description, evidence, 1)
+# ── Helpers ─────────────────────────────────────────────────────────────
 
 
 def _pip_cache_dir() -> Path:
@@ -39,55 +36,55 @@ def _pip_cache_dir() -> Path:
     return Path.home() / ".cache" / "pip"
 
 
-def _scan_pip_cache(results: ScanResults, package: str) -> None:
-    cache_dir = _pip_cache_dir()
+def _scan_cache_dir(
+    results: ScanResults,
+    cache_dir: Path,
+    package: str,
+    label: str,
+    *,
+    check_dirs: bool = False,
+    check_files: bool = True,
+) -> None:
+    """Walk a cache directory for entries matching the package name."""
     if not cache_dir.is_dir():
         return
     try:
         for dirpath, dirnames, filenames in os.walk(cache_dir):
-            for name in dirnames + filenames:
+            items: list[str] = []
+            if check_dirs:
+                items.extend(dirnames)
+            if check_files:
+                items.extend(filenames)
+            for name in items:
                 if package in name.lower():
-                    _add_cache_finding(
-                        results,
-                        f"pip cache: {name}",
+                    results.add_finding(
+                        FindingCategory.CACHE_TRACE,
+                        f"{label}: {name}",
                         os.path.join(dirpath, name),
+                        1,
                     )
                     return  # one hit per cache is enough
     except (PermissionError, OSError):
-        logger.debug("Cannot read pip cache at %s", cache_dir)
+        logger.debug("Cannot read %s at %s", label, cache_dir)
+
+
+def _scan_pip_cache(results: ScanResults, package: str) -> None:
+    _scan_cache_dir(
+        results, _pip_cache_dir(), package, "pip cache",
+        check_dirs=True, check_files=True,
+    )
 
 
 def _scan_npm_cache(results: ScanResults, package: str) -> None:
-    cache_dir = Path.home() / ".npm" / "_cacache"
-    if not cache_dir.is_dir():
-        return
-    try:
-        for dirpath, _, filenames in os.walk(cache_dir):
-            for fn in filenames:
-                if package in fn:
-                    _add_cache_finding(
-                        results,
-                        f"npm cache: {fn}",
-                        os.path.join(dirpath, fn),
-                    )
-                    return
-    except (PermissionError, OSError):
-        logger.debug("Cannot read npm cache")
+    _scan_cache_dir(
+        results, Path.home() / ".npm" / "_cacache", package, "npm cache",
+    )
 
 
 def _scan_pnpm_store(results: ScanResults, package: str) -> None:
-    store = Path.home() / ".local" / "share" / "pnpm" / "store"
-    if not store.is_dir():
-        return
-    try:
-        for dirpath, dirnames, _ in os.walk(store):
-            for d in dirnames:
-                if package in d:
-                    _add_cache_finding(
-                        results,
-                        f"pnpm store: {d}",
-                        os.path.join(dirpath, d),
-                    )
-                    return
-    except (PermissionError, OSError):
-        logger.debug("Cannot read pnpm store")
+    _scan_cache_dir(
+        results,
+        Path.home() / ".local" / "share" / "pnpm" / "store",
+        package, "pnpm store",
+        check_dirs=True, check_files=False,
+    )

scan_supply_chain/config.py

@@ -45,6 +45,18 @@
 SOURCE_SCAN_SKIP_DIRS = _COMMON_SKIP_DIRS | {"site-packages", "node_modules"}
 
 
+def read_if_contains(path: Path, keyword: str) -> str | None:
+    """Read a text file if it mentions *keyword*; return text or ``None``."""
+    if not path.is_file():
+        return None
+    try:
+        text = path.read_text(errors="ignore")
+    except (PermissionError, OSError):
+        logger.debug("Cannot read %s", path)
+        return None
+    return text if keyword in text else None
+
+
 def pruned_walk(
     root: Path, skip_dirs: frozenset[str]
 ) -> Generator[tuple[str, list[str], list[str]], None, None]:

scan_supply_chain/formatting.py

@@ -43,21 +43,25 @@ def _code(escape: str) -> str:
 RESET = _code("\033[0m")
 
 
+_SEPARATOR_WIDTH = 63
+
+
 def print_banner(version: str = ""):
     ver_str = f"v{version}" if version else ""
     title = f"Supply Chain Compromise Scanner {ver_str}".strip()
-    # Pad title to fill the box (59 chars inner width)
-    padded = f"   {title}" + " " * (60 - len(title) - 3)
+    # Pad title to fill the box (inner width = _SEPARATOR_WIDTH - 3)
+    inner = _SEPARATOR_WIDTH - 3
+    padded = f"   {title}" + " " * (inner - len(title))
     print(f"{CYAN}{BOLD}")
-    print("+" + "=" * 63 + "+")
+    print("+" + "=" * _SEPARATOR_WIDTH + "+")
     print(f"|{padded}|")
     print("|   Detects known PyPI and npm supply chain attacks            |")
-    print("+" + "=" * 63 + "+")
+    print("+" + "=" * _SEPARATOR_WIDTH + "+")
     print(RESET)
 
 
 def print_separator():
-    print(f"{CYAN}{'-' * 63}{RESET}")
+    print(f"{CYAN}{'-' * _SEPARATOR_WIDTH}{RESET}")
 
 
 def print_phase_header(number: int, title: str):

scan_supply_chain/history_scanner.py

@@ -2,22 +2,19 @@
 
 from __future__ import annotations
 
-import logging
 from pathlib import Path
 
-from .formatting import print_check_header
-from .models import FindingCategory, ScanResults, track_findings
-
-logger = logging.getLogger(__name__)
+from .config import read_if_contains
+from .models import FindingCategory, ScanResults, scanner_check
 
 _PYPI_INSTALL_CMDS = ("pip install", "pip3 install", "uv pip install", "uv add")
 _NPM_INSTALL_CMDS = ("npm install", "npm i ", "yarn add", "pnpm add", "pnpm install")
 
 
 def scan_history(results: ScanResults, package: str, ecosystem: str) -> None:
     """Search shell history for install commands mentioning the package."""
-    print_check_header("shell history for install commands")
-    with track_findings(results, "No install commands found in shell history"):
+    with scanner_check(results, "shell history for install commands",
+                       "No install commands found in shell history"):
         install_cmds = _PYPI_INSTALL_CMDS if ecosystem == "pypi" else _NPM_INSTALL_CMDS
 
         home = Path.home()
@@ -32,12 +29,8 @@ def _scan_history_file(
     package: str,
     install_cmds: tuple[str, ...],
 ) -> None:
-    if not path.is_file():
-        return
-    try:
-        text = path.read_text(errors="ignore")
-    except (PermissionError, OSError):
-        logger.debug("Cannot read %s", path)
+    text = read_if_contains(path, package)
+    if text is None:
         return
 
     for line in text.splitlines():

scan_supply_chain/ioc_windows.py

@@ -1,17 +1,13 @@
 """Windows-only IOC checks: Registry Run keys, Scheduled Tasks."""
 
-import logging
-import subprocess
-
 from .formatting import (
     BOLD,
     RED,
     RESET,
     print_check_header,
     print_clean,
 )
-
-logger = logging.getLogger(__name__)
+from .subprocess_utils import run_safe
 
 
 def _check_registry_run_keys(
@@ -28,23 +24,18 @@ def _check_registry_run_keys(
         r"HKLM\Software\Microsoft\Windows\CurrentVersion\Run",
     ]
     for key_path in run_keys:
-        try:
-            output = subprocess.run(
-                ["reg", "query", key_path],
-                capture_output=True,
-                text=True,
-                timeout=10,
-            ).stdout.lower()
-            for keyword in keywords:
-                if keyword.lower() in output:
-                    print(
-                        f"  {RED}{BOLD}! SUSPICIOUS REGISTRY ENTRY "
-                        f"in {key_path} (matched: {keyword}){RESET}"
-                    )
-                    results.iocs.append(f"registry:{key_path}:{keyword}")
-                    found = True
-        except (subprocess.TimeoutExpired, OSError):
-            logger.debug("Failed to query registry key %s", key_path)
+        output = run_safe(["reg", "query", key_path], timeout=10)
+        if output is None:
+            continue
+        output = output.lower()
+        for keyword in keywords:
+            if keyword.lower() in output:
+                print(
+                    f"  {RED}{BOLD}! SUSPICIOUS REGISTRY ENTRY "
+                    f"in {key_path} (matched: {keyword}){RESET}"
+                )
+                results.iocs.append(f"registry:{key_path}:{keyword}")
+                found = True
     if not found:
         print_clean("No suspicious Run key entries")
 
@@ -58,13 +49,9 @@ def _check_scheduled_tasks(
         return
     print_check_header("Scheduled Tasks for persistence")
     found = False
-    try:
-        output = subprocess.run(
-            ["schtasks", "/query", "/fo", "CSV"],
-            capture_output=True,
-            text=True,
-            timeout=15,
-        ).stdout.lower()
+    output = run_safe(["schtasks", "/query", "/fo", "CSV"], timeout=15)
+    if output is not None:
+        output = output.lower()
         for keyword in keywords:
             if keyword.lower() in output:
                 print(
@@ -73,8 +60,6 @@ def _check_scheduled_tasks(
                 )
                 results.iocs.append(f"schtask:{keyword}")
                 found = True
-    except (subprocess.TimeoutExpired, OSError):
-        logger.debug("Failed to query scheduled tasks")
     if not found:
         print_clean("No suspicious scheduled tasks")
 

scan_supply_chain/models.py

@@ -129,3 +129,15 @@ def track_findings(
         from .formatting import print_clean
 
         print_clean(clean_message)
+
+
+@contextmanager
+def scanner_check(
+    results: ScanResults, header: str, clean_message: str
+) -> Generator[None, None, None]:
+    """Print check header, then clean_message if no findings were added."""
+    from .formatting import print_check_header
+
+    print_check_header(header)
+    with track_findings(results, clean_message):
+        yield