Intermediate progress: update to sarif-pydantic 0.2.0

Author: drdavella

pyproject.toml

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

src/codemodder/codeql.py

@@ -1,6 +1,6 @@
-import json
 from pathlib import Path
 
+from sarif_pydantic import Sarif
 from typing_extensions import Self
 
 from codemodder.result import LineInfo, ResultSet, SarifLocation, SarifResult
@@ -52,16 +52,17 @@ def rule_url_from_id(cls, result: dict, run: dict, rule_id: str) -> str:
 class CodeQLResultSet(ResultSet):
     @classmethod
     def from_sarif(cls, sarif_file: str | Path, truncate_rule_id: bool = False) -> Self:
-        with open(sarif_file, "r", encoding="utf-8") as f:
-            data = json.load(f)
+        data = Sarif.model_validate_json(
+            Path(sarif_file).read_text(encoding="utf-8-sig")
+        )
 
         result_set = cls()
-        for sarif_run in data["runs"]:
+        for sarif_run in data.runs:
             if CodeQLSarifToolDetector.detect(sarif_run):
-                for sarif_result in sarif_run["results"]:
+                for sarif_result in sarif_run.results or []:
                     codeql_result = CodeQLResult.from_sarif(
                         sarif_result, sarif_run, truncate_rule_id
                     )
                     result_set.add_result(codeql_result)
-                result_set.store_tool_data(sarif_run.get("tool", {}))
+                result_set.store_tool_data(sarif_run.tool.model_dump())
         return result_set

src/codemodder/result.py

@@ -9,6 +9,9 @@
 import libcst as cst
 from boltons.setutils import IndexedSet
 from libcst._position import CodeRange
+from sarif_pydantic import Location as LocationModel
+from sarif_pydantic import Result as ResultModel
+from sarif_pydantic import Run
 from typing_extensions import Self
 
 from codemodder.codetf import Finding, Rule
@@ -41,7 +44,7 @@ def get_snippet(sarif_location) -> str:
         pass
 
     @classmethod
-    def from_sarif(cls, sarif_location) -> Self:
+    def from_sarif(cls, sarif_location: LocationModel) -> Self:
         artifact_location = sarif_location["physicalLocation"]["artifactLocation"]
         file = Path(artifact_location["uri"])
         snippet = cls.get_snippet(sarif_location)
@@ -102,7 +105,7 @@ class SarifResult(SASTResult):
 
     @classmethod
     def from_sarif(
-        cls, sarif_result, sarif_run, truncate_rule_id: bool = False
+        cls, sarif_result: ResultModel, sarif_run: Run, truncate_rule_id: bool = False
     ) -> Self:
         rule_id = cls.extract_rule_id(sarif_result, sarif_run, truncate_rule_id)
         finding_id = cls.extract_finding_id(sarif_result) or rule_id
@@ -124,20 +127,25 @@ def from_sarif(
         )
 
     @classmethod
-    def extract_finding_message(cls, sarif_result: dict, sarif_run: dict) -> str | None:
-        return sarif_result.get("message", {}).get("text", None)
+    def extract_finding_message(
+        cls, sarif_result: ResultModel, sarif_run: Run
+    ) -> str | None:
+        del sarif_run
+        return sarif_result.message.text
 
     @classmethod
-    def rule_url_from_id(cls, result: dict, run: dict, rule_id: str) -> str | None:
+    def rule_url_from_id(
+        cls, result: ResultModel, run: Run, rule_id: str
+    ) -> str | None:
         del result, run, rule_id
         return None
 
     @classmethod
-    def extract_locations(cls, sarif_result) -> Sequence[Location]:
+    def extract_locations(cls, sarif_result: ResultModel) -> Sequence[Location]:
         return tuple(
             [
                 cls.location_type.from_sarif(location)
-                for location in sarif_result["locations"]
+                for location in sarif_result.locations or []
             ]
         )
 
@@ -154,38 +162,41 @@ def extract_related_locations(cls, sarif_result) -> Sequence[LocationWithMessage
         )
 
     @classmethod
-    def extract_code_flows(cls, sarif_result) -> Sequence[Sequence[Location]]:
+    def extract_code_flows(
+        cls, sarif_result: ResultModel
+    ) -> Sequence[Sequence[Location]]:
         return tuple(
             [
                 tuple(
                     [
-                        cls.location_type.from_sarif(locations.get("location"))
-                        for locations in threadflow.get("locations", {})
+                        cls.location_type.from_sarif(locations.location)
+                        for locations in threadflow.locations or []
+                        if locations.location
                     ]
                 )
-                for codeflow in sarif_result.get("codeFlows", {})
-                for threadflow in codeflow.get("threadFlows", {})
+                for codeflow in sarif_result.code_flows or []
+                for threadflow in codeflow.thread_flows or []
             ]
         )
 
     @classmethod
-    def extract_rule_id(cls, result, sarif_run, truncate_rule_id: bool = False) -> str:
-        if rule_id := result.get("ruleId"):
+    def extract_rule_id(
+        cls, result: ResultModel, sarif_run: Run, truncate_rule_id: bool = False
+    ) -> str:
+        if rule_id := result.rule_id:
             return rule_id.split(".")[-1] if truncate_rule_id else rule_id
 
         # it may be contained in the 'rule' field through the tool component in the sarif file
-        if "rule" in result:
-            tool_index = result["rule"]["toolComponent"]["index"]
-            rule_index = result["rule"]["index"]
-            return sarif_run["tool"]["extensions"][tool_index]["rules"][rule_index][
-                "id"
-            ]
+        if (rule := result.rule) and sarif_run.tool.extensions and rule.tool_component:
+            tool_index = rule.tool_component.index
+            rule_index = rule.index
+            return sarif_run.tool.extensions[tool_index].rules[rule_index].id
 
         raise ValueError("Could not extract rule id from sarif result.")
 
     @classmethod
-    def extract_finding_id(cls, result) -> str | None:
-        return result.get("guid") or result.get("correlationGuid")
+    def extract_finding_id(cls, result: ResultModel) -> str | None:
+        return str(result.guid or "") or str(result.correlation_guid or "") or None
 
 
 def same_line(pos: CodeRange, location: Location) -> bool: