Onboards notifications plugin to centralized resource authz

.github/workflows/security-notifications-test-workflow.yml

@@ -17,6 +17,8 @@ jobs:
     strategy:
       matrix:
         java: [ 21 ]
+        # empty string = resource sharing disabled, flag = enabled
+        resource_sharing_flag: ["", "-Dresource_sharing.enabled=true"]
     needs: Get-CI-Image-Tag
     # This job runs on Linux
     runs-on: ubuntu-latest
@@ -46,12 +48,12 @@ jobs:
         run: |
           cd notifications
           chown -R 1000:1000 `pwd`
-          su `id -un 1000` -c "./gradlew integTest -Dsecurity=true -Dhttps=true --tests '*IT'"
+          su `id -un 1000` -c "./gradlew integTest -Dsecurity=true -Dhttps=true ${{ matrix.resource_sharing_flag }} --tests '*IT'"
 
       - name: Upload failed logs
         uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4
         if: failure()
         with:
-          name: logs
+          name: logs-${{ matrix.resource_sharing_flag != '' && 'resource-sharing' || 'no-resource-sharing' }}
           overwrite: 'true'
           path: build/testclusters/integTest-*/logs/*

notifications/notifications/build.gradle

@@ -46,7 +46,7 @@ opensearchplugin {
     name 'opensearch-notifications'
     description 'OpenSearch Notifications Plugin'
     classname 'org.opensearch.notifications.NotificationPlugin'
-    extendedPlugins = ['opensearch-notifications-core']
+    extendedPlugins = ['opensearch-notifications-core', 'opensearch-security;optional=true']
 }
 
 publishing {
@@ -129,6 +129,9 @@ configurations.all {
                     : versions.jackson
                 details.useVersion version
             }
+            if (details.requested.group.startsWith('tools.jackson')) {
+                details.useVersion versions.jackson3
+            }
         }
         force "commons-logging:commons-logging:1.3.5"
         force "commons-codec:commons-codec:1.17.1"
@@ -182,7 +185,8 @@ dependencies {
     // implementation "com.fasterxml.jackson.core:jackson-databind:${versions.jackson_databind}"
     // implementation "com.fasterxml.jackson.core:jackson-annotations:${versions.jackson_annotations}"
 
-    // Needed for security tests
+    // Resource sharing
+    compileOnly group: 'org.opensearch', name:'opensearch-security-spi', version:"${opensearch_build}"
     if (securityEnabled) {
         opensearchPlugin "org.opensearch.plugin:opensearch-security:${opensearch_build}@zip"
     }
@@ -282,6 +286,10 @@ afterEvaluate {
             node.setting("plugins.security.check_snapshot_restore_write_privileges", "true")
             node.setting("plugins.security.restapi.roles_enabled", "[\"all_access\", \"security_rest_api_access\"]")
             node.setting("plugins.security.system_indices.enabled", "true")
+        if (System.getProperty("resource_sharing.enabled") == "true") {
+            node.setting "plugins.security.experimental.resource_sharing.enabled", "true"
+            node.setting "plugins.security.experimental.resource_sharing.protected_types", "[\"notification_config\"]"
+        }
         }
     }
 }
@@ -322,6 +330,7 @@ integTest {
     systemProperty "security", System.getProperty("security")
     systemProperty "user", System.getProperty("user", "admin")
     systemProperty "password", System.getProperty("password", "admin")
+    systemProperty "resource_sharing.enabled", System.getProperty("resource_sharing.enabled")
     // Tell the test JVM if the cluster JVM is running under a debugger so that tests can use longer timeouts for
     // requests. The 'doFirst' delays reading the debug setting on the cluster till execution time.
     doFirst {
@@ -349,6 +358,16 @@ integTest {
         }
     }
 
+    if (System.getProperty("resource_sharing.enabled") != "true") {
+        filter {
+            excludeTestsMatching "org.opensearch.integtest.ResourceSharingNotificationIT"
+        }
+    } else {
+        filter {
+            includeTestsMatching "org.opensearch.integtest.ResourceSharingNotificationIT"
+        }
+    }
+
     if (System.getProperty("tests.rest.bwcsuite") == null) {
         filter {
             excludeTestsMatching "org.opensearch.integtest.bwc.*IT"

notifications/notifications/src/main/kotlin/org/opensearch/notifications/NotificationPlugin.kt

@@ -21,6 +21,7 @@ import org.opensearch.core.common.io.stream.NamedWriteableRegistry
 import org.opensearch.core.xcontent.NamedXContentRegistry
 import org.opensearch.env.Environment
 import org.opensearch.env.NodeEnvironment
+import org.opensearch.identity.PluginSubject
 import org.opensearch.indices.SystemIndexDescriptor
 import org.opensearch.notifications.action.CreateNotificationConfigAction
 import org.opensearch.notifications.action.DeleteNotificationConfigAction
@@ -47,8 +48,10 @@ import org.opensearch.notifications.settings.PluginSettings.REMOTE_METADATA_SERV
 import org.opensearch.notifications.settings.PluginSettings.REMOTE_METADATA_STORE_TYPE
 import org.opensearch.notifications.spi.NotificationCore
 import org.opensearch.notifications.spi.NotificationCoreExtension
+import org.opensearch.notifications.util.PluginClient
 import org.opensearch.notifications.util.SecureIndexClient
 import org.opensearch.plugins.ActionPlugin
+import org.opensearch.plugins.IdentityAwarePlugin
 import org.opensearch.plugins.Plugin
 import org.opensearch.plugins.SystemIndexPlugin
 import org.opensearch.remote.metadata.client.impl.SdkClientFactory
@@ -72,9 +75,10 @@ import java.util.function.Supplier
  * Entry point of the OpenSearch Notifications plugin
  * This class initializes the rest handlers.
  */
