Add setting to ignore source security roles on CCS requests

src/integrationTest/java/org/opensearch/security/privileges/int_tests/CcsIgnoreSourceSecurityRolesIntTests.java

@@ -0,0 +1,203 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ *
+ * Modifications Copyright OpenSearch Contributors. See
+ * GitHub history for details.
+ */
+
+package org.opensearch.security.privileges.int_tests;
+
+import java.util.List;
+
+import com.google.common.collect.ImmutableList;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+import org.opensearch.security.support.ConfigConstants;
+import org.opensearch.test.framework.TestSecurityConfig;
+import org.opensearch.test.framework.certificate.TestCertificates;
+import org.opensearch.test.framework.cluster.ClusterManager;
+import org.opensearch.test.framework.cluster.LocalCluster;
+import org.opensearch.test.framework.cluster.TestRestClient;
+import org.opensearch.test.framework.data.TestIndex;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.opensearch.test.framework.TestSecurityConfig.AuthcDomain.AUTHC_HTTPBASIC_INTERNAL;
+import static org.opensearch.test.framework.matcher.RestMatchers.isForbidden;
+import static org.opensearch.test.framework.matcher.RestMatchers.isOk;
+
+/**
+ * Integration test for the CCS remote role recomputation setting.
+ *
+ * Proves:
+ * 1. With flag=true, source-propagated securityRoles are stripped (CCS query denied for unmapped user)
+ * 2. With flag=true, roles_mapping on the remote still grants access independently (mapped user succeeds)
+ * 3. With flag=false, legacy union behavior is preserved (source roles propagate)
+ * 4. Dynamic setting update via PUT _cluster/settings works at runtime
+ */
+public class CcsIgnoreSourceSecurityRolesIntTests {
+
+    private static final String REMOTE_CLUSTER_FLAG_ON = "remote_flag_on";
+    private static final String REMOTE_CLUSTER_FLAG_OFF = "remote_flag_off";
+    private static final String REMOTE_CLUSTER_DYNAMIC = "remote_dynamic";
+    private static final String INDEX_NAME = "index_r1";
+    private static final String MAPPED_USER_NAME = "mapped_user";
+    private static final String UNMAPPED_USER_NAME = "unmapped_user";
+    private static final String UNLIMITED_ROLE_NAME = "unlimited_role";
+    private static final String READ_ROLE_REMOTE_NAME = "read_role_remote";
+
+    static final TestIndex REMOTE_INDEX = TestIndex.name(INDEX_NAME).documentCount(10).seed(1).build();
+
+    // Role that grants access to ALL indices (assigned to user on local cluster via securityRoles)
+    static final TestSecurityConfig.Role UNLIMITED_ROLE = new TestSecurityConfig.Role(UNLIMITED_ROLE_NAME).clusterPermissions("*")
+        .indexPermissions("*")
+        .on("*");
+
+    // Role that grants read-only on the remote index (mapped via roles_mapping on remote)
+    static final TestSecurityConfig.Role READ_ROLE_REMOTE = new TestSecurityConfig.Role(READ_ROLE_REMOTE_NAME).clusterPermissions(
+        "cluster_composite_ops_ro",
+        "cluster_monitor"
+    ).indexPermissions("read", "indices_monitor", "indices:admin/shards/search_shards").on(INDEX_NAME);
+
+    // User with unlimited access via securityRoles (source cluster), mapped via roles_mapping on remote
+    static final TestSecurityConfig.User MAPPED_USER = new TestSecurityConfig.User(MAPPED_USER_NAME).referencedRoles(UNLIMITED_ROLE);
+
+    // Second user: also has unlimited on source, but NO roles_mapping on remote
+    static final TestSecurityConfig.User UNMAPPED_USER = new TestSecurityConfig.User(UNMAPPED_USER_NAME).referencedRoles(UNLIMITED_ROLE);
+
+    // Roles mapping on remote: maps "mapped_user" -> "read_role_remote"
+    // Note: "unmapped_user" intentionally has NO mapping
+    static final TestSecurityConfig.RoleMapping READ_ROLE_MAPPING = new TestSecurityConfig.RoleMapping(READ_ROLE_REMOTE_NAME).users(
+        MAPPED_USER_NAME
+    );
+
+    static final List<TestSecurityConfig.User> USERS = ImmutableList.of(MAPPED_USER, UNMAPPED_USER);
+
+    static final TestCertificates TEST_CERTIFICATES = new TestCertificates();
+
+    // Remote cluster with flag=TRUE: ignores source securityRoles, uses only its own roles_mapping
+    @ClassRule
+    public static final LocalCluster remoteClusterFlagOn = new LocalCluster.Builder().certificates(TEST_CERTIFICATES)
+        .clusterManager(ClusterManager.SINGLENODE)
+        .clusterName(REMOTE_CLUSTER_FLAG_ON)
+        .authc(AUTHC_HTTPBASIC_INTERNAL)
+        .privilegesEvaluationType("v4")
+        .users(USERS)
+        .roles(UNLIMITED_ROLE, READ_ROLE_REMOTE)
+        .rolesMapping(READ_ROLE_MAPPING)
+        .nodeSetting(ConfigConstants.SECURITY_CCS_IGNORE_SOURCE_SECURITY_ROLES, true)
+        .indices(REMOTE_INDEX)
+        .build();
+
+    // Remote cluster with flag=FALSE: legacy behavior, source securityRoles propagate through
+    @ClassRule
+    public static final LocalCluster remoteClusterFlagOff = new LocalCluster.Builder().certificates(TEST_CERTIFICATES)
+        .clusterManager(ClusterManager.SINGLENODE)
+        .clusterName(REMOTE_CLUSTER_FLAG_OFF)
+        .authc(AUTHC_HTTPBASIC_INTERNAL)
+        .privilegesEvaluationType("v4")
+        .users(USERS)
+        .roles(UNLIMITED_ROLE, READ_ROLE_REMOTE)
+        .nodeSetting(ConfigConstants.SECURITY_CCS_IGNORE_SOURCE_SECURITY_ROLES, false)
+        .indices(REMOTE_INDEX)
+        .build();
+
+    // Dedicated remote cluster for dynamic setting test: starts with flag=FALSE, flipped to TRUE at runtime
+    @ClassRule
+    public static final LocalCluster remoteClusterDynamic = new LocalCluster.Builder().certificates(TEST_CERTIFICATES)
+        .clusterManager(ClusterManager.SINGLENODE)
+        .clusterName(REMOTE_CLUSTER_DYNAMIC)
+        .authc(AUTHC_HTTPBASIC_INTERNAL)
+        .privilegesEvaluationType("v4")
+        .users(USERS)
+        .roles(UNLIMITED_ROLE, READ_ROLE_REMOTE)
+        .nodeSetting(ConfigConstants.SECURITY_CCS_IGNORE_SOURCE_SECURITY_ROLES, false)
+        .indices(REMOTE_INDEX)
+        .build();
+
+    // Local cluster: connects to all three remotes
+    @ClassRule
+    public static final LocalCluster localCluster = new LocalCluster.Builder().certificates(TEST_CERTIFICATES)
+        .clusterManager(ClusterManager.SINGLE_REMOTE_CLIENT)
+        .remote(REMOTE_CLUSTER_FLAG_ON, remoteClusterFlagOn)
+        .remote(REMOTE_CLUSTER_FLAG_OFF, remoteClusterFlagOff)
+        .remote(REMOTE_CLUSTER_DYNAMIC, remoteClusterDynamic)
+        .authc(AUTHC_HTTPBASIC_INTERNAL)
+        .privilegesEvaluationType("v4")
+        .users(USERS)
+        .roles(UNLIMITED_ROLE)
+        .doNotFailOnForbidden(true)
+        .build();
+
+    /**
+     * With flag=true: source's unlimited_role is stripped.
+     * But remote's roles_mapping maps the user -> read_role_remote (read on index_r1).
+     * CCS query should SUCCEED via the remote's own roles_mapping.
+     */
+    @Test
+    public void ccsQuery_withFlagOn_shouldSucceedViaRolesMapping() throws Exception {
+        try (TestRestClient restClient = localCluster.getRestClient(MAPPED_USER)) {
+            TestRestClient.HttpResponse response = restClient.get(REMOTE_CLUSTER_FLAG_ON + ":" + INDEX_NAME + "/_search");
+            assertThat(response, isOk());
+        }
+    }
+
+    /**
+     * With flag=true: source's unlimited_role is stripped.
+     * Unmapped user has NO roles_mapping entry on remote.
+     * CCS query should be FORBIDDEN — proves source securityRoles are actually stripped.
+     */
+    @Test
+    public void ccsQuery_withFlagOn_shouldBeForbidden_whenNoRolesMapping() throws Exception {
+        try (TestRestClient restClient = localCluster.getRestClient(UNMAPPED_USER)) {
+            TestRestClient.HttpResponse response = restClient.get(REMOTE_CLUSTER_FLAG_ON + ":" + INDEX_NAME + "/_search");
+            assertThat(response, isForbidden());
+        }
+    }
+
+    /**
+     * With flag=false (legacy behavior): source's securityRoles propagate through.
+     * Unmapped user has unlimited_role from source — which exists on remote's roles.yml.
+     * CCS query should SUCCEED — proves legacy union behavior is preserved.
+     */
+    @Test
+    public void ccsQuery_withFlagOff_shouldSucceed_whenSourceRolesPropagate() throws Exception {
+        try (TestRestClient restClient = localCluster.getRestClient(UNMAPPED_USER)) {
+            TestRestClient.HttpResponse response = restClient.get(REMOTE_CLUSTER_FLAG_OFF + ":" + INDEX_NAME + "/_search");
+            assertThat(response, isOk());
+        }
+    }
+
+    /**
+     * Dynamic setting update on a dedicated cluster: flip flag from false to true at runtime.
+     * CCS query that previously succeeded should now be forbidden.
+     * Uses a dedicated remote cluster so no other test depends on its state.
+     */
+    @Test
+    public void ccsQuery_withFlagDynamicallyEnabled_shouldBeForbidden() throws Exception {
+        // First: confirm CCS works with flag=false (source roles propagate)
+        try (TestRestClient restClient = localCluster.getRestClient(UNMAPPED_USER)) {
+            TestRestClient.HttpResponse response = restClient.get(REMOTE_CLUSTER_DYNAMIC + ":" + INDEX_NAME + "/_search");
+            assertThat(response, isOk());
+        }
+
+        // Dynamically enable the flag on the dedicated remote cluster
+        try (TestRestClient remoteClient = remoteClusterDynamic.getRestClient(MAPPED_USER)) {
+            TestRestClient.HttpResponse updateResponse = remoteClient.putJson(
+                "_cluster/settings",
+                "{\"transient\": {\"plugins.security.ccs.ignore_source_security_roles\": true}}"
+            );
+            assertThat(updateResponse, isOk());
+        }
+
+        // Now the same CCS query should be forbidden (source roles stripped)
+        try (TestRestClient restClient = localCluster.getRestClient(UNMAPPED_USER)) {
+            TestRestClient.HttpResponse response = restClient.get(REMOTE_CLUSTER_DYNAMIC + ":" + INDEX_NAME + "/_search");
+            assertThat(response, isForbidden());
+        }
+    }
+}

