[8.19] [Transform] Fix transform validation to reject PUT and _start when user lacks remote index permissions (#142403) (#142455)

* [Transform] Fix transform validation to reject PUT and _start when user lacks remote index permissions (#142403)

When a transform is configured with a remote (cross-cluster) source index and the user lacks permissions to access it, the _preview API correctly fails -- but PUT _transform and _start silently succeed, allowing unauthorized transforms to be created and started. The root cause is that validateQuery in AbstractCompositeAggFunction only checks the response status code, which is OK even when IndicesOptions.LENIENT_EXPAND_OPEN causes unauthorized indices to be silently ignored. The search returns null aggregations in this case, but unlike preview(), validateQuery() never checks for that condition.

This PR introduces a SourceAccessDiagnostics class that inspects the SearchResponse for security-related failures at both the CCS cluster level (SKIPPED/FAILED clusters with ElasticsearchSecurityException) and the shard level (FORBIDDEN/UNAUTHORIZED status). A null-aggregation check is added to validateQuery(), but -- critically -- it only rejects the request when a security failure is positively identified. When no security failure is found, validation passes through silently. This distinction avoids the regression that caused PR #95318 to be reverted in #95562: that earlier change unconditionally failed on null aggregations, which broke integrations (such as Elastic Defend) that create and start transforms with wildcard source patterns before any matching indices exist. Since defer_validation only defers from PUT to _start, there was no way for those integrations to bypass the check. Our approach preserves backward compatibility for the empty-indices case while catching the unauthorized-remote-index case. The preview() method also delegates to the same diagnostics class, so all three APIs now produce consistent, actionable error messages when a security failure is detected, falling back to the original generic message otherwise.

The multi-cluster YAML integration tests are updated to verify that both PUT _transform and _start now reject unauthorized remote transforms. A new test case creates a transform with defer_validation: true and confirms that _start catches the permission issue. Unit tests for SourceAccessDiagnostics cover cluster-level SKIPPED/FAILED scenarios, shard-level security exceptions, FORBIDDEN/UNAUTHORIZED status codes, and the fallback to the generic message for non-security failures.

Fixes #95367

(cherry picked from commit 0e44984)

# Conflicts:
#	x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/transforms/common/AbstractCompositeAggFunction.java

* fix compilation error

* Add diagnostics for remote CCS clusters with zero shards

Enhance the SourceAccessDiagnostics class to identify remote CCS clusters that return zero shards due to permission issues. This update includes a new method to check for such scenarios and updates the documentation accordingly. Additionally, new unit tests are added to verify the correct behavior when accessing remote clusters with insufficient permissions, ensuring that appropriate error messages are returned. This change improves the clarity of diagnostics related to security exceptions in cross-cluster searches.

* fix unit test specifics for 8.19

* Update transform configuration in multi-cluster test to include defer_validation and modify description

Author: valeriy42

docs/changelog/142403.yaml

@@ -0,0 +1,6 @@
+area: "Transform"
+issues:
+ - 95367
+pr: 142403
+summary: Fix transform validation to reject PUT and `_start` when user lacks remote index permissions
+type: bug

x-pack/plugin/security/qa/multi-cluster/src/javaRestTest/java/org/elasticsearch/xpack/remotecluster/RemoteClusterSecurityTransformIT.java

@@ -23,6 +23,7 @@
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
 
+import static org.elasticsearch.common.xcontent.support.XContentMapValues.extractValue;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.nullValue;
@@ -201,7 +202,7 @@ public void testCrossClusterTransform() throws Exception {
             // Delete the transform
             assertOK(performRequestWithRemoteTransformUser(new Request("DELETE", "/_transform/simple-remote-transform")));
 
-            // Create a transform targeting an index without permission
+            // Attempt to create a transform targeting an index without permission - should fail validation
             final var putTransformRequest2 = new Request("PUT", "/_transform/invalid");
             putTransformRequest2.setJsonEntity("""
                 {
@@ -213,14 +214,14 @@ public void testCrossClusterTransform() throws Exception {
                   }
                 }
                 """);
-            assertOK(performRequestWithRemoteTransformUser(putTransformRequest2));
-            // It errors when trying to preview it
-            final ResponseException e = expectThrows(
+            final ResponseException putException = expectThrows(
                 ResponseException.class,
-                () -> performRequestWithRemoteTransformUser(new Request("GET", "/_transform/invalid/_preview"))
+                () -> performRequestWithRemoteTransformUser(putTransformRequest2)
             );
-            assertThat(e.getResponse().getStatusLine().getStatusCode(), equalTo(400));
-            assertThat(e.getMessage(), containsString("Source indices have been deleted or closed"));
+            assertThat(putException.getResponse().getStatusLine().getStatusCode(), equalTo(400));
+            // Assert on response body so the test is robust across locales and exception message formatting
+            final String putReason = (String) extractValue("error.reason", entityAsMap(putException.getResponse()));
+            assertThat(putReason, containsString("lacks the required permissions"));
         }
     }
 

x-pack/plugin/transform/qa/multi-cluster-tests-with-security/src/test/resources/rest-api-spec/test/multi_cluster/80_transform.yml

@@ -151,7 +151,7 @@ teardown:
         transform_id: "simple-remote-transform"
 
   - do:
-      catch: /Source indices have been deleted or closed./
+      catch: /lacks the required permissions/
       headers: { Authorization: "Basic Ym9iOnRyYW5zZm9ybS1wYXNzd29yZA==" }  # This is bob
       transform.preview_transform:
         transform_id: "simple-remote-transform"
@@ -306,6 +306,7 @@ teardown:
 ---
 "Batch transform from remote cluster when the user is not authorized":
   - do:
+      catch: /lacks the required permissions/
       headers: { Authorization: "Basic Ym9iOnRyYW5zZm9ybS1wYXNzd29yZA==" }  # This is bob
       transform.put_transform:
         transform_id: "simple-remote-transform-3"
@@ -319,6 +320,30 @@ teardown:
             }
           }
 
