diff --git a/src/detectmatelibrary/detectors/new_sequence_detector.py b/src/detectmatelibrary/detectors/new_sequence_detector.py new file mode 100644 index 00000000..55740fa5 --- /dev/null +++ b/src/detectmatelibrary/detectors/new_sequence_detector.py @@ -0,0 +1,92 @@ +from collections import deque +from typing import Sequence + +from detectmatelibrary.common._config._compile import generate_detector_config +from detectmatelibrary.common.detector import CoreDetectorConfig, CoreDetector +from detectmatelibrary.utils import persistency +from detectmatelibrary.utils.data_buffer import BufferMode +from detectmatelibrary.schemas import ParserSchema, DetectorSchema + +_SEQUENCE_SEPARATOR = "\x1f" + + +def _encode_sequence(sequence: Sequence[int]) -> str: + return _SEQUENCE_SEPARATOR.join(str(event_id) for event_id in sequence) + + +class NewSequenceDetectorConfig(CoreDetectorConfig): + method_type: str = "new_sequence_detector" + max_sequence_length: int = 3 + + +class NewSequenceDetector(CoreDetector): + def __init__( + self, + name: str = "NewSequenceDetector", + config: NewSequenceDetectorConfig = NewSequenceDetectorConfig() + ) -> None: + if isinstance(config, dict): + config = NewSequenceDetectorConfig.from_dict(config, name) + + super().__init__(name=name, buffer_mode=BufferMode.NO_BUF, config=config) + self.config: NewSequenceDetectorConfig + self._window: deque[int] = deque(maxlen=self.config.max_sequence_length) + self.persistency = persistency.EventPersistency( + event_data_class=persistency.EventStabilityTracker, + ) + self.auto_conf_persistency = persistency.EventPersistency( + event_data_class=persistency.EventStabilityTracker + ) + self._register_persistency(self.persistency) + + def train(self, input_: ParserSchema) -> None: # type: ignore + """Train the detector by learning EventID sequences from the input + data.""" + self._window.append(input_["EventID"]) + if len(self._window) < self.config.max_sequence_length: + return + self.persistency.ingest_event( + event_id=_encode_sequence(self._window), + event_template=input_["template"] + ) + + def detect(self, input_: ParserSchema, output_: DetectorSchema) -> bool: # type: ignore + self._window.append(input_["EventID"]) + if len(self._window) < self.config.max_sequence_length: + return False + + if _encode_sequence(self._window) not in self.persistency.get_events_seen(): + output_["score"] = 1.0 + output_["description"] = f"{self.name} detects unknown EventID sequences as anomalies." + output_["alertsObtain"].update({ + f"Sequence {tuple(self._window)}": f"Unknown sequence: {tuple(self._window)}" + }) + return True + return False + + def configure(self, input_: ParserSchema) -> None: # type: ignore + self.auto_conf_persistency.ingest_event( + event_id=input_["EventID"], + event_template=input_["template"] + ) + + def set_configuration(self) -> None: + old_persist = self.config.persist + config_dict = generate_detector_config( + variable_selection={}, + detector_name=self.name, + method_type=self.config.method_type, + max_sequence_length=self.config.max_sequence_length, + ) + self.config = NewSequenceDetectorConfig.from_dict(config_dict, self.name) + self.config.persist = old_persist + self._window = deque(self._window, maxlen=self.config.max_sequence_length) + + def reset_window(self) -> None: + self._window.clear() + + def get_known_sequences(self) -> set[tuple[str, ...]]: + return { + tuple(str(encoded).split(_SEQUENCE_SEPARATOR)) + for encoded in self.persistency.get_events_seen() + } diff --git a/tests/test_detectors/test_new_sequence_detector.py b/tests/test_detectors/test_new_sequence_detector.py new file mode 100644 index 00000000..4be6c7fb --- /dev/null +++ b/tests/test_detectors/test_new_sequence_detector.py @@ -0,0 +1,193 @@ +"""Tests for NewSequenceDetector class. + +This module tests the NewSequenceDetector implementation including: +- Initialization and configuration +- Training functionality to learn known EventID sequences +- Detection logic for unknown sequences +- Window handling (reset_window) and end-to-end regression on audit.log +""" + +from detectmatelibrary.detectors.new_sequence_detector import NewSequenceDetector, \ + NewSequenceDetectorConfig, BufferMode +from detectmatelibrary.parsers.template_matcher import MatcherParser +from detectmatelibrary.helper.from_to import From +import detectmatelibrary.schemas as schemas +from detectmatelibrary.utils.aux import time_test_mode +from tests.test_data import AUDIT_LOG, AUDIT_TEMPLATES, TRAIN_UNTIL + +# Set time test mode for consistent timestamps +time_test_mode() + + +config = { + "detectors": { + "CustomInit": { + "method_type": "new_sequence_detector", + "auto_config": False, + "params": { + "max_sequence_length": 2 + } + }, + "MultipleDetector": { + "method_type": "new_sequence_detector", + "auto_config": False, + "params": { + "max_sequence_length": 2 + } + } + } +} + + +def _make_schema(event_id, template="test template", log_id="1"): + return schemas.ParserSchema({ + "parserType": "test", + "EventID": event_id, + "template": template, + "variables": ["adsasd", "asdasd"], + "logID": log_id, + "parsedLogID": log_id, + "parserID": "test_parser", + "log": "test log message", + "logFormatVariables": {"level": "INFO"} + }) + + +class TestNewSequenceDetectorInitialization: + """Test NewSequenceDetector initialization and configuration.""" + + def test_default_initialization(self): + detector = NewSequenceDetector() + + assert detector.name == "NewSequenceDetector" + assert hasattr(detector, "config") + assert detector.data_buffer.mode == BufferMode.NO_BUF + assert detector.input_schema == schemas.ParserSchema + assert detector.output_schema == schemas.DetectorSchema + assert hasattr(detector, "persistency") + assert detector.config.max_sequence_length == 3 + + def test_custom_config_initialization(self): + detector = NewSequenceDetector(name="CustomInit", config=config) + + assert detector.name == "CustomInit" + assert detector.config.max_sequence_length == 2 + assert hasattr(detector, "persistency") + assert isinstance(detector.persistency.events_data, dict) + + +class TestNewSequenceDetectorTraining: + """Test NewSequenceDetector training functionality.""" + + def test_train_learns_sequence(self): + detector = NewSequenceDetector(config=config, name="MultipleDetector") + + for event_id in [1, 2, 3]: + detector.train(_make_schema(event_id)) + + # max_sequence_length=2 -> sequences (1, 2) and (2, 3) are learned + assert detector.get_known_sequences() == {("1", "2"), ("2", "3")} + + def test_train_below_window_length_learns_nothing(self): + detector = NewSequenceDetector(config=config, name="MultipleDetector") + + # Only one event seen so far, window not yet full (max_sequence_length=2) + detector.train(_make_schema(1)) + + assert detector.get_known_sequences() == set() + + +class TestNewSequenceDetectorDetection: + """Test NewSequenceDetector detection functionality.""" + + def test_detect_known_sequence_no_alert(self): + detector = NewSequenceDetector(config=config, name="MultipleDetector") + + # Repeating the cycle 1 -> 2 -> 3 teaches sequences (1,2), (2,3), (3,1) + for event_id in [1, 2, 3, 1, 2, 3]: + detector.train(_make_schema(event_id)) + # Window now holds the tail (2, 3); appending 1 reproduces the known (3, 1) + output = schemas.DetectorSchema() + result = detector.detect(_make_schema(1, log_id="7"), output) + + assert not result + assert output.score == 0.0 + + def test_detect_unknown_sequence_alert(self): + detector = NewSequenceDetector(config=config, name="MultipleDetector") + + for event_id in [1, 2, 3, 1, 2, 3]: + detector.train(_make_schema(event_id)) + + # (3, 9) was never seen during training + output = schemas.DetectorSchema() + result = detector.detect(_make_schema(9, log_id="7"), output) + + assert result + assert output.score == 1.0 + assert any("Sequence" in key for key in output["alertsObtain"]) + + def test_detect_below_window_length_no_alert(self): + detector = NewSequenceDetector(config=config, name="MultipleDetector") + + # First event with an empty detector: window not yet full, can't be an anomaly + output = schemas.DetectorSchema() + result = detector.detect(_make_schema(1), output) + + assert not result + assert output.score == 0.0 + + +class TestNewSequenceDetectorWindow: + """Test sliding-window behaviour.""" + + def test_reset_window_clears_state(self): + detector = NewSequenceDetector(config=config, name="MultipleDetector") + + detector.train(_make_schema(1)) + assert len(detector._window) == 1 + + detector.reset_window() + assert len(detector._window) == 0 + + +_PARSER_CONFIG = { + "parsers": { + "MatcherParser": { + "method_type": "matcher_parser", + "auto_config": False, + "log_format": "type= msg=audit(