Plugin capabilities

sigma/plugins.py

@@ -280,6 +280,15 @@ class SigmaPluginState(EnumLowercaseStringMixin, Enum):
     ORPHANED = auto()
 
 
+class SigmaPluginCapability(EnumLowercaseStringMixin, Enum):
+    """Sigma plugin capabilities that describe optional plugin functionality."""
+
+    EVENT_COUNT_CORRELATION_CONVERSION = auto()
+    VALUE_COUNT_CORRELATION_CONVERSION = auto()
+    TEMPORAL_CORRELATION_CONVERSION = auto()
+    ORDERED_TEMPORAL_CORRELATION_CONVERSION = auto()
+
+
 @dataclass
 class SigmaPlugin:
     """Sigma plugin description corresponding to https://github.com/SigmaHQ/pySigma-plugin-directory#format"""
@@ -293,6 +302,7 @@ class SigmaPlugin:
     report_issue_url: str
     state: SigmaPluginState
     pysigma_version: Specifier
+    capabilities: Set[SigmaPluginCapability] = field(default_factory=set)
 
     @classmethod
     def from_dict(cls, d: Dict) -> "SigmaPlugin":
@@ -303,6 +313,9 @@ def from_dict(cls, d: Dict) -> "SigmaPlugin":
         kwargs["pysigma_version"] = Specifier(kwargs["pysigma_version"])
         kwargs["type"] = SigmaPluginType[kwargs["type"].upper()]
         kwargs["state"] = SigmaPluginState[kwargs["state"].upper()]
+        kwargs["capabilities"] = {
+            SigmaPluginCapability[c.upper()] for c in kwargs.get("capabilities", [])
+        }
 
         return cls(**kwargs)
 
@@ -316,6 +329,10 @@ def is_compatible(self) -> Optional[bool]:
         except importlib.metadata.PackageNotFoundError:
             return None
 
+    def has_capability(self, capability: SigmaPluginCapability) -> bool:
+        """Checks if the plugin has the specified capability."""
+        return capability in self.capabilities
+
     def install(self):
         """Install plugin with pip."""
         if sys.prefix == sys.base_prefix:  # not in a virtual environment

tests/test_plugins.py

@@ -3,6 +3,7 @@
 from sigma.pipelines.test.pipeline import another_test_pipeline, YetAnotherTestPipeline
 from sigma.plugins import (
     SigmaPlugin,
+    SigmaPluginCapability,
     SigmaPluginDirectory,
     SigmaPluginState,
     SigmaPluginType,
@@ -92,6 +93,10 @@ def sigma_plugin_dict():
         "report_issue_url": "https://github.com/SigmaHQ/pySigma-backend-test/issues/new",
         "state": "testing",
         "pysigma_version": ">=0.9.0",
+        "capabilities": [
+            "event_count_correlation_conversion",
+            "value_count_correlation_conversion",
+        ],
     }
 
 
@@ -107,13 +112,23 @@ def sigma_plugin():
         report_issue_url="https://github.com/SigmaHQ/pySigma-backend-test/issues/new",
         state=SigmaPluginState.TESTING,
         pysigma_version=Specifier(">=0.9.0"),
+        capabilities={
+            SigmaPluginCapability.EVENT_COUNT_CORRELATION_CONVERSION,
+            SigmaPluginCapability.VALUE_COUNT_CORRELATION_CONVERSION,
+        },
     )
 
 
 def test_sigma_plugin_from_dict(sigma_plugin, sigma_plugin_dict):
     assert SigmaPlugin.from_dict(sigma_plugin_dict) == sigma_plugin
 
 
+def test_sigma_plugin_from_dict_without_capabilities(monkeypatch, sigma_plugin, sigma_plugin_dict):
+    monkeypatch.delitem(sigma_plugin_dict, "capabilities")
+    monkeypatch.setattr(sigma_plugin, "capabilities", set())
+    assert SigmaPlugin.from_dict(sigma_plugin_dict) == sigma_plugin
+
+
 def test_sigma_plugin_version_compatible(sigma_plugin):
     pysigma_version = importlib.metadata.version("pysigma")
     sigma_plugin.pysigma_version = Specifier(
@@ -136,6 +151,12 @@ def version_replacement(m):
     assert sigma_plugin.is_compatible() is None
 
 
+def test_sigma_plugin_has_capability(sigma_plugin):
+    assert sigma_plugin.has_capability(SigmaPluginCapability.EVENT_COUNT_CORRELATION_CONVERSION)
+    assert sigma_plugin.has_capability(SigmaPluginCapability.VALUE_COUNT_CORRELATION_CONVERSION)
+    assert not sigma_plugin.has_capability(SigmaPluginCapability.TEMPORAL_CORRELATION_CONVERSION)
+
+
 def check_module(name: str) -> bool:
     # This was the preferred way to test module existence, but it didn't worked in GitHub Actions:
     # return bool(importlib.util.find_spec(name))