This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch camel-4.18.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.18.x by this push:
new 83090ce23791 CAMEL-24061: Wire OAuth credentials onto gRPC stub
channel in camel-zeebe
83090ce23791 is described below
commit 83090ce237916c733416229f8855212da9b4d420
Author: Claus Ibsen <[email protected]>
AuthorDate: Wed Jul 15 09:58:02 2026 +0200
CAMEL-24061: Wire OAuth credentials onto gRPC stub channel in camel-zeebe
Wrap the managedChannel with a ClientInterceptor that applies the
OAuthCredentialsProvider to every outgoing gRPC call when credentials
are configured. Fixes authentication gap where 7 of 9 ZeebeService
operations bypassed OAuth because they used a raw ManagedChannel
instead of the authenticated ZeebeClient.
Closes #24701
Co-Authored-By: Claude Opus 4.6 <[email protected]>
---
.../component/zeebe/internal/ZeebeService.java | 76 +++++++++++++++++-----
1 file changed, 61 insertions(+), 15 deletions(-)
diff --git
a/components/camel-zeebe/src/main/java/org/apache/camel/component/zeebe/internal/ZeebeService.java
b/components/camel-zeebe/src/main/java/org/apache/camel/component/zeebe/internal/ZeebeService.java
index c1e3bb97e1b2..b4571c237d48 100644
---
a/components/camel-zeebe/src/main/java/org/apache/camel/component/zeebe/internal/ZeebeService.java
+++
b/components/camel-zeebe/src/main/java/org/apache/camel/component/zeebe/internal/ZeebeService.java
@@ -17,6 +17,7 @@
package org.apache.camel.component.zeebe.internal;
+import java.io.IOException;
import java.time.Duration;
import com.fasterxml.jackson.core.JsonProcessingException;
@@ -31,8 +32,17 @@ import
io.camunda.zeebe.client.impl.oauth.OAuthCredentialsProvider;
import io.camunda.zeebe.client.impl.oauth.OAuthCredentialsProviderBuilder;
import io.camunda.zeebe.gateway.protocol.GatewayGrpc;
import io.camunda.zeebe.gateway.protocol.GatewayOuterClass;
+import io.grpc.CallOptions;
+import io.grpc.Channel;
+import io.grpc.ClientCall;
+import io.grpc.ClientInterceptor;
+import io.grpc.ClientInterceptors;
+import io.grpc.ForwardingClientCall;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
+import io.grpc.Metadata;
+import io.grpc.MethodDescriptor;
+import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import org.apache.camel.component.zeebe.model.DeploymentRequest;
import org.apache.camel.component.zeebe.model.DeploymentResponse;
@@ -51,6 +61,7 @@ public class ZeebeService {
private ZeebeClient zeebeClient;
private ManagedChannel managedChannel;
+ private Channel grpcChannel;
private ObjectMapper objectMapper;
@@ -77,15 +88,18 @@ public class ZeebeService {
public void doStart() {
String gatewayAddress = String.format("%s:%d", gatewayHost,
gatewayPort);
- if (zeebeClient == null) {
- if (clientId != null) {
- OAuthCredentialsProvider credentialsProvider = new
OAuthCredentialsProviderBuilder()
- .authorizationServerUrl(oAuthAPI)
- .audience(gatewayAddress)
- .clientId(clientId)
- .clientSecret(clientSecret)
- .build();
+ OAuthCredentialsProvider credentialsProvider = null;
+ if (clientId != null) {
+ credentialsProvider = new OAuthCredentialsProviderBuilder()
+ .authorizationServerUrl(oAuthAPI)
+ .audience(gatewayAddress)
+ .clientId(clientId)
+ .clientSecret(clientSecret)
+ .build();
+ }
+ if (zeebeClient == null) {
+ if (credentialsProvider != null) {
zeebeClient = ZeebeClient.newClientBuilder()
.gatewayAddress(gatewayAddress)
.credentialsProvider(credentialsProvider)
@@ -101,6 +115,9 @@ public class ZeebeService {
managedChannel = ManagedChannelBuilder.forAddress(gatewayHost,
gatewayPort)
.usePlaintext()
.build();
+ grpcChannel = credentialsProvider != null
+ ? ClientInterceptors.intercept(managedChannel, new
OAuthInterceptor(credentialsProvider))
+ : managedChannel;
}
}
@@ -148,7 +165,7 @@ public class ZeebeService {
resultMessage.setProcessInstanceKey(processMessage.getProcessInstanceKey());
try {
- GatewayGrpc.GatewayBlockingStub stub =
GatewayGrpc.newBlockingStub(managedChannel);
+ GatewayGrpc.GatewayBlockingStub stub =
GatewayGrpc.newBlockingStub(grpcChannel);
GatewayOuterClass.CancelProcessInstanceResponse
cancelProcessInstanceResponse
=
stub.cancelProcessInstance(GatewayOuterClass.CancelProcessInstanceRequest.newBuilder()
.setProcessInstanceKey(processMessage.getProcessInstanceKey())
@@ -170,7 +187,7 @@ public class ZeebeService {
resultMessage.setCorrelationKey(message.getCorrelationKey());
try {
- GatewayGrpc.GatewayBlockingStub stub =
GatewayGrpc.newBlockingStub(managedChannel);
+ GatewayGrpc.GatewayBlockingStub stub =
GatewayGrpc.newBlockingStub(grpcChannel);
if (message.getCorrelationKey() == null) {
LOG.error("Correlation Key is missing!");
resultMessage.setSuccess(false);
@@ -211,7 +228,7 @@ public class ZeebeService {
JobResponse resultMessage = new JobResponse();
try {
- GatewayGrpc.GatewayBlockingStub stub =
GatewayGrpc.newBlockingStub(managedChannel);
+ GatewayGrpc.GatewayBlockingStub stub =
GatewayGrpc.newBlockingStub(grpcChannel);
GatewayOuterClass.CompleteJobRequest.Builder builder =
GatewayOuterClass.CompleteJobRequest.newBuilder()
.setJobKey(message.getJobKey());
if (!message.getVariables().isEmpty()) {
@@ -239,7 +256,7 @@ public class ZeebeService {
JobResponse resultMessage = new JobResponse();
try {
- GatewayGrpc.GatewayBlockingStub stub =
GatewayGrpc.newBlockingStub(managedChannel);
+ GatewayGrpc.GatewayBlockingStub stub =
GatewayGrpc.newBlockingStub(grpcChannel);
GatewayOuterClass.FailJobRequest.Builder builder =
GatewayOuterClass.FailJobRequest.newBuilder()
.setJobKey(message.getJobKey());
builder = builder.setRetries(message.getRetries());
@@ -262,7 +279,7 @@ public class ZeebeService {
JobResponse resultMessage = new JobResponse();
try {
- GatewayGrpc.GatewayBlockingStub stub =
GatewayGrpc.newBlockingStub(managedChannel);
+ GatewayGrpc.GatewayBlockingStub stub =
GatewayGrpc.newBlockingStub(grpcChannel);
GatewayOuterClass.UpdateJobRetriesRequest.Builder builder =
GatewayOuterClass.UpdateJobRetriesRequest.newBuilder()
.setJobKey(message.getJobKey());
builder = builder.setRetries(message.getRetries());
@@ -284,7 +301,7 @@ public class ZeebeService {
JobResponse resultMessage = new JobResponse();
try {
- GatewayGrpc.GatewayBlockingStub stub =
GatewayGrpc.newBlockingStub(managedChannel);
+ GatewayGrpc.GatewayBlockingStub stub =
GatewayGrpc.newBlockingStub(grpcChannel);
GatewayOuterClass.ThrowErrorRequest.Builder builder =
GatewayOuterClass.ThrowErrorRequest.newBuilder()
.setJobKey(message.getJobKey());
builder = builder.setErrorMessage(message.getErrorMessage());
@@ -307,7 +324,7 @@ public class ZeebeService {
DeploymentResponse resultMessage = new DeploymentResponse();
try {
- GatewayGrpc.GatewayBlockingStub stub =
GatewayGrpc.newBlockingStub(managedChannel);
+ GatewayGrpc.GatewayBlockingStub stub =
GatewayGrpc.newBlockingStub(grpcChannel);
GatewayOuterClass.Resource resource =
GatewayOuterClass.Resource.newBuilder()
.setName(message.getName())
.setContent(ByteString.copyFrom(message.getContent()))
@@ -360,4 +377,33 @@ public class ZeebeService {
public JobWorker registerJobHandler(JobHandler handler, String jobType,
int timeout) {
return
zeebeClient.newWorker().jobType(jobType).handler(handler).timeout(Duration.ofSeconds(timeout)).open();
}
+
+ private static class OAuthInterceptor implements ClientInterceptor {
+
+ private final OAuthCredentialsProvider credentialsProvider;
+
+ OAuthInterceptor(OAuthCredentialsProvider credentialsProvider) {
+ this.credentialsProvider = credentialsProvider;
+ }
+
+ @Override
+ public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
+ MethodDescriptor<ReqT, RespT> method, CallOptions callOptions,
Channel next) {
+ return new
ForwardingClientCall.SimpleForwardingClientCall<>(next.newCall(method,
callOptions)) {
+ @Override
+ public void start(Listener<RespT> responseListener, Metadata
headers) {
+ try {
+ credentialsProvider.applyCredentials((key, value) ->
headers.put(
+ Metadata.Key.of(key,
Metadata.ASCII_STRING_MARSHALLER), value));
+ } catch (IOException e) {
+ throw Status.UNAUTHENTICATED
+ .withDescription("Failed to apply OAuth
credentials to gRPC call")
+ .withCause(e)
+ .asRuntimeException();
+ }
+ super.start(responseListener, headers);
+ }
+ };
+ }
+ }
}
