Add detection and processing for external Semgrep SARIF files (#459)

* Refactor semgrep SARIFs into semgrep module

* Add SARIF detector plugin for semgrep

* Add semgrep SARIF file detection and processing

Author: drdavella

.pre-commit-config.yaml

@@ -14,6 +14,10 @@ repos:
               tests/.*
           )$
     -   id: check-added-large-files
+        exclude: |
+          (?x)^(
+              tests/samples/pygoat.semgrep.sarif.json
+          )$
 -   repo: https://github.com/psf/black
     rev: 24.3.0
     hooks:

pyproject.toml

@@ -83,6 +83,9 @@ core = "core_codemods:registry"
 sonar = "core_codemods:sonar_registry"
 defectdojo = "core_codemods:defectdojo_registry"
 
+[project.entry-points.sarif_detectors]
+"semgrep" = "codemodder.semgrep:SemgrepSarifToolDetector"
+
 [tool.setuptools]
 
 [tool.setuptools.package-data]

src/codemodder/codemods/semgrep.py

@@ -1,13 +1,15 @@
 import io
 import os
 import tempfile
+from functools import cache
 from pathlib import Path
 
 import yaml
 
 from codemodder.codemods.base_detector import BaseDetector
 from codemodder.context import CodemodExecutionContext
 from codemodder.result import ResultSet
+from codemodder.semgrep import SemgrepResultSet
 from codemodder.semgrep import run as semgrep_run
 
 
