diff --git a/.bandit-baseline.json b/.bandit-baseline.json new file mode 100644 index 000000000..f5f086635 --- /dev/null +++ b/.bandit-baseline.json @@ -0,0 +1,4 @@ +{ + "_comment": "Allowlist of accepted bandit HIGH-severity findings, keyed by (path, test_id) — line numbers are intentionally NOT part of the key. Empty means ANY HIGH finding fails the gate. All 18 pre-existing HIGH findings were fixed or converted to justified inline `# nosec` (see .security-suppressions.json); keep this empty unless a genuinely unavoidable HIGH is reviewed and accepted.", + "findings": [] +} diff --git a/.security-suppressions.json b/.security-suppressions.json new file mode 100644 index 000000000..2eb111fff --- /dev/null +++ b/.security-suppressions.json @@ -0,0 +1,15 @@ +{ + "_comment": "Justified security suppressions. Every inline `# nosec S` / `# nosec B` marker in src/gaia MUST have a matching entry here with a real justification. Do not blanket-suppress to quiet the scanner.", + "suppressions": [ + { + "path": "src/gaia/agents/tools/shell_tools.py", + "rule": "B602", + "justification": "Sandboxed shell tool. Every command (and each pipeline segment) is validated against a whitelist via _validate_command before execution; shell=True is enabled ONLY on Windows so cmd.exe can resolve built-ins (dir/cd/type) and pipes that Git-for-Windows tools rely on. Converting to args-list would break piped/whitelisted commands the tool exists to run." + }, + { + "path": "src/gaia/mcp/client/transports/stdio.py", + "rule": "B602", + "justification": "Legacy from_command() API accepts a full shell command STRING (documented contract, e.g. commands using pipes/env-expansion) and needs shell parsing. Command is caller-supplied SDK config, not external/untrusted input; modern from_config() passes an args list and runs shell=False." + } + ] +} diff --git a/src/gaia/agents/tools/shell_tools.py b/src/gaia/agents/tools/shell_tools.py index 50696c445..c10245dd9 100644 --- a/src/gaia/agents/tools/shell_tools.py +++ b/src/gaia/agents/tools/shell_tools.py @@ -644,7 +644,7 @@ def run_shell_command( timeout=timeout, check=False, env=os.environ.copy(), - shell=use_shell, + shell=use_shell, # nosec B602 - Windows-only; command whitelist-validated above, shell needed for cmd.exe built-ins/pipes ) duration = time.monotonic() - start_time diff --git a/src/gaia/cli.py b/src/gaia/cli.py index 2c1976455..e4cf1a8af 100644 --- a/src/gaia/cli.py +++ b/src/gaia/cli.py @@ -3653,9 +3653,11 @@ def main(): system = platform.system() try: if system == "Windows": - subprocess.run( - ["start", "", mailto_url], shell=True, check=True - ) + # os.startfile uses ShellExecute (no shell parsing), + # safe for the user-built mailto URL (which contains + # '&'-separated query params cmd would mis-parse). + # Windows-only attr; guarded by the platform check. + os.startfile(mailto_url) # pylint: disable=no-member elif system == "Darwin": # macOS subprocess.run(["open", mailto_url], check=True) else: # Linux/Unix @@ -4776,11 +4778,14 @@ def main(): def kill_process_by_port(port): """Find and kill a process running on a specific port.""" + try: + port = int(port) + except (ValueError, TypeError): + return {"success": False, "message": f"Invalid port number: {port!r}"} try: if sys.platform.startswith("win"): - # Windows implementation - cmd = f"netstat -ano | findstr :{port}" - output = subprocess.check_output(cmd, shell=True).decode() + # Windows implementation (filter netstat output in Python, no shell pipe) + output = subprocess.check_output(["netstat", "-ano"]).decode() if output: # Split output into lines and process each line for line in output.strip().split("\n"): @@ -4792,7 +4797,9 @@ def kill_process_by_port(port): pid = int(parts[-1]) if pid > 0: # Ensure we don't try to kill PID 0 subprocess.run( - f"taskkill /PID {pid} /F", shell=True, check=True + ["taskkill", "/PID", str(pid), "/F"], + shell=False, + check=True, ) return { "success": True, @@ -4808,8 +4815,9 @@ def kill_process_by_port(port): # Linux/Unix implementation try: # Use lsof to find process using the port - cmd = f"lsof -ti:{port}" - output = subprocess.check_output(cmd, shell=True).decode().strip() + output = ( + subprocess.check_output(["lsof", f"-ti:{port}"]).decode().strip() + ) if output: pids = output.split("\n") killed_pids = [] @@ -4817,7 +4825,9 @@ def kill_process_by_port(port): try: pid = int(pid_str.strip()) if pid > 0: - subprocess.run(f"kill -9 {pid}", shell=True, check=True) + subprocess.run( + ["kill", "-9", str(pid)], shell=False, check=True + ) killed_pids.append(str(pid)) except (ValueError, subprocess.CalledProcessError): continue @@ -4834,8 +4844,8 @@ def kill_process_by_port(port): # If lsof is not available, try netstat + ps approach try: # Use netstat to find the port, then extract PID - cmd = f"netstat -tulpn | grep :{port}" - output = subprocess.check_output(cmd, shell=True).decode() + # (filter output in Python, no shell pipe) + output = subprocess.check_output(["netstat", "-tulpn"]).decode() if output: for line in output.strip().split("\n"): if f":{port}" in line: @@ -4847,8 +4857,8 @@ def kill_process_by_port(port): pid = int(part.split("/")[0]) if pid > 0: subprocess.run( - f"kill -9 {pid}", - shell=True, + ["kill", "-9", str(pid)], + shell=False, check=True, ) return { diff --git a/src/gaia/llm/lemonade_client.py b/src/gaia/llm/lemonade_client.py index bf9150942..17640460f 100644 --- a/src/gaia/llm/lemonade_client.py +++ b/src/gaia/llm/lemonade_client.py @@ -1109,7 +1109,17 @@ def terminate_server(self): # For subprocess.Popen if sys.platform.startswith("win") and self.server_process.pid: # On Windows, use taskkill to ensure process tree is terminated - os.system(f"taskkill /F /PID {self.server_process.pid} /T") + subprocess.run( + [ + "taskkill", + "/F", + "/PID", + str(self.server_process.pid), + "/T", + ], + shell=False, + check=False, + ) elif self.server_process.pid: # On Linux/Unix, kill the process group to terminate child processes try: diff --git a/src/gaia/logger.py b/src/gaia/logger.py index 9c1251804..c4f4b8a74 100644 --- a/src/gaia/logger.py +++ b/src/gaia/logger.py @@ -23,8 +23,12 @@ def configure_console_encoding(): # Also try to set the console code page to UTF-8 try: + # chcp is a cmd.exe builtin; invoke via cmd /c so we avoid shell=True subprocess.run( - ["chcp", "65001"], capture_output=True, shell=True, check=False + ["cmd", "/c", "chcp", "65001"], + capture_output=True, + shell=False, + check=False, ) except (subprocess.SubprocessError, OSError, FileNotFoundError): pass # Ignore if chcp command fails diff --git a/src/gaia/mcp/client/transports/stdio.py b/src/gaia/mcp/client/transports/stdio.py index b750f6069..a65829427 100644 --- a/src/gaia/mcp/client/transports/stdio.py +++ b/src/gaia/mcp/client/transports/stdio.py @@ -80,9 +80,13 @@ def connect(self) -> bool: merged_env = os.environ.copy() merged_env.update(self.env) + # Legacy from_command() path passes a full shell command string + # (documented contract) and needs shell parsing; modern from_config() + # passes an args list and runs shell=False. Command is caller-supplied + # SDK config, not external/untrusted input. self._process = subprocess.Popen( cmd, - shell=use_shell, + shell=use_shell, # nosec B602 - legacy shell-string API, trusted SDK config stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, diff --git a/src/gaia/mcp/context7_cache.py b/src/gaia/mcp/context7_cache.py index 6d052b9aa..66e9fa227 100644 --- a/src/gaia/mcp/context7_cache.py +++ b/src/gaia/mcp/context7_cache.py @@ -114,7 +114,7 @@ def _doc_cache_file(self, library: str, query: str) -> Path: Path to cache file """ key = f"{library}:{query}" - hash_key = hashlib.md5(key.encode()).hexdigest()[:12] + hash_key = hashlib.md5(key.encode(), usedforsecurity=False).hexdigest()[:12] safe_lib = library.replace("/", "_").replace(".", "_") return self.docs_dir / f"{safe_lib}_{hash_key}.json" diff --git a/src/gaia/ui/build.py b/src/gaia/ui/build.py index 9a68d0855..637cd7ce1 100644 --- a/src/gaia/ui/build.py +++ b/src/gaia/ui/build.py @@ -79,20 +79,24 @@ def ensure_webui_built(log_fn=print, warn_fn=None, _webui_dir=None): warn_fn("Warning: npm not found. Cannot auto-rebuild Agent UI frontend.") return False - # On Windows, npm is a .cmd batch file requiring shell execution - _shell = sys.platform == "win32" + # On Windows npm is a .cmd batch file, which CreateProcess can't launch + # directly; invoke via `cmd /c` (args are static) so we avoid shell=True. + def _npm(*args): + if sys.platform == "win32": + return ["cmd", "/c", "npm", *args] + return ["npm", *args] # Step 1 — npm install (only if node_modules/ missing) if not (webui_dir / "node_modules").is_dir(): log_fn("Installing Agent UI frontend dependencies...") try: subprocess.run( - ["npm", "install"], + _npm("install"), cwd=str(webui_dir), check=True, capture_output=True, text=True, - shell=_shell, + shell=False, ) except subprocess.CalledProcessError as e: warn_fn(f"Warning: npm install failed: {e.stderr}") @@ -106,10 +110,10 @@ def ensure_webui_built(log_fn=print, warn_fn=None, _webui_dir=None): log_fn("Building Agent UI frontend...") try: subprocess.run( - ["npm", "run", "build"], + _npm("run", "build"), cwd=str(webui_dir), check=True, - shell=_shell, + shell=False, ) log_fn("Agent UI frontend built successfully.") return True diff --git a/src/gaia/util.py b/src/gaia/util.py index 79075961f..637c0ed56 100644 --- a/src/gaia/util.py +++ b/src/gaia/util.py @@ -15,10 +15,10 @@ def kill_process_on_port(port): raise ValueError(f"Invalid port number: {port!r}") try: if sys.platform.startswith("win"): - # Windows: use netstat + taskkill + # Windows: use netstat + taskkill (filter output in Python, no shell pipe) result = subprocess.run( - f"netstat -ano | findstr :{port}", - shell=True, + ["netstat", "-ano"], + shell=False, capture_output=True, text=True, check=False, @@ -39,7 +39,7 @@ def kill_process_on_port(port): print(f"Found process with PID {pid} on port {port}") try: subprocess.run( - f"taskkill /F /PID {pid}", shell=True, check=False + ["taskkill", "/F", "/PID", pid], shell=False, check=False ) print(f"Killed process with PID {pid}") except Exception as e: @@ -50,8 +50,8 @@ def kill_process_on_port(port): else: # Unix/macOS: use lsof + kill result = subprocess.run( - f"lsof -ti :{port}", - shell=True, + ["lsof", "-ti", f":{port}"], + shell=False, capture_output=True, text=True, check=False, @@ -64,7 +64,9 @@ def kill_process_on_port(port): if pid: print(f"Found process with PID {pid} on port {port}") try: - subprocess.run(f"kill -9 {pid}", shell=True, check=False) + subprocess.run( + ["kill", "-9", pid], shell=False, check=False + ) print(f"Killed process with PID {pid}") except Exception as e: print(f"Error killing PID {pid}: {e}") diff --git a/tests/unit/test_check_security_gates.py b/tests/unit/test_check_security_gates.py new file mode 100644 index 000000000..e0315ef2d --- /dev/null +++ b/tests/unit/test_check_security_gates.py @@ -0,0 +1,154 @@ +# Copyright(C) 2025-2026 Advanced Micro Devices, Inc. All rights reserved. +# SPDX-License-Identifier: MIT + +"""Unit tests for the bandit HIGH security gate (util/check_security_gates.py).""" + +import json +import sys +from pathlib import Path + +import pytest + +# util/ is not a package; add it to the path so we can import the module. +_UTIL_DIR = Path(__file__).resolve().parents[2] / "util" +sys.path.insert(0, str(_UTIL_DIR)) + +import check_security_gates # noqa: E402 +from check_security_gates import ( # noqa: E402 + bandit_finding_key, + load_baseline, + new_bandit_highs, + parse_bandit_json, + run_bandit, +) + + +def _finding(path, test_id, severity="HIGH", line=10): + return { + "filename": path, + "test_id": test_id, + "issue_severity": severity, + "line_number": line, + "issue_text": f"{test_id} issue", + } + + +class TestBanditFindingKey: + def test_key_is_path_and_test_id(self): + f = _finding("src/gaia/util.py", "B602", line=42) + assert bandit_finding_key(f) == ("src/gaia/util.py", "B602") + + def test_key_normalizes_windows_separators(self): + f = _finding("src\\gaia\\util.py", "B602") + assert bandit_finding_key(f) == ("src/gaia/util.py", "B602") + + def test_key_ignores_line_number(self): + a = _finding("src/gaia/util.py", "B602", line=10) + b = _finding("src/gaia/util.py", "B602", line=999) + assert bandit_finding_key(a) == bandit_finding_key(b) + + +class TestNewBanditHighs: + def test_empty_baseline_reports_all_highs(self): + results = [ + _finding("src/gaia/a.py", "B602"), + _finding("src/gaia/b.py", "B605"), + ] + new = new_bandit_highs(results, []) + assert len(new) == 2 + + def test_only_high_severity_reported(self): + results = [ + _finding("src/gaia/a.py", "B602", severity="HIGH"), + _finding("src/gaia/b.py", "B101", severity="LOW"), + _finding("src/gaia/c.py", "B303", severity="MEDIUM"), + ] + new = new_bandit_highs(results, []) + assert [r["filename"] for r in new] == ["src/gaia/a.py"] + + def test_baseline_dict_entries_are_excluded(self): + results = [_finding("src/gaia/a.py", "B602")] + baseline = [{"path": "src/gaia/a.py", "test_id": "B602"}] + assert new_bandit_highs(results, baseline) == [] + + def test_baseline_pair_entries_are_excluded(self): + results = [_finding("src/gaia/a.py", "B602")] + baseline = [["src/gaia/a.py", "B602"]] + assert new_bandit_highs(results, baseline) == [] + + def test_baseline_matches_regardless_of_line_number(self): + # A cosmetic line shift must not re-trip a baselined finding. + results = [_finding("src/gaia/a.py", "B602", line=500)] + baseline = [{"path": "src/gaia/a.py", "test_id": "B602"}] + assert new_bandit_highs(results, baseline) == [] + + def test_baseline_normalizes_windows_paths(self): + results = [_finding("src\\gaia\\a.py", "B602")] + baseline = [{"path": "src/gaia/a.py", "test_id": "B602"}] + assert new_bandit_highs(results, baseline) == [] + + def test_non_baselined_high_still_reported_when_others_allowlisted(self): + results = [ + _finding("src/gaia/a.py", "B602"), # allowlisted + _finding("src/gaia/b.py", "B605"), # new + ] + baseline = [{"path": "src/gaia/a.py", "test_id": "B602"}] + new = new_bandit_highs(results, baseline) + assert [r["filename"] for r in new] == ["src/gaia/b.py"] + + def test_same_test_id_different_file_not_masked(self): + results = [_finding("src/gaia/b.py", "B602")] + baseline = [{"path": "src/gaia/a.py", "test_id": "B602"}] + assert len(new_bandit_highs(results, baseline)) == 1 + + +class TestParseBanditJson: + def test_strips_progress_bar_prefix(self): + raw = 'Working... 100%\n{"results": [], "errors": []}' + assert parse_bandit_json(raw) == {"results": [], "errors": []} + + def test_plain_json(self): + assert parse_bandit_json('{"results": []}') == {"results": []} + + def test_no_json_raises(self): + with pytest.raises(ValueError): + parse_bandit_json("Working... no json here") + + +class TestLoadBaseline: + def test_missing_file_returns_empty(self, tmp_path): + assert load_baseline(tmp_path / "nope.json") == [] + + def test_bare_list(self, tmp_path): + p = tmp_path / "baseline.json" + p.write_text(json.dumps([{"path": "a.py", "test_id": "B602"}])) + assert load_baseline(p) == [{"path": "a.py", "test_id": "B602"}] + + def test_findings_key(self, tmp_path): + p = tmp_path / "baseline.json" + p.write_text(json.dumps({"findings": [{"path": "a.py", "test_id": "B602"}]})) + assert load_baseline(p) == [{"path": "a.py", "test_id": "B602"}] + + +class TestRunBanditErrors: + def test_missing_binary_raises_runtimeerror(self, monkeypatch): + # A missing bandit binary makes subprocess.run raise FileNotFoundError; + # it must surface as an actionable RuntimeError, not crash the caller. + def _boom(*_a, **_k): + raise FileNotFoundError("bandit") + + monkeypatch.setattr(check_security_gates.subprocess, "run", _boom) + with pytest.raises(RuntimeError, match="bandit is not installed"): + run_bandit("src/gaia") + + +class TestRepoBaselineIsEmpty: + """The shipped baseline must stay empty — all pre-existing HIGH were fixed.""" + + def test_repo_baseline_has_no_findings(self): + repo_root = Path(__file__).resolve().parents[2] + baseline = load_baseline(repo_root / ".bandit-baseline.json") + assert baseline == [], ( + "New HIGH findings must be fixed, not baselined. " + f"Unexpected allowlist entries: {baseline}" + ) diff --git a/tests/unit/test_webui_build.py b/tests/unit/test_webui_build.py index cc9e96a23..2bc1a85d8 100644 --- a/tests/unit/test_webui_build.py +++ b/tests/unit/test_webui_build.py @@ -126,10 +126,17 @@ def test_builds_frontend(self): msgs, mock_run, result = self._call(webui_dir) + # On Windows npm is invoked via `cmd /c npm ...` (avoids shell=True); + # on other platforms it's the bare `npm ...`. Match the npm suffix and + # assert every call runs shell=False. called_cmds = [c.args[0] for c in mock_run.call_args_list] self.assertTrue( - any(c == ["npm", "run", "build"] for c in called_cmds), - f"Expected ['npm', 'run', 'build'] call, got: {called_cmds}", + any(c[-3:] == ["npm", "run", "build"] for c in called_cmds), + f"Expected an 'npm run build' call, got: {called_cmds}", + ) + self.assertTrue( + all(c.kwargs.get("shell") is False for c in mock_run.call_args_list), + "All npm calls must run with shell=False", ) self.assertTrue(result, "Expected True when build succeeds") diff --git a/util/check_security_gates.py b/util/check_security_gates.py new file mode 100644 index 000000000..3043a703b --- /dev/null +++ b/util/check_security_gates.py @@ -0,0 +1,197 @@ +# Copyright(C) 2025-2026 Advanced Micro Devices, Inc. All rights reserved. +# SPDX-License-Identifier: MIT + +"""Security gate helpers for GAIA's lint pipeline. + +Currently implements the bandit HIGH-severity blocking gate: any HIGH finding +that is not explicitly allow-listed in ``.bandit-baseline.json`` fails the build. +Findings are keyed by ``(normalized path, test_id)`` -- never the line number, so +an unrelated edit that shifts a flagged line does not spuriously trip the gate. +""" + +from __future__ import annotations + +import json +import subprocess +import sys +from pathlib import Path +from typing import Any, Dict, Iterable, List, Tuple + +BanditKey = Tuple[str, str] + + +def _normalize_path(path: str) -> str: + """Normalize a file path for stable cross-platform comparison.""" + return str(path).replace("\\", "/") + + +def bandit_finding_key(result: Dict[str, Any]) -> BanditKey: + """Stable key for a bandit result: (normalized path, test_id). + + Deliberately excludes the line number so cosmetic line shifts elsewhere in + a file do not change a finding's identity. + """ + return (_normalize_path(result["filename"]), result["test_id"]) + + +def _baseline_keys(baseline: Iterable[Any]) -> set[BanditKey]: + """Build the allow-listed key set from baseline entries. + + Each entry is either a ``{"path": ..., "test_id": ...}`` dict or a + ``[path, test_id]`` pair. + """ + keys: set[BanditKey] = set() + for entry in baseline or []: + if isinstance(entry, dict): + keys.add((_normalize_path(entry["path"]), entry["test_id"])) + else: + path, test_id = entry + keys.add((_normalize_path(path), test_id)) + return keys + + +def new_bandit_highs( + results: Iterable[Dict[str, Any]], baseline: Iterable[Any] +) -> List[Dict[str, Any]]: + """Return HIGH-severity bandit findings not present in the baseline allowlist. + + Args: + results: the ``results`` array from ``bandit -f json``. + baseline: allow-listed findings, each a ``{"path", "test_id"}`` dict or + a ``[path, test_id]`` pair. An empty/omitted baseline means ANY HIGH + finding is reported. + + Returns: + The offending result dicts (HIGH severity, not allow-listed), in input + order. + """ + allow = _baseline_keys(baseline) + offending: List[Dict[str, Any]] = [] + for result in results: + if result.get("issue_severity") != "HIGH": + continue + if bandit_finding_key(result) in allow: + continue + offending.append(result) + return offending + + +def parse_bandit_json(raw: str) -> Dict[str, Any]: + """Parse bandit JSON output, tolerating a leading progress-bar prefix. + + Bandit writes a ``Working... 100%`` progress line to stdout before the JSON + document; skip everything up to the first ``{``. + """ + start = raw.find("{") + if start == -1: + raise ValueError("no JSON object found in bandit output") + return json.loads(raw[start:]) + + +DEFAULT_EXCLUDE = ( + ".git,__pycache__,venv,.venv,.mypy_cache,.tox,.eggs,_build,buck-out,node_modules" +) + + +def _bandit_available() -> bool: + import importlib.util + + return importlib.util.find_spec("bandit") is not None + + +def run_bandit(src_dir: str, exclude: str = DEFAULT_EXCLUDE) -> Dict[str, Any]: + """Run bandit over ``src_dir`` at -ll and return the parsed JSON report.""" + import shutil + + base = ["-r", src_dir, "-ll", "-f", "json", "--exclude", exclude] + # Prefer the locally installed module (fast, offline); fall back to uvx. + if _bandit_available(): + cmd = [sys.executable, "-m", "bandit", *base] + elif shutil.which("uvx"): + cmd = ["uvx", "bandit", *base] + else: + cmd = ["bandit", *base] + + # bandit exits non-zero when it finds issues; we parse the JSON regardless. + try: + proc = subprocess.run(cmd, capture_output=True, text=True, check=False) + except FileNotFoundError as exc: + # Re-raise as RuntimeError so callers only handle one exception type. + raise RuntimeError( + f"bandit is not installed ({exc}). Install it with " + "`uv pip install -e '.[dev]'`, or run lint via `uv run python util/lint.py`." + ) from exc + if not proc.stdout.strip(): + raise RuntimeError( + f"bandit produced no output (exit {proc.returncode}):\n{proc.stderr}" + ) + return parse_bandit_json(proc.stdout) + + +def load_baseline(path: str | Path) -> List[Any]: + """Load the bandit HIGH allowlist; missing file means an empty allowlist.""" + p = Path(path) + if not p.exists(): + return [] + data = json.loads(p.read_text(encoding="utf-8")) + # Accept either a bare list or {"findings": [...]}. + if isinstance(data, dict): + return data.get("findings", []) + return data + + +def format_finding(result: Dict[str, Any]) -> str: + """One-line human-readable rendering of a bandit finding.""" + return ( + f"{result['test_id']} " + f"{_normalize_path(result['filename'])}:{result.get('line_number', '?')} " + f"{result.get('issue_text', '').strip()}" + ) + + +def check_bandit_high_gate( + src_dir: str = "src/gaia", + baseline_path: str | Path = ".bandit-baseline.json", +) -> Tuple[bool, List[Dict[str, Any]]]: + """Run the bandit HIGH gate. + + Returns ``(passed, new_highs)`` where ``passed`` is True iff there are no + HIGH findings outside the baseline allowlist. + """ + report = run_bandit(src_dir) + baseline = load_baseline(baseline_path) + new_highs = new_bandit_highs(report.get("results", []), baseline) + return (not new_highs, new_highs) + + +def main(argv: List[str] | None = None) -> int: + import argparse + + parser = argparse.ArgumentParser(description="GAIA bandit HIGH security gate") + parser.add_argument("--src", default="src/gaia", help="Source dir to scan") + parser.add_argument( + "--baseline", + default=".bandit-baseline.json", + help="Allowlist of accepted HIGH findings", + ) + args = parser.parse_args(argv) + + passed, new_highs = check_bandit_high_gate(args.src, args.baseline) + if passed: + print("[OK] Bandit HIGH gate: 0 new HIGH-severity findings.") + return 0 + + print(f"[FAIL] Bandit HIGH gate: {len(new_highs)} HIGH finding(s) not baselined:") + for finding in new_highs: + print(f" - {format_finding(finding)}") + print( + "\nFix the finding, or (only if genuinely unavoidable) add an inline " + "`# nosec ` with a matching justified entry in " + "`.security-suppressions.json`, and if truly accepted, allowlist it in " + f"{args.baseline}." + ) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/util/lint.py b/util/lint.py index a4f0e32cc..319227d4b 100644 --- a/util/lint.py +++ b/util/lint.py @@ -284,29 +284,59 @@ def check_mypy() -> CheckResult: def check_bandit() -> CheckResult: - """Run Bandit security check (warning only).""" - print("\n[6/10] Running security check with Bandit (warning only)...") + """Run Bandit security check. + + HIGH-severity findings are BLOCKING: any HIGH finding not allow-listed in + ``.bandit-baseline.json`` fails the check. MEDIUM/LOW findings are still + reported (warning-only) so they stay visible without blocking the build. + """ + print("\n[6/10] Running security check with Bandit (HIGH = blocking)...") print("-" * 40) - cmd = uvx("bandit", "-r", SRC_DIR, "-ll", "--exclude", EXCLUDE_DIRS) + from check_security_gates import ( + format_finding, + load_baseline, + new_bandit_highs, + run_bandit, + ) - print(f"[CMD] {' '.join(cmd)}") - exit_code, output = run_command(cmd) + # Scan once; drive both the HIGH gate and the MEDIUM/LOW count off one report. + try: + report = run_bandit(SRC_DIR, EXCLUDE_DIRS) + except (RuntimeError, ValueError) as exc: + print(f"[ERROR] Could not run Bandit: {exc}") + return CheckResult("Security Check (Bandit)", False, False, 1, str(exc)) - if exit_code != 0: - # Count issue lines - issues = output.count(">> Issue:") or 1 - print(f"\n[WARNING] Bandit found security issues (non-blocking):") - lines = output.strip().split("\n")[:30] - for line in lines: - print(line) - if len(output.strip().split("\n")) > 30: - print("... (output truncated, showing first 30 lines)") - print("\nNote: Many are false positives for ML applications.") - return CheckResult("Security Check (Bandit)", False, True, issues, output) + new_highs = new_bandit_highs( + report.get("results", []), load_baseline(".bandit-baseline.json") + ) + passed = not new_highs + + # Surface MEDIUM/LOW counts as an informational warning (non-blocking). + med_low = [ + r + for r in report.get("results", []) + if r.get("issue_severity") in ("MEDIUM", "LOW") + ] + if med_low: + print(f"[INFO] {len(med_low)} MEDIUM/LOW finding(s) (non-blocking).") + + if not passed: + print(f"\n[BLOCKING] Bandit found {len(new_highs)} HIGH-severity issue(s):") + for finding in new_highs: + print(f" - {format_finding(finding)}") + print( + "\nFix the finding, or (only if genuinely unavoidable) add an inline " + "`# nosec ` with a matching justified entry in " + "`.security-suppressions.json`, then allowlist it in " + "`.bandit-baseline.json`." + ) + return CheckResult( + "Security Check (Bandit)", False, False, len(new_highs), str(new_highs) + ) - print("[OK] No security issues found!") - return CheckResult("Security Check (Bandit)", True, True, 0, output) + print("[OK] No HIGH-severity security issues found!") + return CheckResult("Security Check (Bandit)", True, False, 0, "") def check_imports() -> CheckResult: