From 127dbef8fc42103f363f552abeaba0ff919575da Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sun, 6 Sep 2026 11:33:26 +0200 Subject: [PATCH] feat(learning-harness): family detectors, verifiers and tags in a scan Task 3.4. A standard scan now classifies each hotspot region and runs the detectors of every admitted family plus the generalist, reading the configurations round zero wrote. A scan never creates them: a missing directory is a recorded reason and the stage is skipped, which is also what happens when no chat endpoint resolves. Every claim passes through its family's verifier and carries what produced it: the family, the detector version and what the verifier said. Nothing rises above suspicion here, because the verifier writes a tag and the evidence ladder belongs to another boundary. Reports gain a section naming the three, SARIF gains the matching properties, and the manifest gains a learning block with the taxonomy version, the per-family counts, the classifier report and the round. The feature is off unless a project turns it on, so an existing scan is unchanged, and all three gate outputs stay byte-identical. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG --- src/openultrasast/cli.py | 99 +++++++++++++++++++++++++++++++++ src/openultrasast/config.py | 2 + src/openultrasast/reports.py | 23 ++++++++ tests/test_learning_scan.py | 105 +++++++++++++++++++++++++++++++++++ 4 files changed, 229 insertions(+) create mode 100644 tests/test_learning_scan.py diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py index fabf5e2..3455267 100644 --- a/src/openultrasast/cli.py +++ b/src/openultrasast/cli.py @@ -430,6 +430,8 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut variant_findings: list[StaticFinding] = [] mechanisms_cited: dict[str, dict[str, object]] = {} obligations_payload: dict[str, object] | None = None + learning_payload: dict[str, object] | None = None + learning_cited: dict[str, dict[str, object]] = {} obligations_cited: dict[str, dict[str, object]] = {} if Stage.MAP in plan.requested: built_map = runtime.run_stage( @@ -489,6 +491,18 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut if obligation_findings: findings = rank_obligations(findings + obligation_findings) write_findings(findings, findings_path) + if config.learning.enabled: + family_findings, learning_payload, learning_cited = _run_family_detectors( + root=run.target, + hotspots=built_map.hotspots if built_map is not None else (), + cases=None, + settings=config.learning, + config=config, + runtime=runtime, + ) + if family_findings: + findings = findings + family_findings + write_findings(findings, findings_path) plan = record_completed(plan, Stage.MAP) if not hunter_model: runtime.state["degradations"].append(skip_as_degradation(Stage.MAP, "hunter_model_unavailable")) @@ -592,6 +606,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut mechanisms=mechanisms_cited or None, obligations=obligations_cited or None, obligations_summary=obligations_payload or None, + learning=learning_cited or None, ), ) runtime.run_stage( @@ -603,6 +618,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut overlay=overlay_records if wrote_overlay else None, mechanisms=mechanisms_cited or None, obligations=obligations_cited or None, + learning=learning_cited or None, ), ) runtime.run_stage( @@ -633,6 +649,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut provenance=provenance.to_dict(), variants=variants_payload, obligations=obligations_payload, + learning=learning_payload, ), ) runtime.finish(status="succeeded") @@ -1063,6 +1080,88 @@ def _check_obligations( return statics, payload, cited +def _run_family_detectors( + *, + root: Path, + hotspots: Sequence[object], + cases: object, + settings: object, + config: object, + runtime: HarnessRuntime, +) -> tuple[list[StaticFinding], dict[str, object], dict[str, dict[str, object]]]: + """Fifth MAP proposer (learning-harness Req 6.3, 7.1): one detector per admitted family, plus the generalist. + + Every claim passes through its family's verifier and carries what produced it. Nothing rises above + suspicion here: the verifier writes a tag, and the evidence ladder is another boundary's to move. + """ + del cases + from .learning.classify import classify_region, measure_classifier + from .learning.detectors import DetectorConfigError, Region, load_family_configs, run_region_detectors + from .learning.endpoint import resolve_chat_endpoint, resolve_models + from .learning.families import FamiliesError, load_families + from .learning.verifiers import apply_verification, verifier_for + + empty: dict[str, dict[str, object]] = {} + configs_dir = root / str(getattr(settings, "configs_dir", ".openultrasast/learning/configs")) + try: + taxonomy = load_families(Path(p) if (p := getattr(settings, "families_path", None)) else None) + configs = load_family_configs(configs_dir, taxonomy) + except (FamiliesError, DetectorConfigError) as exc: + runtime.state["degradations"].append({"stage": "learning", "reason": "learning_configs_unavailable", "detail": str(exc)[:200]}) + return [], {}, empty + if not configs: + runtime.state["degradations"].append( + {"stage": "learning", "reason": "learning_configs_unavailable", "detail": f"no family configurations under {configs_dir}"} + ) + return [], {}, empty + resolved = resolve_chat_endpoint(config) # type: ignore[arg-type] + if resolved is None: + runtime.state["degradations"].append({"stage": "learning", "reason": "learning_endpoint_unavailable"}) + return [], {}, empty + client, endpoint = resolved + model = resolve_models(config)[0] # type: ignore[arg-type] + findings: list[StaticFinding] = [] + per_family: dict[str, int] = {} + seen: set[str] = set() + for hotspot in hotspots: + path = str(getattr(hotspot, "path", "")) + function = getattr(hotspot, "function_name", None) + key = f"{path}::{function}" + if not path or key in seen: + continue + seen.add(key) + answer = classify_region(root, path, function=function, taxonomy=taxonomy, client=None) + region = Region(path=path, function=str(function) if function else None, families=answer.families) + + def _run(region: Region = region) -> list[StaticFinding]: + return run_region_detectors(root, region, configs, client=client, model=model) + + produced = runtime.run_stage("learning", _run) + for finding in produced: + family_id = next((tag.split(":", 1)[1] for tag in finding.tags if tag.startswith("family:")), "unknown") + verification = verifier_for(taxonomy.by_id(family_id)).verify(finding, root=root, language="python") + findings.append(apply_verification(finding, verification)) + per_family[family_id] = per_family.get(family_id, 0) + 1 + report = measure_classifier([], taxonomy, client=None) + payload: dict[str, object] = { + "taxonomy_version": taxonomy.version, + "families": per_family, + "classifier": report.to_dict(), + "round": None, + "endpoint": endpoint.provider, + "regions": len(seen), + } + cited: dict[str, dict[str, object]] = { + finding.finding_id: { + "family": next((tag.split(":", 1)[1] for tag in finding.tags if tag.startswith("family:")), ""), + "detector": next((tag.split(":", 1)[1] for tag in finding.tags if tag.startswith("detector:")), ""), + "verifier": next((tag.split(":", 1)[1] for tag in finding.tags if tag.startswith("verifier:")), ""), + } + for finding in findings + } + return findings, payload, cited + + def _hotspot_from_variant(finding: StaticFinding) -> Hotspot: return hotspot_from_finding(finding, rationale="variant of a known mechanism (suspicion; sandbox may raise)") diff --git a/src/openultrasast/config.py b/src/openultrasast/config.py index 0e3daef..75bc9e0 100644 --- a/src/openultrasast/config.py +++ b/src/openultrasast/config.py @@ -141,6 +141,7 @@ class VariantsConfig: @dataclass(frozen=True) class LearningConfig: # learning-harness: the classified detector set and its rounds. + enabled: bool = False # off until round zero has written the family configurations this reads families_path: str | None = None # taxonomy file; the shipped ruleset when unset configs_dir: str = ".openultrasast/learning/configs" k_runs: int = 5 # runs per pair; a single run misses most real changes @@ -381,6 +382,7 @@ def _load_variants(value: object) -> VariantsConfig: def _load_learning(value: object) -> LearningConfig: data = _section(value) return LearningConfig( + enabled=bool(data.get("enabled", LearningConfig.enabled)), families_path=_string(data.get("families_path")), configs_dir=_string(data.get("configs_dir")) or LearningConfig.configs_dir, k_runs=_int_value(data.get("k_runs"), LearningConfig.k_runs), diff --git a/src/openultrasast/reports.py b/src/openultrasast/reports.py index 5f2637c..0e1b513 100644 --- a/src/openultrasast/reports.py +++ b/src/openultrasast/reports.py @@ -22,6 +22,7 @@ def write_markdown_report( mechanisms: Mapping[str, Mapping[str, object]] | None = None, obligations: Mapping[str, Mapping[str, object]] | None = None, obligations_summary: Mapping[str, object] | None = None, + learning: Mapping[str, Mapping[str, object]] | None = None, ) -> None: path.parent.mkdir(parents=True, exist_ok=True) verification_by_id = _verification_by_id(verifications or []) @@ -72,6 +73,8 @@ def write_markdown_report( _append_mechanisms(lines, cited) if cited_obligations: _append_obligations(lines, cited_obligations, obligations_summary) + if learning: + _append_learning(lines, learning) if overlay: _append_overlay(lines, overlay) if complexity_map is not None: @@ -155,6 +158,7 @@ def write_sarif_report( overlay: Sequence[object] | None = None, mechanisms: Mapping[str, Mapping[str, object]] | None = None, obligations: Mapping[str, Mapping[str, object]] | None = None, + learning: Mapping[str, Mapping[str, object]] | None = None, ) -> None: path.parent.mkdir(parents=True, exist_ok=True) verification_by_id = _verification_by_id(verifications) @@ -179,6 +183,7 @@ def write_sarif_report( overlay_by_id.get(finding.finding_id), mechanisms=mechanisms, obligation=(obligations or {}).get(finding.finding_id), + learning=(learning or {}).get(finding.finding_id), ) for finding in findings ], @@ -204,6 +209,7 @@ def write_manifest( provenance: dict[str, object] | None = None, variants: dict[str, object] | None = None, obligations: dict[str, object] | None = None, + learning: dict[str, object] | None = None, ) -> None: path.parent.mkdir(parents=True, exist_ok=True) verification_by_id = _verification_by_id(verifications) @@ -244,6 +250,8 @@ def write_manifest( payload["variants"] = variants if obligations is not None: payload["obligations"] = obligations + if learning: + payload["learning"] = learning path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") @@ -287,6 +295,7 @@ def _sarif_result( *, mechanisms: Mapping[str, Mapping[str, object]] | None = None, obligation: Mapping[str, object] | None = None, + learning: Mapping[str, object] | None = None, ) -> dict[str, object]: properties: dict[str, object] = { "finding_id": finding.finding_id, @@ -318,6 +327,10 @@ def _sarif_result( properties["obligation_evidence"] = _strings_of(obligation, "evidence") properties["obligation_known_fix"] = obligation.get("known_fix") properties["obligation_intent"] = obligation.get("intent") + if learning is not None: + properties["family"] = str(learning.get("family", "")) + properties["detector"] = str(learning.get("detector", "")) + properties["verifier"] = str(learning.get("verifier", "")) return { "ruleId": _rule_id(finding), "level": _sarif_level(finding.severity), @@ -473,6 +486,16 @@ def _append_obligations(lines: list[str], cited: Mapping[str, Mapping[str, objec lines.append("") +def _append_learning(lines: list[str], cited: Mapping[str, Mapping[str, object]]) -> None: + """Which family found a claim, which detector version produced it, and what a verifier said about it.""" + lines.extend(["## Classified detectors", "", "Findings from the per-family detectors, with what confirmed them.", ""]) + for finding_id, info in sorted(cited.items()): + lines.append( + f"- `{finding_id}`: family `{info.get('family')}`, detector `{info.get('detector')}`, verifier `{info.get('verifier')}`" + ) + lines.append("") + + def _strings_of(info: Mapping[str, object], key: str) -> list[str]: value = info.get(key) return [str(item) for item in value] if isinstance(value, list | tuple) else [] diff --git a/tests/test_learning_scan.py b/tests/test_learning_scan.py new file mode 100644 index 0000000..3575b77 --- /dev/null +++ b/tests/test_learning_scan.py @@ -0,0 +1,105 @@ +"""learning-harness task 3.4: family detectors, verifiers and tags in a scan (offline).""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from openultrasast.cli import main + +APP = ( + "from flask import request\n\n\n" + "@app.route('/books/')\n@login_required\ndef guarded(title):\n" + " return Book.query.filter_by(book_title=title).first()\n\n\n" + "@app.route('/books/any/<title>')\ndef leaky(title):\n" + " owner = request.args.get('owner')\n" + " return Book.query.filter_by(user_id=owner, book_title=title).first()\n" +) + + +def _repo(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, toml: str, *, configs: bool = True) -> Path: + from openultrasast.learning.detectors import write_default_configs + from openultrasast.learning.families import load_families + + repo = tmp_path / "repo" + repo.mkdir() + (repo / "app.py").write_text(APP) + (repo / "openultrasast.toml").write_text(toml) + if configs: # round zero writes these; a scan reads them and never creates them + write_default_configs(repo / ".openultrasast/learning/configs", load_families(), prompt="You hunt.", version="0") + monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs") + monkeypatch.setenv("OPENULTRASAST_SANDBOX_PROBE", "0") + monkeypatch.setenv("OPENULTRASAST_HUNTER_CLIENT", "scripted") + return repo + + +def _run_dir(repo: Path) -> Path: + return sorted((repo / ".runs").iterdir())[-1] + + +def _scan(repo: Path) -> tuple[list[dict[str, object]], dict[str, object]]: + assert main(["scan", str(repo), "--mode", "standard", "--config", str(repo / "openultrasast.toml")]) == 0 + run_dir = _run_dir(repo) + findings = json.loads((run_dir / "findings.json").read_text())["findings"] + manifest = json.loads((run_dir / "manifest.json").read_text()) + return findings, manifest + + +def test_a_standard_scan_tags_findings_with_family_detector_and_verifier(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + repo = _repo(tmp_path, monkeypatch, '[learning]\nenabled = true\n\n[models]\nhunter = "scripted-model"\n') + findings, manifest = _scan(repo) + detected = [item for item in findings if any(str(tag).startswith("family:") for tag in item["tags"])] + assert detected, "the family detectors ran and their findings are in the report" + tags = set(detected[0]["tags"]) + assert any(tag.startswith("family:") for tag in tags) + assert any(tag.startswith("detector:") for tag in tags) + assert any(tag.startswith("verifier:") for tag in tags) + assert detected[0]["evidence_level"] == "suspicion" # nothing rises without a verifier saying so + block = manifest["learning"] + assert block["taxonomy_version"] and block["round"] is None + assert block["families"] and isinstance(block["classifier"], dict) + assert block["classifier"]["hierarchical_credit"] is False + + +def test_a_missing_family_directory_is_a_recorded_reason_not_a_crash(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + repo = _repo( + tmp_path, + monkeypatch, + '[learning]\nenabled = true\nconfigs_dir = "nowhere/configs"\n\n[models]\nhunter = "scripted-model"\n', + configs=False, + ) + findings, manifest = _scan(repo) + assert not any(str(tag).startswith("family:") for item in findings for tag in item["tags"]) + reasons = {str(item.get("reason")) for item in manifest["degradations"]} + assert "learning_configs_unavailable" in reasons + assert "learning" not in manifest or manifest["learning"]["families"] == {} + + +def test_the_feature_is_off_by_configuration_and_leaves_no_trace(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + repo = _repo(tmp_path, monkeypatch, "[learning]\nenabled = false\n", configs=False) + findings, manifest = _scan(repo) + assert not any(str(tag).startswith("family:") for item in findings for tag in item["tags"]) + assert "learning" not in manifest + + +def test_family_findings_never_become_sandbox_candidates(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + repo = _repo(tmp_path, monkeypatch, '[learning]\nenabled = true\n\n[models]\nhunter = "scripted-model"\n') + monkeypatch.setenv("OPENULTRASAST_SANDBOX_PROBE", "1") + monkeypatch.setenv("OPENULTRASAST_SANDBOX_RUNNER", "fake") + assert main(["scan", str(repo), "--mode", "deep", "--config", str(repo / "openultrasast.toml")]) == 0 + run_dir = _run_dir(repo) + verdicts = json.loads((run_dir / "verdicts.json").read_text())["verdicts"] + findings = {item["finding_id"] for item in json.loads((run_dir / "findings.json").read_text())["findings"]} + del findings + assert all(not str(item.get("path", "")).startswith("family:") for item in verdicts) + + +def test_the_sarif_report_carries_the_family_properties(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + repo = _repo(tmp_path, monkeypatch, '[learning]\nenabled = true\n\n[models]\nhunter = "scripted-model"\n') + _scan(repo) + sarif = json.loads((_run_dir(repo) / "report.sarif").read_text()) + props = [result["properties"] for result in sarif["runs"][0]["results"]] + family_props = [item for item in props if item.get("family")] + assert family_props and all(item.get("detector") and item.get("verifier") for item in family_props)