diff --git a/sample-resource-plugin/src/integrationTest/java/org/opensearch/sample/resource/feature/FeatureFlagSettingTests.java b/sample-resource-plugin/src/integrationTest/java/org/opensearch/sample/resource/feature/FeatureFlagSettingTests.java index eeaf742179..5ea326142b 100644 --- a/sample-resource-plugin/src/integrationTest/java/org/opensearch/sample/resource/feature/FeatureFlagSettingTests.java +++ b/sample-resource-plugin/src/integrationTest/java/org/opensearch/sample/resource/feature/FeatureFlagSettingTests.java @@ -258,7 +258,7 @@ public void testBehaviorAfterEnabling() throws Exception { migrateResponse.assertStatusCode(HttpStatus.SC_OK); assertThat( migrateResponse.bodyAsMap().get("summary"), - equalTo("Migration complete. migrated 1; skippedNoType 0; skippedExisting 0; failed 0") + equalTo("Migration complete. migrated 1; backfilledExisting 0; skippedNoType 0; skippedExisting 0; failed 0") ); } diff --git a/sample-resource-plugin/src/integrationTest/java/org/opensearch/sample/resource/feature/ProtectedTypesSettingTests.java b/sample-resource-plugin/src/integrationTest/java/org/opensearch/sample/resource/feature/ProtectedTypesSettingTests.java index e5aea793d7..e29a3d0c6f 100644 --- a/sample-resource-plugin/src/integrationTest/java/org/opensearch/sample/resource/feature/ProtectedTypesSettingTests.java +++ b/sample-resource-plugin/src/integrationTest/java/org/opensearch/sample/resource/feature/ProtectedTypesSettingTests.java @@ -260,7 +260,7 @@ public void testResourceProtected() throws Exception { migrateResponse.assertStatusCode(HttpStatus.SC_OK); assertThat( migrateResponse.bodyAsMap().get("summary"), - equalTo("Migration complete. migrated 1; skippedNoType 0; skippedExisting 0; failed 0") + equalTo("Migration complete. migrated 1; backfilledExisting 0; skippedNoType 0; skippedExisting 0; failed 0") ); } diff --git a/sample-resource-plugin/src/integrationTest/java/org/opensearch/sample/resource/feature/enabled/ApiAccessTests.java b/sample-resource-plugin/src/integrationTest/java/org/opensearch/sample/resource/feature/enabled/ApiAccessTests.java index ab1aecc8e3..7c504e9e95 100644 --- a/sample-resource-plugin/src/integrationTest/java/org/opensearch/sample/resource/feature/enabled/ApiAccessTests.java +++ b/sample-resource-plugin/src/integrationTest/java/org/opensearch/sample/resource/feature/enabled/ApiAccessTests.java @@ -8,12 +8,14 @@ package org.opensearch.sample.resource.feature.enabled; +import java.time.Duration; import java.util.Map; import java.util.Set; import com.carrotsearch.randomizedtesting.RandomizedRunner; import com.carrotsearch.randomizedtesting.annotations.ThreadLeakScope; import org.apache.http.HttpStatus; +import org.awaitility.Awaitility; import org.junit.After; import org.junit.Before; import org.junit.ClassRule; @@ -28,8 +30,11 @@ import org.opensearch.test.framework.cluster.TestRestClient; import org.opensearch.test.framework.cluster.TestRestClient.HttpResponse; +import tools.jackson.databind.JsonNode; + import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.equalTo; import static org.opensearch.sample.resource.TestUtils.ApiHelper.assertSearchResponse; import static org.opensearch.sample.resource.TestUtils.ApiHelper.searchAllPayload; import static org.opensearch.sample.resource.TestUtils.ApiHelper.searchByNamePayload; @@ -47,6 +52,7 @@ import static org.opensearch.sample.resource.TestUtils.SECURITY_SHARE_ENDPOINT; import static org.opensearch.sample.resource.TestUtils.newCluster; import static org.opensearch.sample.resource.TestUtils.putSharingInfoPayload; +import static org.opensearch.sample.utils.Constants.RESOURCE_INDEX_NAME; import static org.opensearch.sample.utils.Constants.RESOURCE_TYPE; import static org.opensearch.security.api.AbstractApiIntegrationTest.forbidden; import static org.opensearch.security.api.AbstractApiIntegrationTest.ok; @@ -256,6 +262,76 @@ public void testApiAccess_allAccessUser() throws Exception { forbidden(() -> api.deleteResource(adminResId, FULL_ACCESS_USER)); } + @Test + public void testWorkspaceMembership_denyAllowDeny() throws Exception { + // A non-owner, non-shared user sees a resource ONLY while it belongs to a workspace they are a member of. + // FULL_ACCESS_USER's single security role is scoped as user___; SampleResourceExtension + // resolves each security role R to workspace "ws-R", so this is the user's one accessible workspace. + final String userWorkspace = "ws-user_" + FULL_ACCESS_USER.getName() + "__shared_role"; + + // Resource owned by admin, initially in no workspace. FULL_ACCESS_USER is neither owner nor shared-with. + String resId = api.createSampleResourceAs(USER_ADMIN); + api.awaitSharingEntry(resId); + + // DENY: not shared and not in the user's workspace -> the user's search returns no hits. + assertSearchResponse(ok(() -> api.searchResources(FULL_ACCESS_USER)), 0, null); + + // ALLOW: associate the resource with the user's workspace (write the resource doc's workspaces field). The + // index listener reconciles the sharing record to the same set; DLS filters the doc's live field for reads. + setResourceWorkspaces(resId, userWorkspace); + awaitSharingRecordWorkspace(resId, true, userWorkspace); + assertSearchResponse(ok(() -> api.searchResources(FULL_ACCESS_USER)), 1, "sample"); + + // DENY (dissociate): clear the workspace. The listener reconciles the record to empty (removal), so neither + // the read path (DLS) nor the write path (the sharing record's workspace set) retains a stale grant. + setResourceWorkspaces(resId); + awaitSharingRecordWorkspace(resId, false, userWorkspace); + assertSearchResponse(ok(() -> api.searchResources(FULL_ACCESS_USER)), 0, null); + } + + // Sets the resource doc's `workspaces` field to exactly the given ids (empty clears it), as the super admin. + private void setResourceWorkspaces(String resourceId, String... workspaceIds) { + StringBuilder arr = new StringBuilder("["); + for (int i = 0; i < workspaceIds.length; i++) { + if (i > 0) { + arr.append(","); + } + arr.append("\"").append(workspaceIds[i]).append("\""); + } + arr.append("]"); + try (TestRestClient client = cluster.getRestClient(cluster.getAdminCertificate())) { + HttpResponse resp = client.postJson( + RESOURCE_INDEX_NAME + "/_update/" + resourceId + "?refresh=true", + "{\"doc\":{\"workspaces\":" + arr + "}}" + ); + resp.assertStatusCode(HttpStatus.SC_OK); + } + } + + // Waits until the sharing record's `workspaces` set does (or does not) contain the given id, confirming the + // listener reconciled the record to match the resource doc. + private void awaitSharingRecordWorkspace(String resourceId, boolean shouldContain, String workspaceId) { + try (TestRestClient client = cluster.getRestClient(cluster.getAdminCertificate())) { + Awaitility.await("sharing record for " + resourceId + (shouldContain ? " contains " : " excludes ") + workspaceId) + .pollInterval(Duration.ofMillis(500)) + .atMost(Duration.ofSeconds(10)) + .untilAsserted(() -> { + HttpResponse resp = client.get(RESOURCE_SHARING_INDEX + "/_doc/" + resourceId); + resp.assertStatusCode(HttpStatus.SC_OK); + JsonNode ws = resp.bodyAsJsonNode().get("_source").get("workspaces"); + boolean found = false; + if (ws != null && ws.isArray()) { + for (JsonNode n : ws) { + if (workspaceId.equals(n.asString())) { + found = true; + } + } + } + assertThat(found, equalTo(shouldContain)); + }); + } + } + @Test public void testApiAccess_superAdmin() { // can see admin's resource diff --git a/sample-resource-plugin/src/integrationTest/java/org/opensearch/sample/resource/feature/enabled/WorkspaceContainerAccessTests.java b/sample-resource-plugin/src/integrationTest/java/org/opensearch/sample/resource/feature/enabled/WorkspaceContainerAccessTests.java new file mode 100644 index 0000000000..9c1fa75f79 --- /dev/null +++ b/sample-resource-plugin/src/integrationTest/java/org/opensearch/sample/resource/feature/enabled/WorkspaceContainerAccessTests.java @@ -0,0 +1,290 @@ +/* + * 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.sample.resource.feature.enabled; + +import java.time.Duration; +import java.util.List; + +import com.carrotsearch.randomizedtesting.RandomizedRunner; +import com.carrotsearch.randomizedtesting.annotations.ThreadLeakScope; +import org.apache.http.HttpStatus; +import org.awaitility.Awaitility; +import org.junit.After; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.opensearch.sample.resource.TestUtils; +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 tools.jackson.databind.JsonNode; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.equalTo; +import static org.opensearch.sample.resource.TestUtils.FULL_ACCESS_USER; +import static org.opensearch.sample.resource.TestUtils.RESOURCE_SHARING_INDEX; +import static org.opensearch.sample.resource.TestUtils.SAMPLE_READ_ONLY; +import static org.opensearch.sample.resource.TestUtils.SAMPLE_RESOURCE_CREATE_ENDPOINT; +import static org.opensearch.sample.resource.TestUtils.SAMPLE_RESOURCE_UPDATE_ENDPOINT; +import static org.opensearch.sample.resource.TestUtils.newCluster; +import static org.opensearch.sample.utils.Constants.RESOURCE_INDEX_NAME; +import static org.opensearch.sample.utils.Constants.RESOURCE_TYPE; +import static org.opensearch.sample.utils.Constants.WORKSPACE_TYPE; +import static org.opensearch.security.api.AbstractApiIntegrationTest.forbidden; +import static org.opensearch.security.api.AbstractApiIntegrationTest.ok; +import static org.opensearch.test.framework.TestSecurityConfig.User.USER_ADMIN; + +/** + * Exercises the write-path container fan-out: a user with no direct grant can act on a resource because it belongs to a + * workspace that shares access with them, and loses that access the moment the resource is dissociated. The cluster + * registers {@code workspace} as a protected type so {@code checkContainers} resolves the workspace's sharing record. + */ +@RunWith(RandomizedRunner.class) +@ThreadLeakScope(ThreadLeakScope.Scope.NONE) +public class WorkspaceContainerAccessTests { + + @ClassRule + public static LocalCluster cluster = newCluster(true, true, List.of(RESOURCE_TYPE, WORKSPACE_TYPE)); + + private final TestUtils.ApiHelper api = new TestUtils.ApiHelper(cluster); + + @After + public void cleanup() { + api.wipeOutResourceEntries(); + } + + @Test + public void testWorkspaceContainerGrantsAndRevokesDirectAction() throws Exception { + final String workspaceId = "ws-team"; + + // A resource owned by admin. FULL_ACCESS_USER is neither owner nor shared-with. + String resId = api.createSampleResourceAs(USER_ADMIN); + api.awaitSharingEntry(resId); + + // A workspace whose record shares read access with FULL_ACCESS_USER (workspace_read_only -> sampleresource:get). + putWorkspaceSharingRecord(workspaceId, "workspace_read_only", FULL_ACCESS_USER.getName()); + + // DENY: no direct grant and not yet a member -> the action-level check (hasPermission) forbids GET. + forbidden(() -> api.getResource(resId, FULL_ACCESS_USER)); + + // ALLOW: associate the resource with the workspace. The listener reconciles the sharing record's workspace set; + // hasPermission then inherits the action from the workspace container. + setResourceWorkspaces(resId, workspaceId); + awaitSharingRecordWorkspace(resId, true, workspaceId); + HttpResponse getResp = ok(() -> api.getResource(resId, FULL_ACCESS_USER)); + assertThat(getResp.getBody(), containsString("sample")); + + // DENY (dissociate): clear the workspace. The record reconciles to empty (removal), so the container grant is + // gone -- no stale write authorization. + setResourceWorkspaces(resId); + awaitSharingRecordWorkspace(resId, false, workspaceId); + forbidden(() -> api.getResource(resId, FULL_ACCESS_USER)); + } + + @Test + public void testRapidAssociateThenDissociateConvergesToDenied() throws Exception { + // Negative control for out-of-order reconciliation: associate then immediately dissociate WITHOUT awaiting the + // association reconcile. The monotonic guard must ensure the slower association reconcile can never overwrite + // the newer dissociation, so the resource converges to no-workspace and the direct action is denied. + final String workspaceId = "ws-race"; + String resId = api.createSampleResourceAs(USER_ADMIN); + api.awaitSharingEntry(resId); + putWorkspaceSharingRecord(workspaceId, "workspace_read_only", FULL_ACCESS_USER.getName()); + + setResourceWorkspaces(resId, workspaceId); // associate + setResourceWorkspaces(resId); // dissociate immediately, without awaiting the first reconcile + + awaitSharingRecordWorkspace(resId, false, workspaceId); + forbidden(() -> api.getResource(resId, FULL_ACCESS_USER)); + } + + @Test + public void testCreateWithWorkspaceThenImmediateClearConvergesToDenied() throws Exception { + // Negative control for the create race: create a resource already associated with a workspace, then clear it + // immediately. A reconcile that finds the record not yet written must retry (not no-op), and the clear must + // win, so the resource converges to no-workspace and the direct action is denied. + final String workspaceId = "ws-race-create"; + putWorkspaceSharingRecord(workspaceId, "workspace_read_only", FULL_ACCESS_USER.getName()); + + String resId = createResourceWithWorkspacesAs(USER_ADMIN, workspaceId); + setResourceWorkspaces(resId); // clear immediately + + awaitSharingRecordWorkspace(resId, false, workspaceId); + forbidden(() -> api.getResource(resId, FULL_ACCESS_USER)); + } + + @Test + public void testOrdinaryUpdateCannotChangeWorkspaceMembership() throws Exception { + // Trusted-write contract: workspace membership is server-controlled. A user with workspace_read_write (so it + // can update) must not be able to add another workspace through an ordinary update to acquire that workspace's + // stronger access level. The sample update route ignores caller-supplied workspaces. + final String teamWs = "ws-team-rw"; + final String superWs = "ws-super"; + putWorkspaceSharingRecord(teamWs, "workspace_read_write", FULL_ACCESS_USER.getName()); + putWorkspaceSharingRecord(superWs, "workspace_full_access", FULL_ACCESS_USER.getName()); + + String resId = api.createSampleResourceAs(USER_ADMIN); + api.awaitSharingEntry(resId); + // Associate only with ws-team-rw through the server-authorized path. + setResourceWorkspaces(resId, teamWs); + awaitSharingRecordWorkspace(resId, true, teamWs); + + // FULL_ACCESS_USER (workspace_read_write -> can update) tries to add ws-super via an ordinary update. + HttpResponse update = updateResourceWithWorkspacesAs(resId, FULL_ACCESS_USER, "escalate", teamWs, superWs); + update.assertStatusCode(HttpStatus.SC_OK); + + // Membership is unchanged: ws-super was not added, so no escalation to full_access occurred. + awaitSharingRecordWorkspace(resId, false, superWs); + awaitSharingRecordWorkspace(resId, true, teamWs); + } + + // Creates a resource already carrying the given workspaces (create route accepts them; the owner still governs the + // record). Returns the new resource id. + private String createResourceWithWorkspacesAs(TestSecurityConfig.User user, String... workspaceIds) { + String body = "{\"name\":\"sample\",\"resource_type\":\"" + RESOURCE_TYPE + "\",\"workspaces\":" + jsonArray(workspaceIds) + "}"; + try (TestRestClient client = cluster.getRestClient(user)) { + HttpResponse resp = client.putJson(SAMPLE_RESOURCE_CREATE_ENDPOINT, body); + resp.assertStatusCode(HttpStatus.SC_OK); + return resp.getTextFromJsonBody("/message").split(":")[1].trim(); + } + } + + // Attempts an ordinary update carrying a caller-supplied workspaces field (used to prove it is ignored). + private HttpResponse updateResourceWithWorkspacesAs( + String resourceId, + TestSecurityConfig.User user, + String newName, + String... workspaceIds + ) { + String body = "{\"name\":\"" + newName + "\",\"workspaces\":" + jsonArray(workspaceIds) + "}"; + try (TestRestClient client = cluster.getRestClient(user)) { + return client.postJson(SAMPLE_RESOURCE_UPDATE_ENDPOINT + "/" + resourceId, body); + } + } + + private static String jsonArray(String... values) { + StringBuilder arr = new StringBuilder("["); + for (int i = 0; i < values.length; i++) { + if (i > 0) { + arr.append(","); + } + arr.append("\"").append(values[i]).append("\""); + } + return arr.append("]").toString(); + } + + @Test + public void testGuardSurvivesShare() throws Exception { + // share()/revoke()/patch() rewrite the whole sharing record. The monotonic guard (workspaces_seq_no) MUST + // survive that rewrite, otherwise a share would reset it and let an older, still-retrying reconcile re-apply + // stale workspaces. + final String workspaceId = "ws-guard"; + putWorkspaceSharingRecord(workspaceId, "workspace_read_only", FULL_ACCESS_USER.getName()); + String resId = api.createSampleResourceAs(USER_ADMIN); + api.awaitSharingEntry(resId); + + setResourceWorkspaces(resId, workspaceId); + awaitSharingRecordWorkspace(resId, true, workspaceId); + long guardBefore = readWorkspacesSeqNo(resId); + assertThat(guardBefore >= 0L, equalTo(true)); // reconcile stamped a real seq_no + + // Share the resource (as its owner) -> the record is re-indexed via toXContent. + ok(() -> api.shareResource(resId, USER_ADMIN, FULL_ACCESS_USER, SAMPLE_READ_ONLY)); + + // The guard must NOT be reset by the rewrite (it may advance if the share refreshes principals, but never + // drops back to unassigned -- a reset would let an older reconcile re-apply stale workspaces). Membership + // is intact. + long guardAfter = readWorkspacesSeqNo(resId); + assertThat("guard must not reset below its prior value", guardAfter >= guardBefore, equalTo(true)); + awaitSharingRecordWorkspace(resId, true, workspaceId); + } + + private long readWorkspacesSeqNo(String resourceId) { + try (TestRestClient client = cluster.getRestClient(cluster.getAdminCertificate())) { + HttpResponse resp = client.get(RESOURCE_SHARING_INDEX + "/_doc/" + resourceId); + resp.assertStatusCode(HttpStatus.SC_OK); + JsonNode seq = resp.bodyAsJsonNode().get("_source").get("workspaces_seq_no"); + return seq == null ? -2L : seq.asLong(); + } + } + + // Writes a workspace sharing record directly (mirrors how a real workspace backend materializes collaborators), + // sharing the given access level with the given user. + private void putWorkspaceSharingRecord(String workspaceId, String accessLevel, String username) { + String record = "{" + + "\"resource_id\":\"" + + workspaceId + + "\"," + + "\"resource_type\":\"" + + WORKSPACE_TYPE + + "\"," + + "\"created_by\":{\"user\":\"" + + USER_ADMIN.getName() + + "\"}," + + "\"share_with\":{\"" + + accessLevel + + "\":{\"users\":[\"" + + username + + "\"]}}" + + "}"; + try (TestRestClient client = cluster.getRestClient(cluster.getAdminCertificate())) { + HttpResponse resp = client.putJson(RESOURCE_SHARING_INDEX + "/_doc/" + workspaceId + "?refresh=true", record); + assertThat(resp.getStatusCode(), equalTo(HttpStatus.SC_CREATED)); + } + } + + // Sets the resource doc's `workspaces` field to exactly the given ids (empty clears it), as the super admin. + private void setResourceWorkspaces(String resourceId, String... workspaceIds) { + StringBuilder arr = new StringBuilder("["); + for (int i = 0; i < workspaceIds.length; i++) { + if (i > 0) { + arr.append(","); + } + arr.append("\"").append(workspaceIds[i]).append("\""); + } + arr.append("]"); + try (TestRestClient client = cluster.getRestClient(cluster.getAdminCertificate())) { + HttpResponse resp = client.postJson( + RESOURCE_INDEX_NAME + "/_update/" + resourceId + "?refresh=true", + "{\"doc\":{\"workspaces\":" + arr + "}}" + ); + resp.assertStatusCode(HttpStatus.SC_OK); + } + } + + // Waits until the sharing record's `workspaces` set does (or does not) contain the given id. + private void awaitSharingRecordWorkspace(String resourceId, boolean shouldContain, String workspaceId) { + try (TestRestClient client = cluster.getRestClient(cluster.getAdminCertificate())) { + Awaitility.await("sharing record for " + resourceId + (shouldContain ? " contains " : " excludes ") + workspaceId) + .pollInterval(Duration.ofMillis(500)) + .atMost(Duration.ofSeconds(10)) + .untilAsserted(() -> { + HttpResponse resp = client.get(RESOURCE_SHARING_INDEX + "/_doc/" + resourceId); + resp.assertStatusCode(HttpStatus.SC_OK); + JsonNode ws = resp.bodyAsJsonNode().get("_source").get("workspaces"); + boolean found = false; + if (ws != null && ws.isArray()) { + for (JsonNode n : ws) { + if (workspaceId.equals(n.asString())) { + found = true; + } + } + } + assertThat(found, equalTo(shouldContain)); + }); + } + } +} diff --git a/sample-resource-plugin/src/integrationTest/java/org/opensearch/sample/resource/securityapis/MigrateApiTests.java b/sample-resource-plugin/src/integrationTest/java/org/opensearch/sample/resource/securityapis/MigrateApiTests.java index 955243fc0f..b31838eecd 100644 --- a/sample-resource-plugin/src/integrationTest/java/org/opensearch/sample/resource/securityapis/MigrateApiTests.java +++ b/sample-resource-plugin/src/integrationTest/java/org/opensearch/sample/resource/securityapis/MigrateApiTests.java @@ -33,6 +33,7 @@ import org.opensearch.test.framework.cluster.TestRestClient; import org.opensearch.test.framework.matcher.RestMatchers; +import tools.jackson.databind.JsonNode; import tools.jackson.databind.ObjectMapper; import tools.jackson.databind.node.ArrayNode; import tools.jackson.databind.node.ObjectNode; @@ -180,7 +181,7 @@ public void testMigrateAPIWithRestAdmin_valid() { migrateResponse.assertStatusCode(HttpStatus.SC_OK); assertThat( migrateResponse.bodyAsMap().get("summary"), - equalTo("Migration complete. migrated 2; skippedNoType 0; skippedExisting 0; failed 0") + equalTo("Migration complete. migrated 2; backfilledExisting 0; skippedNoType 0; skippedExisting 0; failed 0") ); assertThat(migrateResponse.bodyAsMap().get("resourcesWithDefaultOwner"), equalTo(List.of(resourceIdNoUser))); } @@ -193,7 +194,7 @@ public void testMigrateAPIWithRestAdmin_valid() { assertThat(hitsNode.size(), equalTo(2)); List actualHits = new ArrayList<>(); - hitsNode.forEach(node -> actualHits.add((ObjectNode) node)); + hitsNode.forEach(node -> actualHits.add(stripReconcileFields((ObjectNode) node))); // with custom access level, order-agnostic assertThat( @@ -215,7 +216,7 @@ public void testMigrateAPIWithSuperAdmin_valid() { migrateResponse.assertStatusCode(HttpStatus.SC_OK); assertThat( migrateResponse.bodyAsMap().get("summary"), - equalTo("Migration complete. migrated 2; skippedNoType 0; skippedExisting 0; failed 0") + equalTo("Migration complete. migrated 2; backfilledExisting 0; skippedNoType 0; skippedExisting 0; failed 0") ); assertThat(migrateResponse.bodyAsMap().get("resourcesWithDefaultOwner"), equalTo(List.of(resourceIdNoUser))); @@ -225,7 +226,7 @@ public void testMigrateAPIWithSuperAdmin_valid() { assertThat(hitsNode.size(), equalTo(2)); List actualHits = new ArrayList<>(); - hitsNode.forEach(node -> actualHits.add((ObjectNode) node)); + hitsNode.forEach(node -> actualHits.add(stripReconcileFields((ObjectNode) node))); // with custom access level, order-agnostic assertThat( @@ -247,7 +248,7 @@ public void testMigrateTwice_shouldSkipSecondTime() { migrateResponse.assertStatusCode(HttpStatus.SC_OK); assertThat( migrateResponse.bodyAsMap().get("summary"), - equalTo("Migration complete. migrated 2; skippedNoType 0; skippedExisting 0; failed 0") + equalTo("Migration complete. migrated 2; backfilledExisting 0; skippedNoType 0; skippedExisting 0; failed 0") ); assertThat(migrateResponse.bodyAsMap().get("resourcesWithDefaultOwner"), equalTo(List.of(resourceIdNoUser))); @@ -257,7 +258,7 @@ public void testMigrateTwice_shouldSkipSecondTime() { assertThat(hitsNode.size(), equalTo(2)); final List actualHits = new ArrayList<>(); - hitsNode.forEach(node -> actualHits.add((ObjectNode) node)); + hitsNode.forEach(node -> actualHits.add(stripReconcileFields((ObjectNode) node))); // with custom access level, order-agnostic assertThat( @@ -273,7 +274,7 @@ public void testMigrateTwice_shouldSkipSecondTime() { migrateResponse.assertStatusCode(HttpStatus.SC_OK); assertThat( migrateResponse.bodyAsMap().get("summary"), - equalTo("Migration complete. migrated 0; skippedNoType 0; skippedExisting 2; failed 0") + equalTo("Migration complete. migrated 0; backfilledExisting 0; skippedNoType 0; skippedExisting 2; failed 0") ); assertThat(migrateResponse.bodyAsMap().get("resourcesWithDefaultOwner"), equalTo(List.of(resourceIdNoUser))); @@ -283,7 +284,7 @@ public void testMigrateTwice_shouldSkipSecondTime() { assertThat(hitsNode.size(), equalTo(2)); final List finalActualHits = new ArrayList<>(); - hitsNode.forEach(node -> finalActualHits.add((ObjectNode) node)); + hitsNode.forEach(node -> finalActualHits.add(stripReconcileFields((ObjectNode) node))); // default access-level should not have been updated as record was already migrated assertThat( @@ -294,6 +295,114 @@ public void testMigrateTwice_shouldSkipSecondTime() { } } + @Test + public void testLiveIndexingStampsWorkspacesOnSharingRecord() { + // Creating a resource with a workspaces field stores those workspaces on the sharing record (used by the + // write-path access-level fan-out). all_shared_principals stays usernames/roles only; read-path visibility + // filters the resource's own workspaces field in DLS. + String resourceId = createSampleResourceWithWorkspaces("ws-a", "ws-b"); + + try (TestRestClient client = cluster.getRestClient(cluster.getAdminCertificate())) { + // The sharing record carries the workspaces field (for the write-path fan-out). + TestRestClient.HttpResponse sharingDoc = client.get(RESOURCE_SHARING_INDEX + "/_doc/" + resourceId); + sharingDoc.assertStatusCode(HttpStatus.SC_OK); + ArrayNode ws = (ArrayNode) sharingDoc.bodyAsJsonNode().get("_source").get("workspaces"); + List workspaceIds = new ArrayList<>(); + ws.forEach(n -> workspaceIds.add(n.asString())); + assertThat(workspaceIds, containsInAnyOrder("ws-a", "ws-b")); + + // all_shared_principals is seeded onto the resource doc asynchronously after the sharing record is + // created, so poll until it is present. It stays usernames/roles only -- no workspace denormalization. + Awaitility.await("all_shared_principals seeded on resource doc").untilAsserted(() -> { + TestRestClient.HttpResponse resourceDoc = client.get(RESOURCE_INDEX_NAME + "/_doc/" + resourceId); + resourceDoc.assertStatusCode(HttpStatus.SC_OK); + ArrayNode principals = (ArrayNode) resourceDoc.bodyAsJsonNode().get("_source").get("all_shared_principals"); + List principalList = new ArrayList<>(); + if (principals != null) { + principals.forEach(n -> principalList.add(n.asString())); + } + assertThat(principalList, containsInAnyOrder("user:" + MIGRATION_USER.getName())); + + // The resource doc keeps its own `workspaces` field -- this is what DLS filters on for read visibility. + ArrayNode docWs = (ArrayNode) resourceDoc.bodyAsJsonNode().get("_source").get("workspaces"); + List docWorkspaceIds = new ArrayList<>(); + docWs.forEach(n -> docWorkspaceIds.add(n.asString())); + assertThat(docWorkspaceIds, containsInAnyOrder("ws-a", "ws-b")); + }); + } + } + + @Test + public void testMigrateBackfillsWorkspacesOntoExistingRecord() { + // A pre-existing sharing record that is out of sync with its source doc (e.g. written while the feature was + // off, so the listener never reconciled it). Migration must not re-create the record; it must reconcile the + // record's workspaces to exactly match the source doc -- adding the doc's workspaces and removing stale ones. + String resourceId = createSampleResource(); + + try (TestRestClient client = cluster.getRestClient(cluster.getAdminCertificate())) { + // Put the target workspaces on the source doc. This _update fires the listener, which reconciles the + // record to [ws-a, ws-b]; wait for that so the next step starts from a known state. + TestRestClient.HttpResponse update = client.postJson( + RESOURCE_INDEX_NAME + "/_update/" + resourceId + "?refresh=true", + "{ \"doc\": { \"workspaces\": [\"ws-a\", \"ws-b\"] } }" + ); + update.assertStatusCode(HttpStatus.SC_OK); + Awaitility.await("listener reconciles record to the doc's workspaces").untilAsserted(() -> { + TestRestClient.HttpResponse rec = client.get(RESOURCE_SHARING_INDEX + "/_doc/" + resourceId); + rec.assertStatusCode(HttpStatus.SC_OK); + List recWs = new ArrayList<>(); + ArrayNode arr = (ArrayNode) rec.bodyAsJsonNode().get("_source").get("workspaces"); + if (arr != null) { + arr.forEach(n -> recWs.add(n.asString())); + } + assertThat(recWs, containsInAnyOrder("ws-a", "ws-b")); + }); + + // Now force the record out of sync by writing a stale set directly to the sharing index (no listener runs + // on the sharing index), and reset the monotonic guard (workspaces_seq_no) to a low watermark so it + // resembles a pre-feature record that migration must reconcile to the source doc. Record: [ws-stale]; + // source doc: [ws-a, ws-b]. + TestRestClient.HttpResponse stale = client.postJson( + RESOURCE_SHARING_INDEX + "/_update/" + resourceId + "?refresh=true", + "{ \"doc\": { \"workspaces\": [\"ws-stale\"], \"workspaces_seq_no\": -2 } }" + ); + stale.assertStatusCode(HttpStatus.SC_OK); + + // Migrate: the record exists, so create is skipped; its workspaces differ from the source doc, so it is + // reconciled (reported as backfilledExisting) rather than skipped. + TestRestClient.HttpResponse migrateResponse = client.postJson(RESOURCE_SHARING_MIGRATION_ENDPOINT, migrationPayload_valid()); + migrateResponse.assertStatusCode(HttpStatus.SC_OK); + assertThat( + migrateResponse.bodyAsMap().get("summary"), + equalTo("Migration complete. migrated 0; backfilledExisting 1; skippedNoType 0; skippedExisting 0; failed 0") + ); + + // The sharing record now matches the source doc exactly -- ws-stale removed, ws-a/ws-b present. + TestRestClient.HttpResponse sharingDoc = client.get(RESOURCE_SHARING_INDEX + "/_doc/" + resourceId); + sharingDoc.assertStatusCode(HttpStatus.SC_OK); + ArrayNode ws = (ArrayNode) sharingDoc.bodyAsJsonNode().get("_source").get("workspaces"); + List workspaceIds = new ArrayList<>(); + ws.forEach(n -> workspaceIds.add(n.asString())); + assertThat(workspaceIds, containsInAnyOrder("ws-a", "ws-b")); + + // all_shared_principals stays usernames/roles only -- workspace membership is not denormalized here. + TestRestClient.HttpResponse resourceDoc = client.get(RESOURCE_INDEX_NAME + "/_doc/" + resourceId); + resourceDoc.assertStatusCode(HttpStatus.SC_OK); + ArrayNode principals = (ArrayNode) resourceDoc.bodyAsJsonNode().get("_source").get("all_shared_principals"); + List principalList = new ArrayList<>(); + principals.forEach(n -> principalList.add(n.asString())); + assertThat(principalList, containsInAnyOrder("user:" + MIGRATION_USER.getName())); + + // Idempotency: a second migrate with the same workspaces adds nothing new (skipped, not backfilled). + TestRestClient.HttpResponse secondMigrate = client.postJson(RESOURCE_SHARING_MIGRATION_ENDPOINT, migrationPayload_valid()); + secondMigrate.assertStatusCode(HttpStatus.SC_OK); + assertThat( + secondMigrate.bodyAsMap().get("summary"), + equalTo("Migration complete. migrated 0; backfilledExisting 0; skippedNoType 0; skippedExisting 1; failed 0") + ); + } + } + @Test public void testMigrateAPIWithSuperAdmin_valid_withSpecifiedAccessLevel() { String resourceId = createSampleResource(); @@ -308,7 +417,7 @@ public void testMigrateAPIWithSuperAdmin_valid_withSpecifiedAccessLevel() { migrateResponse.assertStatusCode(HttpStatus.SC_OK); assertThat( migrateResponse.bodyAsMap().get("summary"), - equalTo("Migration complete. migrated 2; skippedNoType 0; skippedExisting 0; failed 0") + equalTo("Migration complete. migrated 2; backfilledExisting 0; skippedNoType 0; skippedExisting 0; failed 0") ); assertThat(migrateResponse.bodyAsMap().get("resourcesWithDefaultOwner"), equalTo(List.of(resourceIdNoUser))); @@ -318,7 +427,7 @@ public void testMigrateAPIWithSuperAdmin_valid_withSpecifiedAccessLevel() { assertThat(hitsNode.size(), equalTo(2)); List actualHits = new ArrayList<>(); - hitsNode.forEach(node -> actualHits.add((ObjectNode) node)); + hitsNode.forEach(node -> actualHits.add(stripReconcileFields((ObjectNode) node))); // with default access level, order-agnostic assertThat( @@ -395,7 +504,7 @@ public void testMigrateAPIWithSuperAdmin_noDefaultAccessLevel_usesRegisteredDefa migrateResponse.assertStatusCode(HttpStatus.SC_OK); assertThat( migrateResponse.bodyAsMap().get("summary"), - equalTo("Migration complete. migrated 2; skippedNoType 0; skippedExisting 0; failed 0") + equalTo("Migration complete. migrated 2; backfilledExisting 0; skippedNoType 0; skippedExisting 0; failed 0") ); TestRestClient.HttpResponse sharingResponse = client.get(RESOURCE_SHARING_INDEX + "/_search"); @@ -404,7 +513,7 @@ public void testMigrateAPIWithSuperAdmin_noDefaultAccessLevel_usesRegisteredDefa assertThat(hitsNode.size(), equalTo(2)); List actualHits = new ArrayList<>(); - hitsNode.forEach(node -> actualHits.add((ObjectNode) node)); + hitsNode.forEach(node -> actualHits.add(stripReconcileFields((ObjectNode) node))); // registered default is sample_read_only assertThat( @@ -645,7 +754,7 @@ public void testMigrateAPI_withGarbageParentId() { migrateResponse.assertStatusCode(HttpStatus.SC_OK); assertThat( migrateResponse.bodyAsMap().get("summary"), - equalTo("Migration complete. migrated 1; skippedNoType 0; skippedExisting 0; failed 0") + equalTo("Migration complete. migrated 1; backfilledExisting 0; skippedNoType 0; skippedExisting 0; failed 0") ); // The sharing record should be created with the garbage parent_id stored as-is @@ -681,7 +790,7 @@ public void testMigrateAPI_withParentHierarchy() { migrateResponse.assertStatusCode(HttpStatus.SC_OK); assertThat( migrateResponse.bodyAsMap().get("summary"), - equalTo("Migration complete. migrated 2; skippedNoType 0; skippedExisting 0; failed 0") + equalTo("Migration complete. migrated 2; backfilledExisting 0; skippedNoType 0; skippedExisting 0; failed 0") ); // Verify the sharing record for the resource has parent_type and parent_id set @@ -762,6 +871,26 @@ private String createSampleResource() { } } + private String createSampleResourceWithWorkspaces(String... workspaceIds) { + try (TestRestClient client = cluster.getRestClient(MIGRATION_USER)) { + StringBuilder wsArray = new StringBuilder("["); + for (int i = 0; i < workspaceIds.length; i++) { + if (i > 0) wsArray.append(","); + wsArray.append("\"").append(workspaceIds[i]).append("\""); + } + wsArray.append("]"); + String sampleResource = ("{\"name\":\"sample_ws\",\"store_user\":true,\"workspaces\":" + wsArray + "}"); + + TestRestClient.HttpResponse response = client.putJson(SAMPLE_RESOURCE_CREATE_ENDPOINT, sampleResource); + response.assertStatusCode(HttpStatus.SC_OK); + String resourceId = response.getTextFromJsonBody("/message").split(":")[1].trim(); + Awaitility.await() + .alias("Wait until resource with workspaces is populated") + .until(() -> client.get(SAMPLE_RESOURCE_GET_ENDPOINT + "/" + resourceId).getStatusCode(), equalTo(200)); + return resourceId; + } + } + private String createSampleResourceNoUser() { try (TestRestClient client = cluster.getRestClient(MIGRATION_USER)) { String sampleResource = """ @@ -794,7 +923,12 @@ private void clearResourceSharingEntries() { } } """; - TestRestClient.HttpResponse response = client.postJson(RESOURCE_SHARING_INDEX + "/_delete_by_query?refresh=true", deleteBody); + // conflicts=proceed: an async workspace-reconcile write may touch a record mid-delete; skip the conflict + // rather than fail (the whole index is dropped next anyway). + TestRestClient.HttpResponse response = client.postJson( + RESOURCE_SHARING_INDEX + "/_delete_by_query?refresh=true&conflicts=proceed", + deleteBody + ); response.assertStatusCode(HttpStatus.SC_OK); @@ -802,6 +936,18 @@ private void clearResourceSharingEntries() { } } + // The workspace-reconcile listener may asynchronously stamp `workspaces` (empty for non-workspace resources) and + // the monotonic-guard `workspaces_seq_no` onto sharing records. These are reconciliation metadata, not sharing + // content, so strip them before comparing records by value against expectedHits(). + private static ObjectNode stripReconcileFields(ObjectNode hit) { + JsonNode src = hit.get("_source"); + if (src instanceof ObjectNode source) { + source.remove("workspaces"); + source.remove("workspaces_seq_no"); + } + return hit; + } + private List expectedHits(String resourceId, String resourceIdNoUser, String accessLevel) { ObjectMapper mapper = new ObjectMapper(); diff --git a/sample-resource-plugin/src/integrationTest/java/org/opensearch/sample/resource/securityapis/ShareableResourceTypesInfoApiTests.java b/sample-resource-plugin/src/integrationTest/java/org/opensearch/sample/resource/securityapis/ShareableResourceTypesInfoApiTests.java index 87376949df..eb1b9d9db0 100644 --- a/sample-resource-plugin/src/integrationTest/java/org/opensearch/sample/resource/securityapis/ShareableResourceTypesInfoApiTests.java +++ b/sample-resource-plugin/src/integrationTest/java/org/opensearch/sample/resource/securityapis/ShareableResourceTypesInfoApiTests.java @@ -8,6 +8,7 @@ package org.opensearch.sample.resource.securityapis; +import java.util.HashMap; import java.util.List; import java.util.Map; @@ -55,19 +56,27 @@ public void testTypesApi_mustListSampleResourceAsAType() { TestRestClient.HttpResponse response = client.get(SECURITY_TYPES_ENDPOINT); response.assertStatusCode(HttpStatus.SC_OK); List types = (List) response.bodyAsMap().get("types"); - assertThat(types.size(), equalTo(2)); - Map firstType = (Map) types.get(0); - assertThat(firstType.get("type"), equalTo("sample-resource")); + // sample-resource, its resource-group container, and the workspace container are all registered types. + assertThat(types.size(), equalTo(3)); + + Map> accessLevelsByType = new HashMap<>(); + for (Object t : types) { + Map type = (Map) t; + accessLevelsByType.put((String) type.get("type"), (List) type.get("access_levels")); + } + assertThat(accessLevelsByType.keySet(), containsInAnyOrder("sample-resource", "sample-resource-group", "workspace")); assertThat( - (List) firstType.get("access_levels"), + accessLevelsByType.get("sample-resource"), containsInAnyOrder("sample_read_only", "sample_read_write", "sample_full_access") ); - Map secondType = (Map) types.get(1); - assertThat(secondType.get("type"), equalTo("sample-resource-group")); assertThat( - (List) secondType.get("access_levels"), + accessLevelsByType.get("sample-resource-group"), containsInAnyOrder("sample_group_read_only", "sample_group_read_write", "sample_group_full_access") ); + assertThat( + accessLevelsByType.get("workspace"), + containsInAnyOrder("workspace_read_only", "workspace_read_write", "workspace_full_access") + ); } } diff --git a/sample-resource-plugin/src/main/java/org/opensearch/sample/SampleResource.java b/sample-resource-plugin/src/main/java/org/opensearch/sample/SampleResource.java index c80ba7bda7..5d523b3995 100644 --- a/sample-resource-plugin/src/main/java/org/opensearch/sample/SampleResource.java +++ b/sample-resource-plugin/src/main/java/org/opensearch/sample/SampleResource.java @@ -12,7 +12,10 @@ package org.opensearch.sample; import java.io.IOException; +import java.util.HashSet; +import java.util.List; import java.util.Map; +import java.util.Set; import org.opensearch.commons.authuser.User; import org.opensearch.core.ParseField; @@ -40,6 +43,9 @@ public class SampleResource implements NamedWriteable, ToXContentObject { private Map attributes; // NOTE: following field is added to specifically test migrate API, for newer resources this field must not be defined private User user; + // Workspace membership; optional, models the multi-valued "workspaces" field a real workspace-aware resource + // would declare. ResourceIndexListener stamps it onto the sharing record; DLS filters on it for read visibility. + private Set workspaces; public SampleResource() throws IOException { super(); @@ -51,6 +57,8 @@ public SampleResource(StreamInput in) throws IOException { this.groupId = in.readOptionalString(); this.attributes = in.readMap(StreamInput::readString, StreamInput::readString); this.user = new User(in); + List ws = in.readOptionalStringList(); + this.workspaces = ws == null ? null : new HashSet<>(ws); } private static final ConstructingObjectParser PARSER = new ConstructingObjectParser<>(RESOURCE_TYPE, true, a -> { @@ -67,6 +75,10 @@ public SampleResource(StreamInput in) throws IOException { // ignore a[3] as we know the type s.setAttributes((Map) a[4]); s.setUser((User) a[5]); + List ws = (List) a[6]; + if (ws != null) { + s.setWorkspaces(new HashSet<>(ws)); + } return s; }); @@ -77,6 +89,7 @@ public SampleResource(StreamInput in) throws IOException { PARSER.declareStringOrNull(optionalConstructorArg(), new ParseField("resource_type")); PARSER.declareObjectOrNull(optionalConstructorArg(), (p, c) -> p.mapStrings(), null, new ParseField("attributes")); PARSER.declareObjectOrNull(optionalConstructorArg(), (p, c) -> User.parse(p), null, new ParseField("user")); + PARSER.declareStringArray(optionalConstructorArg(), new ParseField("workspaces")); } public static SampleResource fromXContent(XContentParser parser) throws IOException { @@ -84,14 +97,18 @@ public static SampleResource fromXContent(XContentParser parser) throws IOExcept } public XContentBuilder toXContent(XContentBuilder builder, ToXContent.Params params) throws IOException { - return builder.startObject() + builder.startObject() .field("name", name) .field("description", description) .field("group_id", groupId) .field("resource_type", RESOURCE_TYPE) .field("attributes", attributes) - .field("user", user) - .endObject(); + .field("user", user); + // Emit workspaces only when set, so docs without it are unchanged. + if (workspaces != null && !workspaces.isEmpty()) { + builder.field("workspaces", workspaces); + } + return builder.endObject(); } public void writeTo(StreamOutput out) throws IOException { @@ -100,6 +117,8 @@ public void writeTo(StreamOutput out) throws IOException { out.writeOptionalString(groupId); out.writeMap(attributes, StreamOutput::writeString, StreamOutput::writeString); user.writeTo(out); + // Symmetric with the StreamInput ctor. Passing null when unset keeps mixed-caller compatibility. + out.writeOptionalStringCollection(workspaces); } public void setName(String name) { @@ -122,6 +141,14 @@ public void setUser(User user) { this.user = user; } + public void setWorkspaces(Set workspaces) { + this.workspaces = workspaces; + } + + public Set getWorkspaces() { + return workspaces; + } + public String getName() { return name; } diff --git a/sample-resource-plugin/src/main/java/org/opensearch/sample/SampleResourceExtension.java b/sample-resource-plugin/src/main/java/org/opensearch/sample/SampleResourceExtension.java index 7678589f90..bca49741c8 100644 --- a/sample-resource-plugin/src/main/java/org/opensearch/sample/SampleResourceExtension.java +++ b/sample-resource-plugin/src/main/java/org/opensearch/sample/SampleResourceExtension.java @@ -11,6 +11,8 @@ package org.opensearch.sample; +import java.util.Collections; +import java.util.HashSet; import java.util.Set; import org.opensearch.sample.client.ResourceSharingClientAccessor; @@ -54,6 +56,7 @@ public String parentType() { public String parentIdField() { return "group_id"; } + // workspacesField() defaults to "workspaces" — no override needed. }); } @@ -61,4 +64,20 @@ public String parentIdField() { public void assignResourceSharingClient(ResourceSharingClient resourceSharingClient) { ResourceSharingClientAccessor.getInstance().setResourceSharingClient(resourceSharingClient); } + + /** + * Sample resolver: maps each of the user's security roles to a workspace id ({@code ws-}). Security roles + * are server-resolved (not user-assertable), satisfying the SPI's trusted-source contract. + */ + @Override + public Set resolveWorkspacesForUser(String username, Set securityRoles, Set backendRoles) { + if (securityRoles == null || securityRoles.isEmpty()) { + return Collections.emptySet(); + } + Set workspaces = new HashSet<>(); + for (String role : securityRoles) { + workspaces.add("ws-" + role); + } + return workspaces; + } } diff --git a/sample-resource-plugin/src/main/java/org/opensearch/sample/SampleWorkspaceExtension.java b/sample-resource-plugin/src/main/java/org/opensearch/sample/SampleWorkspaceExtension.java new file mode 100644 index 0000000000..a0f8d4496f --- /dev/null +++ b/sample-resource-plugin/src/main/java/org/opensearch/sample/SampleWorkspaceExtension.java @@ -0,0 +1,62 @@ +/* + * 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.sample; + +import java.util.Set; + +import org.opensearch.sample.client.ResourceSharingClientAccessor; +import org.opensearch.security.spi.resources.ResourceProvider; +import org.opensearch.security.spi.resources.ResourceSharingExtension; +import org.opensearch.security.spi.resources.client.ResourceSharingClient; + +import static org.opensearch.sample.utils.Constants.RESOURCE_INDEX_NAME; +import static org.opensearch.sample.utils.Constants.WORKSPACE_TYPE; + +/** + * Registers {@code workspace} as a resource type so the write-path container fan-out + * ({@code ResourceAccessHandler.checkContainers}) can be exercised: a resource inherits access from any workspace it + * belongs to. Workspace records share the sample resource index, distinguished by {@code resource_type}; their access + * levels (workspace_read_only/read_write/full_access) map to child actions in resource-access-levels.yml. + */ +public class SampleWorkspaceExtension implements ResourceSharingExtension { + + @Override + public Set getResourceProviders() { + return Set.of(new ResourceProvider() { + @Override + public String resourceType() { + return WORKSPACE_TYPE; + } + + @Override + public String resourceIndexName() { + return RESOURCE_INDEX_NAME; + } + + @Override + public String typeField() { + return "resource_type"; + } + + @Override + public String workspacesField() { + // A workspace does not itself belong to a workspace. + return null; + } + }); + } + + @Override + public void assignResourceSharingClient(ResourceSharingClient resourceSharingClient) { + ResourceSharingClientAccessor.getInstance().setResourceSharingClient(resourceSharingClient); + } +} diff --git a/sample-resource-plugin/src/main/java/org/opensearch/sample/resource/actions/rest/create/CreateResourceRestAction.java b/sample-resource-plugin/src/main/java/org/opensearch/sample/resource/actions/rest/create/CreateResourceRestAction.java index 401aaa9348..acc54a4f49 100644 --- a/sample-resource-plugin/src/main/java/org/opensearch/sample/resource/actions/rest/create/CreateResourceRestAction.java +++ b/sample-resource-plugin/src/main/java/org/opensearch/sample/resource/actions/rest/create/CreateResourceRestAction.java @@ -67,6 +67,7 @@ private RestChannelConsumer updateResource(Map source, String re resource.setDescription(description); resource.setGroupId(groupId); resource.setAttributes(attributes); + resource.setWorkspaces(getWorkspaces(source)); final UpdateResourceRequest updateResourceRequest = new UpdateResourceRequest(resourceId, resource); return channel -> client.executeLocally( UpdateResourceAction.INSTANCE, @@ -86,6 +87,7 @@ private RestChannelConsumer createResource(Map source, NodeClien resource.setName(name); resource.setDescription(description); resource.setAttributes(attributes); + resource.setWorkspaces(getWorkspaces(source)); final CreateResourceRequest createSampleResourceRequest = new CreateResourceRequest(resource, shouldStoreUser); return channel -> client.executeLocally( CreateResourceAction.INSTANCE, @@ -94,6 +96,21 @@ private RestChannelConsumer createResource(Map source, NodeClien ); } + @SuppressWarnings("unchecked") + private java.util.Set getWorkspaces(Map source) { + Object v = source.get("workspaces"); + if (v == null) return null; + java.util.Set out = new java.util.HashSet<>(); + if (v instanceof java.util.List) { + for (Object item : (java.util.List) v) { + if (item != null) out.add(item.toString()); + } + } else { + out.add(v.toString()); + } + return out; + } + @SuppressWarnings("unchecked") private Map getAttributes(Map source) { return source.containsKey("attributes") ? (Map) source.get("attributes") : null; diff --git a/sample-resource-plugin/src/main/java/org/opensearch/sample/resource/actions/transport/UpdateResourceTransportAction.java b/sample-resource-plugin/src/main/java/org/opensearch/sample/resource/actions/transport/UpdateResourceTransportAction.java index e6a8834d54..6684379b1d 100644 --- a/sample-resource-plugin/src/main/java/org/opensearch/sample/resource/actions/transport/UpdateResourceTransportAction.java +++ b/sample-resource-plugin/src/main/java/org/opensearch/sample/resource/actions/transport/UpdateResourceTransportAction.java @@ -8,10 +8,15 @@ package org.opensearch.sample.resource.actions.transport; +import java.util.Collection; +import java.util.HashSet; +import java.util.Set; + import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; import org.apache.logging.log4j.message.ParameterizedMessage; +import org.opensearch.action.get.GetRequest; import org.opensearch.action.index.IndexRequest; import org.opensearch.action.support.ActionFilters; import org.opensearch.action.support.HandledTransportAction; @@ -58,30 +63,52 @@ protected void doExecute(Task task, UpdateResourceRequest request, ActionListene } private void updateResource(UpdateResourceRequest request, ActionListener listener) { - try { - String resourceId = request.getResourceId(); - SampleResource sample = request.getResource(); - try (XContentBuilder builder = jsonBuilder()) { - sample.toXContent(builder, ToXContent.EMPTY_PARAMS); + String resourceId = request.getResourceId(); + SampleResource sample = request.getResource(); + // Workspace membership is server-controlled: because `workspaces` drives read AND write authorization, an + // ordinary update MUST NOT be able to change it (otherwise a user could add a workspace where they hold a + // stronger access level and escalate). Preserve the stored value and ignore any caller-supplied workspaces; + // a real backend routes membership changes through an authorized associate/dissociate path. See + // ResourceProvider#workspacesField. + pluginClient.get(new GetRequest(RESOURCE_INDEX_NAME, resourceId), ActionListener.wrap(getResponse -> { + try { + Set existingWorkspaces = new HashSet<>(); + if (getResponse.isExists() + && getResponse.getSource() != null + && getResponse.getSource().get("workspaces") instanceof Collection c) { + for (Object o : c) { + if (o != null) { + existingWorkspaces.add(o.toString()); + } + } + } + sample.setWorkspaces(existingWorkspaces.isEmpty() ? null : existingWorkspaces); + + try (XContentBuilder builder = jsonBuilder()) { + sample.toXContent(builder, ToXContent.EMPTY_PARAMS); - // because some plugins seem to treat update API calls as index request - IndexRequest ir = new IndexRequest(RESOURCE_INDEX_NAME).id(resourceId) - .setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL) // WAIT_UNTIL because we don't want tests to fail, as they - // execute search right after update - .source(builder); + // because some plugins seem to treat update API calls as index request + IndexRequest ir = new IndexRequest(RESOURCE_INDEX_NAME).id(resourceId) + .setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL) // WAIT_UNTIL because we don't want tests to fail, as they + // execute search right after update + .source(builder); - log.debug("Update Request: {}", ir.toString()); + log.debug("Update Request: {}", ir.toString()); - pluginClient.index(ir, ActionListener.wrap(updateResponse -> { - listener.onResponse( - new CreateResourceResponse("Resource " + request.getResource().getName() + " updated successfully.") + pluginClient.index( + ir, + ActionListener.wrap( + updateResponse -> listener.onResponse( + new CreateResourceResponse("Resource " + sample.getName() + " updated successfully.") + ), + listener::onFailure + ) ); - }, listener::onFailure)); + } + } catch (Exception e) { + log.error(() -> new ParameterizedMessage("Failed to update resource: {}", resourceId), e); + listener.onFailure(e); } - } catch (Exception e) { - log.error(() -> new ParameterizedMessage("Failed to update resource: {}", request.getResourceId()), e); - listener.onFailure(e); - } - + }, listener::onFailure)); } } diff --git a/sample-resource-plugin/src/main/java/org/opensearch/sample/utils/Constants.java b/sample-resource-plugin/src/main/java/org/opensearch/sample/utils/Constants.java index 876bdef2e0..08eb848c5c 100644 --- a/sample-resource-plugin/src/main/java/org/opensearch/sample/utils/Constants.java +++ b/sample-resource-plugin/src/main/java/org/opensearch/sample/utils/Constants.java @@ -15,6 +15,9 @@ public class Constants { public static final String RESOURCE_INDEX_NAME = ".sample_resource"; public static final String RESOURCE_TYPE = "sample-resource"; public static final String RESOURCE_GROUP_TYPE = "sample-resource-group"; + // Must equal ResourceAccessHandler.WORKSPACE_RESOURCE_TYPE: the security plugin resolves workspace containers by + // this exact type name. Registered so the write-path container fan-out can be exercised end-to-end. + public static final String WORKSPACE_TYPE = "workspace"; public static final String SAMPLE_RESOURCE_PLUGIN_PREFIX = "_plugins/sample_plugin"; public static final String SAMPLE_RESOURCE_PLUGIN_API_PREFIX = "/" + SAMPLE_RESOURCE_PLUGIN_PREFIX; diff --git a/sample-resource-plugin/src/main/resources/META-INF/services/org.opensearch.security.spi.resources.ResourceSharingExtension b/sample-resource-plugin/src/main/resources/META-INF/services/org.opensearch.security.spi.resources.ResourceSharingExtension index 07a1a39b61..f5961afe56 100644 --- a/sample-resource-plugin/src/main/resources/META-INF/services/org.opensearch.security.spi.resources.ResourceSharingExtension +++ b/sample-resource-plugin/src/main/resources/META-INF/services/org.opensearch.security.spi.resources.ResourceSharingExtension @@ -1,4 +1,5 @@ # SPDX-License-Identifier: Apache-2.0 org.opensearch.sample.SampleResourceExtension -org.opensearch.sample.SampleResourceGroupExtension \ No newline at end of file +org.opensearch.sample.SampleResourceGroupExtension +org.opensearch.sample.SampleWorkspaceExtension \ No newline at end of file diff --git a/sample-resource-plugin/src/main/resources/mappings.json b/sample-resource-plugin/src/main/resources/mappings.json index b163ee8c11..35e14c5394 100644 --- a/sample-resource-plugin/src/main/resources/mappings.json +++ b/sample-resource-plugin/src/main/resources/mappings.json @@ -9,6 +9,9 @@ }, "all_shared_principals": { "type": "keyword" + }, + "workspaces": { + "type": "keyword" } } } diff --git a/sample-resource-plugin/src/main/resources/resource-access-levels.yml b/sample-resource-plugin/src/main/resources/resource-access-levels.yml index c3d7af2c66..3bc7d8f687 100644 --- a/sample-resource-plugin/src/main/resources/resource-access-levels.yml +++ b/sample-resource-plugin/src/main/resources/resource-access-levels.yml @@ -37,3 +37,21 @@ resource_types: # group full-access grants full access on child resources - "sampleresource:*" - "cluster:admin/security/resource/share" + workspace: + workspace_read_only: + default: true + # workspace read grants read on member resources + allowed_actions: + - "sampleresource:get" + - "sampleresource:search" + + workspace_read_write: + allowed_actions: + - "sampleresource:get" + - "sampleresource:search" + - "sampleresource:update" + + workspace_full_access: + allowed_actions: + - "sampleresource:*" + - "cluster:admin/security/resource/share" diff --git a/spi/src/main/java/org/opensearch/security/spi/resources/ResourceProvider.java b/spi/src/main/java/org/opensearch/security/spi/resources/ResourceProvider.java index adc020d9cb..5794105b5d 100644 --- a/spi/src/main/java/org/opensearch/security/spi/resources/ResourceProvider.java +++ b/spi/src/main/java/org/opensearch/security/spi/resources/ResourceProvider.java @@ -79,4 +79,24 @@ default String ownerBackendRolesPath() { return null; } + /** + * Field on documents of this type holding the set of workspace IDs the resource belongs to (a resource may belong + * to multiple workspaces). Must be mapped as multi-valued {@code keyword}: the read path filters it in DLS with a + * {@code terms} query, and the write path stores it on the sharing record for access-level resolution. Defaults to + * {@code "workspaces"}; return {@code null} to opt out. A document without the field belongs to no workspace. + * + *

Trusted-write contract: this field drives authorization, so a provider MUST NOT let a caller change it + * on an ordinary update — membership changes on an existing resource go through a server-authorized + * associate/dissociate path (else a user could add the resource to a workspace where they hold stronger access and + * escalate). At create time it is owner-governed, but the backend must still validate the creator may add the + * resource to each requested workspace. See {@link ResourceSharingExtension#resolveWorkspacesForUser} for the + * matching contract on user membership. + * + * @return the field name containing the resource's workspace IDs (default {@code "workspaces"}), or + * {@code null} to opt out + */ + default String workspacesField() { + return "workspaces"; + } + } diff --git a/spi/src/main/java/org/opensearch/security/spi/resources/ResourceSharingExtension.java b/spi/src/main/java/org/opensearch/security/spi/resources/ResourceSharingExtension.java index 8465fc3124..4679c6fbdc 100644 --- a/spi/src/main/java/org/opensearch/security/spi/resources/ResourceSharingExtension.java +++ b/spi/src/main/java/org/opensearch/security/spi/resources/ResourceSharingExtension.java @@ -8,6 +8,7 @@ package org.opensearch.security.spi.resources; +import java.util.Collections; import java.util.Set; import org.opensearch.common.Nullable; @@ -37,4 +38,28 @@ public interface ResourceSharingExtension extends SecurityConfigExtension { * @param client the ResourceSharingClient instance, or {@code null} when the feature is disabled */ void assignResourceSharingClient(@Nullable ResourceSharingClient client); + + /** + * Returns the workspace IDs the user is a member of. Called on the privilege hot path when building the DLS filter: + * the returned IDs are matched against each resource's own {@code workspaces} field (not denormalized into + * {@code all_shared_principals}). + * + *

Contract (security-sensitive): + *

    + *
  • MUST come from a trusted, server-set source the user cannot assert (e.g. resolved at authentication time), + * never from user-influenceable input like JWT/proxy claims — the result grants read visibility.
    • + *
  • MUST be I/O-free (privilege hot path): resolve eagerly at authentication time or from an in-memory cache.
    • + *
+ * + *

Defaults to an empty set (workspace-based visibility disabled); only plugins owning an authoritative + * membership source should override. + * + * @param username the authenticated user's name; never {@code null} + * @param securityRoles the user's security roles; never {@code null}, may be empty + * @param backendRoles the user's backend roles; never {@code null}, may be empty + * @return the trusted workspace IDs the user belongs to, or an empty set if none / not implemented + */ + default Set resolveWorkspacesForUser(String username, Set securityRoles, Set backendRoles) { + return Collections.emptySet(); + } } diff --git a/src/main/java/org/opensearch/security/OpenSearchSecurityPlugin.java b/src/main/java/org/opensearch/security/OpenSearchSecurityPlugin.java index 011dbde77f..168ba4d374 100644 --- a/src/main/java/org/opensearch/security/OpenSearchSecurityPlugin.java +++ b/src/main/java/org/opensearch/security/OpenSearchSecurityPlugin.java @@ -210,6 +210,7 @@ import org.opensearch.security.resources.api.share.ShareTransportAction; import org.opensearch.security.resources.settings.ResourceSharingFeatureFlagSetting; import org.opensearch.security.resources.settings.ResourceSharingProtectedResourcesSetting; +import org.opensearch.security.resources.sharing.ResourceSharing; import org.opensearch.security.rest.DashboardsInfoAction; import org.opensearch.security.rest.SecurityConfigUpdateAction; import org.opensearch.security.rest.SecurityHealthAction; @@ -1792,7 +1793,9 @@ public Collection createComponents( public List getNamedWriteables() { return List.of( new NamedWriteableRegistry.Entry(ClusterState.Custom.class, SecurityMetadata.TYPE, SecurityMetadata::new), - new NamedWriteableRegistry.Entry(NamedDiff.class, SecurityMetadata.TYPE, SecurityMetadata::readDiffFrom) + new NamedWriteableRegistry.Entry(NamedDiff.class, SecurityMetadata.TYPE, SecurityMetadata::readDiffFrom), + // Reader for ResourceSharing so ShareResponse's readNamedWriteable(ResourceSharing.class) round-trips. + new NamedWriteableRegistry.Entry(ResourceSharing.class, ResourceSharing.NAME, ResourceSharing::new) ); } diff --git a/src/main/java/org/opensearch/security/configuration/DlsFlsValveImpl.java b/src/main/java/org/opensearch/security/configuration/DlsFlsValveImpl.java index 2c6ac84f1a..f690e7e83c 100644 --- a/src/main/java/org/opensearch/security/configuration/DlsFlsValveImpl.java +++ b/src/main/java/org/opensearch/security/configuration/DlsFlsValveImpl.java @@ -189,7 +189,8 @@ public boolean invoke(PrivilegesEvaluationContext context, final ActionListener< IndexToRuleMap sharedResourceMap = ResourceSharingDlsUtils.resourceRestrictions( namedXContentRegistry, resolvedIndexNames, - user + user, + resourcePluginInfo ); return DlsFilterLevelActionHandler.handle( diff --git a/src/main/java/org/opensearch/security/resources/ResourceAccessHandler.java b/src/main/java/org/opensearch/security/resources/ResourceAccessHandler.java index aecb168245..db6b497d53 100644 --- a/src/main/java/org/opensearch/security/resources/ResourceAccessHandler.java +++ b/src/main/java/org/opensearch/security/resources/ResourceAccessHandler.java @@ -11,7 +11,9 @@ package org.opensearch.security.resources; +import java.util.ArrayList; import java.util.Collections; +import java.util.List; import java.util.Set; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -166,44 +168,101 @@ public void hasPermission( return; } - if (sharingInfo.isCreatedBy(user.getName())) { + if (recordGrantsAction(sharingInfo, resourceType, user, action)) { listener.onResponse(true); return; } - Set accessLevels = sharingInfo.getAccessLevelsForUser(user); - - // no matching access level, either recurse up or fail fast - if (accessLevels.isEmpty()) { - if (sharingInfo.getParentId() != null) { - hasPermission(sharingInfo.getParentId(), sharingInfo.getParentType(), action, listener); - } else { - listener.onResponse(false); - } - return; - } - - // Fetch the static action-groups registered by plugins on bootstrap and check whether any match - final FlattenedActionGroups agForType = resourcePluginInfo.flattenedForType(resourceType); - final Set allowedActions = agForType.resolve(accessLevels); - final WildcardMatcher matcher = WildcardMatcher.from(allowedActions); - - if (matcher.test(action)) { - listener.onResponse(true); - return; - } - - if (sharingInfo.getParentId() != null) { - hasPermission(sharingInfo.getParentId(), sharingInfo.getParentType(), action, listener); - } else { - listener.onResponse(false); - } + // resource itself does not grant the action: fall back to its containers (parent and/or workspaces) + checkContainers(sharingInfo, action, listener); }, e -> { LOGGER.error("Error while checking permission for user {} on resource {}: {}", user.getName(), resourceId, e.getMessage()); listener.onFailure(e); })); } + /** + * Returns whether a single sharing record grants the given user the requested action directly — i.e. the user is + * the creator, or is shared with at an access level whose resolved action-group matches {@code action}. This is a + * pure, in-memory computation (no I/O), factored out so it can be reused both for the resource itself and for each + * container record fetched in a batch. + */ + private boolean recordGrantsAction(ResourceSharing sharingInfo, String resourceType, User user, String action) { + if (sharingInfo.isCreatedBy(user.getName())) { + return true; + } + Set accessLevels = sharingInfo.getAccessLevelsForUser(user); + if (accessLevels.isEmpty()) { + return false; + } + final FlattenedActionGroups agForType = resourcePluginInfo.flattenedForType(resourceType); + final Set allowedActions = agForType.resolve(accessLevels); + return WildcardMatcher.from(allowedActions).test(action); + } + + /** + * Grants access if any of the resource's containers grant it: its single parent (recursed via + * {@link #hasPermission}) or any of its workspaces. Workspace records are fetched in one + * {@link ResourceSharingIndexHandler#fetchSharingInfoForIds mget} and evaluated as leaves (their own + * {@code share_with}), so no per-workspace round trip and no recursion. + *

+ * {@link #WORKSPACE_RESOURCE_TYPE} is a placeholder until the workspace provider is registered via the SPI; if it + * isn't, {@code indexByType} returns null and the workspace branch denies cleanly. + * + * @param sharingInfo the sharing record of the resource whose containers should be consulted + * @param action the action being authorized + * @param listener notified with {@code true} if any container grants access, {@code false} otherwise + */ + private void checkContainers(ResourceSharing sharingInfo, String action, ActionListener listener) { + final User user = getAuthenticatedUser(); + if (user == null) { + listener.onResponse(false); + return; + } + + final List workspaceIds = new ArrayList<>(sharingInfo.getWorkspaces()); + final String workspaceIndex = workspaceIds.isEmpty() ? null : resourcePluginInfo.indexByType(WORKSPACE_RESOURCE_TYPE); + + // Evaluate workspaces (batched) first; fall back to the single parent (recursive) only if no workspace grants. + if (workspaceIndex != null) { + resourceSharingIndexHandler.fetchSharingInfoForIds(workspaceIndex, workspaceIds, ActionListener.wrap(records -> { + for (ResourceSharing wsRecord : records.values()) { + // Resolve against the workspace type's action groups: only they map workspace-level access + // (workspace_read/write) to the child action being authorized. + if (recordGrantsAction(wsRecord, WORKSPACE_RESOURCE_TYPE, user, action)) { + listener.onResponse(true); + return; + } + } + checkParent(sharingInfo, action, listener); + }, listener::onFailure)); + } else { + checkParent(sharingInfo, action, listener); + } + } + + /** + * Resolves access inherited from the single hierarchical parent (if any), recursing via {@link #hasPermission} so + * grandparent chains continue to work. Denies when there is no parent. + */ + private void checkParent(ResourceSharing sharingInfo, String action, ActionListener listener) { + if (sharingInfo.getParentId() != null) { + hasPermission(sharingInfo.getParentId(), sharingInfo.getParentType(), action, listener); + } else { + listener.onResponse(false); + } + } + + /** + * Returns the currently authenticated user from the thread context, or {@code null} if none. + */ + private User getAuthenticatedUser() { + return (User) threadContext.getPersistent(ConfigConstants.OPENDISTRO_SECURITY_AUTHENTICATED_USER); + } + + /** Resource type of a workspace; the workspace provider registers it via the resource-sharing SPI. */ + private static final String WORKSPACE_RESOURCE_TYPE = "workspace"; + /** * Patches the sharing info. It could be either or all 3 of the following possibilities: * 1. Revoke access - remove op diff --git a/src/main/java/org/opensearch/security/resources/ResourceIndexListener.java b/src/main/java/org/opensearch/security/resources/ResourceIndexListener.java index e24da4a256..aca921f4b8 100644 --- a/src/main/java/org/opensearch/security/resources/ResourceIndexListener.java +++ b/src/main/java/org/opensearch/security/resources/ResourceIndexListener.java @@ -9,6 +9,7 @@ package org.opensearch.security.resources; import java.io.IOException; +import java.util.Set; import org.apache.logging.log4j.LogManager; import org.apache.logging.log4j.Logger; @@ -92,6 +93,7 @@ public void postIndex(ShardId shardId, Engine.Index index, Engine.IndexResult re } if (!result.isCreated()) { + // Restore all_shared_principals on the resource doc (guards against a direct write tampering with it). ActionListener listener = ActionListener.wrap(unused -> { log.debug( "postIndex: Successfully updated the resource visibility for resource {} within index {}", @@ -100,6 +102,22 @@ public void postIndex(ShardId shardId, Engine.Index index, Engine.IndexResult re ); }, e -> { log.debug(e.getMessage()); }); this.resourceSharingIndexHandler.fetchAndUpdateResourceVisibility(resourceId, resourceIndex, listener); + + // Reconcile the sharing record's workspaces to the doc's current set (associate/dissociate, incl. removals) + // so the write-path record tracks the read-path field. The op's seq_no is the monotonic guard. + if (provider.workspacesField() != null) { + Set currentWorkspaces = ResourcePluginInfo.extractMultiValuedFieldFromIndexOp(provider.workspacesField(), index); + this.resourceSharingIndexHandler.reconcileWorkspaces( + resourceIndex, + resourceId, + currentWorkspaces, + result.getSeqNo(), + ActionListener.wrap( + changed -> log.debug("postIndex: workspace reconcile for {} changed={}", resourceId, changed), + e -> log.warn("postIndex: failed to reconcile workspaces for {}: {}", resourceId, e.getMessage()) + ) + ); + } return; } @@ -129,6 +147,11 @@ public void postIndex(ShardId shardId, Engine.Index index, Engine.IndexResult re if (parentType != null) { builder.parentType(parentType).parentId(parentId); } + // Stamp the resource's workspaces onto the sharing record (used by the write-path access-level + // fan-out). Providers that declare no workspaces field are unaffected. + if (provider.workspacesField() != null) { + builder.workspaces(ResourcePluginInfo.extractMultiValuedFieldFromIndexOp(provider.workspacesField(), index)); + } this.resourceSharingIndexHandler.indexResourceSharing(resourceIndex, builder.build(), listener); } catch (IOException e) { log.warn("Failed to create a resource sharing entry for resource: {}", resourceId, e); @@ -175,15 +198,18 @@ public void postIndex(ShardId shardId, Engine.Index index, Engine.IndexResult re ); return; } - ResourceSharing sharingInfo = ResourceSharing.builder() + ResourceSharing.Builder childBuilder = ResourceSharing.builder() .resourceId(resourceId) .resourceType(resourceType) .tenant(parentSharing.getTenant()) .createdBy(parentSharing.getCreatedBy()) .parentType(parentType) - .parentId(parentId) - .build(); - this.resourceSharingIndexHandler.indexResourceSharing(resourceIndex, sharingInfo, listener); + .parentId(parentId); + // Stamp the child's own workspaces onto its record; ownership is still inherited from the parent above. + if (provider.workspacesField() != null) { + childBuilder.workspaces(ResourcePluginInfo.extractMultiValuedFieldFromIndexOp(provider.workspacesField(), index)); + } + this.resourceSharingIndexHandler.indexResourceSharing(resourceIndex, childBuilder.build(), listener); }, e -> log.warn( "Failed to create a resource sharing entry for child resource {} in index {}: could not fetch parent {} sharing record: {}", diff --git a/src/main/java/org/opensearch/security/resources/ResourcePluginInfo.java b/src/main/java/org/opensearch/security/resources/ResourcePluginInfo.java index 78a9b7a38f..95354c92c1 100644 --- a/src/main/java/org/opensearch/security/resources/ResourcePluginInfo.java +++ b/src/main/java/org/opensearch/security/resources/ResourcePluginInfo.java @@ -75,6 +75,10 @@ public void setResourceSharingExtensions(Set extension // Enforce resource-type unique-ness Set resourceTypes = new HashSet<>(); + // Providers may share a resource index, but must then agree on the workspaces field: DLS filters a single + // field per index while ingestion uses each provider's declared field, so a disagreement would silently + // mis-scope read visibility. Reject conflicting non-null declarations at registration. + Map indexToWorkspacesField = new HashMap<>(); for (ResourceSharingExtension extension : extensions) { for (var rp : extension.getResourceProviders()) { if (!resourceTypes.contains(rp.resourceType())) { @@ -91,6 +95,22 @@ public void setResourceSharingExtensions(Set extension ) ); } + + String workspacesField = rp.workspacesField(); + if (workspacesField != null) { + String existing = indexToWorkspacesField.putIfAbsent(rp.resourceIndexName(), workspacesField); + if (existing != null && !existing.equals(workspacesField)) { + throw new OpenSearchSecurityException( + String.format( + "Conflicting workspaces fields declared for resource index [%s]: [%s] and [%s]. All providers sharing" + + " an index must declare the same workspaces field (or null to opt out).", + rp.resourceIndexName(), + existing, + workspacesField + ) + ); + } + } } } resourceSharingExtensions.addAll(extensions); @@ -148,6 +168,29 @@ public static String extractFieldFromIndexOp(String fieldName, Engine.Index inde return fieldValue; } + /** + * Extracts all values of a multi-valued field from the Lucene document backing an {@link Engine.Index} op — + * e.g. the set of workspace IDs a resource belongs to (see {@link ResourceProvider#workspacesField()}). Unlike + * {@link #extractFieldFromIndexOp} it collects every {@link IndexableField} for {@code fieldName}, which is how a + * {@code keyword} array surfaces (one field per element). Values only surface for indexed/stored mappings, not + * {@code doc_values}-only fields. + * + * @param fieldName the multi-valued field to extract; must not be {@code null} + * @param indexOp the index op whose parsed document is inspected; must not be {@code null} + * @return the field's values, or empty if none + */ + public static Set extractMultiValuedFieldFromIndexOp(String fieldName, Engine.Index indexOp) { + Set values = new HashSet<>(); + for (IndexableField f : indexOp.parsedDoc().rootDoc().getFields(fieldName)) { + if (f.stringValue() != null) { + values.add(f.stringValue()); + } else if (f.binaryValue() != null) { // e.g., BytesRef-backed + values.add(f.binaryValue().utf8ToString()); + } + } + return values; + } + /** * Resolves the resource type for the given index operation and resource index. *

@@ -192,6 +235,35 @@ public Set getResourceSharingExtensions() { return ImmutableSet.copyOf(resourceSharingExtensions); } + /** + * Aggregates {@link ResourceSharingExtension#resolveWorkspacesForUser} across every registered extension into a + * single trusted set of workspace IDs for the user. Empty if no extension contributes any. Called on the DLS hot + * path — each extension's implementation is required by contract to be I/O-free. + * + * @param user the authenticated user + * @return the union of workspace IDs contributed by all registered extensions + */ + public Set resolveWorkspacesForUser(org.opensearch.security.user.User user) { + lock.readLock().lock(); + try { + if (resourceSharingExtensions.isEmpty()) { + return java.util.Collections.emptySet(); + } + Set securityRoles = user.getSecurityRoles() == null ? java.util.Collections.emptySet() : user.getSecurityRoles(); + Set backendRoles = user.getRoles() == null ? java.util.Collections.emptySet() : user.getRoles(); + Set merged = new HashSet<>(); + for (ResourceSharingExtension extension : resourceSharingExtensions) { + Set contributed = extension.resolveWorkspacesForUser(user.getName(), securityRoles, backendRoles); + if (contributed != null && !contributed.isEmpty()) { + merged.addAll(contributed); + } + } + return merged; + } finally { + lock.readLock().unlock(); + } + } + public void setResourceSharingClient(ResourceSharingClient resourceAccessControlClient) { this.resourceAccessControlClient = resourceAccessControlClient; } @@ -295,6 +367,32 @@ public String getParentType(String resourceType) { } } + /** + * Returns the provider-declared workspaces field name for the resource index (see + * {@link ResourceProvider#workspacesField()}), or {@code null} if no provider on that index declares one. + * Used by DLS to filter workspace membership on the field a provider actually declares, rather than a fixed name. + * When multiple providers share an index, the first declared (non-null) field wins. + */ + /** + * Returns the workspaces field for the given resource index, or {@code null} if no provider on that index + * declares one. Providers sharing an index are required to agree on this field — conflicting non-null + * declarations are rejected at registration (see {@link #setResourceSharingExtensions}) — so the value is + * unambiguous regardless of map iteration order. + */ + public String workspacesFieldForIndex(String index) { + lock.readLock().lock(); + try { + for (ResourceProvider provider : typeToProvider.values()) { + if (provider.resourceIndexName().equals(index) && provider.workspacesField() != null) { + return provider.workspacesField(); + } + } + return null; + } finally { + lock.readLock().unlock(); + } + } + public Set getResourceTypes() { lock.readLock().lock(); try { diff --git a/src/main/java/org/opensearch/security/resources/ResourceSharingDlsUtils.java b/src/main/java/org/opensearch/security/resources/ResourceSharingDlsUtils.java index dd7f756a18..b978d5daa2 100644 --- a/src/main/java/org/opensearch/security/resources/ResourceSharingDlsUtils.java +++ b/src/main/java/org/opensearch/security/resources/ResourceSharingDlsUtils.java @@ -12,6 +12,7 @@ import java.util.ArrayList; import java.util.Collection; import java.util.List; +import java.util.Set; import com.google.common.collect.ImmutableMap; import org.apache.logging.log4j.LogManager; @@ -31,7 +32,8 @@ public class ResourceSharingDlsUtils { public static IndexToRuleMap resourceRestrictions( NamedXContentRegistry xContentRegistry, Collection resolvedIndices, - User user + User user, + ResourcePluginInfo resourcePluginInfo ) { List principals = new ArrayList<>(); @@ -48,24 +50,48 @@ public static IndexToRuleMap resourceRestrictions( user.getRoles().forEach(br -> principals.add("backend:" + br)); } - XContentBuilder builder = null; - DlsRestriction restriction; + // Workspace visibility is expressed as a separate clause on the resource's own `workspaces` field (which OSD + // maintains), rather than by denormalizing workspace: into all_shared_principals. Membership comes from + // ResourceSharingExtension.resolveWorkspacesForUser, whose SPI contract requires a trusted, server-set, + // I/O-free source (see that interface's javadoc). Filtering the live field means associate/dissociate are + // reflected automatically. If no extension implements the resolver, the set is empty and the clause is omitted. + Set userWorkspaces = resourcePluginInfo == null ? Set.of() : resourcePluginInfo.resolveWorkspacesForUser(user); + + // The workspaces clause targets the field each provider actually declares (workspacesField()), resolved per + // index — not a hardcoded name — so it matches the field ingestion reads. Built per index accordingly. + ImmutableMap.Builder mapBuilder = ImmutableMap.builder(); + for (String index : resolvedIndices) { + String workspacesField = resourcePluginInfo == null ? null : resourcePluginInfo.workspacesFieldForIndex(index); + mapBuilder.put(index, buildRestriction(xContentRegistry, principals, workspacesField, userWorkspaces)); + } + return new IndexToRuleMap<>(mapBuilder.build()); + } + + /** + * Builds the per-index DLS restriction: a doc is visible if it is shared with one of the user's principals OR (when + * the index declares a workspaces field and the user has workspace access) it belongs to one of the user's + * workspaces — {@code bool.should[ terms(all_shared_principals), terms() ]}, min_should_match=1. + */ + private static DlsRestriction buildRestriction( + NamedXContentRegistry xContentRegistry, + List principals, + String workspacesField, + Set userWorkspaces + ) { try { - // Build a single `terms` query JSON - builder = XContentFactory.jsonBuilder(); + XContentBuilder builder = XContentFactory.jsonBuilder(); + builder.startObject().startObject("bool").startArray("should"); builder.startObject().startObject("terms").array("all_shared_principals", principals.toArray()).endObject().endObject(); + if (workspacesField != null && !userWorkspaces.isEmpty()) { + builder.startObject().startObject("terms").array(workspacesField, userWorkspaces.toArray()).endObject().endObject(); + } + builder.endArray().field("minimum_should_match", 1).endObject().endObject(); - String dlsJson = builder.toString(); - restriction = new DlsRestriction(List.of(DocumentPrivileges.getRenderedDlsQuery(xContentRegistry, dlsJson))); + return new DlsRestriction(List.of(DocumentPrivileges.getRenderedDlsQuery(xContentRegistry, builder.toString()))); } catch (IOException e) { LOGGER.warn("Received error while applying resource restrictions.", e); - restriction = DlsRestriction.FULL; + return DlsRestriction.FULL; } - - ImmutableMap.Builder mapBuilder = ImmutableMap.builder(); - for (String index : resolvedIndices) { - mapBuilder.put(index, restriction); - } - return new IndexToRuleMap<>(mapBuilder.build()); } + } diff --git a/src/main/java/org/opensearch/security/resources/ResourceSharingIndexHandler.java b/src/main/java/org/opensearch/security/resources/ResourceSharingIndexHandler.java index 628bcf4903..06f8730c59 100644 --- a/src/main/java/org/opensearch/security/resources/ResourceSharingIndexHandler.java +++ b/src/main/java/org/opensearch/security/resources/ResourceSharingIndexHandler.java @@ -13,6 +13,7 @@ import java.util.ArrayList; import java.util.Collection; import java.util.Collections; +import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; @@ -64,6 +65,7 @@ import org.opensearch.index.query.BoolQueryBuilder; import org.opensearch.index.query.MatchAllQueryBuilder; import org.opensearch.index.query.QueryBuilders; +import org.opensearch.index.seqno.SequenceNumbers; import org.opensearch.search.Scroll; import org.opensearch.search.SearchHit; import org.opensearch.search.builder.SearchSourceBuilder; @@ -71,7 +73,6 @@ import org.opensearch.security.resources.api.share.ShareAction; import org.opensearch.security.resources.sharing.CreatedBy; import org.opensearch.security.resources.sharing.Recipient; -import org.opensearch.security.resources.sharing.Recipients; import org.opensearch.security.resources.sharing.ResourceSharing; import org.opensearch.security.resources.sharing.ShareWith; import org.opensearch.security.user.User; @@ -91,6 +92,15 @@ public class ResourceSharingIndexHandler { private static final Logger LOGGER = LogManager.getLogger(ResourceSharingIndexHandler.class); + // Monotonic guard: seq_no of the source-document write last reconciled onto the sharing record's workspaces. + private static final String WORKSPACES_SEQ_NO_FIELD = "workspaces_seq_no"; + // Retries for the share/patch read-modify-write under optimistic concurrency (a concurrent reconcile write). + private static final int SHARING_UPDATE_MAX_ATTEMPTS = 5; + // Bounded retry for reconcile: covers the window where the sharing record is still being created asynchronously, + // and re-reads after a lost optimistic-concurrency compare-and-set. + private static final int WORKSPACE_RECONCILE_MAX_ATTEMPTS = 5; + private static final TimeValue WORKSPACE_RECONCILE_RETRY_DELAY = TimeValue.timeValueMillis(100); + private final Client client; private final ThreadPool threadPool; @@ -148,14 +158,122 @@ public static String getSharingIndex(String resourceIndex) { } /** - * Updates the visibility of a resource document by replacing its {@code principals} field - * with the provided list of principals. The update is executed immediately with - * {@link WriteRequest.RefreshPolicy#IMMEDIATE} to ensure the change is visible in subsequent - * searches. + * Sets a sharing record's {@code workspaces} to exactly the given set (adds and removals), keeping the write path + * in step with the read path. Used by live associate/dissociate and by migration. *

- * The supplied {@link ActionListener} will be invoked with the {@link UpdateResponse} - * on success, or with an exception on failure. + * Monotonic: {@code sourceSeqNo} (the source-doc write's seq_no) is stored as {@code workspaces_seq_no} and a + * reconcile applies only when it is newer, so a slow reconcile can't overwrite a newer one. Guarded by + * if_seq_no/if_primary_term (retried on conflict); a not-yet-created record is retried. {@code created_by}/ + * {@code share_with} are untouched. * + * @param workspaces the exact workspace IDs the record should hold ({@code null}/empty clears membership) + * @param sourceSeqNo the source-doc write's seq_no (monotonic guard) + * @param listener notified {@code true} if membership changed, else {@code false} (stale, in sync, or record + * missing after retries) + */ + public void reconcileWorkspaces( + String resourceIndex, + String resourceId, + Set workspaces, + long sourceSeqNo, + ActionListener listener + ) { + Set target = workspaces == null ? Set.of() : new HashSet<>(workspaces); + reconcileWorkspacesAttempt(getSharingIndex(resourceIndex), resourceId, target, sourceSeqNo, 1, listener); + } + + private void reconcileWorkspacesAttempt( + String resourceSharingIndex, + String resourceId, + Set target, + long sourceSeqNo, + int attempt, + ActionListener listener + ) { + try (ThreadContext.StoredContext ctx = this.threadPool.getThreadContext().stashContext()) { + client.get(new GetRequest(resourceSharingIndex).id(resourceId), ActionListener.wrap(getResponse -> { + ctx.restore(); + if (!getResponse.isExists()) { + // The record is created asynchronously; a reconcile from an immediate follow-up write can arrive + // first. Retry rather than treating a missing record as synchronized. + if (attempt < WORKSPACE_RECONCILE_MAX_ATTEMPTS) { + threadPool.schedule( + () -> reconcileWorkspacesAttempt(resourceSharingIndex, resourceId, target, sourceSeqNo, attempt + 1, listener), + WORKSPACE_RECONCILE_RETRY_DELAY, + ThreadPool.Names.GENERIC + ); + } else { + // Fail loud: the record never appeared, so the write-path record may be out of step with the + // resource's workspaces. Re-run the migrate API to repair once the record exists. + LOGGER.error( + "Sharing record [{}] still missing after {} attempts; workspaces left unreconciled. Re-run " + + "POST _plugins/_security/api/resources/migrate to repair.", + resourceId, + attempt + ); + listener.onResponse(false); + } + return; + } + + Map source = getResponse.getSourceAsMap(); + long storedSeqNo = source.get(WORKSPACES_SEQ_NO_FIELD) instanceof Number n + ? n.longValue() + : SequenceNumbers.UNASSIGNED_SEQ_NO; + // Monotonic guard: an older source operation must never overwrite state written by a newer one. + if (sourceSeqNo <= storedSeqNo) { + listener.onResponse(false); + return; + } + + Set current = workspacesFromSource(source); + boolean contentChanged = !current.equals(target); + try (ThreadContext.StoredContext ctx2 = this.threadPool.getThreadContext().stashContext()) { + Map doc = new HashMap<>(); + doc.put("workspaces", new ArrayList<>(target)); + doc.put(WORKSPACES_SEQ_NO_FIELD, sourceSeqNo); + // WAIT_UNTIL (not IMMEDIATE): write-path reads are realtime GET/mget, so a forced refresh per + // reconcile is unnecessary; wait for the next scheduled refresh instead. + UpdateRequest ur = client.prepareUpdate(resourceSharingIndex, resourceId) + .setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL) + .setDoc(doc) + .setIfSeqNo(getResponse.getSeqNo()) + .setIfPrimaryTerm(getResponse.getPrimaryTerm()) + .request(); + client.update(ur, ActionListener.wrap(updateResponse -> { + ctx2.restore(); + listener.onResponse(contentChanged); + }, e -> { + ctx2.restore(); + // A concurrent reconcile won the compare-and-set; re-read and re-evaluate the guard. + if (ExceptionsHelper.unwrapCause(e) instanceof VersionConflictEngineException + && attempt < WORKSPACE_RECONCILE_MAX_ATTEMPTS) { + reconcileWorkspacesAttempt(resourceSharingIndex, resourceId, target, sourceSeqNo, attempt + 1, listener); + } else { + listener.onFailure(e); + } + })); + } + }, listener::onFailure)); + } + } + + private static Set workspacesFromSource(Map source) { + Object v = source == null ? null : source.get("workspaces"); + Set result = new HashSet<>(); + if (v instanceof Collection c) { + for (Object o : c) { + if (o != null) { + result.add(o.toString()); + } + } + } else if (v instanceof String s && !s.isEmpty()) { + result.add(s); + } + return result; + } + + /** * @param resourceId the unique identifier of the resource document to update * @param resourceIndex the name of the index containing the resource * @param principals the list of principals (e.g. {@code user:alice}, {@code role:admin}) @@ -264,22 +382,19 @@ public void indexResourceSharing(String resourceIndex, ResourceSharing sharingIn ActionListener irListener = ActionListener.wrap(idxResponse -> { ctx.restore(); LOGGER.info("Successfully created {} entry for resource {} in index {}.", resourceSharingIndex, resourceId, resourceIndex); - updateResourceVisibility( - resourceId, - resourceIndex, - List.of("user:" + createdBy.getUsername()), - ActionListener.wrap((updateResponse) -> { - LOGGER.debug( - "postUpdate: Successfully updated visibility for resource {} within index {}", - resourceId, - resourceIndex - ); - listener.onResponse(sharingInfo); - }, (e) -> { - LOGGER.error("Failed to create principals field in [{}] for resource [{}]", resourceIndex, resourceId, e); - listener.onResponse(sharingInfo); - }) - ); + // Seed all_shared_principals from getAllPrincipals() (creator + any share recipients); fall back to + // the creator when empty. + List initialPrincipals = new ArrayList<>(sharingInfo.getAllPrincipals()); + if (initialPrincipals.isEmpty()) { + initialPrincipals.add("user:" + createdBy.getUsername()); + } + updateResourceVisibility(resourceId, resourceIndex, initialPrincipals, ActionListener.wrap((updateResponse) -> { + LOGGER.debug("postUpdate: Successfully updated visibility for resource {} within index {}", resourceId, resourceIndex); + listener.onResponse(sharingInfo); + }, (e) -> { + LOGGER.error("Failed to create principals field in [{}] for resource [{}]", resourceIndex, resourceId, e); + listener.onResponse(sharingInfo); + })); }, (e) -> { if (ExceptionsHelper.unwrapCause(e) instanceof VersionConflictEngineException) { // already exists → skipping @@ -550,6 +665,67 @@ private void processScrollIds( * } * */ + /** + * Fetches multiple resource-sharing records from the sharing index for {@code resourceIndex} in a single + * {@link MultiGetRequest} round-trip, rather than one {@link #fetchSharingInfo} GET per id. + *

+ * This is used on the authorization path when a resource inherits access from a set of container resources + * (e.g. the workspaces it belongs to): all containers live in the same sharing index with known ids, so a single + * mget avoids an N+1 sequential-GET pattern on the privilege hot path. + *

+ * Records that do not exist or fail to parse are simply omitted from the result map; the operation only fails if the + * mget itself fails. + * + * @param resourceIndex the source resource index whose sharing index should be queried + * @param resourceIds the ids of the sharing records to fetch + * @param listener notified with a map of {@code resourceId -> ResourceSharing} for the records that exist + */ + public void fetchSharingInfoForIds( + String resourceIndex, + Collection resourceIds, + ActionListener> listener + ) { + if (StringUtils.isBlank(resourceIndex) || resourceIds == null || resourceIds.isEmpty()) { + listener.onResponse(Collections.emptyMap()); + return; + } + String resourceSharingIndex = getSharingIndex(resourceIndex); + + try (ThreadContext.StoredContext ctx = this.threadPool.getThreadContext().stashContext()) { + final MultiGetRequest mget = new MultiGetRequest(); + for (String id : resourceIds) { + mget.add(new MultiGetRequest.Item(resourceSharingIndex, id)); + } + + client.multiGet(mget, ActionListener.wrap(mres -> { + ctx.restore(); + Map records = new HashMap<>(); + for (MultiGetItemResponse item : mres.getResponses()) { + if (item == null || item.isFailed()) continue; + final GetResponse gr = item.getResponse(); + if (gr == null || !gr.isExists()) continue; + try ( + XContentParser parser = XContentType.JSON.xContent() + .createParser(NamedXContentRegistry.EMPTY, LoggingDeprecationHandler.INSTANCE, gr.getSourceAsString()) + ) { + parser.nextToken(); + ResourceSharing rs = ResourceSharing.fromXContent(parser); + rs.setResourceId(gr.getId()); + records.put(gr.getId(), rs); + } catch (Exception ex) { + LOGGER.warn("Failed to parse resource-sharing doc id={} from {}", gr.getId(), resourceSharingIndex, ex); + } + } + listener.onResponse(records); + }, exception -> { + ctx.restore(); + String failureResponse = "Something went wrong while batch-fetching resource sharing records from " + resourceSharingIndex; + LOGGER.error(failureResponse, exception); + listener.onFailure(new OpenSearchStatusException(failureResponse, RestStatus.INTERNAL_SERVER_ERROR)); + })); + } + } + public void fetchSharingInfo(String resourceIndex, String resourceId, ActionListener listener) { if (StringUtils.isBlank(resourceIndex) || StringUtils.isBlank(resourceId)) { listener.onFailure(new IllegalArgumentException("resourceIndex and resourceId must not be null or empty")); @@ -636,62 +812,139 @@ public void fetchSharingInfo(String resourceIndex, String resourceId, ActionList * @throws RuntimeException if there's an error during the update operation */ public void share(String resourceId, String resourceIndex, ShareWith shareWith, ActionListener listener) { - StepListener sharingInfoListener = new StepListener<>(); - - // Fetch resource sharing doc - fetchSharingInfo(resourceIndex, resourceId, sharingInfoListener); + shareAttempt(resourceId, resourceIndex, shareWith, 1, listener); + } - // build update script - sharingInfoListener.whenComplete(sharingInfo -> { - if (sharingInfo == null) { + private void shareAttempt( + String resourceId, + String resourceIndex, + ShareWith shareWith, + int attempt, + ActionListener listener + ) { + // Re-fetch on each attempt so a version conflict re-applies the mutation to the LATEST record (picking up any + // concurrent workspace reconcile) rather than reverting to a stale snapshot. + fetchSharingInfoWithVersion(resourceIndex, resourceId, ActionListener.wrap(versioned -> { + if (versioned == null) { LOGGER.debug("No sharing record found for resource {}", resourceId); listener.onResponse(null); return; } + ResourceSharing sharingInfo = versioned.sharing(); for (String accessLevel : shareWith.accessLevels()) { - Recipients target = shareWith.atAccessLevel(accessLevel); - sharingInfo.share(accessLevel, target); + sharingInfo.share(accessLevel, shareWith.atAccessLevel(accessLevel)); } if (shareWith.getGeneralAccess() != null) { sharingInfo.setGeneralAccess(shareWith.getGeneralAccess()); } + indexSharingRecordWithCas( + resourceIndex, + sharingInfo, + versioned.seqNo(), + versioned.primaryTerm(), + () -> shareAttempt(resourceId, resourceIndex, shareWith, attempt + 1, listener), + attempt, + listener + ); + }, listener::onFailure)); + } - String resourceSharingIndex = getSharingIndex(resourceIndex); - try (ThreadContext.StoredContext ctx = threadPool.getThreadContext().stashContext()) { - IndexRequest ir = client.prepareIndex(resourceSharingIndex) - .setId(sharingInfo.getResourceId()) - .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE) - .setSource(sharingInfo.toXContent(jsonBuilder(), ToXContent.EMPTY_PARAMS)) - .setOpType(DocWriteRequest.OpType.INDEX) - .request(); + /** + * Read-modify-write of a whole sharing record guarded by optimistic concurrency. The write only succeeds if the + * record hasn't changed since {@code seqNo}/{@code primaryTerm} were read; on a version conflict it invokes + * {@code onConflictRetry} (which re-fetches and re-applies the mutation) so a concurrent workspace reconcile is + * never clobbered. On success it refreshes {@code all_shared_principals} and returns the mutated record. + */ + private void indexSharingRecordWithCas( + String resourceIndex, + ResourceSharing sharingInfo, + long seqNo, + long primaryTerm, + Runnable onConflictRetry, + int attempt, + ActionListener listener + ) throws IOException { + String resourceSharingIndex = getSharingIndex(resourceIndex); + String resourceId = sharingInfo.getResourceId(); + try (ThreadContext.StoredContext ctx = threadPool.getThreadContext().stashContext()) { + IndexRequest ir = client.prepareIndex(resourceSharingIndex) + .setId(resourceId) + .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE) + .setSource(sharingInfo.toXContent(jsonBuilder(), ToXContent.EMPTY_PARAMS)) + .setOpType(DocWriteRequest.OpType.INDEX) + .setIfSeqNo(seqNo) + .setIfPrimaryTerm(primaryTerm) + .request(); - ActionListener irListener = ActionListener.wrap(idxResponse -> { - ctx.restore(); - LOGGER.info( - "Successfully updated {} entry for resource {} in index {}.", - resourceSharingIndex, - resourceId, - resourceIndex - ); - updateResourceVisibility( - resourceId, - resourceIndex, - sharingInfo.getAllPrincipals(), - ActionListener.wrap((updateResponse) -> { - LOGGER.debug("Successfully updated visibility for resource {} within index {}", resourceId, resourceIndex); - listener.onResponse(sharingInfo); - }, (e) -> { - LOGGER.error("Failed to update principals field in [{}] for resource [{}]", resourceIndex, resourceId, e); - listener.onResponse(sharingInfo); - }) + client.index(ir, ActionListener.wrap(idxResponse -> { + ctx.restore(); + LOGGER.info("Successfully updated {} entry for resource {} in index {}.", resourceSharingIndex, resourceId, resourceIndex); + updateResourceVisibility( + resourceId, + resourceIndex, + sharingInfo.getAllPrincipals(), + ActionListener.wrap((updateResponse) -> { + LOGGER.debug("Successfully updated visibility for resource {} within index {}", resourceId, resourceIndex); + listener.onResponse(sharingInfo); + }, (e) -> { + LOGGER.error("Failed to update principals field in [{}] for resource [{}]", resourceIndex, resourceId, e); + listener.onResponse(sharingInfo); + }) + ); + }, e -> { + ctx.restore(); + // A concurrent write (e.g. a workspace reconcile) changed the record; re-fetch and re-apply. + if (ExceptionsHelper.unwrapCause(e) instanceof VersionConflictEngineException && attempt < SHARING_UPDATE_MAX_ATTEMPTS) { + onConflictRetry.run(); + } else { + LOGGER.error(e.getMessage()); + listener.onFailure(e); + } + })); + } + } + + /** + * Like {@link #fetchSharingInfo} but also captures the document's {@code _seq_no}/{@code _primary_term} for a + * subsequent optimistic-concurrency write. Responds with {@code null} when no record exists. + */ + private void fetchSharingInfoWithVersion(String resourceIndex, String resourceId, ActionListener listener) { + if (StringUtils.isBlank(resourceIndex) || StringUtils.isBlank(resourceId)) { + listener.onFailure(new IllegalArgumentException("resourceIndex and resourceId must not be null or empty")); + return; + } + String resourceSharingIndex = getSharingIndex(resourceIndex); + try (ThreadContext.StoredContext ctx = this.threadPool.getThreadContext().stashContext()) { + client.get(new GetRequest(resourceSharingIndex).id(resourceId), ActionListener.wrap(getResponse -> { + ctx.restore(); + if (!getResponse.isExists()) { + listener.onResponse(null); + return; + } + try ( + XContentParser parser = XContentType.JSON.xContent() + .createParser(NamedXContentRegistry.EMPTY, LoggingDeprecationHandler.INSTANCE, getResponse.getSourceAsString()) + ) { + parser.nextToken(); + ResourceSharing resourceSharing = ResourceSharing.fromXContent(parser); + resourceSharing.setResourceId(getResponse.getId()); + listener.onResponse( + new VersionedResourceSharing(resourceSharing, getResponse.getSeqNo(), getResponse.getPrimaryTerm()) ); - }, (failResponse) -> { - LOGGER.error(failResponse.getMessage()); - listener.onFailure(failResponse); - }); - client.index(ir, irListener); - } - }, listener::onFailure); + } catch (Exception e) { + String failure = "Failed to parse sharing record " + resourceId; + LOGGER.error(failure, e); + listener.onFailure(new OpenSearchStatusException(failure, RestStatus.INTERNAL_SERVER_ERROR, e)); + } + }, e -> { + ctx.restore(); + listener.onFailure(e); + })); + } + } + + /** A parsed sharing record together with the document version fields needed for an optimistic-concurrency write. */ + private record VersionedResourceSharing(ResourceSharing sharing, long seqNo, long primaryTerm) { } /** @@ -714,15 +967,28 @@ public void patchSharingInfo( String generalAccess, ActionListener listener ) { + patchAttempt(resourceId, resourceIndex, add, revoke, generalAccessPresent, generalAccess, 1, listener); + } - StepListener sharingInfoListener = new StepListener<>(); - String resourceSharingIndex = getSharingIndex(resourceIndex); - - // Fetch the current ResourceSharing document - fetchSharingInfo(resourceIndex, resourceId, sharingInfoListener); - - // Apply patch and update the document - sharingInfoListener.whenComplete(sharingInfo -> { + private void patchAttempt( + String resourceId, + String resourceIndex, + ShareWith add, + ShareWith revoke, + boolean generalAccessPresent, + String generalAccess, + int attempt, + ActionListener listener + ) { + // Re-fetch on each attempt so a version conflict re-applies the patch to the LATEST record (picking up any + // concurrent workspace reconcile) rather than reverting to a stale snapshot. + fetchSharingInfoWithVersion(resourceIndex, resourceId, ActionListener.wrap(versioned -> { + if (versioned == null) { + LOGGER.debug("No sharing record found for resource {}", resourceId); + listener.onResponse(null); + return; + } + ResourceSharing sharingInfo = versioned.sharing(); if (add != null) { sharingInfo.applyAdd(add); } @@ -732,44 +998,16 @@ public void patchSharingInfo( if (generalAccessPresent) { sharingInfo.setGeneralAccess(generalAccess); } - - try (ThreadContext.StoredContext ctx = this.threadPool.getThreadContext().stashContext()) { - // update the record - IndexRequest ir = client.prepareIndex(resourceSharingIndex) - .setId(resourceId) - .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE) - .setSource(sharingInfo.toXContent(jsonBuilder(), ToXContent.EMPTY_PARAMS)) - .setOpType(DocWriteRequest.OpType.INDEX) - .request(); - - client.index(ir, ActionListener.wrap(idxResponse -> { - ctx.restore(); - LOGGER.info( - "Successfully updated {} resource sharing info for resource {} in index {}.", - resourceSharingIndex, - resourceId, - resourceIndex - ); - - updateResourceVisibility( - resourceId, - resourceIndex, - sharingInfo.getAllPrincipals(), - ActionListener.wrap((updateResponse) -> { - LOGGER.debug("Successfully updated visibility for resource {} within index {}", resourceId, resourceIndex); - listener.onResponse(sharingInfo); - }, (e) -> { - LOGGER.error("Failed to update principals field in [{}] for resource [{}]", resourceIndex, resourceId, e); - listener.onResponse(sharingInfo); - }) - ); - - }, (e) -> { - LOGGER.error(e.getMessage()); - listener.onFailure(e); - })); - } - }, listener::onFailure); + indexSharingRecordWithCas( + resourceIndex, + sharingInfo, + versioned.seqNo(), + versioned.primaryTerm(), + () -> patchAttempt(resourceId, resourceIndex, add, revoke, generalAccessPresent, generalAccess, attempt + 1, listener), + attempt, + listener + ); + }, listener::onFailure)); } /** diff --git a/src/main/java/org/opensearch/security/resources/api/migrate/MigrateResourceSharingInfoApiAction.java b/src/main/java/org/opensearch/security/resources/api/migrate/MigrateResourceSharingInfoApiAction.java index 3db6bea0c2..dadb7234fb 100644 --- a/src/main/java/org/opensearch/security/resources/api/migrate/MigrateResourceSharingInfoApiAction.java +++ b/src/main/java/org/opensearch/security/resources/api/migrate/MigrateResourceSharingInfoApiAction.java @@ -88,7 +88,7 @@ * default_access_level: "" // optional: overrides the default access-level defined in resource-access-levels.yml * } * - Response: - * 200 OK Migration Complete. migrated %d; skippedNoType %s; skippedExisting %s; failed %d // migrate -> successful migration count, skippedNoType -> records with no type, skippedExisting -> records that were already migrated, failed -> records that failed to migrate + * 200 OK Migration Complete. migrated %d; backfilledExisting %d; skippedNoType %s; skippedExisting %s; failed %d // migrate -> newly created records, backfilledExisting -> pre-existing records that gained workspace membership, skippedNoType -> records with no type, skippedExisting -> records already migrated with nothing to backfill, failed -> records that failed to migrate */ public class MigrateResourceSharingInfoApiAction extends AbstractApiAction { @@ -246,7 +246,9 @@ private ValidationResult loadCurrentSharingInfo(RestRequest Scroll scroll = new Scroll(TimeValue.timeValueMinutes(1L)); SearchRequest searchRequest = new SearchRequest(sourceIndex).scroll(scroll) .source( - new SearchSourceBuilder().query(QueryBuilders.matchAllQuery()).size(1_000) // batch size per scroll “page” + new SearchSourceBuilder().query(QueryBuilders.matchAllQuery()) + .size(1_000) // batch size per scroll “page” + .seqNoAndPrimaryTerm(true) // source-doc seq_no is the monotonic guard for workspace reconcile ); // 2) execute first search @@ -291,6 +293,8 @@ private ValidationResult loadCurrentSharingInfo(RestRequest // Extract parent ID if the provider declares a parentIdField String parentId = null; + // Workspace IDs, if the provider declares a workspaces field (see extractWorkspaces). + Set workspaces = Collections.emptySet(); if (type != null) { ResourceProvider hitProvider = resourcePluginInfo.getResourceProvider(type); if (hitProvider != null && hitProvider.parentIdField() != null) { @@ -299,9 +303,12 @@ private ValidationResult loadCurrentSharingInfo(RestRequest parentId = null; } } + if (hitProvider != null && hitProvider.workspacesField() != null) { + workspaces = extractWorkspaces(rec, hitProvider.workspacesField()); + } } - results.add(new SourceDoc(id, username, backendRoles, type, parentId)); + results.add(new SourceDoc(id, username, backendRoles, type, parentId, workspaces, hit.getSeqNo())); } // 4) fetch next batch SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId).scroll(scroll); @@ -345,6 +352,7 @@ private ValidationResult loadCurrentSharingInfo(RestRequest private ValidationResult createNewSharingRecords(ValidationResultArg sourceInfo) throws IOException { AtomicInteger migratedCount = new AtomicInteger(); AtomicInteger skippedExisting = new AtomicInteger(); + AtomicInteger backfilledExisting = new AtomicInteger(); AtomicInteger failureCount = new AtomicInteger(); // Thread-safe sets that we can mutate directly from listeners @@ -403,6 +411,8 @@ private ValidationResult createNewSharingRecords(ValidationResul } // 5) index the new record + final Set docWorkspaces = doc.workspaces; + final long docSeqNo = doc.seqNo; ActionListener listener = ActionListener.wrap(entry -> { if (entry != null) { LOGGER.debug( @@ -412,15 +422,30 @@ private ValidationResult createNewSharingRecords(ValidationResul sourceInfo.sourceIndex ); migratedCount.getAndIncrement(); + migrationStatsLatch.countDown(); } else { - LOGGER.debug( - "Skipping migration of resource sharing record for resource {} within index {} as an entry already exists", + // Record already exists: reconcile its workspaces to exactly match the source doc (adds and + // removals), bringing pre-existing records up to date. The source doc's seq_no is the monotonic + // guard, so a concurrent live update is never overwritten by this migration. + sharingIndexHandler.reconcileWorkspaces( + sourceInfo.sourceIndex, resourceId, - sourceInfo.sourceIndex + docWorkspaces, + docSeqNo, + ActionListener.wrap(changed -> { + if (Boolean.TRUE.equals(changed)) { + backfilledExisting.getAndIncrement(); + } else { + skippedExisting.getAndIncrement(); + } + migrationStatsLatch.countDown(); + }, e -> { + LOGGER.warn("Failed to reconcile workspaces for existing record [{}]: {}", resourceId, e.getMessage()); + failureCount.getAndIncrement(); + migrationStatsLatch.countDown(); + }) ); - skippedExisting.getAndIncrement(); } - migrationStatsLatch.countDown(); }, e -> { LOGGER.debug(e.getMessage()); failureCount.getAndIncrement(); @@ -435,6 +460,10 @@ private ValidationResult createNewSharingRecords(ValidationResul if (doc.parentId != null && provider.parentType() != null) { sharingBuilder.parentId(doc.parentId).parentType(provider.parentType()); } + // Carry the source doc's workspaces onto the record (used by the write-path fan-out). + if (doc.workspaces != null && !doc.workspaces.isEmpty()) { + sharingBuilder.workspaces(doc.workspaces); + } ResourceSharing sharingInfo = sharingBuilder.build(); sharingIndexHandler.indexResourceSharing(sourceInfo.sourceIndex, sharingInfo, listener); @@ -454,8 +483,9 @@ private ValidationResult createNewSharingRecords(ValidationResul } String summary = String.format( - "Migration complete. migrated %d; skippedNoType %s; skippedExisting %s; failed %d", + "Migration complete. migrated %d; backfilledExisting %d; skippedNoType %s; skippedExisting %s; failed %d", migratedCount.get(), + backfilledExisting.get(), skippedNoType.size(), skippedExisting.get(), failureCount.get() @@ -570,6 +600,36 @@ static String jsonPointer(String path) { return path.startsWith("/") ? path : ("/" + path.replace(".", "/")); } + /** + * Reads workspace IDs from a source document at {@code workspacesField} (a JSON array, or a single string). + * Blank ids are ignored. Package-private for testability. + * + * @param rec the parsed source document + * @param workspacesField the provider-declared field path (dot-notation or JSON pointer) + * @return the workspace IDs, or empty if the field is absent/empty + */ + static Set extractWorkspaces(JsonNode rec, String workspacesField) { + if (workspacesField == null) { + return Collections.emptySet(); + } + JsonNode wsNode = rec.at(jsonPointer(workspacesField)); + Set workspaces = new HashSet<>(); + if (wsNode.isArray()) { + for (JsonNode ws : wsNode) { + addIfPresent(workspaces, ws.asText(null)); + } + } else if (wsNode.isTextual()) { + addIfPresent(workspaces, wsNode.asText(null)); + } + return workspaces; + } + + private static void addIfPresent(Set set, String value) { + if (value != null && !value.isEmpty()) { + set.add(value); + } + } + /** * Determine a document's resource type using, in order: *

    @@ -605,7 +665,8 @@ static String classifyDocType( .orElse(null); } - record SourceDoc(String resourceId, String username, List backendRoles, String type, String parentId) { + record SourceDoc(String resourceId, String username, List backendRoles, String type, String parentId, Set workspaces, + long seqNo) { } record ValidationResultArg(String sourceIndex, String defaultOwnerName, Map typeToDefaultAccessLevel, List< diff --git a/src/main/java/org/opensearch/security/resources/sharing/ResourceSharing.java b/src/main/java/org/opensearch/security/resources/sharing/ResourceSharing.java index 52e1dc0cab..9bc44b73f6 100644 --- a/src/main/java/org/opensearch/security/resources/sharing/ResourceSharing.java +++ b/src/main/java/org/opensearch/security/resources/sharing/ResourceSharing.java @@ -22,10 +22,12 @@ import org.apache.logging.log4j.Logger; import org.opensearch.core.common.io.stream.NamedWriteable; +import org.opensearch.core.common.io.stream.StreamInput; import org.opensearch.core.common.io.stream.StreamOutput; import org.opensearch.core.xcontent.ToXContentFragment; import org.opensearch.core.xcontent.XContentBuilder; import org.opensearch.core.xcontent.XContentParser; +import org.opensearch.index.seqno.SequenceNumbers; import org.opensearch.security.user.User; /** @@ -46,6 +48,9 @@ public class ResourceSharing implements ToXContentFragment, NamedWriteable { private final Logger log = LogManager.getLogger(this.getClass()); + /** NamedWriteable name; used both by {@link #getWriteableName()} and the registry entry. */ + public static final String NAME = "resource_sharing"; + /** * The unique identifier of the resource and the resource sharing entry */ @@ -77,6 +82,24 @@ public class ResourceSharing implements ToXContentFragment, NamedWriteable { */ private String parentId; + /** + * The set of workspace IDs this resource belongs to. + * + *

    A single resource may belong to multiple workspaces, so this is a set (unlike {@link #tenant} and + * {@link #parentId}, which are single-valued). Empty for non-workspace resources, which keeps the field + * additive and preserves existing behavior. Used by the write-path access-level fan-out + * ({@code ResourceAccessHandler}) to locate each workspace's sharing record. Read-path visibility is handled + * by filtering the resource's own {@code workspaces} field in DLS, not via {@code all_shared_principals}. + */ + private Set workspaces; + + /** + * Monotonic guard: the source-doc seq_no that last set {@link #workspaces}, so an older reconcile can't overwrite + * a newer one. Modeled here (not just written raw) so it survives whole-record rewrites in share/revoke/patch. + * Defaults to {@link SequenceNumbers#UNASSIGNED_SEQ_NO}. + */ + private long workspacesSeqNo; + /** * Information about who created the resource */ @@ -93,10 +116,29 @@ private ResourceSharing(Builder b) { this.tenant = b.tenant; this.parentType = b.parentType; this.parentId = b.parentId; + this.workspaces = b.workspaces; + this.workspacesSeqNo = b.workspacesSeqNo; this.createdBy = b.createdBy; this.shareWith = b.shareWith; } + /** + * Stream constructor, symmetric with {@link #writeTo(StreamOutput)}. Registered as the reader for the + * {@code resource_sharing} NamedWriteable so {@code readNamedWriteable(ResourceSharing.class)} round-trips. + */ + public ResourceSharing(StreamInput in) throws IOException { + this.resourceId = in.readString(); + this.resourceType = in.readString(); + this.tenant = in.readOptionalString(); + this.parentType = in.readOptionalString(); + this.parentId = in.readOptionalString(); + this.createdBy = new CreatedBy(in); + this.shareWith = in.readBoolean() ? new ShareWith(in) : null; + List ws = in.readOptionalStringList(); + this.workspaces = ws == null ? null : new HashSet<>(ws); + this.workspacesSeqNo = in.readZLong(); + } + public static Builder builder() { return new Builder(); } @@ -133,6 +175,18 @@ public String getParentId() { return parentId; } + public Set getWorkspaces() { + return workspaces == null ? Collections.emptySet() : workspaces; + } + + public void setWorkspaces(Set workspaces) { + this.workspaces = workspaces; + } + + public long getWorkspacesSeqNo() { + return workspacesSeqNo; + } + public void share(String accessLevel, Recipients target) { if (shareWith == null) { Map recs = new HashMap<>(); @@ -191,13 +245,14 @@ public boolean equals(Object o) { && Objects.equals(tenant, that.tenant) && Objects.equals(parentType, that.parentType) && Objects.equals(parentId, that.parentId) + && Objects.equals(getWorkspaces(), that.getWorkspaces()) && Objects.equals(createdBy, that.createdBy) && Objects.equals(shareWith, that.shareWith); } @Override public int hashCode() { - return Objects.hash(resourceId, resourceType, tenant, parentType, parentId, createdBy, shareWith); + return Objects.hash(resourceId, resourceType, tenant, parentType, parentId, getWorkspaces(), createdBy, shareWith); } @Override @@ -218,6 +273,8 @@ public String toString() { + ", parentId='" + parentId + '\'' + + ", workspaces=" + + workspaces + ", createdBy=" + createdBy + ", shareWith=" @@ -227,7 +284,7 @@ public String toString() { @Override public String getWriteableName() { - return "resource_sharing"; + return NAME; } @Override @@ -244,6 +301,12 @@ public void writeTo(StreamOutput out) throws IOException { } else { out.writeBoolean(false); } + // No version guard needed: workspaces ships within the resource-sharing feature (introduced in 3.3), + // which is not yet GA, so there is no older node that speaks the old wire format without this field. + // The symmetric read lives in the ResourceSharing(StreamInput) constructor, registered as the + // resource_sharing NamedWriteable in OpenSearchSecurityPlugin#getNamedWriteables. + out.writeOptionalStringCollection(workspaces == null ? null : new ArrayList<>(workspaces)); + out.writeZLong(workspacesSeqNo); } @Override @@ -260,6 +323,12 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws if (parentId != null) { builder.field("parent_id", parentId); } + if (workspaces != null && !workspaces.isEmpty()) { + builder.field("workspaces", workspaces); + } + if (workspacesSeqNo != SequenceNumbers.UNASSIGNED_SEQ_NO) { + builder.field("workspaces_seq_no", workspacesSeqNo); + } if (shareWith != null) { builder.field("share_with"); shareWith.toXContent(builder, params); @@ -308,6 +377,22 @@ public static ResourceSharing fromXContent(XContentParser parser) throws IOExcep b.parentId(parser.text()); } break; + case "workspaces": + if (token == XContentParser.Token.START_ARRAY) { + Set ws = new HashSet<>(); + while (parser.nextToken() != XContentParser.Token.END_ARRAY) { + ws.add(parser.text()); + } + b.workspaces(ws); + } else if (token == XContentParser.Token.VALUE_NULL) { + b.workspaces(null); + } + break; + case "workspaces_seq_no": + if (token != XContentParser.Token.VALUE_NULL) { + b.workspacesSeqNo(parser.longValue()); + } + break; case "created_by": b.createdBy(CreatedBy.fromXContent(parser)); break; @@ -434,6 +519,9 @@ public List getAllPrincipals() { principals.add("user:" + createdBy.getUsername()); } + // Workspace membership is not a principal: DLS filters the resource's own `workspaces` field instead + // (see ResourceSharingDlsUtils). This list stays usernames/roles only. + // Add shared recipients if (shareWith != null) { if (shareWith.isPublic()) { @@ -472,6 +560,8 @@ public static final class Builder { private String tenant; private String parentType; private String parentId; + private Set workspaces; + private long workspacesSeqNo = SequenceNumbers.UNASSIGNED_SEQ_NO; private CreatedBy createdBy; private ShareWith shareWith; @@ -500,6 +590,16 @@ public Builder parentId(String parentId) { return this; } + public Builder workspaces(Set workspaces) { + this.workspaces = workspaces; + return this; + } + + public Builder workspacesSeqNo(long workspacesSeqNo) { + this.workspacesSeqNo = workspacesSeqNo; + return this; + } + public Builder createdBy(CreatedBy createdBy) { this.createdBy = createdBy; return this; diff --git a/src/test/java/org/opensearch/security/resources/ResourceAccessHandlerTests.java b/src/test/java/org/opensearch/security/resources/ResourceAccessHandlerTests.java index 2ddfd150a5..e8fc913522 100644 --- a/src/test/java/org/opensearch/security/resources/ResourceAccessHandlerTests.java +++ b/src/test/java/org/opensearch/security/resources/ResourceAccessHandlerTests.java @@ -36,6 +36,7 @@ import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.eq; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -158,6 +159,165 @@ public void testHasPermission_noAccessLevelsDenied() { verify(listener).onResponse(false); } + @Test + public void testHasPermission_grantedViaWorkspaceMembership() { + // Resource itself grants the user nothing, but it belongs to workspace "ws-1" and the user has + // "read" access on that workspace's own sharing record -> access is inherited from the workspace container. + User user = new User("erin", ImmutableSet.of("roleA"), ImmutableSet.of("backendA"), null, ImmutableMap.of(), false); + injectUser(user); + when(adminDNs.isAdmin(user)).thenReturn(false); + + final String workspaceIndex = "workspace-index"; + final String workspaceId = "ws-1"; + when(resourcePluginInfo.indexByType("workspace")).thenReturn(workspaceIndex); + + // The resource: no direct access, belongs to ws-1, not created by the user. + ResourceSharing resourceDoc = mock(ResourceSharing.class); + when(resourceDoc.isCreatedBy("erin")).thenReturn(false); + when(resourceDoc.getAccessLevelsForUser(user)).thenReturn(Collections.emptySet()); + when(resourceDoc.getWorkspaces()).thenReturn(Set.of(workspaceId)); + + // The workspace record: shares "read" with the user. + ResourceSharing workspaceDoc = mock(ResourceSharing.class); + when(workspaceDoc.isCreatedBy("erin")).thenReturn(false); + when(workspaceDoc.getAccessLevelsForUser(user)).thenReturn(Set.of("read")); + + FlattenedActionGroups ag = mock(FlattenedActionGroups.class); + when(resourcePluginInfo.flattenedForType("workspace")).thenReturn(ag); + when(ag.resolve(any())).thenReturn(ImmutableSet.of("read")); + + doAnswer(inv -> { + ActionListener l = inv.getArgument(2); + l.onResponse(resourceDoc); + return null; + }).when(sharingIndexHandler).fetchSharingInfo(eq(INDEX), eq(RESOURCE_ID), any()); + + // Workspaces are resolved in a single batched mget, not per-workspace GETs. + doAnswer(inv -> { + ActionListener> l = inv.getArgument(2); + l.onResponse(java.util.Map.of(workspaceId, workspaceDoc)); + return null; + }).when(sharingIndexHandler).fetchSharingInfoForIds(eq(workspaceIndex), any(), any()); + + ActionListener listener = mock(ActionListener.class); + handler.hasPermission(RESOURCE_ID, TYPE, ACTION, listener); + + verify(listener).onResponse(true); + } + + @Test + public void testHasPermission_deniedWhenNoWorkspaceGrantsAccess() { + // Resource grants nothing and belongs to a workspace the user has no access on -> denied. + User user = new User("frank", ImmutableSet.of("roleA"), ImmutableSet.of("backendA"), null, ImmutableMap.of(), false); + injectUser(user); + when(adminDNs.isAdmin(user)).thenReturn(false); + + final String workspaceIndex = "workspace-index"; + final String workspaceId = "ws-9"; + when(resourcePluginInfo.indexByType("workspace")).thenReturn(workspaceIndex); + + ResourceSharing resourceDoc = mock(ResourceSharing.class); + when(resourceDoc.isCreatedBy("frank")).thenReturn(false); + when(resourceDoc.getAccessLevelsForUser(user)).thenReturn(Collections.emptySet()); + when(resourceDoc.getParentId()).thenReturn(null); + when(resourceDoc.getWorkspaces()).thenReturn(Set.of(workspaceId)); + + ResourceSharing workspaceDoc = mock(ResourceSharing.class); + when(workspaceDoc.isCreatedBy("frank")).thenReturn(false); + when(workspaceDoc.getAccessLevelsForUser(user)).thenReturn(Collections.emptySet()); + + doAnswer(inv -> { + ActionListener l = inv.getArgument(2); + l.onResponse(resourceDoc); + return null; + }).when(sharingIndexHandler).fetchSharingInfo(eq(INDEX), eq(RESOURCE_ID), any()); + + doAnswer(inv -> { + ActionListener> l = inv.getArgument(2); + l.onResponse(java.util.Map.of(workspaceId, workspaceDoc)); + return null; + }).when(sharingIndexHandler).fetchSharingInfoForIds(eq(workspaceIndex), any(), any()); + + ActionListener listener = mock(ActionListener.class); + handler.hasPermission(RESOURCE_ID, TYPE, ACTION, listener); + + verify(listener).onResponse(false); + } + + @Test + public void testHasPermission_dissociatedResourceLosesWorkspaceGrant() { + // A workspace record exists that WOULD grant the action, but the resource has been dissociated from it: its + // own workspace set is empty. checkContainers must not fan out to any workspace, so access is denied. This is + // the write-path half of associate/dissociate consistency -- a stale membership would leak authorization. + User user = new User("heidi", ImmutableSet.of("roleA"), ImmutableSet.of("backendA"), null, ImmutableMap.of(), false); + injectUser(user); + when(adminDNs.isAdmin(user)).thenReturn(false); + + // The resource: no direct access, no parent, and NO workspaces (dissociated). + ResourceSharing resourceDoc = mock(ResourceSharing.class); + when(resourceDoc.isCreatedBy("heidi")).thenReturn(false); + when(resourceDoc.getAccessLevelsForUser(user)).thenReturn(Collections.emptySet()); + when(resourceDoc.getParentId()).thenReturn(null); + when(resourceDoc.getWorkspaces()).thenReturn(Collections.emptySet()); + + doAnswer(inv -> { + ActionListener l = inv.getArgument(2); + l.onResponse(resourceDoc); + return null; + }).when(sharingIndexHandler).fetchSharingInfo(eq(INDEX), eq(RESOURCE_ID), any()); + + ActionListener listener = mock(ActionListener.class); + handler.hasPermission(RESOURCE_ID, TYPE, ACTION, listener); + + verify(listener).onResponse(false); + // With no workspaces on the resource, the workspace index is never queried. + verify(sharingIndexHandler, never()).fetchSharingInfoForIds(any(), any(), any()); + } + + @Test + public void testHasPermission_workspaceIsLeafEvaluatedNoRecursion() { + // Workspaces are evaluated as leaves (their own share_with) and never recursed into, so even a malformed + // self-referential workspace terminates: the resource belongs to "ws-loop" which grants nothing, so access + // is denied without following ws-loop's own workspaces. + User user = new User("gwen", ImmutableSet.of("roleA"), ImmutableSet.of("backendA"), null, ImmutableMap.of(), false); + injectUser(user); + when(adminDNs.isAdmin(user)).thenReturn(false); + + final String workspaceIndex = "workspace-index"; + final String loopWs = "ws-loop"; + when(resourcePluginInfo.indexByType("workspace")).thenReturn(workspaceIndex); + + // Resource: no direct access, belongs to ws-loop. + ResourceSharing resourceDoc = mock(ResourceSharing.class); + when(resourceDoc.isCreatedBy("gwen")).thenReturn(false); + when(resourceDoc.getAccessLevelsForUser(user)).thenReturn(Collections.emptySet()); + when(resourceDoc.getParentId()).thenReturn(null); + when(resourceDoc.getWorkspaces()).thenReturn(Set.of(loopWs)); + + // Workspace ws-loop: grants nothing and (malformed) contains itself. + ResourceSharing loopDoc = mock(ResourceSharing.class); + when(loopDoc.isCreatedBy("gwen")).thenReturn(false); + when(loopDoc.getAccessLevelsForUser(user)).thenReturn(Collections.emptySet()); + + doAnswer(inv -> { + ActionListener l = inv.getArgument(2); + l.onResponse(resourceDoc); + return null; + }).when(sharingIndexHandler).fetchSharingInfo(eq(INDEX), eq(RESOURCE_ID), any()); + + doAnswer(inv -> { + ActionListener> l = inv.getArgument(2); + l.onResponse(java.util.Map.of(loopWs, loopDoc)); + return null; + }).when(sharingIndexHandler).fetchSharingInfoForIds(eq(workspaceIndex), any(), any()); + + ActionListener listener = mock(ActionListener.class); + handler.hasPermission(RESOURCE_ID, TYPE, ACTION, listener); + + // Must terminate (no StackOverflow / infinite loop) and deny. + verify(listener).onResponse(false); + } + @Test public void testHasPermission_nullDocumentDenied() { User user = new User("dave", ImmutableSet.of("x"), ImmutableSet.of("y"), null, ImmutableMap.of(), false); diff --git a/src/test/java/org/opensearch/security/resources/ResourcePluginInfoTests.java b/src/test/java/org/opensearch/security/resources/ResourcePluginInfoTests.java index 988e65f89e..182a7f7108 100644 --- a/src/test/java/org/opensearch/security/resources/ResourcePluginInfoTests.java +++ b/src/test/java/org/opensearch/security/resources/ResourcePluginInfoTests.java @@ -18,6 +18,7 @@ import org.junit.Before; import org.junit.Test; +import org.opensearch.OpenSearchSecurityException; import org.opensearch.index.engine.Engine; import org.opensearch.index.mapper.ParsedDocument; import org.opensearch.security.spi.resources.ResourceProvider; @@ -26,6 +27,8 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -102,8 +105,78 @@ public void testUnknownIndexReturnsNull() { assertNull(result); } + @Test + public void testRejectsConflictingWorkspacesFieldsOnSameIndex() { + // Two providers sharing an index but declaring different workspaces fields is a misconfiguration: DLS can only + // filter one field per index. Registration must reject it rather than silently pick one. + ResourceSharingExtension extension = new ResourceSharingExtension() { + @Override + public Set getResourceProviders() { + var providers = new java.util.LinkedHashSet(); + providers.add(workspacesProvider("a", ".shared-index", "workspaces")); + providers.add(workspacesProvider("b", ".shared-index", "ws")); + return providers; + } + + @Override + public void assignResourceSharingClient(ResourceSharingClient client) {} + }; + + OpenSearchSecurityException ex = assertThrows( + OpenSearchSecurityException.class, + () -> resourcePluginInfo.setResourceSharingExtensions(Set.of(extension)) + ); + assertTrue(ex.getMessage().contains("Conflicting workspaces fields")); + } + + @Test + public void testAllowsMatchingWorkspacesFieldsOnSameIndex() { + // Providers sharing an index that agree on the workspaces field (or opt out with null) are accepted. + ResourceSharingExtension extension = new ResourceSharingExtension() { + @Override + public Set getResourceProviders() { + var providers = new java.util.LinkedHashSet(); + providers.add(workspacesProvider("a", ".shared-index", "workspaces")); + providers.add(workspacesProvider("b", ".shared-index", "workspaces")); + providers.add(workspacesProvider("c", ".shared-index", null)); + return providers; + } + + @Override + public void assignResourceSharingClient(ResourceSharingClient client) {} + }; + + resourcePluginInfo.setResourceSharingExtensions(Set.of(extension)); + resourcePluginInfo.updateProtectedTypes(Arrays.asList("a", "b", "c")); + assertEquals("workspaces", resourcePluginInfo.workspacesFieldForIndex(".shared-index")); + } + // ─── Helpers ───────────────────────────────────────────────────────────────── + private ResourceProvider workspacesProvider(String type, String index, String field) { + return new ResourceProvider() { + @Override + public String resourceType() { + return type; + } + + @Override + public String resourceIndexName() { + return index; + } + + @Override + public String typeField() { + return "resource_type"; + } + + @Override + public String workspacesField() { + return field; + } + }; + } + private void registerProviders(List types, String indexName, String sharedTypeField) { ResourceSharingExtension extension = new ResourceSharingExtension() { @Override @@ -224,6 +297,149 @@ private Engine.Index mockWorkflowDoc() { ); } + // ---------- extractMultiValuedFieldFromIndexOp -------------------------------------------------- + + @Test + public void extractMultiValued_collectsAllValuesOfAField() { + Engine.Index indexOp = mockIndexOp( + new StringField("workspaces", "ws-a", Field.Store.NO), + new StringField("workspaces", "ws-b", Field.Store.NO), + new StringField("name", "n", Field.Store.NO) + ); + assertEquals(Set.of("ws-a", "ws-b"), ResourcePluginInfo.extractMultiValuedFieldFromIndexOp("workspaces", indexOp)); + } + + @Test + public void extractMultiValued_emptyWhenFieldAbsent() { + Engine.Index indexOp = mockIndexOp(new StringField("name", "n", Field.Store.NO)); + assertTrue(ResourcePluginInfo.extractMultiValuedFieldFromIndexOp("workspaces", indexOp).isEmpty()); + } + + // ---------- workspacesFieldForIndex ------------------------------------------------------------ + + @Test + public void workspacesFieldForIndex_returnsDeclaredFieldOrNull() { + ResourceSharingExtension ext = new ResourceSharingExtension() { + @Override + public Set getResourceProviders() { + return Set.of(new ResourceProvider() { + @Override + public String resourceType() { + return "dashboard"; + } + + @Override + public String resourceIndexName() { + return ".kibana"; + } + + @Override + public String workspacesField() { + return "ws"; + } + }); + } + + @Override + public void assignResourceSharingClient(ResourceSharingClient client) {} + }; + resourcePluginInfo.setResourceSharingExtensions(Set.of(ext)); + + assertEquals("ws", resourcePluginInfo.workspacesFieldForIndex(".kibana")); + assertNull(resourcePluginInfo.workspacesFieldForIndex(".other-index")); + } + + // ---------- resolveWorkspacesForUser (SPI seam) -------------------------------------------------- + + @Test + public void resolveWorkspaces_returnsEmptyWhenNoExtensionsRegistered() { + // No extensions -> nothing to resolve; must not throw and must not add workspace principals. + Set result = resourcePluginInfo.resolveWorkspacesForUser(mockUser("alice", Set.of("r1"), Set.of())); + assertEquals(java.util.Collections.emptySet(), result); + } + + @Test + public void resolveWorkspaces_returnsEmptyWhenExtensionUsesDefaultImplementation() { + // An extension that does not override the resolver -> empty (default returns Collections.emptySet()). + // This is what preserves BWC for all existing plugins. + registerProviders(List.of("monitor"), ".alerting-config", null); + Set result = resourcePluginInfo.resolveWorkspacesForUser(mockUser("alice", Set.of("r1"), Set.of())); + assertEquals(java.util.Collections.emptySet(), result); + } + + @Test + public void resolveWorkspaces_aggregatesAcrossExtensions() { + // Two extensions both override; the aggregator unions their contributions. + ResourceSharingExtension a = extensionWithResolver((u, sr, br) -> Set.of("ws-a1", "ws-a2")); + ResourceSharingExtension b = extensionWithResolver((u, sr, br) -> Set.of("ws-a2", "ws-b1")); + resourcePluginInfo.setResourceSharingExtensions(Set.of(a, b)); + + Set result = resourcePluginInfo.resolveWorkspacesForUser(mockUser("alice", Set.of("r1"), Set.of())); + assertEquals(Set.of("ws-a1", "ws-a2", "ws-b1"), result); + } + + @Test + public void resolveWorkspaces_forwardsUserFieldsToResolver() { + // Extensions receive the user's name + role sets so a trusted server-set resolver can map on them. + java.util.concurrent.atomic.AtomicReference capturedName = new java.util.concurrent.atomic.AtomicReference<>(); + java.util.concurrent.atomic.AtomicReference> capturedSecRoles = new java.util.concurrent.atomic.AtomicReference<>(); + java.util.concurrent.atomic.AtomicReference> capturedBackendRoles = new java.util.concurrent.atomic.AtomicReference<>(); + ResourceSharingExtension ext = extensionWithResolver((u, sr, br) -> { + capturedName.set(u); + capturedSecRoles.set(sr); + capturedBackendRoles.set(br); + return Set.of("ws-x"); + }); + resourcePluginInfo.setResourceSharingExtensions(Set.of(ext)); + + resourcePluginInfo.resolveWorkspacesForUser(mockUser("bob", Set.of("analyst"), Set.of("ldap-eng"))); + + assertEquals("bob", capturedName.get()); + assertEquals(Set.of("analyst"), capturedSecRoles.get()); + assertEquals(Set.of("ldap-eng"), capturedBackendRoles.get()); + } + + @Test + public void resolveWorkspaces_toleratesNullOrEmptyContributions() { + // An extension returning null (contract-violating but reachable) must not NPE the aggregation. + ResourceSharingExtension nullContrib = extensionWithResolver((u, sr, br) -> null); + ResourceSharingExtension emptyContrib = extensionWithResolver((u, sr, br) -> java.util.Collections.emptySet()); + ResourceSharingExtension realContrib = extensionWithResolver((u, sr, br) -> Set.of("ws-real")); + resourcePluginInfo.setResourceSharingExtensions(Set.of(nullContrib, emptyContrib, realContrib)); + + Set result = resourcePluginInfo.resolveWorkspacesForUser(mockUser("alice", Set.of(), Set.of())); + assertEquals(Set.of("ws-real"), result); + } + + @FunctionalInterface + private interface WorkspaceResolver { + Set resolve(String username, Set securityRoles, Set backendRoles); + } + + private ResourceSharingExtension extensionWithResolver(WorkspaceResolver resolver) { + return new ResourceSharingExtension() { + @Override + public Set getResourceProviders() { + return Set.of(); + } + + @Override + public void assignResourceSharingClient(ResourceSharingClient client) {} + + @Override + public Set resolveWorkspacesForUser(String username, Set securityRoles, Set backendRoles) { + return resolver.resolve(username, securityRoles, backendRoles); + } + }; + } + + private org.opensearch.security.user.User mockUser(String name, Set securityRoles, Set backendRoles) { + // Use a real User (not a mock) — User is a plain class and Mockito default-mocks trip on it. + return new org.opensearch.security.user.User(name).withRoles(backendRoles).withSecurityRoles(securityRoles); + } + + // ---------- fixtures -------------------------------------------------------------------------- + private Engine.Index mockIndexOp(IndexableField... fields) { Engine.Index indexOp = mock(Engine.Index.class); ParsedDocument parsedDoc = mock(ParsedDocument.class); diff --git a/src/test/java/org/opensearch/security/resources/ResourceSharingIndexHandlerTests.java b/src/test/java/org/opensearch/security/resources/ResourceSharingIndexHandlerTests.java new file mode 100644 index 0000000000..10fd61869c --- /dev/null +++ b/src/test/java/org/opensearch/security/resources/ResourceSharingIndexHandlerTests.java @@ -0,0 +1,319 @@ +/* + * 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.resources; + +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.Before; +import org.junit.Test; + +import org.opensearch.action.get.GetRequest; +import org.opensearch.action.get.GetResponse; +import org.opensearch.action.get.MultiGetItemResponse; +import org.opensearch.action.get.MultiGetRequest; +import org.opensearch.action.get.MultiGetResponse; +import org.opensearch.action.index.IndexRequest; +import org.opensearch.action.index.IndexRequestBuilder; +import org.opensearch.action.index.IndexResponse; +import org.opensearch.action.update.UpdateRequest; +import org.opensearch.action.update.UpdateRequestBuilder; +import org.opensearch.action.update.UpdateResponse; +import org.opensearch.common.settings.Settings; +import org.opensearch.common.unit.TimeValue; +import org.opensearch.common.util.concurrent.ThreadContext; +import org.opensearch.common.xcontent.XContentHelper; +import org.opensearch.common.xcontent.XContentType; +import org.opensearch.core.action.ActionListener; +import org.opensearch.core.common.bytes.BytesArray; +import org.opensearch.core.index.Index; +import org.opensearch.core.index.shard.ShardId; +import org.opensearch.core.xcontent.XContentBuilder; +import org.opensearch.index.engine.VersionConflictEngineException; +import org.opensearch.index.get.GetResult; +import org.opensearch.security.resources.sharing.Recipient; +import org.opensearch.security.resources.sharing.Recipients; +import org.opensearch.security.resources.sharing.ResourceSharing; +import org.opensearch.security.resources.sharing.ShareWith; +import org.opensearch.threadpool.ThreadPool; +import org.opensearch.transport.client.Client; + +import org.mockito.ArgumentCaptor; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.atLeast; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Unit tests for the workspace-related read/write helpers on {@link ResourceSharingIndexHandler}: + * {@link ResourceSharingIndexHandler#fetchSharingInfoForIds} and + * {@link ResourceSharingIndexHandler#reconcileWorkspaces}. + */ +public class ResourceSharingIndexHandlerTests { + + private static final String RESOURCE_INDEX = "test-index"; + + private Client client; + private ThreadPool threadPool; + private ResourceSharingIndexHandler handler; + + @Before + public void setUp() { + client = mock(Client.class); + threadPool = mock(ThreadPool.class); + when(threadPool.getThreadContext()).thenReturn(new ThreadContext(Settings.EMPTY)); + handler = new ResourceSharingIndexHandler(client, threadPool, mock(ResourcePluginInfo.class)); + } + + // reconcileWorkspaces reads the sharing record's raw source (workspaces + workspaces_seq_no) and its + // seq_no/primary_term for the optimistic-concurrency guard; stub client.get to return the record. + private void stubRecordGet(boolean exists, String recordJson) { + doAnswer(inv -> { + ActionListener l = inv.getArgument(1); + GetResult getResult = mock(GetResult.class); + when(getResult.getId()).thenReturn("res-1"); + when(getResult.isExists()).thenReturn(exists); + if (exists) { + byte[] bytes = recordJson.getBytes(StandardCharsets.UTF_8); + BytesArray source = new BytesArray(bytes, 0, bytes.length); + when(getResult.sourceRef()).thenReturn(source); + when(getResult.sourceAsString()).thenReturn(recordJson); + when(getResult.sourceAsMap()).thenReturn(XContentHelper.convertToMap(source, false, XContentType.JSON).v2()); + when(getResult.getSeqNo()).thenReturn(1L); + when(getResult.getPrimaryTerm()).thenReturn(1L); + } + l.onResponse(new GetResponse(getResult)); + return null; + }).when(client).get(any(GetRequest.class), any()); + } + + private MultiGetItemResponse existingItem(String id, String sourceJson) { + GetResult getResult = mock(GetResult.class); + when(getResult.getId()).thenReturn(id); + when(getResult.isExists()).thenReturn(true); + byte[] bytes = sourceJson.getBytes(StandardCharsets.UTF_8); + when(getResult.sourceRef()).thenReturn(new BytesArray(bytes, 0, bytes.length)); + when(getResult.sourceAsString()).thenReturn(sourceJson); + return new MultiGetItemResponse(new GetResponse(getResult), null); + } + + private void stubUpdateSucceeds() { + // The update paths use the fluent client.prepareUpdate(idx,id).setRefreshPolicy(..).setDoc(..).request() + // builder; RETURNS_SELF makes every builder call return the same mock, and request() yields a mock request. + UpdateRequestBuilder builder = mock(UpdateRequestBuilder.class, org.mockito.Answers.RETURNS_SELF); + when(builder.request()).thenReturn(mock(UpdateRequest.class)); + when(client.prepareUpdate(anyString(), anyString())).thenReturn(builder); + doAnswer(inv -> { + ActionListener l = inv.getArgument(1); + l.onResponse(mock(UpdateResponse.class)); + return null; + }).when(client).update(any(UpdateRequest.class), any()); + } + + // ---------- fetchSharingInfoForIds ------------------------------------------------------------- + + @Test + public void fetchSharingInfoForIds_returnsEmptyForBlankIndexOrNoIds() { + AtomicReference> out = new AtomicReference<>(); + handler.fetchSharingInfoForIds(RESOURCE_INDEX, List.of(), ActionListener.wrap(out::set, e -> {})); + assertTrue(out.get().isEmpty()); + + out.set(null); + handler.fetchSharingInfoForIds(" ", List.of("a"), ActionListener.wrap(out::set, e -> {})); + assertTrue(out.get().isEmpty()); + + // no client call should have been issued + verify(client, never()).multiGet(any(), any()); + } + + @Test + public void fetchSharingInfoForIds_parsesExistingAndSkipsMissing() { + doAnswer(inv -> { + ActionListener l = inv.getArgument(1); + MultiGetItemResponse exists = existingItem("res-1", "{\"resource_id\":\"res-1\",\"created_by\":{\"user\":\"alice\"}}"); + GetResult missingResult = mock(GetResult.class); + when(missingResult.getId()).thenReturn("res-2"); + when(missingResult.isExists()).thenReturn(false); + MultiGetItemResponse missing = new MultiGetItemResponse(new GetResponse(missingResult), null); + l.onResponse(new MultiGetResponse(new MultiGetItemResponse[] { exists, missing })); + return null; + }).when(client).multiGet(any(MultiGetRequest.class), any()); + + AtomicReference> out = new AtomicReference<>(); + handler.fetchSharingInfoForIds(RESOURCE_INDEX, List.of("res-1", "res-2"), ActionListener.wrap(out::set, e -> {})); + + assertEquals(1, out.get().size()); + assertTrue(out.get().containsKey("res-1")); + assertEquals("alice", out.get().get("res-1").getCreatedBy().getUsername()); + } + + // ---------- reconcileWorkspaces ---------------------------------------------------------------- + + @Test + public void reconcile_appliesWhenNewerSeqNo() { + // No prior guard on the record (workspaces_seq_no absent) -> any source seq_no applies. + stubRecordGet(true, "{\"resource_id\":\"res-1\",\"created_by\":{\"user\":\"alice\"}}"); + stubUpdateSucceeds(); + + AtomicReference out = new AtomicReference<>(); + handler.reconcileWorkspaces(RESOURCE_INDEX, "res-1", Set.of("ws-a", "ws-b"), 5L, ActionListener.wrap(out::set, e -> {})); + + assertTrue(out.get()); + verify(client, times(1)).update(any(UpdateRequest.class), any()); + } + + @Test + public void reconcile_removesWhenDissociated() { + stubRecordGet( + true, + "{\"resource_id\":\"res-1\",\"created_by\":{\"user\":\"alice\"},\"workspaces\":[\"ws-a\",\"ws-b\"],\"workspaces_seq_no\":1}" + ); + stubUpdateSucceeds(); + + AtomicReference out = new AtomicReference<>(); + handler.reconcileWorkspaces(RESOURCE_INDEX, "res-1", Set.of("ws-a"), 5L, ActionListener.wrap(out::set, e -> {})); + + assertTrue(out.get()); + verify(client, times(1)).update(any(UpdateRequest.class), any()); + } + + @Test + public void reconcile_clearsWhenTargetEmpty() { + stubRecordGet( + true, + "{\"resource_id\":\"res-1\",\"created_by\":{\"user\":\"alice\"},\"workspaces\":[\"ws-a\"],\"workspaces_seq_no\":1}" + ); + stubUpdateSucceeds(); + + AtomicReference out = new AtomicReference<>(); + handler.reconcileWorkspaces(RESOURCE_INDEX, "res-1", Set.of(), 5L, ActionListener.wrap(out::set, e -> {})); + + assertTrue(out.get()); + verify(client, times(1)).update(any(UpdateRequest.class), any()); + } + + @Test + public void reconcile_rejectsStaleSeqNo() { + // Monotonic guard: a reconcile older than the last-applied source seq_no must not overwrite newer state. + stubRecordGet( + true, + "{\"resource_id\":\"res-1\",\"created_by\":{\"user\":\"alice\"},\"workspaces\":[\"ws-a\"],\"workspaces_seq_no\":5}" + ); + + AtomicReference out = new AtomicReference<>(); + handler.reconcileWorkspaces(RESOURCE_INDEX, "res-1", Set.of(), 3L, ActionListener.wrap(out::set, e -> {})); + + assertFalse(out.get()); + verify(client, never()).update(any(), any()); + } + + @Test + public void reconcile_advancesGuardWhenContentUnchanged() { + // Content already matches, but a newer seq_no still advances the guard so a later stale reconcile is gated. + stubRecordGet( + true, + "{\"resource_id\":\"res-1\",\"created_by\":{\"user\":\"alice\"},\"workspaces\":[\"ws-a\"],\"workspaces_seq_no\":1}" + ); + stubUpdateSucceeds(); + + AtomicReference out = new AtomicReference<>(); + handler.reconcileWorkspaces(RESOURCE_INDEX, "res-1", Set.of("ws-a"), 5L, ActionListener.wrap(out::set, e -> {})); + + assertFalse(out.get()); // content unchanged + verify(client, times(1)).update(any(UpdateRequest.class), any()); // but the guard was advanced + } + + @Test + public void reconcile_retriesWhenRecordMissing() { + // The record is created asynchronously; a reconcile that finds it missing must schedule a retry rather than + // treat it as synced. (threadPool.schedule is a no-op mock here, so the retry itself does not fire.) + stubRecordGet(false, null); + + handler.reconcileWorkspaces(RESOURCE_INDEX, "res-1", Set.of("ws-a"), 5L, ActionListener.wrap(b -> {}, e -> {})); + + verify(threadPool).schedule(any(Runnable.class), any(TimeValue.class), anyString()); + verify(client, never()).update(any(), any()); + } + + @Test + public void share_reappliesToLatestRecordOnVersionConflict() { + // Interleave: share fetches a stale snapshot ([ws-a], guard 10); before it writes, a reconcile has moved the + // record to []/guard 11 (simulated by a version conflict on the first optimistic-concurrency write). share + // MUST re-fetch the newer record and re-apply share_with to it, so the newer (empty) workspaces and guard 11 + // survive rather than being reverted to [ws-a]/10. + AtomicInteger getCount = new AtomicInteger(); + doAnswer(inv -> { + ActionListener l = inv.getArgument(1); + boolean first = getCount.getAndIncrement() == 0; + String json = first + ? "{\"resource_id\":\"res-1\",\"resource_type\":\"s\",\"created_by\":{\"user\":\"owner\"},\"workspaces\":[\"ws-a\"],\"workspaces_seq_no\":10}" + : "{\"resource_id\":\"res-1\",\"resource_type\":\"s\",\"created_by\":{\"user\":\"owner\"},\"workspaces_seq_no\":11}"; + GetResult gr = mock(GetResult.class); + when(gr.getId()).thenReturn("res-1"); + when(gr.isExists()).thenReturn(true); + when(gr.sourceAsString()).thenReturn(json); + when(gr.getSeqNo()).thenReturn(first ? 10L : 11L); + when(gr.getPrimaryTerm()).thenReturn(1L); + l.onResponse(new GetResponse(gr)); + return null; + }).when(client).get(any(GetRequest.class), any()); + + IndexRequestBuilder indexBuilder = mock(IndexRequestBuilder.class, org.mockito.Answers.RETURNS_SELF); + when(indexBuilder.request()).thenReturn(mock(IndexRequest.class)); + when(client.prepareIndex(anyString())).thenReturn(indexBuilder); + AtomicInteger indexCount = new AtomicInteger(); + doAnswer(inv -> { + ActionListener l = inv.getArgument(1); + if (indexCount.getAndIncrement() == 0) { + // First write loses to a concurrent reconcile. + l.onFailure(new VersionConflictEngineException(new ShardId(new Index("i", "u"), 0), "res-1", "conflict")); + } else { + l.onResponse(mock(IndexResponse.class)); + } + return null; + }).when(client).index(any(IndexRequest.class), any()); + stubUpdateSucceeds(); // updateResourceVisibility on success + + ShareWith shareWith = new ShareWith(Map.of("read", new Recipients(Map.of(Recipient.USERS, Set.of("bob"))))); + AtomicReference out = new AtomicReference<>(); + handler.share("res-1", RESOURCE_INDEX, shareWith, ActionListener.wrap(out::set, e -> {})); + + verify(client, times(2)).get(any(GetRequest.class), any()); // re-fetched the latest on conflict + verify(client, times(2)).index(any(IndexRequest.class), any()); + + // Each write's optimistic-concurrency guard came from the record fetched that attempt (stale 10, then 11). + verify(indexBuilder).setIfSeqNo(10L); + verify(indexBuilder).setIfSeqNo(11L); + verify(indexBuilder, times(2)).setIfPrimaryTerm(1L); + + ArgumentCaptor src = ArgumentCaptor.forClass(XContentBuilder.class); + verify(indexBuilder, atLeast(2)).setSource(src.capture()); + String latest = src.getAllValues().get(src.getAllValues().size() - 1).toString(); + assertFalse("stale workspace must not be re-applied over the newer state", latest.contains("ws-a")); + assertTrue("newer guard must be preserved", latest.contains("\"workspaces_seq_no\":11")); + assertTrue("share_with recipient must be reapplied to the latest record", latest.contains("bob")); + + // The share completed successfully after the retry. + assertNotNull(out.get()); + } +} diff --git a/src/test/java/org/opensearch/security/resources/api/migrate/MigrateResourceSharingInfoApiActionTests.java b/src/test/java/org/opensearch/security/resources/api/migrate/MigrateResourceSharingInfoApiActionTests.java index 9b64459d14..8665f2201b 100644 --- a/src/test/java/org/opensearch/security/resources/api/migrate/MigrateResourceSharingInfoApiActionTests.java +++ b/src/test/java/org/opensearch/security/resources/api/migrate/MigrateResourceSharingInfoApiActionTests.java @@ -135,6 +135,37 @@ public void classifyReturnsNullWhenNothingResolvableAndIndexUnknown() throws Exc assertNull(result); } + @Test + public void extractWorkspacesReadsAllArrayValues() throws Exception { + JsonNode doc = mapper.readTree("{ \"workspaces\": [\"ws-a\", \"ws-b\", \"ws-c\"] }"); + assertEquals(java.util.Set.of("ws-a", "ws-b", "ws-c"), MigrateResourceSharingInfoApiAction.extractWorkspaces(doc, "workspaces")); + } + + @Test + public void extractWorkspacesToleratesSingleScalarValue() throws Exception { + JsonNode doc = mapper.readTree("{ \"workspaces\": \"ws-only\" }"); + assertEquals(java.util.Set.of("ws-only"), MigrateResourceSharingInfoApiAction.extractWorkspaces(doc, "workspaces")); + } + + @Test + public void extractWorkspacesReturnsEmptyWhenFieldAbsent() throws Exception { + JsonNode doc = mapper.readTree("{ \"other\": 1 }"); + assertEquals(Collections.emptySet(), MigrateResourceSharingInfoApiAction.extractWorkspaces(doc, "workspaces")); + } + + @Test + public void extractWorkspacesIgnoresBlankIdsAndNullField() throws Exception { + JsonNode doc = mapper.readTree("{ \"workspaces\": [\"ws-a\", \"\"] }"); + assertEquals(java.util.Set.of("ws-a"), MigrateResourceSharingInfoApiAction.extractWorkspaces(doc, "workspaces")); + assertEquals(Collections.emptySet(), MigrateResourceSharingInfoApiAction.extractWorkspaces(doc, null)); + } + + @Test + public void extractWorkspacesSupportsDotNotationPath() throws Exception { + JsonNode doc = mapper.readTree("{ \"meta\": { \"workspaces\": [\"ws-a\"] } }"); + assertEquals(java.util.Set.of("ws-a"), MigrateResourceSharingInfoApiAction.extractWorkspaces(doc, "meta.workspaces")); + } + @Test public void jsonPointerAcceptsDotNotation() { assertEquals("/monitor/user/name", MigrateResourceSharingInfoApiAction.jsonPointer("monitor.user.name")); diff --git a/src/test/java/org/opensearch/security/resources/sharing/ResourceSharingTests.java b/src/test/java/org/opensearch/security/resources/sharing/ResourceSharingTests.java index 04a22cb8df..f11a2bf0f7 100644 --- a/src/test/java/org/opensearch/security/resources/sharing/ResourceSharingTests.java +++ b/src/test/java/org/opensearch/security/resources/sharing/ResourceSharingTests.java @@ -17,7 +17,9 @@ import org.apache.lucene.tests.util.LuceneTestCase; import org.junit.Test; +import org.opensearch.common.io.stream.BytesStreamOutput; import org.opensearch.common.xcontent.json.JsonXContent; +import org.opensearch.core.common.io.stream.StreamInput; import org.opensearch.core.xcontent.XContentParser; import static org.junit.Assert.assertEquals; @@ -341,4 +343,184 @@ public void fromXContent_parsesTopLevelTenant() throws Exception { assertEquals("owner", sharing.getCreatedBy().getUsername()); } } + + // --- Workspace-awareness ------------------------------------------------------------------------ + + @Test + public void getWorkspaces_defaultsToEmptyWhenAbsent() { + ResourceSharing rs = ResourceSharing.builder().resourceId("r").createdBy(mockCreatedBy("owner")).build(); + assertNotNull(rs.getWorkspaces()); + assertTrue(rs.getWorkspaces().isEmpty()); + } + + @Test + public void getAllPrincipals_doesNotProjectWorkspaces() { + // Workspace membership is NOT denormalized into all_shared_principals; read-path visibility uses the + // resource's own `workspaces` field in DLS instead. getAllPrincipals stays usernames/roles only. + ResourceSharing rs = ResourceSharing.builder() + .resourceId("dash-1") + .resourceType("dashboard") + .createdBy(mockCreatedBy("owner")) + .workspaces(new HashSet<>(Set.of("ws-analytics", "ws-executive"))) + .build(); + + List principals = rs.getAllPrincipals(); + assertTrue(principals.contains("user:owner")); + assertTrue(principals.stream().noneMatch(p -> p.startsWith("workspace:"))); + } + + @Test + public void toXContent_omitsWorkspacesWhenEmpty_andRoundTripsWhenPresent() throws Exception { + // Uses a real CreatedBy (not a mock) because this test drives the real toXContent serialization. + // Empty -> field omitted (byte-identical to pre-change records). + ResourceSharing empty = ResourceSharing.builder() + .resourceId("r") + .resourceType("dashboard") + .createdBy(new CreatedBy("owner")) + .build(); + assertFalse(toJson(empty).contains("workspaces")); + + // Present -> serialized and round-trips through fromXContent. + ResourceSharing withWs = ResourceSharing.builder() + .resourceId("r") + .resourceType("dashboard") + .createdBy(new CreatedBy("owner")) + .workspaces(new HashSet<>(Set.of("ws-a", "ws-b"))) + .build(); + String json = toJson(withWs); + assertTrue(json.contains("workspaces")); + + try (XContentParser parser = JsonXContent.jsonXContent.createParser(null, null, json)) { + parser.nextToken(); + ResourceSharing parsed = ResourceSharing.fromXContent(parser); + assertEquals(Set.of("ws-a", "ws-b"), parsed.getWorkspaces()); + } + } + + // ResourceSharing is a ToXContentFragment that opens/closes its own object, so serialize with a bare + // builder rather than XContentHelper (which would open an outer object and double-wrap). + private static String toJson(ResourceSharing rs) throws Exception { + org.opensearch.core.xcontent.XContentBuilder builder = JsonXContent.contentBuilder(); + rs.toXContent(builder, org.opensearch.core.xcontent.ToXContent.EMPTY_PARAMS); + return builder.toString(); + } + + // Note: CreatedBy/ShareWith use identity equality (no value equals), so these assert fields explicitly + // rather than whole-object ResourceSharing equality. + @Test + public void streamSerialization_roundTripsIncludingWorkspaces() throws Exception { + ResourceSharing original = ResourceSharing.builder() + .resourceId("r1") + .resourceType("dashboard") + .tenant("t1") + .createdBy(new CreatedBy("owner")) + .workspaces(new HashSet<>(Set.of("ws-a", "ws-b"))) + .build(); + + try (BytesStreamOutput out = new BytesStreamOutput()) { + original.writeTo(out); + try (StreamInput in = out.bytes().streamInput()) { + ResourceSharing read = new ResourceSharing(in); + assertEquals("r1", read.getResourceId()); + assertEquals("t1", read.getTenant()); + assertEquals("owner", read.getCreatedBy().getUsername()); + assertEquals(Set.of("ws-a", "ws-b"), read.getWorkspaces()); + } + } + } + + @Test + public void streamSerialization_roundTripsWithNoWorkspaces() throws Exception { + ResourceSharing original = ResourceSharing.builder() + .resourceId("r1") + .resourceType("dashboard") + .createdBy(new CreatedBy("owner")) + .build(); + + try (BytesStreamOutput out = new BytesStreamOutput()) { + original.writeTo(out); + try (StreamInput in = out.bytes().streamInput()) { + ResourceSharing read = new ResourceSharing(in); + assertEquals("r1", read.getResourceId()); + assertEquals("owner", read.getCreatedBy().getUsername()); + assertTrue(read.getWorkspaces().isEmpty()); + } + } + } + + @Test + public void fromXContent_parsesWorkspacesArray() throws Exception { + String json = """ + { + "resource_id": "r1", + "resource_type": "dashboard", + "workspaces": ["ws-1", "ws-2"], + "created_by": { + "user": "owner" + } + } + """; + + try (XContentParser parser = JsonXContent.jsonXContent.createParser(null, null, json)) { + parser.nextToken(); + ResourceSharing sharing = ResourceSharing.fromXContent(parser); + assertEquals(Set.of("ws-1", "ws-2"), sharing.getWorkspaces()); + } + } + + @Test + public void workspacesSeqNo_survivesXContentRoundTrip() throws Exception { + // share()/revoke()/patch() re-index the whole record via toXContent -> fromXContent. The monotonic guard + // (workspaces_seq_no) MUST survive that round-trip, otherwise a share would reset it and let an older, + // still-retrying reconcile re-apply stale workspaces. + ResourceSharing rs = ResourceSharing.builder() + .resourceId("r") + .resourceType("dashboard") + .createdBy(new CreatedBy("owner")) + .workspaces(new HashSet<>(Set.of("ws-a"))) + .workspacesSeqNo(42L) + .build(); + + String json = toJson(rs); + assertTrue(json.contains("workspaces_seq_no")); + try (XContentParser parser = JsonXContent.jsonXContent.createParser(null, null, json)) { + parser.nextToken(); + ResourceSharing parsed = ResourceSharing.fromXContent(parser); + assertEquals(42L, parsed.getWorkspacesSeqNo()); + assertEquals(Set.of("ws-a"), parsed.getWorkspaces()); + } + } + + @Test + public void workspacesSeqNo_omittedWhenUnassigned() throws Exception { + // A record that has never been reconciled carries no guard field (stays byte-clean). + ResourceSharing rs = ResourceSharing.builder().resourceId("r").resourceType("dashboard").createdBy(new CreatedBy("owner")).build(); + assertFalse(toJson(rs).contains("workspaces_seq_no")); + + try (BytesStreamOutput out = new BytesStreamOutput()) { + rs.writeTo(out); + try (StreamInput in = out.bytes().streamInput()) { + ResourceSharing read = new ResourceSharing(in); + assertEquals(org.opensearch.index.seqno.SequenceNumbers.UNASSIGNED_SEQ_NO, read.getWorkspacesSeqNo()); + } + } + } + + @Test + public void streamSerialization_roundTripsWorkspacesSeqNo() throws Exception { + ResourceSharing original = ResourceSharing.builder() + .resourceId("r1") + .resourceType("dashboard") + .createdBy(new CreatedBy("owner")) + .workspaces(new HashSet<>(Set.of("ws-a"))) + .workspacesSeqNo(7L) + .build(); + + try (BytesStreamOutput out = new BytesStreamOutput()) { + original.writeTo(out); + try (StreamInput in = out.bytes().streamInput()) { + assertEquals(7L, new ResourceSharing(in).getWorkspacesSeqNo()); + } + } + } }