From 3fd0057e5f200af2fb088b59635386584ccf0fb7 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Mon, 22 Jun 2026 19:51:04 +0200 Subject: [PATCH] =?UTF-8?q?feat(skills):=20Phase=2014=20=E2=80=94=20Trail?= =?UTF-8?q?=20of=20Bits=20skill=20index=20+=20router?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the hardcoded skill-name stub with a real, routable skill index. - skills.py (zero-dep): SkillDescriptor inventory (~22 ToB skills with routing metadata — languages, tags, vulnerability classes, stages — plus a short operating checklist each), aligned with the spec's routing examples. route_skills scores by vulnerability-class/tag/stage signal with a language filter+bonus, and never selects on a pure language match. route_skills_for_target derives vuln classes from tags; select_skill_context packs checklists into a char budget; skill_chunks chunks the guidance into the `skills` retrieval namespace. - hunter.select_skill_snippets now delegates to the router (existing C-memory and crypto+auth routing expectations preserved). - hunter_harness._describe feeds the budget-capped routed skill context into the HarnessX hunter prompt (the genuine consumer of skill_budget_chars). - 8.4 zero-dep guard extended to cover skills.py. tests/test_skills.py (9 tests: C parser → memory/fuzzing, crypto → constant-time/ vector, language-mismatch + language-only exclusion, stage filtering, budget cap, namespace chunking). openrouter-sast-harness Phase 14 checked off. 188 passed, 4 skipped; ruff + mypy clean; detection gate PASS. Co-Authored-By: Claude Opus 4.8 --- .kiro/specs/openrouter-sast-harness/tasks.md | 14 +- src/openultrasast/hunter.py | 17 +- src/openultrasast/hunter_harness.py | 6 +- src/openultrasast/skills.py | 372 +++++++++++++++++++ tests/test_gate.py | 1 + tests/test_skills.py | 85 +++++ 6 files changed, 476 insertions(+), 19 deletions(-) create mode 100644 src/openultrasast/skills.py create mode 100644 tests/test_skills.py diff --git a/.kiro/specs/openrouter-sast-harness/tasks.md b/.kiro/specs/openrouter-sast-harness/tasks.md index ef3dc8c..d796594 100644 --- a/.kiro/specs/openrouter-sast-harness/tasks.md +++ b/.kiro/specs/openrouter-sast-harness/tasks.md @@ -276,11 +276,15 @@ Tests in `tests/test_fusion.py`. ## Phase 14: Skill Index (`standard_security_harness` prerequisite) -- [ ] Build Trail of Bits skill descriptor inventory. -- [ ] Chunk selected skill operating guidance into retrieval namespace. -- [ ] Implement skill router by language, tags, vulnerability class, and stage. -- [ ] Add prompt budget caps for skill snippets. -- [ ] Add tests for routing C parser targets to memory/fuzzing skills and crypto targets to constant-time/vector skills. +Implemented in `skills.py` (descriptor inventory + scoring router + retrieval-namespace +chunking + budget-capped context); the HarnessX hunter prompt consumes the routed +skill context; tests in `tests/test_skills.py`. + +- [x] Build Trail of Bits skill descriptor inventory. +- [x] Chunk selected skill operating guidance into retrieval namespace. +- [x] Implement skill router by language, tags, vulnerability class, and stage. +- [x] Add prompt budget caps for skill snippets. +- [x] Add tests for routing C parser targets to memory/fuzzing skills and crypto targets to constant-time/vector skills. ## Phase 15: Hardening And Release Readiness (`production` prerequisite) diff --git a/src/openultrasast/hunter.py b/src/openultrasast/hunter.py index 41cdeb9..fd3bf8c 100644 --- a/src/openultrasast/hunter.py +++ b/src/openultrasast/hunter.py @@ -135,19 +135,10 @@ def build_retrieval_context_by_path(root: Path, targets: list[FileTarget], *, ma def select_skill_snippets(target: FileTarget, *, max_snippets: int = 4) -> list[str]: - tags = set(target.tags) - snippets: list[str] = [] - if target.language in {"c", "cpp", "c++"} or "memory_unsafe" in tags: - snippets.extend(["address-sanitizer", "libfuzzer", "harness-writing"]) - if "parser" in tags or "fuzzable" in tags: - snippets.extend(["fuzzing-dictionary", "fuzzing-obstacles"]) - if "crypto" in tags: - snippets.extend(["constant-time-analysis", "wycheproof"]) - if "auth_boundary" in tags or "network_entry" in tags: - snippets.extend(["sharp-edges", "insecure-defaults"]) - if "deserialization" in tags: - snippets.append("semgrep-rule-creator") - return sorted(dict.fromkeys(snippets))[:max_snippets] + """Route the target to its most relevant Trail of Bits skills (ids), via the skill index.""" + from .skills import route_skill_ids + + return route_skill_ids(target, max_snippets=max_snippets) def _run_hunter_task( diff --git a/src/openultrasast/hunter_harness.py b/src/openultrasast/hunter_harness.py index 34d3729..164f80f 100644 --- a/src/openultrasast/hunter_harness.py +++ b/src/openultrasast/hunter_harness.py @@ -126,12 +126,16 @@ class HxScanOrchestrator: return HunterPoolResult(findings=sorted(findings, key=lambda item: item.finding_id), trajectories=trajectories) def _describe(self, task: HunterTask) -> str: + from .skills import select_skill_context + context = task.retrieval_context or f"[repo_code] {task.target.path}" + skills = select_skill_context(task.target, stage="hunt", budget_chars=task.budget.skill_budget_chars) + skill_block = f"\n\nRelevant security skills:\n{skills}" if skills else "" return ( f"Audit `{task.target.path}` ({task.target.language}) for security-relevant vulnerabilities. " f"Report ONLY evidence-backed findings as a JSON array, each object: " f'{{"line": , "title": , "cwe": "CWE-NN" (optional), "rationale": }}. ' - f"Emit `[]` if none.\n\nSource context:\n{context}" + f"Emit `[]` if none.{skill_block}\n\nSource context:\n{context}" ) def _findings_from_result(self, task: HunterTask, result: object, policy: dict[str, CwePolicy]) -> list[StaticFinding]: diff --git a/src/openultrasast/skills.py b/src/openultrasast/skills.py new file mode 100644 index 0000000..89da93b --- /dev/null +++ b/src/openultrasast/skills.py @@ -0,0 +1,372 @@ +"""Trail of Bits skill index + router (Phase 14). + +Security skills are scoped expertise, not global prompt bloat. Each skill is a small +descriptor with routing metadata (languages, tags, vulnerability classes, stages) and +a short operating checklist. The router selects the most relevant skills for a target +and stage; ``select_skill_context`` packs their checklists into a char budget; and +``skill_chunks`` chunks the guidance into the ``skills`` retrieval namespace. Nothing +here imports a model or the optional extra — it is pure routing data. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +from .index import CodeChunk, chunk_text_namespace +from .preprocess import FileTarget + +# Pipeline stages a skill can apply to. +STAGES = ("map", "hunt", "verify", "fix") + +# FileTarget tags → vulnerability classes the router reasons over. +TAG_TO_VULN = { + "memory_unsafe": "memory-safety", + "parser": "parsing", + "fuzzable": "parsing", + "crypto": "crypto", + "deserialization": "deserialization", + "auth_boundary": "access-control", + "network_entry": "access-control", + "privileged": "access-control", +} + + +@dataclass(frozen=True) +class SkillDescriptor: + id: str + title: str + summary: str + checklist: str + languages: tuple[str, ...] = () # empty => language-agnostic + tags: tuple[str, ...] = () + vulnerability_classes: tuple[str, ...] = () + stages: tuple[str, ...] = () # empty => any stage + + def descriptor(self) -> dict[str, object]: + return { + "id": self.id, + "title": self.title, + "summary": self.summary, + "languages": list(self.languages), + "tags": list(self.tags), + "vulnerability_classes": list(self.vulnerability_classes), + "stages": list(self.stages), + } + + +# Curated inventory aligned with the spec's Trail of Bits skill routing examples. +SKILL_INVENTORY: tuple[SkillDescriptor, ...] = ( + SkillDescriptor( + "c-review", + "C/C++ security review", + "Manual review discipline for memory-unsafe code.", + "Audit buffer indexing, object lifetime, integer overflow before length math, and unchecked copies.", + languages=("c", "cpp", "c++"), + tags=("memory_unsafe",), + vulnerability_classes=("memory-safety",), + stages=("map", "hunt"), + ), + SkillDescriptor( + "address-sanitizer", + "AddressSanitizer", + "Detect memory-safety violations at runtime.", + "Build with -fsanitize=address; reproduce the overflow/UAF; capture the ASan trace as crash evidence.", + languages=("c", "cpp", "c++"), + tags=("memory_unsafe",), + vulnerability_classes=("memory-safety",), + stages=("hunt", "verify"), + ), + SkillDescriptor( + "libfuzzer", + "libFuzzer harnessing", + "In-process coverage-guided fuzzing of C/C++ entries.", + "Write a LLVMFuzzerTestOneInput harness over the parser entry; minimize crashing inputs.", + languages=("c", "cpp", "c++"), + tags=("memory_unsafe", "fuzzable", "parser"), + vulnerability_classes=("memory-safety", "parsing"), + stages=("hunt", "verify"), + ), + SkillDescriptor( + "harness-writing", + "Fuzz harness writing", + "Build deterministic harnesses around attacker-controlled entries.", + "Identify the smallest attacker-controlled entry; build a deterministic harness; seed a corpus.", + languages=("c", "cpp", "c++"), + tags=("memory_unsafe", "fuzzable", "parser"), + vulnerability_classes=("memory-safety", "parsing"), + stages=("hunt", "verify"), + ), + SkillDescriptor( + "fuzzing-dictionary", + "Fuzzing dictionary", + "Reach deep parser states with a format dictionary.", + "Extract format tokens/magic bytes into a fuzzing dictionary to reach deep parser states.", + tags=("parser", "fuzzable"), + vulnerability_classes=("parsing",), + stages=("verify",), + ), + SkillDescriptor( + "fuzzing-obstacles", + "Fuzzing obstacles", + "Remove coverage barriers that block the fuzzer.", + "Remove checksums/magic gates that block coverage; document each bypass.", + tags=("parser", "fuzzable"), + vulnerability_classes=("parsing",), + stages=("verify",), + ), + SkillDescriptor( + "atheris", + "Atheris (Python fuzzing)", + "Coverage-guided fuzzing for Python parsers.", + "Fuzz the Python parser entry with Atheris; assert on uncaught exceptions and resource blowups.", + languages=("python",), + tags=("parser", "fuzzable"), + vulnerability_classes=("parsing",), + stages=("hunt", "verify"), + ), + SkillDescriptor( + "property-based-testing", + "Property-based testing", + "Encode parser invariants as properties.", + "Encode parser invariants as Hypothesis properties; shrink counterexamples.", + languages=("python",), + tags=("parser",), + vulnerability_classes=("parsing",), + stages=("verify",), + ), + SkillDescriptor( + "constant-time-analysis", + "Constant-time analysis", + "Find secret-dependent timing in crypto code.", + "Find secret-dependent branches/indexing/early-exit comparisons; require constant-time primitives.", + tags=("crypto",), + vulnerability_classes=("crypto",), + stages=("hunt", "verify"), + ), + SkillDescriptor( + "wycheproof", + "Wycheproof vectors", + "Stress crypto usage with known-edge test vectors.", + "Run Wycheproof test vectors against the crypto usage; flag edge-case acceptance.", + tags=("crypto",), + vulnerability_classes=("crypto",), + stages=("verify",), + ), + SkillDescriptor( + "vector-forge", + "Crypto vector forging", + "Forge boundary vectors to probe crypto APIs.", + "Forge boundary test vectors (zero/identity/malleable signatures) to probe the crypto API.", + tags=("crypto",), + vulnerability_classes=("crypto",), + stages=("verify",), + ), + SkillDescriptor( + "zeroize-audit", + "Secret zeroization audit", + "Verify secrets are wiped after use.", + "Verify secrets are zeroized on drop; check for lingering copies.", + languages=("rust",), + tags=("crypto",), + vulnerability_classes=("crypto",), + stages=("verify",), + ), + SkillDescriptor( + "cargo-fuzz", + "cargo-fuzz", + "Coverage-guided fuzzing for Rust entries.", + "Add a cargo-fuzz target over the entry; run under ASan.", + languages=("rust",), + tags=("fuzzable", "parser"), + vulnerability_classes=("memory-safety", "parsing"), + stages=("hunt", "verify"), + ), + SkillDescriptor( + "sarif-parsing", + "SARIF parsing", + "Normalize external tool output into the map.", + "Normalize SARIF locations/rules/fingerprints into FileTarget static hints.", + vulnerability_classes=(), + stages=("map",), + ), + SkillDescriptor( + "semgrep", + "Semgrep mapping", + "Pattern sinks and iterate variants.", + "Pattern the sink shape; iterate variants; export normalized findings.", + vulnerability_classes=("injection", "deserialization"), + stages=("map",), + ), + SkillDescriptor( + "semgrep-rule-creator", + "Semgrep rule creator", + "Author a Semgrep rule for a sink.", + "Author a Semgrep rule for the sink; validate against the fixture.", + tags=("deserialization",), + vulnerability_classes=("injection", "deserialization"), + stages=("map",), + ), + SkillDescriptor( + "semgrep-rule-variant-creator", + "Semgrep variant creator", + "Widen sink coverage with variants.", + "Generate sink variants (aliases, wrappers) to widen coverage without false positives.", + vulnerability_classes=("injection",), + stages=("map",), + ), + SkillDescriptor( + "codeql", + "CodeQL dataflow", + "Model source→sink taint paths.", + "Model source→sink taint with a CodeQL query; inspect the dataflow path.", + vulnerability_classes=("injection",), + stages=("map",), + ), + SkillDescriptor( + "differential-review", + "Differential review", + "Scope the changed attack surface in a diff.", + "Scope the changed attack surface and new trust-boundary crossings in the diff.", + stages=("map", "fix"), + ), + SkillDescriptor( + "graph-evolution", + "Graph evolution", + "Bound a patch's blast radius.", + "Trace how the call/dataflow graph changed; bound the patch blast radius.", + stages=("fix",), + ), + SkillDescriptor( + "sharp-edges", + "Sharp edges review", + "Map misuse-prone APIs at the boundary.", + "Map misuse-prone APIs, insecure defaults, and confusing configuration at the boundary.", + tags=("auth_boundary", "network_entry"), + vulnerability_classes=("access-control", "api-misuse"), + stages=("map", "hunt"), + ), + SkillDescriptor( + "insecure-defaults", + "Insecure defaults review", + "Flag unsafe defaults exposed at entries.", + "Flag unsafe default flags/permissions/timeouts exposed at the entry.", + tags=("auth_boundary", "network_entry"), + vulnerability_classes=("access-control", "api-misuse"), + stages=("map", "hunt"), + ), +) + +_BY_ID = {skill.id: skill for skill in SKILL_INVENTORY} + + +@dataclass(frozen=True) +class SkillMatch: + skill: SkillDescriptor + score: int + reasons: list[str] = field(default_factory=list) + + +def route_skills( + *, + language: str | None = None, + tags: tuple[str, ...] = (), + vulnerability_classes: tuple[str, ...] = (), + stage: str | None = None, + max_snippets: int = 4, +) -> list[SkillDescriptor]: + """Select the most relevant skills for the given routing signals. + + A skill is excluded when its language list is non-empty and ``language`` is not in + it, or its stage list is non-empty and ``stage`` is not in it. Eligibility requires + at least one topical signal (tag/vulnerability-class/stage) — a pure language match + never selects a skill. Ranked by score, then id, for determinism. + """ + tag_set, vuln_set = set(tags), set(vulnerability_classes) + matches: list[SkillMatch] = [] + for skill in SKILL_INVENTORY: + if skill.languages and language is not None and language not in skill.languages: + continue + if skill.stages and stage is not None and stage not in skill.stages: + continue + signal = 2 * len(vuln_set & set(skill.vulnerability_classes)) + len(tag_set & set(skill.tags)) + if stage is not None and stage in skill.stages: + signal += 1 + if signal == 0: + continue # no topical match — do not select on language alone + rank = signal + (3 if skill.languages and language in skill.languages else 0) + matches.append(SkillMatch(skill=skill, score=rank)) + matches.sort(key=lambda match: (-match.score, match.skill.id)) + return [match.skill for match in matches[:max_snippets]] + + +def _vuln_classes_for_tags(tags: tuple[str, ...]) -> tuple[str, ...]: + return tuple(sorted({TAG_TO_VULN[tag] for tag in tags if tag in TAG_TO_VULN})) + + +def route_skills_for_target(target: FileTarget, *, stage: str | None = None, max_snippets: int = 4) -> list[SkillDescriptor]: + """Route skills for a scan target (language + tags + derived vulnerability classes).""" + tags = tuple(target.tags) + return route_skills( + language=target.language, + tags=tags, + vulnerability_classes=_vuln_classes_for_tags(tags), + stage=stage, + max_snippets=max_snippets, + ) + + +def route_skill_ids(target: FileTarget, *, stage: str | None = None, max_snippets: int = 4) -> list[str]: + return [skill.id for skill in route_skills_for_target(target, stage=stage, max_snippets=max_snippets)] + + +def select_skill_context(target: FileTarget, *, stage: str | None = None, budget_chars: int, max_snippets: int = 4) -> str: + """Pack routed skill checklists into a prompt context capped at ``budget_chars``.""" + if budget_chars <= 0: + return "" + parts: list[str] = [] + used = 0 + for skill in route_skills_for_target(target, stage=stage, max_snippets=max_snippets): + snippet = f"### {skill.title}\n{skill.checklist}" + if not parts and len(snippet) > budget_chars: + return snippet[:budget_chars] # first snippet alone exceeds the cap → truncate + if used + len(snippet) + (1 if parts else 0) > budget_chars: + break + used += len(snippet) + (1 if parts else 0) + parts.append(snippet) + return "\n".join(parts) + + +def skill_chunks(skills: tuple[SkillDescriptor, ...] | None = None, *, max_lines: int = 40) -> list[CodeChunk]: + """Chunk skill operating guidance into the ``skills`` retrieval namespace.""" + chunks: list[CodeChunk] = [] + for skill in skills or SKILL_INVENTORY: + text = f"# {skill.title}\n{skill.summary}\n\nChecklist:\n{skill.checklist}" + chunks.extend( + chunk_text_namespace( + namespace="skills", + path=f"skills/{skill.id}.md", + text=text, + metadata={ + "skill_id": skill.id, + "languages": ",".join(skill.languages), + "tags": ",".join(skill.tags), + "vulnerability_classes": ",".join(skill.vulnerability_classes), + "stages": ",".join(skill.stages), + }, + max_lines=max_lines, + ) + ) + return chunks + + +__all__ = [ + "SKILL_INVENTORY", + "STAGES", + "TAG_TO_VULN", + "SkillDescriptor", + "route_skill_ids", + "route_skills", + "route_skills_for_target", + "select_skill_context", + "skill_chunks", +] diff --git a/tests/test_gate.py b/tests/test_gate.py index 9ce9f48..df984c7 100644 --- a/tests/test_gate.py +++ b/tests/test_gate.py @@ -147,6 +147,7 @@ GOVERNANCE_SCORING_BENCHMARK_PLANES = ( "openultrasast.slot_contract", "openultrasast.stage_processors", "openultrasast.fusion", + "openultrasast.skills", ) diff --git a/tests/test_skills.py b/tests/test_skills.py new file mode 100644 index 0000000..28ead94 --- /dev/null +++ b/tests/test_skills.py @@ -0,0 +1,85 @@ +"""Trail of Bits skill index + router (Phase 14).""" + +from openultrasast.preprocess import FileTarget +from openultrasast.skills import ( + SKILL_INVENTORY, + route_skill_ids, + route_skills, + route_skills_for_target, + select_skill_context, + skill_chunks, +) + + +def _target(language: str, tags: list[str]) -> FileTarget: + return FileTarget(path="t", absolute_path="/repo/t", language=language, loc=10, tags=tags, has_fuzz_entry_point=False) + + +# ---- routing by language / tags / vulnerability class ----------------------- + + +def test_c_parser_target_routes_to_memory_and_fuzzing_skills() -> None: + ids = route_skill_ids(_target("c", ["memory_unsafe", "parser"])) + assert {"address-sanitizer", "harness-writing"} <= set(ids) # memory + fuzzing + assert "libfuzzer" in ids + + +def test_crypto_target_routes_to_constant_time_and_vector_skills() -> None: + ids = route_skill_ids(_target("python", ["crypto"])) + assert "constant-time-analysis" in ids + assert "vector-forge" in ids + + +def test_language_mismatch_excludes_other_language_skills() -> None: + # A C target must not pull Python-only (atheris) or Rust-only (cargo-fuzz) skills. + ids = route_skill_ids(_target("c", ["memory_unsafe", "parser", "fuzzable"])) + assert "atheris" not in ids + assert "cargo-fuzz" not in ids + + +def test_language_only_match_does_not_select() -> None: + # python + crypto must not surface atheris (python but no crypto signal). + ids = route_skills_for_target(_target("python", ["crypto"]), max_snippets=10) + assert "atheris" not in [skill.id for skill in ids] + + +# ---- routing by stage ------------------------------------------------------- + + +def test_stage_filter_restricts_to_stage_skills() -> None: + # fuzzing-dictionary is verify-only; at the map stage it must not appear, while a + # map-stage skill (semgrep) does for an injection signal. + verify = route_skills(tags=("parser",), vulnerability_classes=("parsing",), stage="verify", max_snippets=10) + mapped = route_skills(tags=("parser",), vulnerability_classes=("parsing",), stage="map", max_snippets=10) + assert "fuzzing-dictionary" in {s.id for s in verify} + assert "fuzzing-dictionary" not in {s.id for s in mapped} + + +# ---- prompt budget caps ----------------------------------------------------- + + +def test_select_skill_context_respects_budget() -> None: + target = _target("c", ["memory_unsafe", "parser"]) + full = select_skill_context(target, budget_chars=10_000) + capped = select_skill_context(target, budget_chars=80) + assert len(capped) <= 80 + assert len(full) > len(capped) + + +def test_zero_budget_yields_no_context() -> None: + assert select_skill_context(_target("c", ["memory_unsafe"]), budget_chars=0) == "" + + +# ---- retrieval-namespace chunking + inventory integrity --------------------- + + +def test_skill_chunks_land_in_the_skills_namespace() -> None: + chunks = skill_chunks() + assert chunks + assert all(chunk.namespace == "skills" for chunk in chunks) + assert any(chunk.metadata.get("skill_id") == "address-sanitizer" for chunk in chunks) + + +def test_inventory_ids_are_unique() -> None: + ids = [skill.id for skill in SKILL_INVENTORY] + assert len(ids) == len(set(ids))