diff --git a/salt/kafka/disabled.sls b/salt/kafka/disabled.sls
index 707e953a4..79fd0c261 100644
--- a/salt/kafka/disabled.sls
+++ b/salt/kafka/disabled.sls
@@ -22,4 +22,13 @@ ensure_default_pipeline:
     - 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 %}
\ No newline at end of file
+{% endif %}
+
+{# If Kafka has never been manually enabled, the 'Kafka' user does not exist. In this case certs for Kafka should not exist since they'll be owned by uid 960 #}
+{% for cert in ['kafka-client.crt','kafka-client.key','kafka.crt','kafka.key','kafka-logstash.crt','kafka-logstash.key','kafka-logstash.p12','kafka.p12','elasticfleet-kafka.p8'] %}
+check_kafka_cert_{{cert}}:
+  file.absent:
+    - name: /etc/pki/{{cert}}
+    - onlyif: stat -c %U /etc/pki/{{cert}} | grep -q UNKNOWN
+    - show_changes: False
+{% endfor %}
\ No newline at end of file
diff --git a/salt/kafka/ssl.sls b/salt/kafka/ssl.sls
new file mode 100644
index 000000000..c4e46ac8a
--- /dev/null
+++ b/salt/kafka/ssl.sls
@@ -0,0 +1,196 @@
+# 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 in allowed_states %}
+{% from 'vars/globals.map.jinja' import GLOBALS %}
+
+{% set kafka_password = salt['pillar.get']('kafka:password') %}
+
+include:
+  - ca.dirs
+    {% set global_ca_server = [] %}
+    {% set x509dict = salt['mine.get'](GLOBALS.manager | lower~'*', 'x509.get_pem_entries') %}
+    {% for host in x509dict %}
+      {% if 'manager' in host.split('_')|last or host.split('_')|last == 'standalone' %}
+        {% do global_ca_server.append(host) %}
+      {% endif %}
+    {% endfor %}
+    {% set ca_server = global_ca_server[0] %}
+
+
+{% if GLOBALS.pipeline == "KAFKA" %}
+
+{%   if grains['role'] in ['so-manager', 'so-managersearch', 'so-standalone'] %}
+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 %}
+
+# Standalone needs kafka-logstash for automated testing. Searchnode/manager search need it for logstash to consume from Kafka.
+# Manager will have cert, but be unused until a pipeline is created and logstash enabled.
+{%   if grains['role'] in ['so-standalone', 'so-managersearch', 'so-searchnode', 'so-manager'] %}
+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: 931
+    - group: 939
+
+kafka_logstash_crt_perms:
+  file.managed:
+    - replace: False
+    - name: /etc/pki/kafka-logstash.crt
+    - mode: 640
+    - user: 931
+    - group: 939
+
+kafka_logstash_pkcs12_perms:
+  file.managed:
+    - replace: False
+    - name: /etc/pki/kafka-logstash.p12
+    - mode: 640
+    - user: 931
+    - group: 939
+
+{% endif %}
+
+{% endif %}
\ No newline at end of file
diff --git a/salt/logstash/enabled.sls b/salt/logstash/enabled.sls
index f95a76f13..0f44a3767 100644
--- a/salt/logstash/enabled.sls
+++ b/salt/logstash/enabled.sls
@@ -14,6 +14,11 @@
 include:
 {%   if GLOBALS.role not in ['so-receiver','so-fleet'] %}
   - elasticsearch.ca
+{%   endif %}
+{# Kafka ca runs on nodes that can run logstash for Kafka input / output. Only when Kafka is global pipeline #}
+{%   if GLOBALS.role in ['so-searchnode', 'so-manager', 'so-managersearch', 'so-receiver', 'so-standalone'] and GLOBALS.pipeline == 'KAFKA' %}
+  - kafka.ca
+  - kafka.ssl
 {%   endif %}
   - logstash.config
   - logstash.sostatus
@@ -79,8 +84,9 @@ so-logstash:
       - /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
       {% endif %}
-      {% if GLOBALS.role in ['so-manager', 'so-managersearch', 'so-standalone', 'so-searchnode'] %}
+      {% if GLOBALS.pipeline == "KAFKA" and GLOBALS.role in ['so-manager', 'so-managersearch', 'so-standalone', 'so-searchnode'] %}
       - /etc/pki/kafka-logstash.p12:/usr/share/logstash/kafka-logstash.p12:ro
+      - /opt/so/conf/kafka/kafka-truststore.jks:/etc/pki/kafka-truststore.jks:ro
       {% endif %}
       {% if GLOBALS.role == 'so-eval' %}
       - /nsm/zeek:/nsm/zeek:ro
@@ -105,6 +111,9 @@ so-logstash:
       - file: ls_pipeline_{{assigned_pipeline}}_{{CONFIGFILE.split('.')[0] | replace("/","_") }}
         {% endfor %}
       {% endfor %}
+      {% if GLOBALS.pipeline == 'KAFKA' and GLOBALS.role in ['so-manager', 'so-managersearch', 'so-standalone', 'so-searchnode'] %}
+      - file: kafkacertz
+      {% endif %}
     - require:
       {% if grains['role'] in ['so-manager', 'so-managersearch', 'so-standalone', 'so-import', 'so-heavynode', 'so-receiver'] %}
       - x509: etc_filebeat_crt
@@ -118,6 +127,9 @@ so-logstash:
       - file: cacertz
       - file: capemz
       {% endif %}
+      {% if GLOBALS.pipeline == 'KAFKA' and GLOBALS.role in ['so-manager', 'so-managersearch', 'so-standalone', 'so-searchnode'] %}
+      - file: kafkacertz
+      {% endif %}
 
 delete_so-logstash_so-status.disabled:
   file.uncomment:
diff --git a/salt/ssl/init.sls b/salt/ssl/init.sls
index abcb1a559..f5be34c40 100644
--- a/salt/ssl/init.sls
+++ b/salt/ssl/init.sls
@@ -17,8 +17,6 @@
   {% 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
@@ -666,7 +664,6 @@ elastickeyperms:
 {%- endif %}
 
 {% if grains['role'] in ['so-manager', 'so-managersearch', 'so-standalone'] %}
-
 elasticfleet_kafka_key:
   x509.private_key_managed:
     - name: /etc/pki/elasticfleet-kafka.key
@@ -696,17 +693,13 @@ elasticfleet_kafka_crt:
     - 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
+    - user: 947
     - group: 939
 
 elasticfleet_kafka_key_perms:
@@ -714,187 +707,8 @@ elasticfleet_kafka_key_perms:
     - replace: False
     - name: /etc/pki/elasticfleet-kafka.key
     - mode: 640
-    - user: 960
+    - user: 947
     - 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 %}
-
-# Standalone needs kafka-logstash for automated testing. Searchnode/manager search need it for logstash to consume from Kafka.
-# Manager will have cert, but be unused until a pipeline is created and logstash enabled.
-{% if grains['role'] in ['so-standalone', 'so-managersearch', 'so-searchnode', 'so-manager'] %}
-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 %}