diff --git a/src/metis/cli/utils.py b/src/metis/cli/utils.py index 04fdf178..55c46cd0 100644 --- a/src/metis/cli/utils.py +++ b/src/metis/cli/utils.py @@ -32,7 +32,7 @@ ) from .exporters import export_csv, export_html, export_sarif -from metis.sarif.utils import create_fingerprint +from metis.sarif.triage import apply_triage_annotations from metis.vector_store.retrievers import retriever_query_config try: @@ -377,7 +377,7 @@ def save_output(output_files, data, quiet=False, sarif_payload=None): else: files = list(output_files) json_payload = ( - _merge_triage_annotations(data, sarif_payload) + apply_triage_annotations(data, sarif_payload) if sarif_payload is not None else data ) @@ -431,198 +431,6 @@ def _write_payload(path: Path, payload: object, label: str) -> None: _write_payload(output_path, json_payload, "Results") -def _merge_triage_annotations(report_data, sarif_payload): - if not isinstance(report_data, dict): - return report_data - reviews = report_data.get("reviews") - if not isinstance(reviews, list): - return report_data - runs = sarif_payload.get("runs") if isinstance(sarif_payload, dict) else None - if not isinstance(runs, list): - return report_data - - sarif_results = [] - for run in runs: - if not isinstance(run, dict): - continue - results = run.get("results") - if not isinstance(results, list): - continue - sarif_results.extend(results) - - issue_refs = [] - for file_entry in reviews: - if not isinstance(file_entry, dict): - continue - file_name = str(file_entry.get("file") or file_entry.get("file_path") or "") - issues = file_entry.get("reviews") - if not isinstance(issues, list): - continue - for issue in issues: - if isinstance(issue, dict): - issue_refs.append((issue, file_name)) - - if not sarif_results or not issue_refs: - return report_data - - indexed = [] - fp_map: dict[str, list[int]] = {} - file_line_issue_map: dict[tuple[str, int, str], list[int]] = {} - file_line_map: dict[tuple[str, int], list[int]] = {} - file_issue_map: dict[tuple[str, str], list[int]] = {} - file_line_rule_map: dict[tuple[str, int, str], list[int]] = {} - file_rule_issue_map: dict[tuple[str, str, str], list[int]] = {} - - for idx, result in enumerate(sarif_results): - if not isinstance(result, dict): - continue - properties = result.get("properties") - if not isinstance(properties, dict): - continue - - file_name, line_number = _extract_sarif_location(result) - issue_text = _extract_sarif_issue_text(result) - rule_id = _extract_sarif_rule_id(result) - fingerprint = _extract_sarif_fingerprint(result) - indexed.append((idx, properties)) - if fingerprint: - fp_map.setdefault(fingerprint, []).append(idx) - if file_name and line_number > 0 and rule_id: - file_line_rule_map.setdefault((file_name, line_number, rule_id), []).append( - idx - ) - if file_name and line_number > 0 and issue_text: - file_line_issue_map.setdefault( - (file_name, line_number, issue_text), [] - ).append(idx) - if file_name and rule_id and issue_text: - file_rule_issue_map.setdefault((file_name, rule_id, issue_text), []).append( - idx - ) - if file_name and line_number > 0: - file_line_map.setdefault((file_name, line_number), []).append(idx) - if file_name and issue_text: - file_issue_map.setdefault((file_name, issue_text), []).append(idx) - - unused = {idx for idx, _ in indexed} - - def _take_from(mapping, key): - entries = mapping.get(key) - if not entries: - return None - while entries: - candidate = entries.pop(0) - if candidate in unused: - return candidate - return None - - properties_by_idx = {idx: props for idx, props in indexed} - - for issue, file_name in issue_refs: - line_number = _normalize_issue_line(issue.get("line_number")) - issue_text = str(issue.get("issue") or issue.get("title") or "").strip() - issue_rule = str(issue.get("rule_id") or issue.get("ruleId") or "").strip() - fingerprint = "" - if file_name and line_number > 0: - fingerprint = create_fingerprint(file_name, line_number, "AI001") - - matchers = [] - if fingerprint: - matchers.append((fp_map, fingerprint)) - if file_name and line_number > 0 and issue_rule: - matchers.append((file_line_rule_map, (file_name, line_number, issue_rule))) - if file_name and line_number > 0 and issue_text: - matchers.append((file_line_issue_map, (file_name, line_number, issue_text))) - if file_name and issue_rule and issue_text: - matchers.append((file_rule_issue_map, (file_name, issue_rule, issue_text))) - if file_name and line_number > 0: - matchers.append((file_line_map, (file_name, line_number))) - if file_name and issue_text: - matchers.append((file_issue_map, (file_name, issue_text))) - - match_idx = None - for mapping, key in matchers: - match_idx = _take_from(mapping, key) - if match_idx is not None: - break - if match_idx is None: - continue - - unused.discard(match_idx) - properties = properties_by_idx.get(match_idx) - if not properties: - continue - _apply_triage_properties(issue, properties) - - return report_data - - -def _normalize_issue_line(raw_line) -> int: - try: - parsed = int(raw_line) - except Exception: - return 1 - return parsed if parsed > 0 else 1 - - -def _extract_sarif_fingerprint(result: dict) -> str: - partial = result.get("partialFingerprints") - if not isinstance(partial, dict): - return "" - return str(partial.get("primaryLocationLineHash") or "").strip() - - -def _extract_sarif_location(result: dict) -> tuple[str, int]: - locations = result.get("locations") - if not isinstance(locations, list) or not locations: - return "", 1 - first = locations[0] - if not isinstance(first, dict): - return "", 1 - physical = first.get("physicalLocation") - if not isinstance(physical, dict): - return "", 1 - artifact = physical.get("artifactLocation") - file_name = "" - if isinstance(artifact, dict): - file_name = str(artifact.get("uri") or "") - region = physical.get("region") - properties = result.get("properties") - if isinstance(properties, dict): - reported_line = properties.get("reportedLineNumber") - if reported_line is not None: - return file_name, _normalize_issue_line(reported_line) - if not isinstance(region, dict): - return file_name, 1 - return file_name, _normalize_issue_line(region.get("startLine")) - - -def _extract_sarif_rule_id(result: dict) -> str: - return str(result.get("ruleId") or "").strip() - - -def _extract_sarif_issue_text(result: dict) -> str: - message = result.get("message") - if isinstance(message, dict): - return str(message.get("text") or "").strip() - if isinstance(message, str): - return message.strip() - return "" - - -def _apply_triage_properties(issue: dict, properties: dict) -> None: - if "metisTriaged" in properties: - issue["metisTriaged"] = bool(properties.get("metisTriaged")) - if "metisTriageStatus" in properties: - issue["metisTriageStatus"] = str(properties.get("metisTriageStatus") or "") - if "metisTriageReason" in properties: - issue["metisTriageReason"] = str(properties.get("metisTriageReason") or "") - if "metisTriageTimestamp" in properties: - issue["metisTriageTimestamp"] = str( - properties.get("metisTriageTimestamp") or "" - ) - - def check_file_exists(file_path, quiet=False): if not Path(file_path).is_file(): print_console(f"[red]File not found:[/red] {escape(file_path)}", quiet) diff --git a/src/metis/engine/nodes/finding_dedup/core.py b/src/metis/engine/nodes/finding_dedup/core.py index 322cb1b6..240e69a7 100644 --- a/src/metis/engine/nodes/finding_dedup/core.py +++ b/src/metis/engine/nodes/finding_dedup/core.py @@ -83,7 +83,7 @@ def _unique_candidates(candidates: list[ReviewCandidate]) -> list[ReviewCandidat seen: set[tuple[str, str]] = set() for candidate in candidates: group = candidate.group.model_dump(mode="json", exclude={"reviews"}) - finding = candidate.finding.model_dump(mode="json") + finding = candidate.finding.model_dump(mode="json", exclude={"id"}) identity = ( json.dumps(group, sort_keys=True, separators=(",", ":")), json.dumps(finding, sort_keys=True, separators=(",", ":")), diff --git a/src/metis/engine/nodes/reachability/finding_adapter.py b/src/metis/engine/nodes/reachability/finding_adapter.py index 5e072f7d..a860fefe 100644 --- a/src/metis/engine/nodes/reachability/finding_adapter.py +++ b/src/metis/engine/nodes/reachability/finding_adapter.py @@ -37,6 +37,7 @@ def finding_to_review_item( primary_file = finding.primary_file or finding.sink_file or finding.source_file smap = SourceMap.for_file(codebase_path, primary_file) if primary_file else None item = { + "id": finding.id, "issue": issue, "line_number": line_number, "anchor": dict(finding.primary_anchor) if finding.primary_anchor else None, diff --git a/src/metis/engine/stages/review/models.py b/src/metis/engine/stages/review/models.py index 47331627..05451983 100644 --- a/src/metis/engine/stages/review/models.py +++ b/src/metis/engine/stages/review/models.py @@ -9,6 +9,7 @@ from typing import Literal from typing import NotRequired from typing import Required +from uuid import uuid4 from pydantic import BaseModel from pydantic import ConfigDict @@ -60,6 +61,7 @@ def validate_target(self) -> ReviewCommand: class ReviewFinding(BaseModel): + id: str = Field(default_factory=lambda: uuid4().hex, min_length=1) issue: str = Field(min_length=1) model_config = ConfigDict(extra="allow", frozen=True) diff --git a/src/metis/sarif/triage.py b/src/metis/sarif/triage.py index 7c246f2a..c4b00f30 100644 --- a/src/metis/sarif/triage.py +++ b/src/metis/sarif/triage.py @@ -12,6 +12,7 @@ from metis.json_io import write_json_atomic METIS_TRIAGED_KEY = "metisTriaged" +METIS_FINDING_ID_KEY = "metisFindingId" METIS_TRIAGE_STATUS_KEY = "metisTriageStatus" METIS_TRIAGE_REASON_KEY = "metisTriageReason" METIS_TRIAGE_TIMESTAMP_KEY = "metisTriageTimestamp" @@ -51,6 +52,57 @@ def save_sarif_file(path: str | Path, payload: dict[str, Any]) -> None: write_json_atomic(p, payload, indent=4) +def apply_triage_annotations(report_data: Any, sarif_payload: Any) -> Any: + if not isinstance(report_data, dict) or not isinstance(sarif_payload, dict): + return report_data + reviews = report_data.get("reviews") + runs = sarif_payload.get("runs") + if not isinstance(reviews, list) or not isinstance(runs, list): + return report_data + + properties_by_id: dict[str, dict[str, Any]] = {} + ambiguous_ids: set[str] = set() + for run in runs: + for result in run.get("results", ()) if isinstance(run, dict) else (): + properties = result.get("properties") if isinstance(result, dict) else None + if not isinstance(properties, dict): + continue + finding_id = str(properties.get(METIS_FINDING_ID_KEY) or "").strip() + if not finding_id: + continue + if finding_id in properties_by_id: + ambiguous_ids.add(finding_id) + else: + properties_by_id[finding_id] = properties + for finding_id in ambiguous_ids: + properties_by_id.pop(finding_id, None) + + issues_by_id: dict[str, dict[str, Any]] = {} + ambiguous_report_ids: set[str] = set() + for review in reviews: + issues = review.get("reviews") if isinstance(review, dict) else None + if not isinstance(issues, list): + continue + for issue in issues: + if not isinstance(issue, dict): + continue + finding_id = str(issue.get("id") or "").strip() + if not finding_id: + continue + if finding_id in issues_by_id: + ambiguous_report_ids.add(finding_id) + else: + issues_by_id[finding_id] = issue + for finding_id in ambiguous_report_ids: + issues_by_id.pop(finding_id, None) + + for finding_id, issue in issues_by_id.items(): + properties = properties_by_id.get(finding_id) + if properties is not None: + _apply_triage_properties(issue, properties) + return report_data + + def extract_findings( payload: dict[str, Any], *, include_triaged: bool = False ) -> list[SarifFinding]: @@ -241,3 +293,15 @@ def _apply_triage_metadata( properties[METIS_THREAT_MODEL_POLICY_KEY] = dict(threat_model_policy) else: properties.pop(METIS_THREAT_MODEL_POLICY_KEY, None) + + +def _apply_triage_properties(issue: dict[str, Any], properties: dict[str, Any]) -> None: + if METIS_TRIAGED_KEY in properties: + issue[METIS_TRIAGED_KEY] = bool(properties.get(METIS_TRIAGED_KEY)) + for key in ( + METIS_TRIAGE_STATUS_KEY, + METIS_TRIAGE_REASON_KEY, + METIS_TRIAGE_TIMESTAMP_KEY, + ): + if key in properties: + issue[key] = str(properties.get(key) or "") diff --git a/src/metis/sarif/writer.py b/src/metis/sarif/writer.py index a5a14d8c..650d8614 100644 --- a/src/metis/sarif/writer.py +++ b/src/metis/sarif/writer.py @@ -1,6 +1,9 @@ # SPDX-FileCopyrightText: Copyright 2025 Arm Limited and/or its affiliates # SPDX-License-Identifier: Apache-2.0 +from uuid import uuid4 + +from metis.sarif.triage import METIS_FINDING_ID_KEY from metis.version import __version__ as TOOL_VERSION from metis.sarif.utils import anchor_fingerprint, create_fingerprint, read_file_lines @@ -107,6 +110,7 @@ def generate_sarif( run = sarif["runs"][0] + finding_ids: set[str] = set() for review in results.get("reviews", []): file_path = review.get("file_path") artifact_uri = review.get("file") or file_path or "" @@ -115,6 +119,11 @@ def generate_sarif( total_lines = len(lines) if source_available else 0 for issue in review.get("reviews", []): + finding_id = str(issue.get("id") or "").strip() + if not finding_id or finding_id in finding_ids: + finding_id = uuid4().hex + issue["id"] = finding_id + finding_ids.add(finding_id) text = issue.get("issue", "unspecified") anchor = ( issue.get("anchor") if isinstance(issue.get("anchor"), dict) else None @@ -167,7 +176,7 @@ def generate_sarif( end = line_num + snippet_line_count - 1 context = snippet_text or "" - properties = {} + properties = {METIS_FINDING_ID_KEY: finding_id} cwe_id = issue.get("cwe") if isinstance(cwe_id, str) and cwe_id.strip(): properties["cwe"] = cwe_id.strip() diff --git a/tests/test_cli_triage.py b/tests/test_cli_triage.py index 610fc761..d96a039d 100644 --- a/tests/test_cli_triage.py +++ b/tests/test_cli_triage.py @@ -11,7 +11,6 @@ from metis.cli.command_runtime import CommandRuntime from metis.cli.commands import run_triage from metis.cli.utils import save_output -from metis.sarif.utils import create_fingerprint @pytest.mark.parametrize( @@ -159,12 +158,14 @@ def test_run_triage_accepts_metis_json_input(tmp_path): class _DummyEngine: def execute_triage(self, payload, **kwargs): assert kwargs.get("options") is None - payload["runs"][0]["results"][0]["properties"] = { - "metisTriaged": True, - "metisTriageStatus": "invalid", - "metisTriageReason": "Contradicted by source.", - "metisTriageTimestamp": "2026-01-01T00:00:00Z", - } + payload["runs"][0]["results"][0]["properties"].update( + { + "metisTriaged": True, + "metisTriageStatus": "invalid", + "metisTriageReason": "Contradicted by source.", + "metisTriageTimestamp": "2026-01-01T00:00:00Z", + } + ) return {"formats": ("json",), "sarif": payload} args = SimpleNamespace(quiet=True, output_file=None, include_triaged=False) @@ -235,15 +236,15 @@ def test_run_triage_rejects_non_metis_json_input(tmp_path, monkeypatch): assert any("Metis results object" in message for message in messages) -def test_save_output_json_includes_triage_annotations(tmp_path): +def test_save_output_json_matches_triage_annotations_by_identity(tmp_path): output_path = tmp_path / "results.json" results = { "reviews": [ { "file": "src/a.c", "reviews": [ - {"issue": "Issue A", "line_number": 10}, - {"issue": "Issue B", "line_number": 20}, + {"id": "finding-a", "issue": "Issue A", "line_number": 10}, + {"id": "finding-b", "issue": "Issue B", "line_number": 10}, ], } ] @@ -254,49 +255,21 @@ def test_save_output_json_includes_triage_annotations(tmp_path): { "results": [ { - "ruleId": "AI001", - "message": {"text": "Issue A"}, - "partialFingerprints": { - "primaryLocationLineHash": create_fingerprint( - "src/a.c", 10, "AI001" - ) - }, - "locations": [ - { - "physicalLocation": { - "artifactLocation": {"uri": "src/a.c"}, - "region": {"startLine": 10}, - } - } - ], "properties": { + "metisFindingId": "finding-b", "metisTriaged": True, - "metisTriageStatus": "valid", - "metisTriageReason": "Concrete evidence found.", - "metisTriageTimestamp": "2026-01-01T00:00:00Z", + "metisTriageStatus": "invalid", + "metisTriageReason": "Contradicted by code.", + "metisTriageTimestamp": "2026-01-01T00:00:02Z", }, }, { - "ruleId": "AI001", - "message": {"text": "Issue B"}, - "partialFingerprints": { - "primaryLocationLineHash": create_fingerprint( - "src/a.c", 20, "AI001" - ) - }, - "locations": [ - { - "physicalLocation": { - "artifactLocation": {"uri": "src/a.c"}, - "region": {"startLine": 20}, - } - } - ], "properties": { + "metisFindingId": "finding-a", "metisTriaged": True, - "metisTriageStatus": "inconclusive", - "metisTriageReason": "Alias chain unresolved.", - "metisTriageTimestamp": "2026-01-01T00:00:01Z", + "metisTriageStatus": "valid", + "metisTriageReason": "Concrete evidence found.", + "metisTriageTimestamp": "2026-01-01T00:00:03Z", }, }, ] @@ -308,78 +281,55 @@ def test_save_output_json_includes_triage_annotations(tmp_path): payload = json.loads(output_path.read_text(encoding="utf-8")) issues = payload["reviews"][0]["reviews"] - assert issues[0]["metisTriaged"] is True + assert issues[0]["issue"] == "Issue A" assert issues[0]["metisTriageStatus"] == "valid" - assert issues[0]["metisTriageReason"] == "Concrete evidence found." - assert issues[1]["metisTriageStatus"] == "inconclusive" + assert issues[1]["issue"] == "Issue B" + assert issues[1]["metisTriageStatus"] == "invalid" -def test_save_output_json_matches_triage_annotations_by_identity(tmp_path): +@pytest.mark.parametrize( + ("report_ids", "sarif_ids"), + [ + (("duplicate", "duplicate"), ("duplicate",)), + (("duplicate",), ("duplicate", "duplicate")), + ], +) +def test_save_output_ignores_ambiguous_finding_ids( + tmp_path: Path, + report_ids: tuple[str, ...], + sarif_ids: tuple[str, ...], +) -> None: output_path = tmp_path / "results.json" results = { "reviews": [ { - "file": "src/a.c", "reviews": [ - {"issue": "Issue A", "line_number": 10}, - {"issue": "Issue B", "line_number": 20}, - ], + {"id": finding_id, "issue": f"Issue {index}"} + for index, finding_id in enumerate(report_ids) + ] } ] } - # Intentionally reverse SARIF result order; mapping should still attach by identity. triaged_sarif = { - "version": "2.1.0", "runs": [ { "results": [ { - "message": {"text": "Issue B"}, - "locations": [ - { - "physicalLocation": { - "artifactLocation": {"uri": "src/a.c"}, - "region": {"startLine": 20}, - } - } - ], "properties": { + "metisFindingId": finding_id, "metisTriaged": True, - "metisTriageStatus": "invalid", - "metisTriageReason": "Contradicted by code.", - "metisTriageTimestamp": "2026-01-01T00:00:02Z", - }, - }, - { - "message": {"text": "Issue A"}, - "locations": [ - { - "physicalLocation": { - "artifactLocation": {"uri": "src/a.c"}, - "region": {"startLine": 10}, - } - } - ], - "properties": { - "metisTriaged": True, - "metisTriageStatus": "valid", - "metisTriageReason": "Concrete evidence found.", - "metisTriageTimestamp": "2026-01-01T00:00:03Z", - }, - }, + } + } + for finding_id in sarif_ids ] } - ], + ] } - save_output(str(output_path), results, quiet=True, sarif_payload=triaged_sarif) + save_output(output_path, results, quiet=True, sarif_payload=triaged_sarif) - payload = json.loads(output_path.read_text(encoding="utf-8")) - issues = payload["reviews"][0]["reviews"] - assert issues[0]["issue"] == "Issue A" - assert issues[0]["metisTriageStatus"] == "valid" - assert issues[1]["issue"] == "Issue B" - assert issues[1]["metisTriageStatus"] == "invalid" + saved = json.loads(output_path.read_text(encoding="utf-8")) + assert all("metisTriaged" not in issue for issue in saved["reviews"][0]["reviews"]) def test_triage_debug_callback_enabled_without_verbose(): diff --git a/tests/test_engine_core.py b/tests/test_engine_core.py index 3d333be3..778e17f6 100644 --- a/tests/test_engine_core.py +++ b/tests/test_engine_core.py @@ -373,7 +373,11 @@ def run_review(_service, command, **_kwargs): outputs = engine.execute_review(mode, target=target) - assert outputs["findings"]["reviews"] == [review] + findings = outputs["findings"]["reviews"] + assert findings[0]["file"] == review["file"] + finding = dict(findings[0]["reviews"][0]) + assert finding.pop("id") + assert finding == review["reviews"][0] assert outputs["sarif"]["runs"][0]["results"][0]["message"]["text"] == ( "unchecked input" ) diff --git a/tests/test_sarif.py b/tests/test_sarif.py index 68ede5a6..b64d9ddb 100644 --- a/tests/test_sarif.py +++ b/tests/test_sarif.py @@ -2,9 +2,11 @@ # SPDX-License-Identifier: Apache-2.0 import hashlib +from pathlib import Path from metis.sarif.writer import generate_sarif from metis.sarif.utils import read_file_lines, create_fingerprint +from metis.sarif.triage import METIS_FINDING_ID_KEY def test_read_file_lines(tmp_path): @@ -93,6 +95,65 @@ def test_generate_sarif_single_issue(tmp_path): # Fingerprint matches utility fp_expected = create_fingerprint(str(temp_file), 2, "AI001") assert issue_entry["partialFingerprints"]["primaryLocationLineHash"] == fp_expected + finding_id = results["reviews"][0]["reviews"][0]["id"] + assert issue_entry["properties"][METIS_FINDING_ID_KEY] == finding_id + + +def test_review_finding_id_survives_serialization_and_sarif() -> None: + from metis.engine.stages.review.models import ReviewFinding + + finding = ReviewFinding(issue="Example issue") + restored = ReviewFinding.model_validate(finding.model_dump()) + sarif = generate_sarif( + {"reviews": [{"file": "example.c", "reviews": [restored.model_dump()]}]} + ) + + assert restored.id == finding.id + assert sarif["runs"][0]["results"][0]["properties"][METIS_FINDING_ID_KEY] == ( + finding.id + ) + + +def test_generate_sarif_repairs_duplicate_finding_ids() -> None: + issues = [ + {"id": "duplicate", "issue": "First"}, + {"id": "duplicate", "issue": "Second"}, + ] + + sarif = generate_sarif({"reviews": [{"file": "example.c", "reviews": issues}]}) + + finding_ids = [issue["id"] for issue in issues] + sarif_ids = [ + result["properties"][METIS_FINDING_ID_KEY] + for result in sarif["runs"][0]["results"] + ] + assert len(set(finding_ids)) == 2 + assert sarif_ids == finding_ids + + +def test_reachability_finding_id_survives_review_and_sarif(tmp_path: Path) -> None: + from metis.engine.nodes.reachability.domain import VulnerabilityFinding + from metis.engine.nodes.reachability.finding_adapter import finding_to_review_item + + finding = VulnerabilityFinding( + "reachability-finding", + "other", + "high", + 0.8, + "source", + "example.c", + 1, + "sink", + "example.c", + 2, + ) + item = finding_to_review_item(finding, codebase_path=str(tmp_path)) + sarif = generate_sarif({"reviews": [{"file": "example.c", "reviews": [item]}]}) + + assert item["id"] == finding.id + assert sarif["runs"][0]["results"][0]["properties"][METIS_FINDING_ID_KEY] == ( + finding.id + ) def test_generate_sarif_uses_issue_metadata_when_source_missing():