✨ Implement detections SARIF export

pyproject.toml

@@ -40,6 +40,8 @@ abch_tree_sitter_solidity = "^1.2.0"
 lazy-import = "^0.2.2"
 importlib-metadata = { version = "4.8", python = "<3.10" }
 packaging = ">=22.0"
+sarif-om = "^1.0.4"
+jschema-to-python = "^1.2.3"
 
 pytest-asyncio = { version = "^0.17", optional = true }
 GitPython = { version = "^3.1.20", optional = true }

woke/cli/detect.py

@@ -303,9 +303,11 @@ async def detect_(
     watch: bool,
     ignore_disable_overrides: bool,
 ):
+    from jschema_to_python.to_json import to_json
     from watchdog.observers import Observer
 
     from woke.detectors.api import detect, print_detection
+    from woke.detectors.utils import create_sarif_log
 
     from ..compiler import SolcOutputSelectionEnum, SolidityCompiler
     from ..compiler.build_data_model import ProjectBuild, ProjectBuildInfo
@@ -336,7 +338,7 @@ def callback(build: ProjectBuild, build_info: ProjectBuildInfo):
         else:
             detectors = ctx.invoked_subcommand
 
-        detections, exceptions = detect(
+        used_detectors, detections, exceptions = detect(
             detectors,
             build,
             build_info,
@@ -388,6 +390,13 @@ def callback(build: ProjectBuild, build_info: ProjectBuildInfo):
                 str(config.project_root_path / "woke-detections.ansi"),
                 styles=True,
             )
+        elif export == "sarif":
+            log = create_sarif_log(
+                used_detectors, {name: d[0] for name, d in detections.items()}
+            )
+            (config.project_root_path / "woke-detections.sarif").write_text(
+                to_json(log)
+            )
 
         console.record = False
 
@@ -482,7 +491,7 @@ def callback(build: ProjectBuild, build_info: ProjectBuildInfo):
 )
 @click.option(
     "--export",
-    type=click.Choice(["svg", "html", "text", "ansi"], case_sensitive=False),
+    type=click.Choice(["svg", "html", "text", "ansi", "sarif"], case_sensitive=False),
     help="Export detections to file.",
 )
 @click.option(

woke/detectors/api.py

@@ -186,7 +186,9 @@ def detect(
     capture_exceptions: bool = False,
     logging_handler: Optional[logging.Handler] = None,
 ) -> Tuple[
-    Dict[str, Tuple[List[DetectorResult], List[DetectorResult]]], Dict[str, Exception]
+    List[click.Command],
+    Dict[str, Tuple[List[DetectorResult], List[DetectorResult]]],
+    Dict[str, Exception],
 ]:
     from contextlib import nullcontext
 
@@ -423,7 +425,7 @@ def _callback(*args, **kwargs):
                 raise
             exceptions[detector_name] = e
 
-    return detections, exceptions
+    return detectors, detections, exceptions
 
 
 def print_detection(

woke/detectors/utils.py

@@ -0,0 +1,163 @@
+import os
+import sys
+from inspect import cleandoc
+from typing import Dict, List
+
+from rich_click import Command
+from sarif_om import (
+    ArtifactLocation,
+    Invocation,
+    Location,
+    PhysicalLocation,
+    Region,
+    ReportingDescriptor,
+    ReportingDescriptorReference,
+    Result,
+    Run,
+    SarifLog,
+    Tool,
+    ToolComponent,
+    ToolComponentReference,
+)
+
+from woke.utils import get_package_version
+from woke.utils.keyed_default_dict import KeyedDefaultDict
+
+from .api import DetectionImpact, DetectorResult
+
+# pyright: reportGeneralTypeIssues=false
+
+
+def create_sarif_log(
+    detectors: List[Command], detections: Dict[str, List[DetectorResult]]
+) -> SarifLog:
+    from woke.cli.detect import run_detect
+
+    if sys.version_info < (3, 10):
+        from importlib_metadata import packages_distributions
+    else:
+        from importlib.metadata import packages_distributions
+
+    distributions = packages_distributions()
+
+    driver = ToolComponent(
+        name="woke",
+        semantic_version=get_package_version("woke"),
+        rules=[],
+    )
+    extensions = KeyedDefaultDict(
+        lambda n: ToolComponent(
+            name=n, semantic_version=get_package_version(n), rules=[]
+        )
+    )
+
+    detector_index_mapping: Dict[str, int] = {}
+    for command in detectors:
+        descriptor = ReportingDescriptor(
+            id=command.name,
+            short_description={"text": cleandoc(command.help or "")},
+        )
+
+        source = run_detect.loaded_from_plugins[command.name]
+        # try to infer package name from entry point module name
+        if (
+            isinstance(source, str)
+            and source in source in distributions
+            and len(distributions[source]) == 1
+        ):
+            # loaded from plugin
+            package_name = distributions[source][0]
+            detector_index_mapping[command.name] = len(extensions[package_name].rules)
+            extensions[package_name].rules.append(descriptor)
+        else:
+            # loaded from local path
+            detector_index_mapping[command.name] = len(driver.rules)
+            driver.rules.append(descriptor)
+
+    extensions_list = list(extensions.values())
+    extensions_index_mapping = {e.name: i for i, e in enumerate(extensions_list)}
+
+    impact_to_level = {
+        DetectionImpact.HIGH: "error",
+        DetectionImpact.MEDIUM: "error",
+        DetectionImpact.LOW: "error",
+        DetectionImpact.WARNING: "warning",
+        DetectionImpact.INFO: "note",
+    }
+
+    results = []
+    for detector_name, r in detections.items():
+        for result in r:
+            (
+                start_line,
+                start_col,
+            ) = result.detection.ir_node.source_unit.get_line_col_from_byte_offset(
+                result.detection.ir_node.byte_location[0]
+            )
+            (
+                end_line,
+                end_col,
+            ) = result.detection.ir_node.source_unit.get_line_col_from_byte_offset(
+                result.detection.ir_node.byte_location[1]
+            )
+            rule = ReportingDescriptorReference(
+                id=detector_name,
+                index=detector_index_mapping[detector_name],
+            )
+
+            source = run_detect.loaded_from_plugins[detector_name]
+            if (
+                isinstance(source, str)
+                and source in distributions
+                and len(distributions[source]) == 1
+            ):
+                package_name = distributions[source][0]
+                rule.tool_component = ToolComponentReference(
+                    name=package_name,
+                    index=extensions_index_mapping[package_name],
+                )
+
+            results.append(
+                Result(
+                    rule=rule,
+                    level=impact_to_level[result.impact],
+                    message={"text": result.detection.message},
+                    locations=[
+                        Location(
+                            physical_location=PhysicalLocation(
+                                artifact_location=ArtifactLocation(
+                                    uri=f"file://{result.detection.ir_node.file}",
+                                ),
+                                region=Region(
+                                    start_line=start_line,
+                                    start_column=start_col,
+                                    end_line=end_line,
+                                    end_column=end_col,
+                                ),
+                            )
+                        )
+                    ],
+                )
+            )
+
+    return SarifLog(
+        schema_uri="https://json.schemastore.org/sarif-2.1.0.json",
+        version="2.1.0",
+        runs=[
+            Run(
+                tool=Tool(
+                    driver=driver,
+                    extensions=extensions_list,
+                ),
+                invocations=[
+                    Invocation(
+                        execution_successful=True,
+                        working_directory=ArtifactLocation(
+                            uri=f"file://{os.getcwd()}",
+                        ),
+                    ),
+                ],
+                results=results,
+            ),
+        ],
+    )

woke/lsp/lsp_compiler.py

@@ -1159,7 +1159,7 @@ def __run_detectors(
         logging_handler = LspLoggingHandler(logging_buffer)
 
         if self.__config.lsp.detectors.enable:
-            detections, detector_exceptions = detect(
+            _, detections, detector_exceptions = detect(
                 detector_names,
                 self.last_build,
                 self.last_build_info,