Enabling RBAC check for ACL (#3813)

Co-authored-by: iliax <ikuramshin@provectus.com>

Author: iliax

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/rbac/AccessControlService.java

@@ -108,7 +108,8 @@ && isConsumerGroupAccessible(context, user)
                   && isConnectAccessible(context, user)
                   && isConnectorAccessible(context, user) // TODO connector selectors
                   && isSchemaAccessible(context, user)
-                  && isKsqlAccessible(context, user);
+                  && isKsqlAccessible(context, user)
+                  && isAclAccessible(context, user);
 
           if (!accessGranted) {
             throw new AccessDeniedException("Access denied");
@@ -364,6 +365,23 @@ private boolean isKsqlAccessible(AccessContext context, AuthenticatedUser user)
     return isAccessible(Resource.KSQL, null, user, context, requiredActions);
   }
 
+  private boolean isAclAccessible(AccessContext context, AuthenticatedUser user) {
+    if (!rbacEnabled) {
+      return true;
+    }
+
+    if (context.getAclActions().isEmpty()) {
+      return true;
+    }
+
+    Set<String> requiredActions = context.getAclActions()
+        .stream()
+        .map(a -> a.toString().toUpperCase())
+        .collect(Collectors.toSet());
+
+    return isAccessible(Resource.ACL, null, user, context, requiredActions);
+  }
+
   public Set<ProviderAuthorityExtractor> getOauthExtractors() {
     return oauthExtractors;
   }