[Transform] Add multi-node tests for checking transform permissions

x-pack/plugin/src/yamlRestTest/resources/rest-api-spec/test/transform/transforms_unattended.yml

@@ -47,10 +47,27 @@ teardown:
               "unattended": true
             }
           }
+  - match: { acknowledged: true }
   - do:
-      transform.start_transform:
+      transform.get_transform_stats:
         transform_id: "transform-unattended"
+  - match: { count: 1 }
+  - match: { transforms.0.id: "transform-unattended" }
+  - match: { transforms.0.state: "/started|indexing|stopping|stopped/" }
+  - match: { transforms.0.health.status: "green" }
 
+  - do:
+      transform.start_transform:
+        transform_id: "transform-unattended"
+  - match: { acknowledged: true }
+  - do:
+      transform.get_transform_stats:
+        transform_id: "transform-unattended"
+  - match: { count: 1 }
+  - match: { transforms.0.id: "transform-unattended" }
+  - match: { transforms.0.state: "/started|indexing|stopping|stopped/" }
+  - match: { transforms.0.health.status: "yellow" }
+  - match: { transforms.0.health.issues.0.details: "Validation Failed: 1: no such index [airline-data];" }
 
 ---
 "Test unattended put and start wildcard":
@@ -69,6 +86,8 @@ teardown:
               "unattended": true
             }
           }
+  - match: { acknowledged: true }
   - do:
       transform.start_transform:
         transform_id: "transform-unattended"
+  - match: { acknowledged: true }

x-pack/plugin/transform/qa/multi-node-tests/build.gradle

@@ -45,5 +45,8 @@ testClusters.matching { it.name == 'javaRestTest' }.configureEach {
 
   extraConfigFile nodeKey.name, nodeKey
   extraConfigFile nodeCert.name, nodeCert
+  rolesFile file('roles.yml')
   user username: "x_pack_rest_user", password: "x-pack-test-password"
+  user username: "john_junior", password: "x-pack-test-password", role: "transform_admin"
+  user username: "bill_senior", password: "x-pack-test-password", role: "transform_admin,source_index_access"
 }

x-pack/plugin/transform/qa/multi-node-tests/roles.yml

@@ -0,0 +1,17 @@
+source_index_access:
+  cluster:
+    # This is always required because the REST client uses it to find the version of Elasticsearch it's talking to
+    - cluster:monitor/main
+    - cluster:monitor/tasks/lists
+  indices:
+    # Give access to the source and destination indices because the transform roles alone do not provide access to
+    # non-transform indices
+    - names: [ 'transform-permissions-*' ]
+      privileges:
+        - create_index
+        - indices:admin/refresh
+        - read
+        - write
+        - view_index_metadata
+        - indices:data/write/bulk
+        - indices:data/write/index

x-pack/plugin/transform/qa/multi-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/transform/integration/TransformIT.java

@@ -229,7 +229,7 @@ public void testContinuousTransformUpdate() throws Exception {
                 }
             }
             """, dest, pipelineId);
-        updateConfig(id, update);
+        updateConfig(id, update, RequestOptions.DEFAULT);
 
         // index some more docs
         long timeStamp = Instant.now().toEpochMilli() - 1_000;
@@ -293,7 +293,7 @@ public void testRetentionPolicyDelete() throws Exception {
                 "retention_policy": null
             }
             """;
-        updateConfig(transformId, update);
+        updateConfig(transformId, update, RequestOptions.DEFAULT);
         assertThat(getTransform(transformId), not(hasKey("retention_policy")));
     }
 
