diff --git a/README.md b/README.md index b57f02b..b3112cb 100644 --- a/README.md +++ b/README.md @@ -81,6 +81,7 @@ flowchart LR org --> org_evening[kvb-org-evening] org --> org_sunset[kvb-org-sunset] org --> org_sunrise[kvb-org-sunrise] + think --> think_choice[kvb-think-choice] org --> org_radio[kvb-org-radio] org --> org_queue[kvb-org-queue] agent --> agent_optimize[kvb-agent-optimize] @@ -125,6 +126,7 @@ flowchart LR criticize[kvb-think-criticize] brainstorm[kvb-think-brainstorm] gaps[kvb-think-gaps] + choice[kvb-think-choice] end subgraph cat_create [create] plan[kvb-create-plan] @@ -380,7 +382,7 @@ Copy the skills into a specific project (`/.agents/skills` and `/.cl make project TARGET=/path/to/repo ``` -Both commands also patch the target's `.claude/settings.local.json` โ€” the machine-local, untracked file, so a project's reviewed `.claude/settings.json` is never rewritten by an install: the caveman and ponytail plugins are enabled so their modes auto-activate on session start, and two reminder hooks are registered โ€” `kvb-dream` points each agent at `/kvb-dream` at most once every twelve hours (stamped per session in `~/.claude/.dream-stamps/`), and `kvb-coach` runs its short form at the first prompt of every full hour, once across all agents (stamped in `~/.claude/.coach-hours/`); the global install also enables the `kvb-dream-notify.timer` systemd user timer, which recommends `/kvb-dream` in a GNOME notification daily at 17:00 local time and catches up once after a missed run. A PreToolUse gate from the `kvb-org-organize` skill blocks edits to files in a git worktree until the session holds a claim (`skills/kvb-org-organize/scripts/council.py`), and its resource watch (`skills/kvb-org-organize/scripts/watch.py`) measures disk, Docker and caches, cleans up low-risk data and blocks new heavy jobs before the disk runs out; it stays inactive until `~/.config/agents-council/config.json` exists. The `kvb-org-radio` skill registers three hooks: its prompt hook delivers `broadcast` and `direct` items on the next prompt, a PostToolBatch hook delivers `security`, `pan-pan` and `mayday` items mid-turn, and an `asyncRewake` Stop hook wakes an idle session for a `mayday` or a `direct` item sent with `--wake`. The PreToolUse hook of the `kvb-org-queue` skill sends heavy commands through `queue.py hold` once the config sets `queue_enforce`. Two hooks of the `kvb-org-perform` skill (`skills/kvb-org-perform/scripts/log.py`) log every Agent spawn with its task class, model and effort, and store the quality feedback a sandboxed session prints, in `~/.local/state/agents-council/subagent-model.jsonl`. Existing settings are preserved. +Both commands also patch the target's `.claude/settings.local.json` โ€” the machine-local, untracked file, so a project's reviewed `.claude/settings.json` is never rewritten by an install: the caveman and ponytail plugins are enabled so their modes auto-activate on session start, and two reminder hooks are registered โ€” `kvb-dream` points each agent at `/kvb-dream` at most once every twelve hours (stamped per session in `~/.claude/.dream-stamps/`), and `kvb-coach` runs its short form at the first prompt of every full hour, once across all agents (stamped in `~/.claude/.coach-hours/`); the global install also enables the `kvb-dream-notify.timer` systemd user timer, which recommends `/kvb-dream` in a GNOME notification daily at 17:00 local time and catches up once after a missed run. A PreToolUse gate from the `kvb-org-organize` skill blocks edits to files in a git worktree until the session holds a claim (`skills/kvb-org-organize/scripts/council.py`), and its resource watch (`skills/kvb-org-organize/scripts/watch.py`) measures disk, Docker and caches, cleans up low-risk data and blocks new heavy jobs before the disk runs out; it stays inactive until `~/.config/agents-council/config.json` exists. The `kvb-org-radio` skill registers three hooks: its prompt hook delivers `broadcast` and `direct` items on the next prompt, a PostToolBatch hook delivers `security`, `pan-pan` and `mayday` items mid-turn, and an `asyncRewake` Stop hook wakes an idle session for a `mayday` or a `direct` item sent with `--wake`. The PreToolUse hook of the `kvb-org-queue` skill sends heavy commands through `queue.py hold` once the config sets `queue_enforce`. The global install also installs the MEDUSA security scanner with pipx for `kvb-check-security`, unless `medusa` is already on the PATH. Two hooks of the `kvb-org-perform` skill (`skills/kvb-org-perform/scripts/log.py`) log every Agent spawn with its task class, model and effort, and store the quality feedback a sandboxed session prints, in `~/.local/state/agents-council/subagent-model.jsonl`. Existing settings are preserved. Restart your agent afterwards so it loads the new skills. diff --git a/scripts/install.sh b/scripts/install.sh index f2611a4..40fea09 100644 --- a/scripts/install.sh +++ b/scripts/install.sh @@ -73,6 +73,15 @@ if [[ "${TARGET}" == "${HOME}" ]]; then else warn "skills: codebase-memory-mcp is not runnable from PATH; the code graph stays off (see kvb-agent-optimize)." fi + if command -v medusa &>/dev/null; then + log "skills: the MEDUSA security scanner is installed" + elif ! command -v pipx &>/dev/null; then + warn "skills: pipx not found; install the MEDUSA scanner with 'pip install --user medusa-security'. Until then kvb-check-security reviews by hand." + elif pipx install medusa-security; then + log "skills: installed the MEDUSA security scanner for kvb-check-security" + else + warn "skills: the MEDUSA install failed; kvb-check-security reviews by hand." + fi fi log "skills: done. Restart your agent to load the skills." diff --git a/skills/kvb-check-confidence/SKILL.md b/skills/kvb-check-confidence/SKILL.md index 0935156..987f9aa 100644 --- a/skills/kvb-check-confidence/SKILL.md +++ b/skills/kvb-check-confidence/SKILL.md @@ -15,11 +15,13 @@ and never estimates one itself. **Break risk** (percent): probability it breaks anything else, from the `kvb-check-break-risk` skill. Two numbers, never one blended figure. 2. **Score**: pass both, with efficiency from the `kvb-check-efficiency` - skill and speed from the `kvb-check-speed` skill, to + skill, speed from the `kvb-check-speed` skill and security and the + critical count from `security.py` of the `kvb-check-security` skill, to `${CLAUDE_SKILL_DIR}/../kvb-check-score/scripts/score.py --fix --break - --efficiency --speed ` and show its score line and gate line - verbatim. Exit 1 fails the 90/5 rule; exit 3 means ask the operator again - before the commit, naming the cause the gate line prints. + --efficiency --speed --security --critical ` + and show its score line and gate line verbatim. Exit 4 forbids the commit + (critical finding); exit 1 fails the 90/5 rule; exit 3 means ask the + operator again before the commit, naming the causes the gate line prints. 3. **Evidence**: the strongest proof lines only: what was executed, reproduced, enumerated, or read from an authoritative source. Mark proven (ran it, saw it) apart from derived (reasoned statically). diff --git a/skills/kvb-check-score/SKILL.md b/skills/kvb-check-score/SKILL.md index 0c1167f..5b383dc 100644 --- a/skills/kvb-check-score/SKILL.md +++ b/skills/kvb-check-score/SKILL.md @@ -1,7 +1,7 @@ --- name: kvb-check-score description: > - Score fix, break, efficiency and speed; run wherever confidence is reported. + Score fix, break, efficiency, speed and security; run wherever confidence is reported. --- Run it wherever fix confidence and break risk are reported: every @@ -10,7 +10,7 @@ One script call prints two lines; paste them verbatim, they replace prose numbers. ```bash -python3 ${CLAUDE_SKILL_DIR}/scripts/score.py --fix <0-100> --break <0-100> --efficiency --speed +python3 ${CLAUDE_SKILL_DIR}/scripts/score.py --fix <0-100> --break <0-100> --efficiency --speed --security --critical ``` | Emoji | Value | Source | @@ -19,37 +19,49 @@ python3 ${CLAUDE_SKILL_DIR}/scripts/score.py --fix <0-100> --break <0-100> --eff | ๐Ÿ’ฅ | break risk | `kvb-check-break-risk` | | ๐ŸŒฑ | efficiency | `tokens.py` (skills) or `measure.py` (code) of `kvb-check-efficiency` | | โšก | speed | `measure.py`, see `kvb-check-speed` | +| ๐Ÿ›ก๏ธ | security, critical count | `security.py` of `kvb-check-security`; `~` from a review by hand | | ๐Ÿ† | score | this script | `-` means not measured or not relevant (skills have no speed) and counts as 100; it shows as `โ€“`. `new` (changed skill text without a base) counts as 100 in the score, yet still asks. Never invent a value: measure it or pass `-`. -Score = `fix * (100 - break) / 100 * 100 / efficiency * 100 / speed`, +Score = `fix * (100 - break) / 100 * 100 / efficiency * 100 / speed * security / 100`, capped to 0..100 and rounded. ## Gate -The second line is the verdict: +The second line is the verdict, first match wins: -- `โœ… gate: 90/5, efficiency and speed <= 100` (exit 0): commit as usual. - 90/5 still needs `gate: green` of `kvb-check-gate`. +- `โ›” gate: critical security finding: do not commit` (exit 4): the commit + call is not sent, whatever else holds and whatever the operator approved + before. Fix the finding, rescan, score again. - `โ›” gate: fix % < 90` or `โ›” gate: break % > 5` (exit 1): the 90/5 rule fails, which bars PR labels and the split bar, not a commit: show it in the block and commit only on the operator's word. Close the gap as the last section of `kvb-check-break-risk` says. -- `โš ๏ธ gate: efficiency % > 100: ask the operator before the commit`, the - same for speed (exit 3): the change costs more than its base. Before the - commit, ask the operator with both lines and what the extra cost buys; - commit only on a yes, which never carries over to another commit. +- `โš ๏ธ gate: : ask the operator before the commit` (exit 3), naming + each of `efficiency % > 100`, `speed % > 100` and + `security % < 100`: the change costs more than its base or carries + findings. Before the commit, ask the operator with the block and what the + cost buys or why each finding is acceptable; commit only on a yes, which + never carries over to another commit. +- `โœ… gate: 90/5, efficiency and speed <= 100, security 100 %` (exit 0): + commit as usual. 90/5 still needs `gate: green` of `kvb-check-gate`. - Exit 2: usage error. -## Commit blocks +## Tables -A commit block puts one line above the two: +`--table` prints the same values as one markdown table instead of the two +lines: ```text -๐Ÿ“ ยท ๐ŸŒฟ ยท ๐Ÿ“Š -๐ŸŽฏ fix 95 % ยท ๐Ÿ’ฅ break 2 % ยท ๐ŸŒฑ efficiency 97 % ยท โšก speed โ€“ ยท ๐Ÿ† score 96 -โœ… gate: 90/5, efficiency and speed <= 100 +| ๐ŸŽฏ Fix | ๐Ÿ’ฅ Break | ๐ŸŒฑ Efficiency | โšก Speed | ๐Ÿ›ก๏ธ Security | ๐Ÿ† Score | Gate | ``` + +`--commit-table ` leads that table with `๐Ÿ“ Worktree | ๐ŸŒฟ Branch | +๐Ÿ“Š Diff stat`, read from git: the top level, the current branch (or +`detached at `) and the staged `git diff --cached --shortstat` as +` files +/-`. Every commit block is this table. After the +commit, add `--commit `: the diff stat of that commit and a `๐Ÿ”– Commit` +column. diff --git a/skills/kvb-check-score/scripts/score.py b/skills/kvb-check-score/scripts/score.py index 4492d59..1282c68 100755 --- a/skills/kvb-check-score/scripts/score.py +++ b/skills/kvb-check-score/scripts/score.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""Fold fix confidence, break risk, efficiency and speed into one score line and one gate line. +"""Fold fix confidence, break risk, efficiency, speed and security into one score and one gate verdict. Args: --fix: fix confidence in percent, 0 to 100. @@ -7,29 +7,48 @@ Args: --efficiency: efficiency in percent above 0; `-` when not measured or not relevant, `new` when the change has no base. Both count as 100 in the score. --speed: speed in percent above 0, `-` or `new` as for --efficiency. + --security: security in percent, 0 to 100, from security.py of kvb-check-security; + `~` for a value from a review by hand. + --critical: count of critical security findings. + --table: print one markdown table instead of the two lines. + --commit-table : the same table, led by the worktree, branch and staged + diff stat of that repository. + --commit : with --commit-table, the diff stat of that commit instead of + the staged one, plus a commit column. Output: - Line 1: `๐ŸŽฏ fix % ยท ๐Ÿ’ฅ break % ยท ๐ŸŒฑ efficiency % ยท โšก speed % ยท ๐Ÿ† score `, - `โ€“` in place of ` %` for `-`, `new` for `new`. Score is - fix * (100 - break) / 100 * 100 / efficiency * 100 / speed, capped to 0..100, rounded half up. + Line 1: `๐ŸŽฏ fix % ยท ๐Ÿ’ฅ break % ยท ๐ŸŒฑ efficiency % ยท โšก speed % ยท + ๐Ÿ›ก๏ธ security % ยท ๐Ÿ† score `, `โ€“` in place of ` %` for `-`, `new` for `new`. + Score is fix * (100 - break) / 100 * 100 / efficiency * 100 / speed * security / 100, + capped to 0..100, rounded half up. Line 2: the gate verdict, `โœ… gate: ...`, `โ›” gate: ...` or `โš ๏ธ gate: ...`. Exit codes: 0 gate passed, 1 fix under 90 or break over 5, 2 usage error, - 3 efficiency or speed over 100 or new: ask the operator before the commit. + 3 efficiency or speed over 100 or new, or security under 100: ask the operator + before the commit, 4 a critical security finding: do not commit. """ from __future__ import annotations import argparse import math +import re +import subprocess import sys FIX_MIN = 90 BREAK_MAX = 5 COST_MAX = 100 +SAFE = 100 DASH = "-" NEW = "new" +ESTIMATE = "~" +SCORE_HEADER = ("๐ŸŽฏ Fix", "๐Ÿ’ฅ Break", "๐ŸŒฑ Efficiency", "โšก Speed", "๐Ÿ›ก๏ธ Security", "๐Ÿ† Score", "Gate") +COMMIT_HEADER = ("๐Ÿ“ Worktree", "๐ŸŒฟ Branch", "๐Ÿ“Š Diff stat") +SHORTSTAT = re.compile( + r"(\d+) files? changed(?:, (\d+) insertions?\(\+\))?(?:, (\d+) deletions?\(-\))?" +) def percent(text: str) -> float: @@ -54,6 +73,17 @@ def cost(text: str) -> float | str: return value +def security(text: str) -> tuple[float, str]: + mark = ESTIMATE if text.startswith(ESTIMATE) else "" + return percent(text.removeprefix(ESTIMATE)), mark + + +def count(text: str) -> int: + if not text.isdigit(): + raise argparse.ArgumentTypeError(f"{text!r} is not a count") + return int(text) + + def number(value: float) -> str: return f"{value:g}" @@ -70,8 +100,11 @@ def factor(value: float | str) -> float: return 100 / value if isinstance(value, float) else 1.0 -def score(fix: float, risk: float, efficiency: float | str, speed: float | str) -> int: - raw = fix * (100 - risk) / 100 * factor(efficiency) * factor(speed) +def score(args: argparse.Namespace) -> int: + raw = ( + args.fix * (100 - args.risk) / 100 * factor(args.efficiency) * factor(args.speed) + * args.security[0] / 100 + ) return math.floor(min(max(raw, 0.0), 100.0) + 0.5) @@ -83,18 +116,67 @@ def over(name: str, value: float | str) -> str | None: return None -def gate(fix: float, risk: float, efficiency: float | str, speed: float | str) -> tuple[str, int]: +def gate(args: argparse.Namespace) -> tuple[str, str, int]: + if args.critical > 0: + return "โ›”", "critical security finding: do not commit", 4 failed = [] - if fix < FIX_MIN: - failed.append(f"fix {number(fix)} % < {FIX_MIN}") - if risk > BREAK_MAX: - failed.append(f"break {number(risk)} % > {BREAK_MAX}") + if args.fix < FIX_MIN: + failed.append(f"fix {number(args.fix)} % < {FIX_MIN}") + if args.risk > BREAK_MAX: + failed.append(f"break {number(args.risk)} % > {BREAK_MAX}") + value, mark = args.security + unsafe = f"security {mark}{number(value)} % < {SAFE}" if value < SAFE else None + asks = [text for text in (over("efficiency", args.efficiency), over("speed", args.speed), unsafe) if text] + ask = f"{', '.join(asks)}: ask the operator before the commit" if asks else "" if failed: - return f"โ›” gate: {', '.join(failed)}", 1 - costly = [text for text in (over("efficiency", efficiency), over("speed", speed)) if text] - if costly: - return f"โš ๏ธ gate: {', '.join(costly)}: ask the operator before the commit", 3 - return f"โœ… gate: {FIX_MIN}/{BREAK_MAX}, efficiency and speed <= {COST_MAX}", 0 + return "โ›”", "; ".join(part for part in (", ".join(failed), ask) if part), 1 + if asks: + return "โš ๏ธ", ask, 3 + return "โœ…", f"{FIX_MIN}/{BREAK_MAX}, efficiency and speed <= {COST_MAX}, security {SAFE} %", 0 + + +def git(repo: str, *args: str) -> str: + try: + result = subprocess.run(["git", *args], cwd=repo, capture_output=True, text=True, check=False) + except OSError as error: + print(f"score: git {args[0]} in {repo}: {error}", file=sys.stderr) + sys.exit(2) + if result.returncode != 0: + print(f"score: git {args[0]} in {repo}: {result.stderr.strip()}", file=sys.stderr) + sys.exit(2) + return result.stdout.strip() + + +def diffstat(shortstat: str) -> str: + match = SHORTSTAT.search(shortstat) + files, added, removed = (int(g or 0) for g in match.groups()) if match else (0, 0, 0) + return f"{files} file{'' if files == 1 else 's'} +{added}/-{removed}" + + +def commit_cells(repo: str, commit: str | None) -> tuple[list[str], list[str]]: + branch = git(repo, "branch", "--show-current") or f"detached at {git(repo, 'rev-parse', '--short', 'HEAD')}" + if commit is None: + stat = git(repo, "diff", "--cached", "--shortstat") + return list(COMMIT_HEADER), [git(repo, "rev-parse", "--show-toplevel"), branch, diffstat(stat)] + stat = git(repo, "show", "--shortstat", "--format=", commit) + return ( + [*COMMIT_HEADER, "๐Ÿ”– Commit"], + [git(repo, "rev-parse", "--show-toplevel"), branch, diffstat(stat), git(repo, "rev-parse", "--short", commit)], + ) + + +def cell(text: str) -> str: + return text.replace("|", "\\|") + + +def table(header: list[str], row: list[str]) -> str: + return "\n".join( + ( + "| " + " | ".join(header) + " |", + "|" + " --- |" * len(header), + "| " + " | ".join(cell(value) for value in row) + " |", + ) + ) def main() -> int: @@ -103,14 +185,34 @@ def main() -> int: parser.add_argument("--break", dest="risk", type=percent, required=True) parser.add_argument("--efficiency", type=cost, required=True) parser.add_argument("--speed", type=cost, required=True) + parser.add_argument("--security", type=security, required=True) + parser.add_argument("--critical", type=count, required=True) + layout = parser.add_mutually_exclusive_group() + layout.add_argument("--table", action="store_true") + layout.add_argument("--commit-table", metavar="REPO") + parser.add_argument("--commit", metavar="SHA") args = parser.parse_args() - total = score(args.fix, args.risk, args.efficiency, args.speed) - print( - f"๐ŸŽฏ fix {number(args.fix)} % ยท ๐Ÿ’ฅ break {number(args.risk)} % ยท " - f"๐ŸŒฑ efficiency {shown(args.efficiency)} ยท โšก speed {shown(args.speed)} ยท ๐Ÿ† score {total}" - ) - verdict, code = gate(args.fix, args.risk, args.efficiency, args.speed) - print(verdict) + if args.commit and not args.commit_table: + parser.error("--commit needs --commit-table") + + total = score(args) + emoji, verdict, code = gate(args) + value, mark = args.security + values = [ + f"{number(args.fix)} %", f"{number(args.risk)} %", shown(args.efficiency), shown(args.speed), + f"{mark}{number(value)} %", str(total), f"{emoji} {verdict}", + ] + if args.commit_table: + header, cells = commit_cells(args.commit_table, args.commit) + print(table([*header, *SCORE_HEADER], [*cells, *values])) + elif args.table: + print(table(list(SCORE_HEADER), values)) + else: + print( + f"๐ŸŽฏ fix {values[0]} ยท ๐Ÿ’ฅ break {values[1]} ยท ๐ŸŒฑ efficiency {values[2]} ยท " + f"โšก speed {values[3]} ยท ๐Ÿ›ก๏ธ security {values[4]} ยท ๐Ÿ† score {total}" + ) + print(f"{emoji} gate: {verdict}") return code diff --git a/skills/kvb-check-security/SKILL.md b/skills/kvb-check-security/SKILL.md index 050d245..205164e 100644 --- a/skills/kvb-check-security/SKILL.md +++ b/skills/kvb-check-security/SKILL.md @@ -11,7 +11,7 @@ MEDUSA is a defensive AI-first security scanner (40,000+ patterns: CVEs, prompt ## Setup Check availability with `medusa --version`. -If missing, install it with `pipx install medusa-security` (fallback: `pip install --user medusa-security`). +The skills repository's `make install` installs it with pipx; if it is still missing, `pipx install medusa-security` (fallback: `pip install --user medusa-security`) is the operator's step. ## Commands @@ -22,6 +22,31 @@ If missing, install it with `pipx install medusa-security` (fallback: `pip insta | Find leaked API keys in AI chat + shell history | `medusa secrets scan` | | Scan one exported file | `medusa secrets scan --path ` | +## Security value of a change + +The ๐Ÿ›ก๏ธ value of the `kvb-check-score` block, 100 % meaning safe: + +```bash +python3 ${CLAUDE_SKILL_DIR}/scripts/security.py --repo --base +``` + +It copies the files changed since ``, untracked ones included and +symlinks left out, to a temporary tree, scans it once with +`medusa scan --format json -o --no-cache` and prints one line: +`๐Ÿ›ก๏ธ security % (medusa: findings in files: :, ...)`. +`` is 100 minus 50 per critical, 25 per high, 10 per medium and 2 per low +finding, at least 0. It shows locations and counts only, never a matched +value. Pass `` as `--security` and the critical count as `--critical` +(`--json` holds it) to `score.py`. + +- Exit 0: 100 %. Exit 3: below 100 without a critical finding; the score + gate asks the operator. Exit 1: a critical finding; the score gate forbids + the commit. +- Exit 4: medusa is missing or its report unreadable. Review trust + 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. + ## Rules - Run scans read-only; never pass `--reveal` and never run `medusa secrets purge` unless the user explicitly asks for it. diff --git a/skills/kvb-check-security/scripts/security.py b/skills/kvb-check-security/scripts/security.py new file mode 100755 index 0000000..46d33f8 --- /dev/null +++ b/skills/kvb-check-security/scripts/security.py @@ -0,0 +1,186 @@ +#!/usr/bin/env python3 +"""Scan the files a change touches with MEDUSA and fold the findings into one security percentage. + +Args: + --repo: git repository whose working tree holds the change. + --base: git ref the change starts from. + --json: print the figures as JSON instead of the line. + +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-*.json`: a `findings` list whose entries carry + `severity` (critical, high, medium, low or info), `file` and `line`. + +Output: + `๐Ÿ›ก๏ธ security % (medusa: findings in files: :, ...)`, + 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. + +Exit codes: + 0 security 100, 1 a critical finding, 2 usage error, 3 under 100 without a + critical finding, 4 medusa missing or its report unreadable: review the diff by hand. +""" + +from __future__ import annotations + +import argparse +import json +import shutil +import subprocess +import sys +import tempfile +from pathlib import Path + +WEIGHTS = {"critical": 50, "high": 25, "medium": 10, "low": 2, "info": 0} +SAFE = 100 +SHOWN = 5 +HAND = ( + "review trust boundaries, secrets and injection in the diff by hand, " + "then pass --security '~' to kvb-check-score" +) + + +class Unreadable(Exception): + pass + + +def fail(message: str) -> None: + print(f"security: {message}", file=sys.stderr) + sys.exit(2) + + +def git(repo: Path, *args: str) -> str: + result = subprocess.run(["git", *args], cwd=repo, capture_output=True, check=False) + if result.returncode != 0: + fail(result.stderr.decode(errors="replace").strip() or f"git {args[0]} failed in {repo}") + return result.stdout.decode(errors="replace") + + +def changed(root: Path, base: str) -> list[str]: + # Trip-wire: with rename detection a moved file lists only its new path. + listed = git(root, "diff", "--no-renames", "-z", "--name-only", base, "--") + listed += git(root, "ls-files", "-z", "--others", "--exclude-standard") + return sorted({p for p in listed.split("\0") if p}) + + +def copy(root: Path, paths: list[str], tree: Path) -> int: + copied = 0 + for name in paths: + source = root / name + # Trip-wire: a symlink's target may lie outside the repository; even a copied link lets the scan read it. + if source.is_symlink() or not source.is_file(): + continue + target = tree / name + target.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source, target) + copied += 1 + return copied + + +def located(name: str, tree: Path) -> str: + path = Path(name) + if not path.is_absolute(): + path = tree / path if (tree / path).exists() else tree.parent / path + try: + return path.resolve().relative_to(tree.resolve()).as_posix() + except ValueError: + return name + + +def findings(reports: Path, tree: Path) -> list[dict]: + files = sorted(reports.glob("medusa-scan-*.json")) if reports.is_dir() else [] + if not files: + raise Unreadable(f"no medusa-scan-*.json in {reports}") + try: + report = json.loads(files[-1].read_text(encoding="utf-8")) + except (OSError, ValueError) as error: + raise Unreadable(f"{files[-1].name}: {error}") from None + entries = report.get("findings") if isinstance(report, dict) else None + if not isinstance(entries, list): + raise Unreadable(f"{files[-1].name} holds no findings list") + found = [] + for entry in entries: + 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") + found.append({ + "severity": severity, + "path": located(str(entry.get("file") or "?"), tree), + "line": line if isinstance(line, int) else None, + }) + rank = list(WEIGHTS) + return sorted(found, key=lambda f: (rank.index(f["severity"]), f["path"], f["line"] or 0)) + + +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.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, + ) + try: + return scanned, findings(reports, tree) + except Unreadable as error: + raise Unreadable(f"{error} (medusa exit {result.returncode})") from None + + +def plural(n: int, noun: str) -> str: + return f"{n} {noun}{'' if n == 1 else 's'}" + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--repo", required=True) + parser.add_argument("--base", required=True) + parser.add_argument("--json", action="store_true") + args = parser.parse_args() + + repo = Path(args.repo) + if not repo.is_dir(): + fail(f"{repo} is not a directory") + root = Path(git(repo, "rev-parse", "--show-toplevel").strip()) + if subprocess.run( + ["git", "rev-parse", "--verify", "--quiet", f"{args.base}^{{commit}}"], cwd=root, capture_output=True, check=False + ).returncode != 0: + fail(f"unknown ref {args.base!r}") + + medusa = shutil.which("medusa") + problem = None if medusa else "medusa missing" + if medusa: + try: + scanned, found = scan(medusa, root, args.base) + except Unreadable as error: + problem = f"medusa report unreadable: {error}" + if problem: + if args.json: + print(json.dumps({"security": None, "problem": problem})) + else: + print(f"๐Ÿ›ก๏ธ security โ€“ ({problem}: {HAND})") + return 4 + + 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})) + 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})") + if counts["critical"]: + return 1 + return 3 if value < SAFE else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/skills/kvb-git-commit-next/SKILL.md b/skills/kvb-git-commit-next/SKILL.md index 1530775..f63d5d9 100644 --- a/skills/kvb-git-commit-next/SKILL.md +++ b/skills/kvb-git-commit-next/SKILL.md @@ -19,12 +19,15 @@ description: > 3. Nothing similar found, or the operator picked the staged version: run the kvb-git-commit-no-verify skill on the currently staged changes (kvb-code-no-comments audit and kvb-check-legal check first, then the - full staged diff and its score block of ๐Ÿ“ worktree, ๐ŸŒฟ branch, - ๐Ÿ“Š diff stat, score line and gate line, then `git commit --no-verify`, - exact staging index only). Its approval-prompt duty is binding: the commit - call's approval prompt names worktree, branch, diff stat and the score - values, or the call is not sent. Gate exit 1 is shown and the operator's - word decides; exit 3 asks the operator again first, naming the cause. + full staged diff and the kvb-check-score commit table of ๐Ÿ“ worktree, + ๐ŸŒฟ branch, ๐Ÿ“Š diff stat, scores with ๐Ÿ›ก๏ธ security and gate, then + `git commit --no-verify`, exact staging index only, then the table again + with the hash). Its approval-prompt duty is binding: the commit call's + approval prompt names worktree, branch, diff stat and the score values + with ๐Ÿ›ก๏ธ security, or the call is not sent. Gate exit 4 (critical security + finding): no commit, the call is never sent. Exit 1 is shown and the + operator's word decides; exit 3 asks the operator again first, naming + each cause. 4. Group the remaining uncommitted changes into logical bundles (one concern per bundle: same failure class, mechanism or role) and stage the next one; never mix concerns to save a commit. Staging happens here and nowhere @@ -37,8 +40,11 @@ description: > --file --match `, first with `--dry-run`. 5. For the newly staged bundle run the kvb-think-explain skill (what the changes do and how the pieces interact, file:line-anchored), then the - kvb-check-confidence skill (its kvb-check-score block, evidence, - residuals). -6. Close with the staged bundle's score block (๐Ÿ“ worktree, ๐ŸŒฟ branch, - ๐Ÿ“Š diff stat, score line, gate line), what remains unstaged, and wait for - the operator's commit decision; never auto-commit the newly staged bundle. + kvb-check-confidence skill (its kvb-check-score block with ๐Ÿ›ก๏ธ security + from `security.py` of kvb-check-security, evidence, residuals). +6. Close with the staged bundle's commit table (`score.py ... + --commit-table `: ๐Ÿ“ worktree, ๐ŸŒฟ branch, ๐Ÿ“Š diff stat, scores, + gate), what remains unstaged, and wait for the operator's commit + decision; never auto-commit the newly staged bundle. A gate exit 4 says + up front that the bundle cannot be committed until its critical finding + is fixed. diff --git a/skills/kvb-git-commit-no-verify/SKILL.md b/skills/kvb-git-commit-no-verify/SKILL.md index 19337c8..ab575eb 100644 --- a/skills/kvb-git-commit-no-verify/SKILL.md +++ b/skills/kvb-git-commit-no-verify/SKILL.md @@ -10,24 +10,26 @@ description: > then the kvb-check-legal check over the staging index. A legal blocker stops here. 2. Show in the message right before the `git commit` call, never only after - it, the full staged diff (`git diff --cached`), then the score block: - - ๐Ÿ“ **Worktree**: absolute path from `git rev-parse --show-toplevel`. - - ๐ŸŒฟ **Branch**: `git branch --show-current`, or `detached at `. - - ๐Ÿ“Š **Diff stat** of what gets committed: `git diff --cached - --shortstat`, e.g. `7 files +163/-108`. - - The score line and gate line of the kvb-check-score skill for the staged - bundle, as the kvb-check-confidence skill produces them: break risk - derived through the kvb-check-break-risk skill, never estimated. + it, the full staged diff (`git diff --cached`), then the commit table of + the kvb-check-score skill (`score.py ... --commit-table `): + ๐Ÿ“ worktree, ๐ŸŒฟ branch, ๐Ÿ“Š staged diff stat, then fix, break, efficiency, + speed, security, score and gate of the staged bundle, as the + kvb-check-confidence skill produces them: break risk derived through the + kvb-check-break-risk skill, never estimated. 3. Approval-prompt duty: the commit call's approval prompt (Claude Code: the Bash `description`; other runtimes: their prompt text) names the worktree - path, the branch, the diff stat and the score values (fix, break, - efficiency, speed, score). Missing any item: do not send the call. Gate - exit 1 (below 90/5) is shown, the operator's word decides; exit 3 means - ask the operator again before the commit, naming the cause the gate line - prints. The same duty holds for - every call that creates a commit, `git merge` included (there the - incoming diff stat, and the incoming diff in the message before). + path, the branch, the diff stat and the score values (๐ŸŽฏ fix, ๐Ÿ’ฅ break, + ๐ŸŒฑ efficiency, โšก speed, ๐Ÿ›ก๏ธ security, ๐Ÿ† score). Missing any item: do not + send the call. Gate exit 4 (critical security finding) means no commit: + the call is never sent, not even on the operator's word; fix and rescan + first. Exit 1 (below 90/5) is shown, the operator's word decides; exit 3 + means ask the operator again before the commit, naming each cause the + gate prints. The same duty holds for every call that creates a commit, + `git merge` included (there the incoming diff stat, and the incoming diff + in the message before). 4. Commit exactly the staging index with `git commit --no-verify` and a Conventional-Commits message derived from the staged diff; never add unstaged files, never push. -5. Report the created commit hash and what remains uncommitted. +5. Report the created commit hash, the commit table again with + `--commit ` (the committed diff stat and a ๐Ÿ”– Commit column), and + what remains uncommitted. diff --git a/skills/kvb-think-choice/SKILL.md b/skills/kvb-think-choice/SKILL.md new file mode 100644 index 0000000..06e38b7 --- /dev/null +++ b/skills/kvb-think-choice/SKILL.md @@ -0,0 +1,49 @@ +--- +name: kvb-think-choice +description: > + Offer the open decisions as multiple choice and checkboxes. Trigger on + "auswahl", "offene punkte". +--- + +The operator decides faster by picking than by reading prose. This skill +collects every decision still open and offers each as a choice with the +recommended option first. + +## Step 1: collect + +Gather what waits on the operator, never what an agent can decide itself: + +- This session's own open questions and staged bundles waiting for a commit + word. +- `${CLAUDE_SKILL_DIR}/../kvb-org-operator/scripts/operator.py list --json`: + every row with the check `decision`. +- The `kvb-org-rasci` register (`rasci.py show --wide`): tasks `blocked` on + the Operator as A or C. + +Merge duplicates and drop items a command can settle (those go to +`operator.py run`). Rank by what unblocks most work first. + +## Step 2: shape each item + +- One question per item, two to four options, the recommended option first + and marked as such, each with one line on its consequence. +- Independent yes/no items of one kind (which tasks to start, which rows to + close) become one checkbox question. +- An item whose options need a comparison carries a short preview: the + command, the diff stat or the score line of each option. + +## Step 3: ask + +- **Claude Code**: `AskUserQuestion`, at most four questions per call, + `multiSelect: true` for checkbox questions, `false` otherwise; more items + follow in the next call. +- **Other runtimes**: print the same items as markdown, numbered questions + with lettered options and `- [ ]` boxes, and ask for an answer such as + `1a 2c 3: x, z`. + +## Step 4: act + +Carry out each answer within its own skill's rules (a commit word through +`kvb-git-commit-next`, a decision row through `operator.py done`). A pick +that needs an action outside the sandbox becomes an `operator.py add` row. +An answer is the operator's word for that item only and never carries over. diff --git a/tests/test_score.py b/tests/test_score.py index 2c4ccdf..f73e848 100644 --- a/tests/test_score.py +++ b/tests/test_score.py @@ -1,103 +1,219 @@ -"""Exercise score.py of kvb-check-score: formula, dash and new values, capping, every gate verdict and exit code.""" +"""Exercise score.py of kvb-check-score: formula, dash and new values, security, capping, gate verdicts, tables.""" from __future__ import annotations +import os +import shutil import subprocess import sys +import tempfile import unittest from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[1] SCORE = REPO_ROOT / "skills" / "kvb-check-score" / "scripts" / "score.py" -PASS = "โœ… gate: 90/5, efficiency and speed <= 100" +PASS = "โœ… gate: 90/5, efficiency and speed <= 100, security 100 %" +BASE = {"fix": "95", "break": "2", "efficiency": "100", "speed": "100", "security": "100", "critical": "0"} +HEADER = "| ๐ŸŽฏ Fix | ๐Ÿ’ฅ Break | ๐ŸŒฑ Efficiency | โšก Speed | ๐Ÿ›ก๏ธ Security | ๐Ÿ† Score | Gate |" +ASK = ": ask the operator before the commit" +ENV = {"PATH": os.environ["PATH"]} + + +def flags(values: dict[str, str]) -> list[str]: + return [item for key, value in values.items() for item in (f"--{key}", value)] class TestScore(unittest.TestCase): - def _run(self, *args: str) -> subprocess.CompletedProcess: - return subprocess.run([sys.executable, str(SCORE), *args], capture_output=True, text=True, check=False) + def _run(self, *args: str, cwd: Path) -> subprocess.CompletedProcess: + return subprocess.run( + [sys.executable, str(SCORE), *args], cwd=cwd, env=ENV, capture_output=True, text=True, check=False + ) - def _score(self, fix: str, risk: str, efficiency: str, speed: str) -> tuple[list[str], int]: - result = self._run("--fix", fix, "--break", risk, "--efficiency", efficiency, "--speed", speed) + def _score(self, **values: str) -> tuple[list[str], int]: + result = self._run(*flags({**BASE, **values}), cwd=REPO_ROOT) self.assertEqual(result.stderr, "") return result.stdout.splitlines(), result.returncode def test_formula_and_line_format(self): self.assertEqual( - self._score("95", "4", "100", "100"), - (["๐ŸŽฏ fix 95 % ยท ๐Ÿ’ฅ break 4 % ยท ๐ŸŒฑ efficiency 100 % ยท โšก speed 100 % ยท ๐Ÿ† score 91", PASS], 0), + self._score(fix="95", **{"break": "4"}), + (["๐ŸŽฏ fix 95 % ยท ๐Ÿ’ฅ break 4 % ยท ๐ŸŒฑ efficiency 100 % ยท โšก speed 100 % ยท ๐Ÿ›ก๏ธ security 100 % ยท ๐Ÿ† score 91", + PASS], 0), + ) + lines, code = self._score(efficiency="97", speed="-") + self.assertEqual( + lines[0], "๐ŸŽฏ fix 95 % ยท ๐Ÿ’ฅ break 2 % ยท ๐ŸŒฑ efficiency 97 % ยท โšก speed โ€“ ยท ๐Ÿ›ก๏ธ security 100 % ยท ๐Ÿ† score 96" ) - lines, code = self._score("95", "2", "97", "-") - self.assertEqual(lines[0], "๐ŸŽฏ fix 95 % ยท ๐Ÿ’ฅ break 2 % ยท ๐ŸŒฑ efficiency 97 % ยท โšก speed โ€“ ยท ๐Ÿ† score 96") self.assertEqual(code, 0) - self.assertEqual(self._score("92.5", "0", "100", "100")[0][0].split(" ยท ")[0], "๐ŸŽฏ fix 92.5 %") + self.assertEqual(self._score(fix="92.5")[0][0].split(" ยท ")[0], "๐ŸŽฏ fix 92.5 %") def test_dash_values_count_as_100(self): + lines, code = self._score(fix="92", efficiency="-", speed="-", **{"break": "3"}) self.assertEqual( - self._score("92", "3", "-", "-"), - (["๐ŸŽฏ fix 92 % ยท ๐Ÿ’ฅ break 3 % ยท ๐ŸŒฑ efficiency โ€“ ยท โšก speed โ€“ ยท ๐Ÿ† score 89", PASS], 0), + (lines, code), + (["๐ŸŽฏ fix 92 % ยท ๐Ÿ’ฅ break 3 % ยท ๐ŸŒฑ efficiency โ€“ ยท โšก speed โ€“ ยท ๐Ÿ›ก๏ธ security 100 % ยท ๐Ÿ† score 89", PASS], 0), ) + def test_security_multiplies_and_asks_below_100(self): + lines, code = self._score(fix="100", efficiency="-", speed="-", security="75", **{"break": "0"}) + self.assertEqual( + (lines, code), + (["๐ŸŽฏ fix 100 % ยท ๐Ÿ’ฅ break 0 % ยท ๐ŸŒฑ efficiency โ€“ ยท โšก speed โ€“ ยท ๐Ÿ›ก๏ธ security 75 % ยท ๐Ÿ† score 75", + f"โš ๏ธ gate: security 75 % < 100{ASK}"], 3), + ) + lines, code = self._score(security="~80") + self.assertIn("๐Ÿ›ก๏ธ security ~80 % ยท ๐Ÿ† score 74", lines[0]) + self.assertEqual((lines[1], code), (f"โš ๏ธ gate: security ~80 % < 100{ASK}", 3)) + self.assertEqual(self._score(security="~100")[1], 0) + def test_score_rounds_half_up(self): - for args, total in ( - (("94.5", "0", "-", "-"), 95), - (("90", "5", "100", "100"), 86), - (("91", "5", "-", "-"), 86), - (("97", "0", "97", "-"), 100), + for values, total in ( + ({"fix": "94.5", "break": "0"}, 95), + ({"fix": "90", "break": "5"}, 86), + ({"fix": "91", "break": "5"}, 86), + ({"fix": "97", "break": "0", "efficiency": "97"}, 100), ): - with self.subTest(args=args): - self.assertTrue(self._score(*args)[0][0].endswith(f"๐Ÿ† score {total}")) + with self.subTest(values=values): + self.assertTrue(self._score(**values)[0][0].endswith(f"๐Ÿ† score {total}")) def test_score_is_capped(self): - lines, code = self._score("100", "0", "50", "50") + lines, code = self._score(fix="100", efficiency="50", speed="50", **{"break": "0"}) self.assertTrue(lines[0].endswith("๐Ÿ† score 100")) self.assertEqual((lines[1], code), (PASS, 0)) - lines, code = self._score("0", "100", "-", "-") + lines, code = self._score(fix="0", **{"break": "100"}) self.assertTrue(lines[0].endswith("๐Ÿ† score 0")) self.assertEqual(code, 1) def test_boundaries_pass(self): - self.assertEqual(self._score("90", "5", "100", "100")[1], 0) + self.assertEqual(self._score(fix="90", **{"break": "5"})[1], 0) def test_90_5_failures_exit_1(self): - for args, verdict in ( - (("89.5", "1", "-", "-"), "โ›” gate: fix 89.5 % < 90"), - (("95", "6", "-", "-"), "โ›” gate: break 6 % > 5"), - (("80", "10", "-", "-"), "โ›” gate: fix 80 % < 90, break 10 % > 5"), - (("85", "2", "150", "new"), "โ›” gate: fix 85 % < 90"), + for values, verdict in ( + ({"fix": "89.5", "break": "1"}, "โ›” gate: fix 89.5 % < 90"), + ({"break": "6"}, "โ›” gate: break 6 % > 5"), + ({"fix": "80", "break": "10"}, "โ›” gate: fix 80 % < 90, break 10 % > 5"), + ( + {"fix": "85", "efficiency": "150", "speed": "new", "security": "50"}, + f"โ›” gate: fix 85 % < 90; efficiency 150 % > 100, speed new, security 50 % < 100{ASK}", + ), ): - with self.subTest(args=args): - lines, code = self._score(*args) + with self.subTest(values=values): + lines, code = self._score(**values) self.assertEqual((lines[1], code), (verdict, 1)) - def test_cost_over_100_asks_the_operator_exit_3(self): - cases = { - ("120", "100"): "โš ๏ธ gate: efficiency 120 % > 100: ask the operator before the commit", - ("-", "130"): "โš ๏ธ gate: speed 130 % > 100: ask the operator before the commit", - ("120", "130"): "โš ๏ธ gate: efficiency 120 % > 100, speed 130 % > 100: ask the operator before the commit", - ("100.5", "-"): "โš ๏ธ gate: efficiency 100.5 % > 100: ask the operator before the commit", - ("new", "-"): "โš ๏ธ gate: efficiency new: ask the operator before the commit", - } - for (efficiency, speed), verdict in cases.items(): - with self.subTest(efficiency=efficiency, speed=speed): - lines, code = self._score("95", "2", efficiency, speed) + def test_costs_ask_the_operator_exit_3(self): + for values, verdict in ( + ({"efficiency": "120"}, f"โš ๏ธ gate: efficiency 120 % > 100{ASK}"), + ({"efficiency": "-", "speed": "130"}, f"โš ๏ธ gate: speed 130 % > 100{ASK}"), + ({"efficiency": "100.5", "speed": "-"}, f"โš ๏ธ gate: efficiency 100.5 % > 100{ASK}"), + ({"efficiency": "new", "speed": "-"}, f"โš ๏ธ gate: efficiency new{ASK}"), + ( + {"efficiency": "120", "speed": "130", "security": "90"}, + f"โš ๏ธ gate: efficiency 120 % > 100, speed 130 % > 100, security 90 % < 100{ASK}", + ), + ): + with self.subTest(values=values): + lines, code = self._score(**values) self.assertEqual((lines[1], code), (verdict, 3)) + def test_critical_finding_wins_exit_4(self): + for values in ( + {"critical": "1", "security": "50"}, + {"critical": "2", "fix": "80", "break": "10", "efficiency": "150", "security": "0"}, + ): + with self.subTest(values=values): + lines, code = self._score(**values) + self.assertEqual((lines[1], code), ("โ›” gate: critical security finding: do not commit", 4)) + def test_new_counts_as_100_in_the_score(self): - lines, _ = self._score("95", "0", "new", "-") - self.assertEqual(lines[0], "๐ŸŽฏ fix 95 % ยท ๐Ÿ’ฅ break 0 % ยท ๐ŸŒฑ efficiency new ยท โšก speed โ€“ ยท ๐Ÿ† score 95") + lines, _ = self._score(efficiency="new", speed="-", **{"break": "0"}) + self.assertEqual( + lines[0], "๐ŸŽฏ fix 95 % ยท ๐Ÿ’ฅ break 0 % ยท ๐ŸŒฑ efficiency new ยท โšก speed โ€“ ยท ๐Ÿ›ก๏ธ security 100 % ยท ๐Ÿ† score 95" + ) + + def test_table(self): + result = self._run(*flags({**BASE, "efficiency": "97", "speed": "-"}), "--table", cwd=REPO_ROOT) + self.assertEqual( + result.stdout.splitlines(), + [ + HEADER, + "| --- | --- | --- | --- | --- | --- | --- |", + "| 95 % | 2 % | 97 % | โ€“ | 100 % | 96 | โœ… 90/5, efficiency and speed <= 100, security 100 % |", + ], + ) + self.assertEqual(result.returncode, 0) + result = self._run(*flags({**BASE, "critical": "1"}), "--table", cwd=REPO_ROOT) + self.assertTrue(result.stdout.splitlines()[2].endswith("| โ›” critical security finding: do not commit |")) + self.assertEqual(result.returncode, 4) + + def test_commit_table(self): + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + repo = Path(tmp.name) / "repo" + repo.mkdir() + env = { + "PATH": os.path.dirname(shutil.which("git")), + "HOME": tmp.name, + "GIT_CONFIG_NOSYSTEM": "1", + "GIT_AUTHOR_NAME": "Test", + "GIT_AUTHOR_EMAIL": "test@example.org", + "GIT_COMMITTER_NAME": "Test", + "GIT_COMMITTER_EMAIL": "test@example.org", + } + + def git(*args: str) -> str: + return subprocess.run( + ["git", *args], cwd=repo, env=env, capture_output=True, text=True, check=True + ).stdout.strip() + + git("init", "-q", "-b", "main") + (repo / "a.txt").write_text("one\n", encoding="utf-8") + git("add", "a.txt") + git("commit", "-qm", "base") + (repo / "a.txt").write_text("two\n", encoding="utf-8") + (repo / "b.txt").write_text("three\nfour\n", encoding="utf-8") + git("add", "a.txt", "b.txt") + head = ["๐Ÿ“ Worktree", "๐ŸŒฟ Branch", "๐Ÿ“Š Diff stat"] + score = "| 95 % | 2 % | 100 % | 100 % | 100 % | 93 | โœ… 90/5, efficiency and speed <= 100, security 100 % |" + + result = self._run(*flags(BASE), "--commit-table", str(repo), cwd=repo) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual( + result.stdout.splitlines(), + [ + "| " + " | ".join(head) + " " + HEADER, + "|" + " --- |" * 10, + f"| {repo.resolve()} | main | 2 files +3/-1 " + score, + ], + ) + git("commit", "-qm", "change") + sha = git("rev-parse", "--short", "HEAD") + result = self._run(*flags(BASE), "--commit-table", str(repo), "--commit", "HEAD", cwd=repo) + self.assertEqual(result.stdout.splitlines()[0], "| " + " | ".join([*head, "๐Ÿ”– Commit"]) + " " + HEADER) + self.assertEqual(result.stdout.splitlines()[2], f"| {repo.resolve()} | main | 2 files +3/-1 | {sha} " + score) + git("checkout", "-q", "--detach", "HEAD") + result = self._run(*flags(BASE), "--commit-table", str(repo), cwd=repo) + self.assertEqual(result.stdout.splitlines()[2], f"| {repo.resolve()} | detached at {sha} | 0 files +0/-0 " + score) + self.assertEqual(self._run(*flags(BASE), "--commit-table", tmp.name, cwd=repo).returncode, 2) def test_usage_errors_exit_2(self): - good = {"--fix": "95", "--break": "2", "--efficiency": "100", "--speed": "100"} bad = ( - ("--fix", "101"), ("--fix", "-1"), ("--fix", "nan"), ("--break", "x"), ("--break", "-"), - ("--efficiency", "0"), ("--efficiency", "-5"), ("--efficiency", "abc"), ("--speed", "inf"), + ("fix", "101"), ("fix", "-1"), ("fix", "nan"), ("break", "x"), ("break", "-"), + ("efficiency", "0"), ("efficiency", "-5"), ("efficiency", "abc"), ("speed", "inf"), + ("security", "101"), ("security", "~"), ("security", "~x"), ("security", "-"), + ("critical", "-1"), ("critical", "a"), ("critical", "1.5"), ) - for flag, value in bad: - with self.subTest(flag=flag, value=value): - args = [item for key, val in {**good, flag: value}.items() for item in (key, val)] - result = self._run(*args) + for key, value in bad: + with self.subTest(flag=key, value=value): + result = self._run(*flags({**BASE, key: value}), cwd=REPO_ROOT) self.assertEqual((result.returncode, result.stdout), (2, "")) - self.assertEqual(self._run("--fix", "95", "--break", "2", "--efficiency", "100").returncode, 2) + for missing in ("security", "critical"): + with self.subTest(missing=missing): + values = {k: v for k, v in BASE.items() if k != missing} + self.assertEqual(self._run(*flags(values), cwd=REPO_ROOT).returncode, 2) + for extra in (("--commit", "HEAD"), ("--table", "--commit-table", str(REPO_ROOT))): + with self.subTest(extra=extra): + self.assertEqual(self._run(*flags(BASE), *extra, cwd=REPO_ROOT).returncode, 2) if __name__ == "__main__": diff --git a/tests/test_security.py b/tests/test_security.py new file mode 100644 index 0000000..08ce0a3 --- /dev/null +++ b/tests/test_security.py @@ -0,0 +1,152 @@ +"""Exercise security.py of kvb-check-security against a fake medusa: weights, locations, exit codes, hand review.""" + +from __future__ import annotations + +import json +import os +import shutil +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[1] +SECURITY = REPO_ROOT / "skills" / "kvb-check-security" / "scripts" / "security.py" +GIT_DIR = os.path.dirname(shutil.which("git")) +PLACEHOLDER = "placeholder-value-never-shown" +FAKE = """#!{python} +import json, os, pathlib, sys +args = sys.argv[1:] +pathlib.Path(__file__).with_name("argv.json").write_text(json.dumps(args)) +target, out = pathlib.Path(args[1]), pathlib.Path(args[args.index("-o") + 1]) +out.mkdir(parents=True) +findings = [] +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}}) +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)) +""" +HAND = "review trust boundaries, secrets and injection in the diff by hand, then pass --security '~' to kvb-check-score" + + +class TestSecurity(unittest.TestCase): + def setUp(self): + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + self.base = Path(tmp.name) + self.bin = self.base / "bin" + self.bin.mkdir() + medusa = self.bin / "medusa" + medusa.write_text(FAKE.format(python=sys.executable), encoding="utf-8") + medusa.chmod(0o755) + self.repo = self.base / "repo" + self.repo.mkdir() + self.env = { + "PATH": f"{self.bin}:{GIT_DIR}", + "HOME": str(self.base), + "GIT_CONFIG_NOSYSTEM": "1", + "GIT_AUTHOR_NAME": "Test", + "GIT_AUTHOR_EMAIL": "test@example.org", + "GIT_COMMITTER_NAME": "Test", + "GIT_COMMITTER_EMAIL": "test@example.org", + } + self.git("init", "-q", "-b", "main") + self.write("app.py", "print(1)\n") + self.write("old.py", "x = 1 # MARK-CRITICAL\n") + self.write("gone.py", "y = 2 # MARK-HIGH\n") + self.git("add", "-A") + self.git("commit", "-qm", "base") + + def git(self, *args: str) -> None: + subprocess.run(["git", *args], cwd=self.repo, env=self.env, capture_output=True, check=True) + + def write(self, path: str, text: str) -> None: + (self.repo / path).write_text(text, encoding="utf-8") + + def scan(self, *extra: str, env: dict[str, str]) -> subprocess.CompletedProcess: + return subprocess.run( + [sys.executable, str(SECURITY), "--repo", str(self.repo), "--base", "HEAD", *extra], + cwd=self.base, env=env, capture_output=True, text=True, check=False, + ) + + def test_changed_and_untracked_files_are_weighted(self): + self.write("app.py", f"print(1)\ntoken = '{PLACEHOLDER}' # MARK-HIGH\n") + self.write("new.py", "a = 1 # MARK-LOW\nb = 2\nc = 3 # MARK-MEDIUM\nd = 4 # MARK-INFO\n") + (self.repo / "gone.py").unlink() + (self.repo / "link.py").symlink_to(self.repo / "old.py") + result = self.scan(env=self.env) + self.assertEqual( + result.stdout, + "๐Ÿ›ก๏ธ security 63 % (medusa: 4 findings in 2 files: high app.py:2, medium new.py:3, low new.py:1, " + "info new.py:4)\n", + ) + self.assertEqual(result.returncode, 3, result.stderr) + argv = json.loads((self.bin / "argv.json").read_text(encoding="utf-8")) + self.assertEqual([argv[0], *argv[2:4], argv[4], argv[6]], ["scan", "--format", "json", "-o", "--no-cache"]) + result = self.scan("--json", env=self.env) + self.assertNotIn(PLACEHOLDER, result.stdout + result.stderr) + found = json.loads(result.stdout) + self.assertEqual( + {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}) + + def test_critical_exits_1_and_floors_at_0(self): + self.write("app.py", "a # MARK-CRITICAL\nb # MARK-CRITICAL\nc # MARK-CRITICAL\n") + result = self.scan(env=self.env) + self.assertTrue(result.stdout.startswith("๐Ÿ›ก๏ธ security 0 % (medusa: 3 findings in 1 file: critical app.py:1, ")) + self.assertEqual(result.returncode, 1) + + def test_more_than_five_locations_are_cut(self): + self.write("app.py", "".join(f"v{n} # MARK-LOW\n" for n in range(7))) + result = self.scan(env=self.env) + self.assertEqual( + result.stdout, + "๐Ÿ›ก๏ธ security 86 % (medusa: 7 findings in 1 file: low app.py:1, low app.py:2, low app.py:3, " + "low app.py:4, low app.py:5, +2 more)\n", + ) + self.assertEqual(result.returncode, 3) + + def test_clean_and_empty_changes_are_100(self): + result = self.scan(env=self.env) + self.assertEqual((result.stdout, result.returncode), ("๐Ÿ›ก๏ธ security 100 % (medusa: 0 findings in 0 files)\n", 0)) + self.assertFalse((self.bin / "argv.json").exists()) + self.write("app.py", "print(2)\n") + result = self.scan(env=self.env) + self.assertEqual((result.stdout, result.returncode), ("๐Ÿ›ก๏ธ security 100 % (medusa: 0 findings in 1 file)\n", 0)) + + def test_missing_medusa_asks_for_a_review_by_hand(self): + if shutil.which("medusa", path=GIT_DIR): + self.skipTest("a real medusa sits next to git") + self.write("app.py", "print(2)\n") + result = self.scan(env={**self.env, "PATH": GIT_DIR}) + self.assertEqual((result.stdout, result.returncode), (f"๐Ÿ›ก๏ธ security โ€“ (medusa missing: {HAND})\n", 4)) + result = self.scan("--json", env={**self.env, "PATH": GIT_DIR}) + self.assertEqual(json.loads(result.stdout), {"security": None, "problem": "medusa missing"}) + + def test_unreadable_report_asks_for_a_review_by_hand(self): + self.write("app.py", "print(2)\n") + result = self.scan(env={**self.env, "FAKE_MEDUSA_SHAPE": "other"}) + self.assertEqual(result.returncode, 4) + self.assertRegex(result.stdout, r"^๐Ÿ›ก๏ธ security โ€“ \(medusa report unreadable: .*holds no findings list") + + def test_usage_errors_exit_2(self): + result = subprocess.run( + [sys.executable, str(SECURITY), "--repo", str(self.repo), "--base", "nope"], + env=self.env, capture_output=True, text=True, check=False, + ) + self.assertEqual(result.returncode, 2) + result = subprocess.run( + [sys.executable, str(SECURITY), "--repo", str(self.base / "absent"), "--base", "HEAD"], + env=self.env, capture_output=True, text=True, check=False, + ) + self.assertEqual(result.returncode, 2) + + +if __name__ == "__main__": + unittest.main()