preparing rule detector

docs/detectors.md

@@ -88,6 +88,7 @@ List of detectors:
 * [New Value](detectors/new_value.md): Detect new values in the variables in the logs.
 * [Combo Detector](detectors/combo.md): Detect new combination of variables in the logs.
 * [New Event](detectors/new_event.md): Detect new events in the variables in the logs.
+* [Rule Based](detectors/rule_based.md): Detect anomalies based in a set of rules.
 
 ## Configuration
 

docs/detectors/rule_based.md

@@ -0,0 +1,69 @@
+# Rule-based Detector
+
+The Rule-based Detector raises alerts based on a configurable set of rules.
+
+|            | Schema                     | Description        |
+|------------|----------------------------|--------------------|
+| **Input**  | [ParserSchema](../schemas.md) | Structured log  |
+| **Output** | [DetectorSchema](../schemas.md) | Alert / finding |
+
+## Description
+
+The detector analyzes parsed logs one by one and checks which rules are triggered. When alerts are produced, the triggered rules and their messages are recorded in the `alertsObtain` field of the output schema. The `score` field contains the number of rules that triggered.
+
+### Available rules
+
+| Rule name | Description | Requires arguments | Enabled by default |
+|---|---|---:|:---:|
+| **R001 - TemplateNotFound** | Check whether the parser assigned a template to the log | No | Yes |
+| **R002 - SpecificKeyword** | Check for one or more user-specified keywords in the log content | list of words | No |
+| **R003 - CheckForExceptions** | Check for words commonly associated with exceptions or failures | No | Yes |
+| **R004 - ErrorLevelFound** | If a Level field exists, check whether it indicates an error level | No | No |
+
+Notes on table columns:
+
+- **Rule name**: Identifier used in configuration.
+- **Description**: What the rule checks.
+- **Requires arguments**: Whether the rule needs additional arguments.
+- **Enabled by default**: Whether the rule is active when not explicitly overridden.
+
+## Configuration example
+
+```yaml
+detectors:
+  RuleDetector:
+    method_type: rule_detector
+    auto_config: False
+    params:
+      rules:
+        - rule: "R001 - TemplateNotFound"
+        - rule: "R002 - SpecificKeyword"
+          args:
+            - "critical"
+            - "anomaly"
+```
+
+## Example usage
+
+```python
+import detectmatelibrary.detectors.rule_detector as rd
+from detectmatelibrary import schemas
+
+rule_detector = rd.RuleDetector()
+
+parser_data = schemas.ParserSchema({
+    "parserType": "test",
+    "EventID": 1,
+    "template": "test template",
+    "variables": ["var1"],
+    "logID": "1",
+    "parsedLogID": "1",
+    "parserID": "test_parser",
+    "log": "test log message",
+    "logFormatVariables": {"timestamp": "123456"}
+})
+
+alert = rule_detector.process(parser_data)
+```
+
+Go back [Index](../index.md)

mkdocs.yml

@@ -24,6 +24,8 @@ nav:
     - Random Detector: detectors/random_detector.md
     - New Value: detectors/new_value.md
     - Combo Detector: detectors/combo.md
+    - New Event: detectors/new_event.md
+    - Rule Based: detectors/rule_based.md
   - Auxiliar:
     - Persistency: auxiliar/persistency.md
     - Data buffer: auxiliar/input_buffer.md

src/detectmatelibrary/detectors/rule_detector.py

