mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 03:49:26 +00:00
Marius CiepluchandClaude Opus 5.5
1b28465a61
The 7.8 trace showed the completion loss was a sizer death spiral, not expensive families: after two kills the budget fell to 3 sink visits and never recovered, because portions finishing in 25-30 s sat under an assumed 70 s fixed start and read as "no information", so the budget could only shrink. - taint.sc streams each answer's own cost (`ms`); the backend carries it as a reserved `__timing__` list through shards, partitions and reuse. - The sizer budgets predicted seconds from per-family seconds-per-visit rates learned from those costs, and measures the fixed start as the wall time the answers do not explain (median), instead of assuming it. - A kill names its culprit: the first unanswered id in request order. The ids behind it never started and are re-queued at once; the culprit is asked alone after the rest. One that still fails alone after more than a portion's target is reported as `query_too_expensive`, apart from `query_failed`. - Portions grow id by id from the queue head, and a single-portion scan now goes through the queue too, so a kill there re-asks its remainder. - Fix: the reuse wrapper wrapped only `run_batch`, so since 7.8 every portioned query bypassed the graph-reuse cache and its source-root normalization. `run_batch_once` is wrapped too. Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]> Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo
667 lines
29 KiB
Python
667 lines
29 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from collections.abc import Mapping, Sequence
|
|
from dataclasses import asdict
|
|
from pathlib import Path
|
|
|
|
from .findings import StaticFinding
|
|
from .run import ScanRun
|
|
from .verification import VerificationResult
|
|
|
|
|
|
def write_markdown_report(
|
|
findings: list[StaticFinding],
|
|
path: Path,
|
|
verifications: list[VerificationResult] | None = None,
|
|
*,
|
|
redact: bool = True,
|
|
complexity_map: object | None = None,
|
|
verdicts: Sequence[object] | None = None,
|
|
overlay: Sequence[object] | None = None,
|
|
mechanisms: Mapping[str, Mapping[str, object]] | None = None,
|
|
obligations: Mapping[str, Mapping[str, object]] | None = None,
|
|
obligations_summary: Mapping[str, object] | None = None,
|
|
coverage: Sequence[Mapping[str, object]] | None = None,
|
|
degradations: Sequence[Mapping[str, object]] | None = None,
|
|
) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
verification_by_id = _verification_by_id(verifications or [])
|
|
overlay_by_id = _overlay_by_id(overlay or [])
|
|
cited: dict[str, Mapping[str, object]] = {}
|
|
cited_obligations: dict[str, Mapping[str, object]] = {}
|
|
# THE RULE (contributor-scan 2.11): a report is proportionate to what the engine can say.
|
|
#
|
|
# A finding whose only evidence is a text pattern -- `evidence_level == "static_corroboration"`, nothing
|
|
# corroborating it beyond the match -- is grouped by rule instead of given a section each. libpng emitted
|
|
# 198 sections of which 163 were exactly that, asserting `memcpy` and `strcpy` are PRESENT in a library
|
|
# that uses them correctly throughout. That is the noise developers have learned to skip in every other
|
|
# tool, and shipping it costs the findings beside it their credibility.
|
|
#
|
|
# Grouped, never dropped, and proportionate rather than hidden. A rule with a handful of sites keeps its
|
|
# full sections -- one `python-flask-debug` is a finding a contributor can act on and must not be
|
|
# summarised away. A rule with fifty-four is a property of the codebase. Anything a proposer actually
|
|
# reasoned about -- the model layer, the obligations checker -- keeps its section whatever its rung,
|
|
# because an undecided CLAIM about a site is not an observation that an API exists.
|
|
patterned = _crowded_pattern_rules(findings)
|
|
crowded = {id(item) for item in patterned}
|
|
reasoned = [item for item in findings if id(item) not in crowded]
|
|
|
|
lines = ["# OpenUltraSAST Report", "", f"Findings: {len(findings)}", "", "## Inventory", ""]
|
|
if not findings:
|
|
lines.append("No quick-mode findings were emitted.")
|
|
lines.append("")
|
|
for finding in reasoned:
|
|
verification = verification_by_id.get(finding.finding_id)
|
|
disposition = overlay_by_id.get(finding.finding_id)
|
|
lines.extend(
|
|
[
|
|
f"## {finding.title}",
|
|
"",
|
|
f"- ID: `{finding.finding_id}`",
|
|
f"- Path: `{finding.path}`",
|
|
f"- Line: `{finding.line}`",
|
|
f"- Function: `{finding.function_name or 'unknown'}`",
|
|
f"- Severity: `{finding.severity}`",
|
|
f"- Confidence: `{finding.confidence}`",
|
|
f"- Evidence level: `{finding.evidence_level}`",
|
|
f"- Rung: `{finding.rung}`",
|
|
f"- Verification: `{verification.status if verification else 'not_run'}`",
|
|
f"- Reachability: `{finding.reachability_status}`",
|
|
f"- Conditions: `{', '.join(finding.reachability_conditions) or 'none'}`",
|
|
f"- Ranking priority: `{finding.ranking_priority}`",
|
|
f"- Tags: `{', '.join(finding.tags)}`",
|
|
]
|
|
)
|
|
if disposition is not None:
|
|
lines.append(f"- Disposition: `{getattr(disposition, 'disposition', '')}`")
|
|
lines.append(f"- Overlay reason: `{getattr(disposition, 'reason', '')}`")
|
|
mechanism_id = _mechanism_id_for(finding, disposition)
|
|
if mechanism_id and mechanisms is not None and mechanism_id in mechanisms:
|
|
info = mechanisms[mechanism_id]
|
|
cited[mechanism_id] = info
|
|
lines.append(f"- Mechanism: `{mechanism_id}` — {info.get('summary', '')}")
|
|
lines.append(f"- Known fix guard: `{info.get('guard', 'none')}`")
|
|
lines.append(f"- Learned from pairs: `{', '.join(_pairs_of(info))}`")
|
|
info_o = (obligations or {}).get(finding.finding_id)
|
|
if info_o is not None:
|
|
cited_obligations[finding.finding_id] = info_o
|
|
_append_obligation_lines(lines, info_o, (mechanisms or {}).get(str(info_o.get("known_fix") or "")))
|
|
lines.extend(["", finding.rationale, ""])
|
|
if coverage or degradations:
|
|
_append_coverage(lines, coverage or (), degradations or ())
|
|
if patterned:
|
|
_append_pattern_matches(lines, patterned)
|
|
if cited:
|
|
_append_mechanisms(lines, cited)
|
|
if cited_obligations:
|
|
_append_obligations(lines, cited_obligations, obligations_summary)
|
|
if overlay:
|
|
_append_overlay(lines, overlay)
|
|
if complexity_map is not None:
|
|
_append_complexity_map(lines, complexity_map)
|
|
if verdicts is not None:
|
|
_append_worth_fixing(lines, verdicts)
|
|
report = "\n".join(lines).rstrip() + "\n"
|
|
if redact:
|
|
from .redaction import redact_secrets
|
|
|
|
report = redact_secrets(report)
|
|
path.write_text(report)
|
|
|
|
|
|
def _append_complexity_map(lines: list[str], complexity_map: object) -> None:
|
|
hotspots = tuple(getattr(complexity_map, "hotspots", ()) or ())
|
|
heuristic_only = getattr(complexity_map, "heuristic_only", None)
|
|
summary = f"Hotspots: {len(hotspots)}"
|
|
if heuristic_only is not None:
|
|
summary += f" (heuristic_only={str(bool(heuristic_only)).lower()})"
|
|
lines.extend(["## Complexity map", "", summary, ""])
|
|
if not hotspots:
|
|
lines.extend(["No complexity hotspots were emitted.", ""])
|
|
return
|
|
for hotspot in hotspots:
|
|
identity = _report_identity(getattr(hotspot, "path", ""), getattr(hotspot, "function_name", None))
|
|
lines.extend(
|
|
[
|
|
f"### `{identity}`",
|
|
"",
|
|
f"- Score: `{getattr(hotspot, 'score', '')}`",
|
|
f"- Band: `{getattr(hotspot, 'band', '')}`",
|
|
]
|
|
)
|
|
hint = getattr(hotspot, "test_hint", None)
|
|
if hint is not None:
|
|
lines.append(f"- Gap: `{getattr(hint, 'gap', '')}`")
|
|
kind = getattr(hint, "test_kind", None)
|
|
if kind:
|
|
lines.append(f"- Recommended test: `{kind}`")
|
|
reason = getattr(hint, "reason", "")
|
|
if reason:
|
|
lines.append(f"- Hint: {reason}")
|
|
inventory_ids = tuple(getattr(hotspot, "inventory_finding_ids", ()) or ())
|
|
if inventory_ids:
|
|
lines.append(f"- Inventory: `{', '.join(str(item) for item in inventory_ids)}`")
|
|
rationale = str(getattr(hotspot, "rationale", "") or "")
|
|
lines.extend(["", rationale, ""] if rationale else [""])
|
|
|
|
|
|
def _append_worth_fixing(lines: list[str], verdicts: Sequence[object]) -> None:
|
|
worth = [item for item in verdicts if getattr(item, "worth_fixing", False)]
|
|
lines.extend(["## Worth fixing", "", f"Count: {len(worth)}", ""])
|
|
if not worth:
|
|
lines.extend(["No worth-fixing verdicts were produced.", ""])
|
|
return
|
|
for item in worth:
|
|
identity = _report_identity(getattr(item, "path", ""), getattr(item, "function_name", None))
|
|
lines.extend(
|
|
[
|
|
f"### `{identity}`",
|
|
"",
|
|
f"- Verdict: `{getattr(item, 'verdict', '')}`",
|
|
f"- Reason: `{getattr(item, 'reason', '')}`",
|
|
"",
|
|
]
|
|
)
|
|
|
|
|
|
def _report_identity(path: object, function_name: object) -> str:
|
|
path_text = str(path or "")
|
|
if function_name:
|
|
return f"{path_text}::{function_name}"
|
|
return path_text
|
|
|
|
|
|
def write_sarif_report(
|
|
findings: list[StaticFinding],
|
|
verifications: list[VerificationResult],
|
|
path: Path,
|
|
overlay: Sequence[object] | None = None,
|
|
mechanisms: Mapping[str, Mapping[str, object]] | None = None,
|
|
obligations: Mapping[str, Mapping[str, object]] | None = None,
|
|
) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
verification_by_id = _verification_by_id(verifications)
|
|
overlay_by_id = _overlay_by_id(overlay or [])
|
|
rules = _sarif_rules(findings)
|
|
payload = {
|
|
"version": "2.1.0",
|
|
"$schema": "https://json.schemastore.org/sarif-2.1.0.json",
|
|
"runs": [
|
|
{
|
|
"tool": {
|
|
"driver": {
|
|
"name": "OpenUltraSAST",
|
|
"informationUri": "https://github.com/norandom/OpenUltraSAST",
|
|
"rules": list(rules.values()),
|
|
}
|
|
},
|
|
"results": [
|
|
_sarif_result(
|
|
finding,
|
|
verification_by_id.get(finding.finding_id),
|
|
overlay_by_id.get(finding.finding_id),
|
|
mechanisms=mechanisms,
|
|
obligation=(obligations or {}).get(finding.finding_id),
|
|
)
|
|
for finding in findings
|
|
],
|
|
}
|
|
],
|
|
}
|
|
path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
|
|
|
|
|
|
def write_manifest(
|
|
*,
|
|
run: ScanRun,
|
|
findings: list[StaticFinding],
|
|
verifications: list[VerificationResult],
|
|
artifact_paths: dict[str, Path],
|
|
path: Path,
|
|
score: dict[str, object] | None = None,
|
|
degradations: list[dict[str, object]] | None = None,
|
|
fusion: list[dict[str, object]] | None = None,
|
|
stages: dict[str, object] | None = None,
|
|
complexity: dict[str, object] | None = None,
|
|
worth_fixing: dict[str, object] | None = None,
|
|
provenance: dict[str, object] | None = None,
|
|
variants: dict[str, object] | None = None,
|
|
obligations: dict[str, object] | None = None,
|
|
model: dict[str, object] | None = None,
|
|
) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
verification_by_id = _verification_by_id(verifications)
|
|
payload: dict[str, object] = {
|
|
"scan_id": run.scan_id,
|
|
"target": str(run.target),
|
|
"artifacts": {name: _relative_artifact(run.root, artifact) for name, artifact in sorted(artifact_paths.items())},
|
|
"findings": [
|
|
{
|
|
"finding_id": finding.finding_id,
|
|
"path": finding.path,
|
|
"line": finding.line,
|
|
"severity": finding.severity,
|
|
"evidence_level": finding.evidence_level,
|
|
"rung": finding.rung,
|
|
"verification_status": verification_by_id[finding.finding_id].status
|
|
if finding.finding_id in verification_by_id
|
|
else "not_run",
|
|
"artifact_refs": _artifact_refs(run.root, artifact_paths),
|
|
}
|
|
for finding in findings
|
|
],
|
|
}
|
|
if score is not None:
|
|
payload["score"] = score
|
|
if degradations is not None:
|
|
payload["degradations"] = degradations
|
|
if fusion is not None:
|
|
payload["fusion"] = fusion
|
|
if stages is not None:
|
|
payload["stages"] = stages
|
|
if complexity is not None:
|
|
payload["complexity"] = complexity
|
|
if worth_fixing is not None:
|
|
payload["worth_fixing"] = worth_fixing
|
|
if provenance is not None:
|
|
payload["provenance"] = provenance
|
|
if variants is not None:
|
|
payload["variants"] = variants
|
|
if obligations is not None:
|
|
payload["obligations"] = obligations
|
|
if model is not None:
|
|
payload["model"] = model
|
|
path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
|
|
|
|
|
|
def scan_exit_code(
|
|
findings: list[StaticFinding],
|
|
verifications: list[VerificationResult],
|
|
policy: str,
|
|
worth_fixing_verdicts: Sequence[object] | None = None,
|
|
) -> int:
|
|
if policy == "never":
|
|
return 0
|
|
if policy == "findings":
|
|
return 1 if findings else 0
|
|
if policy == "verified":
|
|
return 1 if any(result.verified for result in verifications) else 0
|
|
if policy == "worth-fixing":
|
|
return 1 if any(getattr(verdict, "worth_fixing", False) for verdict in worth_fixing_verdicts or ()) else 0
|
|
raise ValueError(f"unknown fail policy: {policy}")
|
|
|
|
|
|
def _sarif_rules(findings: list[StaticFinding]) -> dict[str, dict[str, object]]:
|
|
rules: dict[str, dict[str, object]] = {}
|
|
for finding in findings:
|
|
rule_id = _rule_id(finding)
|
|
rules.setdefault(
|
|
rule_id,
|
|
{
|
|
"id": rule_id,
|
|
"name": finding.title,
|
|
"shortDescription": {"text": finding.title},
|
|
"properties": {"tags": finding.tags},
|
|
},
|
|
)
|
|
return rules
|
|
|
|
|
|
def _sarif_result(
|
|
finding: StaticFinding,
|
|
verification: VerificationResult | None,
|
|
overlay: object | None = None,
|
|
*,
|
|
mechanisms: Mapping[str, Mapping[str, object]] | None = None,
|
|
obligation: Mapping[str, object] | None = None,
|
|
) -> dict[str, object]:
|
|
properties: dict[str, object] = {
|
|
"finding_id": finding.finding_id,
|
|
"severity": finding.severity,
|
|
"confidence": finding.confidence,
|
|
"evidence_level": finding.evidence_level,
|
|
"rung": finding.rung,
|
|
"verification": asdict(verification) if verification else None,
|
|
"reachability_status": finding.reachability_status,
|
|
"reachability_conditions": finding.reachability_conditions,
|
|
"ranking_priority": finding.ranking_priority,
|
|
}
|
|
if overlay is not None:
|
|
properties["disposition"] = getattr(overlay, "disposition", "")
|
|
properties["overlay_reason"] = getattr(overlay, "reason", "")
|
|
mechanism_id = _mechanism_id_for(finding, overlay)
|
|
if mechanism_id:
|
|
properties["mechanism_id"] = mechanism_id
|
|
info = (mechanisms or {}).get(mechanism_id)
|
|
if info is not None:
|
|
properties["mechanism_summary"] = str(info.get("summary", ""))
|
|
properties["mechanism_guard"] = str(info.get("guard", "none"))
|
|
properties["mechanism_pairs"] = _pairs_of(info)
|
|
if obligation is not None:
|
|
properties["obligation_kind"] = str(obligation.get("obligation", ""))
|
|
properties["obligation_resource"] = obligation.get("resource")
|
|
properties["obligation_missing"] = str(obligation.get("missing", ""))
|
|
properties["obligation_provenance"] = obligation.get("provenance")
|
|
properties["obligation_label"] = str(obligation.get("label", ""))
|
|
properties["obligation_evidence"] = _strings_of(obligation, "evidence")
|
|
properties["obligation_known_fix"] = obligation.get("known_fix")
|
|
properties["obligation_intent"] = obligation.get("intent")
|
|
return {
|
|
"ruleId": _rule_id(finding),
|
|
"level": _sarif_level(finding.severity),
|
|
"message": {"text": finding.rationale},
|
|
"locations": [
|
|
{
|
|
"physicalLocation": {
|
|
"artifactLocation": {"uri": finding.path},
|
|
"region": {"startLine": finding.line or 1},
|
|
}
|
|
}
|
|
],
|
|
"partialFingerprints": {"openultrasastFindingId": finding.finding_id},
|
|
"properties": properties,
|
|
}
|
|
|
|
|
|
def _rule_id(finding: StaticFinding) -> str:
|
|
return finding.finding_id.split(":", 1)[0]
|
|
|
|
|
|
def _sarif_level(severity: str) -> str:
|
|
if severity in {"critical", "high"}:
|
|
return "error"
|
|
if severity == "medium":
|
|
return "warning"
|
|
return "note"
|
|
|
|
|
|
# How many sites a pattern rule may have before it is summarised instead of listed in full. A rule at or
|
|
# under this keeps the behaviour it always had; over it, the report stops repeating itself.
|
|
_PATTERN_DETAIL_LIMIT = 3
|
|
|
|
|
|
def _crowded_pattern_rules(findings: Sequence[StaticFinding]) -> list[StaticFinding]:
|
|
"""Pattern-only findings belonging to a rule that matched more sites than are worth listing."""
|
|
counts: dict[str, int] = {}
|
|
for finding in findings:
|
|
if _is_pattern_only(finding):
|
|
counts[_rule_key(finding)] = counts.get(_rule_key(finding), 0) + 1
|
|
return [finding for finding in findings if _is_pattern_only(finding) and counts.get(_rule_key(finding), 0) > _PATTERN_DETAIL_LIMIT]
|
|
|
|
|
|
def _append_coverage(lines: list[str], coverage: Sequence[Mapping[str, object]], degradations: Sequence[Mapping[str, object]] = ()) -> None:
|
|
"""What was looked for, and what could not be decided (Req 5.6).
|
|
|
|
A scan that reports nothing has established that nothing it MODELS was found. That is not the same as
|
|
finding nothing, and the difference is the whole reason this section exists: a silent report read as a
|
|
clean bill of health is worse than a noisy one, because the reader acts on it.
|
|
"""
|
|
lines.extend(
|
|
[
|
|
"## What was analysed",
|
|
"",
|
|
"Each family below was offered to at least one region. **A family that reported nothing has "
|
|
"established only that nothing it models was found** — silence here means no fact table covered "
|
|
"the code, not that the code is safe.",
|
|
"",
|
|
"| language | family | regions | operations modelled |",
|
|
"| --- | --- | --- | --- |",
|
|
]
|
|
)
|
|
for row in coverage:
|
|
lines.append(f"| {row.get('language')} | `{row.get('family')}` | {row.get('regions')} | {row.get('modelled')} |")
|
|
lines.append("")
|
|
|
|
stated = [row for row in coverage if str(row.get("limits") or "")]
|
|
if stated:
|
|
lines.extend(["### Stated limits", ""])
|
|
for row in stated:
|
|
lines.extend([f"**`{row.get('family')}` ({row.get('language')})** — {row.get('limits')}", ""])
|
|
|
|
_append_not_analysed(lines, degradations)
|
|
|
|
|
|
# A degradation is a fact about COVERAGE, not a warning to be filed away. Each of these says the scan
|
|
# answered a smaller question than the one it was asked, and a reader who is not told cannot know that the
|
|
# silence in the table above is narrower than it looks.
|
|
_COVERAGE_DEGRADATIONS: dict[str, str] = {
|
|
"cpg_empty": "The code graph came back with no methods in it, so **nothing above was analysed at all**.",
|
|
"cpg_build_failed": "The code graph could not be built, so nothing above was analysed. {detail}",
|
|
"cpg_sharded": (
|
|
"The graph had to be built in {shards} parts, because the frontend could not parse the whole tree at "
|
|
"once. Each part was queried, but **a flow whose source is in one part and whose sink is in another "
|
|
"is not visible to any question asked here**."
|
|
),
|
|
"files_unparsed": "{count} file(s) could not be parsed and are absent from the graph: {files}.",
|
|
"query_failed": "A `{kind}` query failed for {requests} region(s), so those regions were not decided.",
|
|
"query_too_expensive": (
|
|
"{requests} `{kind}` question(s) could not be answered even when asked alone, spending up to "
|
|
"{max_seconds} s each, so those regions were not decided. This is the cost of the question, not a transient failure."
|
|
),
|
|
"budget_exhausted": "The model-call budget ran out with {regions_unasked} region(s) still unasked.",
|
|
"regions_truncated": (
|
|
"Only the **{examined} highest-ranked of {total} regions** were examined. Nothing above says anything "
|
|
"about the rest of the repository."
|
|
),
|
|
"region_failed": "The region `{path}` could not be arbitrated.",
|
|
}
|
|
|
|
|
|
def _append_not_analysed(lines: list[str], degradations: Sequence[Mapping[str, object]]) -> None:
|
|
"""The part of the repository the scan did NOT get to, said plainly.
|
|
|
|
Silence in the coverage table means "no fact table covered this". Silence here would mean something
|
|
else entirely -- that the graph was empty, or split, or that the query never ran -- and the reader has
|
|
no way to tell those apart from the findings alone.
|
|
"""
|
|
rendered: list[str] = []
|
|
for item in degradations:
|
|
template = _COVERAGE_DEGRADATIONS.get(str(item.get("reason") or ""))
|
|
if template is None:
|
|
continue
|
|
fields = dict(item)
|
|
files = fields.get("files")
|
|
if isinstance(files, (list, tuple)):
|
|
fields["files"] = ", ".join(f"`{name}`" for name in files)
|
|
try:
|
|
rendered.append(template.format(**fields))
|
|
except (KeyError, IndexError): # a degradation shaped differently than its template expects
|
|
rendered.append(f"`{item.get('reason')}`")
|
|
if not rendered:
|
|
return
|
|
lines.extend(["### What could not be analysed", ""])
|
|
lines.extend(f"- {line}" for line in rendered)
|
|
lines.append("")
|
|
|
|
|
|
def _is_pattern_only(finding: StaticFinding) -> bool:
|
|
"""Was this proposed with nothing corroborating it?
|
|
|
|
Two shapes qualify, and the second took a second look to see.
|
|
|
|
A **static pattern match** (`static_corroboration` at `suspicion`) says an API is present.
|
|
|
|
A **model suspicion** (`model:` at `suspicion`) is what the enumerator proposed at a site the graph could
|
|
NOT decide, affirmed by a judge that has no coverage there -- a pattern match with an opinion attached.
|
|
Its rationale says so in its own words: "the model resolved no flow here; its sink table may not cover
|
|
this API". It reads like a reasoned claim because a model produced it, which is exactly why it needs
|
|
naming: on vampi it is 38 of 60 findings, roughly one per fourteen lines of a 520-line application.
|
|
|
|
What does NOT qualify is the obligations checker, which also sits at `suspicion`. Its output comes from a
|
|
structural analysis -- an obligated operation with no discharging guard -- rather than from a guess, and
|
|
collapsing it would hide a detector.
|
|
"""
|
|
if str(finding.rung) != "suspicion":
|
|
return False
|
|
if str(finding.evidence_level) == "static_corroboration":
|
|
return True
|
|
return str(finding.finding_id).startswith("model:")
|
|
|
|
|
|
def _rule_key(finding: StaticFinding) -> str:
|
|
"""How a proposal is grouped. Model findings key on their FAMILY, since their rule is always `model`."""
|
|
parts = str(finding.finding_id).split(":")
|
|
if parts and parts[0] == "model" and len(parts) > 1:
|
|
return f"model:{parts[1]}"
|
|
return parts[0] if parts else ""
|
|
|
|
|
|
def _append_pattern_matches(lines: list[str], patterned: Sequence[StaticFinding]) -> None:
|
|
"""Pattern matches, grouped by rule. Proportionate: few sites are named, many are counted."""
|
|
by_rule: dict[str, list[StaticFinding]] = {}
|
|
for finding in patterned:
|
|
by_rule.setdefault(_rule_key(finding), []).append(finding)
|
|
|
|
lines.extend(
|
|
[
|
|
"## Proposed, not corroborated",
|
|
"",
|
|
f"{len(patterned)} site(s) were proposed with nothing corroborating them: a static pattern saying "
|
|
"an API is present, or a candidate the model could not decide and a judge affirmed without "
|
|
"coverage there. Neither says the site is misused. Silence from the model means no fact table "
|
|
"covered it rather than that it is safe, and all of them are in `findings.json`.",
|
|
"",
|
|
]
|
|
)
|
|
for rule, items in sorted(by_rule.items(), key=lambda entry: (-len(entry[1]), entry[0])):
|
|
by_file: dict[str, int] = {}
|
|
for finding in items:
|
|
by_file[str(finding.path)] = by_file.get(str(finding.path), 0) + 1
|
|
top = sorted(by_file.items(), key=lambda entry: (-entry[1], entry[0]))[:3]
|
|
where = ", ".join(f"`{name}` ({count})" for name, count in top)
|
|
more = f", and {len(by_file) - len(top)} more file(s)" if len(by_file) > len(top) else ""
|
|
lines.append(f"- `{rule}` — {len(items)} site(s) across {len(by_file)} file(s): {where}{more}")
|
|
lines.append("")
|
|
|
|
|
|
def _verification_by_id(verifications: list[VerificationResult]) -> dict[str, VerificationResult]:
|
|
return {result.finding_id: result for result in verifications}
|
|
|
|
|
|
def _overlay_by_id(records: Sequence[object]) -> dict[str, object]:
|
|
mapped: dict[str, object] = {}
|
|
for record in records:
|
|
proposal_id = getattr(record, "proposal_id", None)
|
|
if isinstance(proposal_id, str) and proposal_id:
|
|
mapped[proposal_id] = record
|
|
return mapped
|
|
|
|
|
|
def _append_overlay(lines: list[str], records: Sequence[object]) -> None:
|
|
lines.extend(["## Overlay", "", f"Records: {len(records)}", ""])
|
|
labels = ("promoted", "demoted", "unadjudicated", "coverage")
|
|
counts = {label: 0 for label in labels}
|
|
for record in records:
|
|
disposition = str(getattr(record, "disposition", ""))
|
|
if disposition == "promote":
|
|
counts["promoted"] += 1
|
|
elif disposition == "demote":
|
|
counts["demoted"] += 1
|
|
elif disposition == "unadjudicated":
|
|
counts["unadjudicated"] += 1
|
|
elif disposition == "coverage":
|
|
counts["coverage"] += 1
|
|
lines.append("- Labels: " + ", ".join(f"{label}={counts[label]}" for label in labels))
|
|
lines.append("")
|
|
for record in records:
|
|
disposition = str(getattr(record, "disposition", ""))
|
|
proposal_id = str(getattr(record, "proposal_id", ""))
|
|
reason = str(getattr(record, "reason", ""))
|
|
lines.extend(
|
|
[
|
|
f"### `{disposition}` `{proposal_id}`",
|
|
"",
|
|
f"- Path: `{getattr(record, 'path', '')}`",
|
|
f"- Line: `{getattr(record, 'line', None)}`",
|
|
f"- Reason: `{reason}`",
|
|
"",
|
|
]
|
|
)
|
|
|
|
|
|
def _artifact_refs(root: Path, artifact_paths: dict[str, Path]) -> dict[str, str]:
|
|
refs = {
|
|
"findings_json": _relative_artifact(root, artifact_paths["findings"]),
|
|
"verification_json": _relative_artifact(root, artifact_paths["verification"]),
|
|
"markdown": _relative_artifact(root, artifact_paths["markdown"]),
|
|
"sarif": _relative_artifact(root, artifact_paths["sarif"]),
|
|
}
|
|
if "trajectories" in artifact_paths:
|
|
refs["trajectories_jsonl"] = _relative_artifact(root, artifact_paths["trajectories"])
|
|
return refs
|
|
|
|
|
|
def _relative_artifact(root: Path, path: Path) -> str:
|
|
try:
|
|
return path.relative_to(root).as_posix()
|
|
except ValueError:
|
|
return path.as_posix()
|
|
|
|
|
|
def _mechanism_id_for(finding: StaticFinding, overlay: object | None) -> str | None:
|
|
"""Mechanism cited by a finding: a `mechanism:<id>` tag on a variant finding, or the merged overlay record's id."""
|
|
for tag in finding.tags:
|
|
if tag.startswith("mechanism:"):
|
|
return tag.split(":", 1)[1]
|
|
merged = getattr(overlay, "mechanism_id", None)
|
|
return str(merged) if merged else None
|
|
|
|
|
|
def _append_mechanisms(lines: list[str], cited: Mapping[str, Mapping[str, object]]) -> None:
|
|
lines.extend(
|
|
["## Mechanisms", "", "Known mechanisms this report cites (corpus-seeded; variant findings stay suspicions until verified).", ""]
|
|
)
|
|
for mechanism_id, info in sorted(cited.items()):
|
|
pairs = ", ".join(_pairs_of(info)) or "unknown"
|
|
lines.append(
|
|
f"- `{mechanism_id}`: {info.get('summary', '')}; known fix guard `{info.get('guard', 'none')}`; learned from `{pairs}`"
|
|
)
|
|
lines.append("")
|
|
|
|
|
|
def _pairs_of(info: Mapping[str, object]) -> list[str]:
|
|
pairs = info.get("pairs")
|
|
return [str(p) for p in pairs] if isinstance(pairs, list | tuple) else []
|
|
|
|
|
|
def _append_obligation_lines(lines: list[str], info: Mapping[str, object], mechanism: Mapping[str, object] | None) -> None:
|
|
resource = info.get("resource") or "a resource"
|
|
provenance = f" (`{info.get('provenance')}`)" if info.get("provenance") else ""
|
|
evidence = ", ".join(_strings_of(info, "evidence")) or "the operation fact alone"
|
|
lines.append(f"- Obligation: `{info.get('obligation')}` on `{resource}`")
|
|
lines.append(f"- Missing discharger: `{info.get('missing')}`{provenance}")
|
|
lines.append(f"- Evidence: `{info.get('label')}` — {evidence}")
|
|
if info.get("known_fix"):
|
|
guard = mechanism.get("guard") if mechanism else info.get("missing")
|
|
pairs = ", ".join(_pairs_of(mechanism)) if mechanism else "the mechanism store"
|
|
lines.append(f"- Known fix: `{guard}` learned from `{pairs}` ({info.get('known_fix')})")
|
|
if info.get("intent"):
|
|
lines.append(f"- Intent: `{info.get('intent')}`")
|
|
|
|
|
|
def _append_obligations(lines: list[str], cited: Mapping[str, Mapping[str, object]], summary: Mapping[str, object] | None = None) -> None:
|
|
lines.extend(
|
|
["## Obligations", "", "Operations reached without a dominating discharger; suspicions labeled by why the obligation exists.", ""]
|
|
)
|
|
if summary is not None:
|
|
version = summary.get("policy_version")
|
|
lines.append(f"Sibling sets evaluated: {summary.get('sibling_sets', 0)} (under-populated: {summary.get('under_populated', 0)})")
|
|
lines.append(f"Policy version: {f'`{version}`' if version else 'none'}")
|
|
lines.append("")
|
|
sets = summary.get("sets")
|
|
if isinstance(sets, list | tuple):
|
|
for group in sets:
|
|
if isinstance(group, Mapping):
|
|
lines.append(f"- `{group.get('module')}` / `{group.get('resource')}`: {group.get('handlers')} handlers")
|
|
lines.append("")
|
|
for finding_id, info in sorted(cited.items()):
|
|
resource = info.get("resource") or "a resource"
|
|
lines.append(f"- `{finding_id}`: `{info.get('obligation')}` on `{resource}` without `{info.get('missing')}` ({info.get('label')})")
|
|
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 []
|