WIP-Draft: Feature/zeek integration

.gitignore

@@ -325,3 +325,6 @@ cython_debug/
 # Others
 docs/api/
 !/docs/api/index.rst
+
+# requirements.txt
+!*/requirements.*.txt

README.md

@@ -56,19 +56,24 @@
 
 ## About the Project
 
-![Pipeline overview](https://raw.githubusercontent.com/stefanDeveloper/heiDGAF/main/docs/media/heidgaf_overview_detailed.drawio.png?raw=true)
+![Pipeline overview](./assets/heidgaf_architecture.svg)
 
 ## Getting Started
 
 #### Run **heiDGAF** using Docker Compose:
-
 ```sh
-HOST_IP=127.0.0.1 docker compose -f docker/docker-compose.yml up
+HOST_IP=127.0.0.1 docker compose -f docker/docker-compose.yml --profile prod up
 ```
 <p align="center">
   <img src="https://raw.githubusercontent.com/stefanDeveloper/heiDGAF/main/assets/terminal_example.gif?raw=true" alt="Terminal example"/>
 </p>
 
+#### Use the dev profile for testing out changes in docke containers:
+```sh
+HOST_IP=127.0.0.1 docker compose -f docker/docker-compose.yml --profile dev up
+```
+
+
 #### Or run the modules locally on your machine:
 ```sh
 python -m venv .venv
@@ -87,6 +92,8 @@ python src/inspector/inspector.py
 <p align="right">(<a href="#readme-top">back to top</a>)</p>
 
 
+
+
 ## Usage
 
 ### Configuration
@@ -276,6 +283,31 @@ This will create a `rules.txt` file containing the innards of the model, explain
 <p align="right">(<a href="#readme-top">back to top</a>)</p>
 
 
+### Data
+
+> [!IMPORTANT]
+> We support custom schemes.
+
+Depending on your data and usecase, you can customize the data scheme to fit your needs.
+The below configuration is part of the [main configuration file](./config.yaml) which is detailed in our [documentation](https://heidgaf.readthedocs.io/en/latest/usage.html#id2)
+
+```yml
+loglines:
+  fields:
+    - [ "timestamp", RegEx, '^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$' ]
+    - [ "status_code", ListItem, [ "NOERROR", "NXDOMAIN" ], [ "NXDOMAIN" ] ]
+    - [ "src_ip", IpAddress ]
+    - [ "dns_server_ip", IpAddress ]
+    - [ "domain_name", RegEx, '^(?=.{1,253}$)((?!-)[A-Za-z0-9-]{1,63}(?<!-)\.)+[A-Za-z]{2,63}$' ]
+    - [ "record_type", ListItem, [ "A", "AAAA" ] ]
+    - [ "response_ip", IpAddress ]
+    - [ "size", RegEx, '^\d+b$' ]
+```
+
+
+
+<p align="right">(<a href="#readme-top">back to top</a>)</p>
+
 <!-- CONTRIBUTING -->
 ## Contributing
 

config.yaml

@@ -20,26 +20,42 @@ pipeline:
     logserver:
       input_file: "/opt/file.txt"
 
+
+
   log_collection:
-    collector:
-      logline_format:
-        - [ "timestamp", Timestamp, "%Y-%m-%dT%H:%M:%S.%fZ" ]
-        - [ "status_code", ListItem, [ "NOERROR", "NXDOMAIN" ], [ "NXDOMAIN" ] ]
-        - [ "client_ip", IpAddress ]
-        - [ "dns_server_ip", IpAddress ]
-        - [ "domain_name", RegEx, '^(?=.{1,253}$)((?!-)[A-Za-z0-9-]{1,63}(?<!-)\.)+[A-Za-z]{2,63}$' ]
-        - [ "record_type", ListItem, [ "A", "AAAA" ] ]
-        - [ "response_ip", IpAddress ]
-        - [ "size", RegEx, '^\d+b$' ]
-    batch_handler:
-      batch_size: 10000
+    default_batch_handler_config:
+      batch_size: 2000
       batch_timeout: 30.0
       subnet_id:
         ipv4_prefix_length: 24
         ipv6_prefix_length: 64
+    collectors:
+      - name: "dga_collector"
+        protocol_base: dns
+        required_log_information:
+          - [ "ts", Timestamp, "%Y-%m-%dT%H:%M:%S" ]
+          - [ "status_code", ListItem, [ "NOERROR", "NXDOMAIN" ], [ "NXDOMAIN" ] ]
+          - [ "src_ip", IpAddress ]
+          - [ "dns_server_ip", IpAddress ]
+          - [ "domain_name", RegEx, '^(?=.{1,253}$)((?!-)[A-Za-z0-9-]{1,63}(?<!-)\.)+[A-Za-z]{2,63}$' ]
+          - [ "record_type", ListItem, [ "A", "AAAA" ] ]
+          # - [ "response_ip", IpAddress ]
+          - [ "size", RegEx, '^\d+$' ]
+        batch_handler_config_override:
+          batch_timeout: 30.1
+
+  log_filtering:
+    - name: "dga_filter"
+      # method to apply for rule based prefiltering according to the needs
+      relevance_method: check_dga_relevance
+      collector_name: dga_collector
+
 
   data_inspection:
-    inspector:
+    - name: dga_inspector
+      inspector_module_name: "stream_ad_inspector"
+      inspector_class_name: "StreamADInspector"
+      prefilter_name: dga_filter
       mode: univariate # multivariate, ensemble
       # Only used when mode is set to ensemble
       ensemble:
@@ -51,32 +67,51 @@ pipeline:
           module: streamad.model
           model_args:
             is_global: false
-      anomaly_threshold: 0.01
-      score_threshold: 0.5
+      anomaly_threshold: 0.0001
+      score_threshold: 0.005
       time_type: ms
       time_range: 20
 
   data_analysis:
-    detector:
-      model: rf # XGBoost
-      checksum: 021af76b2385ddbc76f6e3ad10feb0bb081f9cf05cff2e52333e31040bbf36cc
+    - name: "RF-dga_detector"
+      detector_module_name: "dga_detector"
+      detector_class_name: "DGADetector"
+      model: rf
+      checksum: 5db8bfb617e80361362c33b1d1afc6d762c28e9fa9275fb11514a3bdef76bb88
       base_url: https://heibox.uni-heidelberg.de/d/0d5cbcbe16cd46a58021/
-      threshold: 0.5
+      threshold: 0.005
+      inspector_name: dga_inspector
 
   monitoring:
     clickhouse_connector:
       batch_size: 50  # do not set higher
       batch_timeout: 2.0
 
+
+  zeek:
+    sensors:
+      zeek1:
+        static_analysis: true
+        protocols:
+          - dns
+        interfaces:
+          - enp2s0f0
+
 environment:
   kafka_brokers:
     - hostname: kafka1
-      port: 19092
+      internal_port: 19092
+      external_port: 8097
+      node_ip: 127.0.0.1
     - hostname: kafka2
-      port: 19093
+      internal_port: 19093
+      external_port: 8098
+      node_ip: 127.0.0.1
     - hostname: kafka3
-      port: 19094
-  kafka_topics:
+      internal_port: 19094
+      external_port: 8099
+      node_ip: 127.0.0.1
+  kafka_topics_prefix:
     pipeline:
       logserver_in: "pipeline-logserver_in"
       logserver_to_collector: "pipeline-logserver_to_collector"

docker/create_tables/alerts.sql

@@ -1,10 +1,10 @@
 CREATE TABLE IF NOT EXISTS alerts (
-    client_ip String NOT NULL,
+    src_ip String NOT NULL,
     alert_timestamp DateTime64(6) NOT NULL,
     suspicious_batch_id UUID NOT NULL,
     overall_score Float32 NOT NULL,
     domain_names String NOT NULL,
     result String,
 )
 ENGINE = MergeTree
-PRIMARY KEY(client_ip, alert_timestamp);
+PRIMARY KEY(src_ip, alert_timestamp);

docker/create_tables/batch_timestamps.sql

@@ -1,10 +1,12 @@
 CREATE TABLE IF NOT EXISTS batch_timestamps (
     batch_id UUID NOT NULL,
+    instance_name String NOT NULL,
     stage String NOT NULL,
     status String NOT NULL,
     timestamp DateTime64(6) NOT NULL,
     message_count UInt32,
     is_active Bool NOT NULL
 )
 ENGINE = MergeTree
+-- keep the PK as the UUID even thogh it is not uinque for indexing reasons
 PRIMARY KEY (batch_id);

docker/create_tables/batch_tree.sql

@@ -0,0 +1,14 @@
+-- Table to be able to reconstruct where the batch was processed in
+-- used in grafana to calculate the elapsed time between stages
+CREATE TABLE IF NOT EXISTS batch_tree (
+    batch_row_id String NOT NULL,
+    batch_id UUID NOT NULL,
+    parent_batch_row_id Nullable(String), -- Default of Null indicates a root element
+    instance_name String NOT NULL,
+    stage String NOT NULL,
+    status String NOT NULL,
+    timestamp DateTime64(6) NOT NULL,
+)
+ENGINE = MergeTree
+-- keep the PK as the UUID even thogh it is not uinque for indexing reasons
+PRIMARY KEY (batch_row_id);

docker/create_tables/failed_dns_loglines.sql → docker/create_tables/failed_loglines.sql

@@ -1,4 +1,4 @@
-CREATE TABLE IF NOT EXISTS failed_dns_loglines (
+CREATE TABLE IF NOT EXISTS failed_loglines (
     message_text String NOT NULL,
     timestamp_in DateTime64(6) NOT NULL,
     timestamp_failed DateTime64(6) NOT NULL,

docker/create_tables/dns_loglines.sql → docker/create_tables/loglines.sql

@@ -1,10 +1,8 @@
-CREATE TABLE IF NOT EXISTS dns_loglines (
+CREATE TABLE IF NOT EXISTS loglines (
     logline_id UUID NOT NULL,
-    subnet_id String NOT NULL,
     timestamp DateTime64(6) NOT NULL,
-    status_code String NOT NULL,
-    client_ip String NOT NULL,
-    record_type String NOT NULL,
+    subnet_id String NOT NULL,
+    src_ip String NOT NULL,
     additional_fields String
 )
 ENGINE = MergeTree

docker/create_tables/suspicious_batch_timestamps.sql

@@ -1,11 +1,13 @@
 CREATE TABLE IF NOT EXISTS suspicious_batch_timestamps (
     suspicious_batch_id UUID NOT NULL,
-    client_ip String NOT NULL,
+    src_ip String NOT NULL,
+    instance_name String NOT NULL,
     stage String NOT NULL,
     status String NOT NULL,
     timestamp DateTime64(6) NOT NULL,
     message_count UInt32,
     is_active Bool NOT NULL
 )
 ENGINE = MergeTree
+-- keep the PK as the UUID even thogh it is not uinque for indexing reasons
 PRIMARY KEY (suspicious_batch_id);