@@ -0,0 +1,106 @@
+"""Copy and paste to create new rules.
+
+def template_rule(
+    input_: schemas.ParserSchema, *args: list[Any]
+) -> tuple[bool, str]:
+
+    raise_alert = False  # Add here rule
+    message = ""  # Rule message
+    return raise_alert, message
+"""
+from detectmatelibrary.common.detector import CoreDetector, CoreDetectorConfig
+
+from detectmatelibrary.utils.data_buffer import BufferMode
+
+from detectmatelibrary import schemas
+
+from typing import List, Any
+
+
+def template_not_found(input_: schemas.ParserSchema, *args: list[Any]) -> tuple[bool, str]:
+    raise_alert = input_["EventID"] == -1
+    message = "Template was not found by the parser"
+    return raise_alert, message
+
+
+def find_keyword(input_: schemas.ParserSchema, args: list[str]) -> tuple[bool, str]:
+    log: str = input_["log"]
+    log = log.lower()
+
+    for k in args:
+        if k in log:
+            return True, f"Found word '{k}' in the logs"
+    return False, ""
+
+
+def exceptions(input_: schemas.ParserSchema, *args: list[Any]) -> tuple[bool, str]:
+    return find_keyword(input_, ["exception", "fail", "error", "raise"])
+
+
+def error_log(input_: schemas.ParserSchema, *args: list[Any]) -> tuple[bool, str]:
+    vars_: dict[str, str] = input_["logFormatVariables"]
+    raise_alert = "Level" in vars_ and "error" == vars_["Level"].lower()
+    message = "Error found"
+    return raise_alert, message
+
+
+rules = {
+    "R001 - TemplateNotFound": template_not_found,
+    "R002 - SpecificKeyword": find_keyword,
+    "R003 - CheckForExceptions": exceptions,
+    "R004 - ErrorLevelFound": error_log,
+}
+
+
+class RuleNotFound(Exception):
+    def __init__(self, rule: str) -> None:
+        super().__init__(f"Rule -> ([{rule}]) not found")
+
+
+class RuleDetectorConfig(CoreDetectorConfig):
+    method_type: str = "rule_detector"
+    rules: list[dict[str, list[str] | str]] = [
+        {"rule": "R001 - TemplateNotFound"},
+        {"rule": "R003 - CheckForExceptions"},
+        {"rule": "R004 - ErrorLevelFound"},
+    ]
+
+
+class RuleDetector(CoreDetector):
+    def __init__(
+        self,
+        name: str = "RuleDetector",
+        config: RuleDetectorConfig | dict[str, Any] = RuleDetectorConfig()
+    ) -> None:
+
+        if isinstance(config, dict):
+            config = RuleDetectorConfig.from_dict(config, name)
+        super().__init__(name=name, buffer_mode=BufferMode.NO_BUF, config=config)
+        self.config: RuleDetectorConfig
+
+        for rule in self.config.rules:
+            if rule["rule"] not in rules:
+                raise RuleNotFound(rule)
+
+    def detect(
+        self,
+        input_: List[schemas.ParserSchema] | schemas.ParserSchema,
+        output_: schemas.DetectorSchema
+    ) -> bool:
+
+        anomaly = False
+        output_["score"] = 0
+
+        for rule in self.config.rules:
+            if "args" in rule:
+                alert, msg = rules[rule["rule"]](input_, rule["args"])  # type: ignore
+            else:
+                alert, msg = rules[rule["rule"]](input_)  # type: ignore
+
+            if alert:
+                output_["alertsObtain"][rule["rule"]] = msg
+                output_["score"] += 1
+
+            anomaly = anomaly or alert
+
+        return anomaly

tests/test_detectors/test_rule_detector.py

