Feature/domainator integration

config.yaml

@@ -196,6 +196,8 @@ pipeline:
       detector_class_name: "DomainatorDetector"
       model: domainator
       use_scaler: false
+      domain_expiration_hours: 1
+      expiry_check_interval_hours: 1
       checksum: a4aac4c585f1e614c3cf0d737e80b960c5de6e87b253f7cdd07125d9ce486476
       base_url: https://heibox.uni-heidelberg.de/d/0d5cbcbe16cd46a58021/
       threshold: 0.5

docs/configuration.rst

@@ -540,6 +540,12 @@ To entirely skip the anomaly detection phase, you can set ``inspector_module_nam
    * - threshold
      - ``0.5``
      - Threshold for the detector's classification.
+   * - domain_expiration_hours
+     - ``1``
+     - Domainator only: discard source/domain windows that have been inactive for more than this many hours, measured using observed message timestamps rather than wall-clock time.
+   * - expiry_check_interval_hours
+     - ``1``
+     - Domainator only: interval between stale-window checks, measured using observed message timestamps.
    * - produce_topics
      - ``(empty)``
      - (Optional) Comma-separated list of alerter topic suffixes to produce alerts to. If left empty, defaults to the ``generic`` alerter topic. Use ``send_to_alerter: false`` or ``produce_topics: []`` for intermediary detectors that should not produce to an alerter.

src/detector/plugins/domainator_detector.py

@@ -1,6 +1,7 @@
 from src.detector.detector import DetectorBase
 import numpy as np
 from collections import defaultdict
