ES-11331 streams params restriction

docs/changelog/132967.yaml

@@ -0,0 +1,5 @@
+pr: 132967
+summary: ES-11331 streams params restriction
+area: Data streams
+type: enhancement
+issues: []

modules/streams/src/yamlRestTest/resources/rest-api-spec/test/streams/logs/20_substream_restrictions.yml

@@ -142,13 +142,16 @@ teardown:
             index.default_pipeline: "reroute-to-logs-foo-success"
   - do:
       bulk:
-        refresh: true
         body: |
           { "index": { "_index": "logs" } }
           { "foo": "bar", "baz": "qux" }
   - match: { errors: false }
   - match: { items.0.index.status: 201 }
 
+  - do:
+      indices.refresh:
+        index: logs.foo
+
   - do:
       delete_by_query:
         index: logs.foo

modules/streams/src/yamlRestTest/resources/rest-api-spec/test/streams/logs/30_param_restrictions.yml

@@ -0,0 +1,175 @@
+---
+teardown:
+  - do:
+      streams.logs_disable: { }
+
+---
+"Check User Can't Use Restricted Params On Logs Endpoint":
+  - do:
+      streams.logs_enable: { }
+  - is_true: acknowledged
+
+  - do:
+      streams.status: { }
+  - is_true: logs.enabled
+
+  - do:
+      bulk:
+        refresh: true
+        body: |
+          { "index": { "_index": "logs"} }
+          { "foo": "bar" }
+          { "index": { "_index": "not-logs" } }
+          { "foo": "bar" }
+  - match: { errors: true }
+  - match: { items.0.index.status: 400 }
+  - match: { items.0.index.error.type: "illegal_argument_exception" }
+  - match: { items.0.index.error.reason: "When writing to a stream, only the following parameters are allowed: [index,op_type,error_trace,timeout]" }
+  - match: { items.1.index.status: 201 }
+
+---
+"Check User Can't Use Restricted Params On Logs Endpoint - Single Doc":
+  - do:
+      streams.logs_enable: { }
+  - is_true: acknowledged
+
+  - do:
+      streams.status: { }
+  - is_true: logs.enabled
+
+  - do:
+      index:
+        refresh: true
+        index: logs
+        body: { "foo": "bar" }
+      catch: '/When writing to a stream, only the following parameters are allowed: \[index,op_type,error_trace,timeout\]/'
+
+---
+"Allowed Params Only - Bulk Should Succeed":
+  - do:
+      streams.logs_enable: { }
+  - is_true: acknowledged
+
+  - do:
+      bulk:
+        error_trace: true
+        timeout: "1m"
+        body: |
+          { "index": { "_index": "logs"} }
+          { "foo": "bar" }
+  - match: { errors: false }
+  - match: { items.0.index.status: 201 }
+
+---
+"Allowed Params Only - Single Doc Should Succeed":
+  - do:
+      streams.logs_enable: { }
+  - is_true: acknowledged
+
+  - do:
+      index:
+        index: logs
+        error_trace: true
+        timeout: "1m"
+        body: |
+          { "foo": "bar" }
+  - match: { _index: "logs" }
+  - match: { result: "created" }
+
+---
+"No Params - Bulk Should Succeed":
+  - do:
+      streams.logs_enable: { }
+  - is_true: acknowledged
+
+  - do:
+      bulk:
+        body: |
+          { "index": { "_index": "logs"} }
+          { "foo": "bar" }
+  - match: { errors: false }
+  - match: { items.0.index.status: 201 }
+
+---
+"No Params - Single Doc Should Succeed":
+  - do:
+      streams.logs_enable: { }
+  - is_true: acknowledged
+  -
+    do:
+      index:
+        index: logs
+        body: { "foo": "bar" }
+  - match: { _index: "logs" }
+  - match: { result: "created" }
+
+---
+"Mixed Allowed and Disallowed Params - Bulk Should Fail":
+  - do:
+      streams.logs_enable: { }
+  - is_true: acknowledged
+
+  - do:
+      bulk:
+        error_trace: true
+        timeout: "1m"
+        routing: "custom-routing"
+        refresh: true
+        body: |
+          { "index": { "_index": "logs"} }
+          { "foo": "bar" }
+  - match: { errors: true }
+  - match: { items.0.index.status: 400 }
+  - match: { items.0.index.error.type: "illegal_argument_exception" }
+  - match: { items.0.index.error.reason: "When writing to a stream, only the following parameters are allowed: [index,op_type,error_trace,timeout]" }
+
+---
+"Mixed Allowed and Disallowed Params - Single Doc Should Fail":
+  - do:
+      streams.logs_enable: { }
+  - is_true: acknowledged
+
+  - do:
+      index:
+        index: logs
+        error_trace: true
+        timeout: "1m"
+        routing: "custom-routing"
+        refresh: true
+        body: { "foo": "bar" }
+      catch: '/When writing to a stream, only the following parameters are allowed: \[index,op_type,error_trace,timeout\]/'
+
+---
+"Multiple Disallowed Params - Bulk Should Fail":
+  - do:
+      streams.logs_enable: { }
+  - is_true: acknowledged
+
+  - do:
+      bulk:
+        routing: "custom-routing"
+        wait_for_active_shards: "2"
+        refresh: true
+        body: |
+          { "index": { "_index": "logs"} }
+          { "foo": "bar" }
+  - match: { errors: true }
+  - match: { items.0.index.status: 400 }
+  - match: { items.0.index.error.type: "illegal_argument_exception" }
+  - match: { items.0.index.error.reason: "When writing to a stream, only the following parameters are allowed: [index,op_type,error_trace,timeout]" }
+
+---
+"Multiple Disallowed Params - Single Doc Should Fail":
+  - do:
+      streams.logs_enable: { }
+  - is_true: acknowledged
+
+  - do:
+      index:
+        index: logs
+        routing: "custom-routing"
+        pipeline: "my-pipeline"
+        wait_for_active_shards: "2"
+        refresh: true
+        body: { "foo": "bar" }
+      catch: '/When writing to a stream, only the following parameters are allowed: \[index,op_type,error_trace,timeout\]/'

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -365,6 +365,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion ESQL_LOOKUP_JOIN_ON_MANY_FIELDS = def(9_139_0_00);
     public static final TransportVersion SIMULATE_INGEST_EFFECTIVE_MAPPING = def(9_140_0_00);
     public static final TransportVersion RESOLVE_INDEX_MODE_ADDED = def(9_141_0_00);
