feat(otlp): Sigma rules

interfaces/otlp/CHANGELOG.md

@@ -46,3 +46,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Separated the package into 2 private modules: _otlp and _rules
 - Enable user-defined rule labels
+
+## [0.6.0]
+
+### Added
+
+- Sigma rules in databag via the `RuleStore` API

interfaces/otlp/pyproject.toml

@@ -16,7 +16,7 @@ classifiers = [
 dynamic = ["version"]
 dependencies = [
     # "ops",
-    "cosl>=1.6.1",
+    "cosl @ git+https://github.com/canonical/cos-lib@feat/sigma-rules-2",
     "requests",
 ]
 
@@ -44,6 +44,9 @@ integration = [  # installed for `just integration otlp`
 requires = ["hatchling"]
 build-backend = "hatchling.build"
 
+[tool.hatch.metadata]
+allow-direct-references = true
+
 [tool.hatch.build.targets.wheel]
 packages = ["src/charmlibs"]
 

interfaces/otlp/src/charmlibs/interfaces/otlp/_otlp.py

@@ -218,6 +218,7 @@ def publish(self):
             'rules': {
                 'logql': self._rules.logql.as_dict(),
                 'promql': self._rules.promql.as_dict(),
+                'sigma': self._rules.sigma.as_dict(),
             },
             'metadata': self._topology.as_dict(),
         })
@@ -337,6 +338,9 @@ def rules(self) -> dict[int, RuleStore]:
                 rules.logql.add(logql_result.rules)
             if promql_result.rules and not promql_result.errmsg:
                 rules.promql.add(promql_result.rules)
+            # Sigma rules: inject topology labels (no expression rewriting needed)
+            if sigma_dict := requirer.rules.sigma:
+                rules.sigma.add(sigma_dict)
             for errmsg in [logql_result.errmsg, promql_result.errmsg]:
                 if errmsg and self._charm.unit.is_leader():
                     relation.data[self._charm.app]['event'] = json.dumps({'errors': errmsg})

interfaces/otlp/src/charmlibs/interfaces/otlp/_rules.py

@@ -21,8 +21,19 @@
 from pathlib import Path
 
 from cosl.juju_topology import JujuTopology
-from cosl.rules import HOST_METRICS_MISSING_RULE_NAME, CosTool, Rules, generic_alert_groups
-from cosl.types import OfficialRuleFileFormat, SingleRuleFormat
+from cosl.rules import (
+    HOST_METRICS_MISSING_RULE_NAME,
+    CosTool,
+    Rules,
+    SigmaRules,
+    generic_alert_groups,
+)
+from cosl.types import (
+    OfficialRuleFileFormat,
+    SigmaRuleFileFormat,
+    SigmaRuleFormat,
+    SingleRuleFormat,
+)
 from ops import CharmBase
 from pydantic import BaseModel, Field
 
@@ -43,10 +54,12 @@ class RuleStore:
     topology: JujuTopology
     logql: Rules = field(init=False)
     promql: Rules = field(init=False)
+    sigma: SigmaRules = field(init=False)
 
     def __post_init__(self):
         self.logql = Rules(query_type='logql', topology=self.topology)
         self.promql = Rules(query_type='promql', topology=self.topology)
