Kafka Settings

FortiSIEM can receive events from (as a Consumer) or send parsed events (as a Producer) to a Kafka message bus. This section specifies this configuration.

As a Producer:

• Make sure you have set up a Kafka Cloud (here) with a specific Topic for FortiSIEM events.
• Make sure you have identified a set of Kafka brokers that FortiSIEM is going to send events to.
• Make sure you have configured Kafka receivers which can parse FortiSIEM events and store in a database. An example would be Logstash receiver (see here) that can store in an Elastic Search database.
• Configure event forwarding in order for FortiSIEM to send events to an external Kafka consumer.
• Supported Kafka version: 0.8

As a Consumer:

• Make sure you have set up a Kafka Cloud (here) with a specific Topic, Consumer Group and a Consumer for sending third party events to FortiSIEM.
• Make sure you have identified a set of Kafka brokers that FortiSIEM will receive events from.
• Supported Kafka version: 0.8

Setting Up Consumer

Complete these steps to configure Kafka for authentication.

Note: Tested with

• kafka_2.13-3.4.0.tgz

1. Download the source code tarball.
   https://archive.apache.org/dist/kafka/3.4.0/kafka_2.13-3.4.0.tgz
2. Generate SSL key
   

   # sudo mkdir /opt/kafka 
   # chown -R admin.admin /opt/kafka 
   # cd /opt/kafka 
   # wget https://github.com/confluentinc/librdkafka/raw/master/tests/gen-ssl-certs.sh
   # bash gen-ssl-certs.sh ca ca-cert CA_CN
   # bash gen-ssl-certs.sh -k server ca-cert broker_ $(hostname)
   # bash gen-ssl-certs.sh -k client ca-cert client_ client  
   # bash gen-ssl-certs.sh client ca-cert client_ client  
   

3. Uncompress the files and enter the "config" folder.
   
4. Modify the configuration files by appending the following to the end of the files:
   

   Note: In the following example, the following is used:
   

   username=alice

   password=alice-secret

   SSL password=abcdefgh

   Remember to replace <Kafka Server IP Address> with your actual Kafka Server IP address.

   # zookeeper.properties
   authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
   requireClientAuthScheme=sasl
    
   # zookeeper_jaas.conf  
   Server {
   org.apache.zookeeper.server.auth.DigestLoginModule required
      user_super="zookeeper"
      user_alice="alice-secret";
   };
    
   # server.properties
   listeners=SASL_SSL://<Kafka Server IP Address>:9092
   advertised.listeners=SASL_SSL://<Kafka Server IP Address>:9092
    
   sasl.enabled.mechanisms=SCRAM-SHA-512
   sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512
   security.inter.broker.protocol=SASL_SSL
   ssl.endpoint.identification.algorithm=
    
   authorizer.class.name=kafka.security.authorizer.AclAuthorizer
   allow.everyone.if.no.acl.found=true
   auto.create.topics.enable=true
   ssl.client.auth=none
   ssl.protocol = TLS
   ssl.enabled.protocols=TLSv1.3,TLSv1.2
   ssl.keystore.type=JKS
   ssl.truststore.type=JKS
   ssl.secure.random.implementation=SHA1PRNG
    
   ssl.keystore.location=/opt/kafka/broker_server.keystore.jks
   ssl.keystore.password=abcdefgh
   ssl.key.password=abcdefgh
   ssl.truststore.location=/opt/kafka/broker_server.truststore.jks
   ssl.truststore.password=abcdefgh
    
   # kafka_server_jaas.conf
   KafkaServer {
   org.apache.kafka.common.security.scram.ScramLoginModule required
   username="alice"
   password="alice-secret"
   user_alice="alice-secret";
   };
   Client {
   org.apache.zookeeper.server.auth.DigestLoginModule required
   username="alice"
   password="alice-secret";
   };
    
   # consumer.properties
   security.protocol=SASL_SSL
   sasl.mechanism=SCRAM-SHA-512
   sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="alice" password="alice-secret";
   ssl.truststore.location=/opt/kafka/client_client.truststore.jks
   ssl.truststore.password=abcdefgh
   ssl.endpoint.identification.algorithm=
    
   # producer.properties
   security.protocol=SASL_SSL
   sasl.mechanism=SCRAM-SHA-512
   sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="alice" password="alice-secret";
   ssl.truststore.location=/opt/kafka/client_client.truststore.jks
   ssl.truststore.password=abcdefgh
   ssl.endpoint.identification.algorithm=
    
   # kafka_client_jaas.conf
   KafkaClient {
   org.apache.kafka.common.security.scram.ScramLoginModule required
   username="alice"
   password="alice-secret"
   user_alice="alice-secret";
   };
   Client {
   org.apache.zookeeper.server.auth.DigestLoginModule required
   username="alice"
   password="alice-secret";
   };
    
   # topic.conf
   sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="alice" password="alice-secret";
   security.protocol=SASL_SSL
   sasl.mechanism=SCRAM-SHA-512
   ssl.truststore.location=/opt/kafka/client_client.truststore.jks
   ssl.truststore.password=abcdefgh
   ssl.endpoint.identification.algorithm=
   
   # zookeeper.properties
   authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
   requireClientAuthScheme=sasl
   jaasLoginRenew=3600000
   

