[Backport 1.3] Prevent OptionalDataException from User data structures

src/integrationTest/java/org/opensearch/security/IndexOperationsHelper.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright OpenSearch Contributors
+ * 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.
+ *
+ */
+package org.opensearch.security;
+
+import java.util.Map;
+import org.opensearch.action.admin.indices.close.CloseIndexRequest;
+import org.opensearch.action.admin.indices.close.CloseIndexResponse;
+import org.opensearch.action.admin.indices.create.CreateIndexRequest;
+import org.opensearch.action.admin.indices.create.CreateIndexResponse;
+import org.opensearch.action.admin.indices.mapping.put.PutMappingRequest;
+import org.opensearch.action.support.master.AcknowledgedResponse;
+import org.opensearch.client.Client;
+import org.opensearch.cluster.metadata.IndexMetadata;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.test.framework.cluster.LocalCluster;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.opensearch.test.framework.matcher.ClusterMatchers.indexExists;
+import static org.opensearch.test.framework.matcher.ClusterMatchers.indexStateIsEqualTo;
+
+public class IndexOperationsHelper {
+
+	public static void createIndex(LocalCluster cluster, String indexName) {
+		createIndex(cluster, indexName, Settings.EMPTY);
+	}
+
+	public static void createIndex(LocalCluster cluster, String indexName, Settings settings) {
+		try (Client client = cluster.getInternalNodeClient()) {
+			CreateIndexResponse createIndexResponse = client.admin()
+					.indices()
+					.create(new CreateIndexRequest(indexName).settings(settings))
+					.actionGet();
+
+			assertThat(createIndexResponse.isAcknowledged(), is(true));
+			assertThat(createIndexResponse.isShardsAcknowledged(), is(true));
+			assertThat(cluster, indexExists(indexName));
+		}
+	}
+
+	public static void closeIndex(LocalCluster cluster, String indexName) {
+		try (Client client = cluster.getInternalNodeClient()) {
+			CloseIndexRequest closeIndexRequest = new CloseIndexRequest(indexName);
+			CloseIndexResponse response = client.admin().indices().close(closeIndexRequest).actionGet();
+
+			assertThat(response.isAcknowledged(), is(true));
+			assertThat(response.isShardsAcknowledged(), is(true));
+			assertThat(cluster, indexStateIsEqualTo(indexName, IndexMetadata.State.CLOSE));
+		}
+	}
+
+	public static void createMapping(LocalCluster cluster, String indexName, Map<String, Object> indexMapping) {
+		try (Client client = cluster.getInternalNodeClient()) {
+			AcknowledgedResponse response = client.admin().indices().putMapping(new PutMappingRequest(indexName).source(indexMapping)).actionGet();
+
+			assertThat(response.isAcknowledged(), is(true));
+		}
+	}
+}

src/integrationTest/java/org/opensearch/security/http/AsyncTests.java

