Merge pull request #13190 from Security-Onion-Solutions/reyesj2/kafka

Initial Kafka support

pillar/kafka/nodes.sls

@@ -0,0 +1,2 @@
+kafka:
+  nodes:

pillar/top.sls

@@ -61,6 +61,9 @@ base:
     - backup.adv_backup
     - minions.{{ grains.id }}
     - minions.adv_{{ grains.id }}
+    - kafka.nodes
+    - kafka.soc_kafka
+    - kafka.adv_kafka
     - stig.soc_stig
 
   '*_sensor':
@@ -176,6 +179,9 @@ base:
     - minions.{{ grains.id }}
     - minions.adv_{{ grains.id }}
     - stig.soc_stig
+    - kafka.nodes
+    - kafka.soc_kafka
+    - kafka.adv_kafka
 
   '*_heavynode':
     - elasticsearch.auth
@@ -220,6 +226,7 @@ base:
     - minions.adv_{{ grains.id }}
     - stig.soc_stig
     - soc.license
+    - kafka.nodes
 
   '*_receiver':
     - logstash.nodes
@@ -232,6 +239,10 @@ base:
     - redis.adv_redis
     - minions.{{ grains.id }}
     - minions.adv_{{ grains.id }}
+    - kafka.nodes
+    - kafka.soc_kafka
+    - kafka.adv_kafka
+    - soc.license
 
   '*_import':
     - secrets

salt/allowed_states.map.jinja

@@ -103,7 +103,8 @@
           'utility',
           'schedule',
           'docker_clean',
