Merge branch '7.2.x'

ksqldb-rest-app/src/main/java/io/confluent/ksql/api/auth/AuthenticationPluginHandler.java

@@ -44,7 +44,7 @@
  * Handler that calls any authentication plugin
  */
 public class AuthenticationPluginHandler implements Handler<RoutingContext> {
-  public static final Set<String> KSQL_AUTHENTICATION_SKIP_PATHS = ImmutableSet
+  private static final Set<String> KSQL_AUTHENTICATION_SKIP_PATHS = ImmutableSet
       .of("/v1/metadata", "/v1/metadata/id", "/healthcheck");
 
   private final Server server;
@@ -57,17 +57,9 @@ public AuthenticationPluginHandler(final Server server,
       final AuthenticationPlugin securityHandlerPlugin) {
     this.server = Objects.requireNonNull(server);
     this.securityHandlerPlugin = Objects.requireNonNull(securityHandlerPlugin);
-    // We add in all the paths that don't require authentication/authorization from
-    // KsqlAuthorizationProviderHandler
-    final Set<String> unauthenticatedPaths = new HashSet<>(KSQL_AUTHENTICATION_SKIP_PATHS);
-    // And then we add anything from the property authentication.skip.paths
-    // This preserves the behaviour from the previous Jetty based implementation
     final List<String> unauthed = server.getConfig()
         .getList(KsqlRestConfig.AUTHENTICATION_SKIP_PATHS_CONFIG);
-    unauthenticatedPaths.addAll(unauthed);
-    final String paths = String.join(",", unauthenticatedPaths);
-    final String converted = convertCommaSeparatedWilcardsToRegex(paths);
-    unauthedPathsPattern = Pattern.compile(converted);
+    unauthedPathsPattern = getAuthorizationSkipPaths(unauthed);
   }
 
   @Override
@@ -98,6 +90,18 @@ public void handle(final RoutingContext routingContext) {
     });
   }
 
