add user role filter for monitor on top of ad result

alerting/src/main/kotlin/com/amazon/opendistroforelasticsearch/alerting/MonitorRunner.kt

@@ -54,6 +54,8 @@ import com.amazon.opendistroforelasticsearch.alerting.settings.AlertingSettings.
 import com.amazon.opendistroforelasticsearch.alerting.settings.DestinationSettings.Companion.ALLOW_LIST
 import com.amazon.opendistroforelasticsearch.alerting.settings.DestinationSettings.Companion.loadDestinationSettings
 import com.amazon.opendistroforelasticsearch.alerting.util.IndexUtils
+import com.amazon.opendistroforelasticsearch.alerting.util.addUserBackendRolesFilter
+import com.amazon.opendistroforelasticsearch.alerting.util.isADMonitor
 import com.amazon.opendistroforelasticsearch.alerting.util.isAllowed
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
@@ -224,9 +226,14 @@ class MonitorRunner(
             logger.error("Error loading alerts for monitor: $id", e)
             return monitorResult.copy(error = e)
         }
-        runBlocking(InjectorContextElement(monitor.id, settings, threadPool.threadContext, roles)) {
-            monitorResult = monitorResult.copy(inputResults = collectInputResults(monitor, periodStart, periodEnd))
+        if (!isADMonitor(monitor)) {
+            runBlocking(InjectorContextElement(monitor.id, settings, threadPool.threadContext, roles)) {
+                monitorResult = monitorResult.copy(inputResults = collectInputResults(monitor, periodStart, periodEnd))
+            }
+        } else {
+            monitorResult = monitorResult.copy(inputResults = collectInputResultsForADMonitor(monitor, periodStart, periodEnd))
         }
+
         val updatedAlerts = mutableListOf<Alert>()
         val triggerResults = mutableMapOf<String, TriggerRunResult>()
         for (trigger in monitor.triggers) {
@@ -338,6 +345,61 @@ class MonitorRunner(
         }
     }
 
+    /**
+     * We moved anomaly result index to system index list. So common user could not directly query
+     * this index any more. This method will stash current thread context to pass security check.
+     * So monitor job can access anomaly result index. We will add monitor user roles filter in
+     * search query to only return documents the monitor user can access.
+     *
+     * On alerting Kibana, monitor users can only see detectors that they have read access. So they
+     * can't create monitor on other user's detector which they have no read access. Even they know
+     * other user's detector id and use it to create monitor, this method will only return anomaly
+     * results they can read.
+     */
+    private suspend fun collectInputResultsForADMonitor(monitor: Monitor, periodStart: Instant, periodEnd: Instant): InputRunResults {
+        return try {
+            val results = mutableListOf<Map<String, Any>>()
+            val input = monitor.inputs[0] as SearchInput
+
+            val searchParams = mapOf("period_start" to periodStart.toEpochMilli(), "period_end" to periodEnd.toEpochMilli())
+            val searchSource = scriptService.compile(Script(ScriptType.INLINE, Script.DEFAULT_TEMPLATE_LANG,
+                    input.query.toString(), searchParams), TemplateScript.CONTEXT)
+                    .newInstance(searchParams)
+                    .execute()
+
+            val searchRequest = SearchRequest().indices(*input.indices.toTypedArray())
+            XContentType.JSON.xContent().createParser(xContentRegistry, LoggingDeprecationHandler.INSTANCE, searchSource).use {
+                searchRequest.source(SearchSourceBuilder.fromXContent(it))
+            }
+
+            // Add user role filter for AD result
+            client.threadPool().threadContext.stashContext().use {
+                // Currently we have no way to verify if user has AD read permission or not. So we always add user
+                // role filter here no matter AD backend role filter enabled or not. If we don't add user role filter
+                // when AD backend filter disabled, user can run monitor on any detector and get anomaly data even
+                // they have no AD read permission. So if domain disabled AD backend role filter, monitor runner
+                // still can't get AD result with different user backend role, even the monitor user has permission
+                // to read AD result. This is a short term solution to trade off between user experience and security.
+                //
+                // Possible long term solution:
+                // 1.Use secure rest client to send request to AD search result API. If no permission exception,
+                // that mean user has read access on AD result. Then don't need to add user role filter when query
+                // AD result if AD backend role filter is disabled.
+                // 2.Security provide some transport action to verify if user has permission to search AD result.
+                // Monitor runner will send transport request to check permission first. If security plugin response
+                // is yes, user has permission to query AD result. If AD role filter enabled, we will add user role
+                // filter to protect data at user role level; otherwise, user can query any AD result.
+                addUserBackendRolesFilter(monitor.user, searchRequest.source())
+                val searchResponse: SearchResponse = client.suspendUntil { client.search(searchRequest, it) }
+                results += searchResponse.convertToMap()
+            }
+            InputRunResults(results.toList())
+        } catch (e: Exception) {
+            logger.info("Error collecting anomaly result inputs for monitor: ${monitor.id}", e)
+            InputRunResults(emptyList(), e)
+        }
+    }
+
     private fun runTrigger(monitor: Monitor, trigger: Trigger, ctx: TriggerExecutionContext): TriggerRunResult {
         return try {
             val triggered = scriptService.compile(trigger.condition, TriggerScript.CONTEXT)

alerting/src/main/kotlin/com/amazon/opendistroforelasticsearch/alerting/model/Monitor.kt

@@ -191,7 +191,7 @@ data class Monitor(
                 when (fieldName) {
                     SCHEMA_VERSION_FIELD -> schemaVersion = xcp.intValue()
                     NAME_FIELD -> name = xcp.text()
-                    USER_FIELD -> user = User.parse(xcp)
+                    USER_FIELD -> user = if (xcp.currentToken() == Token.VALUE_NULL) null else User.parse(xcp)
                     ENABLED_FIELD -> enabled = xcp.booleanValue()
                     SCHEDULE_FIELD -> schedule = Schedule.parse(xcp)
                     INPUTS_FIELD -> {

alerting/src/main/kotlin/com/amazon/opendistroforelasticsearch/alerting/transport/TransportIndexMonitorAction.kt

@@ -31,6 +31,8 @@ import com.amazon.opendistroforelasticsearch.alerting.settings.AlertingSettings.
 import com.amazon.opendistroforelasticsearch.alerting.settings.DestinationSettings.Companion.ALLOW_LIST
 import com.amazon.opendistroforelasticsearch.alerting.util.AlertingException
 import com.amazon.opendistroforelasticsearch.alerting.util.IndexUtils
+import com.amazon.opendistroforelasticsearch.alerting.util.addUserBackendRolesFilter
+import com.amazon.opendistroforelasticsearch.alerting.util.isADMonitor
 import com.amazon.opendistroforelasticsearch.commons.authuser.User
 import com.amazon.opendistroforelasticsearch.commons.authuser.AuthUserRequestBuilder
 import org.apache.logging.log4j.LogManager
@@ -100,7 +102,12 @@ class TransportIndexMonitorAction @Inject constructor(
     }
 
     override fun doExecute(task: Task, request: IndexMonitorRequest, actionListener: ActionListener<IndexMonitorResponse>) {
-        checkIndicesAndExecute(client, actionListener, request)
+        if (!isADMonitor(request.monitor)) {
+            checkIndicesAndExecute(client, actionListener, request)
+        } else {
+            // check if user has access to any anomaly detector for AD monitor
+            checkAnomalyDetectorAndExecute(client, actionListener, request)
+        }
     }
 
     /**
@@ -142,6 +149,41 @@ class TransportIndexMonitorAction @Inject constructor(
         })
     }
 
+    /**
+     * It's no reasonable to create AD monitor if the user has no access to any detector. Otherwise
+     * the monitor will not get any anomaly result. So we will check user has access to at least 1
+     * anomaly detector if they need to create AD monitor.
+     * As anomaly detector index is system index, common user has no permission to query. So we need
+     * to send REST API call to AD REST API.
+     */
+    fun checkAnomalyDetectorAndExecute(
+        client: Client,
+        actionListener: ActionListener<IndexMonitorResponse>,
+        request: IndexMonitorRequest
+    ) {
+        client.threadPool().threadContext.stashContext().use {
+            val searchSourceBuilder = SearchSourceBuilder().size(0)
+            addUserBackendRolesFilter(request.monitor.user, searchSourceBuilder)
+            val searchRequest = SearchRequest().indices(".opendistro-anomaly-detectors").source(searchSourceBuilder)
+            client.search(searchRequest, object : ActionListener<SearchResponse> {
+                override fun onResponse(response: SearchResponse?) {
+                    val totalHits = response?.hits?.totalHits?.value
+                    if (totalHits != null && totalHits > 0L) {
+                        IndexMonitorHandler(client, actionListener, request).resolveUserAndStart()
+                    } else {
+                        actionListener.onFailure(AlertingException.wrap(
+                                ElasticsearchStatusException("User has no available detectors", RestStatus.NOT_FOUND)
+                        ))
+                    }
+                }
+
+                override fun onFailure(t: Exception) {
+                    actionListener.onFailure(AlertingException.wrap(t))
+                }
+            })
+        }
+    }
+
     inner class IndexMonitorHandler(
         private val client: Client,
         private val actionListener: ActionListener<IndexMonitorResponse>,

alerting/src/main/kotlin/com/amazon/opendistroforelasticsearch/alerting/util/AnomalyDetectionUtils.kt

@@ -0,0 +1,77 @@
+/*
+ *   Copyright 2020 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *   Licensed under the Apache License, Version 2.0 (the "License").
+ *   You may not use this file except in compliance with the License.
+ *   A copy of the License is located at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *   or in the "license" file accompanying this file. This file 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 com.amazon.opendistroforelasticsearch.alerting.util
+
+import com.amazon.opendistroforelasticsearch.alerting.core.model.SearchInput
+import com.amazon.opendistroforelasticsearch.alerting.model.Monitor
+import com.amazon.opendistroforelasticsearch.commons.authuser.User
+import org.apache.lucene.search.join.ScoreMode
+import org.elasticsearch.index.query.BoolQueryBuilder
+import org.elasticsearch.index.query.NestedQueryBuilder
+import org.elasticsearch.index.query.QueryBuilders
+import org.elasticsearch.search.builder.SearchSourceBuilder
+
+/**
+ * AD monitor is search input monitor on top of anomaly result index. This method will return
+ * true if monitor input only contains anomaly result index.
+ */
+fun isADMonitor(monitor: Monitor): Boolean {
+    // If monitor has other input than AD result index, it's not AD monitor
+    if (monitor.inputs.size != 1) {
+        return false
+    }
+    val input = monitor.inputs[0]
+    // AD monitor can only have 1 anomaly result index.
+    if (input is SearchInput && input.indices.size == 1 && input.indices[0] == ".opendistro-anomaly-results*") {
+        return true
+    }
+    return false
+}
+
+fun addUserBackendRolesFilter(user: User?, searchSourceBuilder: SearchSourceBuilder): SearchSourceBuilder {
+    var boolQueryBuilder = BoolQueryBuilder()
+    val userFieldName = "user"
+    val userBackendRoleFieldName = "user.backend_roles.keyword"
+    if (user == null) {
+        // For old monitor and detector, they have no user field
+        val userRolesFilterQuery = QueryBuilders.existsQuery(userFieldName)
+        val nestedQueryBuilder = NestedQueryBuilder(userFieldName, userRolesFilterQuery, ScoreMode.None)
+        boolQueryBuilder.mustNot(nestedQueryBuilder)
+    } else if (user.backendRoles.isNullOrEmpty()) {
+        // For simple FGAC user, they may have no backend roles, these users should be able to see detectors
+        // of other users whose backend role is empty.
+        val userRolesFilterQuery = QueryBuilders.existsQuery(userBackendRoleFieldName)
+        val nestedQueryBuilder = NestedQueryBuilder(userFieldName, userRolesFilterQuery, ScoreMode.None)
+
+        val userExistsQuery = QueryBuilders.existsQuery(userFieldName)
+        val userExistsNestedQueryBuilder = NestedQueryBuilder(userFieldName, userExistsQuery, ScoreMode.None)
+
+        boolQueryBuilder.mustNot(nestedQueryBuilder)
+        boolQueryBuilder.must(userExistsNestedQueryBuilder)
+    } else {
+        // For normal case, user should have backend roles.
+        val userRolesFilterQuery = QueryBuilders.termsQuery(userBackendRoleFieldName, user.backendRoles)
+        val nestedQueryBuilder = NestedQueryBuilder(userFieldName, userRolesFilterQuery, ScoreMode.None)
+        boolQueryBuilder.must(nestedQueryBuilder)
+    }
+    val query = searchSourceBuilder.query()
+    if (query == null) {
+        searchSourceBuilder.query(boolQueryBuilder)
+    } else {
+        (query as BoolQueryBuilder).filter(boolQueryBuilder)
+    }
+    return searchSourceBuilder
+}