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
+ * 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.
+ *
+ *
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.
+ *
+ *
The supplier is called on every gRPC request to obtain the
+ * current token, enabling runtime token refresh without restarting the
+ * client. Useful for:
+ *
+ * - File-based tokens updated by infrastructure (K8s projected volumes)
+ * - Environment-based tokens
+ * - Custom refresh logic
+ *
+ *
+ * Automatically enables TLS (same as {@link #withToken}).
+ *
+ * @param supplier a function that returns the current JWT token
+ */
+ public DanubeClientBuilder withTokenSupplier(Supplier 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);
diff --git a/danube-client/src/main/java/com/danubemessaging/client/internal/auth/AuthService.java b/danube-client/src/main/java/com/danubemessaging/client/internal/auth/AuthService.java
index 7fbee4f..478c1a4 100644
--- a/danube-client/src/main/java/com/danubemessaging/client/internal/auth/AuthService.java
+++ b/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.
+ *
+ * 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 } on every gRPC request.
*/
public final class AuthService {
- private static final Duration TOKEN_EXPIRY = Duration.ofHours(1);
private static final Metadata.Key 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 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);
- }
}
diff --git a/danube-client/src/main/java/com/danubemessaging/client/internal/connection/ConnectionOptions.java b/danube-client/src/main/java/com/danubemessaging/client/internal/connection/ConnectionOptions.java
index cf0c36d..af2e8cd 100644
--- a/danube-client/src/main/java/com/danubemessaging/client/internal/connection/ConnectionOptions.java
+++ b/danube-client/src/main/java/com/danubemessaging/client/internal/connection/ConnectionOptions.java
@@ -2,16 +2,25 @@
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 caCertPath,
Optional clientCertPath,
Optional clientKeyPath,
- Optional apiKey) {
+ Optional token,
+ Optional> tokenSupplier) {
public static ConnectionOptions plainText() {
return new ConnectionOptions(
@@ -19,10 +28,28 @@ public static ConnectionOptions plainText() {
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 resolveToken() {
+ if (tokenSupplier.isPresent()) {
+ String supplied = tokenSupplier.get().get();
+ if (supplied != null && !supplied.isBlank()) {
+ return Optional.of(supplied);
+ }
+ }
+ return token.filter(t -> !t.isBlank());
+ }
}
diff --git a/docker/danube_broker.yml b/docker/danube_broker.yml
index 6c58672..bb43bc8 100644
--- a/docker/danube_broker.yml
+++ b/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:
diff --git a/pom.xml b/pom.xml
index d3586b2..53e4ba5 100644
--- a/pom.xml
+++ b/pom.xml
@@ -6,7 +6,7 @@
com.danube-messaging
danube-java
- 0.3.0
+ 0.4.0
pom
Danube Java
Danube Java client multi-module build