Adding role based filtering for rest of APIs

src/main/java/com/amazon/opendistroforelasticsearch/ad/transport/AnomalyDetectorJobTransportAction.java

@@ -15,16 +15,18 @@
 
 package com.amazon.opendistroforelasticsearch.ad.transport;
 
+import static com.amazon.opendistroforelasticsearch.ad.settings.AnomalyDetectorSettings.FILTER_BY_BACKEND_ROLES;
 import static com.amazon.opendistroforelasticsearch.ad.settings.AnomalyDetectorSettings.REQUEST_TIMEOUT;
-
-import java.io.IOException;
+import static com.amazon.opendistroforelasticsearch.ad.util.ParseUtils.getUserContext;
+import static com.amazon.opendistroforelasticsearch.ad.util.ParseUtils.resolveUserAndExecute;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.client.Client;
+import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
@@ -35,61 +37,86 @@
 
 import com.amazon.opendistroforelasticsearch.ad.indices.AnomalyDetectionIndices;
 import com.amazon.opendistroforelasticsearch.ad.rest.handler.IndexAnomalyDetectorJobActionHandler;
+import com.amazon.opendistroforelasticsearch.ad.settings.AnomalyDetectorSettings;
 import com.amazon.opendistroforelasticsearch.ad.util.RestHandlerUtils;
+import com.amazon.opendistroforelasticsearch.commons.authuser.User;
 
 public class AnomalyDetectorJobTransportAction extends HandledTransportAction<AnomalyDetectorJobRequest, AnomalyDetectorJobResponse> {
     private final Logger logger = LogManager.getLogger(AnomalyDetectorJobTransportAction.class);
 
     private final Client client;
+    private final ClusterService clusterService;
     private final Settings settings;
     private final AnomalyDetectionIndices anomalyDetectionIndices;
     private final NamedXContentRegistry xContentRegistry;
+    private volatile Boolean filterByEnabled;
 
     @Inject
     public AnomalyDetectorJobTransportAction(
         TransportService transportService,
         ActionFilters actionFilters,
         Client client,
+        ClusterService clusterService,
         Settings settings,
         AnomalyDetectionIndices anomalyDetectionIndices,
         NamedXContentRegistry xContentRegistry
     ) {
         super(AnomalyDetectorJobAction.NAME, transportService, actionFilters, AnomalyDetectorJobRequest::new);
         this.client = client;
+        this.clusterService = clusterService;
         this.settings = settings;
         this.anomalyDetectionIndices = anomalyDetectionIndices;
         this.xContentRegistry = xContentRegistry;
+        filterByEnabled = AnomalyDetectorSettings.FILTER_BY_BACKEND_ROLES.get(settings);
+        clusterService.getClusterSettings().addSettingsUpdateConsumer(FILTER_BY_BACKEND_ROLES, it -> filterByEnabled = it);
     }
 
     @Override
     protected void doExecute(Task task, AnomalyDetectorJobRequest request, ActionListener<AnomalyDetectorJobResponse> listener) {
         String detectorId = request.getDetectorID();
-        long seqNo = request.getSeqNo();
-        long primaryTerm = request.getPrimaryTerm();
-        String rawPath = request.getRawPath();
-        TimeValue requestTimeout = REQUEST_TIMEOUT.get(settings);
-
         // By the time request reaches here, the user permissions are validated by Security plugin.
-        // Since the detectorID is provided, this can only happen if User is part of a role which has access
-        // to the detector. This is filtered by our Search Detector API.
-
+        User user = getUserContext(client);
         try (ThreadContext.StoredContext context = client.threadPool().getThreadContext().stashContext()) {
-            IndexAnomalyDetectorJobActionHandler handler = new IndexAnomalyDetectorJobActionHandler(
-                client,
-                listener,
-                anomalyDetectionIndices,
+            resolveUserAndExecute(
+                user,
                 detectorId,
-                seqNo,
-                primaryTerm,
-                requestTimeout,
+                filterByEnabled,
+                listener,
+                () -> adJobExecute(request, listener),
+                client,
+                clusterService,
                 xContentRegistry
             );
+        } catch (Exception e) {
+            logger.error(e);
+            listener.onFailure(e);
+        }
+    }
+
+    private void adJobExecute(AnomalyDetectorJobRequest request, ActionListener<AnomalyDetectorJobResponse> listener) {
+        String detectorId = request.getDetectorID();
+        long seqNo = request.getSeqNo();
+        long primaryTerm = request.getPrimaryTerm();
+        String rawPath = request.getRawPath();
+        TimeValue requestTimeout = REQUEST_TIMEOUT.get(settings);
+
+        IndexAnomalyDetectorJobActionHandler handler = new IndexAnomalyDetectorJobActionHandler(
+            client,
+            listener,
+            anomalyDetectionIndices,
+            detectorId,
+            seqNo,
+            primaryTerm,
+            requestTimeout,
+            xContentRegistry
+        );
+        try {
             if (rawPath.endsWith(RestHandlerUtils.START_JOB)) {
                 handler.startAnomalyDetectorJob();
             } else if (rawPath.endsWith(RestHandlerUtils.STOP_JOB)) {
                 handler.stopAnomalyDetectorJob(detectorId);
             }
-        } catch (IOException e) {
+        } catch (Exception e) {
             logger.error(e);
             listener.onFailure(e);
         }

src/main/java/com/amazon/opendistroforelasticsearch/ad/transport/DeleteAnomalyDetectorTransportAction.java

@@ -16,6 +16,9 @@
 package com.amazon.opendistroforelasticsearch.ad.transport;
 
 import static com.amazon.opendistroforelasticsearch.ad.model.AnomalyDetectorJob.ANOMALY_DETECTOR_JOB_INDEX;
+import static com.amazon.opendistroforelasticsearch.ad.settings.AnomalyDetectorSettings.FILTER_BY_BACKEND_ROLES;
+import static com.amazon.opendistroforelasticsearch.ad.util.ParseUtils.getUserContext;
+import static com.amazon.opendistroforelasticsearch.ad.util.ParseUtils.resolveUserAndExecute;
 import static org.elasticsearch.common.xcontent.XContentParserUtils.ensureExpectedToken;
 
 import java.io.IOException;
@@ -35,6 +38,7 @@
 import org.elasticsearch.client.Client;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.common.xcontent.XContentParser;
@@ -47,39 +51,52 @@
 import com.amazon.opendistroforelasticsearch.ad.model.AnomalyDetectorJob;
 import com.amazon.opendistroforelasticsearch.ad.model.DetectorInternalState;
 import com.amazon.opendistroforelasticsearch.ad.rest.handler.AnomalyDetectorFunction;
+import com.amazon.opendistroforelasticsearch.ad.settings.AnomalyDetectorSettings;
 import com.amazon.opendistroforelasticsearch.ad.util.RestHandlerUtils;
+import com.amazon.opendistroforelasticsearch.commons.authuser.User;
 
 public class DeleteAnomalyDetectorTransportAction extends HandledTransportAction<DeleteAnomalyDetectorRequest, DeleteResponse> {
 
     private static final Logger LOG = LogManager.getLogger(DeleteAnomalyDetectorTransportAction.class);
     private final Client client;
     private final ClusterService clusterService;
     private NamedXContentRegistry xContentRegistry;
+    private volatile Boolean filterByEnabled;
 
     @Inject
     public DeleteAnomalyDetectorTransportAction(
         TransportService transportService,
         ActionFilters actionFilters,
         Client client,
         ClusterService clusterService,
+        Settings settings,
         NamedXContentRegistry xContentRegistry
     ) {
         super(DeleteAnomalyDetectorAction.NAME, transportService, actionFilters, DeleteAnomalyDetectorRequest::new);
         this.client = client;
         this.clusterService = clusterService;
         this.xContentRegistry = xContentRegistry;
+        filterByEnabled = AnomalyDetectorSettings.FILTER_BY_BACKEND_ROLES.get(settings);
+        clusterService.getClusterSettings().addSettingsUpdateConsumer(FILTER_BY_BACKEND_ROLES, it -> filterByEnabled = it);
     }
 
     @Override
     protected void doExecute(Task task, DeleteAnomalyDetectorRequest request, ActionListener<DeleteResponse> listener) {
         String detectorId = request.getDetectorID();
         LOG.info("Delete anomaly detector job {}", detectorId);
-
+        User user = getUserContext(client);
         // By the time request reaches here, the user permissions are validated by Security plugin.
-        // Since the detectorID is provided, this can only happen if User is part of a role which has access
-        // to the detector. This is filtered by our Search Detector API.
         try (ThreadContext.StoredContext context = client.threadPool().getThreadContext().stashContext()) {
-            getDetectorJob(detectorId, listener, () -> deleteAnomalyDetectorJobDoc(detectorId, listener));
+            resolveUserAndExecute(
+                user,
+                detectorId,
+                filterByEnabled,
+                listener,
+                () -> getDetectorJob(detectorId, listener, () -> deleteAnomalyDetectorJobDoc(detectorId, listener)),
+                client,
+                clusterService,
+                xContentRegistry
+            );
         } catch (Exception e) {
             LOG.error(e);
             listener.onFailure(e);

src/main/java/com/amazon/opendistroforelasticsearch/ad/transport/GetAnomalyDetectorTransportAction.java

@@ -17,6 +17,9 @@
 
 import static com.amazon.opendistroforelasticsearch.ad.model.AnomalyDetector.ANOMALY_DETECTORS_INDEX;
 import static com.amazon.opendistroforelasticsearch.ad.model.AnomalyDetectorJob.ANOMALY_DETECTOR_JOB_INDEX;
+import static com.amazon.opendistroforelasticsearch.ad.settings.AnomalyDetectorSettings.FILTER_BY_BACKEND_ROLES;
+import static com.amazon.opendistroforelasticsearch.ad.util.ParseUtils.getUserContext;
+import static com.amazon.opendistroforelasticsearch.ad.util.ParseUtils.resolveUserAndExecute;
 import static com.amazon.opendistroforelasticsearch.ad.util.RestHandlerUtils.PROFILE;
 import static org.elasticsearch.common.xcontent.XContentParserUtils.ensureExpectedToken;
 
@@ -37,9 +40,11 @@
 import org.elasticsearch.action.support.ActionFilters;
 import org.elasticsearch.action.support.HandledTransportAction;
 import org.elasticsearch.client.Client;
+import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.CheckedConsumer;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.common.xcontent.XContentParser;
@@ -58,12 +63,14 @@
 import com.amazon.opendistroforelasticsearch.ad.settings.AnomalyDetectorSettings;
 import com.amazon.opendistroforelasticsearch.ad.util.DiscoveryNodeFilterer;
 import com.amazon.opendistroforelasticsearch.ad.util.RestHandlerUtils;
+import com.amazon.opendistroforelasticsearch.commons.authuser.User;
 import com.google.common.collect.Sets;
 
 public class GetAnomalyDetectorTransportAction extends HandledTransportAction<GetAnomalyDetectorRequest, GetAnomalyDetectorResponse> {
 
     private static final Logger LOG = LogManager.getLogger(GetAnomalyDetectorTransportAction.class);
 
+    private final ClusterService clusterService;
     private final Client client;
 
     private final Set<String> allProfileTypeStrs;
@@ -74,16 +81,20 @@ public class GetAnomalyDetectorTransportAction extends HandledTransportAction<Ge
     private final Set<EntityProfileName> defaultEntityProfileTypes;
     private final NamedXContentRegistry xContentRegistry;
     private final DiscoveryNodeFilterer nodeFilter;
+    private volatile Boolean filterByEnabled;
 
     @Inject
     public GetAnomalyDetectorTransportAction(
         TransportService transportService,
         DiscoveryNodeFilterer nodeFilter,
         ActionFilters actionFilters,
+        ClusterService clusterService,
         Client client,
+        Settings settings,
         NamedXContentRegistry xContentRegistry
     ) {
         super(GetAnomalyDetectorAction.NAME, transportService, actionFilters, GetAnomalyDetectorRequest::new);
+        this.clusterService = clusterService;
         this.client = client;
 
         List<DetectorProfileName> allProfiles = Arrays.asList(DetectorProfileName.values());
@@ -100,10 +111,32 @@ public GetAnomalyDetectorTransportAction(
 
         this.xContentRegistry = xContentRegistry;
         this.nodeFilter = nodeFilter;
+        filterByEnabled = AnomalyDetectorSettings.FILTER_BY_BACKEND_ROLES.get(settings);
+        clusterService.getClusterSettings().addSettingsUpdateConsumer(FILTER_BY_BACKEND_ROLES, it -> filterByEnabled = it);
     }
 
     @Override
     protected void doExecute(Task task, GetAnomalyDetectorRequest request, ActionListener<GetAnomalyDetectorResponse> listener) {
+        String detectorID = request.getDetectorID();
+        User user = getUserContext(client);
+        try (ThreadContext.StoredContext context = client.threadPool().getThreadContext().stashContext()) {
+            resolveUserAndExecute(
+                user,
+                detectorID,
+                filterByEnabled,
+                listener,
+                () -> getExecute(request, listener),
+                client,
+                clusterService,
+                xContentRegistry
+            );
+        } catch (Exception e) {
+            LOG.error(e);
+            listener.onFailure(e);
+        }
+    }
+
+    protected void getExecute(GetAnomalyDetectorRequest request, ActionListener<GetAnomalyDetectorResponse> listener) {
         String detectorID = request.getDetectorID();
         Long version = request.getVersion();
         String typesStr = request.getTypeStr();
@@ -112,7 +145,7 @@ protected void doExecute(Task task, GetAnomalyDetectorRequest request, ActionLis
         boolean all = request.isAll();
         boolean returnJob = request.isReturnJob();
 
-        try (ThreadContext.StoredContext context = client.threadPool().getThreadContext().stashContext()) {
+        try {
             if (!Strings.isEmpty(typesStr) || rawPath.endsWith(PROFILE) || rawPath.endsWith(PROFILE + "/")) {
                 if (entityValue != null) {
                     Set<EntityProfileName> entityProfilesToCollect = getEntityProfilesToCollect(typesStr, all);