replace api key with jwt token for secure connection

danube-client-proto/pom.xml

@@ -7,7 +7,7 @@
   <parent>
     <groupId>com.danube-messaging</groupId>
     <artifactId>danube-java</artifactId>
-    <version>0.3.0</version>
+    <version>0.4.0</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

danube-client-proto/src/main/proto/DanubeApi.proto

@@ -209,20 +209,4 @@ message HealthCheckResponse {
         CLOSE = 1;
     }
     ClientStatus status = 1;
-}
-
-// ============================================================================================
-
-service AuthService {
-    rpc Authenticate (AuthRequest) returns (AuthResponse);
-}
-
-message AuthRequest {
-    string api_key = 1;
-}
-
-message AuthResponse {
-    string token = 1;
-}
-
-// ============================================================================================
+}

danube-client/pom.xml

@@ -7,7 +7,7 @@
   <parent>
     <groupId>com.danube-messaging</groupId>
     <artifactId>danube-java</artifactId>
-    <version>0.3.0</version>
+    <version>0.4.0</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

danube-client/src/main/java/com/danubemessaging/client/DanubeClientBuilder.java

@@ -11,6 +11,7 @@
 import java.nio.file.Path;
 import java.util.Optional;
 import java.util.concurrent.Executors;
+import java.util.function.Supplier;
 
 /**
  * Builder for {@link DanubeClient}.
@@ -28,7 +29,8 @@ public final class DanubeClientBuilder {
     private Path caCertPath;
     private Path clientCertPath;
     private Path clientKeyPath;
-    private String apiKey;
+    private String token;
+    private Supplier<String> tokenSupplier;
 
     DanubeClientBuilder() {
     }
@@ -71,15 +73,42 @@ public DanubeClientBuilder withMutualTls(Path caCertPath, Path clientCertPath, P
     }
 
     /**
-     * Enables JWT authentication using an API key.
-     * The client exchanges the API key for a bearer token on first use and caches it
-     * with automatic renewal (default token lifetime: 1 hour).
-     * Calling this method also enables TLS automatically.
+     * Sets the authentication token (JWT) for the client.
      *
-     * @param apiKey the API key issued by the Danube broker
+     * <p>Use {@code danube-admin security tokens create} to generate a token.
+     * Automatically enables TLS. If no TLS config has been set via
+     * {@link #withTls} or {@link #withMutualTls}, a default TLS config using
+     * system root certificates is applied.
+     *
+     * <p>For tokens that expire, consider {@link #withTokenSupplier} instead,
+     * which allows runtime token refresh.
+     *
+     * @param token the JWT token
      */
-    public DanubeClientBuilder withApiKey(String apiKey) {
-        this.apiKey = apiKey;
+    public DanubeClientBuilder withToken(String token) {
+        this.token = token;
+        this.useTls = true;
+        return this;
+    }
+
+    /**
+     * Sets a dynamic token supplier for the client.
+     *
+     * <p>The supplier is called on <b>every gRPC request</b> to obtain the
+     * current token, enabling runtime token refresh without restarting the
+     * client. Useful for:
+     * <ul>
+     *   <li>File-based tokens updated by infrastructure (K8s projected volumes)</li>
+     *   <li>Environment-based tokens</li>
+     *   <li>Custom refresh logic</li>
+     * </ul>
+     *
+     * <p>Automatically enables TLS (same as {@link #withToken}).
+     *
+     * @param supplier a function that returns the current JWT token
+     */
+    public DanubeClientBuilder withTokenSupplier(Supplier<String> supplier) {
+        this.tokenSupplier = supplier;
         this.useTls = true;
         return this;
     }
@@ -91,14 +120,11 @@ public DanubeClient build() {
                 Optional.ofNullable(caCertPath),
                 Optional.ofNullable(clientCertPath),
                 Optional.ofNullable(clientKeyPath),
-                Optional.ofNullable(apiKey));
+                Optional.ofNullable(token),
+                Optional.ofNullable(tokenSupplier));
 
         ConnectionManager connectionManager = new ConnectionManager(options);
