Fix some tests

Author: drdavella

pyproject.toml

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

src/codemodder/codeql.py

@@ -1,5 +1,7 @@
 from pathlib import Path
 
+from sarif_pydantic import Location as LocationModel
+from sarif_pydantic import Result as ResultModel
 from sarif_pydantic import Sarif
 from typing_extensions import Self
 
@@ -14,27 +16,30 @@ def detect(cls, run_data: Run) -> bool:
 
 
 class CodeQLLocation(SarifLocation):
-    @staticmethod
-    def get_snippet(sarif_location) -> str:
-        return ""
-
     @classmethod
-    def from_sarif(cls, sarif_location) -> Self:
-        artifact_location = sarif_location["physicalLocation"]["artifactLocation"]
-        file = Path(artifact_location["uri"])
+    def from_sarif(cls, sarif_location: LocationModel) -> Self:
+        if (physical_location := sarif_location.physical_location) is None:
+            raise ValueError("Location does not contain a physicalLocation")
+        if (artifact_location := physical_location.artifact_location) is None:
+            raise ValueError("PhysicalLocation does not contain an artifactLocation")
+        if (uri := artifact_location.uri) is None:
+            raise ValueError("ArtifactLocation does not contain a uri")
+
+        file = Path(uri)
 
-        try:
-            region = sarif_location["physicalLocation"]["region"]
-        except KeyError:
+        if not (region := physical_location.region):
             # A location without a region indicates a result for the entire file.
             # Use sentinel values of 0 index for start/end
             zero = LineInfo(0)
             return cls(file=file, start=zero, end=zero)
 
-        start = LineInfo(line=region["startLine"], column=region.get("startColumn"))
+        if not region.start_line:
+            raise ValueError("Region does not contain a startLine")
+
+        start = LineInfo(line=region.start_line, column=region.start_column or -1)
         end = LineInfo(
-            line=region.get("endLine", start.line),
-            column=region.get("endColumn", start.column),
+            line=region.end_line or start.line,
+            column=region.end_column or start.column,
         )
         return cls(file=file, start=start, end=end)
 
@@ -43,7 +48,7 @@ class CodeQLResult(SarifResult):
     location_type = CodeQLLocation
 
     @classmethod
-    def rule_url_from_id(cls, result: dict, run: dict, rule_id: str) -> str:
+    def rule_url_from_id(cls, result: ResultModel, run: Run, rule_id: str) -> str:
         del result, run, rule_id
         # TODO: Implement this method to return the specific rule URL
         return "https://codeql.github.com/codeql-query-help/"

src/codemodder/result.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import itertools
-from abc import abstractmethod
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, ClassVar, Sequence, Type
@@ -39,23 +38,30 @@ class Location(ABCDataclass):
 @dataclass(frozen=True)
 class SarifLocation(Location):
     @staticmethod
-    @abstractmethod
-    def get_snippet(sarif_location) -> str:
-        pass
+    def get_snippet(sarif_location: LocationModel) -> str | None:
+        return sarif_location.message.text if sarif_location.message else None
 
     @classmethod
     def from_sarif(cls, sarif_location: LocationModel) -> Self:
-        artifact_location = sarif_location["physicalLocation"]["artifactLocation"]
-        file = Path(artifact_location["uri"])
+        if not (physical_location := sarif_location.physical_location):
+            raise ValueError("Sarif location does not have a physical location")
+        if not (artifact_location := physical_location.artifact_location):
+            raise ValueError("Sarif location does not have an artifact location")
+        if not (region := physical_location.region):
+            raise ValueError("Sarif location does not have a region")
+        if not (uri := artifact_location.uri):
+            raise ValueError("Sarif location does not have a uri")
+
+        file = Path(uri)
         snippet = cls.get_snippet(sarif_location)
         start = LineInfo(
-            line=sarif_location["physicalLocation"]["region"]["startLine"],
-            column=sarif_location["physicalLocation"]["region"]["startColumn"],
+            line=region.start_line or -1,
+            column=region.start_column or -1,
             snippet=snippet,
         )
         end = LineInfo(
-            line=sarif_location["physicalLocation"]["region"]["endLine"],
-            column=sarif_location["physicalLocation"]["region"]["endColumn"],
+            line=region.end_line or -1,
+            column=region.end_column or -1,
             snippet=snippet,
         )
         return cls(file=file, start=start, end=end)
