From b153e44762064bc2c55289407b0b37d38c41b7fa Mon Sep 17 00:00:00 2001 From: davidradl Date: Fri, 16 Jan 2026 17:12:07 +0000 Subject: [PATCH] [FLINK-38935] Change case of Bearer keyword Signed-off-by: davidradl --- .../OIDCAuthHeaderValuePreprocessor.java | 2 +- .../OIDCAuthHeaderValuePreprocessorTest.java | 115 ++++++++++++++++++ 2 files changed, 116 insertions(+), 1 deletion(-) create mode 100644 flink-connector-http/src/test/java/org/apache/flink/connector/http/preprocessor/OIDCAuthHeaderValuePreprocessorTest.java diff --git a/flink-connector-http/src/main/java/org/apache/flink/connector/http/preprocessor/OIDCAuthHeaderValuePreprocessor.java b/flink-connector-http/src/main/java/org/apache/flink/connector/http/preprocessor/OIDCAuthHeaderValuePreprocessor.java index a032f988..4771d012 100644 --- a/flink-connector-http/src/main/java/org/apache/flink/connector/http/preprocessor/OIDCAuthHeaderValuePreprocessor.java +++ b/flink-connector-http/src/main/java/org/apache/flink/connector/http/preprocessor/OIDCAuthHeaderValuePreprocessor.java @@ -59,6 +59,6 @@ public String preprocessHeaderValue(String rawValue) { oidcAuthURL, oidcExpiryReduction); // apply the OIDC authentication by adding the dynamically calculated header value. - return "BEARER " + auth.authenticate(); + return "Bearer " + auth.authenticate(); } } diff --git a/flink-connector-http/src/test/java/org/apache/flink/connector/http/preprocessor/OIDCAuthHeaderValuePreprocessorTest.java b/flink-connector-http/src/test/java/org/apache/flink/connector/http/preprocessor/OIDCAuthHeaderValuePreprocessorTest.java new file mode 100644 index 00000000..56e2bbff --- /dev/null +++ b/flink-connector-http/src/test/java/org/apache/flink/connector/http/preprocessor/OIDCAuthHeaderValuePreprocessorTest.java @@ -0,0 +1,115 @@ +/* + * 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.flink.connector.http.preprocessor; + +import org.apache.flink.connector.http.WireMockServerPortAllocator; + +import com.github.tomakehurst.wiremock.WireMockServer; +import com.github.tomakehurst.wiremock.core.WireMockConfiguration; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.util.Optional; + +import static com.github.tomakehurst.wiremock.client.WireMock.aResponse; +import static com.github.tomakehurst.wiremock.client.WireMock.post; +import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo; +import static org.assertj.core.api.Assertions.assertThat; + +class OIDCAuthHeaderValuePreprocessorTest { + + private static final String TOKEN_ENDPOINT = "/oauth/token"; + + private WireMockServer wireMockServer; + private int serverPort; + + @BeforeEach + public void setup() { + serverPort = WireMockServerPortAllocator.getServerPort(); + wireMockServer = + new WireMockServer(WireMockConfiguration.wireMockConfig().port(serverPort)); + wireMockServer.start(); + } + + @AfterEach + public void tearDown() { + wireMockServer.stop(); + } + + @Test + public void shouldReturnBearerTokenWithCorrectCasing() { + // Setup mock OIDC token endpoint + String accessToken = "test_access_token_12345"; + wireMockServer.stubFor( + post(urlEqualTo(TOKEN_ENDPOINT)) + .willReturn( + aResponse() + .withStatus(200) + .withHeader("Content-Type", "application/json") + .withBody( + "{\"access_token\": \"" + + accessToken + + "\", \"expires_in\": 3600}"))); + + String tokenEndpointUrl = "http://localhost:" + serverPort + TOKEN_ENDPOINT; + String tokenRequest = "grant_type=client_credentials&client_id=test&client_secret=secret"; + + OIDCAuthHeaderValuePreprocessor preprocessor = + new OIDCAuthHeaderValuePreprocessor( + tokenEndpointUrl, tokenRequest, Optional.of(Duration.ofSeconds(1))); + + String headerValue = preprocessor.preprocessHeaderValue("ignored"); + + // Verify the Bearer token uses correct RFC 6750 casing ("Bearer" not "BEARER") + assertThat(headerValue).startsWith("Bearer "); + assertThat(headerValue).isEqualTo("Bearer " + accessToken); + // Explicitly verify it's NOT using uppercase BEARER + assertThat(headerValue).doesNotStartWith("BEARER "); + } + + @Test + public void shouldReturnBearerTokenWithDefaultExpiryReduction() { + // Setup mock OIDC token endpoint + String accessToken = "another_test_token"; + wireMockServer.stubFor( + post(urlEqualTo(TOKEN_ENDPOINT)) + .willReturn( + aResponse() + .withStatus(200) + .withHeader("Content-Type", "application/json") + .withBody( + "{\"access_token\": \"" + + accessToken + + "\", \"expires_in\": 3600}"))); + + String tokenEndpointUrl = "http://localhost:" + serverPort + TOKEN_ENDPOINT; + String tokenRequest = "grant_type=client_credentials"; + + OIDCAuthHeaderValuePreprocessor preprocessor = + new OIDCAuthHeaderValuePreprocessor( + tokenEndpointUrl, tokenRequest, Optional.empty()); + + String headerValue = preprocessor.preprocessHeaderValue("any_raw_value"); + + // Verify correct Bearer casing per RFC 6750 + assertThat(headerValue).startsWith("Bearer "); + assertThat(headerValue).isEqualTo("Bearer " + accessToken); + } +}