Added Secondary token support for Raw Http input

changelog/unreleased/issue-13702.toml

@@ -0,0 +1,5 @@
+type = "a"
+message = "Add secondary authorization token support to HTTP-based inputs."
+
+issues = ["Graylog2/graylog-plugin-enterprise#13702"]
+pulls = ["25544"]

graylog2-server/src/main/java/org/graylog/inputs/otel/transport/OTelHttpHandler.java

@@ -33,6 +33,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.net.InetSocketAddress;
+import java.util.Set;
 import java.util.function.Function;
 import java.util.stream.Stream;
 
@@ -58,9 +59,9 @@ public class OTelHttpHandler extends HttpHandler {
     private final MessageInput input;
 
     public OTelHttpHandler(boolean enableCors, String authorizationHeader,
-                           String authorizationHeaderValue, String path,
+                           Set<String> authorizationHeaderValues, String path,
                            MessageInput input) {
-        super(enableCors, authorizationHeader, authorizationHeaderValue, path);
+        super(enableCors, authorizationHeader, authorizationHeaderValues, path);
         this.input = input;
     }
 
@@ -119,7 +120,7 @@ protected void handleValidPost(ChannelHandlerContext ctx, FullHttpRequest reques
      * Sends an OTLP-conformant error response in the encoding matching the request.
      */
     private void sendOtlpError(ChannelHandlerContext ctx, FullHttpRequest request, boolean keepAlive,
-                                String origin, boolean protobuf, HttpResponseStatus status) {
+                               String origin, boolean protobuf, HttpResponseStatus status) {
         final byte[] body = OtlpHttpUtils.buildErrorStatus(status, null, protobuf);
         final String contentType = protobuf ? OtlpHttpUtils.PROTOBUF_CONTENT_TYPE : OtlpHttpUtils.JSON_CONTENT_TYPE;
         writeResponse(ctx.channel(), keepAlive, request.protocolVersion(), status, origin, body, contentType);

graylog2-server/src/main/java/org/graylog/inputs/otel/transport/OTelHttpTransport.java

@@ -75,7 +75,7 @@ protected LinkedHashMap<String, Callable<? extends ChannelHandler>> getCustomChi
         // input.processRawMessage directly. These cannot be removed without overriding
         // getChildChannelHandlers.
         handlers.replace("http-handler", () -> new OTelHttpHandler(
-                isEnableCors(), getAuthorizationHeader(), getAuthorizationHeaderValue(),
+                isEnableCors(), getAuthorizationHeader(), getAuthorizationHeaderValues(),
                 getPath(), input));
         return handlers;
     }

graylog2-server/src/main/java/org/graylog2/inputs/transports/AbstractHttpTransport.java

@@ -52,6 +52,8 @@
 import java.util.Set;
 import java.util.concurrent.Callable;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import static org.apache.commons.lang3.StringUtils.isBlank;
 import static org.apache.commons.lang3.StringUtils.isNotBlank;
@@ -71,6 +73,8 @@ abstract public class AbstractHttpTransport extends AbstractTcpTransport {
     static final String CK_AUTHORIZATION_HEADER_VALUE = "authorization_header_value";
     private static final String AUTHORIZATION_HEADER_NAME_LABEL = "Authorization Header Name";
     private static final String AUTHORIZATION_HEADER_VALUE_LABEL = "Authorization Header Value";
+    private static final String AUTHORIZATION_HEADER_VALUE_SECONDARY_LABEL = "Authorization Header Value (secondary)";
+    private static final String CK_AUTHORIZATION_HEADER_VALUE_SECONDARY = "authorization_header_value_secondary";
     static final String CK_REAL_IP_HEADER_NAME = "real_ip_header_name";
     static final String CK_ENABLE_FORWARDED_FOR = "enable_forwarded_for";
     static final String CK_REQUIRE_TRUSTED_PROXIES = "require_trusted_proxies";
@@ -81,7 +85,7 @@ abstract public class AbstractHttpTransport extends AbstractTcpTransport {
     protected final int maxChunkSize;
     private final int idleWriterTimeout;
     private final String authorizationHeader;
-    private final String authorizationHeaderValue;
+    private final Set<String> authorizationHeaderValues;
     private final Set<IpSubnet> trustedProxies;
     private final String path;
     private final boolean enableForwardedFor;
@@ -112,7 +116,11 @@ public AbstractHttpTransport(Configuration configuration,
                 ? configuration.getInt(CK_IDLE_WRITER_TIMEOUT, DEFAULT_IDLE_WRITER_TIMEOUT)
                 : DEFAULT_IDLE_WRITER_TIMEOUT;
         this.authorizationHeader = configuration.getString(CK_AUTHORIZATION_HEADER_NAME);
-        this.authorizationHeaderValue = configuration.getString(CK_AUTHORIZATION_HEADER_VALUE);
+        this.authorizationHeaderValues = Stream.of(
+                        configuration.getString(CK_AUTHORIZATION_HEADER_VALUE),
+                        configuration.getString(CK_AUTHORIZATION_HEADER_VALUE_SECONDARY)
+                ).filter(v -> v != null && !v.isBlank())
+                .collect(Collectors.toUnmodifiableSet());
         this.enableForwardedFor = configuration.getBoolean(CK_ENABLE_FORWARDED_FOR);
         this.requireTrustedProxies = configuration.getBoolean(CK_REQUIRE_TRUSTED_PROXIES);
         this.enableRealIpHeader = configuration.getBoolean(CK_ENABLE_REAL_IP_HEADER);
@@ -138,8 +146,8 @@ protected String getAuthorizationHeader() {
         return authorizationHeader;
     }
 
-    protected String getAuthorizationHeaderValue() {
-        return authorizationHeaderValue;
+    protected Set<String> getAuthorizationHeaderValues() {
+        return authorizationHeaderValues;
     }
 
     protected String getPath() {
@@ -168,7 +176,7 @@ protected LinkedHashMap<String, Callable<? extends ChannelHandler>> getCustomChi
         handlers.put("http-forwarded-for-handler", () -> new HttpForwardedForHandler(enableForwardedFor,
                 enableRealIpHeader, realIpHeaders, requireTrustedProxies, trustedProxies));
         handlers.put("http-handler",
-                () -> new HttpHandler(enableCors, authorizationHeader, authorizationHeaderValue, path));
+                () -> new HttpHandler(enableCors, authorizationHeader, authorizationHeaderValues, path));
         if (enableBulkReceiving) {
             handlers.put("http-bulk-newline-decoder",
                     () -> new LenientDelimiterBasedFrameDecoder(maxChunkSize,
@@ -183,11 +191,13 @@ protected LinkedHashMap<String, Callable<? extends ChannelHandler>> getCustomChi
     @Override
     public void launch(MessageInput input, @Nullable InputFailureRecorder inputFailureRecorder)
             throws MisfireException {
-        if (isNotBlank(authorizationHeader) && isBlank(authorizationHeaderValue)) {
-            checkForConfigFieldDependencies(AUTHORIZATION_HEADER_NAME_LABEL,
+        if (isNotBlank(authorizationHeader) && authorizationHeaderValues.isEmpty()) {
+            checkForConfigFieldDependencies(
+                    AUTHORIZATION_HEADER_NAME_LABEL,
                     AUTHORIZATION_HEADER_VALUE_LABEL);
-        } else if (isNotBlank(authorizationHeaderValue) && isBlank(authorizationHeader)) {
-            checkForConfigFieldDependencies(AUTHORIZATION_HEADER_VALUE_LABEL,
+        } else if (!authorizationHeaderValues.isEmpty() && isBlank(authorizationHeader)) {
+            checkForConfigFieldDependencies(
+                    AUTHORIZATION_HEADER_VALUE_LABEL,
                     AUTHORIZATION_HEADER_NAME_LABEL);
         }
         super.launch(input, inputFailureRecorder);
@@ -236,6 +246,13 @@ public ConfigurationRequest getRequestedConfiguration() {
                     "The secret authorization header value which all request must have in order to authenticate successfully. e.g. Bearer: <api-token>N",
                     ConfigurationField.Optional.OPTIONAL,
                     TextField.Attribute.IS_PASSWORD));
+            r.addField(new TextField(
+                    CK_AUTHORIZATION_HEADER_VALUE_SECONDARY,
+                    AUTHORIZATION_HEADER_VALUE_SECONDARY_LABEL,
+                    "",
+                    "Optional secondary authorization header value to accept during token rotation. Remove once all clients have migrated to the new token.",
+                    ConfigurationField.Optional.OPTIONAL,
+                    TextField.Attribute.IS_PASSWORD));
             r.addField(new BooleanField(
                     CK_ENABLE_FORWARDED_FOR,
                     "Take original client IP from X-Forwarded-For or Forwarded headers",

graylog2-server/src/main/java/org/graylog2/inputs/transports/netty/HttpHandler.java

@@ -32,6 +32,8 @@
 import io.netty.handler.codec.http.HttpVersion;
 import jakarta.annotation.Nullable;
 
+import java.util.Set;
+
 import static org.apache.commons.lang.StringUtils.isBlank;
 import static org.apache.commons.lang3.StringUtils.isNotBlank;
 
@@ -40,13 +42,13 @@
 public class HttpHandler extends SimpleChannelInboundHandler<HttpRequest> {
     private final boolean enableCors;
     private final String authorizationHeader;
-    private final String authorizationHeaderValue;
+    private final Set<String> authorizationHeaderValues;
     private final String path;
 
-    public HttpHandler(boolean enableCors, String authorizationHeader, String authorizationHeaderValue, String path) {
+    public HttpHandler(boolean enableCors, String authorizationHeader, Set<String> authorizationHeaderValues, String path) {
         this.enableCors = enableCors;
         this.authorizationHeader = authorizationHeader;
-        this.authorizationHeaderValue = authorizationHeaderValue;
+        this.authorizationHeaderValues = authorizationHeaderValues;
         this.path = path;
     }
 
@@ -65,7 +67,7 @@ protected void channelRead0(ChannelHandlerContext ctx, HttpRequest request) thro
 
         if (isNotBlank(authorizationHeader)) {
             final String suppliedAuthHeaderValue = request.headers().get(authorizationHeader);
-            if (isBlank(suppliedAuthHeaderValue) || !suppliedAuthHeaderValue.equals(authorizationHeaderValue)) {
+            if (isBlank(suppliedAuthHeaderValue) || !authorizationHeaderValues.contains(suppliedAuthHeaderValue)) {
                 writeResponse(channel, keepAlive, httpRequestVersion, HttpResponseStatus.UNAUTHORIZED, origin);
                 return;
             }

graylog2-server/src/test/java/org/graylog/inputs/otel/transport/OTelHttpHandlerTest.java

@@ -46,6 +46,9 @@
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.nio.charset.StandardCharsets;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
@@ -229,7 +232,7 @@ void invalidJsonReturns400WithJsonContentType() {
 
     @Test
     void requestWithValidAuthReturns200() throws Exception {
-        final EmbeddedChannel channel = createChannel(false, "Authorization", "Bearer secret");
+        final EmbeddedChannel channel = createChannel(false, "Authorization", "Bearer secret", null);
         final ExportLogsServiceRequest request = createTestRequest();
 
         final FullHttpRequest httpRequest = new DefaultFullHttpRequest(
@@ -247,9 +250,29 @@ void requestWithValidAuthReturns200() throws Exception {
         response.release();
     }
 
+    @Test
+    void requestWithValidSecondaryAuthReturns200() throws Exception {
+        final EmbeddedChannel channel = createChannel(false, "Authorization", "Bearer primary", "Bearer secondary");
+        final ExportLogsServiceRequest request = createTestRequest();
+
+        final FullHttpRequest httpRequest = new DefaultFullHttpRequest(
+                HttpVersion.HTTP_1_1, HttpMethod.POST, "/v1/logs",
+                Unpooled.wrappedBuffer(request.toByteArray()));
+        httpRequest.headers().set(HttpHeaderNames.CONTENT_TYPE, "application/x-protobuf");
+        httpRequest.headers().set(HttpHeaderNames.CONTENT_LENGTH, request.toByteArray().length);
+        httpRequest.headers().set("Authorization", "Bearer secondary");
+
+        channel.writeInbound(httpRequest);
+
+        final FullHttpResponse response = channel.readOutbound();
+        assertThat(response.status()).isEqualTo(HttpResponseStatus.OK);
+        verify(input).processRawMessage(any());
+        response.release();
+    }
+
     @Test
     void requestWithBadAuthReturns401() {
-        final EmbeddedChannel channel = createChannel(false, "Authorization", "Bearer secret");
+        final EmbeddedChannel channel = createChannel(false, "Authorization", "Bearer secret", null);
 
         final FullHttpRequest httpRequest = new DefaultFullHttpRequest(
                 HttpVersion.HTTP_1_1, HttpMethod.POST, "/v1/logs",
@@ -268,7 +291,7 @@ void requestWithBadAuthReturns401() {
 
     @Test
     void requestWithMissingAuthReturns401() {
-        final EmbeddedChannel channel = createChannel(false, "Authorization", "Bearer secret");
+        final EmbeddedChannel channel = createChannel(false, "Authorization", "Bearer secret", null);
 
         final FullHttpRequest httpRequest = new DefaultFullHttpRequest(
                 HttpVersion.HTTP_1_1, HttpMethod.POST, "/v1/logs",
@@ -286,7 +309,7 @@ void requestWithMissingAuthReturns401() {
 
     @Test
     void optionsPreflightSucceedsEvenWithAuthConfigured() {
-        final EmbeddedChannel channel = createChannel(true, "Authorization", "Bearer secret");
+        final EmbeddedChannel channel = createChannel(true, "Authorization", "Bearer secret", null);
 
         final FullHttpRequest httpRequest = new DefaultFullHttpRequest(
                 HttpVersion.HTTP_1_1, HttpMethod.OPTIONS, "/v1/logs");
@@ -306,7 +329,7 @@ void optionsPreflightSucceedsEvenWithAuthConfigured() {
 
     @Test
     void optionsRequestReturns200() {
-        final EmbeddedChannel channel = createChannel(true, null, null);
+        final EmbeddedChannel channel = createChannel(true, null, null, null);
 
         final FullHttpRequest httpRequest = new DefaultFullHttpRequest(
                 HttpVersion.HTTP_1_1, HttpMethod.OPTIONS, "/v1/logs");
@@ -325,7 +348,7 @@ void optionsRequestReturns200() {
 
     @Test
     void corsHeadersOnSuccessResponse() throws Exception {
-        final EmbeddedChannel channel = createChannel(true, null, null);
+        final EmbeddedChannel channel = createChannel(true, null, null, null);
         final ExportLogsServiceRequest request = createTestRequest();
 
         final FullHttpRequest httpRequest = new DefaultFullHttpRequest(
@@ -390,12 +413,16 @@ void processingFailureReturns500WithMatchingContentType() {
     }
 
     private EmbeddedChannel createChannel() {
-        return createChannel(false, null, null);
+        return createChannel(false, null, null, null);
     }
 
-    private EmbeddedChannel createChannel(boolean enableCors, String authHeader, String authHeaderValue) {
-        return new EmbeddedChannel(new OTelHttpHandler(enableCors, authHeader, authHeaderValue,
-                OTelHttpHandler.LOGS_PATH, input));
+    private EmbeddedChannel createChannel(boolean enableCors, String authHeader,
+                                          String primaryValue, String secondaryValue) {
+        final Set<String> authorizationHeaderValues = Stream.of(primaryValue, secondaryValue)
+                .filter(v -> v != null && !v.isBlank())
+                .collect(Collectors.toUnmodifiableSet());
+        return new EmbeddedChannel(new OTelHttpHandler(enableCors, authHeader,
+                authorizationHeaderValues, OTelHttpHandler.LOGS_PATH, input));
     }
 
     private ExportLogsServiceRequest createTestRequest() {