feat(security): reviewed false positives outside the repo, MEDUSA in the sandbox

security.py runs medusa with HOME inside its temporary directory, since medusa creates ~/.medusa on start and the sandboxed home is read-only. security.py accept records a reviewed finding (rule, path, sha1 of the stripped line, reason) in <council root>/security-accepted.jsonl; the watch hook keeps it in ~/.local/state/agents-council/ next to the queue history, and a scan drops matching findings from the score as accepted until the line changes. Accepted so far: MEDUSA-MM-SCAN-058 x3 and MEDUSA-FTSCAN-231 x2 (false positives in tokens.py and score.py), WEB-CMDI-001 and SAST-EXEC-003 on watch.py shell=True (by design, operator config).

🎯 Fix: 93 %
💥 Break: 3 %
💸 Cost: 119 %
⏱️ Time: –
🛡️ Security: 100 %
💰 Value: ?
🏆 Score: 76
🚦 Gate: ⚠️ cost 119 % > 100: ask the operator before the commit

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Kevin Veen-BirkenbachandClaude Opus 5.5 committed 2026-11-04 21:41:17 +01:00
1 parent b18a481d8c
commit 3dc80a1954
4 files changed
+206 -26

No files matched your search

+7
View File
@@ -46,6 +46,13 @@ value. Pass `<n>` as `--security` and the critical count as `--critical`
boundaries, secrets and injection in the diff by hand, then pass
`--security '~<n>'` (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 <root> --rule
<id> --path <path> --sha1 <sha1> --reason "<one line>"`, 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
`<a> accepted`; editing the line brings it back. Accept only after reading
the rule's pattern, and name each acceptance in the commit message.
## Rules
+109 -13
View File
@@ -6,18 +6,34 @@ Args:
--base: git ref the change starts from.
--json: print the figures as JSON instead of the line.
accept --repo <root> --rule <rule_id> --path <repo path> --sha1 <40 hex> --reason <one line>:
record a reviewed false positive instead; --json of a scan prints rule, path
and sha1 of every finding.
Scan:
The regular files `git diff <ref>` names plus untracked ones, symlinks left
out, copied to a temporary tree and scanned once with
`medusa scan <tree> --format json -o <dir> --no-cache`, read from
`medusa scan <tree> --format json -o <dir> --no-cache`, HOME pointing into
the temporary directory because medusa creates ~/.medusa on start, read from
`<dir>/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
<council root>/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 <n> % (medusa: <k> findings in <m> files: <severity> <path>:<line>, ...)`,
m the files scanned, the first 5 locations by severity and `+<rest> more`.
`🛡️ security <n> % (medusa: <k> findings, <a> accepted in <m> files: <severity> <path>:<line>, ...)`,
`, <a> accepted` only when a is above 0, m the files scanned, the first 5
locations by severity and `+<rest> 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 '~<n>' 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())
+21 -11
View File
@@ -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
<root>/queue-history.jsonl whose id it lacks to
~/.local/state/agents-council/queue-history.jsonl
and append the rows of <root>/queue-history.jsonl
(kvb-org-queue) and <root>/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 (<root>/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()
+69 -2
View File
@@ -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")