Draft: Prepare v1.0.0-rc1

config.yaml

@@ -70,12 +70,12 @@ pipeline:
 
 environment:
   kafka_brokers:
-    - hostname: kafka1
-      port: 19092
-    - hostname: kafka2
-      port: 19093
-    - hostname: kafka3
-      port: 19094
+    - hostname: 127.0.0.1
+      port: 8097
+    - hostname: 127.0.0.1
+      port: 8098
+    - hostname: 127.0.0.1
+      port: 8099
   kafka_topics:
     pipeline:
       logserver_in: "pipeline-logserver_in"

docs/pipeline.rst

@@ -433,17 +433,28 @@ Stage 4: Inspection
 Overview
 --------
 
-The `Inspector` stage is responsible to run time-series based anomaly detection on prefiltered batches. This stage
-is essentiell to reduce the load on the `Detection` stage. Otherwise, resource complexity increases disproportionately.
+The **Inspection** stage performs time-series-based anomaly detection on prefiltered DNS request batches.
+Its primary purpose is to reduce the load on the `Detection` stage by filtering out non-suspicious traffic early.
+
+This stage uses StreamAD models—supporting univariate, multivariate, and ensemble techniques—to detect unusual patterns
+in request volume and packet sizes.
+
 
 Main Class
 ----------
 
 .. py:currentmodule:: src.inspector.inspector
 .. autoclass:: Inspector
+   :members:
+   :undoc-members:
+   :show-inheritance:
+
+The :class:`Inspector` class is responsible for:
 
-The :class:`Inspector` is the primary class to run StreamAD models for time-series based anomaly detection, such as the Z-Score outlier detection.
-In addition, it features fine-tuning settings for models and anomaly thresholds.
+- Loading batches from Kafka
+- Extracting time-series features (e.g., frequency and packet size)
+- Applying anomaly detection models
+- Forwarding suspicious batches to the detector stage
 
 Usage
 -----
@@ -498,9 +509,9 @@ Stage 5: Detection
 Overview
 --------
 
-The `Detector` resembles the heart of heiDGAF. It runs pre-trained machine learning models to get a probability outcome for the DNS requests.
-The pre-trained models are under the EUPL-1.2 license online available.
-In total, we rely on the following data sets for the pre-trained models we offer:
+The **Detection** stage is the core of the heiDGAF pipeline. It consumes **suspicious batches** passed from the `Inspector`, applies **pre-trained ML models** to classify individual DNS requests, and issues alerts based on aggregated probabilities.
+
+The pre-trained models used here are licensed under **EUPL‑1.2** and built from the following datasets:
 
 - `CIC-Bell-DNS-2021 <https://www.unb.ca/cic/datasets/dns-2021.html>`_
 - `DGTA-BENCH - Domain Generation and Tunneling Algorithms for Benchmark <https://data.mendeley.com/datasets/2wzf9bz7xr/1>`_
@@ -511,15 +522,39 @@ Main Class
 
 .. py:currentmodule:: src.detector.detector
 .. autoclass:: Detector
+   :members:
+   :undoc-members:
+   :show-inheritance:
+
+The :class:`Detector` class:
+
+- Consumes a batch flagged as suspicious.
+- Downloads and validates the ML model (if necessary).
+- Extracts features from domain names (e.g. character distributions, entropy, label statistics).
+- Computes a probability per request and an overall risk score per batch.
+- Emits alerts to ClickHouse and logs in ``/tmp/warnings.json`` where applicable.
 
 Usage
 -----
 
-The :class:`Detector` consumes anomalous batches of requests.
-It calculates a probability score for each request, and at last, an overall score of the batch.
-Alerts are log to ``/tmp/warnings.json``.
+1. The `Detector` listens on the Kafka topic from the Inspector (``inspector_to_detector``).  
+2. For each suspicious batch:
+   - Extracts features for every domain request.
+   - Applies the loaded ML model (after scaling) to compute class probabilities.
+   - Marks a request as malicious if its probability exceeds the configured `threshold`.
+3. Computes an **overall score** (e.g. median of malicious probabilities) for the batch.
+4. If malicious requests exist, issues an **alert** record and logs it; otherwise, the batch is filtered.
+
+Alerts are recorded in ClickHouse and also appended to a local JSON file (`warnings.json`) for external monitoring.
 
 Configuration
 -------------
 
-In case you want to load self-trained models, the :class:`Detector` needs a URL path, model name, and SHA256 checksum to download the model during start-up.
+You may use the provided, pre-trained models or supply your own. To use a custom model, specify:
+
+- `base_url`: URL from which to fetch model artifacts  
+- `model`: model name  
+- `checksum`: SHA256 digest for integrity validation  
+- `threshold`: probability threshold for classifying a request as malicious  
+
+These parameters are loaded at startup and used to download, verify, and load the model/scaler if not already cached locally (in temp directory).

src/detector/detector.py