@@ -0,0 +1,180 @@
+import detectmatelibrary.detectors.rule_detector as rd
+import detectmatelibrary.schemas as schemas
+
+import pytest
+
+
+class TestCaseRules:
+    def test_template_not_found(self) -> None:
+        parsed_log = schemas.ParserSchema({"EventID": -1})
+
+        alert, msg = rd.template_not_found(parsed_log)
+        assert alert
+        assert "Template was not found by the parser" == msg
+
+        parsed_log = schemas.ParserSchema({"EventID": 2})
+        alert, _ = rd.template_not_found(parsed_log)
+        assert not alert
+
+    def test_find_keyword(self) -> None:
+        keywords = ["hello", "ciao"]
+
+        parsed_log = schemas.ParserSchema({"log": "hello world"})
+        alert, msg = rd.find_keyword(parsed_log, keywords)
+        assert alert
+        assert "Found word 'hello' in the logs" == msg
+
+        parsed_log = schemas.ParserSchema({"log": "ciao world"})
+        alert, msg = rd.find_keyword(parsed_log, keywords)
+        assert alert
+        assert "Found word 'ciao' in the logs" == msg
+
+        parsed_log = schemas.ParserSchema({"log": "world"})
+        alert, _ = rd.find_keyword(parsed_log, keywords)
+        assert not alert
+
+    def test_find_exceptions(self) -> None:
+        parsed_log = schemas.ParserSchema({"log": "hello Exception"})
+        alert, msg = rd.exceptions(parsed_log)
+        assert alert
+        assert "Found word 'exception' in the logs" == msg
+
+        parsed_log = schemas.ParserSchema({"log": "world"})
+        alert, _ = rd.exceptions(parsed_log)
+        assert not alert
+
+    def test_error_log(self) -> None:
+        parsed_log = schemas.ParserSchema(
+            {"logFormatVariables": {"Level": "Error"}}
+        )
+        alert, msg = rd.error_log(parsed_log)
+        assert alert
+        assert "Error found" == msg
+
+        parsed_log = schemas.ParserSchema(
+            {"logFormatVariables": {"Level": "Info"}}
+        )
+        alert, _ = rd.error_log(parsed_log)
+        assert not alert
+
+        parsed_log = schemas.ParserSchema({"log": "hello world"})
+        alert, _ = rd.error_log(parsed_log)
+        assert not alert
+
+
+class TestCaseRuleDetector:
+    def test_run_all_rules(self) -> None:
+        rule_detector = rd.RuleDetector()
+
+        assert rule_detector.process(schemas.ParserSchema()) is None
+
+        alert1 = rule_detector.process(schemas.ParserSchema(
+            {
+                "EventID": -1,
+                "log": "fail exception",
+                "logFormatVariables": {"Level": "Error"}
+            }
+        ))
+        assert alert1 is not None
+        assert alert1["alertsObtain"] == {
+            "R001 - TemplateNotFound": "Template was not found by the parser",
+            "R003 - CheckForExceptions": "Found word 'exception' in the logs",
+            "R004 - ErrorLevelFound": "Error found",
+        }
+        assert alert1["score"] == 3
+
+    def test_all_rules_serialize(self) -> None:
+        rule_detector = rd.RuleDetector()
+
+        assert rule_detector.process(schemas.ParserSchema().serialize()) is None
+
+        alert1 = rule_detector.process(schemas.ParserSchema(
+            {
+                "EventID": -1,
+                "log": "fail exception",
+                "logFormatVariables": {"Level": "Error"}
+            }
+        ).serialize())
+        assert alert1 is not None
+
+        (alert := schemas.DetectorSchema()).deserialize(alert1)
+        assert alert["alertsObtain"] == {
+            "R001 - TemplateNotFound": "Template was not found by the parser",
+            "R003 - CheckForExceptions": "Found word 'exception' in the logs",
+            "R004 - ErrorLevelFound": "Error found",
+        }
+        assert alert["score"] == 3
+
+    def test_run_single_rule(self) -> None:
+        rule_detector = rd.RuleDetector(
+            name="RuleDetector",
+            config={
+                "detectors": {
+                    "RuleDetector": {
+                        "method_type": "rule_detector",
+                        "auto_config": False,
+                        "params": {
+                            "rules": [
+                                {"rule": "R001 - TemplateNotFound"}
+                            ]
+                        }
+                    }
+                }
+            }
+        )
+
+        assert rule_detector.process(schemas.ParserSchema()) is None
+
+        alert1 = rule_detector.process(schemas.ParserSchema(
+            {
+                "EventID": -1,
+                "log": "fail exception",
+                "logFormatVariables": {"Level": "Error"}
+            }
+        ))
+        assert alert1 is not None
+        assert alert1["alertsObtain"] == {
+            "R001 - TemplateNotFound": "Template was not found by the parser",
+        }
+        assert alert1["score"] == 1
+
+    def test_run_single_rule_with_args(self) -> None:
+        rule_detector = rd.RuleDetector(
+            name="RuleDetector",
+            config={
+                "detectors": {
+                    "RuleDetector": {
+                        "method_type": "rule_detector",
+                        "auto_config": False,
+                        "params": {
+                            "rules": [
+                                {"rule": "R002 - SpecificKeyword", "args": ["hi", "kenobi"]}
+                            ]
+                        }
+                    }
+                }
+            }
+        )
+
+        assert rule_detector.process(schemas.ParserSchema({"log": "ciao"})) is None
+        assert rule_detector.process(schemas.ParserSchema({"log": "hi"})) is not None
+
+    def test_rule_not_found(self) -> None:
+
+        with pytest.raises(rd.RuleNotFound):
+            rd.RuleDetector(
+                name="RuleDetector",
+                config={
+                    "detectors": {
+                        "RuleDetector": {
+                            "method_type": "rule_detector",
+                            "auto_config": False,
+                            "params": {
+                                "rules": [
+                                    {"rule": "Rule made up"}
+                                ]
+                            }
+                        }
+                    }
+                }
+            )