diff --git a/sdk/src/main/java/io/dapr/client/AbstractDaprClient.java b/sdk/src/main/java/io/dapr/client/AbstractDaprClient.java index 52e1fb069d..d079bd4cc4 100644 --- a/sdk/src/main/java/io/dapr/client/AbstractDaprClient.java +++ b/sdk/src/main/java/io/dapr/client/AbstractDaprClient.java @@ -127,7 +127,6 @@ public Mono invokeMethod( InvokeMethodRequest req = new InvokeMethodRequest(appId, methodName) .setBody(data) .setHttpExtension(httpExtension) - .setContentType(objectSerializer.getContentType()) .setMetadata(metadata); return this.invokeMethod(req, type); diff --git a/sdk/src/main/java/io/dapr/client/DaprClientImpl.java b/sdk/src/main/java/io/dapr/client/DaprClientImpl.java index ce1ee6dc0c..b4618a1e4d 100644 --- a/sdk/src/main/java/io/dapr/client/DaprClientImpl.java +++ b/sdk/src/main/java/io/dapr/client/DaprClientImpl.java @@ -143,6 +143,7 @@ import java.util.Iterator; import java.util.List; import java.util.Map; +import java.util.TreeMap; import java.util.function.Consumer; import java.util.function.Function; import java.util.stream.Collectors; @@ -692,7 +693,7 @@ public Mono invokeMethod(InvokeMethodRequest invokeMethodRequest, TypeRef List pathSegments = new ArrayList<>(Arrays.asList(DaprHttp.API_VERSION, "invoke", appId, "method")); pathSegments.addAll(Arrays.asList(methodSegments)); - final Map headers = new HashMap<>(); + final Map headers = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); headers.putAll(httpExtension.getHeaders()); if (metadata != null) { headers.putAll(metadata); @@ -701,7 +702,7 @@ public Mono invokeMethod(InvokeMethodRequest invokeMethodRequest, TypeRef if (contentType != null && !contentType.isEmpty()) { headers.put(io.dapr.client.domain.Metadata.CONTENT_TYPE, contentType); } else { - headers.put(io.dapr.client.domain.Metadata.CONTENT_TYPE, objectSerializer.getContentType()); + headers.putIfAbsent(io.dapr.client.domain.Metadata.CONTENT_TYPE, objectSerializer.getContentType()); } Mono response = Mono.deferContextual( context -> this.httpClient.invokeApi(httpMethod, pathSegments.toArray(new String[0]), diff --git a/sdk/src/main/java/io/dapr/client/DaprHttp.java b/sdk/src/main/java/io/dapr/client/DaprHttp.java index e091a6587a..31378244b3 100644 --- a/sdk/src/main/java/io/dapr/client/DaprHttp.java +++ b/sdk/src/main/java/io/dapr/client/DaprHttp.java @@ -38,6 +38,7 @@ import java.util.Objects; import java.util.Optional; import java.util.Set; +import java.util.TreeMap; import java.util.UUID; import java.util.concurrent.CompletableFuture; @@ -69,6 +70,19 @@ public class DaprHttp implements AutoCloseable { private static final Set ALLOWED_CONTEXT_IN_HEADERS = Set.of("grpc-trace-bin", "traceparent", "tracestate", "baggage"); + /** + * Hop-by-hop headers managed by {@link HttpClient} itself, which rejects them when set explicitly. + * They are dropped so headers received from an inbound request can be forwarded as-is. + */ + private static final Set RESTRICTED_HEADERS = + Set.of("connection", "content-length", "expect", "host", "upgrade"); + + /** + * Headers owned by the SDK that callers cannot override. + */ + private static final Set SDK_OWNED_HEADERS = + Set.of(Headers.DAPR_USER_AGENT.toLowerCase(), Headers.DAPR_API_TOKEN); + /** * Object mapper to parse DaprError with or without details. */ @@ -298,20 +312,7 @@ private CompletableFuture doInvokeApi( HttpRequest.Builder requestBuilder = HttpRequest.newBuilder(); requestBuilder.uri(createUri(uri, pathSegments, urlParameters)); - addHeader(requestBuilder, Headers.DAPR_USER_AGENT, Version.getSdkVersion()); - addHeader(requestBuilder, HEADER_DAPR_REQUEST_ID, UUID.randomUUID().toString()); - addHeader(requestBuilder, "Content-Type", getContentType(headers)); - addHeaders(requestBuilder, headers); - - if (daprApiToken != null) { - addHeader(requestBuilder, Headers.DAPR_API_TOKEN, daprApiToken); - } - - if (context != null) { - context.stream() - .filter(entry -> ALLOWED_CONTEXT_IN_HEADERS.contains(entry.getKey().toString().toLowerCase())) - .forEach(entry -> addHeader(requestBuilder, entry.getKey().toString(), entry.getValue().toString())); - } + buildHeaders(headers, context).forEach(requestBuilder::header); HttpRequest.BodyPublisher body = getBodyPublisher(content); @@ -332,10 +333,41 @@ private CompletableFuture doInvokeApi( .thenApply(this::createResponse); } - private static String getContentType(Map headers) { - String result = headers != null ? headers.get(Metadata.CONTENT_TYPE) : null; + /** + * Merges SDK, tracing and caller headers into a single set, matching names case-insensitively so each + * header is sent once. Precedence (lowest to highest): SDK defaults, tracing context, caller headers, + * SDK-owned headers. Hop-by-hop headers rejected by {@link HttpClient} are dropped. + */ + private Map buildHeaders(Map headers, ContextView context) { + Map result = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); + result.put(HEADER_DAPR_REQUEST_ID, UUID.randomUUID().toString()); + result.put(Metadata.CONTENT_TYPE, MEDIA_TYPE_APPLICATION_JSON); - return result == null ? MEDIA_TYPE_APPLICATION_JSON : result; + if (context != null) { + context.stream() + .filter(entry -> ALLOWED_CONTEXT_IN_HEADERS.contains(entry.getKey().toString().toLowerCase())) + .forEach(entry -> result.put(entry.getKey().toString(), entry.getValue().toString())); + } + + if (headers != null) { + headers.forEach((name, value) -> { + if (name == null || value == null) { + return; + } + + String key = name.toLowerCase(); + if (!RESTRICTED_HEADERS.contains(key) && !SDK_OWNED_HEADERS.contains(key)) { + result.put(name, value); + } + }); + } + + result.put(Headers.DAPR_USER_AGENT, Version.getSdkVersion()); + if (daprApiToken != null) { + result.put(Headers.DAPR_API_TOKEN, daprApiToken); + } + + return result; } private static URI createUri(URI uri, String[] pathSegments, Map> urlParameters) { @@ -418,18 +450,6 @@ private static String encodeQueryParam(String key, String value) { return URLEncoder.encode(key, StandardCharsets.UTF_8) + "=" + URLEncoder.encode(value, StandardCharsets.UTF_8); } - private static void addHeader(HttpRequest.Builder requestBuilder, String name, String value) { - requestBuilder.header(name, value); - } - - private static void addHeaders(HttpRequest.Builder requestBuilder, Map headers) { - if (headers == null || headers.isEmpty()) { - return; - } - - headers.forEach((k, v) -> addHeader(requestBuilder, k, v)); - } - private static HttpRequest.BodyPublisher getBodyPublisher(byte[] content) { return HttpRequest.BodyPublishers.ofByteArray(Objects.requireNonNullElse(content, EMPTY_BYTES)); } diff --git a/sdk/src/test/java/io/dapr/client/DaprClientHttpTest.java b/sdk/src/test/java/io/dapr/client/DaprClientHttpTest.java index 242fe833bf..874a919aab 100644 --- a/sdk/src/test/java/io/dapr/client/DaprClientHttpTest.java +++ b/sdk/src/test/java/io/dapr/client/DaprClientHttpTest.java @@ -46,6 +46,7 @@ import static io.dapr.utils.TestUtils.findFreePort; import static io.dapr.utils.TestUtils.formatIpAddress; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.ArgumentMatchers.any; @@ -656,6 +657,95 @@ public void invokeServiceWithContext() { assertEquals(tracestate, request.headers().firstValue("tracestate").get()); } + @Test + public void invokeServiceForwardsHttpExtensionHeadersAndMetadata() { + HttpRequest request = invokeAndCaptureRequest( + new HttpExtension(DaprHttp.HttpMethods.GET, null, Map.of("aaa", "96969")), + Map.of("bbb", "123")); + + assertEquals(List.of("96969"), request.headers().allValues("aaa")); + assertEquals(List.of("123"), request.headers().allValues("bbb")); + } + + @Test + public void invokeServiceDropsHopByHopHeadersWhenForwardingInboundHeaders() { + Map inboundHeaders = Map.of( + "Host", "127.0.0.1:3000", + "Connection", "Keep-Alive", + "Content-Length", "13", + "x-forwarded-host", "test.local"); + + HttpRequest request = invokeAndCaptureRequest( + new HttpExtension(DaprHttp.HttpMethods.POST, null, inboundHeaders), inboundHeaders); + + assertEquals(List.of("test.local"), request.headers().allValues("x-forwarded-host")); + assertEquals(List.of(), request.headers().allValues("host")); + assertEquals(List.of(), request.headers().allValues("connection")); + assertEquals(List.of(), request.headers().allValues("content-length")); + } + + @Test + public void invokeServiceSendsCallerContentTypeOnce() { + HttpRequest request = invokeAndCaptureRequest( + new HttpExtension(DaprHttp.HttpMethods.POST, null, Map.of("Content-Type", "text/plain")), null); + + assertEquals(List.of("text/plain"), request.headers().allValues("content-type")); + } + + @Test + public void invokeServiceDefaultsContentTypeToSerializer() { + HttpRequest request = invokeAndCaptureRequest(HttpExtension.POST, null); + + assertEquals(List.of("application/json"), request.headers().allValues("content-type")); + } + + @Test + public void invokeServiceExplicitHeaderOverridesTracingContextOnce() { + String contextTraceparent = "00-0af7651916cd43dd8448eb211c80319c-b9c7c989f97918e1-01"; + String inboundTraceparent = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"; + HttpRequest request = invokeAndCaptureRequest( + new HttpExtension(DaprHttp.HttpMethods.POST, null, Map.of("Traceparent", inboundTraceparent)), + null, + Context.of("traceparent", contextTraceparent)); + + assertEquals(List.of(inboundTraceparent), request.headers().allValues("traceparent")); + } + + @Test + public void invokeServiceDoesNotLetCallerOverrideSdkOwnedHeaders() { + DaprHttp tokenedDaprHttp = new DaprHttp(sidecarIp, 3000, "xyz", READ_TIMEOUT, httpClient); + daprClientHttp = buildDaprClient(tokenedDaprHttp); + + HttpRequest request = invokeAndCaptureRequest( + new HttpExtension(DaprHttp.HttpMethods.POST, null, + Map.of(Headers.DAPR_API_TOKEN, "forged", "user-agent", "curl/8.0")), + null); + + assertEquals(List.of("xyz"), request.headers().allValues(Headers.DAPR_API_TOKEN)); + assertEquals(1, request.headers().allValues(Headers.DAPR_USER_AGENT).size()); + assertNotEquals("curl/8.0", request.headers().firstValue(Headers.DAPR_USER_AGENT).orElse(null)); + } + + private HttpRequest invokeAndCaptureRequest(HttpExtension httpExtension, Map metadata) { + return invokeAndCaptureRequest(httpExtension, metadata, Context.empty()); + } + + private HttpRequest invokeAndCaptureRequest( + HttpExtension httpExtension, Map metadata, ContextView context) { + MockHttpResponse mockHttpResponse = new MockHttpResponse(new byte[0], HTTP_OK); + CompletableFuture> mockResponse = CompletableFuture.completedFuture(mockHttpResponse); + ArgumentCaptor requestCaptor = ArgumentCaptor.forClass(HttpRequest.class); + + when(httpClient.sendAsync(any(), any())).thenReturn(mockResponse); + + daprClientHttp.invokeMethod("41", "neworder", "", httpExtension, metadata, byte[].class) + .contextWrite(it -> it.putAll(context)) + .block(); + + verify(httpClient).sendAsync(requestCaptor.capture(), any()); + return requestCaptor.getValue(); + } + @Test public void closeException() { DaprHttp daprHttp = Mockito.mock(DaprHttp.class); diff --git a/sdk/src/test/java/io/dapr/client/DaprHttpTest.java b/sdk/src/test/java/io/dapr/client/DaprHttpTest.java index 739957ea85..45f2b39f44 100644 --- a/sdk/src/test/java/io/dapr/client/DaprHttpTest.java +++ b/sdk/src/test/java/io/dapr/client/DaprHttpTest.java @@ -32,6 +32,7 @@ import java.net.http.HttpRequest; import java.net.http.HttpResponse; import java.time.Duration; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; @@ -108,6 +109,29 @@ public void invokeApi_daprApiToken_present() throws IOException { assertEquals("xyz", request.headers().firstValue(Headers.DAPR_API_TOKEN).get()); } + @Test + public void invokeApi_skipsHeadersWithNullNameOrValue() { + MockHttpResponse mockHttpResponse = new MockHttpResponse(new byte[0], HTTP_OK); + CompletableFuture> mockResponse = CompletableFuture.completedFuture(mockHttpResponse); + ArgumentCaptor requestCaptor = ArgumentCaptor.forClass(HttpRequest.class); + Map headers = new HashMap<>(); + headers.put(null, "orphan-value"); + headers.put("x-tenant-id", null); + headers.put("x-correlation-id", "order-1234"); + + when(httpClient.sendAsync(any(), any())).thenReturn(mockResponse); + + DaprHttp daprHttp = new DaprHttp(sidecarIp, 3500, daprTokenApi, READ_TIMEOUT, httpClient); + daprHttp.invokeApi("POST", "v1.0/state".split("/"), null, (byte[]) null, headers, Context.empty()).block(); + + verify(httpClient).sendAsync(requestCaptor.capture(), any()); + + HttpRequest request = requestCaptor.getValue(); + + assertEquals(List.of("order-1234"), request.headers().allValues("x-correlation-id")); + assertEquals(List.of(), request.headers().allValues("x-tenant-id")); + } + @Test public void invokeApi_daprApiToken_absent() throws IOException { byte[] content = serializer.serialize(EXPECTED_RESULT);