src/integrationTest/java/org/opensearch/security/transport/RemoteClusterIdentityPolicyTest.java

@@ -0,0 +1,86 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ *
+ * Modifications Copyright OpenSearch Contributors. See
+ * GitHub history for details.
+ */
+
+package org.opensearch.security.transport;
+
+import java.util.Arrays;
+
+import com.google.common.collect.ImmutableSet;
+import org.junit.Test;
+
+import org.opensearch.common.settings.Settings;
+import org.opensearch.common.util.concurrent.ThreadContext;
+import org.opensearch.security.support.ConfigConstants;
+import org.opensearch.security.user.User;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+
+public class RemoteClusterIdentityPolicyTest {
+
+    @Test
+    public void sanitize_flagOnAndCcsRequest_stripsSecurityRoles() {
+        RemoteClusterIdentityPolicy policy = new RemoteClusterIdentityPolicy(true);
+        ThreadContext threadContext = createCcsThreadContext();
+        User user = new User("alice").withSecurityRoles(Arrays.asList("all_access"));
+
+        User result = policy.sanitize(user, threadContext);
+
+        assertEquals(ImmutableSet.of(), result.getSecurityRoles());
+        assertEquals("alice", result.getName());
+    }
+
+    @Test
+    public void sanitize_flagOffAndCcsRequest_returnsUnchanged() {
+        RemoteClusterIdentityPolicy policy = new RemoteClusterIdentityPolicy(false);
+        ThreadContext threadContext = createCcsThreadContext();
+        User user = new User("alice").withSecurityRoles(Arrays.asList("all_access"));
+
+        User result = policy.sanitize(user, threadContext);
+
+        assertSame(user, result);
+    }
+
+    @Test
+    public void sanitize_flagOnAndNonCcsRequest_returnsUnchanged() {
+        RemoteClusterIdentityPolicy policy = new RemoteClusterIdentityPolicy(true);
+        ThreadContext threadContext = new ThreadContext(Settings.EMPTY); // no CCS transient
+
+        User user = new User("alice").withSecurityRoles(Arrays.asList("all_access"));
+
+        User result = policy.sanitize(user, threadContext);
+
+        assertSame(user, result);
+    }
+
+    @Test
+    public void sanitize_dynamicUpdate_changesMapBehavior() {
+        RemoteClusterIdentityPolicy policy = new RemoteClusterIdentityPolicy(false);
+        ThreadContext threadContext = createCcsThreadContext();
+        User user = new User("alice").withSecurityRoles(Arrays.asList("all_access"));
+
+        // Flag off: user unchanged
+        assertSame(user, policy.sanitize(user, threadContext));
+
+        // Simulate dynamic settings update
+        policy.setIgnoreSourceSecurityRoles(true);
+
+        // Flag on: securityRoles stripped
+        User result = policy.sanitize(user, threadContext);
+        assertEquals(ImmutableSet.of(), result.getSecurityRoles());
+    }
+
+    private static ThreadContext createCcsThreadContext() {
+        ThreadContext threadContext = new ThreadContext(Settings.EMPTY);
+        threadContext.putTransient(ConfigConstants.OPENDISTRO_SECURITY_SSL_TRANSPORT_TRUSTED_CLUSTER_REQUEST, Boolean.TRUE);
+        return threadContext;
+    }
+}