@@ -416,7 +416,7 @@ public void testContinuousTransformRethrottle() throws Exception {
             }
             """, reqsPerSec, maxPageSize);
 
-        updateConfig(config.getId(), update);
+        updateConfig(config.getId(), update, RequestOptions.DEFAULT);
 
         waitUntilCheckpoint(config.getId(), 1L);
         assertThat(getTransformState(config.getId()), equalTo("started"));

x-pack/plugin/transform/qa/multi-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/transform/integration/TransformInsufficientPermissionsIT.java

@@ -0,0 +1,248 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.transform.integration;
+
+import org.elasticsearch.client.RequestOptions;
+import org.elasticsearch.client.ResponseException;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.core.TimeValue;
+import org.elasticsearch.search.aggregations.AggregationBuilders;
+import org.elasticsearch.search.aggregations.AggregatorFactories;
+import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
+import org.elasticsearch.xpack.core.transform.transforms.QueryConfig;
+import org.elasticsearch.xpack.core.transform.transforms.SettingsConfig;
+import org.elasticsearch.xpack.core.transform.transforms.TimeSyncConfig;
+import org.elasticsearch.xpack.core.transform.transforms.TransformConfig;
+import org.elasticsearch.xpack.core.transform.transforms.pivot.SingleGroupSource;
+import org.elasticsearch.xpack.core.transform.transforms.pivot.TermsGroupSource;
+import org.junit.After;
+
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import static org.elasticsearch.common.xcontent.support.XContentMapValues.extractValue;
+import static org.elasticsearch.test.SecuritySettingsSourceField.TEST_PASSWORD_SECURE_STRING;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.is;
+
+public class TransformInsufficientPermissionsIT extends TransformRestTestCase {
+
+    private static final String TEST_ADMIN_USERNAME = "x_pack_rest_user";
+    private static final String TEST_ADMIN_HEADER = basicAuthHeaderValue(TEST_ADMIN_USERNAME, TEST_PASSWORD_SECURE_STRING);
+    private static final String JUNIOR_USERNAME = "john_junior";
+    private static final String JUNIOR_HEADER = basicAuthHeaderValue(JUNIOR_USERNAME, TEST_PASSWORD_SECURE_STRING);
+    private static final String SENIOR_USERNAME = "bill_senior";
+    private static final String SENIOR_HEADER = basicAuthHeaderValue(SENIOR_USERNAME, TEST_PASSWORD_SECURE_STRING);
+
+    private static final int NUM_USERS = 28;
+
+    @After
+    public void cleanTransforms() throws Exception {
+        cleanUp();
+    }
+
+    /**
+     * defer_validation = false
+     * unattended       = false
+     */
+    public void testTransformPermissionsNoDeferValidationNoUnattended() throws Exception {
+        testTransformPermissionsNoDeferValidation(false);
+    }
+
+    /**
+     * defer_validation = false
+     * unattended       = true
+     */
+    public void testTransformPermissionsNoDeferValidationUnattended() throws Exception {
+        testTransformPermissionsNoDeferValidation(true);
+    }
+
+    private void testTransformPermissionsNoDeferValidation(boolean unattended) throws Exception {
+        String transformId = "transform-permissions-nodefer-" + (unattended ? 1 : 0);
+        String sourceIndexName = transformId + "-index";
+        String destIndexName = sourceIndexName + "-dest";
+        createReviewsIndex(sourceIndexName, 10, NUM_USERS, TransformIT::getUserIdForRow, TransformIT::getDateStringForRow);
+
+        TransformConfig config = createConfig(transformId, sourceIndexName, destIndexName, unattended);
+
+        ResponseException e = expectThrows(
+            ResponseException.class,
+            () -> putTransform(
+                transformId,
+                Strings.toString(config),
+                RequestOptions.DEFAULT.toBuilder()
+                    .addHeader(AUTH_KEY, JUNIOR_HEADER)
+                    .addParameter("defer_validation", String.valueOf(false))
+                    .build()
+            )
+        );
+
+        assertThat(e.getResponse().getStatusLine().getStatusCode(), is(equalTo(403)));
+        assertThat(
+            e.getMessage(),
+            containsString(
+                String.format(
+                    Locale.ROOT,
+                    "Cannot create transform [%s] because user %s lacks the required permissions "
+                        + "[%s:[read, view_index_metadata], %s:[create_index, index, read]]",
+                    transformId,
+                    JUNIOR_USERNAME,
+                    sourceIndexName,
+                    destIndexName
+                )
+            )
+        );
+
+        putTransform(
+            transformId,
+            Strings.toString(config),
+            RequestOptions.DEFAULT.toBuilder()
+                .addHeader(AUTH_KEY, SENIOR_HEADER)
+                .addParameter("defer_validation", String.valueOf(false))
+                .build()
+        );
+    }
+
+    /**
+     * defer_validation = true
+     * unattended       = false
+     */
+    @SuppressWarnings("unchecked")
+    public void testTransformPermissionsDeferValidationNoUnattended() throws Exception {
+        String transformId = "transform-permissions-defer-nounattended";
+        String sourceIndexName = transformId + "-index";
+        String destIndexName = sourceIndexName + "-dest";
+        createReviewsIndex(sourceIndexName, 10, NUM_USERS, TransformIT::getUserIdForRow, TransformIT::getDateStringForRow);
+
+        TransformConfig config = createConfig(transformId, sourceIndexName, destIndexName, false);
+        putTransform(
+            transformId,
+            Strings.toString(config),
+            RequestOptions.DEFAULT.toBuilder().addParameter("defer_validation", String.valueOf(true)).build()
+        );
+        assertThat(extractValue(getTransformStats(transformId), "health", "status"), is(equalTo("green")));
+
+        ResponseException e = expectThrows(
+            ResponseException.class,
+            () -> startTransform(config.getId(), RequestOptions.DEFAULT.toBuilder().addHeader(AUTH_KEY, JUNIOR_HEADER).build())
+        );
+        assertThat(e.getResponse().getStatusLine().getStatusCode(), is(equalTo(500)));
+        assertThat(
+            e.getMessage(),
+            containsString(
+                String.format(Locale.ROOT, "Could not create destination index [%s] for transform [%s]", destIndexName, transformId)
+            )
+        );
+
+        assertThat(extractValue(getTransformStats(transformId), "health", "status"), is(equalTo("green")));
+
+        e = expectThrows(
+            ResponseException.class,
+            () -> startTransform(config.getId(), RequestOptions.DEFAULT.toBuilder().addHeader(AUTH_KEY, SENIOR_HEADER).build())
+        );
+        assertThat(e.getResponse().getStatusLine().getStatusCode(), is(equalTo(500)));
+        assertThat(
+            e.getMessage(),
+            containsString(
+                String.format(Locale.ROOT, "Could not create destination index [%s] for transform [%s]", destIndexName, transformId)
+            )
+        );
+
+        assertThat(extractValue(getTransformStats(transformId), "health", "status"), is(equalTo("green")));
+
+        // update transform's credentials so that the transform has permission to access source/dest indices
+        updateConfig(transformId, "{}", RequestOptions.DEFAULT.toBuilder().addHeader(AUTH_KEY, SENIOR_HEADER).build());
+
+        // _start API now works
+        startTransform(config.getId(), RequestOptions.DEFAULT);
+        waitUntilCheckpoint(transformId, 1);
+
+        assertThat(extractValue(getTransformStats(transformId), "health", "status"), is(equalTo("green")));
+    }
+
+    /**
+     * defer_validation = true
+     * unattended       = true
+     */
+    @SuppressWarnings("unchecked")
+    public void testTransformPermissionsDeferValidationUnattended() throws Exception {
+        String transformId = "transform-permissions-defer-unattended";
+        String sourceIndexName = transformId + "-index";
+        String destIndexName = sourceIndexName + "-dest";
+        createReviewsIndex(sourceIndexName, 10, NUM_USERS, TransformIT::getUserIdForRow, TransformIT::getDateStringForRow);
+
+        TransformConfig config = createConfig(transformId, sourceIndexName, destIndexName, true);
+        putTransform(
+            transformId,
+            Strings.toString(config),
+            RequestOptions.DEFAULT.toBuilder().addParameter("defer_validation", String.valueOf(true)).build()
+        );
+        assertThat(extractValue(getTransformStats(transformId), "health", "status"), is(equalTo("green")));
+
+        startTransform(config.getId(), RequestOptions.DEFAULT);
+
+        // transform is yellow
+        assertBusy(() -> {
+            Map<String, Object> stats = getTransformStats(transformId);
+            assertThat(extractValue(stats, "health", "status"), is(equalTo("yellow")));
+            List<Object> issues = (List<Object>) extractValue(stats, "health", "issues");
+            assertThat(issues, hasSize(1));
+            assertThat(
+                (String) extractValue((Map<String, Object>) issues.get(0), "details"),
+                containsString(String.format(Locale.ROOT, "no such index [%s]", destIndexName))
+            );
+        }, 10, TimeUnit.SECONDS);
+
+        // update transform's credentials so that the transform has permission to access source/dest indices
+        updateConfig(transformId, "{}", RequestOptions.DEFAULT.toBuilder().addHeader(AUTH_KEY, SENIOR_HEADER).build());
+        waitUntilCheckpoint(transformId, 1);
+
+        // transform is green again
+        assertThat(extractValue(getTransformStats(transformId), "health", "status"), is(equalTo("green")));
+    }
+
+    @Override
+    protected Settings restAdminSettings() {
+        return Settings.builder().put(ThreadContext.PREFIX + ".Authorization", TEST_ADMIN_HEADER).build();
+    }
+
+    @Override
+    protected Settings restClientSettings() {
+        return Settings.builder().put(ThreadContext.PREFIX + ".Authorization", JUNIOR_HEADER).build();
+    }
+
+    private TransformConfig createConfig(String transformId, String sourceIndexName, String destIndexName, boolean unattended)
+        throws Exception {
+        Map<String, SingleGroupSource> groups = Map.of(
+            "by-day",
+            createDateHistogramGroupSourceWithCalendarInterval("timestamp", DateHistogramInterval.DAY, null),
+            "by-user",
+            new TermsGroupSource("user_id", null, false),
+            "by-business",
+            new TermsGroupSource("business_id", null, false)
+        );
+
+        AggregatorFactories.Builder aggs = AggregatorFactories.builder()
+            .addAggregator(AggregationBuilders.avg("review_score").field("stars"))
+            .addAggregator(AggregationBuilders.max("timestamp").field("timestamp"));
+
+        TransformConfig config = createTransformConfigBuilder(transformId, destIndexName, QueryConfig.matchAll(), sourceIndexName)
+            .setPivotConfig(createPivotConfig(groups, aggs))
+            .setSyncConfig(new TimeSyncConfig("timestamp", TimeValue.timeValueSeconds(1)))
+            .setSettings(new SettingsConfig.Builder().setAlignCheckpoints(false).setUnattended(unattended).build())
+            .build();
+
+        return config;
+    }
+}