@@ -244,84 +244,77 @@ def _get_features(self, query: str) -> np.ndarray:
         """
 
         # Splitting by dots to calculate label length and max length
+        query = query.strip(".")
         label_parts = query.split(".")
+
+        levels = {
+            "fqdn": query,
+            "secondleveldomain": label_parts[-2] if len(label_parts) >= 2 else "",
+            "thirdleveldomain": (
+                ".".join(label_parts[:-2]) if len(label_parts) > 2 else ""
+            ),
+        }
+
         label_length = len(label_parts)
-        label_max = max(len(part) for part in label_parts)
-        label_average = len(query.strip("."))
+        parts = query.split(".")
+        label_max = len(max(parts, key=str)) if parts else 0
+        label_average = len(query)
+
+        basic_features = np.array(
+            [label_length, label_max, label_average], dtype=np.float64
+        )
 
-        logger.debug("Get letter frequency")
         alc = "abcdefghijklmnopqrstuvwxyz"
+        query_len = len(query)
         freq = np.array(
-            [query.lower().count(i) / len(query) if len(query) > 0 else 0 for i in alc]
+            [query.lower().count(c) / query_len if query_len > 0 else 0.0 for c in alc],
+            dtype=np.float64,
         )
 
         logger.debug("Get full, alpha, special, and numeric count.")
 
         def calculate_counts(level: str) -> np.ndarray:
-            if len(level) == 0:
-                return np.array([0, 0, 0, 0])
-
-            full_count = len(level)
-            alpha_count = sum(c.isalpha() for c in level) / full_count
-            numeric_count = sum(c.isdigit() for c in level) / full_count
-            special_count = (
-                sum(not c.isalnum() and not c.isspace() for c in level) / full_count
+            if not level:
+                return np.array([0.0, 0.0, 0.0, 0.0], dtype=np.float64)
+
+            full_count = len(level) / len(level)
+            alpha_ratio = sum(c.isalpha() for c in level) / len(level)
+            numeric_ratio = sum(c.isdigit() for c in level) / len(level)
+            special_ratio = sum(
+                not c.isalnum() and not c.isspace() for c in level
+            ) / len(level)
+
+            return np.array(
+                [full_count, alpha_ratio, numeric_ratio, special_ratio],
+                dtype=np.float64,
             )
 
-            return np.array([full_count, alpha_count, numeric_count, special_count])
-
-        levels = {
-            "fqdn": query,
-            "thirdleveldomain": label_parts[0] if len(label_parts) > 2 else "",
-            "secondleveldomain": label_parts[1] if len(label_parts) > 1 else "",
-        }
-        counts = {
-            level: calculate_counts(level_value)
-            for level, level_value in levels.items()
-        }
-
-        logger.debug(
-            "Get standard deviation, median, variance, and mean for full, alpha, special, and numeric count."
-        )
-        stats = {}
-        for level, count_array in counts.items():
-            stats[f"{level}_std"] = np.std(count_array)
-            stats[f"{level}_var"] = np.var(count_array)
-            stats[f"{level}_median"] = np.median(count_array)
-            stats[f"{level}_mean"] = np.mean(count_array)
+        fqdn_counts = calculate_counts(levels["fqdn"])
+        third_counts = calculate_counts(levels["thirdleveldomain"])
+        second_counts = calculate_counts(levels["secondleveldomain"])
 
-        logger.debug("Start entropy calculation")
+        level_features = np.hstack([third_counts, second_counts, fqdn_counts])
 
         def calculate_entropy(s: str) -> float:
             if len(s) == 0:
-                return 0
-            probabilities = [float(s.count(c)) / len(s) for c in dict.fromkeys(list(s))]
-            entropy = -sum(p * math.log(p, 2) for p in probabilities)
-            return entropy
-
-        entropy = {level: calculate_entropy(value) for level, value in levels.items()}
-
-        logger.debug("Finished entropy calculation")
+                return 0.0
+            probs = [s.count(c) / len(s) for c in dict.fromkeys(s)]
+            return -sum(p * math.log(p, 2) for p in probs)
 
-        # Final feature aggregation as a NumPy array
-        basic_features = np.array([label_length, label_max, label_average])
-
-        # Flatten counts and stats for each level into arrays
-        level_features = np.hstack([counts[level] for level in levels.keys()])
+        logger.debug("Start entropy calculation")
+        entropy_features = np.array(
+            [
+                calculate_entropy(levels["fqdn"]),
+                calculate_entropy(levels["thirdleveldomain"]),
+                calculate_entropy(levels["secondleveldomain"]),
+            ],
+            dtype=np.float64,
+        )
 
-        # Entropy features
-        entropy_features = np.array([entropy[level] for level in levels.keys()])
+        logger.debug("Entropy features calculated")
 
-        # Concatenate all features into a single numpy array
         all_features = np.concatenate(
-            [
-                basic_features,
-                freq,
-                # freq_features,
-                level_features,
-                # stats_features,
-                entropy_features,
-            ]
+            [basic_features, freq, level_features, entropy_features]
         )
 
         logger.debug("Finished data transformation")
@@ -338,8 +331,9 @@ def detect(self) -> None:  # pragma: no cover
         logger.info("Start detecting malicious requests.")
         for message in self.messages:
             # TODO predict all messages
+            # TODO use scalar: self.scaler.transform(self._get_features(message["domain_name"]))
             y_pred = self.model.predict_proba(
-                self.scaler.transform(self._get_features(message["domain_name"]))
+                self._get_features(message["domain_name"])
             )
             logger.info(f"Prediction: {y_pred}")
             if np.argmax(y_pred, axis=1) == 1 and y_pred[0][1] > THRESHOLD: