MTES-MCT · maximeperraultdev · Dec 3, 2025
diff --git a/.env.dev.defaults b/.env.dev.defaults
@@ -163,6 +163,11 @@ MONITORENV_BATCH_SIZE=30
 ################################################################################
 # KAFKA
 
-MONITORENV_KAFKA_AIS_ENABLED=false
-MONITORENV_KAFKA_AIS_SERVERS=localhost:9092
+MONITORENV_KAFKA_AIS_ENABLED=true
+MONITORENV_KAFKA_AIS_SERVERS=localhost:9093
 MONITORENV_KAFKA_AIS_GROUP_ID=test
+MONITORENV_KAFKA_AIS_TRUSTSTORE=${PWD}/infra/kafka/certs/monitorenv/monitorenv-truststore.jks
+MONITORENV_KAFKA_AIS_KEYSTORE=${PWD}/infra/kafka/certs/monitorenv/monitorenv.p12
+MONITORENV_KAFKA_AIS_TRUSTSTORE_PASSWORD=changeit
+MONITORENV_KAFKA_AIS_KEYSTORE_PASSWORD=changeit
+MONITORENV_KAFKA_AIS_KEY_PASSWORD=changeit
diff --git a/.env.infra.example b/.env.infra.example
@@ -160,3 +160,8 @@ MONITORENV_BATCH_SIZE=100
 MONITORENV_KAFKA_AIS_ENABLED=
 MONITORENV_KAFKA_AIS_SERVERS=
 MONITORENV_KAFKA_AIS_GROUP_ID=
+MONITORENV_KAFKA_AIS_TRUSTSTORE=
+MONITORENV_KAFKA_AIS_KEYSTORE=
+MONITORENV_KAFKA_AIS_TRUSTSTORE_PASSWORD=
+MONITORENV_KAFKA_AIS_KEYSTORE_PASSWORD=
+MONITORENV_KAFKA_AIS_KEY_PASSWORD=
diff --git a/.env.test.defaults b/.env.test.defaults
@@ -157,6 +157,11 @@ MONITORENV_BATCH_SIZE=100
 ################################################################################
 # KAFKA
 
-MONITORENV_KAFKA_AIS_ENABLED=true
-MONITORENV_KAFKA_AIS_SERVERS=localhost:9092
-MONITORENV_KAFKA_AIS_GROUP_ID=test
+MONITORENV_KAFKA_AIS_ENABLED=false
+MONITORENV_KAFKA_AIS_SERVERS=
+MONITORENV_KAFKA_AIS_GROUP_ID=
+MONITORENV_KAFKA_AIS_TRUSTSTORE=
+MONITORENV_KAFKA_AIS_KEYSTORE=
+MONITORENV_KAFKA_AIS_TRUSTSTORE_PASSWORD=
+MONITORENV_KAFKA_AIS_KEYSTORE_PASSWORD=
+MONITORENV_KAFKA_AIS_KEY_PASSWORD=
diff --git a/Makefile b/Makefile
@@ -1,6 +1,7 @@
 BACKEND_CONFIGURATION_FOLDER=$(shell pwd)/infra/configurations/backend/
 HOST_MIGRATIONS_FOLDER=$(shell pwd)/backend/src/main/resources/db/migration
 PIPELINE_TEST_ENV_FILE=$(shell pwd)/pipeline/.env.test
+PWD=$(shell pwd)
 
 ifneq (,$(wildcard .env))
 		include .env

diff --git a/backend/src/main/kotlin/fr/gouv/cacem/monitorenv/config/KafkaAISConsumerConfig.kt b/backend/src/main/kotlin/fr/gouv/cacem/monitorenv/config/KafkaAISConsumerConfig.kt
@@ -1,7 +1,5 @@
 package fr.gouv.cacem.monitorenv.config
 
-import org.apache.kafka.clients.consumer.ConsumerConfig
-import org.apache.kafka.common.serialization.StringDeserializer
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
 import org.springframework.boot.autoconfigure.kafka.KafkaProperties
 import org.springframework.context.annotation.Bean
