[ML] correctly validate permissions when retention policy is configured (#85413)

When a transform has a `retention_policy` it needs to be able to delete documents in the destination index. 

`create_index` does not necessitate that we can delete documents from it. So, even if we create the index, we need to verify that we can delete documents given the `retention_policy` definition.

This is not a crucial bug as the transform will simply fail later. Its nicer to fail sooner.

closes https://github.com/elastic/elasticsearch/issues/85409

docs/changelog/85413.yaml

@@ -0,0 +1,6 @@
+pr: 85413
+summary: Correctly validate permissions when retention policy is configured
+area: Transform
+type: bug
+issues:
+ - 85409

docs/reference/transform/apis/put-transform.asciidoc

@@ -22,7 +22,8 @@ Requires the following privileges:
 * cluster: `manage_transform` (the `transform_admin` built-in role grants this
   privilege)
 * source indices: `read`, `view_index_metadata`
-* destination index: `read`, `create_index`, `index`.
+* destination index: `read`, `create_index`, `index`. If a `retention_policy` is configured, the `delete` privilege is
+  also required.
 
 [[put-transform-desc]]
 == {api-description-title}

docs/reference/transform/apis/start-transform.asciidoc

@@ -22,6 +22,8 @@ Requires the following privileges:
 * cluster: `manage_transform` (the `transform_admin` built-in role grants this
   privilege)
 * source indices: `read`, `view_index_metadata`.
+* destination index: `read`, `create_index`, `index`. If a `retention_policy` is configured, the `delete` privilege is
+  also required.
 
 [[start-transform-desc]]
 == {api-description-title}

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/action/TransformPrivilegeChecker.java

@@ -20,6 +20,7 @@ import org.elasticsearch.xpack.core.security.action.user.HasPrivilegesResponse;
 import org.elasticsearch.xpack.core.security.authz.RoleDescriptor;
 import org.elasticsearch.xpack.core.security.authz.permission.ResourcePrivileges;
 import org.elasticsearch.xpack.core.security.support.Exceptions;
+import org.elasticsearch.xpack.core.transform.transforms.NullRetentionPolicyConfig;
 import org.elasticsearch.xpack.core.transform.transforms.TransformConfig;
 
 import java.util.ArrayList;
@@ -83,7 +84,7 @@ final class TransformPrivilegeChecker {
                 destIndex
             );
 
-            List<String> destPrivileges = new ArrayList<>(3);
+            List<String> destPrivileges = new ArrayList<>(4);
             destPrivileges.add("read");
             destPrivileges.add("index");
             // If the destination index does not exist, we can assume that we may have to create it on start.
@@ -91,6 +92,10 @@ final class TransformPrivilegeChecker {
             if (concreteDest.length == 0) {
                 destPrivileges.add("create_index");
             }
+            if (config.getRetentionPolicyConfig() != null
+                && config.getRetentionPolicyConfig() instanceof NullRetentionPolicyConfig == false) {
+                destPrivileges.add("delete");
+            }
             RoleDescriptor.IndicesPrivileges destIndexPrivileges = RoleDescriptor.IndicesPrivileges.builder()
                 .indices(destIndex)
                 .privileges(destPrivileges)

x-pack/plugin/transform/src/test/java/org/elasticsearch/xpack/transform/action/TransformPrivilegeCheckerTests.java

@@ -18,6 +18,7 @@ import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.metadata.Metadata;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.indices.TestIndexNameExpressionResolver;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.client.NoOpClient;
@@ -28,6 +29,7 @@ import org.elasticsearch.xpack.core.security.authz.RoleDescriptor;
 import org.elasticsearch.xpack.core.security.user.User;
 import org.elasticsearch.xpack.core.transform.transforms.DestConfig;
 import org.elasticsearch.xpack.core.transform.transforms.SourceConfig;
+import org.elasticsearch.xpack.core.transform.transforms.TimeRetentionPolicyConfig;
 import org.elasticsearch.xpack.core.transform.transforms.TransformConfig;
 import org.junit.After;
 import org.junit.Before;
@@ -149,6 +151,40 @@ public class TransformPrivilegeCheckerTests extends ESTestCase {
         );
     }
 
+    public void testCheckPrivileges_CheckDestIndexPrivileges_DestIndexExistsWithRetentionPolicy() {
+        ClusterState clusterState = ClusterState.builder(ClusterName.DEFAULT)
+            .metadata(
+                Metadata.builder()
+                    .put(IndexMetadata.builder(DEST_INDEX_NAME).settings(settings(Version.CURRENT)).numberOfShards(1).numberOfReplicas(0))
+            )
+            .build();
+        TransformConfig config = new TransformConfig.Builder(TRANSFORM_CONFIG).setRetentionPolicyConfig(
+            new TimeRetentionPolicyConfig("foo", TimeValue.timeValueDays(1))
+        ).build();
+        TransformPrivilegeChecker.checkPrivileges(
+            OPERATION_NAME,
+            securityContext,
+            indexNameExpressionResolver,
+            clusterState,
+            client,
+            config,
+            true,
+            ActionListener.wrap(aVoid -> {
+                HasPrivilegesRequest request = client.lastHasPrivilegesRequest;
+                assertThat(request.username(), is(equalTo(USER_NAME)));
+                assertThat(request.applicationPrivileges(), is(emptyArray()));
+                assertThat(request.clusterPrivileges(), is(emptyArray()));
+                assertThat(request.indexPrivileges(), is(arrayWithSize(2)));
+                RoleDescriptor.IndicesPrivileges sourceIndicesPrivileges = request.indexPrivileges()[0];
+                assertThat(sourceIndicesPrivileges.getIndices(), is(arrayContaining(SOURCE_INDEX_NAME)));
+                assertThat(sourceIndicesPrivileges.getPrivileges(), is(arrayContaining("read", "view_index_metadata")));
+                RoleDescriptor.IndicesPrivileges destIndicesPrivileges = request.indexPrivileges()[1];
+                assertThat(destIndicesPrivileges.getIndices(), is(arrayContaining(DEST_INDEX_NAME)));
+                assertThat(destIndicesPrivileges.getPrivileges(), is(arrayContaining("read", "index", "delete")));
+            }, e -> fail(e.getMessage()))
+        );
+    }
+
     private static class MyMockClient extends NoOpClient {
 
         private HasPrivilegesRequest lastHasPrivilegesRequest;