Add StreamCategoryFilter and stream_category to StreamDTO

changelog/unreleased/pr-20110.toml

@@ -0,0 +1,5 @@
+type = "a"
+message = "Added categories to Streams to allow Illuminate content to be scoped to multiple products."
+
+issues = ["graylog-plugin-enterprise#7945"]
+pulls = ["20110"]

graylog2-server/src/main/java/org/graylog/plugins/views/ViewsBindings.java

@@ -56,6 +56,7 @@
 import org.graylog.plugins.views.search.filter.AndFilter;
 import org.graylog.plugins.views.search.filter.OrFilter;
 import org.graylog.plugins.views.search.filter.QueryStringFilter;
+import org.graylog.plugins.views.search.filter.StreamCategoryFilter;
 import org.graylog.plugins.views.search.filter.StreamFilter;
 import org.graylog.plugins.views.search.querystrings.LastUsedQueryStringsService;
 import org.graylog.plugins.views.search.querystrings.MongoLastUsedQueryStringsService;
@@ -177,6 +178,7 @@ protected void configure() {
         registerJacksonSubtype(AndFilter.class);
         registerJacksonSubtype(OrFilter.class);
         registerJacksonSubtype(StreamFilter.class);
+        registerJacksonSubtype(StreamCategoryFilter.class);
         registerJacksonSubtype(QueryStringFilter.class);
 
         // query backends for jackson

graylog2-server/src/main/java/org/graylog/plugins/views/search/Query.java

@@ -32,7 +32,9 @@
 import org.graylog.plugins.views.search.engine.BackendQuery;
 import org.graylog.plugins.views.search.engine.EmptyTimeRange;
 import org.graylog.plugins.views.search.filter.AndFilter;
+import org.graylog.plugins.views.search.filter.StreamCategoryFilter;
 import org.graylog.plugins.views.search.filter.StreamFilter;
+import org.graylog.plugins.views.search.permissions.StreamPermissions;
 import org.graylog.plugins.views.search.rest.ExecutionState;
 import org.graylog.plugins.views.search.rest.ExecutionStateGlobalOverride;
 import org.graylog.plugins.views.search.rest.SearchTypeExecutionState;
@@ -50,13 +52,17 @@
 
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
+import java.util.Collection;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.UUID;
+import java.util.function.Function;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
 
 import static com.google.common.base.MoreObjects.firstNonNull;
@@ -210,6 +216,20 @@ public Set<String> usedStreamIds() {
                 .orElse(Collections.emptySet());
     }
 
+    @SuppressWarnings("UnstableApiUsage")
+    public Set<String> usedStreamCategories() {
+        return Optional.ofNullable(filter())
+                .map(optFilter -> {
+                    final Traverser<Filter> filterTraverser = Traverser.forTree(filter -> firstNonNull(filter.filters(), Collections.emptySet()));
+                    return StreamSupport.stream(filterTraverser.breadthFirst(optFilter).spliterator(), false)
+                            .filter(filter -> filter instanceof StreamCategoryFilter)
+                            .map(streamFilter -> ((StreamCategoryFilter) streamFilter).category())
+                            .filter(Objects::nonNull)
+                            .collect(toSet());
+                })
+                .orElse(Collections.emptySet());
+    }
+
     public Set<String> streamIdsForPermissionsCheck() {
         final Set<String> searchTypeStreamIds = searchTypes().stream()
                 .map(SearchType::streams)
@@ -219,7 +239,7 @@ public Set<String> streamIdsForPermissionsCheck() {
     }
 
     public boolean hasStreams() {
-        return !usedStreamIds().isEmpty();
+        return !(usedStreamIds().isEmpty() && usedStreamCategories().isEmpty());
     }
 
     public boolean hasReferencedStreamFilters() {
@@ -231,6 +251,39 @@ public Query addStreamsToFilter(Set<String> streamIds) {
         return toBuilder().filter(newFilter).build();
     }
 
+    public Query replaceStreamCategoryFilters(Function<Collection<String>, Stream<String>> categoryMappingFunction,
+                                              StreamPermissions streamPermissions) {
+        if (filter() == null) {
+            return this;
+        }
+        return toBuilder()
+                .filter(streamCategoryToStreamFiltersRecursively(filter(), categoryMappingFunction, streamPermissions))
+                .build();
+    }
+
+    private Filter streamCategoryToStreamFiltersRecursively(Filter filter,
+                                                            Function<Collection<String>, Stream<String>> categoryMappingFunction,
+                                                            StreamPermissions streamPermissions) {
+        if (filter.filters() == null || filter.filters().isEmpty()) {
+            return filter;
+        }
+        Set<Filter> mappedFilters = new HashSet<>();
+        for (Filter f : filter.filters()) {
+            Filter mappedFilter = f;
+            if (f instanceof StreamCategoryFilter scf) {
+                mappedFilter = scf.toStreamFilter(categoryMappingFunction, streamPermissions);
+            }
+            if (mappedFilter != null) {
+                mappedFilter = streamCategoryToStreamFiltersRecursively(mappedFilter, categoryMappingFunction, streamPermissions);
+                mappedFilters.add(mappedFilter);
+            }
+        }
+        if (mappedFilters.isEmpty()) {
+            return null;
+        }
+        return filter.toGenericBuilder().filters(mappedFilters.stream().filter(Objects::nonNull).collect(toSet())).build();
+    }
+
     private Filter addStreamsTo(Filter filter, Set<String> streamIds) {
         final Filter streamIdFilter = StreamFilter.anyIdOf(streamIds.toArray(new String[]{}));
         if (filter == null) {

graylog2-server/src/main/java/org/graylog/plugins/views/search/Search.java

@@ -28,6 +28,7 @@
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
 import com.google.common.graph.MutableGraph;
+import org.graylog.plugins.views.search.permissions.StreamPermissions;
 import org.graylog.plugins.views.search.rest.ExecutionState;
 import org.graylog.plugins.views.search.views.PluginMetadataSummary;
 import org.graylog2.contentpacks.ContentPackable;
@@ -43,13 +44,17 @@
 
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
+import java.util.Collection;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.LinkedHashSet;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import static com.google.common.collect.ImmutableSet.toImmutableSet;
 import static java.util.stream.Collectors.toSet;
@@ -146,10 +151,35 @@ public Search addStreamsToQueriesWithoutStreams(Supplier<Set<String>> defaultStr
         return toBuilder().queries(newQueries).build();
     }
 
+    public Search addStreamsToQueriesWithCategories(Function<Collection<String>, Stream<String>> categoryMappingFunction,
+                                                    StreamPermissions streamPermissions) {
+        if (!hasQueriesWithStreamCategories()) {
+            return this;
+        }
+        final Set<Query> withStreamCategories = queries().stream().filter(q -> !q.usedStreamCategories().isEmpty()).collect(toSet());
+        final Set<Query> withoutStreamCategories = Sets.difference(queries(), withStreamCategories);
+        final Set<Query> withMappedStreamCategories = new HashSet<>();
+
+        for (Query query : withStreamCategories) {
+            final Set<String> mappedStreamIds = categoryMappingFunction.apply(query.usedStreamCategories())
+                    .filter(streamPermissions::canReadStream)
+                    .collect(toSet());
+            withMappedStreamCategories.add(query.addStreamsToFilter(mappedStreamIds));
+        }
+
+        final ImmutableSet<Query> newQueries = Sets.union(withMappedStreamCategories, withoutStreamCategories).immutableCopy();
+
+        return toBuilder().queries(newQueries).build();
+    }
+
     private boolean hasQueriesWithoutStreams() {
         return !queries().stream().allMatch(Query::hasStreams);
     }
 
+    private boolean hasQueriesWithStreamCategories() {
+        return queries().stream().anyMatch(q -> !q.usedStreamCategories().isEmpty());
+    }
+
     public abstract Builder toBuilder();
 
     public static Builder builder() {

graylog2-server/src/main/java/org/graylog/plugins/views/search/engine/normalization/PluggableSearchNormalization.java

@@ -24,27 +24,34 @@
 import org.graylog.plugins.views.search.rest.ExecutionState;
 import org.graylog.plugins.views.search.rest.ExecutionStateGlobalOverride;
 import org.graylog2.plugin.Tools;
+import org.graylog2.streams.StreamService;
 import org.joda.time.DateTime;
 
+import java.util.Collection;
 import java.util.Collections;
 import java.util.Optional;
 import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Stream;
 
 import static com.google.common.base.MoreObjects.firstNonNull;
 
 public class PluggableSearchNormalization implements SearchNormalization {
     private final Set<SearchNormalizer> pluggableNormalizers;
     private final Set<SearchNormalizer> postValidationNormalizers;
+    private final Function<Collection<String>, Stream<String>> streamCategoryMapper;
 
     @Inject
     public PluggableSearchNormalization(Set<SearchNormalizer> pluggableNormalizers,
-                                        @PostValidation Set<SearchNormalizer> postValidationNormalizers) {
+                                        @PostValidation Set<SearchNormalizer> postValidationNormalizers,
+                                        StreamService streamService) {
         this.pluggableNormalizers = pluggableNormalizers;
         this.postValidationNormalizers = postValidationNormalizers;
+        this.streamCategoryMapper = (categories) -> streamService.mapCategoriesToIds(categories).stream();
     }
 
-    public PluggableSearchNormalization(Set<SearchNormalizer> pluggableNormalizers) {
-        this(pluggableNormalizers, Collections.emptySet());
+    public PluggableSearchNormalization(Set<SearchNormalizer> pluggableNormalizers, StreamService streamService) {
+        this(pluggableNormalizers, Collections.emptySet(), streamService);
     }
 
     private Search normalize(Search search, Set<SearchNormalizer> normalizers) {
@@ -68,7 +75,9 @@ private Query normalize(final Query query,
 
     @Override
     public Search preValidation(Search search, SearchUser searchUser, ExecutionState executionState) {
-        final Search searchWithStreams = search.addStreamsToQueriesWithoutStreams(() -> searchUser.streams().loadMessageStreamsWithFallback());
+        final Search searchWithStreams = search
+                .addStreamsToQueriesWithoutStreams(() -> searchUser.streams().loadMessageStreamsWithFallback())
+                .addStreamsToQueriesWithCategories(streamCategoryMapper, searchUser);
         final var now = referenceDateFromOverrideOrNow(executionState);
         final var normalizedSearch = searchWithStreams.applyExecutionState(firstNonNull(executionState, ExecutionState.empty()))
                 .withReferenceDate(now);
@@ -93,6 +102,8 @@ public Query preValidation(final Query query, final ParameterProvider parameterP
         Query normalizedQuery = query;
         if (!query.hasStreams()) {
             normalizedQuery = query.addStreamsToFilter(searchUser.streams().loadMessageStreamsWithFallback());
+        } else if (!query.usedStreamCategories().isEmpty()) {
+            normalizedQuery = query.replaceStreamCategoryFilters(streamCategoryMapper, searchUser);
         }
 
         if (!executionState.equals(ExecutionState.empty())) {

graylog2-server/src/main/java/org/graylog/plugins/views/search/filter/StreamCategoryFilter.java

@@ -0,0 +1,104 @@
+/*
+ * Copyright (C) 2020 Graylog, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the Server Side Public License, version 1,
+ * as published by MongoDB, Inc.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * Server Side Public License for more details.
+ *
+ * You should have received a copy of the Server Side Public License
+ * along with this program. If not, see
+ * <http://www.mongodb.com/licensing/server-side-public-license>.
+ */
+package org.graylog.plugins.views.search.filter;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.google.auto.value.AutoValue;
+import org.graylog.plugins.views.search.Filter;
+import org.graylog.plugins.views.search.permissions.StreamPermissions;
+
+import javax.annotation.Nullable;
+import java.util.Collection;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Stream;
+
+@AutoValue
+@JsonTypeName(StreamCategoryFilter.NAME)
+@JsonDeserialize(builder = StreamCategoryFilter.Builder.class)
+public abstract class StreamCategoryFilter implements Filter {
+    public static final String NAME = "stream_category";
+
+    @Override
+    @JsonProperty
+    public abstract String type();
+
+    @Override
+    @Nullable
+    @JsonProperty
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    public abstract Set<Filter> filters();
+
+    @JsonProperty("category")
+    public abstract String category();
+
+    public static Builder builder() {
+        return Builder.create();
+    }
+
+    public abstract Builder toBuilder();
+
+    public static StreamCategoryFilter ofCategory(String category) {
+        return builder().category(category).build();
+    }
+
+    public Filter toStreamFilter(Function<Collection<String>, Stream<String>> categoryMappingFunction,
+                                 StreamPermissions streamPermissions) {
+        String[] mappedStreamIds = categoryMappingFunction.apply(List.of(category()))
+                .filter(streamPermissions::canReadStream)
+                .toArray(String[]::new);
+        // If the streamPermissions do not allow for any of the streams to be read, nullify this filter.
+        if (mappedStreamIds.length == 0) {
+            return null;
+        }
+        // Replace this category with an OrFilter of stream IDs and then add filters if they exist.
+        Filter streamFilter = StreamFilter.anyIdOf(mappedStreamIds).toGenericBuilder().build();
+        if (filters() != null) {
+            streamFilter = streamFilter.toGenericBuilder().filters(filters()).build();
+        }
+        return streamFilter;
+    }
+
+    @Override
+    public Filter.Builder toGenericBuilder() {
+        return toBuilder();
+    }
+
+    @AutoValue.Builder
+    public abstract static class Builder implements Filter.Builder {
+        @JsonProperty
+        public abstract Builder type(String type);
+
+        @JsonProperty
+        public abstract Builder filters(@Nullable Set<Filter> filters);
+
+        @JsonProperty("category")
+        public abstract Builder category(String category);
+
+        public abstract StreamCategoryFilter build();
+
+        @JsonCreator
+        public static Builder create() {
+            return new AutoValue_StreamCategoryFilter.Builder().type(NAME);
+        }
+    }
+}