@@ -150,14 +156,17 @@ def extract_locations(cls, sarif_result: ResultModel) -> Sequence[Location]:
         )
 
     @classmethod
-    def extract_related_locations(cls, sarif_result) -> Sequence[LocationWithMessage]:
+    def extract_related_locations(
+        cls, sarif_result: ResultModel
+    ) -> Sequence[LocationWithMessage]:
         return tuple(
             [
                 LocationWithMessage(
-                    message=rel_location.get("message", {}).get("text", ""),
+                    message=rel_location.message.text,
                     location=cls.location_type.from_sarif(rel_location),
                 )
-                for rel_location in sarif_result.get("relatedLocations", [])
+                for rel_location in sarif_result.related_locations or []
+                if rel_location.message
             ]
         )
 
@@ -187,7 +196,12 @@ def extract_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 := result.rule) and sarif_run.tool.extensions and rule.tool_component:
+        if (
+            (rule := result.rule)
+            and sarif_run.tool.extensions
+            and rule.tool_component
+            and rule.tool_component.index is not None
+        ):
             tool_index = rule.tool_component.index
             rule_index = rule.index
             return sarif_run.tool.extensions[tool_index].rules[rule_index].id

src/codemodder/semgrep.py

@@ -1,38 +1,52 @@
 import itertools
-import json
 import subprocess
 from pathlib import Path
 from tempfile import NamedTemporaryFile
 from typing import Iterable, Optional
 
+from sarif_pydantic import Sarif
 from typing_extensions import Self, override
 
 from codemodder.context import CodemodExecutionContext
 from codemodder.logging import logger
-from codemodder.result import Result, ResultSet, SarifLocation, SarifResult
+from codemodder.result import (
+    LocationModel,
+    Result,
+    ResultModel,
+    ResultSet,
+    SarifLocation,
+    SarifResult,
+)
 from codemodder.sarifs import AbstractSarifToolDetector, Run
 
 
 class SemgrepSarifToolDetector(AbstractSarifToolDetector):
     @classmethod
     def detect(cls, run_data: Run) -> bool:
-        return "semgrep" in run_data.tool.driver.name
+        return "semgrep" in run_data.tool.driver.name.lower()
 
 
 class SemgrepLocation(SarifLocation):
     @staticmethod
-    def get_snippet(sarif_location) -> str:
+    def get_snippet(sarif_location: LocationModel) -> str | None:
         return (
-            sarif_location["physicalLocation"]["region"].get("snippet", {}).get("text")
+            sarif_location.physical_location.region.snippet.text
+            if sarif_location.physical_location
+            and sarif_location.physical_location.region
+            and sarif_location.physical_location.region.snippet
+            else SarifLocation.get_snippet(sarif_location)
         )
 
 
 class SemgrepResult(SarifResult):
     location_type = SemgrepLocation
 
+    @override
     @classmethod
-    def rule_url_from_id(cls, sarif_result, sarif_run, rule_id):
-        del sarif_result, sarif_run
+    def rule_url_from_id(
+        cls, result: ResultModel, run: Run, rule_id: str
+    ) -> str | None:
+        del result, run
         from core_codemods.semgrep.api import semgrep_url_from_id
 
         return semgrep_url_from_id(rule_id)
@@ -41,17 +55,16 @@ def rule_url_from_id(cls, sarif_result, sarif_run, rule_id):
 class SemgrepResultSet(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())
 
         result_set = cls()
-        for sarif_run in data["runs"]:
-            for result in sarif_run["results"]:
+        for sarif_run in data.runs:
+            for result in sarif_run.results or []:
                 sarif_result = SemgrepResult.from_sarif(
                     result, sarif_run, truncate_rule_id
                 )
                 result_set.add_result(sarif_result)
-            result_set.store_tool_data(sarif_run.get("tool", {}))
+            result_set.store_tool_data(sarif_run.tool.model_dump())
         return result_set
 
 

tests/codemods/semgrep/test_semgrep_django_secure_set_cookie.py

@@ -38,6 +38,7 @@ def index(request, template):
         results = {
             "runs": [
                 {
+                    "tool": {"driver": {"name": "Semgrep OSS"}},
                     "results": [
                         {
                             "fingerprints": {"matchBasedId/v1": "123"},
@@ -66,7 +67,7 @@ def index(request, template):
                             "properties": {},
                             "ruleId": "python.django.security.audit.secure-cookies.django-secure-set-cookie",
                         }
-                    ]
+                    ],
                 }
             ]
         }

