Repository: provectus/kafka-ui
Issue: The 'all-topics' view does not display according to the RBAC filter
Labels: type/bug
scope/backend
status/accepted
area/rbac

### Issue submitter TODO list

- [X] I've looked up my issue in [FAQ](https://docs.kafka-ui.provectus.io/faq/common-problems)
- [X] I've searched for an already existing issues [here](https://github.com/provectus/kafka-ui/issues)
- [X] I've tried running `master`-labeled docker image and the issue still persists there
- [X] I'm running a supported version of the application which is listed [here](https://github.com/provectus/kafka-ui/blob/master/SECURITY.md)

### Describe the bug (actual behavior)

I was working with RBAC to filter permissions by topics, and the issue I'm facing is that in the "all-topics" URI, unlike with schemas or other tabs, it doesn't display the filtered topic unless I search for them in the search bar.

### Expected behavior

It should display, like in the rest of the tabs, only the topics that comply with the RBAC configuration.

### Your installation details

1 - 7.0 version
2 - Im not using it
3 - This is the RBAC that I'm using

resource: topic
        value: "dev.*"
        actions: all



### Steps to reproduce

Simply modify the configuration file in the topic permissions section to allow access only to specific topics.

### Screenshots

![image](https://github.com/provectus/kafka-ui/assets/60904917/eee159da-cbd5-45a5-8cd6-6ecd9a7cb8c3)
![image](https://github.com/provectus/kafka-ui/assets/60904917/2ec68880-8d26-44ec-9886-00f8cb82e275)


### Logs

_No response_

### Additional context

1 - I have tried all the methods listed in the documentation to define permissions on a topic

Issue comments:
Hello there javihernanp! 👋

Thank you and congratulations 🎉 for opening your very first issue in this project! 💖

In case you want to claim this issue, please comment down below! We will try to get back to you as soon as we can. 👀
Hey, it seems to be working for me:
```yaml
        - resource: topic
          value: "dev.*"
          actions: all
```

<img width="940" alt="image" src="https://github.com/provectus/kafka-ui/assets/1494347/84eb9b78-665e-47d4-824c-4800d2624f41">

Please provide your full config and responses for `/api/authorization` and `/api/clusters/local/topics` requests, could be found in network tab within browser's dev console.
Hello,

Thank you for your quick response,

This is the config file that I'm using:

```
kafka:
  clusters:
    -
      name: LAB01
      bootstrapServers: cluster1.confluent.cloud:9092
      properties.security.protocol: SASL_SSL
      properties.sasl.mechanism: PLAIN
      properties.sasl.jaas.config: 'org.apache.kafka.common.security.plain.PlainLoginModule required username="****" password="****";'
    -
      name: NON_PROD_AWS_EU-WEST-3
      bootstrapServers: cluster2.confluent.cloud:9092
      properties.security.protocol: SASL_SSL
      properties.sasl.mechanism: PLAIN
      properties.sasl.jaas.config: 'org.apache.kafka.common.security.plain.PlainLoginModule required username="****" password="****";'
      schemaRegistry: https://example
      schemaRegistryAuth:
        username:  username
        password: password

auth:
  type: OAUTH2
  oauth2:
    client:
      github:
        provider: github
        clientId: ****
        clientSecret: ****
        scope:
          - read:org
        user-name-attribute: login
        usernameattribute: login
        custom-params:
          type: github
        authorization-uri: https://example/login/oauth/authorize
        token-uri: https://example/login/oauth/access_token
        tokenuri: https://example/login/oauth/access_token
        user-info-uri: https://example/api/v3/user
        redirect-uri: https://ui.example/login/oauth2/code/github

rbac:
  roles:
    - name: "Admin Team"
      subjects:
      - provider: oauth_github
        type: organization
        value: "organization"
      clusters:
      - LAB01
      - NON_PROD_AWS_EU-WEST-3
      permissions:
      - resource: clusterconfig
        actions: all
      - resource: topic
        value: ".*"
        actions: all
      - resource: consumer
        value: ".*"
        actions: all
      - resource: schema
        value: ".*"
        actions: all
      - resource: connect
        value: ".*"
        actions: all
      - resource: ksql
        actions: all
        value: ".*"
      - resource: acl
        value: ".*"
        actions: all
    - name: "test"
      subjects:
      - provider: oauth_github
        type: user
        value: "fcojavier-hernandez"
      clusters:
      - NON_PROD_AWS_EU-WEST-3
      permissions:
      - resource: clusterconfig
        value: ".*"
        actions: all
      - resource: topic
        value: "dev.*"
        actions: all
      - resource: consumer
        value: ".*"
        actions: all
      - resource: schema
        value: ".*"
        actions: all
      - resource: connect
        value: ".*"
        actions: all
      - resource: ksql
        actions: all
        value: ".*"
      - resource: acl
        value: ".*"
        actions: all

```
For the two outputs you requested, here are both screenshots:

![image](https://github.com/provectus/kafka-ui/assets/60904917/bbd520e2-b621-4e31-b544-ac54f51c729d)
![image](https://github.com/provectus/kafka-ui/assets/60904917/f376768a-6315-40fc-8f6e-80f624bed437)

I would like to ask if you are using the release 0.7

Kind regards.
@javihernanp 

>I would like to ask if you are using the release 0.7
Even better, master branch.

Which cluster stores the topic in question? Your screenshots are cropped, no way to determine that.
Is that the same cluster that returns `topics: []`?


Hello,

Indeed, the cluster I obtain for topic[] is the same as the one I have configured, namely "NON_PROD_AWS_EU-WEST-3". I have continued testing, and it works correctly with a Confluent Cloud Basic cluster. However, if I configure a Confluent Cloud Dedicated cluster, such as "NON_PROD_AWS_EU-WEST-3", it doesn't work correctly. I have checked the Service Accounts I am using, and they have the same permissions in both types of clusters. Furthermore, it is worth mentioning that I am also using the code from the master branch.

Kind regards.
1. Do you have these topics available if RBAC is disabled (no roles specified in config)? 
2. Are there any cluster/topic authorization errors in logs?
Hello,

Do you have these topics available if RBAC is disabled (no roles specified in config)?

 -  Yes

Are there any cluster/topic authorization errors in logs? 

 - I could not see configuring different levels of logs, nothing related to authorization errors

Kind regards
@javihernanp please try running this image:
`public.ecr.aws/provectus/kafka-ui-custom-build:3914`
with the following config property:
`logging.level.com.provectus.kafka.ui.service.rbac=TRACE`
Open the topic list page, also try searching by the topic name as you did, and extract the logs after that.
Hello @Haarolean,

I have been testing by modifying the log level as you indicated, and I have used that version, but I still haven't been able to see anything different in the console output. Would it be possible to have a quick meeting to explain in detail the issue and the different tests we have been conducting?

Kind regards
@javihernanp Sure, let's talk. Got discord? The invite link [here](https://discord.gg/4DWzD7pGE5).
@Haarolean, we have seen the same/similar behavior, but can add a bit more context.  In our case, we RBAC topic filter on "a.*" and "B.*" as a test.  What happens is we still see 39 pages of topics as if there was no RBAC topic filter but most pages are blank.  Topics "B.*" show up on page #1 and topics "a.*" show up on pages 10-11.  So sorted by case and the filter just prevents the topics from showing on the page were it would be.  If you do try to add a non-allowed topic name to the URL directly, you do get a 403.
Hello @Haarolean,

I'm sorry, but at that time I was already disconnected and couldn't see that you sent me a link. I have been testing what @dmalik16 mentioned in the previous comment, and indeed, that is the cause of the error. I have been testing it, and it does show up on the original pages as if all the topics were displayed, just as explained. Since we already know the origin of the issue, I believe it is no longer necessary to have a call to further investigate the issue. Please let me know if you need more information. I would like to know an estimated time for the resolution of the issue., and thank you very much for the support and work done so far.

Best regards.
@javihernanp @dmalik16 could you please try this docker image and tell me if you experience any problems now?
`image: public.ecr.aws/provectus/kafka-ui-custom-build:3946`
@Haarolean, we pulled that image down and tested it out.  It now displays as expected.  In our case the filtered topics appeared over 2 pages total as if they were the only topics.
Thanks!
Hello @Haarolean ,

I have tested the changes, and now it is working correctly with the applied RBAC configuration. Thank you very much for resolving the issue promptly. I was wondering if these changes will be included in the main branch in the next patch. If so, could you please let me know the estimated release date for that patch? Once again, I would like to express my appreciation for the work done and the speed at which it was resolved.

Best regards.

Pull request review comments:
I think it should be `when(mock.filterViewableTopics(any(), any())).then(i -> Mono.just(i.getArgument(0)));`
thanks, done

Patch that resolved the issue:
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/controller/TopicsController.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/controller/TopicsController.java
index a587d53770a..818fab84d68 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/controller/TopicsController.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/controller/TopicsController.java
@@ -167,12 +167,13 @@ public Mono<ResponseEntity<TopicsResponseDTO>> getTopics(String clusterName,
                                                            ServerWebExchange exchange) {
 
     return topicsService.getTopicsForPagination(getCluster(clusterName))
-        .flatMap(existingTopics -> {
+        .flatMap(topics -> accessControlService.filterViewableTopics(topics, clusterName))
+        .flatMap(topics -> {
           int pageSize = perPage != null && perPage > 0 ? perPage : DEFAULT_PAGE_SIZE;
           var topicsToSkip = ((page != null && page > 0 ? page : 1) - 1) * pageSize;
           var comparator = sortOrder == null || !sortOrder.equals(SortOrderDTO.DESC)
               ? getComparatorForTopic(orderBy) : getComparatorForTopic(orderBy).reversed();
-          List<InternalTopic> filtered = existingTopics.stream()
+          List<InternalTopic> filtered = topics.stream()
               .filter(topic -> !topic.isInternal()
                   || showInternal != null && showInternal)
               .filter(topic -> search == null || StringUtils.containsIgnoreCase(topic.getName(), search))
@@ -189,7 +190,6 @@ public Mono<ResponseEntity<TopicsResponseDTO>> getTopics(String clusterName,
 
           return topicsService.loadTopics(getCluster(clusterName), topicsPage)
               .flatMapMany(Flux::fromIterable)
-              .filterWhen(dto -> accessControlService.isTopicAccessible(dto, clusterName))
               .collectList()
               .map(topicsToRender ->
                   new TopicsResponseDTO()
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/rbac/AccessControlService.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/rbac/AccessControlService.java
index b507d4c0810..debd9acd829 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/rbac/AccessControlService.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/rbac/AccessControlService.java
@@ -202,19 +202,23 @@ public boolean isTopicAccessible(AccessContext context, AuthenticatedUser user)
     return isAccessible(Resource.TOPIC, context.getTopic(), user, context, requiredActions);
   }
 
-  public Mono<Boolean> isTopicAccessible(InternalTopic dto, String clusterName) {
+  public Mono<List<InternalTopic>> filterViewableTopics(List<InternalTopic> topics, String clusterName) {
     if (!rbacEnabled) {
-      return Mono.just(true);
+      return Mono.just(topics);
     }
 
-    AccessContext accessContext = AccessContext
-        .builder()
-        .cluster(clusterName)
-        .topic(dto.getName())
-        .topicActions(TopicAction.VIEW)
-        .build();
-
-    return getUser().map(u -> isTopicAccessible(accessContext, u));
+    return getUser()
+        .map(user -> topics.stream()
+            .filter(topic -> {
+                  var accessContext = AccessContext
+                      .builder()
+                      .cluster(clusterName)
+                      .topic(topic.getName())
+                      .topicActions(TopicAction.VIEW)
+                      .build();
+                  return isTopicAccessible(accessContext, user);
+                }
+            ).toList());
   }
 
   private boolean isConsumerGroupAccessible(AccessContext context, AuthenticatedUser user) {


Test patch:
diff --git a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/util/AccessControlServiceMock.java b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/util/AccessControlServiceMock.java
index 263f1367d8c..852b75ae085 100644
--- a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/util/AccessControlServiceMock.java
+++ b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/util/AccessControlServiceMock.java
@@ -5,6 +5,7 @@
 import static org.mockito.Mockito.when;
 
 import com.provectus.kafka.ui.service.rbac.AccessControlService;
+import java.util.Collections;
 import org.mockito.Mockito;
 import reactor.core.publisher.Mono;
 
@@ -16,7 +17,7 @@ public AccessControlService getMock() {
     when(mock.validateAccess(any())).thenReturn(Mono.empty());
     when(mock.isSchemaAccessible(anyString(), anyString())).thenReturn(Mono.just(true));
 
-    when(mock.isTopicAccessible(any(), anyString())).thenReturn(Mono.just(true));
+    when(mock.filterViewableTopics(any(), any())).then(invocation -> Mono.just(invocation.getArgument(0)));
 
     return mock;
   }