Basic authorization support in Druid (#2424)

- Introduce `AuthorizationInfo` interface, specific implementations of which would be provided by extensions
- If the `druid.auth.enabled` is set to `true` then the `isAuthorized` method of `AuthorizationInfo` will be called to perform authorization checks
-  `AuthorizationInfo` object will be created in the servlet filters of specific extension and will be passed as a request attribute with attribute name as `AuthConfig.DRUID_AUTH_TOKEN`
- As per the scope of this PR, all resources that needs to be secured are divided into 3 types - `DATASOURCE`, `CONFIG` and `STATE`. For any type of resource, possible actions are  - `READ` or `WRITE`
- Specific ResourceFilters are used to perform auth checks for all endpoints that corresponds to a specific resource type. This prevents duplication of logic and need to inject HttpServletRequest inside each endpoint. For example
 - `DatasourceResourceFilter` is used for endpoints where the datasource information is present after "datasources" segment in the request Path such as `/druid/coordinator/v1/datasources/`, `/druid/coordinator/v1/metadata/datasources/`, `/druid/v2/datasources/`
 - `RulesResourceFilter` is used where the datasource information is present after "rules" segment in the request Path such as `/druid/coordinator/v1/rules/`
 - `TaskResourceFilter` is used for endpoints is used where the datasource information is present after "task" segment in the request Path such as `druid/indexer/v1/task`
 - `ConfigResourceFilter` is used for endpoints like `/druid/coordinator/v1/config`, `/druid/indexer/v1/worker`, `/druid/worker/v1` etc
 - `StateResourceFilter` is used for endpoints like `/druid/broker/v1/loadstatus`, `/druid/coordinator/v1/leader`, `/druid/coordinator/v1/loadqueue`, `/druid/coordinator/v1/rules` etc
- For endpoints where a list of resources is returned like `/druid/coordinator/v1/datasources`, `/druid/indexer/v1/completeTasks` etc. the list is filtered to return only the resources to which the requested user has access. In these cases, `HttpServletRequest` instance needs to be injected in the endpoint method.

Note -
JAX-RS specification provides an interface called `SecurityContext`. However, we did not use this but provided our own interface `AuthorizationInfo` mainly because it provides more flexibility. For example, `SecurityContext` has a method called `isUserInRole(String role)` which would be used for auth checks and if used then the mapping of what roles can access what resource needs to be modeled inside Druid either using some convention or some other means which is not very flexible as Druid has dynamic resources like datasources. Fixes #2355 with PR #2424

indexing-service/src/main/java/io/druid/indexing/overlord/http/security/TaskResourceFilter.java

@@ -0,0 +1,123 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets 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 io.druid.indexing.overlord.http.security;
+
+import com.google.common.base.Optional;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Predicate;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import com.google.inject.Inject;
+import com.sun.jersey.spi.container.ContainerRequest;
+import io.druid.indexing.common.task.Task;
+import io.druid.indexing.overlord.TaskStorageQueryAdapter;
+import io.druid.server.http.security.AbstractResourceFilter;
+import io.druid.server.security.Access;
+import io.druid.server.security.AuthConfig;
+import io.druid.server.security.AuthorizationInfo;
+import io.druid.server.security.Resource;
+import io.druid.server.security.ResourceType;
+
+import javax.ws.rs.WebApplicationException;
+import javax.ws.rs.core.PathSegment;
+import javax.ws.rs.core.Response;
+import java.util.List;
+
+/**
+ * Use this ResourceFilter when the datasource information is present after "task" segment in the request Path
+ * Here are some example paths where this filter is used -
+ * - druid/indexer/v1/task/{taskid}/...
+ * Note - DO NOT use this filter at MiddleManager resources as TaskStorageQueryAdapter cannot be injected there
+ */
+public class TaskResourceFilter extends AbstractResourceFilter
+{
+  private final TaskStorageQueryAdapter taskStorageQueryAdapter;
+
+  @Inject
+  public TaskResourceFilter(TaskStorageQueryAdapter taskStorageQueryAdapter, AuthConfig authConfig) {
+    super(authConfig);
+    this.taskStorageQueryAdapter = taskStorageQueryAdapter;
+  }
+
+  @Override
+  public ContainerRequest filter(ContainerRequest request)
+  {
+    if (getAuthConfig().isEnabled()) {
+      // This is an experimental feature, see - https://github.com/druid-io/druid/pull/2424
+      final String taskId = Preconditions.checkNotNull(
+          request.getPathSegments()
+                 .get(
+                     Iterables.indexOf(
+                         request.getPathSegments(),
+                         new Predicate<PathSegment>()
+                         {
+                           @Override
+                           public boolean apply(PathSegment input)
+                           {
+                             return input.getPath().equals("task");
+                           }
+                         }
+                     ) + 1
+                 ).getPath()
+      );
+
+      Optional<Task> taskOptional = taskStorageQueryAdapter.getTask(taskId);
+      if (!taskOptional.isPresent()) {
+        throw new WebApplicationException(
+            Response.status(Response.Status.BAD_REQUEST)
+                    .entity(String.format("Cannot find any task with id: [%s]", taskId))
+                    .build()
+        );
+      }
+      final String dataSourceName = Preconditions.checkNotNull(taskOptional.get().getDataSource());
+
+      final AuthorizationInfo authorizationInfo = (AuthorizationInfo) getReq().getAttribute(AuthConfig.DRUID_AUTH_TOKEN);
+      Preconditions.checkNotNull(
+          authorizationInfo,
+          "Security is enabled but no authorization info found in the request"
+      );
+      final Access authResult = authorizationInfo.isAuthorized(
+          new Resource(dataSourceName, ResourceType.DATASOURCE),
+          getAction(request)
+      );
+      if (!authResult.isAllowed()) {
+        throw new WebApplicationException(Response.status(Response.Status.FORBIDDEN)
+                                                  .entity(
+                                                      String.format("Access-Check-Result: %s", authResult.toString())
+                                                  )
+                                                  .build());
+      }
+    }
+
+    return request;
+  }
+
+  @Override
+  public boolean isApplicable(String requestPath)
+  {
+    List<String> applicablePaths = ImmutableList.of("druid/indexer/v1/task/");
+    for (String path : applicablePaths) {
+      if(requestPath.startsWith(path) && !requestPath.equals(path)) {
+        return true;
+      }
+    }
+    return false;
+  }
+}

indexing-service/src/main/java/io/druid/indexing/worker/http/WorkerResource.java

@@ -27,10 +27,13 @@
 import com.google.common.io.ByteSource;
 import com.google.inject.Inject;
 import com.metamx.common.logger.Logger;
+import com.sun.jersey.spi.container.ResourceFilters;
 import io.druid.indexing.overlord.TaskRunner;
 import io.druid.indexing.overlord.TaskRunnerWorkItem;
 import io.druid.indexing.worker.Worker;
 import io.druid.indexing.worker.WorkerCuratorCoordinator;
+import io.druid.server.http.security.ConfigResourceFilter;
+import io.druid.server.http.security.StateResourceFilter;
 import io.druid.tasklogs.TaskLogStreamer;
 
 import javax.ws.rs.DefaultValue;
@@ -73,6 +76,7 @@ public WorkerResource(
   @POST
   @Path("/disable")
   @Produces(MediaType.APPLICATION_JSON)
+  @ResourceFilters(ConfigResourceFilter.class)
   public Response doDisable()
   {
     try {
@@ -93,6 +97,7 @@ public Response doDisable()
   @POST
   @Path("/enable")
   @Produces(MediaType.APPLICATION_JSON)
+  @ResourceFilters(ConfigResourceFilter.class)
   public Response doEnable()
   {
     try {
@@ -107,6 +112,7 @@ public Response doEnable()
   @GET
   @Path("/enabled")
   @Produces(MediaType.APPLICATION_JSON)
+  @ResourceFilters(StateResourceFilter.class)
   public Response isEnabled()
   {
     try {
@@ -122,6 +128,7 @@ public Response isEnabled()
   @GET
   @Path("/tasks")
   @Produces(MediaType.APPLICATION_JSON)
+  @ResourceFilters(StateResourceFilter.class)
   public Response getTasks()
   {
     try {
@@ -149,6 +156,7 @@ public String apply(TaskRunnerWorkItem input)
   @POST
   @Path("/task/{taskid}/shutdown")
   @Produces(MediaType.APPLICATION_JSON)
+  @ResourceFilters(StateResourceFilter.class)
   public Response doShutdown(@PathParam("taskid") String taskid)
   {
     try {
@@ -164,6 +172,7 @@ public Response doShutdown(@PathParam("taskid") String taskid)
   @GET
   @Path("/task/{taskid}/log")
   @Produces("text/plain")
+  @ResourceFilters(StateResourceFilter.class)
   public Response doGetLog(
       @PathParam("taskid") String taskid,
       @QueryParam("offset") @DefaultValue("0") long offset