diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml index 314a3e27c..6d93a6d48 100644 --- a/.github/workflows/lint.yml +++ b/.github/workflows/lint.yml @@ -14,11 +14,13 @@ on: branches: [ main ] paths: - 'src/**' + - 'hub/**' - 'tests/**' - 'setup.py' - 'pyproject.toml' - 'util/lint.py' - 'util/check_*.py' + - '.security-suppressions.json' - '.github/dependabot.yml' - '.github/workflows/lint.yml' pull_request: @@ -26,11 +28,13 @@ on: types: [opened, synchronize, reopened, ready_for_review] paths: - 'src/**' + - 'hub/**' - 'tests/**' - 'setup.py' - 'pyproject.toml' - 'util/lint.py' - 'util/check_*.py' + - '.security-suppressions.json' - '.github/dependabot.yml' - '.github/workflows/lint.yml' merge_group: diff --git a/.security-suppressions.json b/.security-suppressions.json index 2eb111fff..8a0968c0b 100644 --- a/.security-suppressions.json +++ b/.security-suppressions.json @@ -1,6 +1,10 @@ { - "_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.", + "_comment": "Approved security suppressions. Every '# noqa: S' (flake8-bandit) and '# nosec' (bandit) comment in src/ or hub/ MUST be listed here with a justification, or CI fails (util/check_security_gates.py, run by util/lint.py). Adding an entry requires PR review — this is the gate that would have caught the GAIA hub tar-slip (CWE-22), where a '# noqa: S202 - hub artifacts are trusted' silenced an unvalidated tarfile.extractall. Keyed by (path, rule), not line number. See SECURITY.md.", "suppressions": [ + {"path": "src/gaia/eval/scorecard_gate.py", "rule": "S603", "justification": "git is a fixed, trusted executable; args are a constructed list, never shell-interpreted"}, + {"path": "src/gaia/hub/installer.py", "rule": "S603", "justification": "'uv pip install' args are a constructed list (no user string), never shell-interpreted"}, + {"path": "src/gaia/hub/native_launcher.py", "rule": "S603", "justification": "native binary path is an installed hub artifact (operator-controlled); args are constructed, not shell"}, + {"path": "src/gaia/mcp/mcp_bridge.py", "rule": "B104", "justification": "binds 0.0.0.0 only on explicit caller opt-in (documented flag), never by default"}, { "path": "src/gaia/agents/tools/shell_tools.py", "rule": "B602", diff --git a/tests/unit/test_check_security_gates.py b/tests/unit/test_check_security_gates.py index e0315ef2d..65d2d4e05 100644 --- a/tests/unit/test_check_security_gates.py +++ b/tests/unit/test_check_security_gates.py @@ -1,7 +1,17 @@ # 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).""" +"""Unit tests for util/check_security_gates.py. + +Covers both gates the module ships: + +1. The suppression review gate — the scanner (which comment forms count as a + security suppression), the allowlist loader (justification required), the + violation diff, and a backtest proving the gate flags the exact suppression + that hid the hub tar-slip. +2. The bandit HIGH gate — finding keys, the baseline diff, JSON parsing, and the + invariant that the shipped baseline stays empty. +""" import json import sys @@ -9,11 +19,11 @@ 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)) +REPO_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(REPO_ROOT / "util")) import check_security_gates # noqa: E402 +import check_security_gates as gates # noqa: E402 from check_security_gates import ( # noqa: E402 bandit_finding_key, load_baseline, @@ -23,6 +33,12 @@ ) +def _write(tmp_path: Path, rel: str, body: str) -> None: + p = tmp_path / rel + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text(body, encoding="utf-8") + + def _finding(path, test_id, severity="HIGH", line=10): return { "filename": path, @@ -33,6 +49,130 @@ def _finding(path, test_id, severity="HIGH", line=10): } +# =========================================================================== +# Suppression review gate +# =========================================================================== + + +# --------------------------------------------------------------------------- +# find_suppressions +# --------------------------------------------------------------------------- + + +def test_detects_noqa_bandit_code(tmp_path): + _write(tmp_path, "src/a.py", "x = 1 # noqa: S202 - trust me\n") + found = gates.find_suppressions(tmp_path) + assert found == [{"path": "src/a.py", "rule": "S202", "line": 1}] + + +def test_detects_nosec_with_and_without_code(tmp_path): + _write( + tmp_path, + "src/b.py", + "a() # nosec B104 - opt-in\nb() # nosec\n", + ) + found = {(f["path"], f["rule"]) for f in gates.find_suppressions(tmp_path)} + assert ("src/b.py", "B104") in found + assert ("src/b.py", "nosec") in found + + +def test_ignores_non_security_noqa_codes(tmp_path): + # SLF001 (flake8-self) and E402 are not bandit security codes — must be ignored. + _write( + tmp_path, "src/c.py", "z = obj._x # noqa: SLF001\nimport os # noqa: E402\n" + ) + assert gates.find_suppressions(tmp_path) == [] + + +def test_scans_hub_not_just_src(tmp_path): + _write(tmp_path, "hub/agents/x/y.py", "run() # nosec\n") + found = gates.find_suppressions(tmp_path) + assert found and found[0]["path"] == "hub/agents/x/y.py" + + +def test_excludes_vendored_dirs(tmp_path): + _write(tmp_path, "src/node_modules/dep.py", "x = 1 # noqa: S202\n") + assert gates.find_suppressions(tmp_path) == [] + + +# --------------------------------------------------------------------------- +# violations +# --------------------------------------------------------------------------- + + +def test_allowlisted_suppression_is_not_a_violation(): + supp = [{"path": "src/a.py", "rule": "S603", "line": 5}] + allow = {("src/a.py", "S603")} + assert gates.suppression_violations(supp, allow) == [] + + +def test_unlisted_suppression_is_a_violation(): + supp = [{"path": "src/a.py", "rule": "S202", "line": 5}] + assert gates.suppression_violations(supp, set()) == supp + + +# --------------------------------------------------------------------------- +# load_allowlist +# --------------------------------------------------------------------------- + + +def test_allowlist_requires_justification(tmp_path, monkeypatch): + f = tmp_path / ".security-suppressions.json" + f.write_text( + '{"suppressions": [{"path": "src/a.py", "rule": "S603"}]}', encoding="utf-8" + ) + monkeypatch.setattr(gates, "SUPPRESSIONS_FILE", f) + with pytest.raises(ValueError, match="justification"): + gates.load_allowlist() + + +def test_allowlist_loads_valid_entries(tmp_path, monkeypatch): + f = tmp_path / ".security-suppressions.json" + f.write_text( + '{"suppressions": [{"path": "src/a.py", "rule": "S603", ' + '"justification": "constructed args"}]}', + encoding="utf-8", + ) + monkeypatch.setattr(gates, "SUPPRESSIONS_FILE", f) + assert gates.load_allowlist() == {("src/a.py", "S603")} + + +# --------------------------------------------------------------------------- +# Backtest: the gate would have caught the hub tar-slip (CWE-22) +# --------------------------------------------------------------------------- + + +def test_backtest_flags_the_tarslip_suppression(tmp_path): + """The tar-slip shipped as `tf.extractall(...) # noqa: S202 - hub artifacts + are trusted`. With that line present and NOT in the allowlist, the gate must + flag it — i.e. this gate would have forced review before it merged.""" + _write( + tmp_path, + "src/gaia/hub/installer.py", + "with tarfile.open(p) as tf:\n" + " tf.extractall(d) # noqa: S202 - hub artifacts are trusted\n", + ) + found = gates.find_suppressions(tmp_path) + violations = gates.suppression_violations(found, allowlist=set()) + assert any(v["rule"] == "S202" for v in violations) + + +# --------------------------------------------------------------------------- +# The real repo passes its own gate +# --------------------------------------------------------------------------- + + +def test_repo_suppressions_all_reviewed(): + """Every real suppression in the tree is in the committed allowlist.""" + ok, _msgs = gates.check_suppressions() + assert ok is True + + +# =========================================================================== +# Bandit HIGH gate +# =========================================================================== + + class TestBanditFindingKey: def test_key_is_path_and_test_id(self): f = _finding("src/gaia/util.py", "B602", line=42) diff --git a/util/check_security_gates.py b/util/check_security_gates.py index 3043a703b..c9bb002f5 100644 --- a/util/check_security_gates.py +++ b/util/check_security_gates.py @@ -3,23 +3,180 @@ """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. +Two complementary gates live here, both run by ``python util/lint.py --all``: + +1. Suppression review gate (``check_suppressions``). Every ``# noqa: S`` + (flake8-bandit) and ``# nosec`` (bandit) comment under ``src/`` and ``hub/`` + must be listed in ``.security-suppressions.json`` with a justification. A + suppression not listed there fails, forcing a human to review and justify it + in the PR. This is the gate that would have caught the GAIA hub tar-slip + (CWE-22): a ``# noqa: S202 - hub artifacts are trusted`` silenced an + unvalidated ``tarfile.extractall`` from both bandit and the weekly audit. + Entries key on ``(path, rule)`` — never a line number — so the allowlist + survives edits instead of re-firing on every line shift. + +2. Bandit HIGH gate (``check_bandit_high_gate``). Any HIGH-severity bandit + 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. + +Run directly with ``python util/check_security_gates.py`` to execute both gates. """ from __future__ import annotations import json +import re import subprocess import sys from pathlib import Path from typing import Any, Dict, Iterable, List, Tuple +# Anchor to the repo root so the script works regardless of CWD — matches +# util/check_dependabot.py and util/check_doc_versions.py. +REPO_ROOT = Path(__file__).resolve().parents[1] +SUPPRESSIONS_FILE = REPO_ROOT / ".security-suppressions.json" + +# Directories scanned for suppressions. Security suppressions can hide a finding +# anywhere shippable code lives — core (src/) and the hub agents (hub/). +SCAN_DIRS = ("src", "hub") +EXCLUDE_PARTS = { + "node_modules", + ".venv", + "venv", + "__pycache__", + ".mypy_cache", + "build", + "dist", + "site-packages", + ".backup", +} + +# A flake8-bandit security code is 'S' + exactly three digits (S101..S704). +# This deliberately excludes non-security 'S'-prefixed codes like SLF001 +# (flake8-self) or SIM/… which are style, not security. +_BANDIT_NOQA_CODE = re.compile(r"S\d{3}") +_NOQA = re.compile(r"#\s*noqa:\s*([A-Za-z0-9,\s]+)") +_ALL_CODES = re.compile(r"[A-Za-z]+\d+") +_NOSEC = re.compile(r"#\s*nosec\b(.*)") +_NOSEC_BCODE = re.compile(r"B\d{3}") + BanditKey = Tuple[str, str] +# --------------------------------------------------------------------------- +# Suppression review gate +# --------------------------------------------------------------------------- +def _iter_python_files(root: Path): + for scan in SCAN_DIRS: + base = root / scan + if not base.exists(): + continue + for path in base.rglob("*.py"): + if EXCLUDE_PARTS & set(path.parts): + continue + yield path + + +def find_suppressions(root: Path | None = None) -> List[Dict[str, Any]]: + """Return every security suppression found under the scanned dirs. + + Each item is ``{"path": , "rule": , "line": }``. + ``rule`` is a flake8-bandit ``S`` code, a bandit ``B`` code, or the + literal ``"nosec"`` for a bare ``# nosec`` with no explicit code. + """ + root = root or REPO_ROOT + found: List[Dict[str, Any]] = [] + for path in _iter_python_files(root): + rel = path.relative_to(root).as_posix() + try: + lines = path.read_text(encoding="utf-8").splitlines() + except (OSError, UnicodeDecodeError): + continue + for lineno, line in enumerate(lines, start=1): + noqa = _NOQA.search(line) + if noqa: + for code in _ALL_CODES.findall(noqa.group(1)): + if _BANDIT_NOQA_CODE.fullmatch(code): + found.append({"path": rel, "rule": code, "line": lineno}) + nosec = _NOSEC.search(line) + if nosec: + bcodes = _NOSEC_BCODE.findall(nosec.group(1)) + if bcodes: + for b in bcodes: + found.append({"path": rel, "rule": b, "line": lineno}) + else: + found.append({"path": rel, "rule": "nosec", "line": lineno}) + return found + + +def load_allowlist() -> set[Tuple[str, str]]: + """Load the approved ``(path, rule)`` suppression pairs.""" + if not SUPPRESSIONS_FILE.exists(): + raise FileNotFoundError( + f"{SUPPRESSIONS_FILE} is missing. It is the allowlist of approved " + f"security suppressions; the security gate cannot run without it." + ) + try: + data = json.loads(SUPPRESSIONS_FILE.read_text(encoding="utf-8")) + except json.JSONDecodeError as exc: + raise ValueError(f"{SUPPRESSIONS_FILE} is not valid JSON: {exc}") from exc + allow: set[Tuple[str, str]] = set() + for entry in data.get("suppressions", []): + path = entry.get("path") + rule = entry.get("rule") + if not path or not rule: + raise ValueError( + f"{SUPPRESSIONS_FILE}: every suppression needs a 'path' and 'rule' " + f"(offending entry: {entry!r})." + ) + if not entry.get("justification"): + raise ValueError( + f"{SUPPRESSIONS_FILE}: suppression {path}:{rule} has no " + f"'justification' — every approved suppression must say why." + ) + allow.add((path, rule)) + return allow + + +def suppression_violations( + suppressions: List[Dict[str, Any]], allowlist: set[Tuple[str, str]] +) -> List[Dict[str, Any]]: + """Return suppressions whose ``(path, rule)`` is not in the allowlist.""" + return [s for s in suppressions if (s["path"], s["rule"]) not in allowlist] + + +def check_suppressions() -> Tuple[bool, List[str]]: + """Run the suppression gate. Returns ``(ok, messages)``.""" + suppressions = find_suppressions() + allowlist = load_allowlist() + violations = suppression_violations(suppressions, allowlist) + msgs: List[str] = [] + if not violations: + msgs.append( + f"[OK] {len(suppressions)} security suppression(s) - all reviewed in " + f"{SUPPRESSIONS_FILE.name}." + ) + return True, msgs + msgs.append( + f"[FAIL] {len(violations)} security suppression(s) are NOT reviewed in " + f"{SUPPRESSIONS_FILE.name}:" + ) + for v in violations: + msgs.append(f" {v['path']}:{v['line']} {v['rule']}") + msgs.append( + " Every '# noqa: S' / '# nosec' must be justified in " + f"{SUPPRESSIONS_FILE.name} so a human reviews it (this is the gate that " + "would have caught the hub tar-slip). Add an entry with a 'justification', " + "or remove the suppression and fix the finding." + ) + return False, msgs + + +# --------------------------------------------------------------------------- +# Bandit HIGH gate +# --------------------------------------------------------------------------- def _normalize_path(path: str) -> str: """Normalize a file path for stable cross-platform comparison.""" return str(path).replace("\\", "/") @@ -164,33 +321,49 @@ def check_bandit_high_gate( return (not new_highs, new_highs) +# --------------------------------------------------------------------------- +# Entry point — run both gates +# --------------------------------------------------------------------------- def main(argv: List[str] | None = None) -> int: import argparse - parser = argparse.ArgumentParser(description="GAIA bandit HIGH security gate") + parser = argparse.ArgumentParser(description="GAIA security gates") 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", + help="Allowlist of accepted bandit HIGH findings", ) args = parser.parse_args(argv) - passed, new_highs = check_bandit_high_gate(args.src, args.baseline) - if passed: + # Gate 1: suppression review. + try: + supp_ok, supp_msgs = check_suppressions() + except (FileNotFoundError, ValueError) as exc: + print(f"[FAIL] {exc}") + supp_ok = False + supp_msgs = [] + for msg in supp_msgs: + print(msg) + + # Gate 2: bandit HIGH. + bandit_ok, new_highs = check_bandit_high_gate(args.src, args.baseline) + if bandit_ok: 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 + else: + 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 0 if (supp_ok and bandit_ok) else 1 if __name__ == "__main__": diff --git a/util/lint.py b/util/lint.py index 319227d4b..0150e199a 100644 --- a/util/lint.py +++ b/util/lint.py @@ -81,7 +81,7 @@ def check_black(fix: bool = False) -> CheckResult: if fix: print("\n[1/2] Fixing code formatting with Black...") else: - print("\n[1/10] Checking code formatting with Black...") + print("\n[1/11] Checking code formatting with Black...") print("-" * 40) if fix: @@ -155,7 +155,7 @@ def check_isort(fix: bool = False) -> CheckResult: if fix: print("\n[2/2] Fixing import sorting with isort...") else: - print("\n[2/10] Checking import sorting with isort...") + print("\n[2/11] Checking import sorting with isort...") print("-" * 40) if fix: @@ -200,7 +200,7 @@ def check_isort(fix: bool = False) -> CheckResult: def check_pylint() -> CheckResult: """Run Pylint (errors only).""" - print("\n[3/10] Running Pylint (errors only)...") + print("\n[3/11] Running Pylint (errors only)...") print("-" * 40) cmd = uvx( @@ -225,7 +225,7 @@ def check_pylint() -> CheckResult: def check_flake8() -> CheckResult: """Run Flake8.""" - print("\n[4/10] Running Flake8...") + print("\n[4/11] Running Flake8...") print("-" * 40) cmd = uvx( @@ -258,7 +258,7 @@ def check_flake8() -> CheckResult: def check_mypy() -> CheckResult: """Run MyPy type checking (warning only).""" - print("\n[5/10] Running MyPy type checking (warning only)...") + print("\n[5/11] Running MyPy type checking (warning only)...") print("-" * 40) cmd = uvx("mypy", SRC_DIR, "--ignore-missing-imports") @@ -283,6 +283,14 @@ def check_mypy() -> CheckResult: return CheckResult("Type Checking (MyPy)", True, True, 0, output) +def _import_security_gates(): + """Import util/check_security_gates regardless of how lint.py was invoked.""" + sys.path.insert(0, str(Path(__file__).resolve().parent)) + import check_security_gates # noqa: E402 + + return check_security_gates + + def check_bandit() -> CheckResult: """Run Bandit security check. @@ -290,7 +298,7 @@ def check_bandit() -> CheckResult: ``.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("\n[6/11] Running security check with Bandit (HIGH = blocking)...") print("-" * 40) from check_security_gates import ( @@ -339,9 +347,26 @@ def check_bandit() -> CheckResult: return CheckResult("Security Check (Bandit)", True, False, 0, "") +def check_security_suppressions() -> CheckResult: + """Every '# noqa: S' / '# nosec' must be reviewed in .security-suppressions.json.""" + print("\n[7/11] Checking security suppressions are reviewed...") + print("-" * 40) + gates = _import_security_gates() + try: + ok, msgs = gates.check_suppressions() + except (FileNotFoundError, ValueError) as exc: + print(f"[FAIL] {exc}") + return CheckResult("Security Suppressions", False, False, 1, str(exc)) + for m in msgs: + print(m) + return CheckResult( + "Security Suppressions", ok, False, 0 if ok else 1, "\n".join(msgs) + ) + + def check_imports() -> CheckResult: """Test comprehensive SDK imports.""" - print("\n[7/10] Testing comprehensive SDK imports...") + print("\n[8/11] Testing comprehensive SDK imports...") print("-" * 40) # Pre-check: verify gaia is installed @@ -462,7 +487,7 @@ def check_imports() -> CheckResult: def check_agents() -> CheckResult: """Check GAIA agent conventions — inheritance, tests, docs, registry wiring.""" - print("\n[8/10] Checking agent conventions...") + print("\n[9/11] Checking agent conventions...") print("-" * 40) # Import lazily so the check is self-contained and testable standalone. @@ -498,7 +523,7 @@ def check_agents() -> CheckResult: def check_dependabot() -> CheckResult: """Validate .github/dependabot.yml (no soft-disabled entries, npm groups present).""" - print("\n[9/10] Validating .github/dependabot.yml...") + print("\n[10/11] Validating .github/dependabot.yml...") print("-" * 40) try: @@ -529,7 +554,7 @@ def check_dependabot() -> CheckResult: def check_doc_versions() -> CheckResult: """Check documentation version consistency.""" - print("\n[10/10] Checking documentation version consistency...") + print("\n[11/11] Checking documentation version consistency...") print("-" * 40) # Import and run the check @@ -691,6 +716,11 @@ def main(): parser.add_argument("--flake8", action="store_true", help="Run Flake8") parser.add_argument("--mypy", action="store_true", help="Run MyPy") parser.add_argument("--bandit", action="store_true", help="Run Bandit") + parser.add_argument( + "--security", + action="store_true", + help="Check security suppressions are reviewed (.security-suppressions.json)", + ) parser.add_argument("--imports", action="store_true", help="Test imports") parser.add_argument( "--agents", @@ -722,6 +752,7 @@ def main(): args.flake8, args.mypy, args.bandit, + args.security, args.imports, args.agents, args.dependabot, @@ -771,6 +802,9 @@ def main(): if args.bandit or run_all: results.append(check_bandit()) + if args.security or run_all: + results.append(check_security_suppressions()) + if args.agents or run_all: results.append(check_agents())