diff --git a/desloppify/languages/python/_security.py b/desloppify/languages/python/_security.py index 9b623657d..b3bbb4671 100644 --- a/desloppify/languages/python/_security.py +++ b/desloppify/languages/python/_security.py @@ -6,9 +6,14 @@ from desloppify.base.config import load_config from desloppify.base.discovery.source import collect_exclude_dirs -from desloppify.languages._framework.base.types import DetectorCoverageStatus, LangSecurityResult -from desloppify.languages.python.detectors.bandit_adapter import detect_with_bandit +from desloppify.languages._framework.base.types import ( + DetectorCoverageStatus, + LangSecurityResult, +) from desloppify.languages.python._helpers import scan_root_from_files +from desloppify.languages.python.detectors.bandit_adapter import ( + detect_with_bandit_files, +) def missing_bandit_coverage() -> DetectorCoverageStatus: @@ -51,8 +56,11 @@ def detect_python_security(files, zone_map) -> LangSecurityResult: exclude_dirs = collect_exclude_dirs(scan_root) skip_tests = _load_bandit_skip_tests() - result = detect_with_bandit( - scan_root, zone_map, exclude_dirs=exclude_dirs, skip_tests=skip_tests, + result = detect_with_bandit_files( + files, + zone_map, + exclude_dirs=exclude_dirs, + skip_tests=skip_tests, ) coverage = result.status.coverage() return LangSecurityResult( diff --git a/desloppify/languages/python/detectors/bandit_adapter.py b/desloppify/languages/python/detectors/bandit_adapter.py index 361daed3f..dca8848a7 100644 --- a/desloppify/languages/python/detectors/bandit_adapter.py +++ b/desloppify/languages/python/detectors/bandit_adapter.py @@ -23,6 +23,8 @@ import logging import subprocess # nosec B404 import sys +import time +from collections.abc import Iterable from dataclasses import dataclass from pathlib import Path from typing import Literal @@ -57,6 +59,10 @@ "unsafe deserialization, and risky SQL/subprocess patterns." ) +# Keep a batch well below platform command-line limits while ensuring a single +# slow directory cannot consume the adapter's whole timeout budget. +_BANDIT_FILE_BATCH_SIZE = 250 + BanditRunState = Literal["ok", "missing_tool", "timeout", "error", "parse_error"] @@ -185,25 +191,15 @@ def _to_security_entry( } -def detect_with_bandit( - path: Path, +def _run_bandit( + targets: list[Path], zone_map: FileZoneMap | None, timeout: int = 120, exclude_dirs: list[str] | None = None, skip_tests: list[str] | None = None, + require_target_metrics: bool = False, ) -> BanditScanResult: - """Run bandit on *path* and return issues + typed execution status. - - Parameters - ---------- - exclude_dirs: - Absolute directory paths to pass to bandit's ``--exclude`` flag. - When non-empty, bandit will skip these directories during its - recursive scan. - skip_tests: - Bandit test IDs to suppress via ``--skip`` (e.g. ``["B101", "B601"]``). - Allows users to disable entire rule families from ``config.json``. - """ + """Run Bandit for one non-empty set of explicit targets.""" cmd = [ sys.executable, "-m", @@ -217,7 +213,7 @@ def detect_with_bandit( cmd.extend(["--exclude", ",".join(exclude_dirs)]) if skip_tests: cmd.extend(["--skip", ",".join(skip_tests)]) - cmd.append(str(path.resolve())) + cmd.extend(str(target) for target in targets) try: result = subprocess.run( @@ -249,13 +245,28 @@ def detect_with_bandit( status=BanditRunStatus(state="error", detail=str(exc)), ) + returncode = getattr(result, "returncode", 0) + fatal_returncode = isinstance(returncode, int) and returncode > 1 stdout = result.stdout.strip() if not stdout: - # Bandit exits 0 with no output when there's nothing to scan. + if fatal_returncode: + status = BanditRunStatus( + state="error", + detail=f"bandit exited with status {returncode}", + ) + elif require_target_metrics: + status = BanditRunStatus( + state="error", + detail=f"bandit produced no output for {len(targets)} target(s)", + ) + else: + # Bandit exits 0 with no output when a legacy recursive call has + # nothing to scan. + status = BanditRunStatus(state="ok") return BanditScanResult( entries=[], files_scanned=0, - status=BanditRunStatus(state="ok"), + status=status, ) try: @@ -268,8 +279,13 @@ def detect_with_bandit( status=BanditRunStatus(state="parse_error", detail=str(exc)), ) - raw_results: list[dict] = data.get("results", []) - metrics: dict = data.get("metrics", {}) + raw_results = data.get("results", []) + metrics = data.get("metrics", {}) + errors = data.get("errors", []) + if not isinstance(raw_results, list): + raw_results = [] + if not isinstance(metrics, dict): + metrics = {} # Count scanned files from metrics (bandit reports per-file stats). files_scanned = sum( @@ -280,13 +296,164 @@ def detect_with_bandit( entries: list[dict] = [] for res in raw_results: + if not isinstance(res, dict): + continue entry = _to_security_entry(res, zone_map) if entry is not None: entries.append(entry) - logger.debug("bandit: %d issues from %d files", len(entries), files_scanned) + status = BanditRunStatus(state="ok") + if fatal_returncode: + status = BanditRunStatus( + state="error", + detail=f"bandit exited with status {returncode}", + ) + elif errors: + error_count = len(errors) if isinstance(errors, list) else 1 + status = BanditRunStatus( + state="error", + detail=f"bandit reported {error_count} file error(s)", + ) + elif require_target_metrics: + project_root = get_project_root() + metric_paths = { + (Path(path) if Path(path).is_absolute() else project_root / path).resolve() + for path in metrics + if path != "_totals" and not path.endswith("_totals") + } + missing_targets = [target for target in targets if target.resolve() not in metric_paths] + if missing_targets: + status = BanditRunStatus( + state="error", + detail=f"bandit omitted metrics for {len(missing_targets)} target(s)", + ) + + logger.debug( + "bandit: %d issues from %d files (%s)", + len(entries), + files_scanned, + status.state, + ) return BanditScanResult( entries=entries, files_scanned=files_scanned, - status=BanditRunStatus(state="ok"), + status=status, + ) + + +def _file_targets(files: Iterable[str | Path]) -> list[Path]: + """Normalize, filter, and de-duplicate discovered Python file targets.""" + project_root = get_project_root() + targets: list[Path] = [] + seen: set[Path] = set() + for file in files: + target = Path(file) + if target.suffix != ".py": + continue + if not target.is_absolute(): + target = project_root / target + target = target.resolve() + if target in seen: + continue + seen.add(target) + targets.append(target) + return targets + + +def detect_with_bandit_files( + files: Iterable[str | Path], + zone_map: FileZoneMap | None, + timeout: int = 120, + exclude_dirs: list[str] | None = None, + skip_tests: list[str] | None = None, + batch_size: int = _BANDIT_FILE_BATCH_SIZE, +) -> BanditScanResult: + """Run Bandit over the scanner's discovered Python files in safe batches. + + Scanning explicit file targets keeps Bandit's traversal aligned with the + scanner's exclusion-aware source discovery. A failure in any batch retains + findings from successful batches but reports reduced coverage. + """ + if batch_size < 1: + raise ValueError("batch_size must be positive") + + targets = _file_targets(files) + if not targets: + return BanditScanResult( + entries=[], + files_scanned=0, + status=BanditRunStatus(state="ok"), + ) + + batches = [ + targets[index : index + batch_size] + for index in range(0, len(targets), batch_size) + ] + deadline = time.monotonic() + timeout + entries_by_name: dict[str, dict] = {} + files_scanned = 0 + first_failure: BanditRunStatus | None = None + + for batch_number, batch in enumerate(batches, start=1): + remaining_timeout = deadline - time.monotonic() + if remaining_timeout <= 0: + first_failure = first_failure or BanditRunStatus( + state="timeout", + detail=f"total timeout={timeout}s before batch {batch_number}/{len(batches)}", + ) + break + result = _run_bandit( + batch, + zone_map, + timeout=remaining_timeout, + exclude_dirs=exclude_dirs, + skip_tests=skip_tests, + require_target_metrics=True, + ) + files_scanned += result.files_scanned + for entry in result.entries: + entries_by_name.setdefault(str(entry["name"]), entry) + + if result.status.state != "ok" and first_failure is None: + detail = result.status.detail or result.status.state + first_failure = BanditRunStatus( + state=result.status.state, + detail=f"batch {batch_number}/{len(batches)}: {detail}", + ) + if result.status.state == "missing_tool": + break + + status = first_failure or BanditRunStatus(state="ok") + return BanditScanResult( + entries=list(entries_by_name.values()), + files_scanned=files_scanned, + status=status, + ) + + +def detect_with_bandit( + path: Path, + zone_map: FileZoneMap | None, + timeout: int = 120, + exclude_dirs: list[str] | None = None, + skip_tests: list[str] | None = None, +) -> BanditScanResult: + """Run Bandit recursively on *path* for legacy direct callers. + + Parameters + ---------- + exclude_dirs: + Absolute directory paths to pass to bandit's ``--exclude`` flag. + When non-empty, bandit will skip these directories during its + recursive scan. + skip_tests: + Bandit test IDs to suppress via ``--skip`` (e.g. ``["B101", "B601"]``). + Allows users to disable entire rule families from ``config.json``. + """ + return _run_bandit( + [path.resolve()], + zone_map, + timeout=timeout, + exclude_dirs=exclude_dirs, + skip_tests=skip_tests, ) diff --git a/desloppify/languages/python/tests/test_bandit_adapter.py b/desloppify/languages/python/tests/test_bandit_adapter.py index bf08a5564..ef2258313 100644 --- a/desloppify/languages/python/tests/test_bandit_adapter.py +++ b/desloppify/languages/python/tests/test_bandit_adapter.py @@ -2,6 +2,8 @@ from __future__ import annotations +import json +import subprocess from dataclasses import dataclass from pathlib import Path @@ -97,3 +99,207 @@ def _fake_run(cmd, **kwargs): cmd = captured["cmd"] assert isinstance(cmd, list) assert Path(cmd[-1]).is_absolute() + + +def test_detect_with_bandit_files_batches_discovered_files_without_duplicates(monkeypatch, tmp_path): + files = [tmp_path / f"module_{index}.py" for index in range(3)] + calls: list[list[Path]] = [] + commands: list[list[str]] = [] + + class _FakeCompleted: + def __init__(self, targets: list[Path]) -> None: + self.stdout = json.dumps( + { + "results": [ + { + "filename": str(target), + "test_id": "B102", + "issue_severity": "HIGH", + "issue_confidence": "HIGH", + "issue_text": "exec used", + "line_number": 1, + "test_name": "exec_used", + "code": "exec(value)", + "more_info": "https://example.test/B102", + } + for target in targets + ], + "metrics": {str(target): {} for target in targets}, + } + ) + + def _fake_run(cmd, **_kwargs): + targets = [Path(value) for value in cmd if value.endswith(".py")] + calls.append(targets) + commands.append(cmd) + return _FakeCompleted(targets) + + monkeypatch.setattr(adapter_mod.subprocess, "run", _fake_run) + + result = adapter_mod.detect_with_bandit_files( + [*files, files[0]], + zone_map=None, + batch_size=2, + exclude_dirs=[str(tmp_path / ".venv")], + skip_tests=["B101"], + ) + + assert [len(targets) for targets in calls] == [2, 1] + assert all("--exclude" in command for command in commands) + assert all("--skip" in command for command in commands) + assert result.status.state == "ok" + assert result.files_scanned == 3 + assert len(result.entries) == 3 + + +def test_detect_with_bandit_files_retains_completed_batches_on_timeout(monkeypatch, tmp_path): + files = [tmp_path / "first.py", tmp_path / "second.py"] + calls = 0 + + class _FakeCompleted: + stdout = json.dumps( + { + "results": [ + { + "filename": str(files[0]), + "test_id": "B102", + "issue_severity": "HIGH", + "issue_confidence": "HIGH", + "issue_text": "exec used", + "line_number": 1, + "test_name": "exec_used", + "code": "exec(value)", + "more_info": "https://example.test/B102", + } + ], + "metrics": {str(files[0]): {}}, + } + ) + + def _fake_run(_cmd, **_kwargs): + nonlocal calls + calls += 1 + if calls == 2: + raise subprocess.TimeoutExpired("bandit", 120) + return _FakeCompleted() + + monkeypatch.setattr(adapter_mod.subprocess, "run", _fake_run) + + result = adapter_mod.detect_with_bandit_files(files, zone_map=None, batch_size=1) + + assert result.status.state == "timeout" + assert result.status.detail.startswith("batch 2/2: timeout=") + assert result.status.detail.endswith("s") + assert result.files_scanned == 1 + assert len(result.entries) == 1 + + +def test_detect_with_bandit_files_skips_subprocess_for_empty_input(monkeypatch): + def _unexpected_run(*_args, **_kwargs): + raise AssertionError("Bandit should not run for an empty source list") + + monkeypatch.setattr(adapter_mod.subprocess, "run", _unexpected_run) + + result = adapter_mod.detect_with_bandit_files([], zone_map=None) + + assert result.status.state == "ok" + assert result.files_scanned == 0 + assert result.entries == [] + + +def test_detect_with_bandit_files_marks_bandit_errors_as_reduced_coverage(monkeypatch, tmp_path): + file = tmp_path / "missing.py" + + class _FakeCompleted: + stdout = json.dumps( + { + "errors": [{"filename": str(file), "reason": "No such file"}], + "metrics": {}, + "results": [], + } + ) + + monkeypatch.setattr(adapter_mod.subprocess, "run", lambda *_args, **_kwargs: _FakeCompleted()) + + result = adapter_mod.detect_with_bandit_files([file], zone_map=None) + + assert result.status.state == "error" + assert result.status.detail == "batch 1/1: bandit reported 1 file error(s)" + assert result.status.coverage() is not None + + +def test_detect_with_bandit_files_rejects_empty_output(monkeypatch, tmp_path): + file = tmp_path / "module.py" + + class _FakeCompleted: + returncode = 0 + stdout = "" + + monkeypatch.setattr(adapter_mod.subprocess, "run", lambda *_args, **_kwargs: _FakeCompleted()) + + result = adapter_mod.detect_with_bandit_files([file], zone_map=None) + + assert result.status.state == "error" + assert result.status.detail == "batch 1/1: bandit produced no output for 1 target(s)" + assert result.status.coverage() is not None + + +def test_detect_with_bandit_files_rejects_fatal_bandit_exit(monkeypatch, tmp_path): + file = tmp_path / "module.py" + + class _FakeCompleted: + returncode = 2 + stdout = json.dumps( + { + "errors": [], + "metrics": {str(file): {}}, + "results": [], + } + ) + + monkeypatch.setattr(adapter_mod.subprocess, "run", lambda *_args, **_kwargs: _FakeCompleted()) + + result = adapter_mod.detect_with_bandit_files([file], zone_map=None) + + assert result.status.state == "error" + assert result.status.detail == "batch 1/1: bandit exited with status 2" + assert result.status.coverage() is not None + + +def test_detect_with_bandit_files_rejects_missing_target_metrics(monkeypatch, tmp_path): + file = tmp_path / "module.py" + + class _FakeCompleted: + stdout = json.dumps({"errors": [], "metrics": {}, "results": []}) + + monkeypatch.setattr(adapter_mod.subprocess, "run", lambda *_args, **_kwargs: _FakeCompleted()) + + result = adapter_mod.detect_with_bandit_files([file], zone_map=None) + + assert result.status.state == "error" + assert result.status.detail == "batch 1/1: bandit omitted metrics for 1 target(s)" + assert result.status.coverage() is not None + + +def test_detect_with_bandit_files_keeps_the_original_total_timeout(monkeypatch, tmp_path): + files = [tmp_path / "first.py", tmp_path / "second.py"] + observed_timeouts: list[float] = [] + ticks = iter([100.0, 100.0, 221.0]) + + def _fake_run(_targets, _zone_map, *, timeout, **_kwargs): + observed_timeouts.append(timeout) + return adapter_mod.BanditScanResult( + entries=[], + files_scanned=1, + status=adapter_mod.BanditRunStatus(state="ok"), + ) + + monkeypatch.setattr(adapter_mod, "_run_bandit", _fake_run) + monkeypatch.setattr(adapter_mod.time, "monotonic", lambda: next(ticks)) + + result = adapter_mod.detect_with_bandit_files(files, zone_map=None, batch_size=1, timeout=120) + + assert observed_timeouts == [120.0] + assert result.files_scanned == 1 + assert result.status.state == "timeout" + assert result.status.detail == "total timeout=120s before batch 2/2" diff --git a/desloppify/tests/detectors/test_external_adapters.py b/desloppify/tests/detectors/test_external_adapters.py index eb5aec350..1b445208d 100644 --- a/desloppify/tests/detectors/test_external_adapters.py +++ b/desloppify/tests/detectors/test_external_adapters.py @@ -499,12 +499,13 @@ def test_detect_lang_security_passes_exclusions_to_bandit(self): config = PythonConfig() captured_kwargs = {} - def _fake_bandit(path, zone_map, **kwargs): + def _fake_bandit(discovered_files, zone_map, **kwargs): from desloppify.languages.python.detectors.bandit_adapter import ( BanditRunStatus, BanditScanResult, ) + captured_kwargs["files"] = discovered_files captured_kwargs.update(kwargs) return BanditScanResult( entries=[], files_scanned=0, status=BanditRunStatus(state="ok") @@ -513,7 +514,7 @@ def _fake_bandit(path, zone_map, **kwargs): fake_exclude_dirs = ["/project/src/.venv", "/project/src/__pycache__", "/project/src/vendor"] files = ["/project/src/app.py", "/project/src/utils.py"] with patch( - "desloppify.languages.python._security.detect_with_bandit", _fake_bandit + "desloppify.languages.python._security.detect_with_bandit_files", _fake_bandit ), patch( "desloppify.languages.python._security.collect_exclude_dirs", return_value=fake_exclude_dirs, @@ -523,6 +524,7 @@ def _fake_bandit(path, zone_map, **kwargs): exclude_dirs = captured_kwargs.get("exclude_dirs", []) # Should pass through whatever collect_exclude_dirs returns. assert exclude_dirs == fake_exclude_dirs + assert captured_kwargs["files"] == files # ── jscpd adapter ──────────────────────────────────────────────────────────── diff --git a/desloppify/tests/lang/python/test_python_security_dictkeys_and_smells_split_direct.py b/desloppify/tests/lang/python/test_python_security_dictkeys_and_smells_split_direct.py index 8870e9dda..e51d87229 100644 --- a/desloppify/tests/lang/python/test_python_security_dictkeys_and_smells_split_direct.py +++ b/desloppify/tests/lang/python/test_python_security_dictkeys_and_smells_split_direct.py @@ -38,21 +38,29 @@ def coverage(self): monkeypatch.setattr(py_security_mod, "scan_root_from_files", lambda _files: tmp_path) monkeypatch.setattr(py_security_mod, "collect_exclude_dirs", lambda _root: [".venv", "build"]) - monkeypatch.setattr( - py_security_mod, - "detect_with_bandit", - lambda _root, _zone_map, *, exclude_dirs, skip_tests=None: SimpleNamespace( + + captured_files: list[str] = [] + + def _fake_bandit(detected_files, _zone_map, *, exclude_dirs, skip_tests=None): + captured_files.extend(detected_files) + return SimpleNamespace( entries=[{"file": "a.py", "line": 1}], files_scanned=3, status=_Status(), exclude_dirs=exclude_dirs, - ), + ) + + monkeypatch.setattr( + py_security_mod, + "detect_with_bandit_files", + _fake_bandit, ) result = py_security_mod.detect_python_security(["a.py", "b.py"], zone_map=None) assert len(result.entries) == 1 assert result.files_scanned == 3 assert result.coverage["status"] == "full" + assert captured_files == ["a.py", "b.py"] def test_dict_key_shared_helpers_cover_names_keys_and_distance() -> None: