Onboards to centralized resource access control for detectors and forecasters

.github/workflows/test_security.yml

@@ -24,7 +24,7 @@ jobs:
     strategy:
       matrix:
         # empty string = disabled, flag = enabled
-        resource_sharing_flag: ["", "-Dresource_sharing.enabled=true"]
+        resource_sharing_flag: [""]
 
     steps:
       - name: Run start commands

CHANGELOG.md

@@ -7,6 +7,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.1.0/)
 ### Features
 ### Enhancements
 - Support >1 hr intervals ([#1513](https://github.com/opensearch-project/anomaly-detection/pull/1513))
+- Onboards to centralized resource access control for detectors and forecasters ([#1533](https://github.com/opensearch-project/anomaly-detection/pull/1533))
 
 
 ### Bug Fixes

build-tools/coverage.gradle

@@ -6,7 +6,7 @@
 apply plugin: 'jacoco'
 
 jacoco {
-    toolVersion = "0.8.12"
+    toolVersion = "0.8.13"
 }
 
 /**

build.gradle

@@ -163,8 +163,8 @@ dependencies {
     implementation group: 'org.apache.commons', name: 'commons-lang3', version: '3.18.0'
 
 
-    implementation "org.jacoco:org.jacoco.agent:0.8.12"
-    implementation ("org.jacoco:org.jacoco.ant:0.8.12") {
+    implementation "org.jacoco:org.jacoco.agent:0.8.13"
+    implementation ("org.jacoco:org.jacoco.ant:0.8.13") {
         exclude group: 'org.ow2.asm', module: 'asm-commons'
         exclude group: 'org.ow2.asm', module: 'asm'
         exclude group: 'org.ow2.asm', module: 'asm-tree'
@@ -213,7 +213,7 @@ allprojects {
     version = "${opensearch_build}"
 
     plugins.withId('jacoco') {
-        jacoco.toolVersion = '0.8.12'
+        jacoco.toolVersion = '0.8.13'
     }
 }
 
@@ -375,6 +375,7 @@ integTest {
     systemProperty "https", System.getProperty("https")
     systemProperty "user", System.getProperty("user")
     systemProperty "password", System.getProperty("password")
+    systemProperty "resource_sharing.enabled", System.getProperty("resource_sharing.enabled")
 
     // Only rest case can run with remote cluster
     if (System.getProperty("tests.rest.cluster") != null) {

src/main/java/org/opensearch/ad/rest/RestAnomalyDetectorJobAction.java

@@ -21,6 +21,7 @@
 import java.util.Locale;
 
 import org.opensearch.ad.constant.ADCommonMessages;
+import org.opensearch.ad.indices.ADIndex;
 import org.opensearch.ad.settings.ADEnabledSetting;
 import org.opensearch.ad.transport.AnomalyDetectorJobAction;
 import org.opensearch.cluster.service.ClusterService;
@@ -65,7 +66,13 @@ protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient cli
         String rawPath = request.rawPath();
         DateRange detectionDateRange = parseInputDateRange(request);
 
-        JobRequest anomalyDetectorJobRequest = new JobRequest(detectorId, detectionDateRange, historical, rawPath);
+        JobRequest anomalyDetectorJobRequest = new JobRequest(
+            detectorId,
+            ADIndex.CONFIG.getIndexName(),
+            detectionDateRange,
+            historical,
+            rawPath
+        );
 
         return channel -> client
             .execute(AnomalyDetectorJobAction.INSTANCE, anomalyDetectorJobRequest, new RestToXContentListener<>(channel));

src/main/java/org/opensearch/ad/rest/RestDeleteAnomalyDetectorAction.java

@@ -18,6 +18,7 @@
 import java.util.Locale;
 
 import org.opensearch.ad.constant.ADCommonMessages;
+import org.opensearch.ad.indices.ADIndex;
 import org.opensearch.ad.settings.ADEnabledSetting;
 import org.opensearch.ad.transport.DeleteAnomalyDetectorAction;
 import org.opensearch.rest.BaseRestHandler;
@@ -50,7 +51,7 @@ protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient cli
         }
 
         String detectorId = request.param(DETECTOR_ID);
-        DeleteConfigRequest deleteAnomalyDetectorRequest = new DeleteConfigRequest(detectorId);
+        DeleteConfigRequest deleteAnomalyDetectorRequest = new DeleteConfigRequest(detectorId, ADIndex.CONFIG.getIndexName());
         return channel -> client
             .execute(DeleteAnomalyDetectorAction.INSTANCE, deleteAnomalyDetectorRequest, new RestToXContentListener<>(channel));
     }

src/main/java/org/opensearch/ad/rest/RestGetAnomalyDetectorAction.java

@@ -22,6 +22,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.opensearch.ad.constant.ADCommonMessages;
+import org.opensearch.ad.indices.ADIndex;
 import org.opensearch.ad.settings.ADEnabledSetting;
 import org.opensearch.ad.transport.GetAnomalyDetectorAction;
 import org.opensearch.rest.BaseRestHandler;
@@ -64,6 +65,7 @@ protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient cli
         boolean all = request.paramAsBoolean("_all", false);
         GetConfigRequest getConfigRequest = new GetConfigRequest(
             detectorId,
+            ADIndex.CONFIG.getIndexName(),
             RestActions.parseVersion(request),
             returnJob,
             returnTask,

src/main/java/org/opensearch/ad/rest/handler/IndexAnomalyDetectorActionHandler.java

@@ -37,27 +37,27 @@ public class IndexAnomalyDetectorActionHandler extends AbstractAnomalyDetectorAc
     /**
      * Constructor function.
      *
-     * @param clusterService          ClusterService
-     * @param client                  ES node client that executes actions on the local node
-     * @param clientUtil              AD client util
-     * @param transportService        ES transport service
-     * @param anomalyDetectionIndices anomaly detector index manager
-     * @param detectorId              detector identifier
-     * @param seqNo                   sequence number of last modification
-     * @param primaryTerm             primary term of last modification
-     * @param refreshPolicy           refresh policy
-     * @param anomalyDetector         anomaly detector instance
-     * @param requestTimeout          request time out configuration
+     * @param clusterService           ClusterService
+     * @param client                   ES node client that executes actions on the local node
+     * @param clientUtil               AD client util
+     * @param transportService         ES transport service
+     * @param anomalyDetectionIndices  anomaly detector index manager
+     * @param detectorId               detector identifier
+     * @param seqNo                    sequence number of last modification
+     * @param primaryTerm              primary term of last modification
+     * @param refreshPolicy            refresh policy
+     * @param anomalyDetector          anomaly detector instance
+     * @param requestTimeout           request time out configuration
      * @param maxSingleStreamDetectors max single-stream anomaly detectors allowed
-     * @param maxHCDetectors          max HC detectors allowed
-     * @param maxFeatures             max features allowed per detector
-     * @param maxCategoricalFields    max number of categorical fields
-     * @param method                  Rest Method type
-     * @param xContentRegistry        Registry which is used for XContentParser
-     * @param user                    User context
-     * @param adTaskManager           AD Task manager
-     * @param searchFeatureDao        Search feature dao
-     * @param settings                Node settings
+     * @param maxHCDetectors           max HC detectors allowed
+     * @param maxFeatures              max features allowed per detector
+     * @param maxCategoricalFields     max number of categorical fields
+     * @param method                   Rest Method type
+     * @param xContentRegistry         Registry which is used for XContentParser
+     * @param user                     User context
+     * @param adTaskManager            AD Task manager
+     * @param searchFeatureDao         Search feature dao
+     * @param settings                 Node settings
      */
     public IndexAnomalyDetectorActionHandler(
         ClusterService clusterService,

src/main/java/org/opensearch/ad/transport/AnomalyDetectorJobTransportAction.java

@@ -63,7 +63,6 @@ public AnomalyDetectorJobTransportAction(
             FAIL_TO_STOP_DETECTOR,
             AnomalyDetector.class,
             adIndexJobActionHandler,
-            ADIndex.CONFIG.getIndexName(),
             Clock.systemUTC() // inject cannot find clock due to OS limitation
         );
     }

src/main/java/org/opensearch/ad/transport/AnomalyResultRequest.java

@@ -22,6 +22,7 @@
 import org.opensearch.action.ActionRequestValidationException;
 import org.opensearch.ad.constant.ADCommonMessages;
 import org.opensearch.ad.constant.ADCommonName;
+import org.opensearch.ad.indices.ADIndex;
 import org.opensearch.core.common.Strings;
 import org.opensearch.core.common.io.stream.InputStreamStreamInput;
 import org.opensearch.core.common.io.stream.OutputStreamStreamOutput;
@@ -37,7 +38,7 @@ public AnomalyResultRequest(StreamInput in) throws IOException {
     }
 
     public AnomalyResultRequest(String adID, long start, long end) {
-        super(adID, start, end);
+        super(adID, ADIndex.CONFIG.getIndexName(), start, end);
     }
 
     @Override

src/main/java/org/opensearch/ad/transport/IndexAnomalyDetectorRequest.java

@@ -144,4 +144,5 @@ public void writeTo(StreamOutput out) throws IOException {
     public ActionRequestValidationException validate() {
         return null;
     }
+
 }

src/main/java/org/opensearch/ad/transport/IndexAnomalyDetectorTransportAction.java

@@ -16,7 +16,6 @@
 import static org.opensearch.ad.settings.AnomalyDetectorSettings.AD_FILTER_BY_BACKEND_ROLES;
 import static org.opensearch.timeseries.util.ParseUtils.checkFilterByBackendRoles;
 import static org.opensearch.timeseries.util.ParseUtils.getConfig;
-import static org.opensearch.timeseries.util.ParseUtils.shouldUseResourceAuthz;
 import static org.opensearch.timeseries.util.ParseUtils.verifyResourceAccessAndProcessRequest;
 import static org.opensearch.timeseries.util.RestHandlerUtils.wrapRestActionListener;
 
@@ -29,7 +28,6 @@
 import org.opensearch.action.support.ActionFilters;
 import org.opensearch.action.support.HandledTransportAction;
 import org.opensearch.action.support.WriteRequest;
-import org.opensearch.ad.indices.ADIndex;
 import org.opensearch.ad.indices.ADIndexManagement;
 import org.opensearch.ad.model.AnomalyDetector;
 import org.opensearch.ad.rest.handler.IndexAnomalyDetectorActionHandler;
@@ -103,16 +101,9 @@ protected void doExecute(Task task, IndexAnomalyDetectorRequest request, ActionL
         String errorMessage = method == RestRequest.Method.PUT ? FAIL_TO_UPDATE_DETECTOR : FAIL_TO_CREATE_DETECTOR;
         ActionListener<IndexAnomalyDetectorResponse> listener = wrapRestActionListener(actionListener, errorMessage);
 
-        // TODO: Remove following and any other conditional check, post GA for Resource Authz.
-        boolean shouldEvaluateWithNewAuthz = shouldUseResourceAuthz(settings);
-
         try (ThreadContext.StoredContext context = client.threadPool().getThreadContext().stashContext()) {
             verifyResourceAccessAndProcessRequest(
-                user,
-                ADIndex.CONFIG.getIndexName(),
-                detectorId,
-                shouldEvaluateWithNewAuthz,
-                listener,
+                settings,
                 args -> indexDetector(
                     user,
                     detectorId,

src/main/java/org/opensearch/ad/transport/PreviewAnomalyDetectorRequest.java

@@ -16,11 +16,13 @@
 
 import org.opensearch.action.ActionRequest;
 import org.opensearch.action.ActionRequestValidationException;
+import org.opensearch.action.DocRequest;
+import org.opensearch.ad.indices.ADIndex;
 import org.opensearch.ad.model.AnomalyDetector;
 import org.opensearch.core.common.io.stream.StreamInput;
 import org.opensearch.core.common.io.stream.StreamOutput;
 
-public class PreviewAnomalyDetectorRequest extends ActionRequest {
+public class PreviewAnomalyDetectorRequest extends ActionRequest implements DocRequest {
 
     private AnomalyDetector detector;
     private String detectorId;
@@ -73,4 +75,14 @@ public void writeTo(StreamOutput out) throws IOException {
     public ActionRequestValidationException validate() {
         return null;
     }
+
+    @Override
+    public String index() {
+        return ADIndex.CONFIG.getIndexName();
+    }
+
+    @Override
+    public String id() {
+        return detectorId;
+    }
 }

src/main/java/org/opensearch/ad/transport/PreviewAnomalyDetectorTransportAction.java

@@ -17,7 +17,6 @@
 import static org.opensearch.ad.settings.AnomalyDetectorSettings.MAX_CONCURRENT_PREVIEW;
 import static org.opensearch.core.xcontent.XContentParserUtils.ensureExpectedToken;
 import static org.opensearch.timeseries.util.ParseUtils.resolveUserAndExecute;
-import static org.opensearch.timeseries.util.ParseUtils.shouldUseResourceAuthz;
 import static org.opensearch.timeseries.util.ParseUtils.verifyResourceAccessAndProcessRequest;
 import static org.opensearch.timeseries.util.RestHandlerUtils.wrapRestActionListener;
 
@@ -37,7 +36,6 @@
 import org.opensearch.ad.AnomalyDetectorRunner;
 import org.opensearch.ad.constant.ADCommonMessages;
 import org.opensearch.ad.constant.ADCommonName;
-import org.opensearch.ad.indices.ADIndex;
 import org.opensearch.ad.model.AnomalyDetector;
 import org.opensearch.ad.model.AnomalyResult;
 import org.opensearch.ad.settings.AnomalyDetectorSettings;
@@ -111,17 +109,10 @@ protected void doExecute(
         User user = ParseUtils.getUserContext(client);
         ActionListener<PreviewAnomalyDetectorResponse> listener = wrapRestActionListener(actionListener, FAIL_TO_PREVIEW_DETECTOR);
 
-        // TODO: Remove following and any other conditional check, post GA for Resource Authz.
-        boolean shouldEvaluateWithNewAuthz = shouldUseResourceAuthz(settings);
-
         try (ThreadContext.StoredContext context = client.threadPool().getThreadContext().stashContext()) {
             // Call the verifyResourceAccessAndProcessRequest method
             verifyResourceAccessAndProcessRequest(
-                user,
-                ADIndex.CONFIG.getIndexName(),
-                detectorId,
-                shouldEvaluateWithNewAuthz,
-                listener,
+                settings,
                 args -> previewExecute(request, context, listener),
                 new Object[] {},
                 (fallbackArgs) -> resolveUserAndExecute(

src/main/java/org/opensearch/ad/transport/SearchTopAnomalyResultTransportAction.java

@@ -31,6 +31,7 @@
 import org.opensearch.action.search.SearchResponse;
 import org.opensearch.action.support.ActionFilters;
 import org.opensearch.action.support.HandledTransportAction;
+import org.opensearch.ad.indices.ADIndex;
 import org.opensearch.ad.model.ADTask;
 import org.opensearch.ad.model.AnomalyResult;
 import org.opensearch.ad.model.AnomalyResultBucket;
@@ -220,6 +221,7 @@ protected void doExecute(Task task, SearchTopAnomalyResultRequest request, Actio
 
         GetConfigRequest getAdRequest = new GetConfigRequest(
             request.getId(),
+            ADIndex.CONFIG.getIndexName(),
             // The default version value used in org.opensearch.rest.action.RestActions.parseVersion()
             -3L,
             false,

src/main/java/org/opensearch/forecast/rest/RestDeleteForecasterAction.java

@@ -12,6 +12,7 @@
 import java.util.Locale;
 
 import org.opensearch.forecast.constant.ForecastCommonMessages;
+import org.opensearch.forecast.indices.ForecastIndex;
 import org.opensearch.forecast.settings.ForecastEnabledSetting;
 import org.opensearch.forecast.transport.DeleteForecasterAction;
 import org.opensearch.rest.BaseRestHandler;
@@ -42,7 +43,7 @@ protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient cli
 
         try {
             String forecasterId = request.param(FORECASTER_ID);
-            DeleteConfigRequest deleteForecasterRequest = new DeleteConfigRequest(forecasterId);
+            DeleteConfigRequest deleteForecasterRequest = new DeleteConfigRequest(forecasterId, ForecastIndex.CONFIG.getIndexName());
             return channel -> client
                 .execute(DeleteForecasterAction.INSTANCE, deleteForecasterRequest, new RestToXContentListener<>(channel));
         } catch (IllegalArgumentException e) {

src/main/java/org/opensearch/forecast/rest/RestForecasterJobAction.java

@@ -14,6 +14,7 @@
 import java.util.Locale;
 
 import org.opensearch.forecast.constant.ForecastCommonMessages;
+import org.opensearch.forecast.indices.ForecastIndex;
 import org.opensearch.forecast.settings.ForecastEnabledSetting;
 import org.opensearch.forecast.transport.ForecasterJobAction;
 import org.opensearch.rest.RestRequest;
@@ -47,7 +48,7 @@ protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient cli
             DateRange dateRange = parseInputDateRange(request);
 
             // false means we don't support backtesting and thus no need to stop backtesting
-            JobRequest forecasterJobRequest = new JobRequest(forecasterId, dateRange, false, rawPath);
+            JobRequest forecasterJobRequest = new JobRequest(forecasterId, ForecastIndex.CONFIG.getIndexName(), dateRange, false, rawPath);
 
             return channel -> client.execute(ForecasterJobAction.INSTANCE, forecasterJobRequest, new RestToXContentListener<>(channel));
         } catch (IllegalArgumentException e) {

src/main/java/org/opensearch/forecast/rest/RestGetForecasterAction.java

@@ -19,6 +19,7 @@
 import java.util.Locale;
 
 import org.opensearch.forecast.constant.ForecastCommonMessages;
+import org.opensearch.forecast.indices.ForecastIndex;
 import org.opensearch.forecast.settings.ForecastEnabledSetting;
 import org.opensearch.forecast.transport.GetForecasterAction;
 import org.opensearch.rest.BaseRestHandler;
@@ -63,6 +64,7 @@ protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient cli
             boolean all = request.paramAsBoolean("_all", false);
             GetConfigRequest getForecasterRequest = new GetConfigRequest(
                 forecasterId,
+                ForecastIndex.CONFIG.getIndexName(),
                 RestActions.parseVersion(request),
                 returnJob,
                 returnTask,

src/main/java/org/opensearch/forecast/rest/handler/IndexForecasterActionHandler.java

@@ -35,23 +35,23 @@ public class IndexForecasterActionHandler extends AbstractForecasterActionHandle
     /**
      * Constructor function.
      *
-     * @param clusterService          ClusterService
-     * @param client                  OS node client that executes actions on the local node
-     * @param transportService        OS transport service
-     * @param forecastIndices         forecast index manager
-     * @param forecasterId            forecaster identifier
-     * @param seqNo                   sequence number of last modification
-     * @param primaryTerm             primary term of last modification
-     * @param refreshPolicy           refresh policy
-     * @param forecaster              forecaster instance
-     * @param requestTimeout          request time out configuration
-     * @param maxSingleStreamForecasters     max single-stream forecasters allowed
-     * @param maxHCForecasters        max HC forecasters allowed
-     * @param maxForecastFeatures     max features allowed per forecaster
-     * @param maxCategoricalFields    max number of categorical fields
-     * @param method                  Rest Method type
-     * @param xContentRegistry        Registry which is used for XContentParser
-     * @param user                    User context
+     * @param clusterService             ClusterService
+     * @param client                     OS node client that executes actions on the local node
+     * @param transportService           OS transport service
+     * @param forecastIndices            forecast index manager
+     * @param forecasterId               forecaster identifier
+     * @param seqNo                      sequence number of last modification
+     * @param primaryTerm                primary term of last modification
+     * @param refreshPolicy              refresh policy
+     * @param forecaster                 forecaster instance
+     * @param requestTimeout             request time out configuration
+     * @param maxSingleStreamForecasters max single-stream forecasters allowed
+     * @param maxHCForecasters           max HC forecasters allowed
+     * @param maxForecastFeatures        max features allowed per forecaster
+     * @param maxCategoricalFields       max number of categorical fields
+     * @param method                     Rest Method type
+     * @param xContentRegistry           Registry which is used for XContentParser
+     * @param user                       User context
      */
     public IndexForecasterActionHandler(
         ClusterService clusterService,