+  public static Pattern getAuthorizationSkipPaths(final List<String> unauthed) {
+    // We add in all the paths that don't require authentication/authorization from
+    // KsqlAuthorizationProviderHandler
+    final Set<String> unauthenticatedPaths = new HashSet<>(KSQL_AUTHENTICATION_SKIP_PATHS);
+    // And then we add anything from the property authentication.skip.paths
+    // This preserves the behaviour from the previous Jetty based implementation
+    unauthenticatedPaths.addAll(unauthed);
+    final String paths = String.join(",", unauthenticatedPaths);
+    final String converted = convertCommaSeparatedWilcardsToRegex(paths);
+    return Pattern.compile(converted);
+  }
+
   private static class AuthPluginUser implements ApiUser {
 
     private final DefaultKsqlPrincipal principal;

ksqldb-rest-app/src/main/java/io/confluent/ksql/api/auth/KsqlAuthorizationProviderHandler.java

@@ -15,16 +15,18 @@
 
 package io.confluent.ksql.api.auth;
 
-import static io.confluent.ksql.api.auth.AuthenticationPluginHandler.KSQL_AUTHENTICATION_SKIP_PATHS;
 import static io.netty.handler.codec.http.HttpResponseStatus.FORBIDDEN;
 
+import io.confluent.ksql.api.server.Server;
+import io.confluent.ksql.rest.server.KsqlRestConfig;
 import io.confluent.ksql.security.KsqlAuthorizationProvider;
 import io.vertx.core.AsyncResult;
 import io.vertx.core.Handler;
 import io.vertx.core.Promise;
 import io.vertx.core.WorkerExecutor;
 import io.vertx.ext.auth.User;
 import io.vertx.ext.web.RoutingContext;
+import java.util.regex.Pattern;
 
 /**
  * Handler that calls a KsqlAuthorizationProvider plugin that can be used for custom authorization
@@ -33,19 +35,22 @@ public class KsqlAuthorizationProviderHandler implements Handler<RoutingContext>
 
   private final WorkerExecutor workerExecutor;
   private final KsqlAuthorizationProvider ksqlAuthorizationProvider;
+  private final Pattern unauthenticatedpaths;
 
-  public KsqlAuthorizationProviderHandler(final WorkerExecutor workerExecutor,
+  public KsqlAuthorizationProviderHandler(final Server server,
       final KsqlAuthorizationProvider ksqlAuthorizationProvider) {
-    this.workerExecutor = workerExecutor;
+    this.workerExecutor = server.getWorkerExecutor();
     this.ksqlAuthorizationProvider = ksqlAuthorizationProvider;
+    this.unauthenticatedpaths = AuthenticationPluginHandler.getAuthorizationSkipPaths(
+        server.getConfig().getList(KsqlRestConfig.AUTHENTICATION_SKIP_PATHS_CONFIG));
   }
 
   @Override
   public void handle(final RoutingContext routingContext) {
 
     final String path = routingContext.normalisedPath();
 
-    if (KSQL_AUTHENTICATION_SKIP_PATHS.contains(path)) {
+    if (unauthenticatedpaths.matcher(path).matches()) {
       routingContext.next();
       return;
     }

ksqldb-rest-app/src/main/java/io/confluent/ksql/api/server/AuthHandlers.java

@@ -60,8 +60,7 @@ static void setupAuthHandlers(final Server server, final Router router,
       // For authorization use auth provider configured via security extension (if any)
       securityExtension.getAuthorizationProvider()
           .ifPresent(ksqlAuthorizationProvider -> router.route()
-              .handler(new KsqlAuthorizationProviderHandler(server.getWorkerExecutor(),
-                  ksqlAuthorizationProvider)));
+              .handler(new KsqlAuthorizationProviderHandler(server, ksqlAuthorizationProvider)));
 
       router.route().handler(AuthHandlers::resumeHandler);
     }

ksqldb-rest-app/src/test/java/io/confluent/ksql/api/auth/AuthenticationPluginHandlerTest.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2020 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.api.auth;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+
+import com.google.common.collect.ImmutableList;
+import java.util.List;
+import java.util.regex.Pattern;
+import org.junit.Test;
+
+public class AuthenticationPluginHandlerTest {
+
+  @Test
+  public void shouldBuildRegexThatRespectsSkipPaths() {
+    // Given:
+    final List<String> configs = ImmutableList.of("/heartbeat", "/lag");
+
+    // When:
+    final Pattern skips = AuthenticationPluginHandler.getAuthorizationSkipPaths(configs);
+
+    // Then:
+    assertThat(skips.matcher("/heartbeat").matches(), is(true));
+    assertThat(skips.matcher("/foo").matches(), is(false));
+  }
+
+}

ksqldb-rest-app/src/test/java/io/confluent/ksql/api/auth/KsqlAuthorizationProviderHandlerTest.java

@@ -0,0 +1,85 @@
+/*
+ * Copyright 2020 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (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.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.api.auth;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import io.confluent.ksql.api.server.Server;
+import io.confluent.ksql.rest.server.KsqlRestConfig;
+import io.confluent.ksql.security.KsqlAuthorizationProvider;
+import io.vertx.core.WorkerExecutor;
+import io.vertx.ext.web.RoutingContext;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class KsqlAuthorizationProviderHandlerTest {
+
+  @Mock
+  private Server server;
+  @Mock
+  private KsqlAuthorizationProvider authProvider;
+  @Mock
+  private WorkerExecutor workerExecutor;
+  @Mock
+  private RoutingContext routingContext;
+
+  @Test
+  public void shouldRespectServerAuthSkipPathsConfig() {
+    // Given:
+    Mockito.when(server.getWorkerExecutor()).thenReturn(workerExecutor);
+    Mockito.when(server.getConfig()).thenReturn(new KsqlRestConfig(
+        ImmutableMap.of(
+            KsqlRestConfig.AUTHENTICATION_SKIP_PATHS_CONFIG,
+            ImmutableList.of("/heartbeat")
+        )
+    ));
+    Mockito.when(routingContext.normalisedPath()).thenReturn("/heartbeat");
+    KsqlAuthorizationProviderHandler handler = new KsqlAuthorizationProviderHandler(server, authProvider);
+
+    // When:
+    handler.handle(routingContext);
+
+    // Then (make sure the authorization "work" is skipped):
+    Mockito.verify(routingContext).next();
+    Mockito.verifyZeroInteractions(workerExecutor);
+  }
+
+  @Test
+  public void shouldNotSkipNonAuthenticatedPaths() {
+    // Given:
+    Mockito.when(server.getWorkerExecutor()).thenReturn(workerExecutor);
+    Mockito.when(server.getConfig()).thenReturn(new KsqlRestConfig(
+        ImmutableMap.of(
+            KsqlRestConfig.AUTHENTICATION_SKIP_PATHS_CONFIG,
+            ImmutableList.of("/heartbeat")
+        )
+    ));
+    Mockito.when(routingContext.normalisedPath()).thenReturn("/foo");
+    KsqlAuthorizationProviderHandler handler = new KsqlAuthorizationProviderHandler(server, authProvider);
+
+    // When:
+    handler.handle(routingContext);
+
+    // Then (make sure the authorization "work" is not skipped):
+    Mockito.verify(routingContext, Mockito.never()).next();
+    Mockito.verify(workerExecutor).executeBlocking(Mockito.any(), Mockito.any());
+  }
+
+}