Fix authorization check in supervisor history API (#5460)

indexing-service/src/main/java/io/druid/indexing/overlord/supervisor/SupervisorResource.java

@@ -235,7 +235,7 @@ public class SupervisorResource
                   }
                 }
             );
-            return Response.ok(supervisorHistory).build();
+            return Response.ok(authorizedSupervisorHistory).build();
           }
         }
     );

indexing-service/src/test/java/io/druid/indexing/overlord/supervisor/SupervisorResourceTest.java

@@ -28,9 +28,13 @@ import com.google.common.collect.Maps;
 import io.druid.indexing.overlord.DataSourceMetadata;
 import io.druid.indexing.overlord.TaskMaster;
 import io.druid.java.util.common.DateTimes;
+import io.druid.server.security.Access;
+import io.druid.server.security.Action;
 import io.druid.server.security.AuthConfig;
-import io.druid.server.security.AuthTestUtils;
 import io.druid.server.security.AuthenticationResult;
+import io.druid.server.security.Authorizer;
+import io.druid.server.security.AuthorizerMapper;
+import io.druid.server.security.Resource;
 import org.easymock.Capture;
 import org.easymock.EasyMock;
 import org.easymock.EasyMockRunner;
@@ -64,7 +68,34 @@ public class SupervisorResourceTest extends EasyMockSupport
   @Before
   public void setUp() throws Exception
   {
-    supervisorResource = new SupervisorResource(taskMaster, new AuthConfig(), AuthTestUtils.TEST_AUTHORIZER_MAPPER);
+    supervisorResource = new SupervisorResource(
+        taskMaster,
+        new AuthConfig(),
+        new AuthorizerMapper(null) {
+          @Override
+          public Authorizer getAuthorizer(String name)
+          {
+            return new Authorizer()
+            {
+              @Override
+              public Access authorize(
+                  AuthenticationResult authenticationResult, Resource resource, Action action
+              )
+              {
+                if (authenticationResult.getIdentity().equals("druid")) {
+                  return Access.OK;
+                } else {
+                  if (resource.getName().equals("datasource2")) {
+                    return new Access(false, "not authorized.");
+                  } else {
+                    return Access.OK;
+                  }
+                }
+              }
+            };
+          }
+        }
+    );
   }
 
   @Test
@@ -303,6 +334,60 @@ public class SupervisorResourceTest extends EasyMockSupport
     Assert.assertEquals(503, response.getStatus());
   }
 
+  @Test
+  public void testSpecGetAllHistoryWithAuthFailureFiltering() throws Exception
+  {
+    Map<String, List<VersionedSupervisorSpec>> history = Maps.newHashMap();
+    history.put("id1", null);
+    history.put("id2", null);
+
+    EasyMock.expect(taskMaster.getSupervisorManager()).andReturn(Optional.of(supervisorManager)).times(2);
+    EasyMock.expect(supervisorManager.getSupervisorHistory()).andReturn(history);
+    SupervisorSpec spec1 = new TestSupervisorSpec("id1", null) {
+
+      @Override
+      public List<String> getDataSources()
+      {
+        return Lists.newArrayList("datasource1");
+      }
+    };
+    SupervisorSpec spec2 = new TestSupervisorSpec("id2", null) {
+
+      @Override
+      public List<String> getDataSources()
+      {
+        return Lists.newArrayList("datasource2");
+      }
+    };
+    EasyMock.expect(supervisorManager.getSupervisorSpec("id1")).andReturn(Optional.of(spec1)).atLeastOnce();
+    EasyMock.expect(supervisorManager.getSupervisorSpec("id2")).andReturn(Optional.of(spec2)).atLeastOnce();
+    EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED)).andReturn(null).atLeastOnce();
+    EasyMock.expect(request.getAttribute(AuthConfig.DRUID_AUTHENTICATION_RESULT)).andReturn(
+        new AuthenticationResult("wronguser", "druid", null)
+    ).atLeastOnce();
+    request.setAttribute(AuthConfig.DRUID_AUTHORIZATION_CHECKED, true);
+    EasyMock.expectLastCall().anyTimes();
+    replayAll();
+
+    Response response = supervisorResource.specGetAllHistory(request);
+
+    Map<String, List<VersionedSupervisorSpec>> filteredHistory = Maps.newHashMap();
+    filteredHistory.put("id1", null);
+
+    Assert.assertEquals(200, response.getStatus());
+    Assert.assertEquals(filteredHistory, response.getEntity());
+
+    resetAll();
+
+    EasyMock.expect(taskMaster.getSupervisorManager()).andReturn(Optional.<SupervisorManager>absent());
+    replayAll();
+
+    response = supervisorResource.specGetAllHistory(request);
+    verifyAll();
+
+    Assert.assertEquals(503, response.getStatus());
+  }
+
   @Test
   public void testSpecGetHistory() throws Exception
   {