Dev fixes 2

config.yaml

@@ -200,7 +200,7 @@ pipeline:
       threshold: 0.05
       consume_from: inspector
       inspector_name: "domainator_inspector"
-      next_detectors: 
+      next_detectors:
         - "domainator_attributor_behaviour"
         - "domainator_attributor_identification_behaviour"
         - "domainator_attributor_identification"

docker/create_tables/alerts.sql

@@ -9,4 +9,4 @@ CREATE TABLE IF NOT EXISTS alerts (
 ENGINE = MergeTree
 PARTITION BY toYYYYMM(alert_timestamp)
 ORDER BY (alert_timestamp, src_ip, suspicious_batch_id)
-TTL toDateTime(alert_timestamp) + INTERVAL 60 DAY;
+TTL toDateTime(alert_timestamp) + INTERVAL 60 DAY;

docker/create_tables/logline_to_batches.sql

@@ -7,5 +7,3 @@ ENGINE = MergeTree
 ORDER BY (timestamp, batch_id, logline_id)
 PARTITION BY toYYYYMM(timestamp)
 TTL toDateTime(timestamp) + INTERVAL 1 DAY;
-
-

docker/create_tables/server_log_to_logline.sql

@@ -7,4 +7,3 @@ ENGINE = MergeTree
 ORDER BY (timestamp, message_id, logline_id)
 PARTITION BY toYYYYMM(timestamp)
 TTL toDateTime(timestamp) + INTERVAL 1 DAY;
-

docker/create_tables/server_logs.sql

@@ -6,4 +6,4 @@ CREATE TABLE IF NOT EXISTS server_logs (
 ENGINE = MergeTree
 PARTITION BY toYYYYMM(timestamp_in)
 ORDER BY (timestamp_in, message_id)
-TTL toDateTime(timestamp_in) + INTERVAL 1 DAY;
+TTL toDateTime(timestamp_in) + INTERVAL 1 DAY;

docker/create_tables/server_logs_timestamps.sql

@@ -6,4 +6,4 @@ CREATE TABLE IF NOT EXISTS server_logs_timestamps (
 ENGINE = MergeTree
 PARTITION BY toYYYYMM(event_timestamp)
 ORDER BY (event, event_timestamp, message_id)
-TTL toDateTime(event_timestamp) + INTERVAL 1 DAY;
+TTL toDateTime(event_timestamp) + INTERVAL 1 DAY;

docker/create_tables/suspicious_batch_timestamps.sql

@@ -11,4 +11,4 @@ CREATE TABLE IF NOT EXISTS suspicious_batch_timestamps (
 ENGINE = MergeTree
 PARTITION BY toYYYYMM(timestamp)
 ORDER BY (stage, status, timestamp, instance_name, suspicious_batch_id, src_ip)
-TTL toDateTime(timestamp) + INTERVAL 1 DAY;
+TTL toDateTime(timestamp) + INTERVAL 1 DAY;

docs/api/base.rst

@@ -2,4 +2,3 @@ Base
 ====
 
 .. automodule:: src.base
-

docs/api/detector.rst

@@ -2,4 +2,3 @@ Detector
 ========
 
 .. automodule:: src.detector
-

docs/api/inspector.rst

@@ -2,4 +2,3 @@ Inspector
 =========
 
 .. automodule:: src.inspector
-

docs/api/logcollector.rst

@@ -2,4 +2,3 @@ Log Collector
 =============
 
 .. automodule:: src.logcollector
-

docs/api/logserver.rst

@@ -2,4 +2,3 @@ Log Server
 ==========
 
 .. automodule:: src.logserver
-

docs/api/monitoring.rst

@@ -2,4 +2,3 @@ Monitoring
 ==========
 
 .. automodule:: src.monitoring
-

docs/api/prefilter.rst

@@ -2,4 +2,3 @@ Prefilter
 =========
 
 .. automodule:: src.prefilter
-

docs/api/train.rst

@@ -2,4 +2,3 @@ Train
 =====
 
 .. automodule:: src.train
-

docs/api/version.rst

@@ -2,4 +2,3 @@ Version
 =======
 
 .. automodule:: src.version
-

docs/index.rst

@@ -2,7 +2,7 @@ Welcome to HAMSTRING's documentation!
 ======================================
 
 **HAMSTRING** is a CIDS framework to run signature and machine learning-based IDS classifiers. It employs several string and anomaly-based filtering techniques
-to maximize detection efficiency. Currently the tool focueses on DNS attacks, as it incorporates heiDGAF (DGA detection) & Domainator (DNS Tunneling Detector) 
+to maximize detection efficiency. Currently the tool focueses on DNS attacks, as it incorporates heiDGAF (DGA detection) & Domainator (DNS Tunneling Detector)
 :cite:p:`petrov_domainator_2025` :cite:p:`machmeier_heidgaf`.
 
 Check out the :doc:`usage` section for further information on how to use the software, including how to

src/alerter/alerter.py

@@ -348,9 +348,7 @@ async def start(self):
 
 def build_alerter_worker(alerter_config, consume_topic, worker_id=None):
     class_name = alerter_config.get("alerter_class_name", "GenericAlerter")
-    alerter_module_name = alerter_config.get(
-        "alerter_module_name", "generic_alerter"
-    )
+    alerter_module_name = alerter_config.get("alerter_module_name", "generic_alerter")
     plugin_module_name = f"{PLUGIN_PATH}.{alerter_module_name}"
     plugin_module = importlib.import_module(plugin_module_name)
     alerter_class = getattr(plugin_module, class_name)

src/base/execution.py

@@ -97,7 +97,9 @@ async def start_pipeline_worker_replicas(
         return
 
     if process_entrypoint is None:
-        raise ValueError("process_entrypoint is required for process and hybrid scaling")
+        raise ValueError(
+            "process_entrypoint is required for process and hybrid scaling"
+        )
 
     loop = asyncio.get_running_loop()
     executor = ProcessPoolExecutor(max_workers=executor_config.processes)

src/base/kafka_handler.py

@@ -53,8 +53,8 @@
 )
 KAFKA_TOPIC_STAGE_CONFIG = KAFKA_TOPIC_CONFIG.get("stages", {})
 KAFKA_TOPIC_EXACT_CONFIG = KAFKA_TOPIC_CONFIG.get("topics", {})
-KAFKA_PIPELINE_TOPIC_PREFIXES = config["environment"].get("kafka_topics_prefix", {}).get(
-    "pipeline", {}
+KAFKA_PIPELINE_TOPIC_PREFIXES = (
+    config["environment"].get("kafka_topics_prefix", {}).get("pipeline", {})
 )
 
 
@@ -390,17 +390,23 @@ def _reset_producer(self) -> None:
             if self.producer:
                 self.producer.flush(5)
         except Exception as exception:
-            logger.warning("Ignoring Kafka producer flush failure during reconnect: %s", exception)
+            logger.warning(
+                "Ignoring Kafka producer flush failure during reconnect: %s", exception
+            )
         self.producer = self._new_producer()
 
-    def _with_producer_retry(self, description: str, operation: Callable[[], None]) -> None:
+    def _with_producer_retry(
+        self, description: str, operation: Callable[[], None]
+    ) -> None:
         def attempt():
             try:
                 operation()
             except Exception as exception:
                 if not _is_retriable_kafka_exception(exception):
                     raise
-                logger.warning("%s failed, recreating Kafka producer: %s", description, exception)
+                logger.warning(
+                    "%s failed, recreating Kafka producer: %s", description, exception
+                )
                 self._reset_producer()
                 raise
 
@@ -599,7 +605,9 @@ def operation():
                 try:
                     self.producer.abort_transaction()
                 except Exception as abort_exception:
-                    logger.warning("Kafka transaction abort failed: %s", abort_exception)
+                    logger.warning(
+                        "Kafka transaction abort failed: %s", abort_exception
+                    )
                 logger.error("Transaction aborted.")
                 logger.error(e)
                 raise
@@ -711,7 +719,9 @@ def connect():
             )
             target_partitions_by_topic = ensure_topics(admin_client, self.topics)
 
-            if not self._all_topics_created(self.topics, target_partitions_by_topic, consumer):
+            if not self._all_topics_created(
+                self.topics, target_partitions_by_topic, consumer
+            ):
                 try:
                     consumer.close()
                 except Exception:
@@ -721,14 +731,18 @@ def connect():
             consumer.subscribe(self.topics)
             return consumer
 
-        self.consumer = retry_forever(connect, f"Kafka consumer setup for {self.topics}")
+        self.consumer = retry_forever(
+            connect, f"Kafka consumer setup for {self.topics}"
+        )
 
     def _reset_consumer(self) -> None:
         try:
             if self.consumer:
                 self.consumer.close()
         except Exception as exception:
-            logger.warning("Ignoring Kafka consumer close failure during reconnect: %s", exception)
+            logger.warning(
+                "Ignoring Kafka consumer close failure during reconnect: %s", exception
+            )
         self._last_consumed_message = None
         self._connect_consumer()
 
@@ -842,7 +856,9 @@ def _poll_message(self):
             try:
                 msg = self.consumer.poll(timeout=1.0)
             except (KafkaException, RuntimeError, OSError) as exception:
-                logger.warning("Kafka consumer poll failed, reconnecting: %s", exception)
+                logger.warning(
+                    "Kafka consumer poll failed, reconnecting: %s", exception
+                )
                 self._reset_consumer()
                 continue
 
@@ -853,7 +869,10 @@ def _poll_message(self):
                 if msg.error().code() == KafkaError._PARTITION_EOF:
                     return None
                 if _is_retriable_kafka_error(msg.error()):
-                    logger.warning("Kafka consumer error is retriable, reconnecting: %s", msg.error())
+                    logger.warning(
+                        "Kafka consumer error is retriable, reconnecting: %s",
+                        msg.error(),
+                    )
                     self._reset_consumer()
                     return None
 

src/base/retry.py

@@ -21,11 +21,11 @@
 
 def resilience_config() -> dict[str, Any]:
     config = setup_config()
-    retry_config = (
-        config.get("pipeline", {})
-        .get("resilience", {})
-        .get("retry", {})
-    )
+    if not isinstance(config, dict):
+        return dict(_DEFAULT_CONFIG)
+    retry_config = config.get("pipeline", {}).get("resilience", {}).get("retry", {})
+    if not isinstance(retry_config, dict):
+        retry_config = {}
     merged = dict(_DEFAULT_CONFIG)
     merged.update(retry_config)
     return merged
@@ -37,7 +37,7 @@ def retry_forever(
     retry_config: dict[str, Any] | None = None,
     retryable: tuple[type[BaseException], ...] = (Exception,),
 ) -> T:
-    config = retry_config or resilience_config()
+    config = retry_config if retry_config is not None else resilience_config()
     initial_delay = _float_setting(config, "initial_delay_seconds")
     max_delay = _float_setting(config, "max_delay_seconds")
     multiplier = max(1.0, _float_setting(config, "backoff_multiplier"))

src/detector/detector.py

@@ -61,7 +61,9 @@ def _normalize_topic_suffixes(topic_suffixes) -> list[str]:
     if topic_suffixes is None:
         return []
     if isinstance(topic_suffixes, str):
-        return [suffix.strip() for suffix in topic_suffixes.split(",") if suffix.strip()]
+        return [
+            suffix.strip() for suffix in topic_suffixes.split(",") if suffix.strip()
+        ]
     if isinstance(topic_suffixes, (list, tuple, set)):
         return [str(suffix).strip() for suffix in topic_suffixes if str(suffix).strip()]
     return [str(topic_suffixes).strip()]
@@ -85,7 +87,9 @@ def build_alerter_topics(detector_config: dict) -> list[str]:
 def build_downstream_detector_topics(detector_config: dict) -> list[str]:
     detector_names = []
     for config_key in ("next_detectors", "produce_detector_topics"):
-        detector_names.extend(_normalize_topic_suffixes(detector_config.get(config_key)))
+        detector_names.extend(
+            _normalize_topic_suffixes(detector_config.get(config_key))
+        )
 
     return [
         f"{DETECTOR_TO_DETECTOR_TOPIC_PREFIX}-{detector_name}"
@@ -352,7 +356,6 @@ def _sha256sum(self, file_path: str) -> str:
 
         return h.hexdigest()
 
-
     def get_model_download_url(self):
         """
         Generate the complete URL for downloading the Domainator detection model.
@@ -387,8 +390,6 @@ def get_scaler_download_url(self):
         )
         return f"{self.model_base_url}/files/?p=%2F{self.model_name}%2F{self.checksum}%2Fscaler.pickle&dl=1"
 
-
-
     def _get_model(self):
         """
         Download and validate the detection model.
@@ -410,7 +411,9 @@ def _get_model(self):
             requests.HTTPError: If there's an error downloading the model.
         """
         logger.info(f"Get model: {self.model_name} with checksum {self.checksum}")
-        scaler_download_url = self.get_scaler_download_url() if self.use_scaler else None
+        scaler_download_url = (
+            self.get_scaler_download_url() if self.use_scaler else None
+        )
 
         if not os.path.isfile(self.model_path):
             model_download_url = self.get_model_download_url()
@@ -460,7 +463,7 @@ def detect(self) -> None:
         then just overwrite the method and append to `self.warnings` a warning if a given messag is to be regarded malicious.
 
         Note:
-        
+
         """
         logger.info("general")
         logger.info("Start detecting malicious requests.")
@@ -479,7 +482,7 @@ def detect(self) -> None:
                     "sha256": self.checksum,
                 }
                 self.warnings.append(warning)
-                
+
     def clear_data(self):
         """Clears the data in the internal data structures."""
         self.messages = []
@@ -646,8 +649,7 @@ def _send_detector_batch(self, parent_row_id, messages) -> None:
 
     def _build_persisted_warnings(self) -> list[dict]:
         return [
-            self._normalize_warning_for_storage(warning)
-            for warning in self.warnings
+            self._normalize_warning_for_storage(warning) for warning in self.warnings
         ]
 
     def _normalize_warning_for_storage(self, warning: dict) -> dict:
@@ -658,11 +660,7 @@ def _normalize_warning_for_storage(self, warning: dict) -> dict:
         request_messages = (
             self._flatten_messages(request) if request is not None else []
         )
-        detector_name = (
-            warning.get("detector_name")
-            or warning.get("name")
-            or self.name
-        )
+        detector_name = warning.get("detector_name") or warning.get("name") or self.name
         score = warning.get("score", warning.get("probability"))
         domains = self._extract_warning_domains(warning)
         logline_ids = self._extract_message_values(request_messages, "logline_id")
@@ -753,9 +751,7 @@ def _extract_warning_domains(self, warning) -> list[str]:
                     field_value = item.get(field_name)
                     if isinstance(field_value, list):
                         domains.extend(
-                            domain
-                            for domain in field_value
-                            if isinstance(domain, str)
+                            domain for domain in field_value if isinstance(domain, str)
                         )
 
         return sorted(set(domains))

src/detector/plugins/dga_detector.py

@@ -152,7 +152,6 @@ def calculate_entropy(s: str) -> float:
 
         return all_features.reshape(1, -1)
 
-
     def detect(self) -> None:
         """
         Process messages to detect malicious requests.
@@ -178,4 +177,4 @@ def detect(self) -> None:
                     "name": self.name,
                     "sha256": self.checksum,
                 }
-                self.warnings.append(warning)
+                self.warnings.append(warning)