Require streams:edit when connecting pipeline and streams (#23308)

* require streams:edit

* CL

* delete extraneous change

* sonar fixes

Author: patrickmann

changelog/unreleased/pr-23308.toml

@@ -0,0 +1,5 @@
+type = "fixed"
+message = "Connecting a pipeline to a stream requires edit permissions on the stream."
+
+issues = ["Graylog/graylog2-enterprise#10654"]
+pulls = ["23308"]

graylog2-server/src/main/java/org/graylog/plugins/pipelineprocessor/rest/PipelineConnectionsResource.java

@@ -76,17 +76,17 @@ public PipelineConnectionsResource(PipelineStreamConnectionsService connectionsS
         this.entityScopeService = entityScopeService;
     }
 
-    @ApiOperation(value = "Connect processing pipelines to a stream", notes = "")
+    @ApiOperation(value = "Connect processing pipelines to a stream")
     @POST
     @Path("/to_stream")
     @RequiresPermissions(PipelineRestPermissions.PIPELINE_CONNECTION_EDIT)
     @AuditEvent(type = PipelineProcessorAuditEventTypes.PIPELINE_CONNECTION_UPDATE)
     public PipelineConnections connectPipelines(@ApiParam(name = "Json body", required = true) @NotNull PipelineConnections connection) throws NotFoundException {
         final String streamId = connection.streamId();
 
+        // verify the stream exists and is editable
         checkNotEditable(streamId, "Cannot connect pipeline to non editable stream");
-        // verify the stream exists
-        checkPermission(RestPermissions.STREAMS_READ, streamId);
+        checkPermission(RestPermissions.STREAMS_EDIT, streamId);
         streamService.load(streamId);
 
         // verify the pipelines exist
@@ -97,7 +97,7 @@ public PipelineConnections connectPipelines(@ApiParam(name = "Json body", requir
         return connectionsService.save(connection);
     }
 
-    @ApiOperation(value = "Connect streams to a processing pipeline", notes = "")
+    @ApiOperation(value = "Connect streams to a processing pipeline")
     @POST
     @Path("/to_pipeline")
     @RequiresPermissions(PipelineRestPermissions.PIPELINE_CONNECTION_EDIT)
@@ -106,7 +106,7 @@ public Set<PipelineConnections> connectStreams(@ApiParam(name = "Json body", req
         final String pipelineId = connection.pipelineId();
         final Set<PipelineConnections> updatedConnections = Sets.newHashSet();
 
-        // verify the pipeline exists and is editable
+        // verify the pipeline exists
         checkPermission(PipelineRestPermissions.PIPELINE_READ, pipelineId);
         checkScope(pipelineService.load(pipelineId));
 
@@ -118,6 +118,7 @@ public Set<PipelineConnections> connectStreams(@ApiParam(name = "Json body", req
         connection.streamIds().forEach(streamId ->
                 checkNotEditable(streamId, "Cannot connect pipeline to non editable stream")
         );
+
         // remove deleted pipeline connections
         for (PipelineConnections pipelineConnection : pipelineConnections) {
             if (!connection.streamIds().contains(pipelineConnection.streamId())) {
@@ -132,8 +133,8 @@ public Set<PipelineConnections> connectStreams(@ApiParam(name = "Json body", req
 
         // update pipeline connections
         for (String streamId : connection.streamIds()) {
-            // verify the stream exist
-            checkPermission(RestPermissions.STREAMS_READ, streamId);
+            // verify the stream exists and is editable
+            checkPermission(RestPermissions.STREAMS_EDIT, streamId);
             streamService.load(streamId);
 
             PipelineConnections updatedConnection;
@@ -177,7 +178,7 @@ public PipelineConnections getPipelinesForStream(@ApiParam(name = "streamId") @P
     @ApiOperation("Get all pipeline connections")
     @GET
     @RequiresPermissions(PipelineRestPermissions.PIPELINE_CONNECTION_READ)
-    public Set<PipelineConnections> getAll() throws NotFoundException {
+    public Set<PipelineConnections> getAll() {
         final Set<PipelineConnections> pipelineConnections = connectionsService.loadAll();
 
         final Set<PipelineConnections> filteredConnections = Sets.newHashSetWithExpectedSize(pipelineConnections.size());