JWT Authentication support for devon4j-kafka (#264)

* initial commit

* fix formatting

* Move kafka-jwt-auth to devon4j-security-jwt

* Change order of logging and token validation

* removed

* Fix annotation packagename

* Acknowledge msg with invalid token

* Include auto config for jwt auth aspect for kafka

* added loggers for retry

* log message key instead of contents

* changed from debug to info

* aspect junits

* wip-integration tests

* initial commit

* fix formatting

* Move kafka-jwt-auth to devon4j-security-jwt

* Change order of logging and token validation

* removed

* Fix annotation packagename

* Acknowledge msg with invalid token

* Include auto config for jwt auth aspect for kafka

* added loggers for retry

* log message key instead of contents

* changed from debug to info

* aspect junits

* wip-integration tests

* wip integration test

* finished integrated test and added resources

* missing test case and javadocs

* remove prefix from token

* added access control confog. build fix

* fix indentation

* Add documentation

* improve doc

Co-authored-by: Simon Spielmann <[email protected]>
Co-authored-by: Simon Spielmann <[email protected]>
Co-authored-by: Jörg Hohwiller <[email protected]>

Author: 3 people

documentation/guide-jwt.asciidoc

@@ -103,3 +103,7 @@ To allow a client to login with username and password to get a JWT for sub-seque
     http.addFilterBefore(getJwtLoginFilter(), UsernamePasswordAuthenticationFilter.class);
   }
 ----
+
+== Authentication with Kafka
+
+Authentication with JWT and Kafka is explained in the link:guide-kafka.asciidoc[Kafka guide].

documentation/guide-kafka.asciidoc

@@ -314,6 +314,37 @@ messaging.kafka.health.topicsToCheck=employeeapp-employee-v1-delete,employeeapp-
 
 These properties are provided with default values except the topicsToCheck and health check will do happen only when the property is `management.endpoint.health.enabled=true`.
 
+== Authentication
+
+=== JSON Web Token (JWT)
+
+devon4j-kafka supports authentication via JSON Web Tokens (JWT) out-of-the-box.
+To use it add a dependency to the devon4j-starter-security-jwt:
+
+[source,xml]
+-----
+<dependency>
+  <groupId>com.devonfw.java.starters</groupId>
+  <artifactId>devon4j-starter-security-jwt</artifactId>
+</dependency>
+-----
+
+The authentication via JWT needs some configuration, e.g. a keystore to verify the token signature. This is explained in the link:guide-jwt.asciidoc[JWT  documentation]. 
+
+To secure a message listener with jwt add the `@JwtAuthentication`:
+
+[source,java]
+-----
+  @JwtAuthentication
+  @KafkaListener(topics = "employeeapp-employee-v1-delete", groupId = "${messaging.kafka.consumer.groupId}")
+  public void consumer(ConsumerRecord<K, V> consumerRecord, Acknowledgment acknowledgment) {
+...
+    }
+  }
+-----
+
+With this annotation in-place each message will be checked for a valid JWT in a message header with the name `Authorization`. If a valid annotation is found the spring security context will be initialized with the user roles and "normal" authorization e.g. with `@RolesAllowed` may be used. This is also demonstrated in the kafka sample application.
+
 == Using Kafka for internal parallel processing
 Apart from the use of Kafka as "communication channel" it sometimes helpful to use Kafka internally to do parallel processing:
 

modules/kafka/pom.xml

@@ -33,7 +33,6 @@
       <artifactId>devon4j-logging</artifactId>
     </dependency>
 
-
     <dependency>
       <groupId>org.springframework.kafka</groupId>
       <artifactId>spring-kafka-test</artifactId>

modules/kafka/src/main/java/com/devonfw/module/kafka/common/messaging/api/config/MessageReceiverConfig.java