@@ -47,3 +49,25 @@ def apply(
         yaml_files = self.get_yaml_files(codemod_id)
         with context.timer.measure("semgrep"):
             return semgrep_run(context, yaml_files, files_to_analyze)
+
+
+class SemgrepSarifFileDetector(BaseDetector):
+    def apply(
+        self,
+        codemod_id: str,
+        context: CodemodExecutionContext,
+        files_to_analyze: list[Path],
+    ) -> ResultSet:
+        del codemod_id
+        del files_to_analyze
+        return process_semgrep_findings(
+            tuple(context.tool_result_files_map.get("semgrep", ()))
+        )  # Convert list to tuple for cache hashability
+
+
+@cache
+def process_semgrep_findings(semgrep_sarif_files: tuple[str]) -> ResultSet:
+    results = SemgrepResultSet()
+    for file in semgrep_sarif_files or ():
+        results |= SemgrepResultSet.from_sarif(file)
+    return results

src/codemodder/sarifs.py

@@ -3,14 +3,10 @@
 from collections import defaultdict
 from importlib.metadata import entry_points
 from pathlib import Path
-from typing import DefaultDict, Optional
-
-from typing_extensions import Self
+from typing import DefaultDict
 
 from codemodder.logging import logger
 
-from .result import LineInfo, Location, Result, ResultSet
-
 
 class AbstractSarifToolDetector(metaclass=ABCMeta):
     @classmethod
@@ -39,65 +35,3 @@ def detect_sarif_tools(filenames: list[Path]) -> DefaultDict[str, list[str]]:
                     continue
 
     return results
-
-
-def extract_rule_id(result, sarif_run) -> Optional[str]:
-    if "ruleId" in result:
-        # semgrep preprends the folders into the rule-id, we want the base name only
-        return result["ruleId"].rsplit(".")[-1]
-
-    # 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"]
-
-    return None
-
-
-# NOTE: These Sarif classes are actually specific to Semgrep and should be moved elsewhere
-class SarifLocation(Location):
-    @classmethod
-    def from_sarif(cls, sarif_location) -> Self:
-        artifact_location = sarif_location["physicalLocation"]["artifactLocation"]
-        file = Path(artifact_location["uri"])
-        start = LineInfo(
-            line=sarif_location["physicalLocation"]["region"]["startLine"],
-            column=sarif_location["physicalLocation"]["region"]["startColumn"],
-            snippet=sarif_location["physicalLocation"]["region"]["snippet"]["text"],
-        )
-        end = LineInfo(
-            line=sarif_location["physicalLocation"]["region"]["endLine"],
-            column=sarif_location["physicalLocation"]["region"]["endColumn"],
-            snippet=sarif_location["physicalLocation"]["region"]["snippet"]["text"],
-        )
-        return cls(file=file, start=start, end=end)
-
-
-class SarifResult(Result):
-    @classmethod
-    def from_sarif(cls, sarif_result, sarif_run) -> Self:
-        rule_id = extract_rule_id(sarif_result, sarif_run)
-        if not rule_id:
-            raise ValueError("Could not extract rule id from sarif result.")
-
-        locations: list[Location] = []
-        for location in sarif_result["locations"]:
-            artifact_location = SarifLocation.from_sarif(location)
-            locations.append(artifact_location)
-        return cls(rule_id=rule_id, locations=locations)
-
-
-class SarifResultSet(ResultSet):
-    @classmethod
-    def from_sarif(cls, sarif_file: str | Path) -> Self:
-        with open(sarif_file, "r", encoding="utf-8") as f:
-            data = json.load(f)
-
-        result_set = cls()
-        for sarif_run in data["runs"]:
-            for result in sarif_run["results"]:
-                sarif_result = SarifResult.from_sarif(result, sarif_run)
-                result_set.add_result(sarif_result)
-
-        return result_set

src/codemodder/semgrep.py

@@ -1,19 +1,93 @@
 import itertools
+import json
 import subprocess
 from pathlib import Path
 from tempfile import NamedTemporaryFile
 from typing import Iterable, Optional
 
+from typing_extensions import Self
+
 from codemodder.context import CodemodExecutionContext
 from codemodder.logging import logger
-from codemodder.sarifs import SarifResultSet
+from codemodder.result import LineInfo, Location, Result, ResultSet
+from codemodder.sarifs import AbstractSarifToolDetector
+
+
+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 extract_rule_id(result, sarif_run) -> Optional[str]:
+    if "ruleId" in result:
+        # semgrep preprends the folders into the rule-id, we want the base name only
+        return result["ruleId"].rsplit(".")[-1]
+
+    # 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"]
+
+    return None
+
+
+class SemgrepLocation(Location):
+    @classmethod
+    def from_sarif(cls, sarif_location) -> Self:
+        artifact_location = sarif_location["physicalLocation"]["artifactLocation"]
+        file = Path(artifact_location["uri"])
+        start = LineInfo(
+            line=sarif_location["physicalLocation"]["region"]["startLine"],
+            column=sarif_location["physicalLocation"]["region"]["startColumn"],
+            snippet=sarif_location["physicalLocation"]["region"]["snippet"]["text"],
+        )
+        end = LineInfo(
+            line=sarif_location["physicalLocation"]["region"]["endLine"],
+            column=sarif_location["physicalLocation"]["region"]["endColumn"],
+            snippet=sarif_location["physicalLocation"]["region"]["snippet"]["text"],
+        )
+        return cls(file=file, start=start, end=end)
+
+
+class SemgrepResult(Result):
+    @classmethod
+    def from_sarif(cls, sarif_result, sarif_run) -> Self:
+        rule_id = extract_rule_id(sarif_result, sarif_run)
+        if not rule_id:
+            raise ValueError("Could not extract rule id from sarif result.")
+
+        locations: list[Location] = []
+        for location in sarif_result["locations"]:
+            artifact_location = SemgrepLocation.from_sarif(location)
+            locations.append(artifact_location)
+        return cls(rule_id=rule_id, locations=locations)
+
+
+class SemgrepResultSet(ResultSet):
+    @classmethod
+    def from_sarif(cls, sarif_file: str | Path) -> Self:
+        with open(sarif_file, "r", encoding="utf-8") as f:
+            data = json.load(f)
+
+        result_set = cls()
+        for sarif_run in data["runs"]:
+            for result in sarif_run["results"]:
+                sarif_result = SemgrepResult.from_sarif(result, sarif_run)
+                result_set.add_result(sarif_result)
+
+        return result_set
 
 
 def run(
     execution_context: CodemodExecutionContext,
     yaml_files: Iterable[Path],
     files_to_analyze: Optional[Iterable[Path]] = None,
-) -> SarifResultSet:
+) -> SemgrepResultSet:
     """
     Runs Semgrep and outputs a dict with the results organized by rule_id.
     """
@@ -49,5 +123,5 @@ def run(
             if not execution_context.verbose:
                 logger.error("captured semgrep stderr: %s", call.stderr)
             raise subprocess.CalledProcessError(call.returncode, command)
-        results = SarifResultSet.from_sarif(temp_sarif_file.name)
+        results = SemgrepResultSet.from_sarif(temp_sarif_file.name)
         return results

src/core_codemods/sonar/api.py

@@ -59,6 +59,8 @@ def apply(
         context: CodemodExecutionContext,
         files_to_analyze: list[Path],
     ) -> ResultSet:
+        del codemod_id
+        del files_to_analyze
         sonar_findings = process_sonar_findings(
             tuple(
                 context.tool_result_files_map.get("sonar", ())