Add restricted param check to single doc endpoint

Author: lukewhiting

server/src/main/java/org/elasticsearch/action/ActionModule.java

@@ -459,6 +459,7 @@ public class ActionModule extends AbstractModule {
     private final RequestValidators<IndicesAliasesRequest> indicesAliasesRequestRequestValidators;
     private final ReservedClusterStateService reservedClusterStateService;
     private final RestExtension restExtension;
+    private final ClusterService clusterService;
 
     public ActionModule(
         Settings settings,
@@ -534,6 +535,7 @@ public ActionModule(
             reservedProjectStateHandlers
         );
         this.restExtension = restExtension;
+        this.clusterService = clusterService;
     }
 
     private static <T> T getRestServerComponent(
@@ -927,9 +929,9 @@ public void initRestHandlers(Supplier<DiscoveryNodes> nodesInCluster, Predicate<
         registerHandler.accept(new RestResolveClusterAction());
         registerHandler.accept(new RestResolveIndexAction());
 
-        registerHandler.accept(new RestIndexAction());
-        registerHandler.accept(new CreateHandler());
-        registerHandler.accept(new AutoIdHandler());
+        registerHandler.accept(new RestIndexAction(clusterService, projectIdResolver));
+        registerHandler.accept(new CreateHandler(clusterService, projectIdResolver));
+        registerHandler.accept(new AutoIdHandler(clusterService, projectIdResolver));
         registerHandler.accept(new RestGetAction());
         registerHandler.accept(new RestGetSourceAction());
         registerHandler.accept(new RestMultiGetAction(settings));

server/src/main/java/org/elasticsearch/action/bulk/TransportAbstractBulkAction.java

@@ -42,6 +42,7 @@
 import org.elasticsearch.indices.SystemIndices;
 import org.elasticsearch.ingest.IngestService;
 import org.elasticsearch.node.NodeClosedException;
+import org.elasticsearch.rest.action.document.RestBulkAction;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
@@ -412,14 +413,43 @@ private void applyPipelinesAndDoInternalExecute(
         while (bulkRequestModifier.hasNext()) {
             req = bulkRequestModifier.next();
             i++;
+            doStreamsChecks(bulkRequest, projectMetadata, req, bulkRequestModifier, i);
+        }
+
+        var wrappedListener = bulkRequestModifier.wrapActionListenerIfNeeded(listener);
 
-            for (StreamType streamType : StreamType.getEnabledStreamTypesForProject(projectMetadata)) {
-                if (req instanceof IndexRequest ir && streamType.matchesStreamPrefix(req.index()) && ir.isPipelineResolved() == false) {
-                    IllegalArgumentException e = new IllegalArgumentException(
+        if (applyPipelines(task, bulkRequestModifier.getBulkRequest(), executor, wrappedListener) == false) {
+            doInternalExecute(task, bulkRequestModifier.getBulkRequest(), executor, wrappedListener, relativeStartTimeNanos);
+        }
+    }
+
+    private void doStreamsChecks(
+        BulkRequest bulkRequest,
+        ProjectMetadata projectMetadata,
+        DocWriteRequest<?> req,
+        BulkRequestModifier bulkRequestModifier,
+        int i
+    ) {
+        for (StreamType streamType : StreamType.getEnabledStreamTypesForProject(projectMetadata)) {
+            if (req instanceof IndexRequest ir && ir.isPipelineResolved() == false) {
+                IllegalArgumentException e = null;
+                if (streamType.matchesStreamPrefix(req.index())) {
+                    e = new IllegalArgumentException(
                         "Direct writes to child streams are prohibited. Index directly into the ["
                             + streamType.getStreamName()
                             + "] stream instead"
                     );
+                }
+
+                if (e == null && bulkRequest.streamsRestrictedParamsUsed() && req.index().equals(streamType.getStreamName())) {
+                    e = new IllegalArgumentException(
+                        "When writing to a stream, only the following parameters are allowed: ["
+                            + String.join(",", RestBulkAction.STREAMS_ALLOWED_PARAMS)
+                            + "]"
+                    );
+                }
+
+                if (e != null) {
                     Boolean failureStoreEnabled = resolveFailureStore(req.index(), projectMetadata, threadPool.absoluteTimeInMillis());
 
                     if (featureService.clusterHasFeature(clusterService.state(), DataStream.DATA_STREAM_FAILURE_STORE_FEATURE)) {
@@ -438,12 +468,6 @@ private void applyPipelinesAndDoInternalExecute(
                 }
             }
         }
-
-        var wrappedListener = bulkRequestModifier.wrapActionListenerIfNeeded(listener);
-
-        if (applyPipelines(task, bulkRequestModifier.getBulkRequest(), executor, wrappedListener) == false) {
-            doInternalExecute(task, bulkRequestModifier.getBulkRequest(), executor, wrappedListener, relativeStartTimeNanos);
-        }
     }
 
     /**

server/src/main/java/org/elasticsearch/rest/action/document/RestIndexAction.java

@@ -15,7 +15,12 @@
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.support.ActiveShardCount;
 import org.elasticsearch.client.internal.node.NodeClient;
+import org.elasticsearch.cluster.metadata.ProjectMetadata;
+import org.elasticsearch.cluster.project.ProjectIdResolver;
+import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.bytes.ReleasableBytesReference;
+import org.elasticsearch.common.streams.StreamType;
+import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.index.VersionType;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestRequest;
@@ -40,6 +45,13 @@ public class RestIndexAction extends BaseRestHandler {
         + "index requests is deprecated, use the typeless endpoints instead (/{index}/_doc/{id}, /{index}/_doc, "
         + "or /{index}/_create/{id}).";
     private final Set<String> capabilities = Set.of(FAILURE_STORE_STATUS_CAPABILITY);
+    private final ClusterService clusterService;
+    private final ProjectIdResolver projectIdResolver;
+
+    public RestIndexAction(ClusterService clusterService, ProjectIdResolver projectIdResolver) {
+        this.clusterService = clusterService;
+        this.projectIdResolver = projectIdResolver;
+    }
 
     @Override
     public List<Route> routes() {
@@ -54,6 +66,10 @@ public String getName() {
     @ServerlessScope(Scope.PUBLIC)
     public static final class CreateHandler extends RestIndexAction {
 
+        public CreateHandler(ClusterService clusterService, ProjectIdResolver projectIdResolver) {
+            super(clusterService, projectIdResolver);
+        }
+
         @Override
         public String getName() {
             return "document_create_action";
@@ -81,7 +97,9 @@ static void validateOpType(String opType) {
     @ServerlessScope(Scope.PUBLIC)
     public static final class AutoIdHandler extends RestIndexAction {
 
-        public AutoIdHandler() {}
+        public AutoIdHandler(ClusterService clusterService, ProjectIdResolver projectIdResolver) {
+            super(clusterService, projectIdResolver);
+        }
 
         @Override
         public String getName() {
@@ -105,7 +123,28 @@ public RestChannelConsumer prepareRequest(RestRequest request, final NodeClient
     @Override
     public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         ReleasableBytesReference source = request.requiredContent();
-        IndexRequest indexRequest = new IndexRequest(request.param("index"));
+
+        String index = request.param("index");
+        ProjectMetadata projectMetadata = null;
+
+        for (StreamType streamType : StreamType.values()) {
+            if (index.equals(streamType.getStreamName())) {
+                if (projectMetadata == null) {
+                    projectMetadata = clusterService.state().projectState(projectIdResolver.getProjectId()).metadata();
+                }
+
+                if (streamType.streamTypeIsEnabled(projectMetadata)
+                    && Sets.difference(request.params().keySet(), RestBulkAction.STREAMS_ALLOWED_PARAMS).isEmpty() == false) {
+                    throw new IllegalArgumentException(
+                        "When writing to a stream, only the following parameters are allowed: ["
+                            + String.join(",", RestBulkAction.STREAMS_ALLOWED_PARAMS)
+                            + "]"
+                    );
+                }
+            }
+        }
+
+        IndexRequest indexRequest = new IndexRequest(index);
         indexRequest.id(request.param("id"));
         indexRequest.routing(request.param("routing"));
         indexRequest.setPipeline(request.param("pipeline"));

server/src/test/java/org/elasticsearch/rest/action/document/RestIndexActionTests.java

@@ -37,13 +37,13 @@ public final class RestIndexActionTests extends RestActionTestCase {
 
     @Before
     public void setUpAction() {
-        controller().registerHandler(new RestIndexAction());
-        controller().registerHandler(new CreateHandler());
-        controller().registerHandler(new AutoIdHandler());
+        controller().registerHandler(new RestIndexAction(null, null));
+        controller().registerHandler(new CreateHandler(null, null));
+        controller().registerHandler(new AutoIdHandler(null, null));
     }
 
     public void testCreateOpTypeValidation() {
-        RestIndexAction.CreateHandler create = new CreateHandler();
+        RestIndexAction.CreateHandler create = new CreateHandler(null, null);
 
         String opType = randomFrom("CREATE", null);
         CreateHandler.validateOpType(opType);