-          'stig'
+          'stig',
+          'kafka'
           ],
       'so-managersearch': [
           'salt.master',
@@ -125,7 +126,8 @@
           'utility',
           'schedule',
           'docker_clean',
-          'stig'
+          'stig',
+          'kafka'
           ],
       'so-searchnode': [
           'ssl',
@@ -159,7 +161,8 @@
           'schedule',
           'tcpreplay',
           'docker_clean',
-          'stig'
+          'stig',
+          'kafka'
           ],
       'so-sensor': [
           'ssl',
@@ -190,7 +193,9 @@
           'telegraf',
           'firewall',
           'schedule',
-          'docker_clean'
+          'docker_clean',
+          'kafka',
+          'elasticsearch.ca'
           ],
       'so-desktop': [
           'ssl',

salt/ca/files/signing_policies.conf

@@ -70,3 +70,17 @@ x509_signing_policies:
     - authorityKeyIdentifier: keyid,issuer:always
     - days_valid: 820
     - copypath: /etc/pki/issued_certs/
+  kafka:
+    - minions: '*'
+    - signing_private_key: /etc/pki/ca.key
+    - signing_cert: /etc/pki/ca.crt
+    - C: US
+    - ST: Utah
+    - L: Salt Lake City
+    - basicConstraints: "critical CA:false"
+    - keyUsage: "digitalSignature, keyEncipherment"
+    - subjectKeyIdentifier: hash
+    - authorityKeyIdentifier: keyid,issuer:always
+    - extendedKeyUsage: "serverAuth, clientAuth"
+    - days_valid: 820
+    - copypath: /etc/pki/issued_certs/

salt/common/tools/sbin/so-image-common

@@ -50,6 +50,7 @@ container_list() {
       "so-idh"
       "so-idstools"
       "so-influxdb"
+      "so-kafka"
       "so-kibana"
       "so-kratos"
       "so-logstash"

salt/docker/defaults.yaml

@@ -187,3 +187,12 @@ docker:
       custom_bind_mounts: []
       extra_hosts: []
       extra_env: []
+    'so-kafka':
+      final_octet: 88
+      port_bindings:
+        - 0.0.0.0:9092:9092
+        - 0.0.0.0:9093:9093
+        - 0.0.0.0:8778:8778
+      custom_bind_mounts: []
+      extra_hosts: []
+      extra_env: []

salt/docker/soc_docker.yaml

@@ -101,3 +101,4 @@ docker:
         multiline: True
         forcedType: "[]string"
     so-zeek: *dockerOptions
+    so-kafka: *dockerOptions

salt/elasticfleet/tools/sbin_jinja/so-elastic-fleet-outputs-update

@@ -21,64 +21,104 @@ function update_logstash_outputs() {
     # Update Logstash Outputs
     curl -K /opt/so/conf/elasticsearch/curl.config -L -X PUT "localhost:5601/api/fleet/outputs/so-manager_logstash" -H 'kbn-xsrf: true' -H 'Content-Type: application/json' -d "$JSON_STRING" | jq
 }
+function update_kafka_outputs() {
+  # Make sure SSL configuration is included in policy updates for Kafka output. SSL is configured in so-elastic-fleet-setup
+  SSL_CONFIG=$(curl -K /opt/so/conf/elasticsearch/curl.config -L "http://localhost:5601/api/fleet/outputs/so-manager_kafka" | jq -r '.item.ssl')
 
-# Get current list of Logstash Outputs
+  JSON_STRING=$(jq -n \
-RAW_JSON=$(curl -K /opt/so/conf/elasticsearch/curl.config 'http://localhost:5601/api/fleet/outputs/so-manager_logstash')
+    --arg UPDATEDLIST "$NEW_LIST_JSON" \
+    --argjson SSL_CONFIG "$SSL_CONFIG" \
+    '{"name": "grid-kafka","type": "kafka","hosts": $UPDATEDLIST,"is_default": true,"is_default_monitoring": true,"config_yaml": "","ssl": $SSL_CONFIG}')
+  # Update Kafka outputs
+  curl -K /opt/so/conf/elasticsearch/curl.config -L -X PUT "localhost:5601/api/fleet/outputs/so-manager_kafka" -H 'kbn-xsrf: true' -H 'Content-Type: application/json' -d "$JSON_STRING" | jq
+}
 
-# Check to make sure that the server responded with good data - else, bail from script
+{% if GLOBALS.pipeline == "KAFKA" %}
-CHECKSUM=$(jq -r '.item.id' <<< "$RAW_JSON")
+  # Get current list of Kafka Outputs
-if [ "$CHECKSUM" != "so-manager_logstash" ]; then
+  RAW_JSON=$(curl -K /opt/so/conf/elasticsearch/curl.config 'http://localhost:5601/api/fleet/outputs/so-manager_kafka')
- printf "Failed to query for current Logstash Outputs..."
- exit 1
-fi
 
-# Get the current list of Logstash outputs & hash them
+  # Check to make sure that the server responded with good data - else, bail from script
-CURRENT_LIST=$(jq -c -r '.item.hosts' <<<  "$RAW_JSON")
+  CHECKSUM=$(jq -r '.item.id' <<< "$RAW_JSON")
-CURRENT_HASH=$(sha1sum <<< "$CURRENT_LIST" | awk '{print $1}')
+  if [ "$CHECKSUM" != "so-manager_kafka" ]; then
+   printf "Failed to query for current Kafka Outputs..."
+   exit 1
+  fi
 
-declare -a NEW_LIST=()
+  # Get the current list of kafka outputs & hash them
+  CURRENT_LIST=$(jq -c -r '.item.hosts' <<<  "$RAW_JSON")
+  CURRENT_HASH=$(sha1sum <<< "$CURRENT_LIST" | awk '{print $1}')
+
+  declare -a NEW_LIST=()
+
+  # Query for the current Grid Nodes that are running kafka
+  KAFKANODES=$(salt-call --out=json pillar.get kafka:nodes | jq '.local')
+
+  # Query for Kafka nodes with Broker role and add hostname to list
+  while IFS= read -r line; do
+    NEW_LIST+=("$line")
+  done < <(jq -r 'to_entries | .[] | select(.value.role | contains("broker")) | .key + ":9092"' <<< $KAFKANODES)
+
+  {# If global pipeline isn't set to KAFKA then assume default of REDIS / logstash #}
+{% else %}
+  # Get current list of Logstash Outputs
+  RAW_JSON=$(curl -K /opt/so/conf/elasticsearch/curl.config 'http://localhost:5601/api/fleet/outputs/so-manager_logstash')
+
+  # Check to make sure that the server responded with good data - else, bail from script
+  CHECKSUM=$(jq -r '.item.id' <<< "$RAW_JSON")
+  if [ "$CHECKSUM" != "so-manager_logstash" ]; then
+   printf "Failed to query for current Logstash Outputs..."
+   exit 1
+  fi
+
+  # Get the current list of Logstash outputs & hash them
+  CURRENT_LIST=$(jq -c -r '.item.hosts' <<<  "$RAW_JSON")
+  CURRENT_HASH=$(sha1sum <<< "$CURRENT_LIST" | awk '{print $1}')
+
+  declare -a NEW_LIST=()
+
+  {# If we select to not send to manager via SOC, then omit the code that adds manager to NEW_LIST #}
+  {% if ELASTICFLEETMERGED.enable_manager_output %}
+  # Create array & add initial elements
+  if [ "{{ GLOBALS.hostname }}" = "{{ GLOBALS.url_base }}" ]; then
+      NEW_LIST+=("{{ GLOBALS.url_base }}:5055")
+  else
+      NEW_LIST+=("{{ GLOBALS.url_base }}:5055" "{{ GLOBALS.hostname }}:5055")
+  fi
+  {% endif %}
+
+  # Query for FQDN entries & add them to the list
+  {% if ELASTICFLEETMERGED.config.server.custom_fqdn | length > 0 %}
+  CUSTOMFQDNLIST=('{{ ELASTICFLEETMERGED.config.server.custom_fqdn | join(' ') }}')
+  readarray -t -d ' ' CUSTOMFQDN < <(printf '%s' "$CUSTOMFQDNLIST")
+  for CUSTOMNAME in "${CUSTOMFQDN[@]}"
+  do
+   NEW_LIST+=("$CUSTOMNAME:5055")
+  done
+  {% endif %}
+
+  # Query for the current Grid Nodes that are running Logstash
+  LOGSTASHNODES=$(salt-call --out=json pillar.get logstash:nodes | jq '.local')
+
+  # Query for Receiver Nodes & add them to the list
+  if grep -q "receiver" <<< $LOGSTASHNODES; then
+     readarray -t RECEIVERNODES < <(jq -r ' .receiver | keys_unsorted[]'  <<< $LOGSTASHNODES)
+     for NODE in "${RECEIVERNODES[@]}"
+     do
+      NEW_LIST+=("$NODE:5055")
+     done
+  fi
+
+  # Query for Fleet Nodes & add them to the list
+  if grep -q "fleet" <<< $LOGSTASHNODES; then
+     readarray -t FLEETNODES < <(jq -r ' .fleet | keys_unsorted[]'  <<< $LOGSTASHNODES)
+     for NODE in "${FLEETNODES[@]}"
+     do
+      NEW_LIST+=("$NODE:5055")
+     done
+  fi
 
-{# If we select to not send to manager via SOC, then omit the code that adds manager to NEW_LIST #}
-{% if ELASTICFLEETMERGED.enable_manager_output %}
-# Create array & add initial elements
-if [ "{{ GLOBALS.hostname }}" = "{{ GLOBALS.url_base }}" ]; then
-    NEW_LIST+=("{{ GLOBALS.url_base }}:5055")
-else
-    NEW_LIST+=("{{ GLOBALS.url_base }}:5055" "{{ GLOBALS.hostname }}:5055")
-fi
 {% endif %}
 
-# Query for FQDN entries & add them to the list
-{% if ELASTICFLEETMERGED.config.server.custom_fqdn | length > 0 %}
-CUSTOMFQDNLIST=('{{ ELASTICFLEETMERGED.config.server.custom_fqdn | join(' ') }}')
-readarray -t -d ' ' CUSTOMFQDN < <(printf '%s' "$CUSTOMFQDNLIST")
-for CUSTOMNAME in "${CUSTOMFQDN[@]}"
-do
- NEW_LIST+=("$CUSTOMNAME:5055")
-done
-{% endif %}
-
-# Query for the current Grid Nodes that are running Logstash
-LOGSTASHNODES=$(salt-call --out=json pillar.get logstash:nodes | jq '.local')
-
-# Query for Receiver Nodes & add them to the list
-if grep -q "receiver" <<< $LOGSTASHNODES; then
-   readarray -t RECEIVERNODES < <(jq -r ' .receiver | keys_unsorted[]'  <<< $LOGSTASHNODES)
-   for NODE in "${RECEIVERNODES[@]}"
-   do
-    NEW_LIST+=("$NODE:5055")
-   done
-fi
-
-# Query for Fleet Nodes & add them to the list
-if grep -q "fleet" <<< $LOGSTASHNODES; then
-   readarray -t FLEETNODES < <(jq -r ' .fleet | keys_unsorted[]'  <<< $LOGSTASHNODES)
-   for NODE in "${FLEETNODES[@]}"
-   do
-    NEW_LIST+=("$NODE:5055")
-   done
-fi
-
 # Sort & hash the new list of Logstash Outputs
 NEW_LIST_JSON=$(jq --compact-output --null-input '$ARGS.positional' --args -- "${NEW_LIST[@]}")
 NEW_HASH=$(sha1sum <<< "$NEW_LIST_JSON" | awk '{print $1}')
@@ -87,9 +127,28 @@ NEW_HASH=$(sha1sum <<< "$NEW_LIST_JSON" | awk '{print $1}')
 if [ "$NEW_HASH" = "$CURRENT_HASH" ]; then
     printf "\nHashes match - no update needed.\n"
     printf "Current List: $CURRENT_LIST\nNew List: $NEW_LIST_JSON\n"
+
+    # Since output can be KAFKA or LOGSTASH, we need to check if the policy set as default matches the value set in GLOBALS.pipeline and update if needed
+    printf "Checking if the correct output policy is set as default\n"
+    OUTPUT_DEFAULT=$(jq -r '.item.is_default' <<< $RAW_JSON)
+    OUTPUT_DEFAULT_MONITORING=$(jq -r '.item.is_default_monitoring' <<< $RAW_JSON)
+    if [[ "$OUTPUT_DEFAULT" = "false" || "$OUTPUT_DEFAULT_MONITORING" = "false" ]]; then
+      printf "Default output policy needs to be updated.\n"
+    {%- if GLOBALS.pipeline == "KAFKA" and 'gmd' in salt['pillar.get']('features', []) %}
+      update_kafka_outputs
+    {%- else %}
+      update_logstash_outputs
+    {%- endif %}
+    else
+      printf "Default output policy is set - no update needed.\n"
+    fi
     exit 0
 else
     printf "\nHashes don't match - update needed.\n"
     printf "Current List: $CURRENT_LIST\nNew List: $NEW_LIST_JSON\n"
+    {%- if GLOBALS.pipeline == "KAFKA" and 'gmd' in salt['pillar.get']('features', []) %}
+    update_kafka_outputs
+    {%- else %}
     update_logstash_outputs
+    {%- endif %}
 fi

salt/elasticfleet/tools/sbin_jinja/so-elastic-fleet-setup

@@ -77,6 +77,11 @@ curl -K /opt/so/conf/elasticsearch/curl.config -L -X POST "localhost:5601/api/fl
 printf "\n\n"
 {%- endif %}
 
+printf "\nCreate Kafka Output Config if node is not an Import or Eval install\n"
+{% if grains.role not in ['so-import', 'so-eval'] %}
+salt-call state.apply kafka.elasticfleet queue=True
+{% endif %}
+
 # Add Manager Hostname & URL Base to Fleet Host URLs
 printf "\nAdd SO-Manager Fleet URL\n"
 if [ "{{ GLOBALS.hostname }}" = "{{ GLOBALS.url_base }}" ]; then

salt/elasticsearch/ca.sls

@@ -4,7 +4,7 @@
 # Elastic License 2.0.
 
 {% from 'allowed_states.map.jinja' import allowed_states %}
-{% if sls.split('.')[0] in allowed_states %}
+{% if sls.split('.')[0] in allowed_states or sls in allowed_states %}
 {%   from 'vars/globals.map.jinja' import GLOBALS %}
 
 # Move our new CA over so Elastic and Logstash can use SSL with the internal CA

salt/elasticsearch/files/ingest/.fleet_final_pipeline-1

@@ -84,6 +84,7 @@
     { "community_id":{ "if": "ctx.event?.dataset == 'endpoint.events.network'", "ignore_failure":true }  },
     { "set":        { "if": "ctx.event?.module == 'fim'", "override": true, "field": "event.module", "value": "file_integrity" } },
     { "rename":        { "if": "ctx.winlog?.provider_name == 'Microsoft-Windows-Windows Defender'", "ignore_missing": true, "field": "winlog.event_data.Threat Name", "target_field": "winlog.event_data.threat_name" } },
+    { "set":        { "if": "ctx?.metadata?.kafka != null" , "field": "kafka.id", "value": "{{metadata.kafka.partition}}{{metadata.kafka.offset}}{{metadata.kafka.timestamp}}", "ignore_failure": true } },
     { "remove":     { "field": [ "message2", "type", "fields", "category", "module", "dataset", "event.dataset_temp", "dataset_tag_temp", "module_temp" ], "ignore_missing": true, "ignore_failure": true } } 
   ],
   "on_failure": [

salt/firewall/containers.map.jinja

@@ -27,6 +27,7 @@
      'so-elastic-fleet',
      'so-elastic-fleet-package-registry',
      'so-influxdb',
+     'so-kafka',
      'so-kibana',
      'so-kratos',
      'so-logstash',
@@ -80,6 +81,7 @@
 {% set NODE_CONTAINERS = [
      'so-logstash',
      'so-redis',
+     'so-kafka'
 ] %}
 
 {% elif GLOBALS.role == 'so-idh' %}

salt/firewall/defaults.yaml

@@ -77,6 +77,7 @@ firewall:
     elastic_agent_data:
       tcp:
         - 5055
+        - 9092
       udp: []
     elastic_agent_update:
       tcp:
@@ -90,6 +91,10 @@ firewall:
       tcp:
         - 8086
       udp: []
+    kafka:
+      tcp:
+        - 9093
+      udp: []
     kibana:
       tcp:
         - 5601
@@ -364,6 +369,7 @@ firewall:
                 - elastic_agent_update
                 - localrules
                 - sensoroni
+                - kafka
             fleet:
               portgroups:
                 - elasticsearch_rest
@@ -434,6 +440,7 @@ firewall:
                 - elastic_agent_data
                 - elastic_agent_update
                 - sensoroni
+                - kafka
             analyst:
               portgroups:
                 - nginx
@@ -558,6 +565,7 @@ firewall:
                 - elastic_agent_update
                 - localrules
                 - sensoroni
+                - kafka
             fleet:
               portgroups:
                 - elasticsearch_rest
@@ -626,6 +634,7 @@ firewall:
                 - elastic_agent_data
                 - elastic_agent_update
                 - sensoroni
+                - kafka
             analyst:
               portgroups:
                 - nginx
@@ -753,7 +762,7 @@ firewall:
                 - beats_5044
                 - beats_5644
                 - beats_5056
-                - redis
+                - kafka
                 - elasticsearch_node
                 - elastic_agent_control
                 - elastic_agent_data
@@ -823,6 +832,7 @@ firewall:
                 - elastic_agent_data
                 - elastic_agent_update
                 - sensoroni
+                - kafka
             analyst:
               portgroups:
                 - nginx
@@ -1287,17 +1297,21 @@ firewall:
               portgroups:
                 - redis
                 - elastic_agent_data
+                - kafka
             manager:
               portgroups:
                 - elastic_agent_data
+                - kafka
             managersearch:
               portgroups:
                 - redis
                 - elastic_agent_data
+                - kafka
             self:
               portgroups:
                 - redis
                 - elastic_agent_data
+                - kafka
             beats_endpoint:
               portgroups:
                 - beats_5044

salt/firewall/soc_firewall.yaml

@@ -120,6 +120,9 @@ firewall:
     influxdb:
       tcp: *tcpsettings
       udp: *udpsettings
+    kafka:
+      tcp: *tcpsettings
+      udp: *udpsettings
     kibana:
       tcp: *tcpsettings
       udp: *udpsettings
@@ -939,7 +942,6 @@ firewall:
               portgroups: *portgroupshost
             customhostgroup9: 
               portgroups: *portgroupshost
-
     idh:
       chain:
         DOCKER-USER:

salt/global/defaults.yaml

@@ -1,2 +1,3 @@
 global:
   pcapengine: STENO
+  pipeline: REDIS

salt/global/soc_global.yaml

@@ -36,9 +36,10 @@ global:
     global: True
     advanced: True
   pipeline:
-    description: Sets which pipeline technology for events to use. Currently only Redis is supported.
+    description: Sets which pipeline technology for events to use. Currently only Redis is fully supported. Kafka is experimental and requires a Security Onion Pro license.
+    regex: ^(REDIS|KAFKA)$
+    regexFailureMessage: You must enter either REDIS or KAFKA.
     global: True
-    readonly: True
     advanced: True
   repo_host:
     description: Specify the host where operating system packages will be served from.

salt/kafka/config.map.jinja

@@ -0,0 +1,91 @@
+{# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+   or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at
+   https://securityonion.net/license; you may not use this file except in compliance with the
+   Elastic License 2.0. #}
+{% from 'kafka/map.jinja' import KAFKAMERGED %}
+{% from 'vars/globals.map.jinja' import GLOBALS %}
+
+{% set KAFKA_NODES_PILLAR = salt['pillar.get']('kafka:nodes') %}
+{% set KAFKA_PASSWORD = salt['pillar.get']('kafka:password') %}
+
+{# Create list of KRaft controllers #}
+{% set controllers = [] %}
+
+{# Check for Kafka nodes with controller in process_x_roles #}
+{% for node in KAFKA_NODES_PILLAR %}
+{%   if 'controller' in KAFKA_NODES_PILLAR[node].role %}
+{%     do controllers.append(KAFKA_NODES_PILLAR[node].nodeid ~ "@" ~ node ~ ":9093") %}
+{%   endif %}
+{% endfor %}
+
+{% set kafka_controller_quorum_voters = ','.join(controllers) %}
+
+{# By default all Kafka eligible nodes are given the role of broker, except for
+     grid MANAGER (broker,controller) until overridden through SOC UI #}
+{% set node_type = salt['pillar.get']('kafka:nodes:'+ GLOBALS.hostname + ':role') %}
+
+{# Generate server.properties for 'broker' , 'controller', 'broker,controller' node types
+    anything above this line is a configuration needed for ALL Kafka nodes #}
+{% if node_type == 'broker' %}
+{%   do KAFKAMERGED.config.broker.update({'advertised_x_listeners': 'BROKER://'+ GLOBALS.node_ip +':9092' }) %}
+{%   do KAFKAMERGED.config.broker.update({'controller_x_quorum_x_voters': kafka_controller_quorum_voters }) %}
+{%   do KAFKAMERGED.config.broker.update({'node_x_id': salt['pillar.get']('kafka:nodes:'+ GLOBALS.hostname +':nodeid') }) %}
+{%   do KAFKAMERGED.config.broker.update({'ssl_x_keystore_x_password': KAFKA_PASSWORD }) %}
+
+{# Nodes with only the 'broker' role need to have the below settings for communicating with controller nodes #}
+{%   do KAFKAMERGED.config.broker.update({'controller_x_listener_x_names': KAFKAMERGED.config.controller.controller_x_listener_x_names }) %}
+{%   do KAFKAMERGED.config.broker.update({
+       'listener_x_security_x_protocol_x_map': KAFKAMERGED.config.broker.listener_x_security_x_protocol_x_map
+        + ',' + KAFKAMERGED.config.controller.listener_x_security_x_protocol_x_map })
+  %}
+{% endif %}
+
+{% if node_type == 'controller' %}
+{%   do KAFKAMERGED.config.controller.update({'controller_x_quorum_x_voters': kafka_controller_quorum_voters }) %}
+{%   do KAFKAMERGED.config.controller.update({'node_x_id': salt['pillar.get']('kafka:nodes:'+ GLOBALS.hostname +':nodeid') }) %}
+{%   do KAFKAMERGED.config.controller.update({'ssl_x_keystore_x_password': KAFKA_PASSWORD }) %}
+
+{% endif %}
+
+{# Kafka nodes of this type are not recommended for use outside of development / testing. #}
+{% if node_type == 'broker,controller' %}
+{%   do KAFKAMERGED.config.broker.update({'advertised_x_listeners': 'BROKER://'+ GLOBALS.node_ip +':9092' }) %}
+{%   do KAFKAMERGED.config.broker.update({'controller_x_listener_x_names': KAFKAMERGED.config.controller.controller_x_listener_x_names }) %}
+{%   do KAFKAMERGED.config.broker.update({'controller_x_quorum_x_voters': kafka_controller_quorum_voters }) %}
+{%   do KAFKAMERGED.config.broker.update({'node_x_id': salt['pillar.get']('kafka:nodes:'+ GLOBALS.hostname +':nodeid') }) %}
+{%   do KAFKAMERGED.config.broker.update({'process_x_roles': 'broker,controller' }) %}
+{%   do KAFKAMERGED.config.broker.update({'ssl_x_keystore_x_password': KAFKA_PASSWORD }) %}
+
+{%   do KAFKAMERGED.config.broker.update({
+       'listeners': KAFKAMERGED.config.broker.listeners + ',' + KAFKAMERGED.config.controller.listeners })
+  %}
+
+{%   do KAFKAMERGED.config.broker.update({
+       'listener_x_security_x_protocol_x_map': KAFKAMERGED.config.broker.listener_x_security_x_protocol_x_map
+        + ',' + KAFKAMERGED.config.controller.listener_x_security_x_protocol_x_map })
+  %}
+
+{% endif %}
+
+{# If a password other than PLACEHOLDER isn't set remove it from the server.properties #}
+{% if KAFKAMERGED.config.broker.ssl_x_truststore_x_password == 'PLACEHOLDER' %}
+{%   do KAFKAMERGED.config.broker.pop('ssl_x_truststore_x_password') %}
+{% endif %}
+
+{% if KAFKAMERGED.config.controller.ssl_x_truststore_x_password == 'PLACEHOLDER' %}
+{%   do KAFKAMERGED.config.controller.pop('ssl_x_truststore_x_password') %}
+{% endif %}
+
+{# Client properties stuff #}
+{% if KAFKAMERGED.config.client.ssl_x_truststore_x_password == 'PLACEHOLDER' %}
+{%   do KAFKAMERGED.config.client.pop('ssl_x_truststore_x_password') %}
+{% endif %}
+{% do KAFKAMERGED.config.client.update({'ssl_x_keystore_x_password': KAFKA_PASSWORD }) %}
+
+{% if 'broker' in node_type %}
+{%   set KAFKACONFIG = KAFKAMERGED.config.broker %}
+{% else %}
+{%   set KAFKACONFIG = KAFKAMERGED.config.controller %}
+{% endif %}
+
+{% set KAFKACLIENT = KAFKAMERGED.config.client %}

salt/kafka/config.sls

@@ -0,0 +1,80 @@
+# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at 
+# https://securityonion.net/license; you may not use this file except in compliance with the
+# Elastic License 2.0.
+
+{% from 'allowed_states.map.jinja' import allowed_states %}
+{% if sls.split('.')[0] in allowed_states %}
+{%   from 'vars/globals.map.jinja' import GLOBALS %}
+
+include:
+  - ssl
+
+kafka_group:
+  group.present:
+    - name: kafka
+    - gid: 960
+
+kafka:
+  user.present:
+    - uid: 960
+    - gid: 960
+
+kafka_sbin_tools:
+  file.recurse:
+    - name: /usr/sbin
+    - source: salt://kafka/tools/sbin
+    - user: 960
+    - group: 960
+    - file_mode: 755
+
+kafka_sbin_jinja_tools:
+  file.recurse:
+    - name: /usr/sbin
+    - source: salt://kafka/tools/sbin_jinja
+    - user: 960
+    - group: 960
+    - file_mode: 755
+    - template: jinja
+    - defaults:
+        GLOBALS: {{ GLOBALS }}
+
+kakfa_log_dir:
+  file.directory:
+    - name: /opt/so/log/kafka
+    - user: 960
+    - group: 960
+    - makedirs: True
+
+kafka_data_dir:
+  file.directory:
+    - name: /nsm/kafka/data
+    - user: 960
+    - group: 960
+    - makedirs: True
+
+{%   for sc in ['server', 'client'] %}
+kafka_kraft_{{sc}}_properties:
+  file.managed:
+    - source: salt://kafka/etc/{{sc}}.properties.jinja
+    - name: /opt/so/conf/kafka/{{sc}}.properties
+    - template: jinja
+    - user: 960
+    - group: 960
+    - makedirs: True
+    - show_changes: False
+{%   endfor %}
+
+reset_quorum_on_changes:
+  cmd.run:
+    - name: rm -f /nsm/kafka/data/__cluster_metadata-0/quorum-state
+    - onchanges:
+      - file: /opt/so/conf/kafka/server.properties
+
+{% else %}
+
+{{sls}}_state_not_allowed:
+  test.fail_without_changes:
+    - name: {{sls}}_state_not_allowed
+
+{% endif %}

salt/kafka/defaults.yaml

@@ -0,0 +1,62 @@
+kafka:
+  enabled: False
+  cluster_id:
+  password:
+  controllers:
+  reset:
+  config:
+    broker:
+      advertised_x_listeners:
+      auto_x_create_x_topics_x_enable: true
+      controller_x_quorum_x_voters:
+      default_x_replication_x_factor: 1
+      inter_x_broker_x_listener_x_name: BROKER
+      listeners: BROKER://0.0.0.0:9092
+      listener_x_security_x_protocol_x_map: BROKER:SSL
+      log_x_dirs: /nsm/kafka/data
+      log_x_retention_x_check_x_interval_x_ms: 300000
+      log_x_retention_x_hours: 168
+      log_x_segment_x_bytes: 1073741824
+      node_x_id:
+      num_x_io_x_threads: 8
+      num_x_network_x_threads: 3
+      num_x_partitions: 3
+      num_x_recovery_x_threads_x_per_x_data_x_dir: 1
+      offsets_x_topic_x_replication_x_factor: 1
+      process_x_roles: broker
+      socket_x_receive_x_buffer_x_bytes: 102400
+      socket_x_request_x_max_x_bytes: 104857600
+      socket_x_send_x_buffer_x_bytes: 102400
+      ssl_x_keystore_x_location: /etc/pki/kafka.p12
+      ssl_x_keystore_x_type: PKCS12
+      ssl_x_keystore_x_password:
+      ssl_x_truststore_x_location: /etc/pki/java/sos/cacerts
+      ssl_x_truststore_x_password: PLACEHOLDER
+      ssl_x_truststore_x_type: PEM
+      transaction_x_state_x_log_x_min_x_isr: 1
+      transaction_x_state_x_log_x_replication_x_factor: 1
+    client:
+      security_x_protocol: SSL
+      ssl_x_truststore_x_location: /etc/pki/java/sos/cacerts
+      ssl_x_truststore_x_password: PLACEHOLDER
+      ssl_x_truststore_x_type: PEM
+      ssl_x_keystore_x_location: /etc/pki/kafka.p12
+      ssl_x_keystore_x_type: PKCS12
+      ssl_x_keystore_x_password:
+    controller:
+      controller_x_listener_x_names: CONTROLLER
+      controller_x_quorum_x_voters:
+      listeners: CONTROLLER://0.0.0.0:9093
+      listener_x_security_x_protocol_x_map: CONTROLLER:SSL
+      log_x_dirs: /nsm/kafka/data
+      log_x_retention_x_check_x_interval_x_ms: 300000
+      log_x_retention_x_hours: 168
+      log_x_segment_x_bytes: 1073741824
+      node_x_id:
+      process_x_roles: controller
+      ssl_x_keystore_x_location: /etc/pki/kafka.p12
+      ssl_x_keystore_x_type: PKCS12
+      ssl_x_keystore_x_password:
+      ssl_x_truststore_x_location: /etc/pki/java/sos/cacerts
+      ssl_x_truststore_x_password: PLACEHOLDER
+      ssl_x_truststore_x_type: PEM

salt/kafka/disabled.sls

@@ -0,0 +1,21 @@
+# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at 
+# https://securityonion.net/license; you may not use this file except in compliance with the
+# Elastic License 2.0.
+
+so-kafka:
+  docker_container.absent:
+    - force: True
+
+so-kafka_so-status.disabled:
+  file.comment:
+    - name: /opt/so/conf/so-status/so-status.conf
+    - regex: ^so-kafka$
+
+{% if grains.role in ['so-manager','so-managersearch','so-standalone'] %}
+ensure_default_pipeline:
+  cmd.run:
+    - name: |
+        /usr/sbin/so-yaml.py replace /opt/so/saltstack/local/pillar/kafka/soc_kafka.sls kafka.enabled False;
+        /usr/sbin/so-yaml.py replace /opt/so/saltstack/local/pillar/global/soc_global.sls global.pipeline REDIS
+{% endif %}

salt/kafka/elasticfleet.sls

@@ -0,0 +1,24 @@
+# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at 
+# https://securityonion.net/license; you may not use this file except in compliance with the
+# Elastic License 2.0.
+
+{%   from 'vars/globals.map.jinja' import GLOBALS %}
+include:
+  - elasticfleet.enabled
+
+{# Create Kafka output policy if it doesn't exist #}
+update_kafka_output_policy_script:
+  file.managed:
+    - name: /usr/sbin/so-kafka-fleet-output-policy
+    - source: salt://kafka/tools/sbin_jinja/so-kafka-fleet-output-policy
+    - user: root
+    - mode: 755
+    - template: jinja
+    - defaults:
+        GLOBALS: {{ GLOBALS }}
+
+create_kafka_output_policy:
+  cmd.run:
+    - name: 'so-kafka-fleet-output-policy > /dev/null 2>&1'
+    - show_changes: false

salt/kafka/enabled.sls

@@ -0,0 +1,86 @@
+# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at
+# https://securityonion.net/license; you may not use this file except in compliance with the
+# Elastic License 2.0.
+#
+# Note: Per the Elastic License 2.0, the second limitation states:
+#
+#   "You may not move, change, disable, or circumvent the license key functionality
+#    in the software, and you may not remove or obscure any functionality in the
+#    software that is protected by the license key."
+
+{% from 'allowed_states.map.jinja' import allowed_states %}
+{% if sls.split('.')[0] in allowed_states %}
+{%   from 'vars/globals.map.jinja' import GLOBALS %}
+{%   from 'docker/docker.map.jinja' import DOCKER %}
+{%   set KAFKANODES = salt['pillar.get']('kafka:nodes') %}
+{%   if 'gmd' in salt['pillar.get']('features', []) %}
+
+include:
+  - elasticsearch.ca
+  - kafka.sostatus
+  - kafka.config
+  - kafka.storage
+
+so-kafka:
+  docker_container.running:
+    - image: {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-kafka:{{ GLOBALS.so_version }}
+    - hostname: so-kafka
+    - name: so-kafka
+    - networks:
+      - sobridge:
+        - ipv4_address: {{ DOCKER.containers['so-kafka'].ip }}
+    - user: kafka
+    - environment:
+        KAFKA_HEAP_OPTS: -Xmx2G -Xms1G
+        KAFKA_OPTS: -javaagent:/opt/jolokia/agents/jolokia-agent-jvm-javaagent.jar=port=8778,host={{ DOCKER.containers['so-kafka'].ip }},policyLocation=file:/opt/jolokia/jolokia.xml
+    - extra_hosts:
+      {% for node in KAFKANODES %}
+      - {{ node }}:{{ KAFKANODES[node].ip }}
+      {% endfor %}
+      {% if DOCKER.containers['so-kafka'].extra_hosts %}
+      {%   for XTRAHOST in DOCKER.containers['so-kafka'].extra_hosts %}
+      - {{ XTRAHOST }}
+      {%   endfor %}
+      {% endif %}
+    - port_bindings:
+      {% for BINDING in DOCKER.containers['so-kafka'].port_bindings %}
+      - {{ BINDING }}
+      {% endfor %}
+    - binds:
+      - /etc/pki/kafka.p12:/etc/pki/kafka.p12:ro
+      - /etc/pki/tls/certs/intca.crt:/etc/pki/java/sos/cacerts:ro
+      - /nsm/kafka/data/:/nsm/kafka/data/:rw
+      - /opt/so/log/kafka:/opt/kafka/logs/:rw
+      - /opt/so/conf/kafka/server.properties:/opt/kafka/config/kraft/server.properties:ro
+      - /opt/so/conf/kafka/client.properties:/opt/kafka/config/kraft/client.properties
+    - watch:
+      {% for sc in ['server', 'client'] %}
+      - file: kafka_kraft_{{sc}}_properties
+      {% endfor %}
+
+delete_so-kafka_so-status.disabled:
+  file.uncomment:
+    - name: /opt/so/conf/so-status/so-status.conf
+    - regex: ^so-kafka$
+
+{%   else %}
+
+{{sls}}_no_license_detected:
+  test.fail_without_changes:
+    - name: {{sls}}_no_license_detected
+    - comment:
+      - "Kafka for Guaranteed Message Delivery is a feature supported only for customers with a valid license.
+      Contact Security Onion Solutions, LLC via our website at https://securityonionsolutions.com
+      for more information about purchasing a license to enable this feature."
+include:
+  - kafka.disabled
+{%   endif %}
+
+{% else %}
+
+{{sls}}_state_not_allowed:
+  test.fail_without_changes:
+    - name: {{sls}}_state_not_allowed
+
+{% endif %}

salt/kafka/etc/client.properties.jinja

@@ -0,0 +1,7 @@
+{# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+   or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at 
+   https://securityonion.net/license; you may not use this file except in compliance with the
+   Elastic License 2.0. #}
+
+{% from 'kafka/config.map.jinja' import KAFKACLIENT -%}
+{{ KAFKACLIENT | yaml(False) | replace("_x_", ".") }}

salt/kafka/etc/server.properties.jinja

@@ -0,0 +1,7 @@
+{# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+   or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at 
+   https://securityonion.net/license; you may not use this file except in compliance with the
+   Elastic License 2.0. #}
+
+{% from 'kafka/config.map.jinja' import KAFKACONFIG -%}
+{{ KAFKACONFIG | yaml(False) | replace("_x_", ".") }}

salt/kafka/files/managed_node_pillar.jinja

@@ -0,0 +1,10 @@
+kafka:
+  nodes:
+{% for node, values in COMBINED_KAFKANODES.items() %}
+    {{ node }}:
+      ip: {{ values['ip'] }}
+      nodeid: {{ values['nodeid'] }}
+{%-   if values['role'] != none %}
+      role: {{ values['role'] }}
+{%-   endif %}
+{% endfor %}

salt/kafka/init.sls

@@ -0,0 +1,25 @@
+# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at
+# https://securityonion.net/license; you may not use this file except in compliance with the
+# Elastic License 2.0.
+#
+# Note: Per the Elastic License 2.0, the second limitation states:
+#
+#   "You may not move, change, disable, or circumvent the license key functionality
+#    in the software, and you may not remove or obscure any functionality in the
+#    software that is protected by the license key."
+
+{% from 'kafka/map.jinja' import KAFKAMERGED %}
+{% from 'vars/globals.map.jinja' import GLOBALS %}
+
+include:
+{# Run kafka/nodes.sls before Kafka is enabled, so kafka nodes pillar is setup #}
+{% if grains.role in ['so-manager','so-managersearch', 'so-standalone'] %}
+  - kafka.nodes
+  - kafka.elasticfleet
+{% endif %}
+{% if GLOBALS.pipeline == "KAFKA" and KAFKAMERGED.enabled %}
+  - kafka.enabled
+{% else %}
+  - kafka.disabled
+{% endif %}

salt/kafka/map.jinja

@@ -0,0 +1,10 @@
+{# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+   or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at 
+   https://securityonion.net/license; you may not use this file except in compliance with the
+   Elastic License 2.0. #}
+
+{# This is only used to determine if Kafka is enabled / disabled. Configuration is found in kafka/config.map.jinja #}
+{# kafka/config.map.jinja depends on there being a kafka nodes pillar being populated #}
+
+{% import_yaml 'kafka/defaults.yaml' as KAFKADEFAULTS %}
+{% set KAFKAMERGED = salt['pillar.get']('kafka', KAFKADEFAULTS.kafka, merge=True) %}

salt/kafka/nodes.map.jinja

@@ -0,0 +1,92 @@
+{# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+   or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at 
+   https://securityonion.net/license; you may not use this file except in compliance with the
+   Elastic License 2.0. #}
+
+{# USED TO GENERATE PILLAR/KAFKA/NODES.SLS. #}
+{% import_yaml 'kafka/defaults.yaml' as KAFKADEFAULTS %}
+{% from 'vars/globals.map.jinja' import GLOBALS %}
+
+{% set process_x_roles = KAFKADEFAULTS.kafka.config.broker.process_x_roles %}
+
+{% set current_kafkanodes = salt.saltutil.runner(
+        'mine.get',
+         tgt='G@role:so-manager or G@role:so-managersearch or G@role:so-standalone or G@role:so-receiver',
+         fun='network.ip_addrs',
+         tgt_type='compound') %}
+
+{% set STORED_KAFKANODES = salt['pillar.get']('kafka:nodes', default=None) %}
+{% set KAFKA_CONTROLLERS_PILLAR = salt['pillar.get']('kafka:controllers', default=None) %}
+
+{% set existing_ids = [] %}
+
+{# Check STORED_KAFKANODES for existing kafka nodes and pull their IDs so they are not reused across the grid #}
+{% if STORED_KAFKANODES != none %}
+{%   for node, values in STORED_KAFKANODES.items() %}
+{%     if values.get('nodeid') %}
+{%       do existing_ids.append(values['nodeid']) %}
+{%     endif %}
+{%   endfor %}
+{% endif %}
+
+{# Create list of possible node ids #}
+{% set all_possible_ids = range(1, 65536)|list %}
+
+{# Create list of available node ids by looping through all_possible_ids and ensuring it isn't in existing_ids #}
+{% set available_ids = [] %}
+{% for id in all_possible_ids %}
+{%   if id not in existing_ids %}
+{%     do available_ids.append(id) %}
+{%   endif %}
+{% endfor %}
+
+{# Collect kafka eligible nodes and check if they're already in STORED_KAFKANODES to avoid potentially reassigning a nodeid #}
+{% set NEW_KAFKANODES = {} %}
+{% for minionid, ip in current_kafkanodes.items() %}
+{%   set hostname = minionid.split('_')[0] %}
+{%   if STORED_KAFKANODES != none and hostname not in STORED_KAFKANODES.items() %}
+{%     set new_id = available_ids.pop(0) %}
+{%     do NEW_KAFKANODES.update({hostname: {'nodeid': new_id, 'ip': ip[0], 'role': process_x_roles }}) %}
+{%   endif %}
+{%   if hostname not in NEW_KAFKANODES.items() %}
+{%     set new_id = available_ids.pop(0) %}
+{%     do NEW_KAFKANODES.update({hostname: {'nodeid': new_id, 'ip': ip[0], 'role': process_x_roles }}) %}
+{%   endif %}
+{% endfor %}
+
+{# Combine STORED_KAFKANODES and NEW_KAFKANODES for writing to the pillar/kafka/nodes.sls #}
+{% set COMBINED_KAFKANODES = {} %}
+{% for node, details in NEW_KAFKANODES.items() %}
+{%   do COMBINED_KAFKANODES.update({node: details}) %}
+{% endfor %}
+{% if STORED_KAFKANODES != none %}
+{%  for node, details in STORED_KAFKANODES.items() %}
+{%    do COMBINED_KAFKANODES.update({node: details}) %}
+{%  endfor %}
+{% endif %}
+
+{# Update the process_x_roles value for any host in the kafka_controllers_pillar configured from SOC UI #}
+{% set ns = namespace(has_controller=false) %}
+{% if KAFKA_CONTROLLERS_PILLAR != none %}
+{%   set KAFKA_CONTROLLERS_PILLAR_LIST = KAFKA_CONTROLLERS_PILLAR.split(',') %}
+{%   for hostname in KAFKA_CONTROLLERS_PILLAR_LIST %}
+{%     if hostname in COMBINED_KAFKANODES %}
+{%       do COMBINED_KAFKANODES[hostname].update({'role': 'controller'}) %}
+{%       set ns.has_controller = true %}
+{%     endif %}
+{%   endfor %}
+{%   for hostname in COMBINED_KAFKANODES %}
+{%     if hostname not in KAFKA_CONTROLLERS_PILLAR_LIST %}
+{%       do COMBINED_KAFKANODES[hostname].update({'role': 'broker'}) %}
+{%     endif %}
+{%   endfor %}
+{# If the kafka_controllers_pillar is NOT empty check that atleast one node contains the controller role.
+     otherwise default to GLOBALS.manager having broker,controller role #}
+{%   if not ns.has_controller %}
+{%     do COMBINED_KAFKANODES[GLOBALS.manager].update({'role': 'broker,controller'}) %}
+{%   endif %}
+{# If kafka_controllers_pillar is empty, default to having grid manager as 'broker,controller'
+   so there is always atleast 1 controller in the cluster #}
+{% else %}
+{%   do COMBINED_KAFKANODES[GLOBALS.manager].update({'role': 'broker,controller'}) %}
+{% endif %}

salt/kafka/nodes.sls

@@ -0,0 +1,18 @@
+# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at 
+# https://securityonion.net/license; you may not use this file except in compliance with the
+# Elastic License 2.0.
+
+{% from 'kafka/nodes.map.jinja' import COMBINED_KAFKANODES %}
+{% set kafka_cluster_id = salt['pillar.get']('kafka:cluster_id', default=None) %}
+
+{# Write Kafka pillar, so all grid members have access to nodeid of other kafka nodes and their roles #}
+write_kafka_pillar_yaml:
+  file.managed:
+    - name: /opt/so/saltstack/local/pillar/kafka/nodes.sls
+    - mode: 644
+    - user: socore
+    - source: salt://kafka/files/managed_node_pillar.jinja
+    - template: jinja
+    - context:
+        COMBINED_KAFKANODES: {{ COMBINED_KAFKANODES }}

salt/kafka/reset.sls

@@ -0,0 +1,9 @@
+# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at 
+# https://securityonion.net/license; you may not use this file except in compliance with the
+# Elastic License 2.0.
+
+wipe_kafka_data:
+  file.absent:
+    - name: /nsm/kafka/data/
+    - force: True

salt/kafka/soc_kafka.yaml

@@ -0,0 +1,209 @@
+kafka:
+  enabled:
+    description: Enable or disable Kafka. Recommended to have desired configuration staged prior to enabling Kafka. Configure controllers with the hostnames of the nodes you want to act as controllers, join all receiver nodes to grid that will be converted to Kafka nodes, and configure the default_replication_factor to the desired value for your redundancy needs.
+    helpLink: kafka.html
+  cluster_id:
+    description: The ID of the Kafka cluster.
+    readonly: True
+    advanced: True
+    sensitive: True
+    helpLink: kafka.html
+  password:
+    description: The password to use for the Kafka certificates.
+    sensitive: True
+    helpLink: kafka.html
+  controllers:
+    description: A comma-seperated list of Security Onion hosts that will act as Kafka controllers. These hosts will be responsible for managing the Kafka cluster. WARNING - The hostnames of receiver nodes intended to be controllers should be added here BEFORE they have joined the Security Onion grid or BEFORE enabling KAFKA. This is to ensure that data is not lost by converting a data broker to a controller. Failure to do so may result in topics becoming unavailable and requiring manual intervention to repair or resetting Kafka data.
+    forcedType: "string"
+    helpLink: kafka.html
+  reset:
+    description: Disable and reset the Kafka cluster. This will remove all Kafka data including logs that may have not yet been ingested into Elasticsearch and reverts the grid to using REDIS as the global pipeline. This is useful when testing different Kafka configurations such as rearranging Kafka brokers / controllers allowing you to reset the cluster rather than manually fixing any issues arising from attempting to reassign a Kafka broker into a controller. Enter 'YES_RESET_KAFKA' and submit to disable and reset Kafka. Make any configuration changes required and re-enable Kafka when ready. This action CANNOT be reversed.
+    advanced: True
+    helpLink: kafka.html
+  config:
+    broker:
+      advertised_x_listeners:
+        description: Specify the list of listeners (hostname and port) that Kafka brokers provide to clients for communication.
+        title: advertised.listeners
+        helpLink: kafka.html
+      auto_x_create_x_topics_x_enable:
+        description: Enable the auto creation of topics.
+        title: auto.create.topics.enable
+        forcedType: bool
+        helpLink: kafka.html
+      default_x_replication_x_factor:
+        description: The default replication factor for automatically created topics. This value must be less than the amount of brokers in the cluster. Hosts specified in controllers should not be counted towards total broker count.
+        title: default.replication.factor
+        forcedType: int
+        helpLink: kafka.html
+      inter_x_broker_x_listener_x_name:
+        description: The name of the listener used for inter-broker communication.
+        title: inter.broker.listener.name
+        helpLink: kafka.html
+      listeners:
+        description: Set of URIs that is listened on and the listener names in a comma-seperated list.
+        helpLink: kafka.html
+      listener_x_security_x_protocol_x_map:
+        description: Comma-seperated mapping of listener name and security protocols.
+        title: listener.security.protocol.map
+        helpLink: kafka.html
+      log_x_dirs:
+        description: Where Kafka logs are stored within the Docker container.
+        title: log.dirs
+        helpLink: kafka.html
+      log_x_retention_x_check_x_interval_x_ms:
+        description: Frequency at which log files are checked if they are qualified for deletion.
+        title: log.retention.check.interval.ms
+        helpLink: kafka.html
+      log_x_retention_x_hours:
+        description: How long, in hours, a log file is kept.
+        title: log.retention.hours
+        forcedType: int
+        helpLink: kafka.html
+      log_x_segment_x_bytes:
+        description: The maximum allowable size for a log file.
+        title: log.segment.bytes
+        forcedType: int
+        helpLink: kafka.html
+      num_x_io_x_threads:
+        description: The number of threads used by Kafka.
+        title: num.io.threads
+        forcedType: int
+        helpLink: kafka.html
+      num_x_network_x_threads:
+        description: The number of threads used for network communication.
+        title: num.network.threads
+        forcedType: int
+        helpLink: kafka.html
+      num_x_partitions:
+        description: The number of log partitions assigned per topic.
+        title: num.partitions
+        forcedType: int
+        helpLink: kafka.html
+      num_x_recovery_x_threads_x_per_x_data_x_dir:
+        description: The number of threads used for log recuperation at startup and purging at shutdown. This ammount of threads is used per data directory.
+        title: num.recovery.threads.per.data.dir
+        forcedType: int
+        helpLink: kafka.html
+      offsets_x_topic_x_replication_x_factor:
+        description: The offsets topic replication factor.
+        title: offsets.topic.replication.factor
+        forcedType: int
+        helpLink: kafka.html
+      process_x_roles:
+        description: The role performed by Kafka brokers.
+        title: process.roles
+        readonly: True
+        helpLink: kafka.html
+      socket_x_receive_x_buffer_x_bytes:
+        description: Size, in bytes of the SO_RCVBUF buffer. A value of -1 will use the OS default.
+        title: socket.receive.buffer.bytes
+        #forcedType: int - soc needs to allow -1 as an int before we can use this
+        helpLink: kafka.html
+      socket_x_request_x_max_x_bytes:
+        description: The maximum bytes allowed for a request to the socket.
+        title: socket.request.max.bytes
+        forcedType: int
+        helpLink: kafka.html
+      socket_x_send_x_buffer_x_bytes:
+        description: Size, in bytes of the SO_SNDBUF buffer. A value of -1 will use the OS default.
+        title: socket.send.buffer.byte
+        #forcedType: int - soc needs to allow -1 as an int before we can use this
+        helpLink: kafka.html
+      ssl_x_keystore_x_location:
+        description: The key store file location within the Docker container.
+        title: ssl.keystore.location
+        helpLink: kafka.html
+      ssl_x_keystore_x_password:
+        description: The key store file password. Invalid for PEM format.
+        title: ssl.keystore.password
+        sensitive: True
+        helpLink: kafka.html
+      ssl_x_keystore_x_type: 
+        description: The key store file format.
+        title: ssl.keystore.type
+        regex: ^(JKS|PKCS12|PEM)$
+        helpLink: kafka.html
+      ssl_x_truststore_x_location:
+        description: The trust store file location within the Docker container.
+        title: ssl.truststore.location
+        helpLink: kafka.html
+      ssl_x_truststore_x_password:
+        description: The trust store file password. If null, the trust store file is still use, but integrity checking is disabled. Invalid for PEM format.
+        title: ssl.truststore.password
+        sensitive: True
+        helpLink: kafka.html
+      transaction_x_state_x_log_x_min_x_isr:
+        description: Overrides min.insync.replicas for the transaction topic. When a producer configures acks to "all" (or "-1"), this setting determines the minimum number of replicas required to acknowledge a write as successful. Failure to meet this minimum triggers an exception (either NotEnoughReplicas or NotEnoughReplicasAfterAppend). When used in conjunction, min.insync.replicas and acks enable stronger durability guarantees. For instance, creating a topic with a replication factor of 3, setting min.insync.replicas to 2, and using acks of "all" ensures that the producer raises an exception if a majority of replicas fail to receive a write.
+        title: transaction.state.log.min.isr
+        forcedType: int
+        helpLink: kafka.html
+      transaction_x_state_x_log_x_replication_x_factor:
+        description: Set the replication factor higher for the transaction topic to ensure availability. Internal topic creation will not proceed until the cluster size satisfies this replication factor prerequisite.
+        title: transaction.state.log.replication.factor
+        forcedType: int
+        helpLink: kafka.html
+    client:
+      security_x_protocol:
+        description: 'Broker communication protocol. Options are: SASL_SSL, PLAINTEXT, SSL, SASL_PLAINTEXT'
+        title: security.protocol
+        regex: ^(SASL_SSL|PLAINTEXT|SSL|SASL_PLAINTEXT)
+        helpLink: kafka.html
+      ssl_x_keystore_x_location:
+        description: The key store file location within the Docker container.
+        title: ssl.keystore.location
+        helpLink: kafka.html
+      ssl_x_keystore_x_password:
+        description: The key store file password. Invalid for PEM format.
+        title: ssl.keystore.password
+        sensitive: True
+        helpLink: kafka.html
+      ssl_x_keystore_x_type:
+        description: The key store file format.
+        title: ssl.keystore.type
+        regex: ^(JKS|PKCS12|PEM)$
+        helpLink: kafka.html
+      ssl_x_truststore_x_location:
+        description: The trust store file location within the Docker container.
+        title: ssl.truststore.location
+        helpLink: kafka.html
+      ssl_x_truststore_x_password:
+        description: The trust store file password. If null, the trust store file is still use, but integrity checking is disabled. Invalid for PEM format.
+        title: ssl.truststore.password
+        sensitive: True
+        helpLink: kafka.html
+    controller:
+      controller_x_listener_x_names:
+        description: Set listeners used by the controller in a comma-seperated list.
+        title: controller.listener.names
+        helpLink: kafka.html
+      listeners:
+        description: Set of URIs that is listened on and the listener names in a comma-seperated list.
+        helpLink: kafka.html
+      listener_x_security_x_protocol_x_map:
+        description: Comma-seperated mapping of listener name and security protocols.
+        title: listener.security.protocol.map
+        helpLink: kafka.html
+      log_x_dirs:
+        description: Where Kafka logs are stored within the Docker container.
+        title: log.dirs
+        helpLink: kafka.html
+      log_x_retention_x_check_x_interval_x_ms:
+        description: Frequency at which log files are checked if they are qualified for deletion.
+        title: log.retention.check.interval.ms
+        helpLink: kafka.html
+      log_x_retention_x_hours:
+        description: How long, in hours, a log file is kept.
+        title: log.retention.hours
+        forcedType: int
+        helpLink: kafka.html
+      log_x_segment_x_bytes:
+        description: The maximum allowable size for a log file.
+        title: log.segment.bytes
+        forcedType: int
+        helpLink: kafka.html
+      process_x_roles:
+        description: The role performed by controller node.
+        title: process.roles
+        readonly: True
+        helpLink: kafka.html

salt/kafka/sostatus.sls

@@ -0,0 +1,21 @@
+# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at 
+# https://securityonion.net/license; you may not use this file except in compliance with the
+# Elastic License 2.0.
+
+{% from 'allowed_states.map.jinja' import allowed_states %}
+{% if sls.split('.')[0] in allowed_states %}
+
+append_so-kafka_so-status.conf:
+  file.append:
+    - name: /opt/so/conf/so-status/so-status.conf
+    - text: so-kafka
+    - unless: grep -q so-kafka /opt/so/conf/so-status/so-status.conf
+
+{% else %}
+
+{{sls}}_state_not_allowed:
+  test.fail_without_changes:
+    - name: {{sls}}_state_not_allowed
+
+{% endif %}

salt/kafka/storage.sls

@@ -0,0 +1,30 @@
+# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at 
+# https://securityonion.net/license; you may not use this file except in compliance with the
+# Elastic License 2.0.
+
+{% from 'allowed_states.map.jinja' import allowed_states %}
+{% if sls.split('.')[0] in allowed_states %}
+{%   from 'vars/globals.map.jinja' import GLOBALS %}
+{%   set kafka_cluster_id = salt['pillar.get']('kafka:cluster_id') %}
+
+{# Initialize kafka storage if it doesn't already exist. Just looking for meta.properties in /nsm/kafka/data #}
+{%   if not salt['file.file_exists']('/nsm/kafka/data/meta.properties') %}
+kafka_storage_init:
+  cmd.run:
+    - name: |
+        docker run -v /nsm/kafka/data:/nsm/kafka/data -v /opt/so/conf/kafka/server.properties:/opt/kafka/config/kraft/newserver.properties --name so-kafkainit --user root --entrypoint /opt/kafka/bin/kafka-storage.sh {{ GLOBALS.registry_host }}:5000/{{ GLOBALS.image_repo }}/so-kafka:{{ GLOBALS.so_version }} format -t {{ kafka_cluster_id }} -c /opt/kafka/config/kraft/newserver.properties
+kafka_rm_kafkainit:
+  cmd.run:
+    - name: |
+        docker rm so-kafkainit
+{%   endif %}
+
+
+{% else %}
+
+{{sls}}_state_not_allowed:
+  test.fail_without_changes:
+    - name: {{sls}}_state_not_allowed
+
+{% endif %}

salt/kafka/tools/sbin/so-kafka-cli

@@ -0,0 +1,47 @@
+#! /bin/bash
+# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at 
+# https://securityonion.net/license; you may not use this file except in compliance with the
+# Elastic License 2.0.
+
+if [ -z "$NOROOT" ]; then
+	# Check for prerequisites
+	if [ "$(id -u)" -ne 0 ]; then
+		echo "This script must be run using sudo!"
+		exit 1
+	fi
+fi
+
+function usage() {
+  echo -e "\nUsage: $0 <script> [options]"
+  echo ""
+  echo "Available scripts:"
+  show_available_kafka_cli_tools
+}
+
+function show_available_kafka_cli_tools(){
+  docker exec so-kafka ls /opt/kafka/bin | grep kafka
+}
+
+if [ -z $1 ]; then
+  usage
+  exit 1
+fi
+
+available_tools=$(show_available_kafka_cli_tools)
+script_exists=false
+
+for script in $available_tools; do
+  if [ "$script" == "$1" ]; then
+    script_exists=true
+    break
+  fi
+done
+
+if [ "$script_exists" == true ]; then
+  docker exec so-kafka /opt/kafka/bin/$1 "${@:2}"
+else
+  echo -e  "\nInvalid script: $1"
+  usage
+  exit 1
+fi

salt/kafka/tools/sbin/so-kafka-config-update

@@ -0,0 +1,87 @@
+#! /bin/bash
+# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at 
+# https://securityonion.net/license; you may not use this file except in compliance with the
+# Elastic License 2.0.
+
+if [ -z "$NOROOT" ]; then
+	# Check for prerequisites
+	if [ "$(id -u)" -ne 0 ]; then
+		echo "This script must be run using sudo!"
+		exit 1
+	fi
+fi
+
+usage() {
+  cat <<USAGE_EOF
+
+  Usage: $0 <operation> [parameters]
+
+  Where <operation> is one of the following:
+
+    topic-partitions: Increase the number of partitions for a Kafka topic
+      Required arguments: topic-partitions <topic name> <# partitions>
+      Example: $0 topic-partitions suricata-topic 6
+
+	list-topics: List of Kafka topics
+	  Example: $0 list-topics
+
+USAGE_EOF
+  exit 1
+}
+
+if [[ $# -lt 1 || $1 == --help || $1 == -h ]]; then
+  usage
+fi
+
+kafka_client_config="/opt/kafka/config/kraft/client.properties"
+
+too_few_arguments() {
+  echo -e "\nMissing one or more required arguments!\n"
+  usage
+}
+
+get_kafka_brokers() {
+	brokers_cache="/opt/so/state/kafka_brokers"
+  broker_port="9092"
+  if [[ ! -f "$brokers_cache" ]] || [[ $(find "/$brokers_cache" -mmin +120) ]]; then
+    echo "Refreshing Kafka brokers list"
+	  salt-call pillar.get kafka:nodes --out=json | jq -r --arg broker_port "$broker_port" '.local | to_entries[] | select(.value.role | contains("broker")) | "\(.value.ip):\($broker_port)"' | paste -sd "," - > "$brokers_cache"
+  else
+    echo "Using cached Kafka brokers list"
+  fi
+  brokers=$(cat "$brokers_cache")
+}
+
+increase_topic_partitions() {
+  get_kafka_brokers
+	command=$(so-kafka-cli kafka-topics.sh --bootstrap-server $brokers --command-config $kafka_client_config --alter --topic $topic --partitions $partition_count)
+  if $command; then
+    echo -e "Successfully increased the number of partitions for topic $topic to $partition_count\n"
+    so-kafka-cli kafka-topics.sh --bootstrap-server $brokers --command-config $kafka_client_config --describe --topic $topic
+  fi
+}
+
+get_kafka_topics_list() {
+	get_kafka_brokers
+	so-kafka-cli kafka-topics.sh --bootstrap-server $brokers --command-config $kafka_client_config --exclude-internal --list | sort
+}
+
+operation=$1
+case "${operation}" in
+  "topic-partitions")
+	if [[ $# -lt 3 ]]; then
+	  too_few_arguments
+	fi
+	topic=$2
+	partition_count=$3
+	increase_topic_partitions
+	;;
+  "list-topics")
+	get_kafka_topics_list
+	;;
+  *)
+	usage
+	;;
+esac
+

salt/kafka/tools/sbin_jinja/so-kafka-fleet-output-policy

@@ -0,0 +1,23 @@
+#!/bin/bash
+# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at
+# https://securityonion.net/license; you may not use this file except in compliance with the
+# Elastic License 2.0.
+
+output=$(curl -K /opt/so/conf/elasticsearch/curl.config -L "http://localhost:5601/api/fleet/outputs" | jq -r .items[].id)
+
+if ! echo "$output" | grep -q "so-manager_kafka"; then
+  KAFKACRT=$(openssl x509 -in /etc/pki/elasticfleet-kafka.crt)
+  KAFKAKEY=$(openssl rsa -in  /etc/pki/elasticfleet-kafka.key)
+  KAFKACA=$(openssl x509 -in  /etc/pki/tls/certs/intca.crt)
+  KAFKA_OUTPUT_VERSION="2.6.0"
+  JSON_STRING=$( jq -n \
+                --arg KAFKACRT "$KAFKACRT" \
+                --arg KAFKAKEY "$KAFKAKEY" \
+                --arg KAFKACA "$KAFKACA" \
+                --arg MANAGER_IP "{{ GLOBALS.manager_ip }}:9092" \
+                --arg KAFKA_OUTPUT_VERSION "$KAFKA_OUTPUT_VERSION" \
+                    '{ "name": "grid-kafka", "id": "so-manager_kafka", "type": "kafka", "hosts": [ $MANAGER_IP ], "is_default": false, "is_default_monitoring": false, "config_yaml": "", "ssl": { "certificate_authorities": [ $KAFKACA ], "certificate": $KAFKACRT, "key": $KAFKAKEY, "verification_mode": "full" }, "proxy_id": null, "client_id": "Elastic", "version": $KAFKA_OUTPUT_VERSION, "compression": "none", "auth_type": "ssl", "partition": "round_robin", "round_robin": { "group_events": 1 }, "topics":[{"topic":"%{[event.module]}-securityonion","when":{"type":"regexp","condition":"event.module:.+"}},{"topic":"default-securityonion"}], "headers": [ { "key": "", "value": "" } ], "timeout": 30, "broker_timeout": 30, "required_acks": 1 }'
+                )
+  curl -K /opt/so/conf/elasticsearch/curl.config -L -X POST "localhost:5601/api/fleet/outputs" -H 'kbn-xsrf: true' -H 'Content-Type: application/json' -d "$JSON_STRING" 2&1> /dev/null
+fi

salt/logstash/defaults.yaml

@@ -37,6 +37,7 @@ logstash:
       - so/0900_input_redis.conf.jinja
       - so/9805_output_elastic_agent.conf.jinja
       - so/9900_output_endgame.conf.jinja
+      - so/0800_input_kafka.conf.jinja
     custom0: []
     custom1: []
     custom2: []

salt/logstash/enabled.sls

@@ -75,9 +75,10 @@ so-logstash:
       {% else %}
       - /etc/pki/tls/certs/intca.crt:/usr/share/filebeat/ca.crt:ro
       {% endif %}
-      {% if GLOBALS.role in ['so-manager', 'so-managersearch', 'so-standalone', 'so-import', 'so-heavynode', 'so-searchnode'] %}
+      {% if GLOBALS.role in ['so-manager', 'so-managersearch', 'so-standalone', 'so-import', 'so-heavynode', 'so-searchnode' ] %}
       - /opt/so/conf/ca/cacerts:/etc/pki/ca-trust/extracted/java/cacerts:ro
       - /opt/so/conf/ca/tls-ca-bundle.pem:/etc/pki/ca-trust/extracted/pem/tls-ca-bundle.pem:ro
+      - /etc/pki/kafka-logstash.p12:/usr/share/logstash/kafka-logstash.p12:ro
       {% endif %}
       {% if GLOBALS.role == 'so-eval' %}
       - /nsm/zeek:/nsm/zeek:ro

salt/logstash/init.sls

@@ -4,9 +4,13 @@
 # Elastic License 2.0.
 
 {% from 'logstash/map.jinja' import LOGSTASH_MERGED %}
+{% from 'kafka/map.jinja' import KAFKAMERGED %}
 
 include:
-{% if LOGSTASH_MERGED.enabled %}
+{# Disable logstash when Kafka is enabled except when the role is standalone #}
+{% if LOGSTASH_MERGED.enabled and grains.role == 'so-standalone' %}
+  - logstash.enabled
+{% elif LOGSTASH_MERGED.enabled and not KAFKAMERGED.enabled %}
   - logstash.enabled
 {% else %}
   - logstash.disabled

salt/logstash/pipelines/config/so/0800_input_kafka.conf.jinja

@@ -0,0 +1,36 @@
+{%- set kafka_password = salt['pillar.get']('kafka:password') %}
+{%- set kafka_brokers = salt['pillar.get']('kafka:nodes', {}) %}
+{%- set brokers = [] %}
+
+{%- for key, values in kafka_brokers.items() %}
+{%-   if 'broker' in values['role'] %}
+{%-     do brokers.append(key ~ ':9092') %}
+{%-   endif %}
+{%- endfor %}
+{%- set bootstrap_servers = ','.join(brokers) %}
+
+input {
+    kafka {
+        codec => json
+        topics_pattern => '.*-securityonion$'
+        group_id => 'searchnodes'
+        consumer_threads => 3
+        client_id => '{{ GLOBALS.hostname }}'
+        security_protocol => 'SSL'
+        bootstrap_servers => '{{ bootstrap_servers }}'
+        ssl_keystore_location => '/usr/share/logstash/kafka-logstash.p12'
+        ssl_keystore_password => '{{ kafka_password }}'
+        ssl_keystore_type => 'PKCS12'
+        ssl_truststore_location => '/etc/pki/ca-trust/extracted/java/cacerts'
+        ssl_truststore_password => 'changeit'
+        decorate_events => true
+        tags => [ "elastic-agent", "input-{{ GLOBALS.hostname}}", "kafka" ]
+    }
+}
+filter {
+  if ![metadata] {
+    mutate {
+      rename => { "@metadata" => "metadata" }
+    }
+  }
+}

salt/manager/tools/sbin/so-minion

@@ -604,6 +604,10 @@ function updateMineAndApplyStates() {
 	if [[ "$NODETYPE" == "SEARCHNODE" || "$NODETYPE" == "HEAVYNODE" ]]; then
 		salt-run state.orch orch.container_download pillar="{'setup': {'newnode': $MINION_ID }}" > /dev/null 2>&1 &
 	fi
+	if [[ "$NODETYPE" == "RECEIVER" ]]; then
+		# Setup nodeid for Kafka
+		salt-call state.apply kafka.nodes queue=True
+	fi
 	# $MINIONID is the minion id of the manager and $MINION_ID is the target node or the node being configured
 	salt-run state.orch orch.deploy_newnode pillar="{'setup': {'manager': $MINIONID, 'newnode': $MINION_ID }}" > /dev/null 2>&1 &
 }

salt/manager/tools/sbin/soup

@@ -625,11 +625,27 @@ up_to_2.4.70() {
   suricata_idstools_migration
   toggle_telemetry
   add_detection_test_pillars
+
   INSTALLEDVERSION=2.4.70
 }
 
 up_to_2.4.80() {
   phases_pillar_2_4_80
+  # Kafka configuration changes
+
+  # Global pipeline changes to REDIS or KAFKA
+  echo "Removing global.pipeline pillar configuration"
+  sed -i '/pipeline:/d' /opt/so/saltstack/local/pillar/global/soc_global.sls
+    # Kafka pillars
+  mkdir -p /opt/so/saltstack/local/pillar/kafka
+  touch /opt/so/saltstack/local/pillar/kafka/soc_kafka.sls
+  touch /opt/so/saltstack/local/pillar/kafka/adv_kafka.sls
+  echo 'kafka: ' > /opt/so/saltstack/local/pillar/kafka/soc_kafka.sls
+  kafka_cluster_id=$(get_random_value 22)
+  echo '    cluster_id: '$kafka_cluster_id >> /opt/so/saltstack/local/pillar/kafka/soc_kafka.sls
+  kafkapass=$(get_random_value)
+  echo '    password: '$kafkapass >> /opt/so/saltstack/local/pillar/kafka/soc_kafka.sls
+
   INSTALLEDVERSION=2.4.80
 }
 

salt/redis/init.sls

@@ -4,9 +4,10 @@
 # Elastic License 2.0.
 
 {% from 'redis/map.jinja' import REDISMERGED %}
+{% from 'vars/globals.map.jinja' import GLOBALS %}
 
 include:
-{% if REDISMERGED.enabled %}
+{% if GLOBALS.pipeline == "REDIS" and REDISMERGED.enabled %}
   - redis.enabled
 {% else %}
   - redis.disabled

salt/salt/files/engines.conf

@@ -30,4 +30,65 @@ engines:
                   '*':
                   - cmd.run:
                       cmd: /usr/sbin/so-rule-update
+        - files:
+            - /opt/so/saltstack/local/pillar/global/soc_global.sls
+            - /opt/so/saltstack/local/pillar/global/adv_global.sls
+          pillar: global.pipeline
+          default: REDIS
+          actions:
+            from:
+              '*':
+                to:
+                  'KAFKA':
+                  - cmd.run:
+                      cmd: /usr/sbin/so-yaml.py replace /opt/so/saltstack/local/pillar/kafka/soc_kafka.sls kafka.enabled True
+                  - cmd.run:
+                      cmd: salt -C 'G@role:so-standalone or G@role:so-manager or G@role:so-managersearch or G@role:so-receiver' saltutil.kill_all_jobs
+                  - cmd.run:
+                      cmd: salt-call state.apply kafka.nodes
+                  - cmd.run:
+                      cmd: salt -C 'G@role:so-standalone or G@role:so-manager or G@role:so-managersearch or G@role:so-receiver' state.highstate
+              'KAFKA':
+                to:
+                  'REDIS':
+                  - cmd.run:
+                      cmd: /usr/sbin/so-yaml.py replace /opt/so/saltstack/local/pillar/kafka/soc_kafka.sls kafka.enabled False
+                  - cmd.run:
+                      cmd: salt -C 'G@role:so-standalone or G@role:so-manager or G@role:so-managersearch or G@role:so-receiver' saltutil.kill_all_jobs
+                  - cmd.run:
+                      cmd: salt -C 'G@role:so-standalone or G@role:so-manager or G@role:so-managersearch or G@role:so-receiver' state.highstate
+        - files:
+            - /opt/so/saltstack/local/pillar/kafka/soc_kafka.sls
+            - /opt/so/saltstack/local/pillar/kafka/adv_kafka.sls
+          pillar: kafka.controllers
+          default: ''
+          actions:
+            from:
+              '*':
+                to:
+                  '*':
+                  - cmd.run:
+                      cmd: salt -C 'G@role:so-standalone or G@role:so-manager or G@role:so-managersearch or G@role:so-receiver' saltutil.kill_all_jobs
+                  - cmd.run:
+                      cmd: salt-call state.apply kafka.nodes
+                  - cmd.run:
+                      cmd: salt -C 'G@role:so-standalone or G@role:so-manager or G@role:so-managersearch or G@role:so-receiver' state.apply kafka
+                  - cmd.run:
+                      cmd: salt-call state.apply elasticfleet
+        - files:
+            - /opt/so/saltstack/local/pillar/kafka/soc_kafka.sls
+            - /opt/so/saltstack/local/pillar/kafka/adv_kafka.sls
+          pillar: kafka.reset
+          default: ''
+          actions:
+            from:
+              '*':
+                to:
+                  'YES_RESET_KAFKA':
+                  - cmd.run:
+                      cmd: salt -C 'G@role:so-standalone or G@role:so-manager or G@role:so-managersearch or G@role:so-receiver' saltutil.kill_all_jobs
+                  - cmd.run:
+                      cmd: salt -C 'G@role:so-standalone or G@role:so-manager or G@role:so-managersearch or G@role:so-receiver' state.apply kafka.disabled,kafka.reset
+                  - cmd.run:
+                      cmd: /usr/sbin/so-yaml.py remove /opt/so/saltstack/local/pillar/kafka/soc_kafka.sls kafka.reset
       interval: 10

salt/salt/minion.sls

@@ -13,6 +13,9 @@ include:
   - systemd.reload
   - repo.client
   - salt.mine_functions
+{% if GLOBALS.role in GLOBALS.manager_roles %}
+  - ca
+{% endif %}
 
 {% if INSTALLEDSALTVERSION|string != SALTVERSION|string %}
 
@@ -98,5 +101,8 @@ salt_minion_service:
       - file: mine_functions
 {% if INSTALLEDSALTVERSION|string == SALTVERSION|string %}
       - file: set_log_levels
+{% endif %}
+{% if GLOBALS.role in GLOBALS.manager_roles %}
+      - file: /etc/salt/minion.d/signing_policies.conf
 {% endif %}
     - order: last

salt/ssl/init.sls

@@ -17,6 +17,8 @@
   {% set COMMONNAME = GLOBALS.manager %}
 {% endif %}
 
+{% set kafka_password = salt['pillar.get']('kafka:password') %}
+
 {% if grains.id.split('_')|last in ['manager', 'managersearch', 'eval', 'standalone', 'import'] %}
 include:
   - ca
@@ -662,8 +664,295 @@ elastickeyperms:
     - mode: 640
     - group: 930
 
+kafka_logstash_key:
+  x509.private_key_managed:
+    - name: /etc/pki/kafka-logstash.key
+    - keysize: 4096
+    - backup: True
+    - new: True
+    {% if salt['file.file_exists']('/etc/pki/kafka-logstash.key') -%}
+    - prereq:
+      - x509: /etc/pki/kafka-logstash.crt
+    {%- endif %}
+    - retry:
+        attempts: 5
+        interval: 30
+
+kafka_logstash_crt:
+  x509.certificate_managed:
+    - name: /etc/pki/kafka-logstash.crt
+    - ca_server: {{ ca_server }}
+    - subjectAltName: DNS:{{ GLOBALS.hostname }}, IP:{{ GLOBALS.node_ip }}
+    - signing_policy: kafka
+    - private_key: /etc/pki/kafka-logstash.key
+    - CN: {{ GLOBALS.hostname }}
+    - days_remaining: 0
+    - days_valid: 820
+    - backup: True
+    - timeout: 30
+    - retry:
+        attempts: 5
+        interval: 30
+  cmd.run:
+    - name: "/usr/bin/openssl pkcs12 -inkey /etc/pki/kafka-logstash.key -in /etc/pki/kafka-logstash.crt -export -out /etc/pki/kafka-logstash.p12 -nodes -passout pass:{{ kafka_password }}"
+    - onchanges:
+      - x509: /etc/pki/kafka-logstash.key
+
+kafka_logstash_key_perms:
+  file.managed:
+    - replace: False
+    - name: /etc/pki/kafka-logstash.key
+    - mode: 640
+    - user: 960
+    - group: 939
+
+kafka_logstash_crt_perms:
+  file.managed:
+    - replace: False
+    - name: /etc/pki/kafka-logstash.crt
+    - mode: 640
+    - user: 960
+    - group: 939
+
+kafka_logstash_pkcs12_perms:
+  file.managed:
+    - replace: False
+    - name: /etc/pki/kafka-logstash.p12
+    - mode: 640
+    - user: 960
+    - group: 931
+
 {%- endif %}
 
+{% if grains['role'] in ['so-manager', 'so-managersearch', 'so-standalone'] %}
+
+elasticfleet_kafka_key:
+  x509.private_key_managed:
+    - name: /etc/pki/elasticfleet-kafka.key
+    - keysize: 4096
+    - backup: True
+    - new: True
+    {% if salt['file.file_exists']('/etc/pki/elasticfleet-kafka.key') -%}
+    - prereq:
+      - x509: elasticfleet_kafka_crt
+    {%- endif %}
+    - retry:
+        attempts: 5
+        interval: 30
+
+elasticfleet_kafka_crt:
+  x509.certificate_managed:
+    - name: /etc/pki/elasticfleet-kafka.crt
+    - ca_server: {{ ca_server }}
+    - signing_policy: kafka
+    - private_key: /etc/pki/elasticfleet-kafka.key
+    - CN: {{ GLOBALS.hostname }}
+    - subjectAltName: DNS:{{ GLOBALS.hostname }}, IP:{{ GLOBALS.node_ip }}
+    - days_remaining: 0
+    - days_valid: 820
+    - backup: True
+    - timeout: 30
+    - retry:
+        attempts: 5
+        interval: 30
+  cmd.run:
+  - name: "/usr/bin/openssl pkcs8 -in /etc/pki/elasticfleet-kafka.key -topk8 -out /etc/pki/elasticfleet-kafka.p8 -nocrypt"
+  - onchanges:
+    - x509: elasticfleet_kafka_key
+
+elasticfleet_kafka_cert_perms:
+  file.managed:
+    - replace: False
+    - name: /etc/pki/elasticfleet-kafka.crt
+    - mode: 640
+    - user: 960
+    - group: 939
+
+elasticfleet_kafka_key_perms:
+  file.managed:
+    - replace: False
+    - name: /etc/pki/elasticfleet-kafka.key
+    - mode: 640
+    - user: 960
+    - group: 939
+
+elasticfleet_kafka_pkcs8_perms:
+  file.managed:
+    - replace: False
+    - name: /etc/pki/elasticfleet-kafka.p8
+    - mode: 640
+    - user: 960
+    - group: 939
+
+kafka_client_key:
+  x509.private_key_managed:
+    - name: /etc/pki/kafka-client.key
+    - keysize: 4096
+    - backup: True
+    - new: True
+    {% if salt['file.file_exists']('/etc/pki/kafka-client.key') -%}
+    - prereq:
+      - x509: /etc/pki/kafka-client.crt
+    {%- endif %}
+    - retry:
+        attempts: 5
+        interval: 30
+
+kafka_client_crt:
+  x509.certificate_managed:
+    - name: /etc/pki/kafka-client.crt
+    - ca_server: {{ ca_server }}
+    - subjectAltName: DNS:{{ GLOBALS.hostname }}, IP:{{ GLOBALS.node_ip }}
+    - signing_policy: kafka
+    - private_key: /etc/pki/kafka-client.key
+    - CN: {{ GLOBALS.hostname }}
+    - days_remaining: 0
+    - days_valid: 820
+    - backup: True
+    - timeout: 30
+    - retry:
+        attempts: 5
+        interval: 30
+
+kafka_client_key_perms:
+  file.managed:
+    - replace: False
+    - name: /etc/pki/kafka-client.key
+    - mode: 640
+    - user: 960
+    - group: 939
+
+kafka_client_crt_perms:
+  file.managed:
+    - replace: False
+    - name: /etc/pki/kafka-client.crt
+    - mode: 640
+    - user: 960
+    - group: 939
+
+{% endif %}
+
+{% if grains['role'] in ['so-manager', 'so-managersearch','so-receiver', 'so-standalone'] %}
+
+kafka_key:
+  x509.private_key_managed:
+    - name: /etc/pki/kafka.key
+    - keysize: 4096
+    - backup: True
+    - new: True
+    {% if salt['file.file_exists']('/etc/pki/kafka.key') -%}
+    - prereq:
+      - x509: /etc/pki/kafka.crt
+    {%- endif %}
+    - retry:
+        attempts: 5
+        interval: 30
+
+kafka_crt:
+  x509.certificate_managed:
+    - name: /etc/pki/kafka.crt
+    - ca_server: {{ ca_server }}
+    - subjectAltName: DNS:{{ GLOBALS.hostname }}, IP:{{ GLOBALS.node_ip }}
+    - signing_policy: kafka
+    - private_key: /etc/pki/kafka.key
+    - CN: {{ GLOBALS.hostname }}
+    - days_remaining: 0
+    - days_valid: 820
+    - backup: True
+    - timeout: 30
+    - retry:
+        attempts: 5
+        interval: 30
+  cmd.run:
+    - name: "/usr/bin/openssl pkcs12 -inkey /etc/pki/kafka.key -in /etc/pki/kafka.crt -export -out /etc/pki/kafka.p12 -nodes -passout pass:{{ kafka_password }}"
+    - onchanges:
+      - x509: /etc/pki/kafka.key
+kafka_key_perms:
+  file.managed:
+    - replace: False
+    - name: /etc/pki/kafka.key
+    - mode: 640
+    - user: 960
+    - group: 939
+
+kafka_crt_perms:
+  file.managed:
+    - replace: False
+    - name: /etc/pki/kafka.crt
+    - mode: 640
+    - user: 960
+    - group: 939
+
+kafka_pkcs12_perms:
+  file.managed:
+    - replace: False
+    - name: /etc/pki/kafka.p12
+    - mode: 640
+    - user: 960
+    - group: 939
+
+{% endif %}
+{# For automated testing standalone will need kafka-logstash key to pull logs from Kafka #}
+{% if grains['role'] == 'so-standalone' %}
+kafka_logstash_key:
+  x509.private_key_managed:
+    - name: /etc/pki/kafka-logstash.key
+    - keysize: 4096
+    - backup: True
+    - new: True
+    {% if salt['file.file_exists']('/etc/pki/kafka-logstash.key') -%}
+    - prereq:
+      - x509: /etc/pki/kafka-logstash.crt
+    {%- endif %}
+    - retry:
+        attempts: 5
+        interval: 30
+
+kafka_logstash_crt:
+  x509.certificate_managed:
+    - name: /etc/pki/kafka-logstash.crt
+    - ca_server: {{ ca_server }}
+    - subjectAltName: DNS:{{ GLOBALS.hostname }}, IP:{{ GLOBALS.node_ip }}
+    - signing_policy: kafka
+    - private_key: /etc/pki/kafka-logstash.key
+    - CN: {{ GLOBALS.hostname }}
+    - days_remaining: 0
+    - days_valid: 820
+    - backup: True
+    - timeout: 30
+    - retry:
+        attempts: 5
+        interval: 30
+  cmd.run:
+    - name: "/usr/bin/openssl pkcs12 -inkey /etc/pki/kafka-logstash.key -in /etc/pki/kafka-logstash.crt -export -out /etc/pki/kafka-logstash.p12 -nodes -passout pass:{{ kafka_password }}"
+    - onchanges:
+      - x509: /etc/pki/kafka-logstash.key
+
+kafka_logstash_key_perms:
+  file.managed:
+    - replace: False
+    - name: /etc/pki/kafka-logstash.key
+    - mode: 640
+    - user: 960
+    - group: 939
+
+kafka_logstash_crt_perms:
+  file.managed:
+    - replace: False
+    - name: /etc/pki/kafka-logstash.crt
+    - mode: 640
+    - user: 960
+    - group: 939
+
+kafka_logstash_pkcs12_perms:
+  file.managed:
+    - replace: False
+    - name: /etc/pki/kafka-logstash.p12
+    - mode: 640
+    - user: 960
+    - group: 931
+{% endif %}
+
 {% else %}
 
 {{sls}}_state_not_allowed:

salt/ssl/remove.sls

@@ -71,3 +71,20 @@ fleet_crt:
 fbcertdir:
   file.absent:
     - name: /opt/so/conf/filebeat/etc/pki
+
+kafka_crt:
+  file.absent:
+    - name: /etc/pki/kafka.crt
+kafka_key:
+  file.absent:
+    - name: /etc/pki/kafka.key
+
+kafka_logstash_crt:
+  file.absent:
+    - name: /etc/pki/kafka-logstash.crt
+kafka_logstash_key:
+  file.absent:
+    - name: /etc/pki/kafka-logstash.key
+kafka_logstash_keystore:
+  file.absent:
+    - name: /etc/pki/kafka-logstash.p12

salt/telegraf/etc/telegraf.conf

@@ -243,6 +243,40 @@
   password = "{{ salt['pillar.get']('elasticsearch:auth:users:so_logstash_user:pass') }}"
 {%- endif %}
 
+{% if grains.role in ['so-manager','so-managersearch','so-standalone','so-receiver'] and GLOBALS.pipeline == "KAFKA" -%}
+[[inputs.jolokia2_agent]]
+  name_prefix= "kafka_"
+  urls = ["http://localhost:8778/jolokia"]
+
+[[inputs.jolokia2_agent.metric]]
+  name         = "topics"
+  mbean        = "kafka.server:name=*,type=BrokerTopicMetrics"
+  field_prefix = "$1."
+
+[[inputs.jolokia2_agent.metric]]
+  name         = "topic"
+  mbean        = "kafka.server:name=*,topic=*,type=BrokerTopicMetrics"
+  field_prefix = "$1."
+  tag_keys     = ["topic"]
+
+[[inputs.jolokia2_agent.metric]]
+  name         = "controller"
+  mbean        = "kafka.controller:name=*,type=*"
+  field_prefix = "$1."
+
+[[inputs.jolokia2_agent.metric]]
+  name       = "partition"
+  mbean      = "kafka.log:name=*,partition=*,topic=*,type=Log"
+  field_name = "$1"
+  tag_keys   = ["topic", "partition"]
+
+[[inputs.jolokia2_agent.metric]]
+  name       = "partition"
+  mbean      = "kafka.cluster:name=UnderReplicated,partition=*,topic=*,type=Partition"
+  field_name = "UnderReplicatedPartitions"
+  tag_keys   = ["topic", "partition"]
+
+{%- endif %}
 # # Read metrics from one or more commands that can output to stdout
 {%- if 'sostatus.sh' in TELEGRAFMERGED.scripts[GLOBALS.role.split('-')[1]] %}
 {%-   do TELEGRAFMERGED.scripts[GLOBALS.role.split('-')[1]].remove('sostatus.sh') %}

salt/telegraf/map.jinja

@@ -22,3 +22,10 @@
 {%   endif %}
 
 {% endif %}
+
+{% if GLOBALS.pipeline != 'REDIS' %}
+{# When global pipeline is not REDIS remove redis.sh script. KAFKA metrics are collected via jolokia agent. Config in telegraf.conf #}
+{%   if GLOBALS.role in ['so-standalone', 'so-manager', 'so-managersearch', 'so-receiver', 'so-heavynode'] %}
+{%     do TELEGRAFMERGED.scripts[GLOBALS.role.split('-')[1]].remove('redis.sh') %}
+{%   endif %}
+{% endif %}

salt/top.sls

@@ -107,6 +107,7 @@ base:
     - utility
     - elasticfleet
     - stig
+    - kafka
 
   '*_standalone and G@saltversion:{{saltversion}}':
     - match: compound
@@ -141,6 +142,7 @@ base:
     - utility
     - elasticfleet
     - stig
+    - kafka
 
   '*_searchnode and G@saltversion:{{saltversion}}':
     - match: compound
@@ -238,6 +240,7 @@ base:
     - logstash
     - redis
     - elasticfleet.install_agent_grid
+    - kafka
 
   '*_idh and G@saltversion:{{saltversion}}':
     - match: compound

salt/vars/globals.map.jinja

@@ -23,7 +23,7 @@
     'manager_ip': INIT.PILLAR.global.managerip,
     'md_engine': INIT.PILLAR.global.mdengine,
     'pcap_engine': GLOBALMERGED.pcapengine,
-    'pipeline': INIT.PILLAR.global.pipeline,
+    'pipeline': GLOBALMERGED.pipeline,
     'so_version': INIT.PILLAR.global.soversion,
     'so_docker_gateway': DOCKER.gateway,
     'so_docker_range': DOCKER.range,

setup/so-functions

@@ -788,6 +788,7 @@ create_manager_pillars() {
 	patch_pillar
 	nginx_pillar
 	kibana_pillar
+	kafka_pillar
 }
 
 create_repo() {
@@ -1176,6 +1177,18 @@ kibana_pillar() {
 	logCmd "touch $kibana_pillar_file"
 }
 
+kafka_pillar() {
+	KAFKACLUSTERID=$(get_random_value 22)
+	KAFKAPASS=$(get_random_value)
+	logCmd "mkdir -p $local_salt_dir/pillar/kakfa"
+	logCmd "touch $adv_kafka_pillar_file"
+	logCmd "touch $kafka_pillar_file"
+	printf '%s\n'\
+		"kafka:"\
+		"  cluster_id: $KAFKACLUSTERID"\
+		"  password: $KAFKAPASS" > $kafka_pillar_file
+}
+
 logrotate_pillar() {
 	logCmd "mkdir -p $local_salt_dir/pillar/logrotate"
 	logCmd "touch $adv_logrotate_pillar_file"
@@ -1332,7 +1345,6 @@ create_global() {
 
 	# Continue adding other details	
 	echo "  imagerepo: '$IMAGEREPO'" >> $global_pillar_file
-	echo "  pipeline: 'redis'" >> $global_pillar_file
 	echo "  repo_host: '$HOSTNAME'" >> $global_pillar_file
 	echo "  influxdb_host: '$HOSTNAME'" >> $global_pillar_file
 	echo "  registry_host: '$HOSTNAME'" >> $global_pillar_file
@@ -1402,7 +1414,7 @@ make_some_dirs() {
 	mkdir -p $local_salt_dir/salt/firewall/portgroups
 	mkdir -p $local_salt_dir/salt/firewall/ports
 
-	for THEDIR in bpf pcap elasticsearch ntp firewall redis backup influxdb strelka sensoroni soc docker zeek suricata nginx telegraf logstash soc manager kratos idstools idh elastalert stig global;do
+	for THEDIR in bpf pcap elasticsearch ntp firewall redis backup influxdb strelka sensoroni soc docker zeek suricata nginx telegraf logstash soc manager kratos idstools idh elastalert stig global kafka;do
 	mkdir -p $local_salt_dir/pillar/$THEDIR
 	touch $local_salt_dir/pillar/$THEDIR/adv_$THEDIR.sls
 	touch $local_salt_dir/pillar/$THEDIR/soc_$THEDIR.sls

setup/so-variables

@@ -178,6 +178,12 @@ export redis_pillar_file
 adv_redis_pillar_file="$local_salt_dir/pillar/redis/adv_redis.sls"
 export adv_redis_pillar_file
 
+kafka_pillar_file="local_salt_dir/pillar/kafka/soc_kafka.sls"
+export kafka_pillar_file
+
+adv_kafka_pillar_file="$local_salt_dir/pillar/kafka/adv_kafka.sls"
+export kafka_pillar_file
+
 idh_pillar_file="$local_salt_dir/pillar/idh/soc_idh.sls"
 export idh_pillar_file