Repository: strimzi/strimzi-kafka-bridge (java)

Issue:
Ability to Enable CORS on Bridge
Is there an option to enable CORS on the Kafka Bridge?    I’m trying to do a super simple Vue.js app to display some of the messages.  When the page is loaded, I receive the No 'Access-Control-Allow-Origin' header is present on the requested resource.

It looks like there are annotations on the standard Ingress for CORS, (nginx.ingress.kubernetes.io/enable-cors: "true”, etc), but wondering if there is something similar for the Kafka Bridge and if that would resolve the issue.

Pull request:
Enable CORS
Added support for CORS.
This PR fixes #378 

CORS parameters are defined in application.properties:

#if cors enabled
http.cors.enabled=<true|false>

#list of allowed origins; you can use java regular expression
http.cors.allowedOrigins=https://strimzi.io

#list of allowed HTTP methods, separated by comma
http.cors.allowedMethods=GET,POST,PUT,DELETE,OPTIONS,PATCH

A new test class specific for CORS has been created:
io.strimzi.kafka.bridge.http.HttpCorsTests

Patch:
diff --git a/bin/docker/kafka_bridge_config_generator.sh b/bin/docker/kafka_bridge_config_generator.sh
index 70b2787..0e464f9 100755
--- a/bin/docker/kafka_bridge_config_generator.sh
+++ b/bin/docker/kafka_bridge_config_generator.sh
@@ -122,6 +122,9 @@ HTTP_PROPERTIES=$(cat <<-EOF
 http.enabled=${KAFKA_BRIDGE_HTTP_ENABLED}
 http.host=${KAFKA_BRIDGE_HTTP_HOST}
 http.port=${KAFKA_BRIDGE_HTTP_PORT}
+http.cors.enabled=${KAFKA_BRIDGE_CORS_ENABLED}
+http.cors.allowedOrigins=${KAFKA_BRIDGE_CORS_ALLOWED_ORIGINS}
+http.cors.allowedMethods=${KAFKA_BRIDGE_CORS_ALLOWED_METHODS}
 EOF
 )
 AMQP_PROPERTIES=$(cat <<-EOF
diff --git a/config/application.properties b/config/application.properties
index 74b09e0..9bfed2b 100644
--- a/config/application.properties
+++ b/config/application.properties
@@ -24,4 +24,8 @@ amqp.messageConverter=io.strimzi.kafka.bridge.amqp.converter.AmqpDefaultMessageC
 #HTTP related settings
 http.enabled=true
 http.host=0.0.0.0
-http.port=8080
\ No newline at end of file
+http.port=8080
+#Enable CORS
+http.cors.enabled=false
+http.cors.allowedOrigins=*
+http.cors.allowedMethods=GET,POST,PUT,DELETE,OPTIONS,PATCH
\ No newline at end of file
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java b/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java
index b267839..43c6ba6 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java
@@ -19,16 +19,18 @@ import io.strimzi.kafka.bridge.http.model.HttpBridgeError;
 import io.vertx.core.AbstractVerticle;
 import io.vertx.core.Future;
 import io.vertx.core.file.FileSystem;
-import io.vertx.core.http.HttpConnection;
 import io.vertx.core.http.HttpServer;
 import io.vertx.core.http.HttpServerOptions;
+import io.vertx.core.http.HttpMethod;
 import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.http.HttpConnection;
 import io.vertx.core.json.JsonObject;
 import io.vertx.ext.web.Router;
 import io.vertx.ext.web.RoutingContext;
 import io.vertx.ext.web.api.contract.openapi3.OpenAPI3RouterFactory;
 import io.vertx.ext.web.api.validation.ValidationException;
 
+import io.vertx.ext.web.handler.CorsHandler;
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.common.serialization.ByteArrayDeserializer;
@@ -36,9 +38,17 @@ import org.apache.kafka.common.serialization.ByteArraySerializer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.HashMap;
-import java.util.Iterator;
 import java.util.Map;
+import java.util.Set;
+import java.util.Iterator;
+import java.util.HashMap;
+import java.util.HashSet;
+
+import static io.netty.handler.codec.http.HttpHeaderNames.ACCEPT;
+import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_TYPE;
+import static io.netty.handler.codec.http.HttpHeaderNames.ORIGIN;
+import static io.netty.handler.codec.http.HttpHeaderNames.ACCESS_CONTROL_ALLOW_ORIGIN;
+import static io.netty.handler.codec.http.HttpHeaderNames.ACCESS_CONTROL_ALLOW_METHODS;
 
 /**
  * Main bridge class listening for connections and handling HTTP requests.
@@ -148,6 +158,34 @@ public class HttpBridge extends AbstractVerticle implements HealthCheckable {
                 this.router.errorHandler(HttpResponseStatus.BAD_REQUEST.code(), this::errorHandler);
                 this.router.errorHandler(HttpResponseStatus.NOT_FOUND.code(), this::errorHandler);
 
+                //enable cors
+                if (this.bridgeConfig.getHttpConfig().isCorsEnabled()) {
+                    Set<String> allowedHeaders = new HashSet<>();
+                    //set predefined headers
+                    allowedHeaders.add("x-requested-with");
+                    allowedHeaders.add(ACCESS_CONTROL_ALLOW_ORIGIN.toString());
+                    allowedHeaders.add(ACCESS_CONTROL_ALLOW_METHODS.toString());
+                    allowedHeaders.add(ORIGIN.toString());
+                    allowedHeaders.add(CONTENT_TYPE.toString());
+                    allowedHeaders.add(ACCEPT.toString());
+
+                    //set allowed methods from property http.cors.allowedMethods
+                    Set<HttpMethod> allowedMethods = new HashSet<>();
+                    String configAllowedMethods = this.bridgeConfig.getHttpConfig().getCorsAllowedMethods();
+                    String[] configAllowedMethodsArray = configAllowedMethods.split(",");
+                    for (String method: configAllowedMethodsArray)
+                        allowedMethods.add(HttpMethod.valueOf(method));
+
+                    //set allowed origins from property http.cors.allowedOrigins
+                    String allowedOrigins = this.bridgeConfig.getHttpConfig().getCorsAllowedOrigins();
+
+                    log.info("Allowed origins for Cors: {}", allowedOrigins);
+
+                    this.router.route().handler(CorsHandler.create(allowedOrigins)
+                            .allowedHeaders(allowedHeaders)
+                            .allowedMethods(allowedMethods));
+                }
+
                 log.info("Starting HTTP-Kafka bridge verticle...");
                 this.httpBridgeContext = new HttpBridgeContext<>();
                 this.bindHttpServer(startFuture);
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/HttpConfig.java b/src/main/java/io/strimzi/kafka/bridge/http/HttpConfig.java
index 1fc9d59..bcfc0e2 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpConfig.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpConfig.java
@@ -18,6 +18,9 @@ public class HttpConfig extends AbstractConfig {
     public static final String HTTP_CONFIG_PREFIX = "http.";
 
     public static final String HTTP_ENABLED = HTTP_CONFIG_PREFIX + "enabled";
+    public static final String HTTP_CORS_ENABLED = HTTP_CONFIG_PREFIX + "cors.enabled";
+    public static final String HTTP_CORS_ALLOWED_ORIGINS = HTTP_CONFIG_PREFIX + "cors.allowedOrigins";
+    public static final String HTTP_CORS_ALLOWED_METHODS = HTTP_CONFIG_PREFIX + "cors.allowedMethods";
     public static final String HTTP_HOST = HTTP_CONFIG_PREFIX + "host";
     public static final String HTTP_PORT = HTTP_CONFIG_PREFIX + "port";
     public static final String HTTP_CONSUMER_TIMEOUT = HTTP_CONFIG_PREFIX + "timeoutSeconds";
@@ -64,6 +67,27 @@ public class HttpConfig extends AbstractConfig {
         return Long.parseLong(this.config.getOrDefault(HTTP_CONSUMER_TIMEOUT, DEFAULT_CONSUMER_TIMEOUT).toString());
     }
 
+    /**
+     * @return if CORS is enabled
+     */
+    public boolean isCorsEnabled() {
+        return Boolean.valueOf(this.config.getOrDefault(HTTP_CORS_ENABLED, false).toString());
+    }
+
+    /**
+     * @return list of CORS Allowed Origins (default *)
+     */
+    public String getCorsAllowedOrigins() {
+        return (String) this.config.getOrDefault(HTTP_CORS_ALLOWED_ORIGINS, "*");
+    }
+
+    /**
+     * @return list of CORS Allowed Methods (default GET,POST,PUT,DELETE,OPTIONS,PATCH)
+     */
+    public String getCorsAllowedMethods() {
+        return (String) this.config.getOrDefault(HTTP_CORS_ALLOWED_METHODS, "GET,POST,PUT,DELETE,OPTIONS,PATCH");
+    }
+
     /**
      * Loads HTTP related configuration parameters from a related map
      *
diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties
index 74b09e0..9bfed2b 100644
--- a/src/main/resources/application.properties
+++ b/src/main/resources/application.properties
@@ -24,4 +24,8 @@ amqp.messageConverter=io.strimzi.kafka.bridge.amqp.converter.AmqpDefaultMessageC
 #HTTP related settings
 http.enabled=true
 http.host=0.0.0.0
-http.port=8080
\ No newline at end of file
+http.port=8080
+#Enable CORS
+http.cors.enabled=false
+http.cors.allowedOrigins=*
+http.cors.allowedMethods=GET,POST,PUT,DELETE,OPTIONS,PATCH
\ No newline at end of file


Test patch:
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/HttpCorsTests.java b/src/test/java/io/strimzi/kafka/bridge/http/HttpCorsTests.java
new file mode 100644
index 0000000..cfd2c5f
--- /dev/null
+++ b/src/test/java/io/strimzi/kafka/bridge/http/HttpCorsTests.java
@@ -0,0 +1,217 @@
+/*
+ * Copyright Strimzi authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+
+package io.strimzi.kafka.bridge.http;
+
+import io.strimzi.kafka.bridge.HealthChecker;
+import io.strimzi.kafka.bridge.config.BridgeConfig;
+import io.strimzi.kafka.bridge.facades.KafkaFacade;
+import io.strimzi.kafka.bridge.utils.Urls;
+import io.vertx.core.Vertx;
+import io.vertx.core.VertxOptions;
+import io.vertx.core.http.HttpMethod;
+import io.vertx.ext.web.client.WebClient;
+import io.vertx.ext.web.client.WebClientOptions;
+import io.vertx.junit5.VertxExtension;
+import io.vertx.junit5.VertxTestContext;
+import org.apache.log4j.LogManager;
+import org.apache.log4j.Logger;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.not;
+import static org.hamcrest.Matchers.hasItem;
+
+@ExtendWith(VertxExtension.class)
+@SuppressWarnings({"checkstyle:JavaNCSS"})
+public class HttpCorsTests {
+
+    static final Logger LOGGER = LogManager.getLogger(HttpCorsTests.class);
+    static Map<String, Object> config = new HashMap<>();
+    static long timeout = 5L;
+
+    static {
+        config.put(HttpConfig.HTTP_CONSUMER_TIMEOUT, timeout);
+        config.put(BridgeConfig.BRIDGE_ID, "my-bridge");
+    }
+
+    static Vertx vertx;
+    static HttpBridge httpBridge;
+    static WebClient client;
+
+    static BridgeConfig bridgeConfig;
+    static KafkaFacade kafkaCluster = new KafkaFacade();
+
+    @BeforeAll
+    static void beforeAll() {
+        kafkaCluster.start();
+    }
+
+    @BeforeEach
+    void prepare() {
+        VertxOptions options = new VertxOptions();
+        options.setMaxEventLoopExecuteTime(Long.MAX_VALUE);
+        vertx = Vertx.vertx(options);
+    }
+
+    @AfterEach
+    void cleanup() {
+        vertx.close();
+    }
+
+    @AfterAll
+    static void afterAll() {
+        kafkaCluster.stop();
+    }
+
+
+    @Test
+    public void testCorsNotEnabled(VertxTestContext context) {
+        createWebClient();
+        configureBridge(false, null);
+
+        if (!"TRUE".equalsIgnoreCase(System.getenv("STRIMZI_USE_SYSTEM_BRIDGE"))) {
+            vertx.deployVerticle(httpBridge, context.succeeding(id -> client
+                    .request(HttpMethod.OPTIONS, 8080, "localhost", "/consumers/1/instances/1/subscription")
+                    .putHeader("Origin", "https://evil.io")
+                    .putHeader("Access-Control-Request-Method", "POST")
+                    .send(ar -> context.verify(() -> {
+                        assertThat(ar.result().statusCode(), is(405));
+                        client.request(HttpMethod.POST, 8080, "localhost", "/consumers/1/instances/1/subscription")
+                                .putHeader("Origin", "https://evil.io")
+                                .send(ar2 -> context.verify(() -> {
+                                    assertThat(ar2.result().statusCode(), is(405));
+                                    context.completeNow();
+                                }));
+                    }))));
+        } else {
+            context.completeNow();
+        }
+    }
+
+    /**
+     * Real requests (GET, POST, PUT, DELETE) for domains not trusted are not allowed
+     */
+    @Test
+    public void testCorsForbidden(VertxTestContext context) {
+        createWebClient();
+        configureBridge(true, null);
+
+        if (!"TRUE".equalsIgnoreCase(System.getenv("STRIMZI_USE_SYSTEM_BRIDGE"))) {
+            vertx.deployVerticle(httpBridge, context.succeeding(id -> client
+                    .request(HttpMethod.OPTIONS, 8080, "localhost", "/consumers/1/instances/1/subscription")
+                    .putHeader("Origin", "https://evil.io")
+                    .putHeader("Access-Control-Request-Method", "POST")
+                    .send(ar -> context.verify(() -> {
+                        assertThat(ar.result().statusCode(), is(403));
+                        assertThat(ar.result().statusMessage(), is("CORS Rejected - Invalid origin"));
+                        client.request(HttpMethod.POST, 8080, "localhost", "/consumers/1/instances/1/subscription")
+                                .putHeader("Origin", "https://evil.io")
+                                .send(ar2 -> context.verify(() -> {
+                                    assertThat(ar2.result().statusCode(), is(403));
+                                    assertThat(ar2.result().statusMessage(), is("CORS Rejected - Invalid origin"));
+                                    context.completeNow();
+                                }));
+                    }))));
+
+        } else {
+            context.completeNow();
+        }
+    }
+
+    /**
+     * Real requests (GET, POST, PUT, DELETE) for domains trusted are allowed
+     */
+    @Test
+    public void testCorsOriginAllowed(VertxTestContext context) {
+        createWebClient();
+        configureBridge(true, null);
+
+        final String origin = "https://strimzi.io";
+
+        if (!"TRUE".equalsIgnoreCase(System.getenv("STRIMZI_USE_SYSTEM_BRIDGE"))) {
+            vertx.deployVerticle(httpBridge, context.succeeding(id -> client
+                    .request(HttpMethod.OPTIONS, 8080, "localhost", "/consumers/1/instances/1/subscription")
+                    .putHeader("Origin", "https://strimzi.io")
+                    .putHeader("Access-Control-Request-Method", "POST")
+                    .send(ar -> context.verify(() -> {
+                        assertThat(ar.result().statusCode(), is(200));
+                        assertThat(ar.result().getHeader("access-control-allow-origin"), is(origin));
+                        assertThat(ar.result().getHeader("access-control-allow-headers"), is("access-control-allow-origin,origin,x-requested-with,content-type,access-control-allow-methods,accept"));
+                        List<String> list = Arrays.asList(ar.result().getHeader("access-control-allow-methods").split(","));
+                        assertThat(list, hasItem("POST"));
+                        client.request(HttpMethod.POST, 8080, "localhost", "/consumers/1/instances/1/subscription")
+                                .putHeader("Origin", "https://strimzi.io")
+                                .send(ar2 -> context.verify(() -> {
+                                    //we are not creating a topic, so we will get a 404 status code
+                                    assertThat(ar2.result().statusCode(), is(404));
+                                    context.completeNow();
+                                }));
+                    }))));
+        } else {
+            context.completeNow();
+        }
+    }
+
+    /**
+     * Real requests (GET, POST, PUT, DELETE) for domains listed are allowed but not on specific HTTP methods.
+     * Browsers will control the list of allowed methods.
+     */
+    @Test
+    public void testCorsMethodNotAllowed(VertxTestContext context) {
+        createWebClient();
+        configureBridge(true, "GET,PUT,DELETE,OPTIONS,PATCH");
+
+        final String origin = "https://strimzi.io";
+
+        if (!"TRUE".equalsIgnoreCase(System.getenv("STRIMZI_USE_SYSTEM_BRIDGE"))) {
+            vertx.deployVerticle(httpBridge, context.succeeding(id -> client
+                    .request(HttpMethod.OPTIONS, 8080, "localhost", "/consumers/1/instances/1/subscription")
+                    .putHeader("Origin", "https://strimzi.io")
+                    .putHeader("Access-Control-Request-Method", "POST")
+                    .send(ar -> context.verify(() -> {
+                        assertThat(ar.result().statusCode(), is(200));
+                        assertThat(ar.result().getHeader("access-control-allow-origin"), is(origin));
+                        assertThat(ar.result().getHeader("access-control-allow-headers"), is("access-control-allow-origin,origin,x-requested-with,content-type,access-control-allow-methods,accept"));
+                        List<String> list = Arrays.asList(ar.result().getHeader("access-control-allow-methods").split(","));
+                        assertThat(list, not(hasItem("POST")));
+                        context.completeNow();
+                    }))));
+        } else {
+            context.completeNow();
+        }
+    }
+
+    private void createWebClient() {
+        client = WebClient.create(vertx, new WebClientOptions()
+                .setDefaultHost(Urls.BRIDGE_HOST)
+                .setDefaultPort(Urls.BRIDGE_PORT)
+        );
+
+    }
+
+    private void configureBridge(boolean corsEnabled, String methodsAllowed) {
+        if (!"TRUE".equalsIgnoreCase(System.getenv("STRIMZI_USE_SYSTEM_BRIDGE"))) {
+            config.put(HttpConfig.HTTP_CORS_ENABLED, String.valueOf(corsEnabled));
+            config.put(HttpConfig.HTTP_CORS_ALLOWED_ORIGINS, "https://strimzi.io");
+            config.put(HttpConfig.HTTP_CORS_ALLOWED_METHODS, methodsAllowed != null ? methodsAllowed : "GET,POST,PUT,DELETE,OPTIONS,PATCH");
+
+            bridgeConfig = BridgeConfig.fromMap(config);
+            httpBridge = new HttpBridge(bridgeConfig);
+            httpBridge.setHealthChecker(new HealthChecker());
+        }
+    }
+}