@@ -64,14 +64,18 @@ public KafkaListenerContainerProperties messageKafkaListenerContainerProperties(
   /**
    * Creates the bean for {@link MessageListenerLoggingAspect}.
    *
+   * @param <K> the key type
+   * @param <V> the value type
+   *
    * @param messageSpanExtractor the {@link MessageSpanExtractor}
    *
    * @return the {@link MessageListenerLoggingAspect}.
    */
   @Bean
-  public MessageListenerLoggingAspect messageListenerLoggingAspect(MessageSpanExtractor messageSpanExtractor) {
+  public <K, V> MessageListenerLoggingAspect<K, V> messageListenerLoggingAspect(
+      MessageSpanExtractor<K, V> messageSpanExtractor) {
 
-    MessageListenerLoggingAspect messageListenerLoggingAspect = new MessageListenerLoggingAspect();
+    MessageListenerLoggingAspect<K, V> messageListenerLoggingAspect = new MessageListenerLoggingAspect<>();
     messageListenerLoggingAspect.setTracer(this.tracer);
     messageListenerLoggingAspect.setSpanExtractor(messageSpanExtractor);
     return messageListenerLoggingAspect;

modules/kafka/src/main/java/com/devonfw/module/kafka/common/messaging/logging/impl/EventKey.java

@@ -8,7 +8,7 @@ public enum EventKey {
   /**
    * MESSAGE_SENT_SUCCESSFULLY
    */
-  MESSAGE_SENT_SUCCESSFULLY("The message {} ​​was placed in the topic {}, partition {}, offset {}."),
+  MESSAGE_SENT_SUCCESSFULLY("The message with ID {} ​​was placed in the topic {}, partition {}, offset {}."),
 
   /**
    * MESSAGE_NOT_SENT

modules/kafka/src/main/java/com/devonfw/module/kafka/common/messaging/logging/impl/MessageListenerLoggingAspect.java

@@ -30,10 +30,13 @@
 /**
  * This aspect class is used to listen the {@link KafkaListener} and to log the {@link ConsumerRecord} received.
  *
+ * @param <K> the key type
+ * @param <V> the value type
+ *
  */
 @Aspect
 @Order(0)
-public class MessageListenerLoggingAspect {
+public class MessageListenerLoggingAspect<K, V> {
 
   private static final Logger LOG = LoggerFactory.getLogger(MessageListenerLoggingAspect.class);
 
@@ -51,7 +54,7 @@ public class MessageListenerLoggingAspect {
   /**
    * The {@link MessageSpanExtractor}.
    */
-  protected MessageSpanExtractor spanExtractor;
+  protected MessageSpanExtractor<K, V> spanExtractor;
 
   /**
    * Set the {@link Tracer}.
@@ -68,7 +71,7 @@ public void setTracer(Tracer tracer) {
    *
    * @param spanExtractor .
    */
-  public void setSpanExtractor(MessageSpanExtractor spanExtractor) {
+  public void setSpanExtractor(MessageSpanExtractor<K, V> spanExtractor) {
 
     this.spanExtractor = spanExtractor;
   }
@@ -83,8 +86,7 @@ public void setSpanExtractor(MessageSpanExtractor spanExtractor) {
    * @throws Throwable the {@link Throwable}
    */
   @Around("@annotation(org.springframework.kafka.annotation.KafkaListener) && args(kafkaRecord,..)")
-  public Object logMessageProcessing(ProceedingJoinPoint call, ConsumerRecord<Object, Object> kafkaRecord)
-      throws Throwable {
+  public Object logMessageProcessing(ProceedingJoinPoint call, ConsumerRecord<K, V> kafkaRecord) throws Throwable {
 
     openSpan(kafkaRecord);
 
@@ -116,7 +118,7 @@ public Object logMessageProcessing(ProceedingJoinPoint call, ConsumerRecord<Obje
     }
   }
 
-  private void openSpan(ConsumerRecord<Object, Object> kafkaRecord) {
+  private void openSpan(ConsumerRecord<K, V> kafkaRecord) {
 
     if (ObjectUtils.isEmpty(this.tracer)) {
       return;
@@ -131,7 +133,7 @@ private void openSpan(ConsumerRecord<Object, Object> kafkaRecord) {
         (span.context().parentId() != null ? toLowerHex(span.context().parentId()) : "null"));
   }
 
-  private long determineLengthOfStayInTopic(ConsumerRecord<Object, Object> kafkaRecord) {
+  private long determineLengthOfStayInTopic(ConsumerRecord<K, V> kafkaRecord) {
 
     if (kafkaRecord.timestampType() != TimestampType.NO_TIMESTAMP_TYPE) {
       return Instant.now().toEpochMilli() - kafkaRecord.timestamp();

modules/kafka/src/main/java/com/devonfw/module/kafka/common/messaging/logging/impl/MessageLoggingSupport.java

@@ -11,15 +11,15 @@ public class MessageLoggingSupport {
    * This method is used to log event when message is sent successfully.
    *
    * @param logger the {@link Logger}
-   * @param value the message value.
+   * @param key the message key.
    * @param topic the topic of the message.
    * @param partition the partition
    * @param offset the offset
    *
    */
-  public void logMessageSent(Logger logger, String value, String topic, Integer partition, Long offset) {
+  public void logMessageSent(Logger logger, String key, String topic, Integer partition, Long offset) {
 
-    logger.info(EventKey.MESSAGE_SENT_SUCCESSFULLY.getMessage(), value, topic, partition, offset);
+    logger.info(EventKey.MESSAGE_SENT_SUCCESSFULLY.getMessage(), key, topic, partition, offset);
   }
 
   /**

modules/kafka/src/main/java/com/devonfw/module/kafka/common/messaging/logging/impl/ProducerLoggingListener.java

@@ -32,12 +32,16 @@ public ProducerLoggingListener(MessageLoggingSupport loggingSupport) {
   @Override
   public void onSuccess(String topic, Integer partition, Object key, Object value, RecordMetadata recordMetadata) {
 
+    String messageKey = "<no message key>";
+    if (key != null) {
+      key = key.toString();
+    }
     if (recordMetadata != null) {
-      this.loggingSupport.logMessageSent(LOG, value.toString(), recordMetadata.topic(), recordMetadata.partition(),
+      this.loggingSupport.logMessageSent(LOG, messageKey, recordMetadata.topic(), recordMetadata.partition(),
           recordMetadata.offset());
 
     } else {
-      this.loggingSupport.logMessageSent(LOG, value.toString(), topic, partition, null);
+      this.loggingSupport.logMessageSent(LOG, messageKey, topic, partition, null);
     }
   }
 

modules/kafka/src/main/java/com/devonfw/module/kafka/common/messaging/retry/impl/DefaultRetryPolicy.java

@@ -77,6 +77,9 @@ public boolean canRetry(ConsumerRecord<K, V> consumerRecord, MessageRetryContext
       throw new IllegalArgumentException("The \"ex \" parameter cannot be null.");
     }
 
+    LOG.info("proceeding with retry for the message {} and due to {}", consumerRecord.value().toString(),
+        ex.getMessage());
+
     if (retryContext != null && retryContext.getRetryUntil() != null
         && retryContext.getCurrentRetryCount() < this.retryCount) {
       return canRetry(retryContext, ex);

modules/kafka/src/main/java/com/devonfw/module/kafka/common/messaging/trace/api/config/TraceConfig.java

@@ -27,13 +27,16 @@ public MessageSpanInjector messageSpanInjector() {
 
   /**
    * Creates bean for the {@link MessageSpanExtractor}
+   * 
+   * @param <K> the key type
+   * @param <V> the value type
    *
    * @return the {@link MessageSpanExtractor}
    */
   @Bean
-  public MessageSpanExtractor messageSpanExtractor() {
+  public <K, V> MessageSpanExtractor<K, V> messageSpanExtractor() {
 
-    return new MessageSpanExtractor();
+    return new MessageSpanExtractor<>();
   }
 
 }