+---
+"Batch transform _start from remote cluster when the user is not authorized":
+  - do:
+      headers: { Authorization: "Basic Ym9iOnRyYW5zZm9ybS1wYXNzd29yZA==" }  # This is bob
+      transform.put_transform:
+        transform_id: "simple-remote-transform-unauthorized-start"
+        defer_validation: true
+        body: >
+          {
+            "source": { "index": "my_remote_cluster:remote_test_index" },
+            "dest": { "index": "simple-remote-transform-unauthorized-start" },
+            "pivot": {
+              "group_by": { "user": {"terms": {"field": "user"}}},
+              "aggs": { "avg_stars": {"avg": {"field": "stars"}}}
+            }
+          }
+  - match: { acknowledged: true }
+
+  - do:
+      catch: /lacks the required permissions/
+      headers: { Authorization: "Basic Ym9iOnRyYW5zZm9ybS1wYXNzd29yZA==" }  # This is bob
+      transform.start_transform:
+        transform_id: "simple-remote-transform-unauthorized-start"
+
 ---
 "Batch transform update from remote cluster when the user is not authorized":
   - do:
@@ -339,16 +364,17 @@ teardown:
       headers: { Authorization: "Basic Ym9iOnRyYW5zZm9ybS1wYXNzd29yZA==" }  # This is bob
       transform.update_transform:
         transform_id: "simple-remote-transform-2"
+        defer_validation: true
         body: >
           {
-            "source": { "index": "my_remote_cluster:remote_test_index" },
+            "description": "Updated by bob",
             "dest": { "index": "simple-remote-transform-2" }
           }
 
 ---
 "Batch transform preview from remote cluster when the user is not authorized":
   - do:
-      catch: /Source indices have been deleted or closed./
+      catch: /lacks the required permissions/
       headers: { Authorization: "Basic Ym9iOnRyYW5zZm9ybS1wYXNzd29yZA==" }  # This is bob
       transform.preview_transform:
         body: >
@@ -361,7 +387,7 @@ teardown:
             }
           }
   - do:
-      catch: /Source indices have been deleted or closed./
+      catch: /lacks the required permissions/
       headers: { Authorization: "Basic Ym9iOnRyYW5zZm9ybS1wYXNzd29yZA==" }  # This is bob
       transform.preview_transform:
         body: >

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/transforms/common/AbstractCompositeAggFunction.java

@@ -87,7 +87,7 @@ public void preview(
                     final InternalAggregations aggregations = r.getAggregations();
                     if (aggregations == null) {
                         listener.onFailure(
-                            new ElasticsearchStatusException("Source indices have been deleted or closed.", RestStatus.BAD_REQUEST)
+                            new ElasticsearchStatusException(SourceAccessDiagnostics.diagnoseSourceAccessFailure(r), RestStatus.BAD_REQUEST)
                         );
                         return;
                     }
@@ -139,6 +139,19 @@ public void validateQuery(
                     );
                     return;
                 }