+    public static final TransportVersion STREAMS_ENDPOINT_PARAM_RESTRICTIONS = def(9_142_00_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,

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/BulkRequest.java

@@ -86,6 +86,7 @@ public class BulkRequest extends LegacyActionRequest
     private Boolean globalRequireAlias;
     private Boolean globalRequireDatsStream;
     private boolean includeSourceOnError = true;
+    private boolean streamsRestrictedParamsUsed = false;
 
     private long sizeInBytes = 0;
 
@@ -107,7 +108,10 @@ public BulkRequest(StreamInput in) throws IOException {
         }
         if (in.getTransportVersion().onOrAfter(TransportVersions.INGEST_REQUEST_INCLUDE_SOURCE_ON_ERROR)) {
             includeSourceOnError = in.readBoolean();
-        } // else default value is true
+        }
+        if (in.getTransportVersion().onOrAfter(TransportVersions.STREAMS_ENDPOINT_PARAM_RESTRICTIONS)) {
+            streamsRestrictedParamsUsed = in.readBoolean();
+        }
     }
 
     public BulkRequest(@Nullable String globalIndex) {
@@ -474,6 +478,9 @@ public void writeTo(StreamOutput out) throws IOException {
         if (out.getTransportVersion().onOrAfter(TransportVersions.INGEST_REQUEST_INCLUDE_SOURCE_ON_ERROR)) {
             out.writeBoolean(includeSourceOnError);
         }
+        if (out.getTransportVersion().onOrAfter(TransportVersions.STREAMS_ENDPOINT_PARAM_RESTRICTIONS)) {
+            out.writeBoolean(streamsRestrictedParamsUsed);
+        }
     }
 
     @Override
@@ -516,6 +523,14 @@ public boolean isSimulated() {
         return false; // Always false, but may be overridden by a subclass
     }
 
+    public boolean streamsRestrictedParamsUsed() {
+        return streamsRestrictedParamsUsed;
+    }
+
+    public void streamsRestrictedParamsUsed(boolean streamsRestrictedParamsUsed) {
+        this.streamsRestrictedParamsUsed = streamsRestrictedParamsUsed;
+    }
+
     /*
      * Returns any component template substitutions that are to be used as part of this bulk request. We would likely only have
      * substitutions in the event of a simulated request.
@@ -554,6 +569,7 @@ public BulkRequest shallowClone() {
         bulkRequest.routing(routing());
         bulkRequest.requireAlias(requireAlias());
         bulkRequest.requireDataStream(requireDataStream());
+        bulkRequest.streamsRestrictedParamsUsed(streamsRestrictedParamsUsed());
         return bulkRequest;
     }
 }

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

@@ -62,12 +62,25 @@ public IncrementalBulkService(Client client, IndexingPressure indexingPressure,
 
     public Handler newBulkRequest() {
         ensureEnabled();
-        return newBulkRequest(null, null, null);
+        return newBulkRequest(null, null, null, false);
     }
 
-    public Handler newBulkRequest(@Nullable String waitForActiveShards, @Nullable TimeValue timeout, @Nullable String refresh) {
+    public Handler newBulkRequest(
+        @Nullable String waitForActiveShards,
+        @Nullable TimeValue timeout,
+        @Nullable String refresh,
+        boolean usesStreamsRestrictedParams
+    ) {
         ensureEnabled();
-        return new Handler(client, indexingPressure, waitForActiveShards, timeout, refresh, chunkWaitTimeMillisHistogram);
+        return new Handler(
+            client,
+            indexingPressure,
+            waitForActiveShards,
+            timeout,
+            refresh,
+            chunkWaitTimeMillisHistogram,
+            usesStreamsRestrictedParams
+        );
     }
 
     private void ensureEnabled() {
@@ -105,6 +118,7 @@ public static class Handler implements Releasable {
         private final Client client;
         private final ActiveShardCount waitForActiveShards;
         private final TimeValue timeout;
+        private final boolean usesStreamsRestrictedParams;
         private final String refresh;
 
         private final ArrayList<Releasable> releasables = new ArrayList<>(4);
@@ -125,12 +139,14 @@ protected Handler(
             @Nullable String waitForActiveShards,
             @Nullable TimeValue timeout,
             @Nullable String refresh,
-            LongHistogram chunkWaitTimeMillisHistogram
+            LongHistogram chunkWaitTimeMillisHistogram,
+            boolean usesStreamsRestrictedParams
         ) {
             this.client = client;
             this.waitForActiveShards = waitForActiveShards != null ? ActiveShardCount.parseString(waitForActiveShards) : null;
             this.timeout = timeout;
             this.refresh = refresh;
+            this.usesStreamsRestrictedParams = usesStreamsRestrictedParams;
             this.incrementalOperation = indexingPressure.startIncrementalCoordinating(0, 0, false);
             this.chunkWaitTimeMillisHistogram = chunkWaitTimeMillisHistogram;
             createNewBulkRequest(EMPTY_STATE);
@@ -310,6 +326,7 @@ private void createNewBulkRequest(BulkRequest.IncrementalState incrementalState)
             if (refresh != null) {
                 bulkRequest.setRefreshPolicy(refresh);
             }
+            bulkRequest.streamsRestrictedParamsUsed(usesStreamsRestrictedParams);
         }
     }
 }