diff --git a/skills/kvb-check-security/SKILL.md b/skills/kvb-check-security/SKILL.md index 205164e..ddb9068 100644 --- a/skills/kvb-check-security/SKILL.md +++ b/skills/kvb-check-security/SKILL.md @@ -46,6 +46,13 @@ value. Pass `` as `--security` and the critical count as `--critical` boundaries, secrets and injection in the diff by hand, then pass `--security '~'` (quoted: zsh expands a bare `~100` as a user's home) and the critical findings you found as `--critical`. - Exit 2: usage error. +- **Accepted false positives**: `security.py accept --repo --rule + --path --sha1 --reason ""`, rule, path and + sha1 from `--json`, appends a row outside the repository: to the council + root, which the `kvb-org-organize` watch hook keeps in + `~/.local/state/agents-council/`. A match leaves the score and shows as + ` accepted`; editing the line brings it back. Accept only after reading + the rule's pattern, and name each acceptance in the commit message. ## Rules diff --git a/skills/kvb-check-security/scripts/security.py b/skills/kvb-check-security/scripts/security.py index 46d33f8..17e0630 100755 --- a/skills/kvb-check-security/scripts/security.py +++ b/skills/kvb-check-security/scripts/security.py @@ -6,18 +6,34 @@ Args: --base: git ref the change starts from. --json: print the figures as JSON instead of the line. + accept --repo --rule --path --sha1 <40 hex> --reason : + record a reviewed false positive instead; --json of a scan prints rule, path + and sha1 of every finding. + Scan: The regular files `git diff ` names plus untracked ones, symlinks left out, copied to a temporary tree and scanned once with - `medusa scan --format json -o --no-cache`, read from + `medusa scan --format json -o --no-cache`, HOME pointing into + the temporary directory because medusa creates ~/.medusa on start, read from `/medusa-scan-*.json`: a `findings` list whose entries carry - `severity` (critical, high, medium, low or info), `file` and `line`. + `severity` (critical, high, medium, low or info), `rule_id`, `file` and `line`. + +Accepted: + One JSON row per reviewed false positive (id, rule, path, sha1 of the stripped + source line, reason, repo, date), appended by accept to + /security-accepted.jsonl of the kvb-org-organize watch.py, whose + hook keeps the rows in ~/.local/state/agents-council/security-accepted.jsonl; + a scan reads both. A finding with that rule, path and line text is accepted and + leaves the score; an edit of the line brings it back. A row without rule, path + and sha1 is a usage error. Output: - `🛡️ security % (medusa: findings in files: :, ...)`, - m the files scanned, the first 5 locations by severity and `+ more`. + `🛡️ security % (medusa: findings, accepted in files: :, ...)`, + `, accepted` only when a is above 0, m the files scanned, the first 5 + locations by severity and `+ more`. n = 100 - 50 * critical - 25 * high - 10 * medium - 2 * low, at least 0. Matched - values are never printed. --json adds the counts per severity and every location. + values are never printed. --json adds the counts per severity, every location + with its rule and sha1, and the accepted ones. Exit codes: 0 security 100, 1 a critical finding, 2 usage error, 3 under 100 without a @@ -27,7 +43,13 @@ Exit codes: from __future__ import annotations import argparse +import datetime +import fcntl +import hashlib +import importlib.util import json +import os +import re import shutil import subprocess import sys @@ -37,6 +59,8 @@ from pathlib import Path WEIGHTS = {"critical": 50, "high": 25, "medium": 10, "low": 2, "info": 0} SAFE = 100 SHOWN = 5 +SHA1 = re.compile(r"[0-9a-f]{40}") +WATCH = Path(__file__).resolve().parents[2] / "kvb-org-organize" / "scripts" / "watch.py" HAND = ( "review trust boundaries, secrets and injection in the diff by hand, " "then pass --security '~' to kvb-check-score" @@ -90,6 +114,65 @@ def located(name: str, tree: Path) -> str: return name +def fingerprint(tree: Path, path: str, line: int | None) -> str | None: + try: + rows = (tree / path).read_text(encoding="utf-8", errors="replace").splitlines() + except OSError: + return None + if line is None or not 1 <= line <= len(rows): + return None + return hashlib.sha1(rows[line - 1].strip().encode()).hexdigest() + + +def watch(): + if not WATCH.is_file(): + fail(f"{WATCH} is missing; its council paths hold the accepted findings") + spec = importlib.util.spec_from_file_location("kvb_security_watch", WATCH) + loaded = importlib.util.module_from_spec(spec) + spec.loader.exec_module(loaded) + return loaded + + +def accepted(listings: tuple[Path, ...]) -> set[tuple[str, str, str]]: + entries = set() + for listing in listings: + if not listing.is_file(): + continue + for number, line in enumerate(listing.read_text(encoding="utf-8").splitlines(), 1): + if not line.strip(): + continue + try: + row = json.loads(line) + entries.add((row["rule"], row["path"], row["sha1"])) + except (ValueError, KeyError, TypeError): + fail(f"{listing}:{number}: want a JSON row with rule, path and sha1") + return entries + + +def accept(argv: list[str]) -> int: + parser = argparse.ArgumentParser(prog="security.py accept", description="Record a reviewed false positive.") + for flag in ("--repo", "--rule", "--path", "--sha1", "--reason"): + parser.add_argument(flag, required=True) + args = parser.parse_args(argv) + if not SHA1.fullmatch(args.sha1): + parser.error("--sha1 must be 40 lowercase hex digits") + if not args.reason.strip() or "\n" in args.reason: + parser.error("--reason must be one non-empty line") + root = git(Path(args.repo), "rev-parse", "--show-toplevel").strip() + key = f"{args.rule}\t{args.path}\t{args.sha1}" + row = { + "id": hashlib.sha1(key.encode()).hexdigest(), "rule": args.rule, "path": args.path, "sha1": args.sha1, + "reason": args.reason, "repo": root, "date": datetime.datetime.now(datetime.UTC).isoformat(timespec="seconds"), + } + target = watch().ACCEPTED + target.parent.mkdir(parents=True, exist_ok=True) + with target.open("a", encoding="utf-8") as listing: + fcntl.flock(listing, fcntl.LOCK_EX) + listing.write(json.dumps(row) + "\n") + print(f"accepted {args.rule} {args.path} ({args.sha1[:12]}) in {target}") + return 0 + + def findings(reports: Path, tree: Path) -> list[dict]: files = sorted(reports.glob("medusa-scan-*.json")) if reports.is_dir() else [] if not files: @@ -106,11 +189,14 @@ def findings(reports: Path, tree: Path) -> list[dict]: severity = str(entry.get("severity", "")).lower() if isinstance(entry, dict) else "" if severity not in WEIGHTS: raise Unreadable(f"{files[-1].name}: a finding without a known severity") - line = entry.get("line") + line = entry.get("line") if isinstance(entry.get("line"), int) else None + path = located(str(entry.get("file") or "?"), tree) found.append({ "severity": severity, - "path": located(str(entry.get("file") or "?"), tree), - "line": line if isinstance(line, int) else None, + "rule": entry.get("rule_id") if isinstance(entry.get("rule_id"), str) else None, + "path": path, + "line": line, + "sha1": fingerprint(tree, path, line), }) rank = list(WEIGHTS) return sorted(found, key=lambda f: (rank.index(f["severity"]), f["path"], f["line"] or 0)) @@ -119,14 +205,16 @@ def findings(reports: Path, tree: Path) -> list[dict]: def scan(medusa: str, root: Path, base: str) -> tuple[int, list[dict]]: with tempfile.TemporaryDirectory(prefix="kvb-security-") as scratch: work = Path(scratch) - tree, reports = work / "tree", work / "reports" + tree, reports, home = work / "tree", work / "reports", work / "home" tree.mkdir() + home.mkdir() scanned = copy(root, changed(root, base), tree) if not scanned: return 0, [] result = subprocess.run( [medusa, "scan", str(tree), "--format", "json", "-o", str(reports), "--no-cache"], cwd=work, stdin=subprocess.DEVNULL, capture_output=True, check=False, + env={**os.environ, "HOME": str(home)}, ) try: return scanned, findings(reports, tree) @@ -154,11 +242,13 @@ def main() -> int: ).returncode != 0: fail(f"unknown ref {args.base!r}") + council = watch() + reviewed = accepted((council.ACCEPTED, council.KEPT_ACCEPTED)) medusa = shutil.which("medusa") problem = None if medusa else "medusa missing" if medusa: try: - scanned, found = scan(medusa, root, args.base) + scanned, every = scan(medusa, root, args.base) except Unreadable as error: problem = f"medusa report unreadable: {error}" if problem: @@ -168,19 +258,25 @@ def main() -> int: print(f"🛡️ security – ({problem}: {HAND})") return 4 + passed = [f for f in every if (f["rule"], f["path"], f["sha1"]) in reviewed] + found = [f for f in every if f not in passed] counts = {severity: sum(f["severity"] == severity for f in found) for severity in WEIGHTS} value = max(0, SAFE - sum(WEIGHTS[s] * n for s, n in counts.items())) if args.json: - print(json.dumps({"security": value, "files": scanned, **counts, "findings": found})) + print(json.dumps({"security": value, "files": scanned, **counts, "findings": found, "accepted": passed})) else: places = [f"{f['severity']} {f['path']}:{f['line'] if f['line'] is not None else '?'}" for f in found] listed = ", ".join(places[:SHOWN]) + (f", +{len(places) - SHOWN} more" if len(places) > SHOWN else "") detail = f": {listed}" if places else "" - print(f"🛡️ security {value} % (medusa: {plural(len(found), 'finding')} in {plural(scanned, 'file')}{detail})") + waived = f", {len(passed)} accepted" if passed else "" + print( + f"🛡️ security {value} % (medusa: {plural(len(found), 'finding')}{waived} " + f"in {plural(scanned, 'file')}{detail})" + ) if counts["critical"]: return 1 return 3 if value < SAFE else 0 if __name__ == "__main__": - sys.exit(main()) + sys.exit(accept(sys.argv[2:]) if sys.argv[1:2] == ["accept"] else main()) diff --git a/skills/kvb-org-organize/scripts/watch.py b/skills/kvb-org-organize/scripts/watch.py index bfe8bc8..567cf35 100755 --- a/skills/kvb-org-organize/scripts/watch.py +++ b/skills/kvb-org-organize/scripts/watch.py @@ -4,9 +4,10 @@ Usage: watch.py watch hook (UserPromptSubmit, PostToolUse): measure at most once per interval machine-wide, act on the stage, show alerts, - and append the kvb-org-queue history rows of - /queue-history.jsonl whose id it lacks to - ~/.local/state/agents-council/queue-history.jsonl + and append the rows of /queue-history.jsonl + (kvb-org-queue) and /security-accepted.jsonl + (kvb-check-security) whose id it lacks to the same name + under ~/.local/state/agents-council/ watch.py block hook (PreToolUse on Bash): exit 2 for a heavy job while the stage is stop or critical, or while a kvb-org-radio mayday holds radio silence (/radio-silence.json, until > now) @@ -60,8 +61,12 @@ STAGES = ("ok", "warn", "stop", "critical") STATE = ROOT / "watch.json" SILENCE = ROOT / "radio-silence.json" ALERTS = ROOT / "alerts.jsonl" +KEPT = Path.home() / ".local" / "state" / "agents-council" HISTORY = ROOT / "queue-history.jsonl" -KEPT_HISTORY = Path.home() / ".local" / "state" / "agents-council" / "queue-history.jsonl" +KEPT_HISTORY = KEPT / "queue-history.jsonl" +ACCEPTED = ROOT / "security-accepted.jsonl" +KEPT_ACCEPTED = KEPT / "security-accepted.jsonl" +MIRRORED = ((HISTORY, KEPT_HISTORY), (ACCEPTED, KEPT_ACCEPTED)) ALERT_TTL = 3600 LOCK_TTL = 1800 GB = 1e9 @@ -200,28 +205,33 @@ def history_rows(text: str) -> list[tuple[str, str]]: return rows -def keep_history() -> None: +def keep_rows(origin: Path, target: Path) -> None: # The agents' sandbox cannot write ~/.local/state and /tmp is wiped on reboot, so this hook keeps the rows. # The kept file takes the source's mtime: an unchanged source is skipped without parsing. try: - source = HISTORY.stat().st_mtime_ns - if KEPT_HISTORY.exists() and KEPT_HISTORY.stat().st_mtime_ns == source: + source = origin.stat().st_mtime_ns + if target.exists() and target.stat().st_mtime_ns == source: return - KEPT_HISTORY.parent.mkdir(parents=True, exist_ok=True) - with KEPT_HISTORY.open("a+", encoding="utf-8") as kept: + target.parent.mkdir(parents=True, exist_ok=True) + with target.open("a+", encoding="utf-8") as kept: fcntl.flock(kept, fcntl.LOCK_EX) kept.seek(0) known = {row_id for row_id, _ in history_rows(kept.read())} - for row_id, line in history_rows(HISTORY.read_text(encoding="utf-8")): + for row_id, line in history_rows(origin.read_text(encoding="utf-8")): if row_id not in known: known.add(row_id) kept.write(line + "\n") kept.flush() - os.utime(KEPT_HISTORY, ns=(source, source)) + os.utime(target, ns=(source, source)) except (OSError, ValueError, TypeError): pass +def keep_history() -> None: + for origin, target in MIRRORED: + keep_rows(origin, target) + + def cmd_watch(cfg: dict) -> int: payload = json.load(sys.stdin) keep_history() diff --git a/tests/test_security.py b/tests/test_security.py index 08ce0a3..30d0a87 100644 --- a/tests/test_security.py +++ b/tests/test_security.py @@ -2,6 +2,7 @@ from __future__ import annotations +import hashlib import json import os import shutil @@ -19,6 +20,8 @@ FAKE = """#!{python} import json, os, pathlib, sys args = sys.argv[1:] pathlib.Path(__file__).with_name("argv.json").write_text(json.dumps(args)) +pathlib.Path(__file__).with_name("home.txt").write_text(os.environ["HOME"]) +pathlib.Path(os.environ["HOME"], ".medusa").mkdir() target, out = pathlib.Path(args[1]), pathlib.Path(args[args.index("-o") + 1]) out.mkdir(parents=True) findings = [] @@ -26,7 +29,10 @@ for path in sorted(p for p in target.rglob("*") if p.is_file()): for number, text in enumerate(path.read_text().splitlines(), 1): for severity in ("CRITICAL", "HIGH", "MEDIUM", "LOW", "INFO"): if "MARK-" + severity in text: - findings.append({{"severity": severity, "file": str(path), "line": number, "code": text}}) + findings.append({{ + "severity": severity, "rule_id": "FAKE-" + severity, "file": str(path), "line": number, + "code": text, + }}) report = {{"issues": findings}} if os.environ.get("FAKE_MEDUSA_SHAPE") == "other" else {{"findings": findings}} (out / "medusa-scan-20261104-120000.json").write_text(json.dumps(report)) """ @@ -48,6 +54,7 @@ class TestSecurity(unittest.TestCase): self.env = { "PATH": f"{self.bin}:{GIT_DIR}", "HOME": str(self.base), + "AGENTS_COUNCIL_ROOT": str(self.base / "council"), "GIT_CONFIG_NOSYSTEM": "1", "GIT_AUTHOR_NAME": "Test", "GIT_AUTHOR_EMAIL": "test@example.org", @@ -94,7 +101,67 @@ class TestSecurity(unittest.TestCase): {k: found[k] for k in ("security", "files", "critical", "high", "medium", "low", "info")}, {"security": 63, "files": 2, "critical": 0, "high": 1, "medium": 1, "low": 1, "info": 1}, ) - self.assertEqual(found["findings"][0], {"severity": "high", "path": "app.py", "line": 2}) + self.assertEqual( + found["findings"][0], + { + "severity": "high", "rule": "FAKE-HIGH", "path": "app.py", "line": 2, + "sha1": hashlib.sha1(f"token = '{PLACEHOLDER}' # MARK-HIGH".encode()).hexdigest(), + }, + ) + self.assertEqual(found["accepted"], []) + home = Path((self.bin / "home.txt").read_text(encoding="utf-8")) + self.assertNotEqual(home, self.base) + self.assertTrue(home.name == "home" and home.parent.name.startswith("kvb-security-"), home) + + def accept(self, *flags: str) -> subprocess.CompletedProcess: + return subprocess.run( + [sys.executable, str(SECURITY), "accept", "--repo", str(self.repo), *flags], + cwd=self.base, env=self.env, capture_output=True, text=True, check=False, + ) + + def test_accepted_false_positive_leaves_the_score_until_its_line_changes(self): + line = "x = compute() # MARK-CRITICAL" + self.write("app.py", f"print(1)\n{line}\n") + sha1 = hashlib.sha1(line.encode()).hexdigest() + result = self.accept("--rule", "FAKE-CRITICAL", "--path", "app.py", "--sha1", sha1, "--reason", "fixture marker") + self.assertEqual(result.returncode, 0, result.stderr) + listing = self.base / "council" / "security-accepted.jsonl" + row = json.loads(listing.read_text(encoding="utf-8")) + self.assertEqual( + {k: row[k] for k in ("rule", "path", "sha1", "reason", "repo")}, + {"rule": "FAKE-CRITICAL", "path": "app.py", "sha1": sha1, "reason": "fixture marker", + "repo": str(self.repo.resolve())}, + ) + result = self.scan(env=self.env) + self.assertEqual( + (result.stdout, result.returncode), + ("🛡️ security 100 % (medusa: 0 findings, 1 accepted in 1 file)\n", 0), + ) + accepted = json.loads(self.scan("--json", env=self.env).stdout)["accepted"] + self.assertEqual([(a["rule"], a["path"], a["line"]) for a in accepted], [("FAKE-CRITICAL", "app.py", 2)]) + self.write("app.py", "print(1)\nx = compute(2) # MARK-CRITICAL\n") + result = self.scan(env=self.env) + self.assertEqual( + (result.stdout, result.returncode), + ("🛡️ security 50 % (medusa: 1 finding in 1 file: critical app.py:2)\n", 1), + ) + + def test_kept_rows_count_and_broken_rows_stop(self): + line = "y = other() # MARK-HIGH" + self.write("app.py", f"{line}\n") + kept = self.base / ".local" / "state" / "agents-council" / "security-accepted.jsonl" + kept.parent.mkdir(parents=True) + sha1 = hashlib.sha1(line.encode()).hexdigest() + kept.write_text(json.dumps({"rule": "FAKE-HIGH", "path": "app.py", "sha1": sha1}) + "\n", encoding="utf-8") + result = self.scan(env=self.env) + self.assertEqual(result.stdout, "🛡️ security 100 % (medusa: 0 findings, 1 accepted in 1 file)\n") + kept.write_text('{"rule": "FAKE-HIGH"}\n', encoding="utf-8") + result = self.scan(env=self.env) + self.assertEqual(result.returncode, 2) + self.assertIn("security-accepted.jsonl:1: want a JSON row with rule, path and sha1", result.stderr) + result = self.accept("--rule", "R", "--path", "app.py", "--sha1", "XYZ", "--reason", "r") + self.assertEqual(result.returncode, 2) + self.assertIn("--sha1 must be 40 lowercase hex digits", result.stderr) def test_critical_exits_1_and_floors_at_0(self): self.write("app.py", "a # MARK-CRITICAL\nb # MARK-CRITICAL\nc # MARK-CRITICAL\n")