src/integrationTest/java/org/opensearch/security/user/UserTest.java

@@ -135,4 +135,23 @@ public void withRequestedTenant_unmodified() {
     public void illegalName() {
         new User("");
     }
+
+    @Test
+    public void withoutSecurityRoles_stripsRoles() {
+        User original = new User("test_user").withSecurityRoles(Arrays.asList("all_access", "read_only"));
+        User stripped = original.withoutSecurityRoles();
+
+        assertEquals(ImmutableSet.of("all_access", "read_only"), original.getSecurityRoles());
+        assertEquals(ImmutableSet.of(), stripped.getSecurityRoles());
+        assertEquals(original.getName(), stripped.getName());
+        assertEquals(original.getRoles(), stripped.getRoles());
+    }
+
+    @Test
+    public void withoutSecurityRoles_alreadyEmpty_returnsSameInstance() {
+        User original = new User("test_user");
+        User stripped = original.withoutSecurityRoles();
+
+        assertSame(original, stripped);
+    }
 }

src/main/java/org/opensearch/security/OpenSearchSecurityPlugin.java

@@ -240,6 +240,7 @@
 import org.opensearch.security.support.SecuritySettings;
 import org.opensearch.security.transport.DefaultInterClusterRequestEvaluator;
 import org.opensearch.security.transport.InterClusterRequestEvaluator;
