Added CustomHttpResponseHeadersFilter

application.example.yml

@@ -40,6 +40,12 @@ akhq:
       enabled: true # true by default
       name: org.akhq.log.access # Logger name
       format: "[Date: {}] [Duration: {} ms] [Url: {} {}] [Status: {}] [Ip: {}] [User: {}]" # Logger format
+    # Custom HTTP response headers configuration
+    customHttpResponseHeaders:
+      - name: "Content-Security-Policy"
+        value: "default-src 'none'; frame-src 'self'; script-src 'self'; connect-src 'self'; img-src 'self'; style-src 'self'; frame-ancestors 'self'; form-action 'self'; upgrade-insecure-requests"
+      - name: "X-Permitted-Cross-Domain-Policies"
+        value: "none"
 
   # default kafka properties for each clients, available for admin / producer / consumer (optional)
   clients-defaults:
@@ -289,3 +295,4 @@ akhq:
         - username: header-admin
           groups:
             - admin
+

docs/docs/configuration/akhq.md

@@ -54,3 +54,16 @@ akhq:
       }
     </style>
 ```
+
+## Custom HTTP response headers
+To add headers to every response please add the headers like in following example:
+```yaml
+akhq:
+  server:
+    customHttpResponseHeaders:
+      - name: "Content-Security-Policy"
+        value: "default-src 'none'; frame-src 'self'; script-src 'self'; connect-src 'self'; img-src 'self'; style-src 'self'; frame-ancestors 'self'; form-action 'self'; upgrade-insecure-requests"
+      - name: "X-Permitted-Cross-Domain-Policies"
+        value: "none"
+```
+

src/main/java/org/akhq/configs/Server.java

@@ -0,0 +1,38 @@
+package org.akhq.configs;
+
+import io.micronaut.context.annotation.ConfigurationProperties;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Configuration for AKHQ server configuration (mainly for {@code akhq.server.custom-http-response-headers}).
+ *
+ * This does NOT include settings for {@link org.akhq.middlewares.HttpServerAccessLogFilter} like
+ * {@code akhq.server.access-log.enabled} or {@code akhq.server.access-log.format} which are directly referenced in that class.
+ */
+@ConfigurationProperties("akhq.server")
+public class Server {
+
+    private static final String HEADER_KEY = "name";
+    private static final String HEADER_VALUE = "value";
+    private final Map<String,String> customHttpResponseHeaders = new HashMap<>();
+
+    public Map<String, String> getCustomHttpResponseHeaders() {
+        return customHttpResponseHeaders;
+    }
+
+    /**
+     * Convert the property entries to a {@code Map} as they are read from configuration as an {@link ArrayList}.
+     *
+     * @param customHttpResponseHeaders the list of maps from application configuration definied by property
+     * {@code akhq.server.custom-http-response-headers}.
+     */
+    public void setCustomHttpResponseHeaders(List<Map<String, String>> customHttpResponseHeaders) {
+        customHttpResponseHeaders.forEach(header ->
+                this.customHttpResponseHeaders.put(header.get(HEADER_KEY), header.get(HEADER_VALUE))
+        );
+    }
+}

src/main/java/org/akhq/middlewares/CustomHttpResponseHeadersFilter.java

@@ -0,0 +1,62 @@
+package org.akhq.middlewares;
+
+
+import io.micronaut.context.ApplicationContext;
+import io.micronaut.core.async.publisher.Publishers;
+import io.micronaut.http.HttpRequest;
+import io.micronaut.http.MutableHttpResponse;
+import io.micronaut.http.annotation.Filter;
+import io.micronaut.http.filter.HttpServerFilter;
+import io.micronaut.http.filter.ServerFilterChain;
+import org.akhq.configs.Server;
+import org.reactivestreams.Publisher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Add or replace custom HTTP response headers and remove some headers that are not allowed according to the
+ * configuration property {@code akhq.server.custom-http-response-headers}.
+ */
+@Filter("/**")
+public class CustomHttpResponseHeadersFilter implements HttpServerFilter {
+
+    private static final Logger LOG = LoggerFactory.getLogger(CustomHttpResponseHeadersFilter.class);
+    private static final String REMOVE_HEADER_VALUE = "REMOVE_HEADER";
+
+    private final Server server;
+
+    public CustomHttpResponseHeadersFilter(ApplicationContext context, Server server) {
+        this.server = server;
+        LOG.trace("Created instance of " + CustomHttpResponseHeadersFilter.class);
+    }
+
+    @Override
+    public Publisher<MutableHttpResponse<?>> doFilter(HttpRequest<?> request, ServerFilterChain chain) {
+        Publisher<MutableHttpResponse<?>> responsePublisher = chain.proceed(request);
+        LOG.trace("Adding custom headers to response.");
+        return Publishers.map(responsePublisher, mutableHttpResponse -> {
+            this.server.getCustomHttpResponseHeaders().entrySet().forEach(responseHeader -> {
+                if (responseHeader.getValue().equals(REMOVE_HEADER_VALUE)) {
+                    if (mutableHttpResponse.getHeaders().contains(responseHeader.getKey())) {
+                        String existingHeaderValue = mutableHttpResponse.getHeaders().get(responseHeader.getKey());
+                        mutableHttpResponse.getHeaders().remove(responseHeader.getKey());
+                        LOG.trace("Removed header '{}' (value was '{}')", responseHeader.getKey(), existingHeaderValue);
+                    } else {
+                        LOG.trace("Header '{}' to be removed did not exist", responseHeader.getKey());
+                    }
+                } else {
+                    if (mutableHttpResponse.getHeaders().contains(responseHeader.getKey())) {
+                        String existingHeaderValue = mutableHttpResponse.getHeaders().get(responseHeader.getKey());
+                        mutableHttpResponse.getHeaders().set(responseHeader.getKey(), responseHeader.getValue());
+                        LOG.trace("Replaced existing header '{}' by value {} (value was '{}')", responseHeader.getKey(), responseHeader.getValue(), existingHeaderValue);
+                    } else {
+                        mutableHttpResponse.getHeaders().add(responseHeader.getKey(), responseHeader.getValue());
+                        LOG.trace("Added custom header '{}' with value '{}'", responseHeader.getKey(), responseHeader.getValue());
+                    }
+                }
+            });
+            return mutableHttpResponse;
+        });
+
+    }
+}

src/test/java/org/akhq/middlewares/CustomHttpResponseHeadersFilterTest.java

@@ -0,0 +1,107 @@
+package org.akhq.middlewares;
+
+import io.micronaut.context.ApplicationContext;
+import io.micronaut.http.HttpResponse;
+import io.micronaut.http.HttpStatus;
+import io.micronaut.http.client.HttpClient;
+import io.micronaut.http.client.exceptions.HttpClientResponseException;
+import io.micronaut.runtime.server.EmbeddedServer;
+import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
+import org.junit.jupiter.api.*;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+@MicronautTest
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+public class CustomHttpResponseHeadersFilterTest {
+
+    private static EmbeddedServer server;
+    private static HttpClient client;
+
+    protected static final String REQUIRED_HEADER_CONTENT_SECURITY_POLICY = "Content-Security-Policy";
+    protected static final String REQUIRED_HEADER_X_PERMITTED_CROSS_DOMAIN_POLICIES = "X-Permitted-Cross-Domain-Policies";
+    protected static final String REQUIRED_HEADER_CROSS_ORIGIN_EMBEDDER_POLICY = "Cross-Origin-Embedder-Policy";
+    protected static final String REQUIRED_HEADER_CROSS_ORIGIN_OPENER_POLICY = "Cross-Origin-Opener-Policy";
+    protected static final String REQUIRED_HEADER_CROSS_ORIGIN_RESOURCE_POLICY = "Cross-Origin-Resource-Policy";
+    protected static final String REQUIRED_HEADER_FEATURE_POLICY = "Feature-Policy";
+    protected static final String REQUIRED_HEADER_PERMISSIONS_POLICY = "Permissions-Policy";
+    protected static final String REQUIRED_HEADER_CONTENT_SECURITY_POLICY_VALUE = "default-src 'none'; frame-src 'self'; script-src 'self'; connect-src 'self'; img-src 'self'; style-src 'self'; frame-ancestors 'self'; form-action 'self'; upgrade-insecure-requests";
+    protected static final String REQUIRED_HEADER_X_PERMITTED_CROSS_DOMAIN_POLICIES_VALUE = "none";
+    protected static final String REQUIRED_HEADER_CROSS_ORIGIN_EMBEDDER_POLICY_VALUE = "require-corp";
+    protected static final String REQUIRED_HEADER_CROSS_ORIGIN_OPENER_POLICY_VALUE = "same-origin";
+    protected static final String REQUIRED_HEADER_CROSS_ORIGIN_RESOURCE_POLICY_VALUE = "same-origin";
+    protected static final String REQUIRED_HEADER_FEATURE_POLICY_VALUE = "microphone 'none'; geolocation 'none'; usb 'none'; payment 'none'; document-domain 'none'; camera 'none'; display-capture 'none'; ambient-light-sensor 'none'";
+    protected static final String REQUIRED_HEADER_PERMISSIONS_POLICY_VALUE = "microphone=(), geolocation=(), usb=(), payment=(), document-domain=(), camera=(), display-capture=(), ambient-light-sensor=()";
+    protected static final String FORBIDDEN_HEADER_X_POWERED_BY = "X-Powered-By";
+    protected static final String FORBIDDEN_HEADER_ACCESS_CONTROL_ALLOW_ORIGIN = "Access-Control-Allow-Origin";
+    protected static final String FORBIDDEN_HEADER_VIA = "Via";
+    protected static final String FORBIDDEN_HEADER_SERVER = "Server";
+
+    @BeforeAll
+    public static void setupServer() {
+        server = ApplicationContext.run(EmbeddedServer.class, "custom-http-response-headers");
+        client = server.getApplicationContext().createBean(HttpClient.class, server.getURL());
+    }
+
+    @AfterAll
+    public static void stopServer() {
+        if (client != null) {
+            client.stop();
+        }
+        if (server != null) {
+            server.stop();
+        }
+    }
+
+    /**
+     * Test the {@link CustomHttpResponseHeadersFilter} for a valid URI. Check the response for existence of the custom headers.
+     */
+    @Test
+    @Order(1)
+    void testFilterCheckHeaders() {
+        HttpResponse<?> response = client.toBlocking().exchange("/issues/66");
+        assertEquals(HttpStatus.OK.getCode(), response.getStatus().getCode());
+        assertHeaders(response);
+    }
+
+    /**
+     * Test the {@link CustomHttpResponseHeadersFilter} for an invalid URI. But the response must nevertheless be added.
+     */
+    @Test
+    @Order(2)
+    void testFilterCheckHeadersForInvalidURI() {
+        HttpClientResponseException exception = assertThrows(HttpClientResponseException.class, () -> client.toBlocking().exchange("/issues/"));
+        assertEquals(HttpStatus.NOT_FOUND.getCode(), exception.getStatus().getCode());
+        assertHeaders(exception.getResponse());
+    }
+
+    private static void assertHeaders(HttpResponse<?> response) {
+
+        Assertions.assertEquals(REQUIRED_HEADER_CONTENT_SECURITY_POLICY_VALUE, response.getHeaders().get(REQUIRED_HEADER_CONTENT_SECURITY_POLICY),
+                "Wrong value " + response.getHeaders().get(REQUIRED_HEADER_CONTENT_SECURITY_POLICY) + " for " + REQUIRED_HEADER_CONTENT_SECURITY_POLICY);
+
+        Assertions.assertEquals(REQUIRED_HEADER_FEATURE_POLICY_VALUE, response.getHeaders().get(REQUIRED_HEADER_FEATURE_POLICY),
+                "Wrong value " + response.getHeaders().get(REQUIRED_HEADER_FEATURE_POLICY) + " for " + REQUIRED_HEADER_FEATURE_POLICY);
+
+        Assertions.assertEquals(REQUIRED_HEADER_CROSS_ORIGIN_OPENER_POLICY_VALUE, response.getHeaders().get(REQUIRED_HEADER_CROSS_ORIGIN_OPENER_POLICY),
+                "Wrong value " + response.getHeaders().get(REQUIRED_HEADER_CROSS_ORIGIN_OPENER_POLICY) + " for " + REQUIRED_HEADER_CROSS_ORIGIN_OPENER_POLICY);
+
+        Assertions.assertEquals(REQUIRED_HEADER_CROSS_ORIGIN_EMBEDDER_POLICY_VALUE, response.getHeaders().get(REQUIRED_HEADER_CROSS_ORIGIN_EMBEDDER_POLICY),
+                "Wrong value " + response.getHeaders().get(REQUIRED_HEADER_CROSS_ORIGIN_EMBEDDER_POLICY) + " for " + REQUIRED_HEADER_CROSS_ORIGIN_EMBEDDER_POLICY);
+
+        Assertions.assertEquals(REQUIRED_HEADER_CROSS_ORIGIN_RESOURCE_POLICY_VALUE, response.getHeaders().get(REQUIRED_HEADER_CROSS_ORIGIN_RESOURCE_POLICY),
+                "Wrong value " + response.getHeaders().get(REQUIRED_HEADER_CROSS_ORIGIN_RESOURCE_POLICY) + " for " + REQUIRED_HEADER_CROSS_ORIGIN_RESOURCE_POLICY);
+
+        Assertions.assertEquals(REQUIRED_HEADER_PERMISSIONS_POLICY_VALUE, response.getHeaders().get(REQUIRED_HEADER_PERMISSIONS_POLICY),
+                "Wrong value " + response.getHeaders().get(REQUIRED_HEADER_PERMISSIONS_POLICY) + " for " + REQUIRED_HEADER_PERMISSIONS_POLICY);
+
+        Assertions.assertEquals(REQUIRED_HEADER_X_PERMITTED_CROSS_DOMAIN_POLICIES_VALUE, response.getHeaders().get(REQUIRED_HEADER_X_PERMITTED_CROSS_DOMAIN_POLICIES),
+                "Wrong value " + response.getHeaders().get(REQUIRED_HEADER_X_PERMITTED_CROSS_DOMAIN_POLICIES) + " for " + REQUIRED_HEADER_X_PERMITTED_CROSS_DOMAIN_POLICIES);
+
+        assertFalse(response.getHeaders().contains(FORBIDDEN_HEADER_SERVER), FORBIDDEN_HEADER_SERVER + " erroneously exists as header");
+        assertFalse(response.getHeaders().contains(FORBIDDEN_HEADER_VIA), FORBIDDEN_HEADER_VIA + " erroneously exists as header");
+        assertFalse(response.getHeaders().contains(FORBIDDEN_HEADER_X_POWERED_BY), FORBIDDEN_HEADER_X_POWERED_BY + " erroneously exists as header");
+        assertFalse(response.getHeaders().contains(FORBIDDEN_HEADER_ACCESS_CONTROL_ALLOW_ORIGIN), FORBIDDEN_HEADER_ACCESS_CONTROL_ALLOW_ORIGIN + " erroneously exists as header");
+    }
+
+}

src/test/java/org/akhq/middlewares/CustomHttpResponseHeadersFilterTestPage.java

@@ -0,0 +1,17 @@
+package org.akhq.middlewares;
+
+import io.micronaut.http.HttpResponse;
+import io.micronaut.http.annotation.Controller;
+import io.micronaut.http.annotation.Get;
+import io.micronaut.http.annotation.PathVariable;
+
+@Controller("/issues")
+public class CustomHttpResponseHeadersFilterTestPage {
+
+    @Get("/{number}")
+    public HttpResponse<String> issue(@PathVariable Integer number) {
+        return HttpResponse.ok("Issue # " + number + "!")
+                .header("Cross-Origin-Opener-Policy", "test if being replaced")
+                .header("Via", "test if being removed");
+    }
+}

src/test/java/org/akhq/middlewares/NoCustomHttpResponseHeadersFilterTest.java

@@ -0,0 +1,46 @@
+package org.akhq.middlewares;
+
+import io.micronaut.context.ApplicationContext;
+import io.micronaut.http.HttpResponse;
+import io.micronaut.http.HttpStatus;
+import io.micronaut.http.client.HttpClient;
+import io.micronaut.runtime.server.EmbeddedServer;
+import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
+import org.junit.jupiter.api.*;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+@MicronautTest
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+public class NoCustomHttpResponseHeadersFilterTest {
+
+    private static EmbeddedServer server;
+    private static HttpClient client;
+
+    @BeforeAll
+    public static void setupServer() {
+        server = ApplicationContext.run(EmbeddedServer.class, "no-custom-http-response-headers");
+        client = server.getApplicationContext().createBean(HttpClient.class, server.getURL());
+    }
+
+    @AfterAll
+    public static void stopServer() {
+        if (client != null) {
+            client.stop();
+        }
+        if (server != null) {
+            server.stop();
+        }
+    }
+
+    /**
+     * Test the {@link CustomHttpResponseHeadersFilter} for a valid URI. Check the response for existence of the custom headers.
+     */
+    @Test
+    @Order(1)
+    void testFilterCheckNoHeaders() {
+        HttpResponse<?> response = client.toBlocking().exchange("/issues/66");
+        assertEquals(HttpStatus.OK.getCode(), response.getStatus().getCode());
+    }
+}
+

src/test/resources/application-custom-http-response-headers.yml

@@ -0,0 +1,36 @@
+micronaut:
+  application:
+    name: akhq
+  security:
+    enabled: false
+  server:
+    port: -1
+akhq:
+  connections:
+    dummy-kafka-cluster:
+      properties:
+        bootstrap.servers: "localhost:9121"
+  server:
+    customHttpResponseHeaders:
+      - name: "Content-Security-Policy"
+        value: "default-src 'none'; frame-src 'self'; script-src 'self'; connect-src 'self'; img-src 'self'; style-src 'self'; frame-ancestors 'self'; form-action 'self'; upgrade-insecure-requests"
+      - name: "X-Permitted-Cross-Domain-Policies"
+        value: "none"
+      - name: "Cross-Origin-Embedder-Policy"
+        value: "require-corp"
+      - name: "Cross-Origin-Opener-Policy"
+        value: "same-origin"
+      - name: "Cross-Origin-Resource-Policy"
+        value: "same-origin"
+      - name: "Feature-Policy"
+        value: "microphone 'none'; geolocation 'none'; usb 'none'; payment 'none'; document-domain 'none'; camera 'none'; display-capture 'none'; ambient-light-sensor 'none'"
+      - name: "Permissions-Policy"
+        value: "microphone=(), geolocation=(), usb=(), payment=(), document-domain=(), camera=(), display-capture=(), ambient-light-sensor=()"
+      - name: "X-Powered-By"
+        value: "REMOVE_HEADER"
+      - name: "Access-Control-Allow-Origin"
+        value: "REMOVE_HEADER"
+      - name: "Via"
+        value: "REMOVE_HEADER"
+      - name: "Server"
+        value: "REMOVE_HEADER"

src/test/resources/application-no-custom-http-response-headers.yml

@@ -0,0 +1,12 @@
+micronaut:
+  application:
+    name: akhq
+  security:
+    enabled: false
+  server:
+    port: -1
+akhq:
+  connections:
+    dummy-kafka-cluster:
+      properties:
+        bootstrap.servers: "localhost:9121"