+                // Null aggregations may indicate permission issues when accessing remote indices.
+                // We only fail validation when a security failure is positively identified (e.g.,
+                // ElasticsearchSecurityException in cluster or shard failures). We deliberately do NOT
+                // fail when no security failure is found, because null aggregations can also occur when
+                // a local wildcard index pattern resolves to zero indices -- a legitimate scenario for
+                // integrations that start transforms before source data exists (see #95562).
+                if (response.getAggregations() == null) {
+                    String diagnosis = SourceAccessDiagnostics.diagnoseSourceAccessFailure(response);
+                    if (diagnosis.equals(SourceAccessDiagnostics.SOURCE_INDICES_MISSING) == false) {
+                        listener.onFailure(new ValidationException().addValidationError(diagnosis));
+                        return;
+                    }
+                }
                 listener.onResponse(true);
             }, e -> {
                 Throwable unwrapped = ExceptionsHelper.unwrapCause(e);

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/transforms/common/SourceAccessDiagnostics.java

@@ -0,0 +1,142 @@
+/*
+ * 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.transforms.common;
+
+import org.elasticsearch.ElasticsearchSecurityException;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.action.search.ShardSearchFailure;
+import org.elasticsearch.rest.RestStatus;
+
+/**
+ * Diagnoses the cause of an empty or failed search response against transform source indices.
+ * Inspects CCS cluster-level and shard-level failures to distinguish permission errors
+ * from genuinely missing or closed indices.
+ */
+final class SourceAccessDiagnostics {
+
+    static final String SOURCE_INDICES_MISSING = "Source indices have been deleted or closed.";
+
+    private SourceAccessDiagnostics() {}
+
+    /**
+     * Inspects a {@link SearchResponse} that returned null aggregations and produces
+     * a human-readable error message identifying the likely cause.
+     *
+     * <p>The method checks, in order:
+     * <ol>
+     *   <li>CCS cluster-level failures for security exceptions (SKIPPED or FAILED clusters)</li>
+     *   <li>Top-level shard failures for security exceptions</li>
+     *   <li>Remote CCS clusters with 0 total shards (security may have silently filtered all indices)</li>
+     *   <li>Falls back to a generic "deleted or closed" message</li>
+     * </ol>
+     */
+    static String diagnoseSourceAccessFailure(SearchResponse response) {
+        String clusterSecurityError = findClusterSecurityFailure(response);
+        if (clusterSecurityError != null) {
+            return clusterSecurityError;
+        }
+
+        String shardSecurityError = findShardSecurityFailure(response);
+        if (shardSecurityError != null) {
+            return shardSecurityError;
+        }
+
+        String zeroShardsError = findRemoteClusterWithZeroShards(response);
+        if (zeroShardsError != null) {
+            return zeroShardsError;
+        }
+
+        return SOURCE_INDICES_MISSING;
+    }
+
+    /**
+     * Checks CCS cluster-level information for clusters that were SKIPPED or FAILED
+     * due to a security exception (e.g., user lacks permissions on the remote cluster).
+     */
+    private static String findClusterSecurityFailure(SearchResponse response) {
+        SearchResponse.Clusters clusters = response.getClusters();
+        if (clusters == null || clusters.getTotal() == 0) {
+            return null;
+        }
+
+        for (String alias : clusters.getClusterAliases()) {
+            SearchResponse.Cluster cluster = clusters.getCluster(alias);
+            if (cluster == null) {
+                continue;
+            }
+            if (cluster.getStatus() != SearchResponse.Cluster.Status.SKIPPED
+                && cluster.getStatus() != SearchResponse.Cluster.Status.FAILED) {
+                continue;
+            }
+            for (ShardSearchFailure failure : cluster.getFailures()) {
+                if (isSecurityFailure(failure)) {
+                    return "User lacks the required permissions to read source indices on cluster [" + alias + "].";
+                }
+            }
+        }
+
+        return null;
+    }
+
+    /**
+     * Checks top-level shard failures for security exceptions (e.g., user lacks
+     * permissions on a specific index).
+     */
+    private static String findShardSecurityFailure(SearchResponse response) {
+        for (ShardSearchFailure failure : response.getShardFailures()) {
+            if (isSecurityFailure(failure)) {
+                String index = failure.index();
+                if (index != null) {
+                    return "User lacks the required permissions to read source index [" + index + "].";
+                }
+                return "User lacks the required permissions to read from the source indices.";
+            }
+        }
+        return null;
+    }
+
+    /**
+     * Checks for remote CCS clusters that completed successfully but searched zero shards.
+     * This pattern typically occurs when the security module silently filters out all
+     * matching indices because the user lacks the required permissions. With lenient
+     * index options, unauthorized indices are treated as unavailable, so no security
+     * exception is thrown — the search simply returns zero results.
+     *
+     * <p>Only remote clusters are inspected. A local cluster with zero shards is considered
+     * legitimate (e.g., a wildcard pattern before data exists — see #95562).
+     */
+    private static String findRemoteClusterWithZeroShards(SearchResponse response) {
+        SearchResponse.Clusters clusters = response.getClusters();
+        if (clusters == null || clusters.getTotal() == 0) {
+            return null;
+        }
+
+        for (String alias : clusters.getClusterAliases()) {
+            if (alias.isEmpty()) {
+                continue; // skip the local cluster
+            }
+            SearchResponse.Cluster cluster = clusters.getCluster(alias);
+            if (cluster == null) {
+                continue;
+            }
+            Integer totalShards = cluster.getTotalShards();
+            if (cluster.getStatus() == SearchResponse.Cluster.Status.SUCCESSFUL && totalShards != null && totalShards == 0) {
+                return "User lacks the required permissions to read source indices on cluster [" + alias + "].";
+            }
+        }
+
+        return null;
+    }
+
+    private static boolean isSecurityFailure(ShardSearchFailure failure) {
+        if (failure.getCause() instanceof ElasticsearchSecurityException) {
+            return true;
+        }
+        return failure.status() == RestStatus.FORBIDDEN || failure.status() == RestStatus.UNAUTHORIZED;
+    }
+}