+import org.opensearch.security.transport.RemoteClusterIdentityPolicy;
 import org.opensearch.security.transport.SecurityInterceptor;
 import org.opensearch.security.user.User;
 import org.opensearch.security.user.UserFactory;
@@ -1713,6 +1714,15 @@ public Collection<Object> createComponents(
 
         cr.setDynamicConfigFactory(dcf);
 
+        RemoteClusterIdentityPolicy remoteClusterIdentityPolicy = new RemoteClusterIdentityPolicy(
+            settings.getAsBoolean(ConfigConstants.SECURITY_CCS_IGNORE_SOURCE_SECURITY_ROLES, false)
+        );
+        clusterService.getClusterSettings()
+            .addSettingsUpdateConsumer(SecuritySettings.CCS_IGNORE_SOURCE_SECURITY_ROLES_SETTING, newValue -> {
+                log.info("CCS ignore source security roles dynamically set to {}", newValue);
+                remoteClusterIdentityPolicy.setIgnoreSourceSecurityRoles(newValue);
+            });
+
         si = new SecurityInterceptor(
             settings,
             threadPool,
@@ -1725,7 +1735,8 @@ public Collection<Object> createComponents(
             Objects.requireNonNull(cih),
             SSLConfig,
             OpenSearchSecurityPlugin::isActionTraceEnabled,
-            userFactory
+            userFactory,
+            remoteClusterIdentityPolicy
         );
         components.add(principalExtractor);
 
@@ -1809,6 +1820,9 @@ public List<Setting<?>> getSettings() {
 
         settings.add(Setting.boolSetting(ConfigConstants.SECURITY_SSL_ONLY, false, Property.NodeScope, Property.Filtered));
 
+        // CCS: allow remote cluster to ignore source-propagated security roles
+        settings.add(SecuritySettings.CCS_IGNORE_SOURCE_SECURITY_ROLES_SETTING);
+
         // currently dual mode is supported only when ssl_only is enabled, but this stance would change in future
         settings.add(SecuritySettings.SSL_DUAL_MODE_SETTING);
         settings.add(SecuritySettings.LEGACY_OPENDISTRO_SSL_DUAL_MODE_SETTING);

src/main/java/org/opensearch/security/support/ConfigConstants.java

@@ -454,6 +454,8 @@ public class ConfigConstants {
     public static final List<String> OPENSEARCH_RESOURCE_SHARING_PROTECTED_TYPES_DEFAULT = List.of(); // defaults to no registered types as
                                                                                                       // protected
 
+    public static final String SECURITY_CCS_IGNORE_SOURCE_SECURITY_ROLES = SECURITY_SETTINGS_PREFIX + "ccs.ignore_source_security_roles";
+
     public static Set<String> getSettingAsSet(
         final Settings settings,
         final String key,

src/main/java/org/opensearch/security/support/SecuritySettings.java

@@ -284,4 +284,12 @@ public class SecuritySettings {
         AUDIT_CONFIG_PREFIX + "action_groups.",
         Setting.Property.NodeScope
     );
+
+    // CCS: ignore source-propagated security roles on the remote cluster
+    public static final Setting<Boolean> CCS_IGNORE_SOURCE_SECURITY_ROLES_SETTING = Setting.boolSetting(
+        ConfigConstants.SECURITY_CCS_IGNORE_SOURCE_SECURITY_ROLES,
+        false,
+        Setting.Property.NodeScope,
+        Setting.Property.Dynamic
+    );
 }