+from datetime import datetime, timedelta
 
 from src.base.log_config import get_logger
 from src.detector.plugins.domainator_utils import (
@@ -45,11 +46,64 @@ def __init__(
             consume_topic (str): Kafka topic from which the detector will consume messages.
         """
         self.model_base_url = detector_config["base_url"]
-        self.message_queues = defaultdict(list)
+        # How long a domain has to have been stale for before it gets removed from the dictionary
+        self.domain_expiration_hours = timedelta(
+            hours=detector_config.get("domain_expiration_hours", 1)
+        )
+        # How often the detector checks whether entries have expired
+        self.expiry_check_interval = timedelta(
+            hours=detector_config.get("expiry_check_interval_hours", 1)
+        )
+
+        self.message_queues = defaultdict(
+            lambda: defaultdict(lambda: {"timestamp": None, "messages": []})
+        )
+        self.latest_message_timestamp = None
+        self.latest_expiry_timestamp = None
+
+        self.requests_detect_threshold = 10
+        self.max_window_requests = 10
+        self.max_window_cut = 5
+
         super().__init__(
             detector_config, consume_topic, produce_topics, downstream_detector_topics
         )
 
+    def expire_stale_domains(self):
+        """
+        Drop windows for src_ip/domain pairs that have not been seen for a certain period of time.
+
+        The method deliberately avoids using the current time and instead tracks the
+        timestamp of the observed data. This keeps the tracking correct for both live
+        traffic and historical data.
+
+        Returns:
+            int: Number of ``(src_ip, domain)`` entries removed.
+        """
+        if self.latest_message_timestamp is None:
+            return 0
+
+        removed_domains = 0
+
+        stale_entries = []
+        for src, domains in self.message_queues.items():
+            for domain, entry in domains.items():
+                entry_timestamp = datetime.fromisoformat(entry["timestamp"])
+                if (
+                    self.latest_message_timestamp - entry_timestamp
+                    > self.domain_expiration_hours
+                ):
+                    stale_entries.append((src, domain))
+
+        for src, domain in stale_entries:
+            del self.message_queues[src][domain]
+            removed_domains += 1
+
+            if not self.message_queues[src]:
+                del self.message_queues[src]
+
+        return removed_domains
+
     def predict(self, messages):
         """
         Process a window of messages and predict if the domain is likely to be used
@@ -63,19 +117,31 @@ def predict(self, messages):
                 a "domain_name" key with the domain to analyze.
 
         Returns:
-            np.ndarray: Prediction probabilities for each class. Typically a 2D array
-                where the shape is (1, 2) for binary classification (benign/malicious).
+            dict: Prediction probabilities keyed by the model's class labels.
         """
         queries = [message["domain_name"] for message in messages]
 
         y_pred = self.model.predict_proba(get_domainator_features(queries))
+        y_pred = dict(zip(self.model.classes_, y_pred[0]))
+
         return y_pred
 
     def detect(self):
         logger.info("Start detecting malicious requests.")
         for message in self.messages:
+            message_src = message["src_ip"]
             message_domain = strip_domain(message["domain_name"])
-            self.message_queues[message_domain].append(message)
+            message_queue = self.message_queues[message_src][message_domain]
+
+            message_queue["messages"].append(message)
+            message_queue["timestamp"] = message["ts"]
+
+            datetime_timestamp = datetime.fromisoformat(message_queue["timestamp"])
+            if (
+                self.latest_message_timestamp is None
+                or datetime_timestamp > self.latest_message_timestamp
+            ):
+                self.latest_message_timestamp = datetime_timestamp
 
             # currently receives 1 message batch at a time iterates over it and adds to the queue
             # right now let's say we fetch 100 messages in a batch, we iterate over the messages to add to the dict, worst case: 97 predictions necessary (len of 3, sliding window over all 100 messsages)
@@ -84,19 +150,35 @@ def detect(self):
             # improvement suggestion: try to predict only at the end for each domain seen in the batch IF len > 3. If prediction happened maybe clear the queue ? --> suboptimal for no sliding window approach
             # further improvement: kafka partitoin key shold not be src_ip but rather domain name --> make adjustable for other detectors as well !!
 
-            if len(self.message_queues[message_domain]) >= 3:
-                y_pred = self.predict(self.message_queues[message_domain])
+            if len(message_queue["messages"]) >= self.requests_detect_threshold:
+                y_pred = self.predict(message_queue["messages"])
                 logger.info(f"Prediction: {y_pred}")
-                if np.argmax(y_pred, axis=1) == 1 and y_pred[0][1] > self.threshold:
+                if (
+                    y_pred["malicious"] == max(y_pred.values())
+                    and y_pred["malicious"] > self.threshold
+                ):
                     logger.info("Append malicious request domain to warning.")
                     warning = {
-                        "request": self.message_queues[message_domain].copy(),
-                        "probability": float(y_pred[0][1]),
+                        "request": message_queue["messages"].copy(),
+                        "probability": float(y_pred["malicious"]),
                         "name": self.name,
                         "sha256": self.checksum,
                     }
                     self.warnings.append(warning)
 
-                if len(self.message_queues[message_domain]) >= 5:
-                    del self.message_queues[message_domain][0]
+                # The statement here can be removed if the minimum and maximum window sizes are kept the same
+                if len(message_queue["messages"]) >= self.max_window_requests:
+                    del message_queue["messages"][0 : self.max_window_cut]
+
+        # The check is done based on the message's timestamps to support historical data
+        if self.latest_message_timestamp is not None and (
+            self.latest_expiry_timestamp is None
+            or self.latest_message_timestamp - self.latest_expiry_timestamp
+            > self.expiry_check_interval
+        ):
+            removed_domains = self.expire_stale_domains()
+            self.latest_expiry_timestamp = self.latest_message_timestamp
+            if removed_domains:
+                logger.info(f"Expired {removed_domains} stale window(s).")
+
         logger.debug(f"Domainator Message queue length: {len(self.message_queues)}")

src/detector/plugins/domainator_utils.py

@@ -36,7 +36,7 @@ def strip_domain(query: str):
     domain = ""
 
     if len(query) >= 2:
-        domain = query[-2]
+        domain = ".".join(query[-2:])
 
     return domain
 
@@ -56,15 +56,23 @@ def get_domainator_features(queries: list) -> pd.DataFrame:
     """
 
     queries = [query.strip(".") for query in queries]
-    subdomains = [".".join(domain.split(".")[:-2]) for domain in queries]
+    subdomains = ["".join(domain.split(".")[:-2]) for domain in queries]
 
     metrics = {column: [] for column in DOMAINATOR_FEATURE_COLUMNS}
 
     # if subdomains:
     cartesian = list(itertools.combinations(subdomains, 2))
 
     metrics["levenshtein"] = np.mean(
-        [Levenshtein.ratio(product[0], product[1]) for product in cartesian]
+        [
+            (
+                Levenshtein.distance(product[0], product[1])
+                / max(len(product[0]), len(product[1]))
+                if len(product[0]) and len(product[1])
+                else 0.0
+            )
+            for product in cartesian
+        ]
     )
     metrics["jaro"] = np.mean(
         [Levenshtein.jaro(product[0], product[1]) for product in cartesian]
@@ -91,7 +99,7 @@ def get_domainator_features(queries: list) -> pd.DataFrame:
         [
             (
                 pylcs.lcs_sequence_length(product[0], product[1])
-                / ((len(product[0]) + len(product[1])) / 2)
+                / (max(len(product[0]), len(product[1])))
                 if len(product[0]) and len(product[1])
                 else 0.0
             )
@@ -102,7 +110,7 @@ def get_domainator_features(queries: list) -> pd.DataFrame:
         [
             (
                 pylcs.lcs_string_length(product[0], product[1])
-                / ((len(product[0]) + len(product[1])) / 2)
+                / (max(len(product[0]), len(product[1])))
                 if len(product[0]) and len(product[1])
                 else 0.0
             )

tests/detector/test_domainator_attributor.py

@@ -125,7 +125,7 @@ def test_detect_snapshots_each_warning_request(self):
         )
         self.assertEqual(
             ["1", "2", "3", "4"],
-            [message["logline_id"] for message in sut.message_queues["example"]],
+            [message["logline_id"] for message in sut.message_queues["example.org"]],
         )
 
     def test_detect_emits_for_attribution_class_other_than_index_one(self):
@@ -248,7 +248,7 @@ def test_get_features_basic_attributes(self):
         jaro_dist = features.iloc[0, 1]  # Jaro distance
         lcs = features.iloc[0, 6]  # Longest common string
 
-        self.assertEqual(leven_dist, 0.75)
+        self.assertEqual(leven_dist, 0.25)
         self.assertAlmostEqual(jaro_dist, 0.833, 3)  # Rounded to 3 decimal places
         self.assertEqual(lcs, 0.75)
 
@@ -262,8 +262,8 @@ def test_get_features_empty_domains(self):
 
         # Basic features
         self.assertEqual(
-            features.iloc[0, 0], 1.0
-        )  # Levenshtein distance of empty strings is 1
+            features.iloc[0, 0], 0.0
+        )  # Normalized Levenshtein distance is 0 for empty subdomains
         self.assertEqual(
             features.iloc[0, 1], 1.0
         )  # Jaro distance of empty strings is 1
@@ -293,8 +293,8 @@ def test_get_features_single_same_character(self):
 
         # Basic features
         self.assertEqual(
-            features.iloc[0, 0], 1.0
-        )  # Levenshtein distance of same strings is 1
+            features.iloc[0, 0], 0.0
+        )  # Identical subdomains have zero Levenshtein distance
         self.assertEqual(features.iloc[0, 1], 1.0)  # Jaro distance of same strings is 1
         self.assertEqual(
             features.iloc[0, 2], 1.0