diff --git a/interfaces/otlp/src/charmlibs/interfaces/otlp/_otlp.py b/interfaces/otlp/src/charmlibs/interfaces/otlp/_otlp.py index 7d849f24b..d11d532d9 100644 --- a/interfaces/otlp/src/charmlibs/interfaces/otlp/_otlp.py +++ b/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}) diff --git a/interfaces/otlp/tests/unit/conftest.py b/interfaces/otlp/tests/unit/conftest.py index 2a94efdc3..49742a3c5 100644 --- a/interfaces/otlp/tests/unit/conftest.py +++ b/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, diff --git a/interfaces/otlp/tests/unit/test_rules.py b/interfaces/otlp/tests/unit/test_rules.py index d7b9e27d5..fed20bed5 100644 --- a/interfaces/otlp/tests/unit/test_rules.py +++ b/interfaces/otlp/tests/unit/test_rules.py @@ -26,10 +26,12 @@ from charmlibs.interfaces.otlp._rules import _RulesModel, duplicate_rules_per_unit from conftest import ( PEERS_ENDPOINT, + SIGMA_COLLECTION, SINGLE_LOGQL_ALERT, SINGLE_LOGQL_RECORD, SINGLE_PROMQL_ALERT, SINGLE_PROMQL_RECORD, + SINGLE_SIGMA_RULE, ) MODEL_NAME = 'foo-model' @@ -89,6 +91,41 @@ def test_rules_compression(otlp_requirer_ctx: testing.Context[ops.CharmBase]): assert set(_RulesModel.model_fields.keys()).issubset(decompressed.keys()) +def test_published_rules_are_byte_stable_across_hooks( + otlp_requirer_ctx: testing.Context[ops.CharmBase], +): + # GIVEN a requirer that publishes rules (including Sigma rules with topology tags) + state = State(relations=[Relation(SEND)], leader=True) + + # WHEN the publish event fires twice, as it would across two separate hooks + first = otlp_requirer_ctx.run(otlp_requirer_ctx.on.update_status(), state=state) + first_rules = next(iter(first.relations)).local_app_data.get('rules') + + second = otlp_requirer_ctx.run(otlp_requirer_ctx.on.update_status(), state=first) + second_rules = next(iter(second.relations)).local_app_data.get('rules') + + # THEN the serialized databag value is byte-identical, so Juju sees no change and + # does not fire a spurious relation-changed event. + assert first_rules is not None + assert first_rules == second_rules + + +def test_sigma_rule_id_is_preserved_in_published_databag( + otlp_requirer_ctx: testing.Context[ops.CharmBase], +): + # GIVEN a requirer publishing a Sigma rule that carries a fixed UUID + state = State(relations=[Relation(SEND)], leader=True) + + # WHEN the rules are published + state_out = otlp_requirer_ctx.run(otlp_requirer_ctx.on.update_status(), state=state) + rules = next(iter(state_out.relations)).local_app_data.get('rules') + sigma_rules = _decompress(json.loads(rules))['sigma']['rules'] + + # THEN the original UUID survives untouched (it is never regenerated per hook) + ids = {r.get('id') for r in sigma_rules} + assert SINGLE_SIGMA_RULE['id'] in ids + + @pytest.mark.parametrize('subordinate', [True, False]) def test_duplicate_rules_per_unit(subordinate: bool): # GIVEN the charm is (or is not) a subordinate @@ -423,3 +460,182 @@ def test_rulestore_combine_returns_self(): # THEN combine returns the target (self) for chaining assert result is target + + +# --- Sigma rules --- + + +def test_rulestore_add_single_sigma_rule(): + # GIVEN a RuleStore + store = _make_store() + + # WHEN a single sigma rule is added + store.add_sigma(SINGLE_SIGMA_RULE) + + # THEN the sigma collection contains 1 rule + sigma = store.sigma.as_dict() + assert len(sigma.get('rules', [])) == 1 + # AND topology labels are injected + labels = sigma['rules'][0].get('labels', {}) + assert labels['juju_model'] == MODEL_NAME + assert labels['juju_application'] == 'test-app' + + +def test_rulestore_add_sigma_collection(): + # GIVEN a RuleStore + store = _make_store() + + # WHEN a sigma collection is added + store.add_sigma(SIGMA_COLLECTION) + + # THEN the sigma collection contains 2 rules + sigma = store.sigma.as_dict() + assert len(sigma.get('rules', [])) == 2 + titles = {r['title'] for r in sigma['rules']} + assert titles == {'Failed SSH Login Attempt', 'High CPU Usage'} + + +def test_rulestore_add_sigma_returns_self(): + # GIVEN a RuleStore + store = _make_store() + + # WHEN add_sigma is called + result = store.add_sigma(SINGLE_SIGMA_RULE) + + # THEN it returns self for chaining + assert result is store + + +def test_rulestore_combine_sigma(): + # GIVEN a source RuleStore with sigma rules + source = _make_store().add_sigma(SINGLE_SIGMA_RULE) + # AND an empty target RuleStore + target = _make_store() + + # WHEN combined + target.combine(source) + + # THEN the target now contains the sigma rules + sigma = target.sigma.as_dict() + assert len(sigma.get('rules', [])) == 1 + assert sigma['rules'][0]['title'] == 'Failed SSH Login Attempt' + + +def test_rulestore_combine_empty_sigma_does_not_clear_target(): + # GIVEN a target RuleStore with sigma rules + target = _make_store().add_sigma(SINGLE_SIGMA_RULE) + # AND an empty source RuleStore + source = _make_store() + + # WHEN combined + target.combine(source) + + # THEN the target sigma rules are unchanged + assert len(target.sigma.as_dict().get('rules', [])) == 1 + + +def test_sigma_rules_in_databag(otlp_requirer_ctx: testing.Context[ops.CharmBase]): + # GIVEN a send-otlp relation + state = State(relations=[Relation(SEND)], leader=True, model=MODEL) + + # WHEN the update_status event is fired + state_out = otlp_requirer_ctx.run(otlp_requirer_ctx.on.update_status(), state=state) + for relation in list(state_out.relations): + # THEN the sigma rules appear in the compressed databag + decompressed = _decompress(relation.local_app_data.get('rules')) + assert decompressed + sigma_rules = decompressed.get('sigma', {}).get('rules', []) + # 1 single rule + 2 from collection = 3 + assert len(sigma_rules) == 3 + + +def test_sigma_topology_in_databag(otlp_requirer_ctx: testing.Context[ops.CharmBase]): + # GIVEN a send-otlp relation + state = State(relations=[Relation(SEND)], leader=True, model=MODEL) + + # WHEN the update_status event is fired + state_out = otlp_requirer_ctx.run(otlp_requirer_ctx.on.update_status(), state=state) + for relation in list(state_out.relations): + decompressed = _decompress(relation.local_app_data.get('rules')) + sigma_rules = decompressed.get('sigma', {}).get('rules', []) + for rule in sigma_rules: + # THEN each sigma rule has topology labels + labels = rule.get('labels', {}) + assert labels.get('juju_model') == MODEL_NAME + assert labels.get('juju_application') == 'otlp-requirer' + + +def test_provider_sigma_rules(otlp_provider_ctx: testing.Context[ops.CharmBase]): + # GIVEN a requirer offers sigma rules in the databag + rules = { + 'logql': {}, + 'promql': {}, + 'sigma': { + 'rules': [SINGLE_SIGMA_RULE], + }, + } + metadata = { + 'model': MODEL_NAME, + 'model_uuid': MODEL_UUID, + 'application': 'otlp-requirer', + 'charm_name': 'otlp-requirer', + 'unit': 'otlp-requirer/0', + } + receiver = Relation( + RECEIVE, remote_app_data={'rules': json.dumps(rules), 'metadata': json.dumps(metadata)} + ) + state = State(leader=True, relations=[receiver], model=MODEL) + with otlp_provider_ctx(otlp_provider_ctx.on.update_status(), state=state) as mgr: + # WHEN the provider aggregates the rules from the databag + rule_store = OtlpProvider(mgr.charm, RECEIVE).rules[receiver.id] + sigma = rule_store.sigma.as_dict() + + # THEN the sigma rules exist in the RuleStore + assert len(sigma.get('rules', [])) == 1 + rule = sigma['rules'][0] + assert rule['title'] == 'Failed SSH Login Attempt' + + # AND the rule has topology labels from the provider + labels = rule.get('labels', {}) + assert labels.get('juju_model') == MODEL_NAME + assert labels.get('juju_model_uuid') == MODEL_UUID + assert labels.get('juju_application') == 'otlp-provider' + + +@pytest.mark.parametrize( + 'otlp_requirer_ctx,expected_labels', + [ + pytest.param( + {'aggregator': True, 'extra_alert_labels': {'env': 'prod'}}, + {'env': 'prod'}, + id='single-label', + ), + pytest.param( + {'aggregator': True, 'extra_alert_labels': {'env': 'prod', 'team': 'core'}}, + {'env': 'prod', 'team': 'core'}, + id='multiple-labels', + ), + ], + indirect=['otlp_requirer_ctx'], +) +def test_extra_alert_labels_injected_into_sigma( + otlp_requirer_ctx: testing.Context[ops.CharmBase], expected_labels: dict[str, str] +): + # GIVEN a send-otlp relation with peers + peers = PeerRelation(endpoint=PEERS_ENDPOINT) + state = State(relations=[peers, Relation(SEND)], leader=True, model=MODEL) + + # WHEN the update_status event is fired + state_out = otlp_requirer_ctx.run(otlp_requirer_ctx.on.update_status(), state=state) + for relation in list(state_out.relations): + if relation.endpoint != SEND: + continue + decompressed = _decompress(relation.local_app_data.get('rules')) + sigma_rules = decompressed.get('sigma', {}).get('rules', []) + assert sigma_rules + + # THEN each sigma rule has the extra alert labels + for rule in sigma_rules: + labels = rule.get('labels', {}) + for k, v in expected_labels.items(): + assert labels.get(k) == v