tests/codemods/semgrep/test_semgrep_enable_jinja2_autoescape.py

@@ -27,6 +27,7 @@ def test_import(self, tmpdir):
         results = {
             "runs": [
                 {
+                    "tool": {"driver": {"name": "Semgrep OSS"}},
                     "results": [
                         {
                             "fingerprints": {"matchBasedId/v1": "123"},
@@ -55,7 +56,7 @@ def test_import(self, tmpdir):
                             "properties": {},
                             "ruleId": "python.flask.security.xss.audit.direct-use-of-jinja2.direct-use-of-jinja2",
                         }
-                    ]
+                    ],
                 }
             ]
         }

tests/codemods/semgrep/test_semgrep_harden_pyyaml.py

@@ -26,6 +26,7 @@ def test_pyyaml(self, tmpdir):
         results = {
             "runs": [
                 {
+                    "tool": {"driver": {"name": "Semgrep OSS"}},
                     "results": [
                         {
                             "fingerprints": {"matchBasedId/v1": "123"},
@@ -54,7 +55,7 @@ def test_pyyaml(self, tmpdir):
                             "properties": {},
                             "ruleId": "python.lang.security.deserialization.avoid-pyyaml-load.avoid-pyyaml-load",
                         }
-                    ]
+                    ],
                 }
             ]
         }
@@ -84,6 +85,7 @@ def index(request):
         results = {
             "runs": [
                 {
+                    "tool": {"driver": {"name": "Semgrep OSS"}},
                     "results": [
                         {
                             "fingerprints": {"matchBasedId/v1": "123"},
@@ -112,7 +114,7 @@ def index(request):
                             "properties": {},
                             "ruleId": "python.django.security.audit.avoid-insecure-deserialization.avoid-insecure-deserialization",
                         }
-                    ]
+                    ],
                 }
             ]
         }

tests/codemods/semgrep/test_semgrep_jwt_decode_verify.py

@@ -25,6 +25,7 @@ def test_import(self, tmpdir):
         results = {
             "runs": [
                 {
+                    "tool": {"driver": {"name": "Semgrep OSS"}},
                     "results": [
                         {
                             "fingerprints": {"matchBasedId/v1": "123"},
@@ -52,7 +53,7 @@ def test_import(self, tmpdir):
                             },
                             "ruleId": "python.jwt.security.unverified-jwt-decode.unverified-jwt-decode",
                         }
-                    ]
+                    ],
                 }
             ]
         }

tests/codemods/semgrep/test_semgrep_nan_injection.py

@@ -38,6 +38,7 @@ def home(request):
         results = {
             "runs": [
                 {
+                    "tool": {"driver": {"name": "Semgrep OSS"}},
                     "results": [
                         {
                             "fingerprints": {"matchBasedId/v1": "1932"},
@@ -112,6 +113,7 @@ def view(request):
         results = {
             "runs": [
                 {
+                    "tool": {"driver": {"name": "Semgrep OSS"}},
                     "results": [
                         {
                             "fingerprints": {"matchBasedId/v1": "1fdbd5a"},
@@ -247,6 +249,7 @@ def view(request):
         results = {
             "runs": [
                 {
+                    "tool": {"driver": {"name": "Semgrep OSS"}},
                     "results": [
                         {
                             "fingerprints": {"matchBasedId/v1": "asdfg"},
@@ -304,6 +307,7 @@ def view(request):
         results = {
             "runs": [
                 {
+                    "tool": {"driver": {"name": "Semgrep OSS"}},
                     "results": [
                         {
                             "fingerprints": {"matchBasedId/v1": "q324"},
@@ -359,6 +363,7 @@ def view(request):
         results = {
             "runs": [
                 {
+                    "tool": {"driver": {"name": "Semgrep OSS"}},
                     "results": [
                         {
                             "fingerprints": {"matchBasedId/v1": "asdtg"},
@@ -416,6 +421,7 @@ def view(request):
         results = {
             "runs": [
                 {
+                    "tool": {"driver": {"name": "Semgrep OSS"}},
                     "results": [
                         {
                             "fingerprints": {"matchBasedId/v1": "asd2"},

tests/codemods/semgrep/test_semgrep_no_csrf_exempt.py

@@ -52,6 +52,7 @@ def foo():
         results = {
             "runs": [
                 {
+                    "tool": {"driver": {"name": "Semgrep OSS"}},
                     "results": [
                         {
                             "fingerprints": {"matchBasedId/v1": "a3ca2"},