-        AuthService authService = new AuthService(connectionManager, options);
-
-        if (apiKey != null && !apiKey.isBlank()) {
-            authService.authenticateClient(uri, apiKey);
-        }
+        AuthService authService = new AuthService(options);
 
         LookupService lookupService = new LookupService(connectionManager, authService);
         RetryManager retryManager = new RetryManager(0, 0, 0);

danube-client/src/main/java/com/danubemessaging/client/internal/auth/AuthService.java

@@ -1,86 +1,44 @@
 package com.danubemessaging.client.internal.auth;
 
-import com.danubemessaging.client.errors.DanubeClientException;
-import com.danubemessaging.client.internal.connection.ConnectionManager;
 import com.danubemessaging.client.internal.connection.ConnectionOptions;
-import danube.AuthServiceGrpc;
-import danube.DanubeApi;
 import io.grpc.Metadata;
 import io.grpc.stub.AbstractStub;
 import io.grpc.stub.MetadataUtils;
 import java.net.URI;
-import java.time.Duration;
-import java.time.Instant;
-import java.util.concurrent.locks.ReentrantLock;
 
 /**
- * Handles API-key authentication and bearer token caching.
+ * Handles JWT token insertion into gRPC request metadata.
+ *
+ * <p>With JWT-first authentication, the client uses a pre-generated JWT token
+ * (from {@code danube-admin security tokens create}) that is sent as
+ * {@code Authorization: Bearer <token>} on every gRPC request.
  */
 public final class AuthService {
-    private static final Duration TOKEN_EXPIRY = Duration.ofHours(1);
     private static final Metadata.Key<String> AUTHORIZATION = Metadata.Key.of("authorization",
             Metadata.ASCII_STRING_MARSHALLER);
 
-    private final ConnectionManager connectionManager;
     private final ConnectionOptions connectionOptions;
-    private final ReentrantLock tokenLock = new ReentrantLock();
 
-    private volatile String token;
-    private volatile Instant tokenExpiry;
-
-    public AuthService(ConnectionManager connectionManager, ConnectionOptions connectionOptions) {
-        this.connectionManager = connectionManager;
+    public AuthService(ConnectionOptions connectionOptions) {
         this.connectionOptions = connectionOptions;
     }
 
-    public String authenticateClient(URI address, String apiKey) {
-        var grpcConnection = connectionManager.getConnection(address, address);
-        var client = AuthServiceGrpc.newBlockingStub(grpcConnection.grpcChannel());
-        var request = DanubeApi.AuthRequest.newBuilder().setApiKey(apiKey).build();
-
-        try {
-            var response = client.authenticate(request);
-            cacheToken(response.getToken());
-            return response.getToken();
-        } catch (Exception e) {
-            throw new DanubeClientException("Authentication failed", e);
-        }
-    }
-
-    public String getValidToken(URI address, String apiKey) {
-        String currentToken = token;
-        Instant expiry = tokenExpiry;
-        if (currentToken != null && expiry != null && Instant.now().isBefore(expiry)) {
-            return currentToken;
-        }
-
-        tokenLock.lock();
-        try {
-            currentToken = token;
-            expiry = tokenExpiry;
-            if (currentToken != null && expiry != null && Instant.now().isBefore(expiry)) {
-                return currentToken;
-            }
-            return authenticateClient(address, apiKey);
-        } finally {
-            tokenLock.unlock();
-        }
-    }
-
+    /**
+     * Inserts the Bearer token header into the given metadata if a token is
+     * configured (static or via supplier).
+     */
     public void insertTokenIfNeeded(Metadata metadata, URI address) {
         connectionOptions
-                .apiKey()
-                .ifPresent(apiKey -> metadata.put(AUTHORIZATION, "Bearer " + getValidToken(address, apiKey)));
+                .resolveToken()
+                .ifPresent(token -> metadata.put(AUTHORIZATION, "Bearer " + token));
     }
 
+    /**
+     * Returns the stub with auth headers attached if a token is configured.
+     */
     public <T extends AbstractStub<T>> T attachAuthIfNeeded(T stub, URI address) {
         Metadata metadata = new Metadata();
         insertTokenIfNeeded(metadata, address);
         return stub.withInterceptors(MetadataUtils.newAttachHeadersInterceptor(metadata));
     }
-
-    private void cacheToken(String tokenValue) {
-        token = tokenValue;
-        tokenExpiry = Instant.now().plus(TOKEN_EXPIRY);
-    }
 }

danube-client/src/main/java/com/danubemessaging/client/internal/connection/ConnectionOptions.java

@@ -2,27 +2,54 @@
 
 import java.nio.file.Path;
 import java.util.Optional;
+import java.util.function.Supplier;
 
 /**
  * Connection settings used when opening gRPC channels.
+ *
+ * @param useTls         whether to use TLS for the connection
+ * @param caCertPath     optional path to the CA certificate file (PEM)
+ * @param clientCertPath optional path to the client certificate file (mTLS)
+ * @param clientKeyPath  optional path to the client private key file (mTLS)
+ * @param token          optional static JWT token for authentication
+ * @param tokenSupplier  optional dynamic token supplier called per-request
  */
 public record ConnectionOptions(
         boolean useTls,
         Optional<Path> caCertPath,
         Optional<Path> clientCertPath,
         Optional<Path> clientKeyPath,
-        Optional<String> apiKey) {
+        Optional<String> token,
+        Optional<Supplier<String>> tokenSupplier) {
 
     public static ConnectionOptions plainText() {
         return new ConnectionOptions(
                 false,
                 Optional.empty(),
                 Optional.empty(),
                 Optional.empty(),
+                Optional.empty(),
                 Optional.empty());
     }
 
     public boolean isMutualTls() {
         return clientCertPath.isPresent() && clientKeyPath.isPresent();
     }
+
+    /**
+     * Resolves the current token. If a supplier is set, calls it to get a
+     * fresh token (enabling runtime rotation). Otherwise falls back to the
+     * static token.
+     *
+     * @return the resolved token, or empty if no authentication is configured
+     */
+    public Optional<String> resolveToken() {
+        if (tokenSupplier.isPresent()) {
+            String supplied = tokenSupplier.get().get();
+            if (supplied != null && !supplied.isBlank()) {
+                return Optional.of(supplied);
+            }
+        }
+        return token.filter(t -> !t.isBlank());
+    }
 }

docker/danube_broker.yml

@@ -28,18 +28,27 @@ bootstrap_namespaces:
 # Allow producers to auto-create topics when missing
 auto_create_topics: true
 
+# Enable TLS on the admin API (default: false).
+# Set to true for remote cluster management over untrusted networks.
+# When true, uses the same cert/key from auth.tls below.
+# admin_tls: false
+
 # Security Configuration
+# mode: none — no authentication, no encryption (development/testing only)
+# mode: tls  — full security: TLS + JWT for clients, mTLS for inter-broker & Raft
+# For production with TLS enabled, see config/danube_broker_secure.yml
 auth:
-  mode: none # Options: none, tls, tlswithjwt
-#  tls:
-#    cert_file: "./cert/server-cert.pem"
-#    key_file: "./cert/server-key.pem"
-#    ca_file: "./cert/ca-cert.pem"
-#    verify_client: false
-#  jwt:
-#    secret_key: "your-secret-key"
-#    issuer: "danube-auth"
-#    expiration_time: 3600 # in seconds
+  mode: none
+  # tls:
+  #   cert_file: "./cert/server-cert.pem"
+  #   key_file: "./cert/server-key.pem"
+  #   ca_file: "./cert/ca-cert.pem"
+  # jwt:
+  #   secret_key: "your-secret-key"
+  #   issuer: "danube-auth"
+  #   expiration_time: 3600
+  # super_admins:
+  #   - "admin"
 
 # Load Manager Configuration (Automated Proactive Rebalancing)
 load_manager:

pom.xml

@@ -6,7 +6,7 @@
 
   <groupId>com.danube-messaging</groupId>
   <artifactId>danube-java</artifactId>
-  <version>0.3.0</version>
+  <version>0.4.0</version>
   <packaging>pom</packaging>
   <name>Danube Java</name>
   <description>Danube Java client multi-module build</description>