@@ -0,0 +1,107 @@
+/*
+ * Copyright OpenSearch Contributors
+ * 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.
+ *
+ */
+
+package org.opensearch.security.http;
+
+import com.carrotsearch.randomizedtesting.annotations.ThreadLeakScope;
+
+import java.util.HashMap;
+import org.apache.http.HttpStatus;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.opensearch.security.IndexOperationsHelper;
+import org.opensearch.security.support.ConfigConstants;
+import org.opensearch.test.framework.AsyncActions;
+import org.opensearch.test.framework.RolesMapping;
+import org.opensearch.test.framework.TestSecurityConfig;
+import org.opensearch.test.framework.cluster.LocalCluster;
+import org.opensearch.test.framework.cluster.TestRestClient;
+import org.opensearch.test.framework.cluster.TestRestClient.HttpResponse;
+
+import java.util.Map;
+import java.util.List;
+import java.util.ArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.CompletableFuture;
+
+import static org.opensearch.test.framework.TestSecurityConfig.AuthcDomain.AUTHC_HTTPBASIC_INTERNAL;
+import static org.opensearch.test.framework.TestSecurityConfig.Role.ALL_ACCESS;
+
+@RunWith(com.carrotsearch.randomizedtesting.RandomizedRunner.class)
+@ThreadLeakScope(ThreadLeakScope.Scope.NONE)
+public class AsyncTests {
+	private static final TestSecurityConfig.User ADMIN_USER = new TestSecurityConfig.User("admin").roles(ALL_ACCESS).backendRoles("admin");
+
+	private static Map<String, Object> createTestNodeSettings() {
+		Map<String, Object> testNodeSettings = new HashMap<>();
+		List<String> rolesEnabled = new ArrayList<>();
+		rolesEnabled.add(ALL_ACCESS.getName());
+		testNodeSettings.put(ConfigConstants.SECURITY_RESTAPI_ROLES_ENABLED, rolesEnabled);
+		return testNodeSettings;
+	}
+
+	@ClassRule
+	public static LocalCluster cluster = new LocalCluster.Builder().singleNode()
+			.authc(AUTHC_HTTPBASIC_INTERNAL)
+			.users(ADMIN_USER)
+			.rolesMapping(new RolesMapping(ALL_ACCESS).backendRoles("admin"))
+			.anonymousAuth(false)
+			.nodeSettings(createTestNodeSettings())
+			.build();
+
+	@Test
+	public void testBulkAndCacheInvalidationMixed() throws Exception {
+		String indexName = "test-index";
+		final String invalidateCachePath = "_plugins/_security/api/cache";
+		final String nodesPath = "_nodes";
+		final String bulkPath = "_bulk";
+		int repeatCount = 5;
+		String docContents = "{ \"index\": { \"_index\": \"" + indexName + "\" }}\n{ \"foo\": \"bar\" }\n";
+
+		StringBuilder builder = new StringBuilder();
+		for (int i = 0; i < repeatCount; i++) {
+			builder.append(docContents);
+		}
+
+		String document = builder.toString();
+		final int parallelism = 5;
+		final int totalNumberOfRequests = 30;
+
+		try (final TestRestClient client = cluster.getRestClient(ADMIN_USER)) {
+			IndexOperationsHelper.createIndex(cluster, indexName);
+
+			final CountDownLatch countDownLatch = new CountDownLatch(1);
+
+			List<CompletableFuture<HttpResponse>> allRequests = new ArrayList<CompletableFuture<HttpResponse>>();
+
+			allRequests.addAll(AsyncActions.generate(() -> {
+				countDownLatch.await();
+				return client.delete(invalidateCachePath);
+			}, parallelism, totalNumberOfRequests));
+
+			allRequests.addAll(AsyncActions.generate(() -> {
+				countDownLatch.await();
+				return client.postJson(bulkPath, document);
+			}, parallelism, totalNumberOfRequests));
+
+			allRequests.addAll(AsyncActions.generate(() -> {
+				countDownLatch.await();
+				return client.get(nodesPath);
+			}, parallelism, totalNumberOfRequests));
+
+			// Make sure all requests start at the same time
+			countDownLatch.countDown();
+
+			AsyncActions.getAll(allRequests, 30, TimeUnit.SECONDS).forEach((response) -> { response.assertStatusCode(HttpStatus.SC_OK); });
+		}
+	}
+}

src/integrationTest/java/org/opensearch/security/http/BasicAuthTests.java