5. Start zookeeper.
   

   cd ..
   export KAFKA_OPTS="-Djava.security.auth.login.config=$(\pwd)/config/zookeeper_jaas.conf"
   bin/zookeeper-server-start.sh config/zookeeper.properties
   
   (In another shell window)
   bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-512=[password=alice-secret]' --entity-type users --entity-name alice

6. Start the server (In another shell window)
   

   export KAFKA_OPTS="-Djava.security.auth.login.config=$(\pwd)/config/kafka_server_jaas.conf"
   bin/kafka-server-start.sh config/server.properties
   

7. Create topic (name=test1) (In another shell window)
   

   bin/kafka-topics.sh --create --topic test1 --bootstrap-server <Kafka Server IP Address>:9092 --partitions 3 --replication-factor 1 --command-config config/topic.conf
   

8. Start consumer.
   

   export KAFKA_OPTS="-Djava.security.auth.login.config=$(\pwd)/config/kafka_client_jaas.conf"
   bin/kafka-console-consumer.sh --topic test1 --bootstrap-server=<Kafka Server IP Address>:9092 --consumer.config=config/consumer.properties
   

   At this point, when FortiSIEM forwards events to this client, contents can be seen in the consumer window.

9. (Optional) Start producer.

   export KAFKA_OPTS="-Djava.security.auth.login.config=$(\pwd)/config/kafka_client_jaas.conf"
   bin/kafka-console-producer.sh --topic test1 --broker-list <Kafka Server IP Address>:9092 --producer.config config/producer.properties

On FortiSIEM, take the following actions.

1. Copy the following files to /etc/pki/kafka to your FortiSIEM Supervisor and to every FortiSIEM Collector.
   • ca-cert
   • client_client.pem
   • client_client.key
2. On the FortiSIEM Supervisor and Collector, make sure to set up the value properly in /opt/phoenix/config/phoenix_config.txt for the path of the above files, so FortiSIEM can query the Simple Authentication and Security Layer (SASL) with correct results for the Supervisor and each Collector.

   grep sasl /opt/phoenix/config/phoenix_config.txt

   sasl_ssl_ca_cert=/etc/pki/kafka/ca-cert

   sasl_ssl_cert_file=/etc/pki/kafka/client_client.pem

   sasl_ssl_key_file=/etc/pki/kafka/client_client.key

   sasl_ssl_password=abcdefgh

   sasl_ssl_verify=false

Setting Up FortiSIEM

Complete these steps for configuring Kafka settings in FortiSIEM:

1. Go to Admin > Settings > System > Kafka tab.
2. Click +.
3. Enter the Name and Topic.
4. Select or search the Organization from the drop-down.
5. Add Brokers by clicking + icon.
   1. Enter IP address or Host name of the broker.
   2. Enter Broker port (default 9092).
6. Click Save.
7. Select the Client Type to Producer or Consumer.
8. If the Consumer is selected in step 7, enter the Consumer Name and Group Name fields.
9. Enable Authentication if you want to apply Kafka authentication by adding a checkmark to the Authentication checkbox, then take the following steps:
   1. Select your Protocol: SASL_PLAINTEXT, SASL_SSL.
   2. Select your authentication mechanism: PLAIN, SCRAM-SHA-256, or SCRAM-SHA-512.
   3. In the User Name field, enter the user name to authenticate for the Kafka servers.
   4. In the Password field, enter the password associated with the user name to authenticate for the Kafka servers.
   5. In the Confirm Password field, re-enter the password associated with the user name to authenticate for the Kafka servers.
10. Click Save.