+        self.sigma = SigmaRules(topology=self.topology)
 
     def add_logql(
         self,
@@ -110,12 +123,36 @@ def add_promql_path(self, dir_path: str | Path, *, recursive: bool = False) -> '
         self.promql.add_path(dir_path, recursive=recursive)
         return self
 
+    def add_sigma(self, rule_dict: SigmaRuleFileFormat | SigmaRuleFormat) -> 'RuleStore':
+        """Add rules from dict to the existing Sigma ruleset.
+
+        Args:
+            rule_dict: a single sigma rule dict or a ``{"rules": [...]}`` collection.
+        """
+        self.sigma.add(rule_dict)
+        return self
+
+    def add_sigma_path(self, dir_path: str | Path, *, recursive: bool = False) -> 'RuleStore':
+        """Add Sigma rules from a dir path.
+
+        All rules from files are aggregated into a single collection.
+        Topology labels are injected into each rule.
+
+        Args:
+            dir_path: either a rules file or a dir of rules files.
+            recursive: whether to read files recursively or not (no impact if `path` is a file).
+        """
+        self.sigma.add_path(dir_path, recursive=recursive)
+        return self
+
     def combine(self, other: 'RuleStore') -> 'RuleStore':
         """Combine rules from another RuleStore with this RuleStore."""
         if other_logql := other.logql.as_dict():
             self.logql.add(other_logql)
         if other_promql := other.promql.as_dict():
             self.promql.add(other_promql)
+        if other_sigma := other.sigma.as_dict():
+            self.sigma.add(other_sigma)
         return self
 
 
@@ -132,6 +169,10 @@ class _RulesModel(BaseModel):
         'OfficialRuleFileFormat from cos-lib.',
         default_factory=OfficialRuleFileFormat,
     )
+    sigma: SigmaRuleFileFormat = Field(
+        description='Sigma detection rules, following the SigmaRuleFileFormat from cos-lib.',
+        default_factory=SigmaRuleFileFormat,
+    )
 
 
 def duplicate_rules_per_unit(
@@ -235,4 +276,6 @@ def inject_extra_labels_into_rules(
         for group in rule_groups.get('groups', []):
             for rule in group.get('rules', []):
                 rule.setdefault('labels', {}).update(extra_alert_labels)
+    for sigma_rule in rules_copy.sigma.as_dict().get('rules', []):
+        sigma_rule.setdefault('labels', {}).update(extra_alert_labels)
     return rules_copy

interfaces/otlp/src/charmlibs/interfaces/otlp/_version.py

@@ -12,4 +12,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-__version__ = '0.5.0'
+__version__ = '0.6.0'

interfaces/otlp/tests/unit/conftest.py

@@ -25,7 +25,12 @@
 import ops
 import pytest
 from cosl.juju_topology import JujuTopology
-from cosl.types import AlertingRuleFormat, OfficialRuleFileFormat, RecordingRuleFormat
+from cosl.types import (
+    AlertingRuleFormat,
+    OfficialRuleFileFormat,
+    RecordingRuleFormat,
+    SigmaRuleFormat,
+)
 from ops import testing
 from ops.charm import CharmBase
 
@@ -74,6 +79,26 @@
         },
     ]
 )
+SINGLE_SIGMA_RULE: SigmaRuleFormat = {
+    'title': 'Failed SSH Login Attempt',
+    'id': '5f3a4e20-1b2c-4d5e-9f8a-7b6c3d4e5f6a',
+    'status': 'experimental',
+    'logsource': {'category': 'authentication', 'product': 'linux', 'service': 'sshd'},
+    'detection': {'selection': {'event_type': 'authentication_failure'}, 'condition': 'selection'},
+    'level': 'medium',
+}
+SIGMA_COLLECTION: dict = {
+    'rules': [
+        SINGLE_SIGMA_RULE,
+        {
+            'title': 'High CPU Usage',
+            'logsource': {'category': 'process_monitoring', 'product': 'linux'},
+            'detection': {'selection': {'metric': 'cpu_percent'}, 'condition': 'selection'},
+            'level': 'high',
+        },
+    ]
+}
+
 ALL_PROTOCOLS: list[Literal['grpc', 'http']] = ['grpc', 'http']
 ALL_TELEMETRIES: list[Literal['logs', 'metrics', 'traces']] = ['logs', 'metrics', 'traces']
 
@@ -102,6 +127,8 @@ def _publish_rules(self, _: ops.EventBase) -> None:
                 .add_promql(deepcopy(SINGLE_PROMQL_RECORD), group_name='test_promql_record')
                 .add_logql(deepcopy(OFFICIAL_LOGQL_RULES))
                 .add_promql(deepcopy(OFFICIAL_PROMQL_RULES))
+                .add_sigma(deepcopy(SINGLE_SIGMA_RULE))
+                .add_sigma(deepcopy(SIGMA_COLLECTION))
             )
         OtlpRequirer(
             self,