camel-docling: send the bearer token in the Authorization header

components/camel-ai/camel-docling/src/main/java/org/apache/camel/component/docling/BearerTokenDoclingServeClient.java

@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.docling;
+
+import java.net.http.HttpRequest;
+
+import ai.docling.serve.api.DoclingServeApi;
+import ai.docling.serve.client.DoclingServeClient;
+import ai.docling.serve.client.JsonReadException;
+import ai.docling.serve.client.operations.RequestContext;
+import tools.jackson.core.JacksonException;
+import tools.jackson.databind.json.JsonMapper;
+
+/**
+ * A {@link DoclingServeClient} that sends the access token in the {@code Authorization: Bearer} header.
+ * <p>
+ * The docling-serve client only supports API key authentication, which it always sends in the {@code X-Api-Key} header,
+ * and it offers no way to configure any other header. A bearer token - for example one resolved from the
+ * {@code oauthProfile} option, or one issued by an OAuth proxy sitting in front of docling-serve - therefore never
+ * reaches the server, and every request fails with a 401 even though the endpoint is configured with credentials.
+ * <p>
+ * The client leaves {@link #createRequestBuilder(RequestContext)} open for subclasses, which is where the header is
+ * added here. Everything else, including the JSON codec, mirrors the client the builder itself would have created.
+ * <p>
+ * Upstream support for a bearer token is proposed in
+ * <a href="https://github.com/docling-project/docling-java/pull/785">docling-java#785</a>; this subclass can be
+ * replaced by the client option once it lands.
+ */
+final class BearerTokenDoclingServeClient extends DoclingServeClient {
+
+    private final JsonMapper jsonMapper = JsonMapper.builder().build();
+    private final String authorization;
+
+    BearerTokenDoclingServeClient(DoclingServeClientBuilder<?, ?> builder, String token) {
+        super(builder);
+        this.authorization = "Bearer " + token;
+    }
+
+    @Override
+    protected <T> T readValue(String value, Class<T> type) {
+        try {
+            return jsonMapper.readValue(value, type);
+        } catch (JacksonException e) {
+            throw new JsonReadException(e);
+        }
+    }
+
+    @Override
+    protected <T> String writeValueAsString(T value) {
+        // Jackson 3 makes JacksonException unchecked, so it propagates as it is, as in the client of the builder
+        if (prettyPrintJson()) {
+            return jsonMapper.writerWithDefaultPrettyPrinter().writeValueAsString(value);
+        }
+        return jsonMapper.writeValueAsString(value);
+    }
+
+    @Override
+    protected <I, O> HttpRequest.Builder createRequestBuilder(RequestContext<I, O> context) {
+        return super.createRequestBuilder(context).header("Authorization", authorization);
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public <T extends DoclingServeApi, B extends DoclingServeApi.DoclingApiBuilder<T, B>> B toBuilder() {
+        // the token is not part of DoclingServeApiConfig, so a builder cannot carry it over
+        return (B) DoclingServeClient.builder().config(config());
+    }
+}

components/camel-ai/camel-docling/src/main/java/org/apache/camel/component/docling/DoclingProducer.java

@@ -196,13 +196,19 @@ protected void doStart() throws Exception {
                     .asyncPollInterval(Duration.ofMillis(configuration.getAsyncPollInterval()))
                     .asyncTimeout(Duration.ofMillis(configuration.getAsyncTimeout()));
 
-            // Apply API key authentication if configured
+            // Apply the authentication configured on the endpoint
             if (configuration.getAuthenticationScheme() == AuthenticationScheme.API_KEY
                     && configuration.getAuthenticationToken() != null) {
                 builder.apiKey(configuration.getAuthenticationToken());
             }
 
-            doclingServeApi = builder.build();
+            if (configuration.getAuthenticationScheme() == AuthenticationScheme.BEARER
+                    && configuration.getAuthenticationToken() != null) {
+                // the client has no bearer token option, so the Authorization header is added by this subclass
+                doclingServeApi = new BearerTokenDoclingServeClient(builder, configuration.getAuthenticationToken());
+            } else {
+                doclingServeApi = builder.build();
+            }
 
             LOG.info("DoclingProducer configured to use docling-serve API at: {} with authentication: {} (async mode: {})",
                     configuration.getDoclingServeUrl(),
@@ -222,6 +228,10 @@ protected void doStop() throws Exception {
         }
     }
 
+    DoclingServeApi getDoclingServeApi() {
+        return doclingServeApi;
+    }
+
     private void resolveOAuthToken() throws Exception {
         if (ObjectHelper.isEmpty(configuration.getOauthProfile())) {
             return;

components/camel-ai/camel-docling/src/test/java/org/apache/camel/component/docling/DoclingProducerAuthTest.java

@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.docling;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import ai.docling.serve.api.health.HealthCheckResponse;
+import com.sun.net.httpserver.Headers;
+import com.sun.net.httpserver.HttpServer;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Asserts which credentials the producer puts on the wire for each {@link AuthenticationScheme}, using a stub
+ * docling-serve instead of a real one.
+ */
+class DoclingProducerAuthTest extends CamelTestSupport {
+
+    private final List<Headers> requests = new CopyOnWriteArrayList<>();
+    private HttpServer server;
+
+    @BeforeEach
+    void startStubServer() throws IOException {
+        requests.clear();
+        server = HttpServer.create(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0), 0);
+        server.createContext("/", exchange -> {
+            requests.add(exchange.getRequestHeaders());
+            byte[] body = "{\"status\":\"ok\"}".getBytes(StandardCharsets.UTF_8);
+            exchange.getResponseHeaders().add("Content-Type", "application/json");
+            exchange.sendResponseHeaders(200, body.length);
+            try (OutputStream out = exchange.getResponseBody()) {
+                out.write(body);
+            }
+        });
+        server.start();
+    }
+
+    @AfterEach
+    void stopStubServer() {
+        if (server != null) {
+            server.stop(0);
+            server = null;
+        }
+    }
+
+    @Test
+    void bearerTokenIsSentInTheAuthorizationHeader() throws Exception {
+        assertThat(healthStatus("BEARER", "secret-token")).isEqualTo("ok");
+
+        assertThat(requests).hasSize(1);
+        assertThat(requests.get(0).getFirst("Authorization")).isEqualTo("Bearer secret-token");
+        assertThat(requests.get(0).getFirst("X-Api-Key")).isNull();
+    }
+
+    @Test
+    void apiKeyIsSentInTheApiKeyHeader() throws Exception {
+        assertThat(healthStatus("API_KEY", "secret-token")).isEqualTo("ok");
+
+        assertThat(requests).hasSize(1);
+        assertThat(requests.get(0).getFirst("X-Api-Key")).isEqualTo("secret-token");
+        assertThat(requests.get(0).getFirst("Authorization")).isNull();
+    }
+
+    @Test
+    void noCredentialsAreSentWhenAuthenticationIsDisabled() throws Exception {
+        assertThat(healthStatus("NONE", null)).isEqualTo("ok");
+
+        assertThat(requests).hasSize(1);
+        assertThat(requests.get(0).getFirst("Authorization")).isNull();
+        assertThat(requests.get(0).getFirst("X-Api-Key")).isNull();
+    }
+
+    private String healthStatus(String authenticationScheme, String authenticationToken) throws Exception {
+        String uri = "docling:CONVERT_TO_MARKDOWN?useDoclingServe=true&doclingServeUrl=http://localhost:"
+                     + server.getAddress().getPort()
+                     + "&authenticationScheme=" + authenticationScheme
+                     + (authenticationToken != null ? "&authenticationToken=" + authenticationToken : "");
+        DoclingEndpoint endpoint = context.getEndpoint(uri, DoclingEndpoint.class);
+        DoclingProducer producer = (DoclingProducer) endpoint.createProducer();
+        producer.start();
+        try {
+            HealthCheckResponse health = producer.getDoclingServeApi().health();
+            return health.getStatus();
+        } finally {
+            producer.stop();
+        }
+    }
+}