Use Sarif data model for detection

Author: drdavella

pyproject.toml

@@ -25,6 +25,7 @@ dependencies = [
     "tomlkit~=0.13.0",
     "wrapt~=1.17.0",
     "chardet~=5.2.0",
+    "sarif-pydantic~=0.1.0",
     "setuptools~=78.1",
 ]
 keywords = ["codemod", "codemods", "security", "fix", "fixes"]

src/codemodder/codeql.py

@@ -4,13 +4,13 @@
 from typing_extensions import Self
 
 from codemodder.result import LineInfo, ResultSet, SarifLocation, SarifResult
-from codemodder.sarifs import AbstractSarifToolDetector
+from codemodder.sarifs import AbstractSarifToolDetector, Run
 
 
 class CodeQLSarifToolDetector(AbstractSarifToolDetector):
     @classmethod
-    def detect(cls, run_data: dict) -> bool:
-        return "tool" in run_data and "CodeQL" in run_data["tool"]["driver"]["name"]
+    def detect(cls, run_data: Run) -> bool:
+        return "CodeQL" in run_data.tool.driver.name
 
 
 class CodeQLLocation(SarifLocation):

src/codemodder/sarifs.py

@@ -1,17 +1,19 @@
-import json
 from abc import ABCMeta, abstractmethod
 from collections import defaultdict
 from importlib.metadata import entry_points
 from pathlib import Path
 from typing import DefaultDict
 
+from pydantic import ValidationError
+from sarif_pydantic import Run, Sarif
+
 from codemodder.logging import logger
 
 
 class AbstractSarifToolDetector(metaclass=ABCMeta):
     @classmethod
     @abstractmethod
-    def detect(cls, run_data: dict) -> bool:
+    def detect(cls, run_data: Run) -> bool:
         pass
 
 
@@ -24,18 +26,16 @@ def detect_sarif_tools(filenames: list[Path]) -> DefaultDict[str, list[Path]]:
     }
     for fname in filenames:
         try:
-            data = json.loads(fname.read_text(encoding="utf-8-sig"))
-        except json.JSONDecodeError:
-            logger.exception("Malformed JSON file: %s", fname)
+            data = Sarif.model_validate_json(fname.read_text(encoding="utf-8-sig"))
+        except ValidationError:
+            logger.exception("Invalid SARIF file: %s", fname)
             raise
-        for name, det in detectors.items():
-            try:
-                runs = data["runs"]
-            except KeyError:
-                logger.exception("Sarif file without `runs` data: %s", fname)
-                raise
 
-            for run in runs:
+        if not data.runs:
+            raise ValueError(f"SARIF file without `runs` data: {fname}")
+
+        for name, det in detectors.items():
+            for run in data.runs:
                 try:
                     if det.detect(run):
                         logger.debug("detected %s sarif: %s", name, fname)

src/codemodder/semgrep.py

@@ -10,16 +10,13 @@
 from codemodder.context import CodemodExecutionContext
 from codemodder.logging import logger
 from codemodder.result import Result, ResultSet, SarifLocation, SarifResult
-from codemodder.sarifs import AbstractSarifToolDetector
+from codemodder.sarifs import AbstractSarifToolDetector, Run
 
 
 class SemgrepSarifToolDetector(AbstractSarifToolDetector):
     @classmethod
-    def detect(cls, run_data: dict) -> bool:
-        return (
-            "tool" in run_data
-            and "semgrep" in run_data["tool"]["driver"]["name"].lower()
-        )
+    def detect(cls, run_data: Run) -> bool:
+        return "semgrep" in run_data.tool.driver.name
 
 
 class SemgrepLocation(SarifLocation):

tests/test_sarif_processing.py

@@ -3,6 +3,7 @@
 from pathlib import Path
 
 import pytest
+from pydantic import ValidationError
 
 from codemodder.codemods.semgrep import process_semgrep_findings
 from codemodder.sarifs import detect_sarif_tools
@@ -119,9 +120,9 @@ def test_bad_sarif(self, tmpdir, caplog):
             # remove all { to make a badly formatted json
             f.write(sarif_file.read_text(encoding="utf-8").replace("{", ""))
 
-        with pytest.raises(json.JSONDecodeError):
+        with pytest.raises(ValidationError):
             detect_sarif_tools([bad_json])
-        assert f"Malformed JSON file: {str(bad_json)}" in caplog.text
+        assert f"Invalid SARIF file: {str(bad_json)}" in caplog.text
 
     def test_bad_sarif_no_runs_data(self, tmpdir, caplog):
         bad_json = tmpdir / "bad.sarif"
@@ -134,9 +135,9 @@ def test_bad_sarif_no_runs_data(self, tmpdir, caplog):
         with open(bad_json, "w") as f:
             f.write(data)
 
-        with pytest.raises(KeyError):
+        with pytest.raises(ValidationError):
             detect_sarif_tools([bad_json])
-        assert f"Sarif file without `runs` data: {str(bad_json)}" in caplog.text
+        assert f"Invalid SARIF file: {str(bad_json)}" in caplog.text
 
     def test_two_sarifs_different_tools(self):
         results = detect_sarif_tools(