@@ -18,26 +16,8 @@ class KafkaAISConsumerConfig(
     val kafkaProperties: KafkaProperties,
 ) {
     @Bean
-    fun consumerFactory(): ConsumerFactory<String?, String?> {
-        val props: MutableMap<String?, Any?> = HashMap()
-        props.put(
-            ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
-            kafkaProperties.bootstrapServers,
-        )
-        props.put(
-            ConsumerConfig.GROUP_ID_CONFIG,
-            kafkaProperties.consumer.groupId,
-        )
-        props.put(
-            ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
-            StringDeserializer::class.java,
-        )
-        props.put(
-            ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
-            StringDeserializer::class.java,
-        )
-        return DefaultKafkaConsumerFactory<String?, String?>(props)
-    }
+    fun consumerFactory(): ConsumerFactory<String?, String?> =
+        DefaultKafkaConsumerFactory(kafkaProperties.buildConsumerProperties())
 
     @Bean
     fun kafkaListenerContainerFactory(): ConcurrentKafkaListenerContainerFactory<String?, String?> {

diff --git a/backend/src/main/resources/application.yml b/backend/src/main/resources/application.yml
@@ -1,11 +1,11 @@
 server:
-  port: 8880
-  forward-headers-strategy: framework
-  compression:
-    enabled: true
-    excluded-user-agents: ""
-    mime-types: text/html,text/xml,text/plain,text/css,text/javascript,application/javascript
-    min-response-size: 2048
+    port: 8880
+    forward-headers-strategy: framework
+    compression:
+        enabled: true
+        excluded-user-agents: ""
+        mime-types: text/html,text/xml,text/plain,text/css,text/javascript,application/javascript
+        min-response-size: 2048
 
 host:
   ip: ${MONITORENV_URL}
@@ -56,52 +56,63 @@ monitorenv:
         password: ${MONITORENV_LEGICEM_PASSWORD}
 
 spring:
-  jmx:
-    enabled: false
-  mvc:
-    static-path-pattern: "/**"
-  web:
-    resources:
-      static-locations: file:${STATIC_FILES_PATH}
-  flyway:
-    locations: ${FLYWAY_MIGRATION_PATH}
-  jpa:
-    hibernate:
-      ddl-auto: validate
-    properties:
-      hibernate:
-        temp:
-          use_jdbc_metadata_defaults: false
-        dialect: org.hibernate.spatial.dialect.postgis.PostgisPG95Dialect
-        jdbc:
-          time_zone: UTC
-        default_batch_fetch_size: ${MONITORENV_BATCH_SIZE:50}
-  datasource:
-    url: ${ENV_DB_URL}
-    driver-class-name: org.postgresql.Driver
-    hikari:
-      maxLifetime: 60000
-  kafka:
-    bootstrap-servers: ${MONITORENV_KAFKA_AIS_SERVERS}
-    consumer:
-      group-id: ${MONITORENV_KAFKA_AIS_GROUP_ID:test}
-      auto-offset-reset: latest
+    jmx:
+        enabled: false
+    mvc:
+        static-path-pattern: "/**"
+    web:
+        resources:
+            static-locations: file:${STATIC_FILES_PATH}
+    flyway:
+        locations: ${FLYWAY_MIGRATION_PATH}
+    jpa:
+        hibernate:
+            ddl-auto: validate
+        properties:
+            hibernate:
+                temp:
+                    use_jdbc_metadata_defaults: false
+                dialect: org.hibernate.spatial.dialect.postgis.PostgisPG95Dialect
+                jdbc:
+                    time_zone: UTC
+                default_batch_fetch_size: ${MONITORENV_BATCH_SIZE:50}
+    datasource:
+        url: ${ENV_DB_URL}
+        driver-class-name: org.postgresql.Driver
+        hikari:
+            maxLifetime: 60000
+    kafka:
+        bootstrap-servers: ${MONITORENV_KAFKA_AIS_SERVERS}
+        consumer:
+            group-id: ${MONITORENV_KAFKA_AIS_GROUP_ID:test}
+            auto-offset-reset: latest
+        properties:
+            security.protocol: SSL
+
+            ssl.keystore.location: ${MONITORENV_KAFKA_AIS_KEYSTORE}
+            ssl.keystore.password: ${MONITORENV_KAFKA_AIS_KEYSTORE_PASSWORD}
+            ssl.key.password: ${MONITORENV_KAFKA_AIS_KEY_PASSWORD}
+            ssl.keystore.type: PKCS12
+
+            ssl.truststore.location: ${MONITORENV_KAFKA_AIS_TRUSTSTORE}
+            ssl.truststore.password: ${MONITORENV_KAFKA_AIS_TRUSTSTORE_PASSWORD}
+            ssl.truststore.type: JKS
 
 management:
-  endpoints:
-    web:
-      exposure:
-        include: health
-  endpoint:
-    health:
-      show-details: always
-    metrics:
-      enable: false
+    endpoints:
+        web:
+            exposure:
+                include: health
+    endpoint:
+        health:
+            show-details: always
+        metrics:
+            enable: false
 
 monitorfish:
-  url: ${MONITORFISH_URL}
-  xApiKey: ${MONITORFISH_API_KEY}
+    url: ${MONITORFISH_URL}
+    xApiKey: ${MONITORFISH_API_KEY}
 
 rapportnav:
-  url: ${RAPPORTNAV_URL}
-  timeout: 3000
+    url: ${RAPPORTNAV_URL}
+    timeout: 3000
diff --git a/docker-compose.yml b/docker-compose.yml
@@ -37,6 +37,11 @@ services:
       - MONITORENV_KAFKA_AIS_ENABLED=${MONITORENV_KAFKA_AIS_ENABLED}
       - MONITORENV_KAFKA_AIS_SERVERS=${MONITORENV_KAFKA_AIS_SERVERS}
       - MONITORENV_KAFKA_AIS_GROUP_ID=${MONITORENV_KAFKA_AIS_GROUP_ID}
+      - MONITORENV_KAFKA_AIS_KEYSTORE=${MONITORENV_KAFKA_AIS_KEYSTORE}
+      - MONITORENV_KAFKA_AIS_KEYSTORE_PASSWORD=${MONITORENV_KAFKA_AIS_KEYSTORE_PASSWORD}
+      - MONITORENV_KAFKA_AIS_KEY_PASSWORD=${MONITORENV_KAFKA_AIS_KEY_PASSWORD}
+      - MONITORENV_KAFKA_AIS_TRUSTSTORE=${MONITORENV_KAFKA_AIS_TRUSTSTORE}
+      - MONITORENV_KAFKA_AIS_TRUSTSTORE_PASSWORD=${MONITORENV_KAFKA_AIS_TRUSTSTORE_PASSWORD}
       - MONITORENV_OIDC_ENABLED=${MONITORENV_OIDC_ENABLED}
       - MONITORENV_OIDC_LOGIN_URL=${MONITORENV_OIDC_LOGIN_URL}
       - MONITORENV_OIDC_SUCCESS_URL=${MONITORENV_OIDC_SUCCESS_URL}
@@ -196,37 +201,58 @@ services:
       retries: 30
 
   kafka:
+    container_name: kafka
     profiles: [ dev, test ]
     image: confluentinc/cp-kafka:8.1.0
     ports:
-      - "9092:9092"
+      - "9093:9093"
+    volumes:
+      - ./infra/kafka/certs/kafka/:/etc/kafka/secrets
     environment:
       KAFKA_PROCESS_ROLES: broker,controller
       KAFKA_NODE_ID: 1
-      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
-      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "SSL:SSL,CONTROLLER:SSL"
+      KAFKA_LISTENERS: SSL://0.0.0.0:9093,CONTROLLER://0.0.0.0:9092
+      KAFKA_ADVERTISED_LISTENERS: SSL://localhost:9093
       KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
       KAFKA_NUM_PARTITIONS: 1
       KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
       KAFKA_BROKER_ID: 1
-      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
+      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9092
       KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
       KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
+      KAFKA_SECURITY_INTER_BROKER_PROTOCOL: SSL
+      KAFKA_SSL_KEYSTORE_TYPE: PKCS12
+      KAFKA_SSL_KEYSTORE_LOCATION: /etc/kafka/secrets/kafka.p12
+      KAFKA_SSL_TRUSTSTORE_LOCATION: /etc/kafka/secrets/kafka-truststore.jks
+      KAFKA_SSL_KEYSTORE_FILENAME: kafka.p12
+      KAFKA_SSL_KEYSTORE_PASSWORD: changeit
+      KAFKA_SSL_TRUSTSTORE_FILENAME: kafka-truststore.jks
+      KAFKA_SSL_TRUSTSTORE_PASSWORD: changeit
+      KAFKA_SSL_KEY_CREDENTIALS: kafka.keypass
+      KAFKA_SSL_KEYSTORE_CREDENTIALS: kafka.keypass
+      KAFKA_SSL_TRUSTSTORE_CREDENTIALS: kafka.keypass
+      KAFKA_SSL_CLIENT_AUTH: required
+    #      KAFKA_OPTS: "-Djavax.net.debug=ssl,handshake,certpath"
     command: >
       bash -c "
         /etc/confluent/docker/run &
 
-        echo 'Attente du broker Kafka...';
-        until kafka-topics --bootstrap-server localhost:9092 --list >/dev/null 2>&1; do
-          sleep 2;
+        echo 'Attente du broker Kafka SSL...';
+
+        # On attend que le broker SSL soit prêt
+        until kafka-topics --bootstrap-server localhost:9093 --command-config /etc/kafka/secrets/client.properties --list >/dev/null 2>&1; do
+            sleep 2;
         done;
 
-        echo 'Kafka est prêt — début envoi.';
+        echo 'Kafka SSL est prêt — début envoi.';
         i=0;
+
         while true; do
-          echo \"auto message $i\" | kafka-console-producer --bootstrap-server localhost:9092 --topic ais;
+          echo \"auto message $i\" | kafka-console-producer --bootstrap-server localhost:9093 --topic ais --producer.config /etc/kafka/secrets/client.properties
+
           echo \"Message envoyé: auto message $i\";
-          i=\$((i+1));
+          i=$((i+1));
           sleep 15;
         done
       "

diff --git a/infra/kafka/certs/kafka/client.properties b/infra/kafka/certs/kafka/client.properties
@@ -0,0 +1,8 @@
+security.protocol=SSL
+ssl.truststore.location=/etc/kafka/secrets/monitorenv-truststore.jks
+ssl.truststore.password=changeit
+ssl.keystore.location=/etc/kafka/secrets/monitorenv.p12
+ssl.keystore.password=changeit
+ssl.key.password=changeit
+ssl.keystore.type=PKCS12
+ssl.truststore.type=JKS
diff --git a/infra/kafka/certs/monitorenv/.gitkeep b/infra/kafka/certs/monitorenv/.gitkeep
diff --git a/infra/kafka/scripts/create_certificate.sh b/infra/kafka/scripts/create_certificate.sh
@@ -0,0 +1,34 @@
+# 1. Create CA
+openssl genrsa -out ../certs/kafka/ca.key 4096
+openssl req -x509 -new -key ../certs/kafka/ca.key -days 3650 -out ../certs/kafka/ca.crt -subj "/CN=MyKafkaCA"
+
+# 2. Server key + CSR
+openssl genrsa -out ../certs/kafka/kafka.key 4096
+openssl req -new -key ../certs/kafka/kafka.key -out ../certs/kafka/kafka.csr -subj "/CN=localhost"
+
+# 3. Sign server cert with CA
+openssl x509 -req -in ../certs/kafka/kafka.csr -CA ../certs/kafka/ca.crt -CAkey ../certs/kafka/ca.key -CAcreateserial \
+  -out ../certs/kafka/kafka.crt -days 3650
+
+# 4. Client key + CSR
+openssl genrsa -out ../certs/monitorenv/monitorenv.key 4096
+openssl req -new -key ../certs/monitorenv/monitorenv.key -out ../certs/monitorenv/monitorenv.csr -subj "/CN=monitorenv"
+
+openssl x509 -req -in ../certs/monitorenv/monitorenv.csr -CA ../certs/kafka/ca.crt -CAkey ../certs/kafka/ca.key -CAcreateserial \
+  -out ../certs/monitorenv/monitorenv.crt -days 3650
+
+# 5. PKCS12 keystores
+openssl pkcs12 -export -in ../certs/kafka/kafka.crt -inkey ../certs/kafka/kafka.key -out ../certs/kafka/kafka.p12 -name kafka -password pass:changeit
+openssl pkcs12 -export -in ../certs/monitorenv/monitorenv.crt -inkey ../certs/monitorenv/monitorenv.key -out ../certs/monitorenv/monitorenv.p12 -name monitorenv -password pass:changeit
+
+# 6. Truststores
+keytool -import -trustcacerts -alias CARoot -file ../certs/kafka/ca.crt -keystore ../certs/kafka/kafka-truststore.jks -storepass changeit -noprompt
+keytool -import -trustcacerts -alias CARoot -file ../certs/kafka/ca.crt -keystore ../certs/monitorenv/monitorenv-truststore.jks -storepass changeit -noprompt
+
+# 7. Copy Client truststore and keystore
+cp ../certs/monitorenv/monitorenv-truststore.jks ../certs/kafka/monitorenv-truststore.jks
+cp ../certs/monitorenv/monitorenv.p12 ../certs/kafka/monitorenv.p12
+
+# 8. Create keypass that contains password
+touch ../certs/kafka/kafka.keypass
+echo "changeit" > ../certs/kafka/kafka.keypass