Add REST API authentication and permissions (#15)

Fixes #14

src/main/java/org/graylog/plugins/pipelineprocessor/PipelineProcessorModule.java

@@ -20,6 +20,7 @@
 import org.graylog.plugins.pipelineprocessor.processors.PipelineInterpreter;
 import org.graylog.plugins.pipelineprocessor.rest.PipelineConnectionsResource;
 import org.graylog.plugins.pipelineprocessor.rest.PipelineResource;
+import org.graylog.plugins.pipelineprocessor.rest.PipelineRestPermissions;
 import org.graylog.plugins.pipelineprocessor.rest.RuleResource;
 import org.graylog2.plugin.PluginConfigBean;
 import org.graylog2.plugin.PluginModule;
@@ -40,6 +41,7 @@ protected void configure() {
         addRestResource(RuleResource.class);
         addRestResource(PipelineResource.class);
         addRestResource(PipelineConnectionsResource.class);
+        addPermissions(PipelineRestPermissions.class);
 
         install(new ProcessorFunctionsModule());
     }

src/main/java/org/graylog/plugins/pipelineprocessor/rest/PipelineConnectionsResource.java

@@ -16,18 +16,21 @@
  */
 package org.graylog.plugins.pipelineprocessor.rest;
 
+import com.google.common.collect.Sets;
 import com.google.common.eventbus.EventBus;
 import io.swagger.annotations.Api;
 import io.swagger.annotations.ApiOperation;
 import io.swagger.annotations.ApiParam;
 import org.apache.shiro.authz.annotation.RequiresAuthentication;
+import org.apache.shiro.authz.annotation.RequiresPermissions;
 import org.graylog.plugins.pipelineprocessor.db.PipelineService;
 import org.graylog.plugins.pipelineprocessor.db.PipelineStreamConnectionsService;
 import org.graylog.plugins.pipelineprocessor.events.PipelineConnectionsChangedEvent;
 import org.graylog2.database.NotFoundException;
 import org.graylog2.events.ClusterEventBus;
 import org.graylog2.plugin.rest.PluginRestResource;
 import org.graylog2.shared.rest.resources.RestResource;
+import org.graylog2.shared.security.RestPermissions;
 import org.graylog2.streams.StreamService;
 
 import javax.inject.Inject;
@@ -41,6 +44,7 @@
 import javax.ws.rs.core.MediaType;
 import java.util.Collections;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 @Api(value = "Pipelines/Connections", description = "Stream connections of processing pipelines")
 @Path("/system/pipelines/connections")
@@ -67,14 +71,17 @@ public PipelineConnectionsResource(PipelineStreamConnectionsService connectionsS
 
     @ApiOperation(value = "Connect processing pipelines to a stream", notes = "")
     @POST
+    @RequiresPermissions(PipelineRestPermissions.PIPELINE_CONNECTION_EDIT)
     public PipelineConnections connectPipelines(@ApiParam(name = "Json body", required = true) @NotNull PipelineConnections connection) throws NotFoundException {
         final String streamId = connection.streamId();
         // the default stream doesn't exist as an entity
         if (!streamId.equalsIgnoreCase("default")) {
+            checkPermission(RestPermissions.STREAMS_READ, streamId);
             streamService.load(streamId);
         }
         // verify the pipelines exist
         for (String s : connection.pipelineIds()) {
+            checkPermission(PipelineRestPermissions.PIPELINE_READ, s);
             pipelineService.load(s);
         }
         final PipelineConnections save = connectionsService.save(connection);
@@ -85,19 +92,51 @@ public PipelineConnections connectPipelines(@ApiParam(name = "Json body", requir
     @ApiOperation("Get pipeline connections for the given stream")
     @GET
     @Path("/{streamId}")
+    @RequiresPermissions(PipelineRestPermissions.PIPELINE_CONNECTION_READ)
     public PipelineConnections getPipelinesForStream(@ApiParam(name = "streamId") @PathParam("streamId") String streamId) throws NotFoundException {
-        return connectionsService.load(streamId);
+        // the user needs to at least be able to read the stream
+        checkPermission(RestPermissions.STREAMS_READ, streamId);
+
+        final PipelineConnections connections = connectionsService.load(streamId);
+        // filter out all pipelines the user does not have enough permissions to see
+        return PipelineConnections.create(
+                connections.id(),
+                connections.streamId(),
+                connections.pipelineIds()
+                        .stream()
+                        .filter(id -> isPermitted(PipelineRestPermissions.PIPELINE_READ, id))
+                        .collect(Collectors.toSet())
+        );
     }
 
     @ApiOperation("Get all pipeline connections")
     @GET
+    @RequiresPermissions(PipelineRestPermissions.PIPELINE_CONNECTION_READ)
     public Set<PipelineConnections> getAll() throws NotFoundException {
-        Set<PipelineConnections> pipelineConnections = connectionsService.loadAll();
+        final Set<PipelineConnections> pipelineConnections = connectionsService.loadAll();
+
+        final Set<PipelineConnections> filteredConnections = Sets.newHashSetWithExpectedSize(pipelineConnections.size());
+        for (PipelineConnections pc : pipelineConnections) {
+            // only include the streams the user can see
+            if (isPermitted(RestPermissions.STREAMS_READ, pc.streamId())) {
+                // filter out all pipelines the user does not have enough permissions to see
+                filteredConnections.add(PipelineConnections.create(
+                        pc.id(),
+                        pc.streamId(),
+                        pc.pipelineIds()
+                                .stream()
+                                .filter(id -> isPermitted(PipelineRestPermissions.PIPELINE_READ, id))
+                                .collect(Collectors.toSet()))
+                );
+            }
+        }
+
+
         // to simplify clients, we always return the default stream, until we have it as a true entity
-        if (!pipelineConnections.stream().anyMatch(pc -> pc.streamId().equals("default"))) {
-            pipelineConnections.add(PipelineConnections.create(null, "default", Collections.emptySet()));
+        if (!filteredConnections.stream().anyMatch(pc -> pc.streamId().equals("default"))) {
+            filteredConnections.add(PipelineConnections.create(null, "default", Collections.emptySet()));
         }
-        return pipelineConnections;
+        return filteredConnections;
     }
 
 }

src/main/java/org/graylog/plugins/pipelineprocessor/rest/PipelineResource.java

@@ -22,6 +22,7 @@
 import io.swagger.annotations.ApiOperation;
 import io.swagger.annotations.ApiParam;
 import org.apache.shiro.authz.annotation.RequiresAuthentication;
+import org.apache.shiro.authz.annotation.RequiresPermissions;
 import org.graylog.plugins.pipelineprocessor.ast.Pipeline;
 import org.graylog.plugins.pipelineprocessor.db.PipelineDao;
 import org.graylog.plugins.pipelineprocessor.db.PipelineService;
@@ -76,6 +77,7 @@ public PipelineResource(PipelineService pipelineService,
 
     @ApiOperation(value = "Create a processing pipeline from source", notes = "")
     @POST
+    @RequiresPermissions(PipelineRestPermissions.PIPELINE_CREATE)
     public PipelineSource createFromParser(@ApiParam(name = "pipeline", required = true) @NotNull PipelineSource pipelineSource) throws ParseException {
         final Pipeline pipeline;
         try {
@@ -121,7 +123,9 @@ public Collection<PipelineSource> getAll() {
         final Collection<PipelineDao> daos = pipelineService.loadAll();
         final ArrayList<PipelineSource> results = Lists.newArrayList();
         for (PipelineDao dao : daos) {
-            results.add(PipelineSource.fromDao(pipelineRuleParser, dao));
+            if (isPermitted(PipelineRestPermissions.PIPELINE_READ, dao.id())) {
+                results.add(PipelineSource.fromDao(pipelineRuleParser, dao));
+            }
         }
 
         return results;
@@ -131,6 +135,7 @@ public Collection<PipelineSource> getAll() {
     @Path("/{id}")
     @GET
     public PipelineSource get(@ApiParam(name = "id") @PathParam("id") String id) throws NotFoundException {
+        checkPermission(PipelineRestPermissions.PIPELINE_READ, id);
         final PipelineDao dao = pipelineService.load(id);
         return PipelineSource.fromDao(pipelineRuleParser, dao);
     }
@@ -140,6 +145,8 @@ public PipelineSource get(@ApiParam(name = "id") @PathParam("id") String id) thr
     @PUT
     public PipelineSource update(@ApiParam(name = "id") @PathParam("id") String id,
                              @ApiParam(name = "pipeline", required = true) @NotNull PipelineSource update) throws NotFoundException {
+        checkPermission(PipelineRestPermissions.PIPELINE_EDIT, id);
+
         final PipelineDao dao = pipelineService.load(id);
         final Pipeline pipeline;
         try {
@@ -163,6 +170,8 @@ public PipelineSource update(@ApiParam(name = "id") @PathParam("id") String id,
     @Path("/{id}")
     @DELETE
     public void delete(@ApiParam(name = "id") @PathParam("id") String id) throws NotFoundException {
+        checkPermission(PipelineRestPermissions.PIPELINE_DELETE, id);
+
         pipelineService.load(id);
         pipelineService.delete(id);
         clusterBus.post(PipelinesChangedEvent.deletedPipelineId(id));

src/main/java/org/graylog/plugins/pipelineprocessor/rest/PipelineRestPermissions.java

@@ -0,0 +1,53 @@
+package org.graylog.plugins.pipelineprocessor.rest;
+
+import com.google.common.collect.ImmutableSet;
+import org.graylog2.plugin.security.Permission;
+import org.graylog2.plugin.security.PluginPermissions;
+
+import java.util.Collections;
+import java.util.Set;
+
+import static org.graylog2.plugin.security.Permission.create;
+
+public class PipelineRestPermissions implements PluginPermissions {
+
+    /* pipelines */
+    public static final String PIPELINE_CREATE = "pipeline:create";
+    public static final String PIPELINE_READ = "pipeline:read";
+    public static final String PIPELINE_EDIT = "pipeline:edit";
+    public static final String PIPELINE_DELETE = "pipeline:delete";
+
+    /* rules */
+    public static final String PIPELINE_RULE_CREATE = "pipeline_rule:create";
+    public static final String PIPELINE_RULE_READ = "pipeline_rule:read";
+    public static final String PIPELINE_RULE_EDIT = "pipeline_rule:edit";
+    public static final String PIPELINE_RULE_DELETE = "pipeline_rule:delete";
+
+    /* connections */
+    public static final String PIPELINE_CONNECTION_READ = "pipeline_connection:read";
+    public static final String PIPELINE_CONNECTION_EDIT = "pipeline_connection:edit";
+
+
+    @Override
+    public Set<Permission> permissions() {
+        return ImmutableSet.of(
+                create(PIPELINE_CREATE, "Create new processing pipeline"),
+                create(PIPELINE_READ, "Read a processing pipeline"),
+                create(PIPELINE_EDIT, "Update a processing pipeline"),
+                create(PIPELINE_DELETE, "Delete a processing pipeline"),
+
+                create(PIPELINE_RULE_CREATE, "Create new processing rule"),
+                create(PIPELINE_RULE_READ, "Read a processing rule"),
+                create(PIPELINE_RULE_EDIT, "Update a processing rule"),
+                create(PIPELINE_RULE_DELETE, "Delete a processing rule"),
+
+                create(PIPELINE_CONNECTION_READ, "Read a pipeline stream connection"),
+                create(PIPELINE_CONNECTION_EDIT, "Update a pipeline stream connections")
+                );
+    }
+
+    @Override
+    public Set<Permission> readerBasePermissions() {
+        return Collections.emptySet();
+    }
+}

src/main/java/org/graylog/plugins/pipelineprocessor/rest/RuleResource.java

@@ -21,6 +21,7 @@
 import io.swagger.annotations.ApiOperation;
 import io.swagger.annotations.ApiParam;
 import org.apache.shiro.authz.annotation.RequiresAuthentication;
+import org.apache.shiro.authz.annotation.RequiresPermissions;
 import org.graylog.plugins.pipelineprocessor.ast.Rule;
 import org.graylog.plugins.pipelineprocessor.db.RuleDao;
 import org.graylog.plugins.pipelineprocessor.db.RuleService;
@@ -76,6 +77,7 @@ public RuleResource(RuleService ruleService,
 
     @ApiOperation(value = "Create a processing rule from source", notes = "")
     @POST
+    @RequiresPermissions(PipelineRestPermissions.PIPELINE_RULE_CREATE)
     public RuleSource createFromParser(@ApiParam(name = "rule", required = true) @NotNull RuleSource ruleSource) throws ParseException {
         final Rule rule;
         try {
@@ -119,6 +121,7 @@ public RuleSource parse(@ApiParam(name = "rule", required = true) @NotNull RuleS
 
     @ApiOperation(value = "Get all processing rules")
     @GET
+    @RequiresPermissions(PipelineRestPermissions.PIPELINE_RULE_READ)
     public Collection<RuleSource> getAll() {
         final Collection<RuleDao> ruleDaos = ruleService.loadAll();
         return ruleDaos.stream()
@@ -130,6 +133,7 @@ public Collection<RuleSource> getAll() {
     @Path("/{id}")
     @GET
     public RuleSource get(@ApiParam(name = "id") @PathParam("id") String id) throws NotFoundException {
+        checkPermission(PipelineRestPermissions.PIPELINE_RULE_READ, id);
         return RuleSource.fromDao(pipelineRuleParser, ruleService.load(id));
     }
 
@@ -141,6 +145,7 @@ public Collection<RuleSource> getBulk(@ApiParam("rules") BulkRuleRequest rules)
 
         return ruleDaos.stream()
                 .map(ruleDao -> RuleSource.fromDao(pipelineRuleParser, ruleDao))
+                .filter(rule -> isPermitted(PipelineRestPermissions.PIPELINE_RULE_READ, rule.id()))
                 .collect(Collectors.toList());
     }
 
@@ -149,6 +154,8 @@ public Collection<RuleSource> getBulk(@ApiParam("rules") BulkRuleRequest rules)
     @PUT
     public RuleSource update(@ApiParam(name = "id") @PathParam("id") String id,
                              @ApiParam(name = "rule", required = true) @NotNull RuleSource update) throws NotFoundException {
+        checkPermission(PipelineRestPermissions.PIPELINE_RULE_EDIT, id);
+
         final RuleDao ruleDao = ruleService.load(id);
         final Rule rule;
         try {
@@ -174,6 +181,7 @@ public RuleSource update(@ApiParam(name = "id") @PathParam("id") String id,
     @Path("/{id}")
     @DELETE
     public void delete(@ApiParam(name = "id") @PathParam("id") String id) throws NotFoundException {
+        checkPermission(PipelineRestPermissions.PIPELINE_RULE_DELETE, id);
         ruleService.load(id);
         ruleService.delete(id);