mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 03:49:26 +00:00
Marius CiepluchandClaude Opus 5
6dc25d77e4
contributor-scan 2.10. Every one of libpng's twenty-five entry-point regions was a `main()` -- twenty-three under contrib/, plus example.c and pngtest.c -- and those regions carried the fifty interprocedural taint requests that dominated a 300s query phase. A library has no entry points, so entry-point-driven analysis had nothing to anchor to except its own example programs. The task said to derive this rather than hardcode `contrib/`, and a project already declares it. Makefile.am names installed targets in `lib_LTLIBRARIES` and `bin_PROGRAMS`, each target's `<target>_SOURCES` names its files, and `check_PROGRAMS`/`noinst_PROGRAMS` are automake's own word for "not installed" -- exactly the distinction wanted. Taking every `_SOURCES` would have called libpng's whole test suite shipped code, since contrib/libtests declares its sources the same way the library does. interprocedural regions 25 -> 1 taint query 300.1s -> 125.0s total query 311.9s -> 138.5s regions 121 -> 121, unchanged by design The surviving interprocedural region is contrib/tools/pngfix.c:main, which libpng genuinely installs. Two rules keep it honest. **Silence is not exclusion**: a project with no build file this can read gets None back and nothing changes, so vampi's numbers are untouched. **Not shipped is not unscanned**: those regions are still arbitrated function-locally, because a contributor editing a test must still get an answer -- what they lose is the expensive treatment and first claim on the budget. Entailed findings went 0 -> 3, and all three are false positives at contrib/gregbook/wpng.c:374, `strcpy(outname+len, ".png")`. It is safe: outname is char[256] and line 325 guards with `if ((len = strlen(inname)) > 250)`. They were not introduced so much as uncovered -- a different callDepth meant a different sink set, and the 26 fgets false positives were masking them. The structural cause is recorded as 2.12: c/injection and c/memory carry ZERO sanitizers, so no C flow can ever be discharged and every `argv -> strcpy` entails however carefully it is bounded. The discharge here is not a call but a guard that dominates the sink, which is precisely what dominance.sc already does for access control and is wired to that one family. Also adds Group 7, five tasks for stopping the same false positive twice. Two of today's three false-positive classes were one bug in three matchers -- `resolve` in `resolveUrl`, `user` in `users`, `gets` in `fgets` -- each fixed locally and each next one surviving because nobody carried the fix across. A model may group counter-examples by cause and draft an edit; only a split-respecting measurement may adopt one, because the predecessor's improve lever admitted shapes taught by the holdout pairs they then recovered and turned a -0.059 Youden into +0.059. Gates byte-identical. Co-Authored-By: Claude Opus 5 (1M context) <[email protected]> Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG
1164 lines
51 KiB
Python
1164 lines
51 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import tempfile
|
|
from collections.abc import Sequence
|
|
from dataclasses import dataclass, replace
|
|
from pathlib import Path
|
|
from time import perf_counter
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from . import tool_hunter
|
|
from .benchmark import (
|
|
create_benchmark_run,
|
|
evaluate_benchmark,
|
|
load_baseline_findings,
|
|
load_benchmark_manifest,
|
|
load_findings,
|
|
resolve_benchmark_source,
|
|
write_benchmark_artifacts,
|
|
)
|
|
from .calibration import (
|
|
FalsePositiveLearning,
|
|
calibrate_rankings,
|
|
fp_reachability_overrides,
|
|
learnings_from_verifications,
|
|
load_false_positive_learnings,
|
|
merge_false_positive_learnings,
|
|
write_false_positive_learnings,
|
|
write_ranking_calibrations,
|
|
)
|
|
from .complexity import map as complexity_map
|
|
from .complexity.ledger import persist_verdicts
|
|
from .config import ModelLayerConfig, ObligationsConfig, ResolvedConfig, load_config, load_dotenv
|
|
from .findings import StaticFinding, quick_scan_findings, write_findings
|
|
from .fusion import FusionDecision, fuse_findings_dispatch
|
|
from .gate import FALSE_POSITIVE_CEILING, RECALL_FLOOR
|
|
from .harness import HarnessRuntime, HarnessTraceWriter, write_harness_config
|
|
from .harness_ext import has_harnessx
|
|
from .hunter import run_hunter_pool, write_hunter_trajectories
|
|
from .hunter_harness import HxScanOrchestrator
|
|
from .improve import RoundOutcome, run_improvement
|
|
from .index import build_code_chunks
|
|
from .mapping import analyze_entry_points, attach_reachability_hints, ingest_sarif, write_entry_points, write_static_hints
|
|
from .pair_gate import print_pair_metrics
|
|
from .pairs import (
|
|
DEFAULT_CATALOG,
|
|
SLICE_NAMES,
|
|
evaluate_catalog,
|
|
load_pair_catalog,
|
|
result_payload,
|
|
select_profile,
|
|
select_slice,
|
|
select_split,
|
|
select_vendored,
|
|
)
|
|
from .policy import assert_rules_resolve, load_policy
|
|
from .preprocess import FileTarget, preprocess_repository, write_preprocess_artifact
|
|
from .provenance import fingerprint
|
|
from .rank import rank_obligations, rank_targets, write_rankings
|
|
from .regress import TRIGGERABLE, CandidateVerdict, run_regression, write_verdicts
|
|
from .reports import scan_exit_code, write_manifest, write_markdown_report, write_sarif_report
|
|
from .repos import (
|
|
DEFAULT_REPO_DIR,
|
|
RepoUnavailable,
|
|
checkout_path,
|
|
load_repo_recipes,
|
|
recipe_payload,
|
|
)
|
|
from .repos import resolve as resolve_repo
|
|
from .repos import select as select_repo
|
|
from .repos import verify as verify_repo
|
|
from .ruleset import DEFAULT_RULESET_DIR, load_ruleset
|
|
from .run import ScanRun, create_scan_run
|
|
from .sandbox import resolve_sandbox_probe, resolve_sandbox_runner
|
|
from .scoring import build_score_artifact
|
|
from .semantic import (
|
|
OverlayRecord,
|
|
adjudicate,
|
|
filter_promoted_hotspots,
|
|
finding_from_coverage,
|
|
load_facts,
|
|
write_overlay,
|
|
)
|
|
from .semantic.facts import FactLoadError, SemanticFacts
|
|
from .stages import Stage, plan_for_mode, record_completed, record_skip, skip_as_degradation, stages_payload
|
|
from .verification import VerificationResult, verify_findings, write_verification_results
|
|
from .verify_judge import verify_findings_dispatch
|
|
|
|
CALIBRATION_DIR = ".openultrasast/calibration"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ScanOutcome:
|
|
scan_id: str
|
|
run_dir: Path
|
|
file_target_count: int
|
|
ranked_target_count: int
|
|
finding_count: int
|
|
calibrations_applied: int
|
|
exit_code: int
|
|
|
|
|
|
if TYPE_CHECKING:
|
|
pass
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
load_dotenv()
|
|
parser = argparse.ArgumentParser(prog="ousast")
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
|
|
scan = subparsers.add_parser("scan", help="scan a local repository")
|
|
scan.add_argument("path", type=Path)
|
|
scan.add_argument("--mode", choices=("quick", "standard", "deep"), default="quick")
|
|
scan.add_argument("--config", type=Path, default=Path("openultrasast.toml"))
|
|
scan.add_argument("--fail-on", choices=("never", "findings", "verified", "worth-fixing"), default="never")
|
|
|
|
index = subparsers.add_parser("index", help="chunk a local repository for embedding index construction")
|
|
index.add_argument("path", type=Path)
|
|
index.add_argument("--config", type=Path, default=Path("openultrasast.toml"))
|
|
index.add_argument("--chunk-lines", type=int, default=80)
|
|
|
|
benchmark = subparsers.add_parser("benchmark", help="run a benchmark manifest and write scoreboard artifacts")
|
|
benchmark.add_argument("manifest", type=Path)
|
|
benchmark.add_argument("--mode", choices=("quick", "standard", "deep"), default="quick")
|
|
benchmark.add_argument("--config", type=Path, default=Path("openultrasast.toml"))
|
|
|
|
improve = subparsers.add_parser(
|
|
"improve",
|
|
help="run the bounded self-improvement loop against a benchmark manifest and update the loop-owned ruleset ledger",
|
|
)
|
|
improve.add_argument("manifest", type=Path)
|
|
improve.add_argument("--max-rounds", type=int, default=5)
|
|
improve.add_argument("--recall-floor", type=float, default=RECALL_FLOOR)
|
|
improve.add_argument("--fp-ceiling", type=float, default=FALSE_POSITIVE_CEILING)
|
|
improve.add_argument(
|
|
"--ledger", type=Path, default=None, help="ruleset ledger path (default: <target>/.openultrasast/calibration/rule_policy.json)"
|
|
)
|
|
improve.add_argument("--journal", type=Path, default=None, help="improvement journal path (default: alongside the ledger)")
|
|
improve.add_argument(
|
|
"--ruleset-dir", type=Path, default=DEFAULT_RULESET_DIR, help="ruleset directory to improve (default: the bundled ruleset)"
|
|
)
|
|
improve.add_argument("--dry-run", action="store_true", help="run rounds against a throwaway ledger; never touch the target's ledger")
|
|
improve.add_argument(
|
|
"--pair-catalog",
|
|
type=Path,
|
|
default=DEFAULT_CATALOG,
|
|
help="pair catalog whose holdout split gates every round per provenance profile",
|
|
)
|
|
improve.add_argument(
|
|
"--no-pair-gate", action="store_true", help="skip the per-profile holdout clause (faster; not for accepted ledgers)"
|
|
)
|
|
improve.add_argument(
|
|
"--profile-tolerance", type=float, default=0.0, help="allowed per-profile drop in pair_correct/Youden before rejecting"
|
|
)
|
|
improve.add_argument("--min-holdout-pairs", type=int, default=5, help="profiles with fewer holdout pairs are reported, not gated")
|
|
|
|
pairs = subparsers.add_parser(
|
|
"pairs",
|
|
help="scan isolated vuln-vs-fixed pairs (TP on vuln, silent on fix) and emit improve-loop signals",
|
|
)
|
|
pairs.add_argument("--catalog", type=Path, default=DEFAULT_CATALOG)
|
|
pairs.add_argument("--slice", choices=SLICE_NAMES, default="all")
|
|
pairs.add_argument(
|
|
"--profile", choices=("all", "human", "agent", "mixed", "synthetic"), default="all", help="filter pairs by provenance profile"
|
|
)
|
|
pairs.add_argument("--split", choices=("all", "train", "holdout"), default="all", help="filter pairs by declared split")
|
|
pairs.add_argument("--pointers", action="store_true", help="allow network for non-vendored pointer pairs this run (nightly; CI never)")
|
|
pairs.add_argument("--json", action="store_true", help="print the pair scoreboard as JSON")
|
|
|
|
repos = subparsers.add_parser(
|
|
"repos",
|
|
help="pinned known-vulnerable checkouts: list them, and verify a recipe against the code it names",
|
|
)
|
|
repos.add_argument("--dir", type=Path, default=DEFAULT_REPO_DIR)
|
|
repos.add_argument("--repo", default="all", help="one recipe by name, or all")
|
|
repos.add_argument("--fetch", action="store_true", help="allow network to materialise missing checkouts (CI never)")
|
|
repos.add_argument("--json", action="store_true")
|
|
|
|
subparsers.add_parser("mcp", help="run the narrow MCP server over stdio for OpenCode integration")
|
|
model_cmd = subparsers.add_parser("model", help="maintainer: the model layer that arbitrates detector claims")
|
|
model_sub = model_cmd.add_subparsers(dest="model_command", required=True)
|
|
candidates = model_sub.add_parser("candidates", help="offline: what the candidate enumerator can reach, per family and per slice")
|
|
candidates.add_argument("--out", type=Path, default=Path(".openultrasast/model"), help="where artifacts are written")
|
|
candidates.add_argument("--catalog", type=Path, default=DEFAULT_CATALOG)
|
|
candidates.add_argument("--slice", choices=SLICE_NAMES, default="all")
|
|
candidates.add_argument("--generator", choices=("ir", "ruleset"), default="ir", help="which pass enumerates the candidate sites")
|
|
candidates.add_argument("--json", action="store_true")
|
|
|
|
args = parser.parse_args(argv)
|
|
if args.command == "scan":
|
|
return _scan(args.path, args.config, args.mode, args.fail_on)
|
|
if args.command == "index":
|
|
return _index(args.path, args.config, args.chunk_lines)
|
|
if args.command == "benchmark":
|
|
return _benchmark(args.manifest, args.config, args.mode)
|
|
if args.command == "improve":
|
|
return _improve(
|
|
args.manifest,
|
|
max_rounds=args.max_rounds,
|
|
recall_floor=args.recall_floor,
|
|
fp_ceiling=args.fp_ceiling,
|
|
ledger=args.ledger,
|
|
journal=args.journal,
|
|
ruleset_dir=args.ruleset_dir,
|
|
dry_run=args.dry_run,
|
|
pair_catalog=None if args.no_pair_gate else args.pair_catalog,
|
|
profile_tolerance=args.profile_tolerance,
|
|
min_holdout_pairs=args.min_holdout_pairs,
|
|
)
|
|
if args.command == "pairs":
|
|
return _pairs(
|
|
args.catalog,
|
|
args.slice,
|
|
json_out=args.json,
|
|
profile=args.profile,
|
|
split=args.split,
|
|
pointers=args.pointers,
|
|
)
|
|
if args.command == "repos":
|
|
return _repos(args.dir, args.repo, fetch=args.fetch, json_out=args.json)
|
|
if args.command == "model":
|
|
return _model_candidates(args)
|
|
if args.command == "mcp":
|
|
from .mcp import serve # lazy: keeps the import cycle (mcp -> cli) one-directional
|
|
|
|
return serve()
|
|
return 2
|
|
|
|
|
|
def _scan(path: Path, config_path: Path, mode: str, fail_on: str) -> int:
|
|
outcome = _run_scan(path, config_path, mode, fail_on)
|
|
_print_scan_outcome(outcome)
|
|
return outcome.exit_code
|
|
|
|
|
|
def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOutcome:
|
|
if not path.exists() or not path.is_dir():
|
|
raise SystemExit(f"scan path is not a directory: {path}")
|
|
|
|
plan = plan_for_mode(mode)
|
|
config = load_config(config_path if config_path.exists() else None)
|
|
# Capability gates for the optional HarnessX agentic plane. When the extra is
|
|
# absent (default/CI), both stay False and the deterministic path runs unchanged.
|
|
hunter_model = config.models.hunter
|
|
verifier_model = config.models.verifier
|
|
harnessx_present = has_harnessx()
|
|
hx_hunter = mode == "standard" and bool(hunter_model) and harnessx_present
|
|
hx_verify = mode == "standard" and bool(verifier_model) and harnessx_present
|
|
run = create_scan_run(path, config)
|
|
runtime = HarnessRuntime(
|
|
scan_id=run.scan_id,
|
|
config=config,
|
|
trace_writer=HarnessTraceWriter(run.root / "trace" / "events.jsonl", redact=config.hardening.redact_secrets),
|
|
)
|
|
write_harness_config(config=config, processors=[], contract_mode="strict", path=run.root / "harness.json")
|
|
runtime.start(mode=mode, target=run.target)
|
|
policy = runtime.run_stage("policy_load", load_policy)
|
|
ledger_path = run.target / CALIBRATION_DIR / "false_positive_learnings.json"
|
|
rule_ledger_path = run.target / CALIBRATION_DIR / "rule_policy.json"
|
|
ruleset = runtime.run_stage("ruleset_load", lambda: load_ruleset(DEFAULT_RULESET_DIR, rule_ledger_path))
|
|
runtime.run_stage("policy_check", lambda: assert_rules_resolve(ruleset, policy))
|
|
static_hints = runtime.run_stage("static_mapping", lambda: _load_static_hints(config.static_analysis.sarif_paths))
|
|
write_static_hints(static_hints, run.root / "mapping" / "static_hints.json")
|
|
snapshot, targets = runtime.run_stage(
|
|
"preprocess",
|
|
lambda: preprocess_repository(run.target, run.root / "preprocess" / "file_targets.json", static_hints),
|
|
)
|
|
entry_points = runtime.run_stage("entry_point_mapping", lambda: analyze_entry_points(run.target, targets))
|
|
write_entry_points(entry_points, run.root / "mapping" / "entry_points.json")
|
|
targets = attach_reachability_hints(targets, entry_points)
|
|
write_preprocess_artifact(snapshot, targets, run.root / "preprocess" / "file_targets.json")
|
|
provenance = runtime.run_stage("provenance", lambda: fingerprint(run.target))
|
|
rankings = runtime.run_stage("rank", lambda: rank_targets(targets))
|
|
prior_learnings = load_false_positive_learnings(ledger_path)
|
|
rankings, calibrations = runtime.run_stage("calibrate", lambda: calibrate_rankings(rankings, prior_learnings))
|
|
applied_calibrations = [calibration for calibration in calibrations if calibration.applied_learning_ids]
|
|
write_rankings(rankings, run.root / "rank" / "ranking.json")
|
|
write_ranking_calibrations(calibrations, run.root / "calibration" / "applied_calibrations.json")
|
|
if mode == "standard":
|
|
if hx_hunter and hunter_model:
|
|
orchestrator = HxScanOrchestrator(
|
|
provider_model=hunter_model,
|
|
provider=config.harnessx.provider,
|
|
max_cost_usd=config.harnessx.max_cost_usd,
|
|
token_threshold=config.harnessx.token_threshold,
|
|
)
|
|
hunter_result = runtime.run_stage(
|
|
"hunter_pool",
|
|
lambda: orchestrator.run_pool(
|
|
run.target, targets, rankings, scan_id=run.scan_id, ruleset=ruleset, policy=policy, emit=runtime.emit
|
|
),
|
|
)
|
|
else:
|
|
if bool(hunter_model) and not harnessx_present:
|
|
runtime.state["degradations"].append(
|
|
{"stage": "hunter_pool", "requested": "harnessx", "reason": "harnessx_extra_unavailable", "fallback": "run_hunter_pool"}
|
|
)
|
|
hunter_result = runtime.run_stage(
|
|
"hunter_pool",
|
|
lambda: run_hunter_pool(run.target, targets, rankings, scan_id=run.scan_id, ruleset=ruleset, policy=policy),
|
|
)
|
|
findings = hunter_result.findings
|
|
trajectories = hunter_result.trajectories
|
|
else:
|
|
findings = runtime.run_stage(
|
|
"quick_findings",
|
|
lambda: quick_scan_findings(
|
|
run.target,
|
|
targets,
|
|
rankings,
|
|
ruleset,
|
|
policy,
|
|
min_emit_priority=config.ruleset.min_emit_priority,
|
|
min_emit_precision=config.ruleset.min_emit_precision,
|
|
),
|
|
)
|
|
trajectories = []
|
|
# Shadow-status rules fire but are excluded from the report and the score;
|
|
# their outcomes are preserved separately for precision tracking.
|
|
shadow_findings = [finding for finding in findings if finding.status == "shadow"]
|
|
if shadow_findings:
|
|
findings = [finding for finding in findings if finding.status != "shadow"]
|
|
write_findings(shadow_findings, run.root / "shadow_findings.json")
|
|
# Bounded-CI budget: cap the reported finding count (0 = unlimited). Findings are
|
|
# severity/priority-sorted, so truncation keeps the most severe and is disclosed.
|
|
max_findings = config.hardening.max_findings
|
|
if max_findings and len(findings) > max_findings:
|
|
runtime.state["degradations"].append(
|
|
{"stage": "budget", "reason": "max_findings_exceeded", "requested": max_findings, "actual": len(findings)}
|
|
)
|
|
findings = findings[:max_findings]
|
|
findings_path = run.root / "findings.json"
|
|
verification_path = run.root / "verification.json"
|
|
markdown_path = run.root / "report.md"
|
|
sarif_path = run.root / "report.sarif"
|
|
manifest_path = run.root / "manifest.json"
|
|
trajectories_path = run.root / "traces" / "hunter_trajectories.jsonl"
|
|
write_findings(findings, findings_path)
|
|
if trajectories:
|
|
write_hunter_trajectories(trajectories, trajectories_path)
|
|
if mode == "standard" and bool(verifier_model) and not harnessx_present:
|
|
runtime.state["degradations"].append(
|
|
{"stage": "verify", "requested": "harnessx", "reason": "harnessx_extra_unavailable", "fallback": "structural_verifier"}
|
|
)
|
|
verifications = runtime.run_stage(
|
|
"verify",
|
|
lambda: verify_findings_dispatch(
|
|
findings,
|
|
verifier_model=verifier_model,
|
|
verifier_provider=config.harnessx.provider,
|
|
use_harnessx=hx_verify,
|
|
),
|
|
)
|
|
write_verification_results(verifications, verification_path)
|
|
runtime.run_stage(
|
|
"record_calibration",
|
|
lambda: _persist_calibration_feedback(run, ledger_path, prior_learnings, findings, verifications),
|
|
)
|
|
# Fusion: two-panel adjudication for triggered findings (standard mode). Runs
|
|
# deterministically by default; routes panels through the configured provider when
|
|
# a panel model is set and the extra is present, else falls back + records a degradation.
|
|
fusion_decisions: list[FusionDecision] = []
|
|
fusion_path = run.root / "fusion.json"
|
|
if mode == "standard" and config.fusion.enabled:
|
|
hx_fusion = bool(config.fusion.panel_model) and harnessx_present
|
|
if bool(config.fusion.panel_model) and not harnessx_present:
|
|
runtime.state["degradations"].append(
|
|
{"stage": "fusion", "requested": "harnessx", "reason": "harnessx_extra_unavailable", "fallback": "deterministic_panels"}
|
|
)
|
|
fusion_decisions = runtime.run_stage(
|
|
"fusion",
|
|
lambda: fuse_findings_dispatch(
|
|
findings,
|
|
verifications,
|
|
panel_model=config.fusion.panel_model,
|
|
decider_model=config.fusion.decider_model,
|
|
provider=config.harnessx.provider,
|
|
use_harnessx=hx_fusion,
|
|
high_assurance=config.fusion.high_assurance,
|
|
),
|
|
)
|
|
if fusion_decisions:
|
|
fusion_path.write_text(json.dumps([decision.to_dict() for decision in fusion_decisions], indent=2, sort_keys=True) + "\n")
|
|
cwe_by_rule = {rule.rule_id: rule.cwe for rule in ruleset}
|
|
rule_cwe_by_id = {finding.finding_id: cwe_by_rule.get(finding.finding_id.split(":", 1)[0], "") for finding in findings}
|
|
# A confirmed false positive (prior-scan learning) lowers a finding's effective
|
|
# reachability multiplier in the score instead of deleting the rule.
|
|
reachability_override = fp_reachability_overrides(findings, prior_learnings)
|
|
score_artifact = runtime.run_stage(
|
|
"score",
|
|
lambda: build_score_artifact(
|
|
findings,
|
|
rule_cwe_by_id,
|
|
policy,
|
|
k=config.score.k,
|
|
min_score=config.score.min_score,
|
|
block_severity_reachable=config.score.block_severity_reachable,
|
|
blocking=config.score.blocking,
|
|
reachability_override=reachability_override,
|
|
),
|
|
)
|
|
score_path = run.root / "score.json"
|
|
score_path.write_text(json.dumps(score_artifact.to_dict(), indent=2, sort_keys=True) + "\n")
|
|
plan = record_completed(plan, Stage.STATIC)
|
|
complexity_payload: dict[str, object] | None = None
|
|
complexity_map_path = run.root / "complexity_map.json"
|
|
overlay_path = run.root / "overlay.json"
|
|
built_map = None
|
|
overlay_records: list[OverlayRecord] = []
|
|
wrote_overlay = False
|
|
obligations_payload: dict[str, object] | None = None
|
|
model_payload: dict[str, object] | None = None
|
|
obligations_cited: dict[str, dict[str, object]] = {}
|
|
if Stage.MAP in plan.requested:
|
|
built_map = runtime.run_stage(
|
|
"map",
|
|
lambda: complexity_map.build_complexity_map(
|
|
targets,
|
|
findings,
|
|
complexity_map_path,
|
|
repo_files=[target.path for target in targets],
|
|
ledger_path=run.target / CALIBRATION_DIR / "complexity_ledger.json",
|
|
),
|
|
)
|
|
facts_or_error: SemanticFacts | FactLoadError
|
|
try:
|
|
facts_or_error = load_facts()
|
|
except FactLoadError as exc:
|
|
facts_or_error = exc
|
|
runtime.state["degradations"].append({"stage": "overlay", "reason": "facts_unavailable", "detail": str(exc)})
|
|
overlay_records = list(
|
|
runtime.run_stage(
|
|
"overlay",
|
|
lambda: adjudicate(root=run.target, targets=targets, findings=findings, facts=facts_or_error),
|
|
)
|
|
)
|
|
write_overlay(overlay_records, overlay_path)
|
|
wrote_overlay = True
|
|
coverage_findings = [finding_from_coverage(record) for record in overlay_records if record.disposition == "coverage"]
|
|
if coverage_findings:
|
|
seen = {finding.finding_id for finding in findings}
|
|
extra = [item for item in coverage_findings if item.finding_id not in seen]
|
|
if extra:
|
|
findings = findings + extra
|
|
write_findings(findings, findings_path)
|
|
if config.obligations.enabled:
|
|
obligation_findings, obligations_payload, obligations_cited = _check_obligations(
|
|
root=run.target,
|
|
targets=targets,
|
|
entries=entry_points,
|
|
facts=facts_or_error,
|
|
settings=config.obligations,
|
|
runtime=runtime,
|
|
hunter_model=hunter_model,
|
|
)
|
|
if obligation_findings:
|
|
findings = rank_obligations(findings + obligation_findings)
|
|
write_findings(findings, findings_path)
|
|
# Fifth MAP proposer: the model layer (contributor-scan Req 1). Additive -- it adds findings beside
|
|
# the pattern, overlay and obligation ones and removes none of them.
|
|
model_findings, model_payload = _run_model_layer(
|
|
root=run.target,
|
|
targets=targets,
|
|
entries=entry_points,
|
|
settings=config.model,
|
|
runtime=runtime,
|
|
config=config,
|
|
)
|
|
if model_findings:
|
|
findings = findings + model_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"))
|
|
else:
|
|
built_map, findings, verifications = _attach_tool_hunter(
|
|
root=run.target,
|
|
built_map=built_map,
|
|
findings=findings,
|
|
verifications=verifications,
|
|
hunter_model=hunter_model,
|
|
max_hotspots=config.complexity.max_hunter_hotspots,
|
|
max_findings=max_findings,
|
|
findings_path=findings_path,
|
|
verification_path=verification_path,
|
|
complexity_map_path=complexity_map_path,
|
|
runtime=runtime,
|
|
)
|
|
complexity_payload = {
|
|
"hotspot_count": len(built_map.hotspots),
|
|
"heuristic_only": built_map.heuristic_only,
|
|
}
|
|
verdicts_path = run.root / "verdicts.json"
|
|
wrote_verdicts = False
|
|
verdict_records: tuple[CandidateVerdict, ...] = ()
|
|
if Stage.REGRESS in plan.requested:
|
|
probe = resolve_sandbox_probe()
|
|
if not probe.available():
|
|
plan = record_skip(plan, Stage.REGRESS, "sandbox_unavailable")
|
|
runtime.state["degradations"].append(skip_as_degradation(Stage.REGRESS, "sandbox_unavailable"))
|
|
else:
|
|
sandbox = resolve_sandbox_runner()
|
|
hotspots = built_map.hotspots if built_map is not None else ()
|
|
hotspots = filter_promoted_hotspots(hotspots, overlay_records, findings)
|
|
promoted = [
|
|
finding
|
|
for finding in findings
|
|
if finding.finding_id in {record.proposal_id for record in overlay_records if record.disposition == "promote"}
|
|
]
|
|
cwe_by_rule_id = {rule.rule_id: rule.cwe for rule in ruleset}
|
|
rule_cwe_by_finding = {finding.finding_id: cwe_by_rule_id.get(finding.finding_id.split(":", 1)[0], "") for finding in promoted}
|
|
languages_by_path = {target.path: target.language for target in targets}
|
|
records = runtime.run_stage(
|
|
"regress",
|
|
lambda: run_regression(
|
|
hotspots,
|
|
promoted,
|
|
max_candidates=config.regress.max_candidates,
|
|
policy=policy,
|
|
rule_cwe=rule_cwe_by_finding,
|
|
languages_by_path=languages_by_path,
|
|
repo_root=run.target,
|
|
sandbox=sandbox,
|
|
sandbox_limits=config.sandbox,
|
|
images=dict(config.regress.images),
|
|
),
|
|
)
|
|
write_verdicts(records, verdicts_path)
|
|
persist_verdicts(run.target / CALIBRATION_DIR / "complexity_ledger.json", records)
|
|
# Req 6.4: proof rungs live on verdicts, not on StaticFinding, so obligations re-rank once the sandbox has spoken.
|
|
proven_ids = {fid for record in records if record.verdict == TRIGGERABLE for fid in record.inventory_finding_ids}
|
|
if proven_ids and any(finding.finding_id.startswith("obligation:") for finding in findings):
|
|
findings = rank_obligations(findings, proven_ids=proven_ids)
|
|
write_findings(findings, findings_path)
|
|
verdict_records = records
|
|
wrote_verdicts = True
|
|
plan = record_completed(plan, Stage.REGRESS)
|
|
worth_fixing_payload = _worth_fixing_payload(verdict_records) if wrote_verdicts else None
|
|
runtime.run_stage(
|
|
"report",
|
|
lambda: write_markdown_report(
|
|
findings,
|
|
markdown_path,
|
|
verifications,
|
|
redact=config.hardening.redact_secrets,
|
|
complexity_map=built_map,
|
|
verdicts=verdict_records if wrote_verdicts else None,
|
|
overlay=overlay_records if wrote_overlay else None,
|
|
obligations=obligations_cited or None,
|
|
obligations_summary=obligations_payload or None,
|
|
),
|
|
)
|
|
runtime.run_stage(
|
|
"sarif",
|
|
lambda: write_sarif_report(
|
|
findings,
|
|
verifications,
|
|
sarif_path,
|
|
overlay=overlay_records if wrote_overlay else None,
|
|
obligations=obligations_cited or None,
|
|
),
|
|
)
|
|
runtime.run_stage(
|
|
"manifest",
|
|
lambda: write_manifest(
|
|
run=run,
|
|
findings=findings,
|
|
verifications=verifications,
|
|
artifact_paths=_artifact_paths(
|
|
findings=findings_path,
|
|
verification=verification_path,
|
|
markdown=markdown_path,
|
|
sarif=sarif_path,
|
|
score=score_path,
|
|
trajectories=trajectories_path if trajectories else None,
|
|
fusion=fusion_path if fusion_decisions else None,
|
|
complexity_map=complexity_map_path if complexity_payload is not None else None,
|
|
overlay=overlay_path if wrote_overlay else None,
|
|
verdicts=verdicts_path if wrote_verdicts else None,
|
|
),
|
|
path=manifest_path,
|
|
score=score_artifact.to_dict(),
|
|
degradations=runtime.state["degradations"] or None,
|
|
fusion=[_fusion_summary(decision) for decision in fusion_decisions] or None,
|
|
stages=stages_payload(plan),
|
|
complexity=complexity_payload,
|
|
worth_fixing=worth_fixing_payload,
|
|
provenance=provenance.to_dict(),
|
|
obligations=obligations_payload,
|
|
model=model_payload,
|
|
),
|
|
)
|
|
runtime.finish(status="succeeded")
|
|
return ScanOutcome(
|
|
scan_id=run.scan_id,
|
|
run_dir=run.root,
|
|
file_target_count=len(targets),
|
|
ranked_target_count=len(rankings),
|
|
finding_count=len(findings),
|
|
calibrations_applied=len(applied_calibrations),
|
|
exit_code=scan_exit_code(findings, verifications, fail_on, worth_fixing_verdicts=verdict_records),
|
|
)
|
|
|
|
|
|
def _attach_tool_hunter(
|
|
*,
|
|
root: Path,
|
|
built_map: complexity_map.ComplexityMap,
|
|
findings: list[StaticFinding],
|
|
verifications: list[VerificationResult],
|
|
hunter_model: str,
|
|
max_hotspots: int,
|
|
max_findings: int,
|
|
findings_path: Path,
|
|
verification_path: Path,
|
|
complexity_map_path: Path,
|
|
runtime: HarnessRuntime,
|
|
) -> tuple[complexity_map.ComplexityMap, list[StaticFinding], list[VerificationResult]]:
|
|
client = tool_hunter.resolve_hunter_client()
|
|
if client is None:
|
|
return built_map, findings, verifications
|
|
hunter_hotspots = list(built_map.hotspots[: max(max_hotspots, 0)])
|
|
hunter_findings = runtime.run_stage(
|
|
"tool_hunter",
|
|
lambda: tool_hunter.run_tool_hunter(
|
|
root,
|
|
hunter_hotspots,
|
|
client=client,
|
|
model=hunter_model,
|
|
max_steps=tool_hunter.DEFAULT_MAX_STEPS,
|
|
),
|
|
)
|
|
seen = {finding.finding_id for finding in findings}
|
|
extra = [finding for finding in hunter_findings if finding.finding_id not in seen]
|
|
if extra:
|
|
findings = findings + extra
|
|
if max_findings and len(findings) > max_findings:
|
|
if not any(entry.get("reason") == "max_findings_exceeded" for entry in runtime.state["degradations"]):
|
|
runtime.state["degradations"].append(
|
|
{"stage": "budget", "reason": "max_findings_exceeded", "requested": max_findings, "actual": len(findings)}
|
|
)
|
|
findings = findings[:max_findings]
|
|
kept = {finding.finding_id for finding in findings}
|
|
extra = [finding for finding in extra if finding.finding_id in kept]
|
|
if extra:
|
|
verifications = verifications + verify_findings(extra)
|
|
write_verification_results(verifications, verification_path)
|
|
write_findings(findings, findings_path)
|
|
built_map = replace(built_map, heuristic_only=False)
|
|
complexity_map.write_complexity_map(built_map, complexity_map_path)
|
|
return built_map, findings, verifications
|
|
|
|
|
|
def _persist_calibration_feedback(
|
|
run: ScanRun,
|
|
ledger_path: Path,
|
|
prior_learnings: list[FalsePositiveLearning],
|
|
findings: list[StaticFinding],
|
|
verifications: list[VerificationResult],
|
|
) -> list[FalsePositiveLearning]:
|
|
# Store rejected or held findings as scoped learnings for the next scan.
|
|
new_learnings = learnings_from_verifications(findings, verifications)
|
|
merged = merge_false_positive_learnings(prior_learnings, new_learnings)
|
|
write_false_positive_learnings(merged, ledger_path)
|
|
write_false_positive_learnings(new_learnings, run.root / "calibration" / "false_positive_learnings.json")
|
|
return merged
|
|
|
|
|
|
def _print_scan_outcome(outcome: ScanOutcome) -> None:
|
|
print(f"scan_id={outcome.scan_id}")
|
|
print(f"run_dir={outcome.run_dir}")
|
|
print(f"file_targets={outcome.file_target_count}")
|
|
print(f"ranked_targets={outcome.ranked_target_count}")
|
|
print(f"findings={outcome.finding_count}")
|
|
print(f"calibrations_applied={outcome.calibrations_applied}")
|
|
|
|
|
|
def _load_static_hints(sarif_paths: tuple[str, ...]) -> list[object]:
|
|
hints: list[object] = []
|
|
for sarif_path in sarif_paths:
|
|
path = Path(sarif_path)
|
|
if path.exists():
|
|
hints.extend(ingest_sarif(path))
|
|
return hints
|
|
|
|
|
|
def _worth_fixing_payload(verdicts: tuple[CandidateVerdict, ...]) -> dict[str, object]:
|
|
worth = [record for record in verdicts if record.worth_fixing]
|
|
return {
|
|
"count": len(worth),
|
|
"verdicts": [
|
|
{
|
|
"path": record.path,
|
|
"function_name": record.function_name,
|
|
"verdict": record.verdict,
|
|
"reason": record.reason,
|
|
"worth_fixing": True,
|
|
}
|
|
for record in worth
|
|
],
|
|
}
|
|
|
|
|
|
def _artifact_paths(**paths: Path | None) -> dict[str, Path]:
|
|
return {name: path for name, path in paths.items() if path is not None}
|
|
|
|
|
|
def _fusion_summary(decision: FusionDecision) -> dict[str, object]:
|
|
return {
|
|
"finding_id": decision.finding_id,
|
|
"disposition": str(decision.disposition),
|
|
"decision_source": decision.decision_source,
|
|
"triggers": decision.triggers,
|
|
}
|
|
|
|
|
|
def _benchmark(manifest_path: Path, config_path: Path, mode: str) -> int:
|
|
if not manifest_path.exists() or not manifest_path.is_file():
|
|
raise SystemExit(f"benchmark manifest is not a file: {manifest_path}")
|
|
manifest = load_benchmark_manifest(manifest_path)
|
|
target = resolve_benchmark_source(manifest_path, manifest)
|
|
run = create_benchmark_run(target, manifest)
|
|
started_at = perf_counter()
|
|
outcome = _run_scan(target, config_path, mode, "never")
|
|
runtime_seconds = perf_counter() - started_at
|
|
findings = _load_scan_findings(outcome.run_dir)
|
|
baseline_findings = load_baseline_findings(manifest_path, manifest)
|
|
result = evaluate_benchmark(
|
|
run=run,
|
|
mode=mode,
|
|
findings=findings,
|
|
scan_id=outcome.scan_id,
|
|
scan_run_dir=outcome.run_dir,
|
|
runtime_seconds=runtime_seconds,
|
|
baseline_findings=baseline_findings,
|
|
)
|
|
write_benchmark_artifacts(run, manifest_path, result)
|
|
_print_scan_outcome(outcome)
|
|
print(f"benchmark_run_id={run.benchmark_run_id}")
|
|
print(f"benchmark_run_dir={run.root}")
|
|
print(f"benchmark_expected={result.metrics.expected_findings_total}")
|
|
print(f"benchmark_matched={result.metrics.matched_findings_total}")
|
|
print(f"benchmark_missed={result.metrics.missed_findings_total}")
|
|
return 0
|
|
|
|
|
|
def _improve(
|
|
manifest_path: Path,
|
|
*,
|
|
max_rounds: int,
|
|
recall_floor: float,
|
|
fp_ceiling: float,
|
|
ledger: Path | None,
|
|
journal: Path | None,
|
|
ruleset_dir: Path,
|
|
dry_run: bool,
|
|
pair_catalog: Path | None = None,
|
|
profile_tolerance: float = 0.0,
|
|
min_holdout_pairs: int = 5,
|
|
) -> int:
|
|
if not manifest_path.exists() or not manifest_path.is_file():
|
|
raise SystemExit(f"benchmark manifest is not a file: {manifest_path}")
|
|
pair_cases: tuple[object, ...] = ()
|
|
if pair_catalog is not None:
|
|
if not pair_catalog.is_file():
|
|
raise SystemExit(f"pair catalog is not a file: {pair_catalog} (pass --no-pair-gate to skip the per-profile holdout clause)")
|
|
pair_cases = select_split(load_pair_catalog(pair_catalog), "holdout")
|
|
manifest = load_benchmark_manifest(manifest_path)
|
|
target = resolve_benchmark_source(manifest_path, manifest)
|
|
policy = load_policy()
|
|
|
|
# The loop writes its accepted ledger where `scan`/`benchmark` read it, so a
|
|
# subsequent scan of this target automatically picks up the improved ruleset.
|
|
default_ledger = target / CALIBRATION_DIR / "rule_policy.json"
|
|
with tempfile.TemporaryDirectory(prefix="ousast-improve-") as scratch:
|
|
if dry_run:
|
|
ledger_path = Path(scratch) / "rule_policy.json"
|
|
journal_path = Path(scratch) / "improve_journal.json"
|
|
else:
|
|
ledger_path = ledger or default_ledger
|
|
journal_path = journal or ledger_path.with_name("improve_journal.json")
|
|
|
|
outcomes = run_improvement(
|
|
target,
|
|
manifest,
|
|
ledger_path=ledger_path,
|
|
journal_path=journal_path,
|
|
ruleset_dir=ruleset_dir,
|
|
policy=policy,
|
|
max_rounds=max_rounds,
|
|
recall_floor=recall_floor,
|
|
fp_ceiling=fp_ceiling,
|
|
pair_cases=pair_cases,
|
|
profile_tolerance=profile_tolerance,
|
|
min_holdout_pairs=min_holdout_pairs,
|
|
)
|
|
_print_improve_outcomes(outcomes, manifest_path, target, ledger_path, dry_run=dry_run)
|
|
return 0
|
|
|
|
|
|
def _pairs(
|
|
catalog: Path,
|
|
slice_name: str,
|
|
*,
|
|
json_out: bool,
|
|
profile: str = "all",
|
|
split: str = "all",
|
|
pointers: bool = False,
|
|
) -> int:
|
|
if not catalog.exists() or not catalog.is_file():
|
|
raise SystemExit(f"pair catalog is not a file: {catalog}")
|
|
cases = select_split(select_profile(select_slice(load_pair_catalog(catalog), slice_name), profile), split)
|
|
try:
|
|
result = evaluate_catalog(cases, pointers=True if pointers else None)
|
|
except ValueError as exc:
|
|
raise SystemExit(f"--k-runs: {exc}") from exc
|
|
if json_out:
|
|
print(json.dumps(result_payload(result), indent=2, sort_keys=True))
|
|
return 0
|
|
print_pair_metrics("pair eval", result)
|
|
print(f"signals={len(result.signals)}")
|
|
return 0
|
|
|
|
|
|
def _repos(directory: Path, name: str, *, fetch: bool, json_out: bool) -> int:
|
|
"""List the pinned checkouts and check each recipe against the code it claims to describe.
|
|
|
|
Offline by default: without ``--fetch`` a missing checkout is reported as missing, not fetched. Exit is
|
|
non-zero only when a materialised checkout CONTRADICTS its recipe -- a recipe that merely has not been
|
|
fetched is not a failure, or CI could never run this.
|
|
"""
|
|
recipes = select_repo(load_repo_recipes(directory), name)
|
|
if not recipes:
|
|
raise SystemExit(f"no repository recipe matches {name!r} in {directory}")
|
|
|
|
rows: list[dict[str, object]] = []
|
|
contradicted = 0
|
|
for recipe in recipes:
|
|
problems: tuple[str, ...] = ()
|
|
try:
|
|
root = resolve_repo(recipe, fetch=True if fetch else None)
|
|
except RepoUnavailable:
|
|
state, path = "absent", checkout_path(recipe)
|
|
else:
|
|
problems = verify_repo(recipe, root)
|
|
state, path = ("contradicted" if problems else "verified"), root
|
|
contradicted += 1 if problems else 0
|
|
rows.append({**recipe_payload(recipe), "state": state, "path": str(path), "problems": list(problems)})
|
|
|
|
if not json_out:
|
|
in_scope = len(recipe.in_scope)
|
|
measures = ",".join(recipe.measures)
|
|
print(f"{recipe.name:10} {state:13} {recipe.language:8} measures={measures} known={len(recipe.known)} in_scope={in_scope}")
|
|
for problem in problems:
|
|
print(f" ! {problem}")
|
|
if state == "absent":
|
|
print(f" fetch with: ousast repos --repo {recipe.name} --fetch")
|
|
|
|
if json_out:
|
|
print(json.dumps(rows, indent=2, sort_keys=True))
|
|
return 1 if contradicted else 0
|
|
|
|
|
|
def _model_skipped(reason: str) -> dict[str, object]:
|
|
"""A skipped stage reports the same SHAPE as a run one, plus why it skipped.
|
|
|
|
A consumer reading this should not have to branch on whether the stage ran; a payload that is sometimes
|
|
a full record and sometimes a single key is how a report ends up silently omitting a section.
|
|
"""
|
|
return {**_model_payload(None), "skipped": reason}
|
|
|
|
|
|
def _model_payload(result: object, *, unjudged_paths: tuple[str, ...] = ()) -> dict[str, object]:
|
|
"""What the scan records about the model layer.
|
|
|
|
`unjudged_sample` names the regions the budget did not reach, lowest-ranked first. Counting them is the
|
|
honesty requirement (Req 3.3); NAMING them is what makes an exclusion list for a large repository
|
|
something a maintainer can write from evidence rather than guesswork.
|
|
"""
|
|
return {
|
|
"by_rung": dict(getattr(result, "by_rung", {}) or {}),
|
|
"regions_scanned": getattr(result, "regions_scanned", 0),
|
|
"regions_unjudged": getattr(result, "regions_unjudged", 0),
|
|
"regions_unasked": getattr(result, "regions_unasked", 0),
|
|
"unjudged_sample": list(unjudged_paths),
|
|
"model_calls": getattr(result, "model_calls", 0),
|
|
"cost_usd": getattr(result, "cost_usd", 0.0),
|
|
"seconds": getattr(result, "seconds", 0.0),
|
|
"build_seconds": getattr(result, "build_seconds", 0.0),
|
|
"query_seconds": getattr(result, "query_seconds", 0.0),
|
|
"query_seconds_by_kind": dict(getattr(result, "query_seconds_by_kind", {}) or {}),
|
|
"arbitrate_seconds": getattr(result, "arbitrate_seconds", 0.0),
|
|
}
|
|
|
|
|
|
def _run_model_layer(
|
|
*,
|
|
root: Path,
|
|
targets: list[FileTarget],
|
|
entries: Sequence[object],
|
|
settings: ModelLayerConfig,
|
|
runtime: HarnessRuntime,
|
|
config: ResolvedConfig,
|
|
) -> tuple[list[StaticFinding], dict[str, object]]:
|
|
"""Fifth MAP proposer (contributor-scan Req 1): the model layer, arbitrating regions of the repository.
|
|
|
|
Additive by construction. Without a CPG engine this records a reason and returns nothing, and the scan is
|
|
byte-for-byte what it is today -- the model layer must never make the tool worse for someone who does not
|
|
have Joern.
|
|
"""
|
|
if not settings.enabled:
|
|
return [], _model_skipped("disabled")
|
|
|
|
from .cpg.backend import resolve_cpg_backend
|
|
from .model.endpoint import resolve_chat_endpoint, resolve_models
|
|
from .model.regions import regions_for
|
|
from .model.scan import ScanBudget, scan_repository
|
|
from .model.shipped import declared_sources
|
|
|
|
backend = resolve_cpg_backend()
|
|
if not backend.available():
|
|
runtime.state["degradations"].append({"stage": "model", "reason": "cpg_unavailable"})
|
|
return [], _model_skipped("cpg_unavailable")
|
|
|
|
# The project's own build declaration, when it makes one, decides what the budget is spent on first.
|
|
regions = regions_for(entries, targets, shipped=declared_sources(root))
|
|
if not regions:
|
|
return [], _model_skipped("no_regions")
|
|
|
|
resolved = resolve_chat_endpoint(config)
|
|
client = resolved[0] if resolved else None
|
|
if client is None:
|
|
# Not a failure: entailment still works and only the suspicion band goes unasked (Req 11.3).
|
|
runtime.state["degradations"].append({"stage": "model", "reason": "learning_endpoint_unavailable"})
|
|
|
|
result = runtime.run_stage(
|
|
"model",
|
|
lambda: scan_repository(
|
|
root,
|
|
regions,
|
|
backend=backend,
|
|
client=client,
|
|
model=resolve_models(config)[0],
|
|
budget=ScanBudget(max_model_calls=settings.max_model_calls, max_regions=settings.max_regions),
|
|
),
|
|
)
|
|
for degradation in result.degradations:
|
|
runtime.state["degradations"].append(dict(degradation))
|
|
|
|
# The regions the budget did not reach, weakest first -- the evidence an exclusion list is built from.
|
|
unjudged = tuple(region.path for region in regions[result.regions_scanned :][:10])
|
|
findings = [_finding_from_model(item) for item in result.findings]
|
|
return findings, _model_payload(result, unjudged_paths=unjudged)
|
|
|
|
|
|
def _finding_from_model(item: object) -> StaticFinding:
|
|
"""A model finding as a StaticFinding, carrying its rung and witness (Req 5.1)."""
|
|
site = str(getattr(item, "site", ""))
|
|
path, _, rest = site.partition(":")
|
|
line, _, function = rest.partition(":")
|
|
rung = getattr(item, "rung", None)
|
|
witness = str(getattr(item, "witness", "") or getattr(item, "contradiction", ""))
|
|
family = str(getattr(item, "family", ""))
|
|
return StaticFinding(
|
|
finding_id=f"model:{family}:{site}",
|
|
path=path or site,
|
|
title=f"{family} ({getattr(rung, 'value', rung)})",
|
|
severity="medium",
|
|
confidence="medium",
|
|
evidence_level="suspicion",
|
|
rationale=witness or f"the model layer reported a {family} candidate",
|
|
line=int(line) if line.isdigit() else None,
|
|
function_name=function or None,
|
|
reachability_status="unknown",
|
|
reachability_evidence=[],
|
|
reachability_conditions=[],
|
|
tags=[f"family:{family}", f"rung:{getattr(rung, 'value', rung)}"],
|
|
ranking_priority=0.0,
|
|
rung=str(getattr(rung, "value", rung) or "suspicion"),
|
|
)
|
|
|
|
|
|
def _check_obligations(
|
|
*,
|
|
root: Path,
|
|
targets: list[FileTarget],
|
|
entries: Sequence[object],
|
|
facts: SemanticFacts | FactLoadError,
|
|
settings: ObligationsConfig,
|
|
runtime: HarnessRuntime,
|
|
hunter_model: str | None = None,
|
|
) -> tuple[list[StaticFinding], dict[str, object], dict[str, dict[str, object]]]:
|
|
"""Fourth MAP proposer (authorization-obligations Req 3-6): obligations without a dominating discharger, at suspicion.
|
|
|
|
Function-local until path records exist; the declared policy is read from the target; obligation findings never
|
|
enter the sandbox candidate set.
|
|
"""
|
|
from .semantic.ir import parse_file
|
|
from .semantic.obligations import load_obligation_facts
|
|
from .semantic.obligations.check import check_obligations, findings_to_static
|
|
from .semantic.obligations.dominance import OrderDominance
|
|
from .semantic.obligations.policy import PolicyError, load_declared_policy
|
|
|
|
payload: dict[str, object] = {"sibling_sets": 0, "under_populated": 0, "operations": 0, "findings_by_label": {}, "policy_version": None}
|
|
if isinstance(facts, FactLoadError):
|
|
runtime.state["degradations"].append({"stage": "obligations", "reason": "facts_unavailable"})
|
|
return [], payload, {}
|
|
irs: dict[str, tuple[Any, str]] = {}
|
|
texts: dict[str, str] = {}
|
|
for target in targets:
|
|
try:
|
|
text = (root / target.path).read_text(errors="ignore")
|
|
except OSError:
|
|
continue
|
|
texts[target.path] = text
|
|
irs[target.path] = (parse_file(target.path, text, target.language), text)
|
|
policy = None
|
|
policy_path = root / settings.policy_path
|
|
try:
|
|
policy = load_declared_policy(policy_path)
|
|
except PolicyError as exc:
|
|
runtime.state["degradations"].append({"stage": "obligations", "reason": "policy_invalid", "detail": str(exc)[:200]})
|
|
result = runtime.run_stage(
|
|
"obligations",
|
|
lambda: check_obligations(
|
|
irs=irs,
|
|
entries=entries,
|
|
facts=load_obligation_facts(),
|
|
flow_facts=facts,
|
|
policy=policy,
|
|
paths=(),
|
|
dominance=OrderDominance(texts=texts),
|
|
store_shapes=(), # known-fix hints came from the deleted mechanism store
|
|
min_siblings=settings.min_siblings,
|
|
),
|
|
)
|
|
from .semantic.obligations.intent import adjudicate_intent
|
|
|
|
result = adjudicate_intent(result, client=tool_hunter.resolve_hunter_client(), model=hunter_model or "", texts=texts)
|
|
for degradation in result.degradations:
|
|
runtime.state["degradations"].append(degradation)
|
|
by_label: dict[str, int] = {}
|
|
for finding in result.findings:
|
|
by_label[finding.label] = by_label.get(finding.label, 0) + 1
|
|
payload = {
|
|
"sibling_sets": len(result.sibling_sets),
|
|
"under_populated": len(result.under_populated),
|
|
"operations": result.operations,
|
|
"findings_by_label": by_label,
|
|
"policy_version": result.policy_version,
|
|
"degradations": [dict(item) for item in result.degradations],
|
|
"sets": [{"module": group.key[0], "resource": group.key[1], "handlers": len(group.handlers)} for group in result.sibling_sets],
|
|
}
|
|
statics = findings_to_static(result)
|
|
cited: dict[str, dict[str, object]] = {}
|
|
for finding, static in zip(result.findings, statics, strict=True):
|
|
cited[static.finding_id] = {
|
|
"obligation": finding.operation.kind,
|
|
"resource": finding.operation.resource,
|
|
"missing": finding.missing,
|
|
"provenance": finding.provenance,
|
|
"label": finding.label,
|
|
"evidence": list(finding.evidence),
|
|
"known_fix": finding.known_fix,
|
|
"intent": finding.intent,
|
|
"intent_rationale": finding.intent_rationale,
|
|
}
|
|
return statics, payload, cited
|
|
|
|
|
|
def _first_source(root: Path) -> str:
|
|
from .preprocess import preprocess_repository
|
|
|
|
_, targets = preprocess_repository(root)
|
|
return targets[0].path if targets else "."
|
|
|
|
|
|
def _print_improve_outcomes(outcomes: list[RoundOutcome], manifest_path: Path, target: Path, ledger_path: Path, *, dry_run: bool) -> None:
|
|
print(f"manifest={manifest_path}")
|
|
print(f"target={target}")
|
|
accepted = [o for o in outcomes if o.accepted]
|
|
for outcome in outcomes:
|
|
edits = ", ".join(f"{e.rule_id}:{e.from_status}->{e.to_status}" for e in outcome.edits) or "-"
|
|
print(
|
|
f"round {outcome.round}: {outcome.reason} | "
|
|
f"recall {outcome.recall_before:.2%}->{outcome.recall_after:.2%} "
|
|
f"fp {outcome.fp_before:.2%}->{outcome.fp_after:.2%} "
|
|
f"score {outcome.score_before}->{outcome.score_after} | edits: {edits}"
|
|
)
|
|
if outcome.profile_regressions or outcome.profiles_under_minimum:
|
|
print(f" profiles: regressed={outcome.profile_regressions or '-'} under_minimum={outcome.profiles_under_minimum or '-'}")
|
|
print(f"rounds={len(outcomes)} accepted={len(accepted)}")
|
|
if dry_run:
|
|
print("dry_run=true (no ledger or mechanism store written)")
|
|
elif accepted:
|
|
print(f"ledger={ledger_path}")
|
|
else:
|
|
print("ledger=unchanged (no round accepted)")
|
|
|
|
|
|
def _load_scan_findings(run_dir: Path) -> list[StaticFinding]:
|
|
return load_findings(run_dir / "findings.json")
|
|
|
|
|
|
def _model_candidates(args: argparse.Namespace) -> int:
|
|
"""What the candidate enumerator can reach, per family and per slice. No model is called anywhere here.
|
|
|
|
The ceiling is a property of the corpus and the enumerator, so it can end a design before a cent is spent
|
|
on it — which is what it did for the pattern-ruleset generator (12.1% against the IR's 96.6%).
|
|
"""
|
|
from .model.candidates import ceiling
|
|
from .model.taxonomy import load_families
|
|
|
|
cases = select_vendored(select_slice(load_pair_catalog(args.catalog), args.slice))
|
|
coverage = ceiling(cases, taxonomy=load_families(), generator=args.generator)
|
|
out: Path = args.out
|
|
out.mkdir(parents=True, exist_ok=True)
|
|
payload = coverage.to_dict()
|
|
(out / "candidate-ceiling.json").write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
|
|
if args.json:
|
|
print(json.dumps(payload, indent=2, sort_keys=True))
|
|
else:
|
|
for family, block in sorted(coverage.per_family.items()):
|
|
ceiling_value = float(block["ceiling"]) if isinstance(block["ceiling"], int | float) else 0.0
|
|
print(
|
|
f"model candidates {family}: {block['with_candidate']}/{block['pairs']} "
|
|
f"= {ceiling_value:.1%} ceiling, sites {block['sites']}"
|
|
)
|
|
for slice_name, block in sorted(coverage.per_slice.items()):
|
|
ceiling_value = float(block["ceiling"]) if isinstance(block["ceiling"], int | float) else 0.0
|
|
print(f" slice {slice_name}: {ceiling_value:.1%} over {block['pairs']} pairs")
|
|
return 0
|
|
|
|
|
|
def _index(path: Path, config_path: Path, chunk_lines: int) -> int:
|
|
if not path.exists() or not path.is_dir():
|
|
raise SystemExit(f"index path is not a directory: {path}")
|
|
config = load_config(config_path if config_path.exists() else None)
|
|
run = create_scan_run(path, config)
|
|
snapshot, targets = preprocess_repository(run.target, run.root / "preprocess" / "file_targets.json")
|
|
chunks = build_code_chunks(run.target, targets, max_lines=chunk_lines)
|
|
payload = {
|
|
"store": config.embeddings.store or "json-local",
|
|
"embedding_model": config.embeddings.model,
|
|
"repo_root": snapshot.root,
|
|
"repo_commit": snapshot.commit,
|
|
"chunk_count": len(chunks),
|
|
"chunks": [chunk.__dict__ for chunk in chunks],
|
|
}
|
|
path_out = run.root / "index" / "chunks.json"
|
|
path_out.parent.mkdir(parents=True, exist_ok=True)
|
|
path_out.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
|
|
print(f"scan_id={run.scan_id}")
|
|
print(f"run_dir={run.root}")
|
|
print(f"chunks={len(chunks)}")
|
|
print(f"index_artifact={path_out}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|