-class NotificationPlugin : ActionPlugin, Plugin(), NotificationCoreExtension, SystemIndexPlugin {
+class NotificationPlugin : ActionPlugin, Plugin(), NotificationCoreExtension, SystemIndexPlugin, IdentityAwarePlugin {
 
     lateinit var clusterService: ClusterService // initialized in createComponents()
+    private var pluginClient: PluginClient? = null
 
     internal companion object {
         private val log by logger(NotificationPlugin::class.java)
@@ -146,10 +150,29 @@ class NotificationPlugin : ActionPlugin, Plugin(), NotificationCoreExtension, Sy
             client.threadPool().executor(ThreadPool.Names.GENERIC)
         )
         PluginSettings.addSettingsUpdateConsumer(clusterService)
-        NotificationConfigIndex.initialize(sdkClient, client, clusterService)
+        val pluginClientInstance = PluginClient(client)
+        this.pluginClient = pluginClientInstance
+        val searchSdkClient = SdkClientFactory.createSdkClient(
+            pluginClientInstance,
+            xContentRegistry,
+            mapOf(
+                REMOTE_METADATA_TYPE_KEY to REMOTE_METADATA_STORE_TYPE.get(settings),
+                REMOTE_METADATA_ENDPOINT_KEY to REMOTE_METADATA_ENDPOINT.get(settings),
+                REMOTE_METADATA_REGION_KEY to REMOTE_METADATA_REGION.get(settings),
+                REMOTE_METADATA_SERVICE_NAME_KEY to REMOTE_METADATA_SERVICE_NAME.get(settings),
+                TENANT_AWARE_KEY to MULTI_TENANCY_ENABLED.get(settings).toString(),
+                TENANT_ID_FIELD_KEY to "tenant_id"
+            ),
+            client.threadPool().executor(ThreadPool.Names.GENERIC)
+        )
+        NotificationConfigIndex.initialize(sdkClient, searchSdkClient, client, clusterService)
         ConfigIndexingActions.initialize(NotificationConfigIndex, UserAccessManager)
         SendMessageActionHelper.initialize(NotificationConfigIndex, UserAccessManager)
-        return listOf(sdkClient)
+        return listOf(sdkClient, pluginClientInstance)
+    }
+
+    override fun assignSubject(pluginSubject: PluginSubject) {
+        pluginClient?.setSubject(pluginSubject)
     }
 
     /**

notifications/notifications/src/main/kotlin/org/opensearch/notifications/NotificationsResourceSharingExtension.kt

@@ -0,0 +1,30 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.opensearch.notifications
+
+import org.opensearch.notifications.index.NotificationConfigIndex
+import org.opensearch.security.spi.resources.ResourceProvider
+import org.opensearch.security.spi.resources.ResourceSharingExtension
+import org.opensearch.security.spi.resources.client.ResourceSharingClient
+
+class NotificationsResourceSharingExtension : ResourceSharingExtension {
+
+    companion object {
+        const val RESOURCE_TYPE = "notification_config"
+    }
+
+    override fun getResourceProviders(): Set<ResourceProvider> {
+        return setOf(
+            object : ResourceProvider {
+                override fun resourceType(): String = RESOURCE_TYPE
+                override fun resourceIndexName(): String = NotificationConfigIndex.INDEX_NAME
+            }
+        )
+    }
+
+    override fun assignResourceSharingClient(client: ResourceSharingClient?) {
+        ResourceSharingClientAccessor.setResourceSharingClient(client)
+    }
+}

notifications/notifications/src/main/kotlin/org/opensearch/notifications/ResourceSharingClientAccessor.kt

@@ -0,0 +1,29 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.opensearch.notifications
+
+import org.opensearch.security.spi.resources.client.ResourceSharingClient
+
+/**
+ * Accessor for resource sharing client
+ */
+object ResourceSharingClientAccessor {
+
+    @Volatile
+    private var client: ResourceSharingClient? = null
+
+    @JvmStatic
+    fun setResourceSharingClient(client: ResourceSharingClient?) {
+        this.client = client
+    }
+
+    @JvmStatic
+    fun getResourceSharingClient(): ResourceSharingClient? = client
+
+    @JvmStatic
+    fun clear() {
+        client = null
+    }
+}

notifications/notifications/src/main/kotlin/org/opensearch/notifications/index/ConfigIndexingActions.kt

@@ -38,6 +38,7 @@ import org.opensearch.notifications.metrics.Metrics
 import org.opensearch.notifications.model.DocMetadata
 import org.opensearch.notifications.model.NotificationConfigDoc
 import org.opensearch.notifications.security.UserAccess
+import org.opensearch.notifications.security.UserAccessManager
 import java.time.Instant
 /**
  * NotificationConfig indexing operation actions.
@@ -310,6 +311,9 @@ object ConfigIndexingActions {
      */
     private suspend fun info(configIds: Set<String>, user: User?): GetNotificationConfigResponse {
         log.info("$LOG_PREFIX:NotificationConfig-info $configIds")
+        configIds.forEach { id ->
+            UserAccessManager.verifyResourceAccess(id, "cluster:admin/opensearch/notifications/configs/get")
+        }
         val configDocs = operations.getNotificationConfigs(configIds)
         if (configDocs.size != configIds.size) {
             val mutableSet = configIds.toMutableSet()
@@ -441,6 +445,9 @@ object ConfigIndexingActions {
     private suspend fun delete(configIds: Set<String>, user: User?): DeleteNotificationConfigResponse {
         log.info("$LOG_PREFIX:NotificationConfig-delete $configIds")
         userAccess.validateUser(user)
+        configIds.forEach { id ->
+            UserAccessManager.verifyResourceAccess(id, "cluster:admin/opensearch/notifications/configs/delete")
+        }
         val configDocs = operations.getNotificationConfigs(configIds)
         if (configDocs.size != configIds.size) {
             val mutableSet = configIds.toMutableSet()

notifications/notifications/src/main/kotlin/org/opensearch/notifications/index/NotificationConfigIndex.kt

@@ -83,6 +83,7 @@ internal object NotificationConfigIndex : ConfigOperations {
     private lateinit var client: Client
     private lateinit var clusterService: ClusterService
     private lateinit var sdkClient: SdkClient
+    private lateinit var searchSdkClient: SdkClient
 
     private val searchHitParser = object : SearchResults.SearchHitParser<NotificationConfigInfo> {
         override fun parse(searchHit: SearchHit): NotificationConfigInfo {
@@ -105,10 +106,11 @@ internal object NotificationConfigIndex : ConfigOperations {
     /**
      * {@inheritDoc}
      */
-    fun initialize(sdkClient: SdkClient, client: Client, clusterService: ClusterService) {
+    fun initialize(sdkClient: SdkClient, searchSdkClient: SdkClient, client: Client, clusterService: ClusterService) {
         NotificationConfigIndex.client = SecureIndexClient(client)
         NotificationConfigIndex.clusterService = clusterService
         NotificationConfigIndex.sdkClient = sdkClient
+        NotificationConfigIndex.searchSdkClient = searchSdkClient
     }
 
     private fun getSchemaVersionFromIndexMapping(indexMapping: Map<String, Any>?): Int {
@@ -290,8 +292,8 @@ internal object NotificationConfigIndex : ConfigOperations {
             .searchSourceBuilder(sourceBuilder)
             .build()
 
-        val response: SearchResponse = sdkClient.suspendUntilTimeout(PluginSettings.operationTimeoutMs) {
-            sdkClient.searchDataObjectAsync(searchRequest).whenComplete(it)
+        val response: SearchResponse = searchSdkClient.suspendUntilTimeout(PluginSettings.operationTimeoutMs) {
+            searchSdkClient.searchDataObjectAsync(searchRequest).whenComplete(it)
         }
         val result = NotificationConfigSearchResult(request.fromIndex.toLong(), response, searchHitParser)
         log.info(

notifications/notifications/src/main/kotlin/org/opensearch/notifications/security/UserAccessManager.kt

@@ -5,22 +5,34 @@
 
 package org.opensearch.notifications.security
 
+import kotlinx.coroutines.suspendCancellableCoroutine
 import org.opensearch.OpenSearchStatusException
 import org.opensearch.commons.authuser.User
+import org.opensearch.core.action.ActionListener
 import org.opensearch.core.rest.RestStatus
+import org.opensearch.notifications.NotificationsResourceSharingExtension.Companion.RESOURCE_TYPE
+import org.opensearch.notifications.ResourceSharingClientAccessor
 import org.opensearch.notifications.settings.FilterByBackendRolesAccessStrategy
 import org.opensearch.notifications.settings.PluginSettings
+import kotlin.coroutines.resume
+import kotlin.coroutines.resumeWithException
 
 /**
  * Class for checking/filtering user access.
  */
 internal object UserAccessManager : UserAccess {
     const val ADMIN_ROLE = "all_access"
 
+    private fun isResourceSharingEnabled(): Boolean {
+        val client = ResourceSharingClientAccessor.getResourceSharingClient()
+        return client != null && client.isFeatureEnabledForType(RESOURCE_TYPE)
+    }
+
     /**
      * {@inheritDoc}
      */
     override fun validateUser(user: User?) {
+        if (isResourceSharingEnabled()) return
         if (PluginSettings.isRbacEnabled() && user?.backendRoles.isNullOrEmpty()) {
             throw OpenSearchStatusException(
                 "User doesn't have backend roles configured. Contact administrator.",
@@ -33,7 +45,7 @@ internal object UserAccessManager : UserAccess {
      * {@inheritDoc}
      */
     override fun getAllAccessInfo(user: User?): List<String> {
-        if (user == null) { // Filtering is disabled
+        if (isResourceSharingEnabled() || user == null) {
             return listOf()
         }
         return user.backendRoles
@@ -43,7 +55,7 @@ internal object UserAccessManager : UserAccess {
      * {@inheritDoc}
      */
     override fun getSearchAccessInfo(user: User?): List<String> {
-        if (user == null || !PluginSettings.isRbacEnabled() || user.roles.contains(ADMIN_ROLE)) { // Filtering is disabled
+        if (isResourceSharingEnabled() || user == null || !PluginSettings.isRbacEnabled() || user.roles.contains(ADMIN_ROLE)) {
             return listOf()
         }
         return user.backendRoles
@@ -67,11 +79,36 @@ internal object UserAccessManager : UserAccess {
      * {@inheritDoc}
      */
     override fun doesUserHaveAccess(user: User?, access: List<String>): Boolean {
-        if (user == null || !PluginSettings.isRbacEnabled()) { // Filtering is disabled
+        if (isResourceSharingEnabled() || user == null || !PluginSettings.isRbacEnabled()) {
             return true
         }
-        // User has access to resource if resource is public i.e. no access roles attached, user is an admin user or there is any intersection
-        // between user backend roles and access roles
         return access.isEmpty() || user.roles.contains(ADMIN_ROLE) || checkUserBackendRolesAccess(user.backendRoles, access)
     }
+
+    /**
+     * Verify resource access via the security plugin's ResourceSharingClient.
+     * Used for multi-ID requests where DocRequest.id() returns null and the
+     * transport-level ResourceAccessEvaluator is skipped.
+     */
+    suspend fun verifyResourceAccess(resourceId: String, action: String) {
+        val client = ResourceSharingClientAccessor.getResourceSharingClient() ?: return
+        if (!client.isFeatureEnabledForType(RESOURCE_TYPE)) return
+        val hasAccess = suspendCancellableCoroutine { cont ->
+            client.verifyAccess(
+                resourceId,
+                RESOURCE_TYPE,
+                action,
+                object : ActionListener<Boolean> {
+                    override fun onResponse(response: Boolean) = cont.resume(response)
+                    override fun onFailure(e: Exception) = cont.resumeWithException(e)
+                }
+            )
+        }
+        if (!hasAccess) {
+            throw OpenSearchStatusException(
+                "no permissions for [$action] on resource [$resourceId]",
+                RestStatus.FORBIDDEN
+            )
+        }
+    }
 }