@@ -40,7 +40,7 @@ public class BasicAuthTests {
     static final User TEST_USER = new User("test_user").password("s3cret");
 
     public static final String CUSTOM_ATTRIBUTE_NAME = "superhero";
-    static final User SUPER_USER = new User("super-user").password("super-password").attr(CUSTOM_ATTRIBUTE_NAME, true);
+    static final User SUPER_USER = new User("super-user").password("super-password").attr(CUSTOM_ATTRIBUTE_NAME, "true");
     public static final String NOT_EXISTING_USER = "not-existing-user";
     public static final String INVALID_PASSWORD = "secret-password";
 

src/integrationTest/java/org/opensearch/test/framework/TestSecurityConfig.java

@@ -76,6 +76,10 @@
 */
 public class TestSecurityConfig {
 
+    public final static TestSecurityConfig.User USER_ADMIN = new TestSecurityConfig.User("admin").roles(
+            new Role("allaccess").indexPermissions("*").on("*").clusterPermissions("*")
+    );
+
     private static final Logger log = LogManager.getLogger(TestSecurityConfig.class);
 
     private Config config = new Config();
@@ -256,13 +260,14 @@ public XContentBuilder toXContent(XContentBuilder xContentBuilder, Params params
     public static class User implements UserCredentialsHolder, ToXContentObject {
 
         public final static TestSecurityConfig.User USER_ADMIN = new TestSecurityConfig.User("admin").roles(
-            new Role("allaccess").indexPermissions("*").on("*").clusterPermissions("*")
+                new Role("allaccess").indexPermissions("*").on("*").clusterPermissions("*")
         );
 
         String name;
         private String password;
         List<Role> roles = new ArrayList<>();
-        private Map<String, Object> attributes = new HashMap<>();
+        List<String> backendRoles = new ArrayList<>();
+        private Map<String, String> attributes = new HashMap<>();
 
         public User(String name) {
             this.name = name;
@@ -283,7 +288,12 @@ public User roles(Role... roles) {
             return this;
         }
 
-        public User attr(String key, Object value) {
+        public User backendRoles(String... backendRoles) {
+            this.backendRoles.addAll(Arrays.asList(backendRoles));
+            return this;
+        }
+
+        public User attr(String key, String value) {
             this.attributes.put(key, value);
             return this;
         }
@@ -316,6 +326,10 @@ public XContentBuilder toXContent(XContentBuilder xContentBuilder, Params params
                 xContentBuilder.field("opendistro_security_roles", roleNames);
             }
 
+            if (!backendRoles.isEmpty()) {
+                xContentBuilder.field("backend_roles", backendRoles);
+            }
+
             if (attributes != null && attributes.size() != 0) {
                 xContentBuilder.field("attributes", attributes);
             }

src/integrationTest/java/org/opensearch/test/framework/matcher/ClusterMatchers.java

@@ -56,4 +56,8 @@ public static Matcher<LocalCluster> indexExists(String expectedIndexName) {
     public static Matcher<LocalCluster> indexSettingsContainValues(String expectedIndexName, Settings expectedSettings) {
         return new IndexSettingsContainValuesMatcher(expectedIndexName, expectedSettings);
     }
+
+    public static Matcher<LocalCluster> indexStateIsEqualTo(String expectedIndexName, IndexMetadata.State expectedState) {
+        return new IndexStateIsEqualToMatcher(expectedIndexName, expectedState);
+    }
 }

src/integrationTest/java/org/opensearch/test/framework/matcher/IndexStateIsEqualToMatcher.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright OpenSearch Contributors
+ * 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.
+ *
+ */
+package org.opensearch.test.framework.matcher;
+
+import org.hamcrest.Description;
+import org.hamcrest.TypeSafeDiagnosingMatcher;
+import org.opensearch.action.admin.cluster.state.ClusterStateRequest;
+import org.opensearch.action.admin.cluster.state.ClusterStateResponse;
+import org.opensearch.client.Client;
+import org.opensearch.cluster.metadata.IndexMetadata;
+import org.opensearch.common.collect.ImmutableOpenMap;
+import org.opensearch.test.framework.cluster.LocalCluster;
+import static java.util.Objects.requireNonNull;
+
+class IndexStateIsEqualToMatcher extends TypeSafeDiagnosingMatcher<LocalCluster> {
+
+	private final String expectedIndexName;
+	private final IndexMetadata.State expectedState;
+
+	IndexStateIsEqualToMatcher(String expectedIndexName, IndexMetadata.State expectedState) {
+		this.expectedIndexName = requireNonNull(expectedIndexName);
+		this.expectedState = requireNonNull(expectedState);
+	}
+
+	@Override
+	protected boolean matchesSafely(LocalCluster cluster, Description mismatchDescription) {
+		try (Client client = cluster.getInternalNodeClient()) {
+			ClusterStateRequest clusterStateRequest = new ClusterStateRequest().indices(expectedIndexName);
+			ClusterStateResponse clusterStateResponse = client.admin().cluster().state(clusterStateRequest).actionGet();
+
+			ImmutableOpenMap<String, IndexMetadata> indicesMetadata = clusterStateResponse.getState().getMetadata().indices();
+			if (!indicesMetadata.containsKey(expectedIndexName)) {
+				mismatchDescription.appendValue("Index does not exist");
+			}
+			IndexMetadata indexMetadata = indicesMetadata.get(expectedIndexName);
+			if (expectedState != indexMetadata.getState()) {
+				mismatchDescription.appendValue("Actual index state is equal to ").appendValue(indexMetadata.getState().name());
+				return false;
+			}
+			return true;
+		}
+	}
+
+	@Override
+	public void describeTo(Description description) {
+		description.appendText("Index: ")
+				.appendValue(expectedIndexName)
+				.appendText(" . State should be equal to ")
+				.appendValue(expectedState.name());
+	}
+}

src/main/java/org/opensearch/security/user/User.java

@@ -66,18 +66,18 @@ public class User implements Serializable, Writeable, CustomAttributesAware {
     /**
      * roles == backend_roles
      */
-    private final Set<String> roles = new HashSet<String>();
-    private final Set<String> openDistroSecurityRoles = new HashSet<String>();
+    private final Set<String> roles = Collections.synchronizedSet(new HashSet<String>());
+    private final Set<String> openDistroSecurityRoles = Collections.synchronizedSet(new HashSet<String>());
     private String requestedTenant;
-    private Map<String, String> attributes = new HashMap<>();
+    private Map<String, String> attributes = Collections.synchronizedMap(new HashMap<>());
     private boolean isInjected = false;
 
     public User(final StreamInput in) throws IOException {
         super();
         name = in.readString();
         roles.addAll(in.readList(StreamInput::readString));
         requestedTenant = in.readString();
-        attributes = in.readMap(StreamInput::readString, StreamInput::readString);
+        attributes = Collections.synchronizedMap(in.readMap(StreamInput::readString, StreamInput::readString));
         openDistroSecurityRoles.addAll(in.readList(StreamInput::readString));
     }
 
@@ -162,7 +162,7 @@ public final boolean isUserInRole(final String role) {
 
     /**
      * Associate this user with a set of backend roles
-     * 
+     *
      * @param roles The backend roles
      */
     public final void addAttributes(final Map<String,String> attributes) {
@@ -254,7 +254,7 @@ public void writeTo(StreamOutput out) throws IOException {
      */
     public synchronized final Map<String, String> getCustomAttributesMap() {
         if(attributes == null) {
-            attributes = new HashMap<>();
+            attributes = Collections.synchronizedMap(new HashMap<>());
         }
         return attributes;
     }
@@ -266,6 +266,6 @@ public final void addSecurityRoles(final Collection<String> securityRoles) {
     }
 
     public final Set<String> getSecurityRoles() {
-        return this.openDistroSecurityRoles == null ? Collections.emptySet() : Collections.unmodifiableSet(this.openDistroSecurityRoles);
+        return this.openDistroSecurityRoles == null ? Collections.synchronizedSet(Collections.emptySet()) : Collections.unmodifiableSet(this.openDistroSecurityRoles);
     }
 }