Add row-level security filter in query

extensions-contrib/grpc-query/src/main/java/org/apache/druid/grpc/server/QueryDriver.java

@@ -49,6 +49,7 @@
 import org.apache.druid.server.QueryLifecycleFactory;
 import org.apache.druid.server.security.Access;
 import org.apache.druid.server.security.AuthenticationResult;
+import org.apache.druid.server.security.AuthorizationResult;
 import org.apache.druid.server.security.ForbiddenException;
 import org.apache.druid.sql.DirectStatement;
 import org.apache.druid.sql.DirectStatement.ResultSet;
@@ -146,8 +147,8 @@ private QueryResponse runNativeQuery(QueryRequest request, AuthenticationResult
     final String currThreadName = Thread.currentThread().getName();
     try {
       queryLifecycle.initialize(query);
-      Access authorizationResult = queryLifecycle.authorize(authResult);
-      if (!authorizationResult.isAllowed()) {
+      AuthorizationResult authorizationResult = queryLifecycle.authorize(authResult);
+      if (authorizationResult.getPermissionErrorMessage(true).isPresent()) {
         throw new ForbiddenException(Access.DEFAULT_ERROR_MESSAGE);
       }
       queryResponse = queryLifecycle.execute();

extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/BasicSecurityResourceFilter.java

@@ -23,7 +23,7 @@
 import com.sun.jersey.spi.container.ContainerRequest;
 import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.server.http.security.AbstractResourceFilter;
-import org.apache.druid.server.security.Access;
+import org.apache.druid.server.security.AuthorizationResult;
 import org.apache.druid.server.security.AuthorizationUtils;
 import org.apache.druid.server.security.AuthorizerMapper;
 import org.apache.druid.server.security.Resource;
@@ -54,20 +54,20 @@ public ContainerRequest filter(ContainerRequest request)
         getAction(request)
     );
 
-    final Access authResult = AuthorizationUtils.authorizeResourceAction(
+    final AuthorizationResult authResult = AuthorizationUtils.authorizeResourceAction(
         getReq(),
         resourceAction,
         getAuthorizerMapper()
     );
 
-    if (!authResult.isAllowed()) {
+    authResult.getPermissionErrorMessage(true).ifPresent(error -> {
       throw new WebApplicationException(
           Response.status(Response.Status.FORBIDDEN)
                   .type(MediaType.TEXT_PLAIN)
-                  .entity(StringUtils.format("Access-Check-Result: %s", authResult.toString()))
+                  .entity(StringUtils.format("Access-Check-Result: %s", error))
                   .build()
       );
-    }
+    });
 
     return request;
   }

extensions-core/druid-catalog/src/main/java/org/apache/druid/catalog/http/CatalogResource.java

@@ -34,8 +34,8 @@
 import org.apache.druid.java.util.common.IAE;
 import org.apache.druid.java.util.common.Pair;
 import org.apache.druid.java.util.common.StringUtils;
-import org.apache.druid.server.security.Access;
 import org.apache.druid.server.security.Action;
+import org.apache.druid.server.security.AuthorizationResult;
 import org.apache.druid.server.security.AuthorizationUtils;
 import org.apache.druid.server.security.AuthorizerMapper;
 import org.apache.druid.server.security.ForbiddenException;
@@ -56,7 +56,6 @@
 import javax.ws.rs.core.Context;
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
-
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -108,17 +107,17 @@ public CatalogResource(
    * </ul>
    *
    * @param schemaName The name of the Druid schema, which must be writable
-   *        and the user must have at least read access.
-   * @param tableName The name of the table definition to modify. The user must
-   *        have write access to the table.
-   * @param spec The new table definition.
-   * @param version the expected version of an existing table. The version must
-   *        match. If not (or if the table does not exist), returns an error.
-   * @param overwrite if {@code true}, then overwrites any existing table.
-   *        If {@code false}, then the operation fails if the table already exists.
-   *        Ignored if a version is specified.
-   * @param req the HTTP request used for authorization.
-    */
+   *                   and the user must have at least read access.
+   * @param tableName  The name of the table definition to modify. The user must
+   *                   have write access to the table.
+   * @param spec       The new table definition.
+   * @param version    the expected version of an existing table. The version must
+   *                   match. If not (or if the table does not exist), returns an error.
+   * @param overwrite  if {@code true}, then overwrites any existing table.
+   *                   If {@code false}, then the operation fails if the table already exists.
+   *                   Ignored if a version is specified.
+   * @param req        the HTTP request used for authorization.
+   */
   @POST
   @Path("/schemas/{schema}/tables/{name}")
   @Consumes(MediaType.APPLICATION_JSON)
@@ -181,9 +180,9 @@ public Response postTable(
    * the definition is created before the datasource itself.)
    *
    * @param schemaName The Druid schema. The user must have read access.
-   * @param tableName The name of the table within the schema. The user must have
-   *        read access.
-   * @param req the HTTP request used for authorization.
+   * @param tableName  The name of the table within the schema. The user must have
+   *                   read access.
+   * @param req        the HTTP request used for authorization.
    * @return the definition for the table, if any.
    */
   @GET
@@ -211,8 +210,8 @@ public Response getTable(
    * for the given schema and table.
    *
    * @param schemaName The name of the schema that holds the table.
-   * @param tableName The name of the table definition to delete. The user must have
-   *             write access.
+   * @param tableName  The name of the table definition to delete. The user must have
+   *                   write access.
    */
   @DELETE
   @Path("/schemas/{schema}/tables/{name}")
@@ -247,9 +246,9 @@ public Response deleteTable(
    * the table spec changed between the time it was retrieve and the edit operation
    * is submitted.
    *
-   * @param schemaName The name of the schema that holds the table.
-   * @param tableName The name of the table definition to delete. The user must have
-   *             write access.
+   * @param schemaName  The name of the schema that holds the table.
+   * @param tableName   The name of the table definition to delete. The user must have
+   *                    write access.
    * @param editRequest The operation to perform. See the classes for details.
    */
   @POST
@@ -281,7 +280,7 @@ public Response editTable(
    * Retrieves the list of all Druid schema names.
    *
    * @param format the format of the response. See the code for the
-   *        available formats
+   *               available formats
    */
   @GET
   @Path("/schemas")
@@ -318,9 +317,9 @@ public Response getSchemas(
    * the read-only schemas, there will be no table definitions.
    *
    * @param schemaName The name of the Druid schema to query. The user must
-   *        have read access.
-   * @param format the format of the response. See the code for the
-   *        available formats
+   *                   have read access.
+   * @param format     the format of the response. See the code for the
+   *                   available formats
    */
   @GET
   @Path("/schemas/{schema}/tables")
@@ -360,7 +359,7 @@ public Response getSchemaTables(
    * table definitions known to the catalog. Used to prime a cache on first access.
    * After that, the Coordinator will push updates to Brokers. Returns the full
    * list of table details.
-   *
+   * <p>
    * It is expected that the number of table definitions will be of small or moderate
    * size, so no provision is made to handle very large lists.
    */
@@ -467,9 +466,9 @@ private Response listAllTableMetadata(final HttpServletRequest req)
     List<Pair<SchemaSpec, TableMetadata>> tables = new ArrayList<>();
     for (SchemaSpec schema : catalog.schemaRegistry().schemas()) {
       tables.addAll(catalog.tables().tablesInSchema(schema.name())
-          .stream()
-          .map(table -> Pair.of(schema, table))
-          .collect(Collectors.toList()));
+                           .stream()
+                           .map(table -> Pair.of(schema, table))
+                           .collect(Collectors.toList()));
 
     }
     Iterable<Pair<SchemaSpec, TableMetadata>> filtered = AuthorizationUtils.filterAuthorizedResources(
@@ -483,9 +482,9 @@ private Response listAllTableMetadata(final HttpServletRequest req)
     );
 
     List<TableMetadata> metadata = Lists.newArrayList(filtered)
-        .stream()
-        .map(pair -> pair.rhs)
-        .collect(Collectors.toList());
+                                        .stream()
+                                        .map(pair -> pair.rhs)
+                                        .collect(Collectors.toList());
     return Response.ok().entity(metadata).build();
   }
 
@@ -499,9 +498,9 @@ private Response tableNamesInSchema(
         req,
         tables,
         name ->
-          Collections.singletonList(
-              resourceAction(schema, name, Action.READ)),
-          authorizerMapper
+            Collections.singletonList(
+                resourceAction(schema, name, Action.READ)),
+        authorizerMapper
     );
     return Response.ok().entity(Lists.newArrayList(filtered)).build();
   }
@@ -581,13 +580,13 @@ private void authorizeTable(
 
   private void authorize(String resource, String key, Action action, HttpServletRequest request)
   {
-    final Access authResult = authorizeAccess(resource, key, action, request);
-    if (!authResult.isAllowed()) {
-      throw new ForbiddenException(authResult.toString());
-    }
+    final AuthorizationResult authResult = authorizeAccess(resource, key, action, request);
+    authResult.getPermissionErrorMessage(true).ifPresent(error -> {
+      throw new ForbiddenException(error);
+    });
   }
 
-  private Access authorizeAccess(String resource, String key, Action action, HttpServletRequest request)
+  private AuthorizationResult authorizeAccess(String resource, String key, Action action, HttpServletRequest request)
   {
     return AuthorizationUtils.authorizeResourceAction(
         request,

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/dart/controller/http/DartSqlResource.java

@@ -36,9 +36,9 @@
 import org.apache.druid.server.DruidNode;
 import org.apache.druid.server.ResponseContextConfig;
 import org.apache.druid.server.initialization.ServerConfig;
-import org.apache.druid.server.security.Access;
 import org.apache.druid.server.security.Action;
 import org.apache.druid.server.security.AuthenticationResult;
+import org.apache.druid.server.security.AuthorizationResult;
 import org.apache.druid.server.security.AuthorizationUtils;
 import org.apache.druid.server.security.AuthorizerMapper;
 import org.apache.druid.server.security.Resource;
@@ -144,7 +144,7 @@ public GetQueriesResponse doGetRunningQueries(
   )
   {
     final AuthenticationResult authenticationResult = AuthorizationUtils.authenticationResultFromRequest(req);
-    final Access stateReadAccess = AuthorizationUtils.authorizeAllResourceActions(
+    final AuthorizationResult stateReadAccess = AuthorizationUtils.authorizeAllResourceActions(
         authenticationResult,
         Collections.singletonList(new ResourceAction(Resource.STATE_RESOURCE, Action.READ)),
         authorizerMapper
@@ -175,7 +175,7 @@ public GetQueriesResponse doGetRunningQueries(
     queries.sort(Comparator.comparing(DartQueryInfo::getStartTime).thenComparing(DartQueryInfo::getDartQueryId));
 
     final GetQueriesResponse response;
-    if (stateReadAccess.isAllowed()) {
+    if (stateReadAccess.isUserWithNoRestriction()) {
       // User can READ STATE, so they can see all running queries, as well as authentication details.
       response = new GetQueriesResponse(queries);
     } else {
@@ -245,9 +245,9 @@ public Response cancelQuery(
       return Response.status(Response.Status.ACCEPTED).build();
     }
 
-    final Access access = authorizeCancellation(req, cancelables);
+    final AuthorizationResult authResult = authorizeCancellation(req, cancelables);
 
-    if (access.isAllowed()) {
+    if (!authResult.getPermissionErrorMessage(true).isPresent()) {
       sqlLifecycleManager.removeAll(sqlQueryId, cancelables);
 
       // Don't call cancel() on the cancelables. That just cancels native queries, which is useless here. Instead,

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/dart/controller/sql/DartQueryMaker.java

@@ -52,6 +52,7 @@
 import org.apache.druid.msq.sql.MSQTaskQueryMaker;
 import org.apache.druid.segment.column.ColumnType;
 import org.apache.druid.server.QueryResponse;
+import org.apache.druid.server.security.ForbiddenException;
 import org.apache.druid.sql.calcite.planner.PlannerContext;
 import org.apache.druid.sql.calcite.rel.DruidQuery;
 import org.apache.druid.sql.calcite.run.QueryMaker;
@@ -127,6 +128,9 @@ public DartQueryMaker(
   @Override
   public QueryResponse<Object[]> runQuery(DruidQuery druidQuery)
   {
+    plannerContext.getAuthorizationResult().getPermissionErrorMessage(true).ifPresent(error -> {
+      throw new ForbiddenException(error);
+    });
     final MSQSpec querySpec = MSQTaskQueryMaker.makeQuerySpec(
         null,
         druidQuery,

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/rpc/MSQResourceUtils.java

@@ -19,7 +19,7 @@
 
 package org.apache.druid.msq.rpc;
 
-import org.apache.druid.server.security.Access;
+import org.apache.druid.server.security.AuthorizationResult;
 import org.apache.druid.server.security.AuthorizationUtils;
 import org.apache.druid.server.security.AuthorizerMapper;
 import org.apache.druid.server.security.ForbiddenException;
@@ -41,11 +41,15 @@ public static void authorizeAdminRequest(
   {
     final List<ResourceAction> resourceActions = permissionMapper.getAdminPermissions();
 
-    Access access = AuthorizationUtils.authorizeAllResourceActions(request, resourceActions, authorizerMapper);
+    AuthorizationResult authResult = AuthorizationUtils.authorizeAllResourceActions(
+        request,
+        resourceActions,
+        authorizerMapper
+    );
 
-    if (!access.isAllowed()) {
-      throw new ForbiddenException(access.toString());
-    }
+    authResult.getPermissionErrorMessage(true).ifPresent(error -> {
+      throw new ForbiddenException(error);
+    });
   }
 
   public static void authorizeQueryRequest(
@@ -57,10 +61,14 @@ public static void authorizeQueryRequest(
   {
     final List<ResourceAction> resourceActions = permissionMapper.getQueryPermissions(queryId);
 
-    Access access = AuthorizationUtils.authorizeAllResourceActions(request, resourceActions, authorizerMapper);
+    AuthorizationResult authResult = AuthorizationUtils.authorizeAllResourceActions(
+        request,
+        resourceActions,
+        authorizerMapper
+    );
 
-    if (!access.isAllowed()) {
-      throw new ForbiddenException(access.toString());
-    }
+    authResult.getPermissionErrorMessage(true).ifPresent(error -> {
+      throw new ForbiddenException(error);
+    });
   }
 }

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/sql/MSQTaskQueryMaker.java

@@ -54,6 +54,7 @@
 import org.apache.druid.segment.column.ColumnType;
 import org.apache.druid.server.QueryResponse;
 import org.apache.druid.server.lookup.cache.LookupLoadingSpec;
+import org.apache.druid.server.security.ForbiddenException;
 import org.apache.druid.sql.calcite.parser.DruidSqlIngest;
 import org.apache.druid.sql.calcite.parser.DruidSqlInsert;
 import org.apache.druid.sql.calcite.parser.DruidSqlReplace;
@@ -116,6 +117,9 @@ public class MSQTaskQueryMaker implements QueryMaker
   @Override
   public QueryResponse<Object[]> runQuery(final DruidQuery druidQuery)
   {
+    plannerContext.getAuthorizationResult().getPermissionErrorMessage(true).ifPresent(error -> {
+      throw new ForbiddenException(error);
+    });
     Hook.QUERY_PLAN.run(druidQuery.getQuery());
     plannerContext.dispatchHook(DruidHook.NATIVE_PLAN, druidQuery.getQuery());