Expose cluster ACL list

documentation/compose/jaas/kafka_server.conf

@@ -11,4 +11,8 @@ KafkaClient {
     user_admin="admin-secret";
 };
 
-Client {};
+Client {
+       org.apache.zookeeper.server.auth.DigestLoginModule required
+       username="zkuser"
+       password="zkuserpassword";
+};

documentation/compose/jaas/zookeeper_jaas.conf

@@ -0,0 +1,4 @@
+Server {
+       org.apache.zookeeper.server.auth.DigestLoginModule required
+       user_zkuser="zkuserpassword";
+};

documentation/compose/kafka-ui-sasl.yaml

@@ -8,15 +8,26 @@ services:
     ports:
       - 8080:8080
     depends_on:
+      - zookeeper
       - kafka
     environment:
       KAFKA_CLUSTERS_0_NAME: local
       KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:29092
       KAFKA_CLUSTERS_0_PROPERTIES_SECURITY_PROTOCOL: SASL_PLAINTEXT
       KAFKA_CLUSTERS_0_PROPERTIES_SASL_MECHANISM: PLAIN
+#      KAFKA_CLUSTERS_0_PROPERTIES_SASL_JAAS_CONFIG: 'org.apache.kafka.common.security.plain.PlainLoginModule required username="enzo" password="cisternino";'
       KAFKA_CLUSTERS_0_PROPERTIES_SASL_JAAS_CONFIG: 'org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin-secret";'
       DYNAMIC_CONFIG_ENABLED: true # not necessary for sasl auth, added for tests
 
+  zookeeper:
+    image: wurstmeister/zookeeper:3.4.6
+    environment:
+      JVMFLAGS: "-Djava.security.auth.login.config=/etc/zookeeper/zookeeper_jaas.conf"
+    volumes:
+      - ./jaas/zookeeper_jaas.conf:/etc/zookeeper/zookeeper_jaas.conf
+    ports:
+      - 2181:2181
+
   kafka:
     image: confluentinc/cp-kafka:7.2.1
     hostname: kafka
@@ -26,27 +37,25 @@ services:
       - "9997:9997"
     environment:
       KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
       KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,SASL_PLAINTEXT:SASL_PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT'
       KAFKA_ADVERTISED_LISTENERS: 'SASL_PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092'
       KAFKA_OPTS: "-Djava.security.auth.login.config=/etc/kafka/jaas/kafka_server.conf"
+      KAFKA_AUTHORIZER_CLASS_NAME: "kafka.security.authorizer.AclAuthorizer"
       KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
       KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
       KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
       KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
       KAFKA_JMX_PORT: 9997
       KAFKA_JMX_HOSTNAME: localhost
-      KAFKA_PROCESS_ROLES: 'broker,controller'
       KAFKA_NODE_ID: 1
       KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka:29093'
       KAFKA_LISTENERS: 'SASL_PLAINTEXT://kafka:29092,CONTROLLER://kafka:29093,PLAINTEXT_HOST://0.0.0.0:9092'
       KAFKA_INTER_BROKER_LISTENER_NAME: 'SASL_PLAINTEXT'
       KAFKA_SASL_ENABLED_MECHANISMS: 'PLAIN'
       KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL: 'PLAIN'
-      KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
-      KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs'
       KAFKA_SECURITY_PROTOCOL: 'SASL_PLAINTEXT'
-      KAFKA_SUPER_USERS: 'User:admin,User:enzo'
+      KAFKA_SUPER_USERS: 'User:admin'
     volumes:
       - ./scripts/update_run.sh:/tmp/update_run.sh
       - ./jaas:/etc/kafka/jaas
-    command: "bash -c 'if [ ! -f /tmp/update_run.sh ]; then echo \"ERROR: Did you forget the update_run.sh file that came with this docker-compose.yml file?\" && exit 1 ; else /tmp/update_run.sh && /etc/confluent/docker/run ; fi'"

kafka-ui-api/src/main/java/com/provectus/kafka/ui/controller/AclsController.java

@@ -0,0 +1,94 @@
+package com.provectus.kafka.ui.controller;
+
+import com.provectus.kafka.ui.api.AclsApi;
+import com.provectus.kafka.ui.mapper.ClusterMapper;
+import com.provectus.kafka.ui.model.KafkaAclDTO;
+import com.provectus.kafka.ui.model.rbac.AccessContext;
+import com.provectus.kafka.ui.model.rbac.permission.AclAction;
+import com.provectus.kafka.ui.service.acl.AclsService;
+import com.provectus.kafka.ui.service.rbac.AccessControlService;
+import lombok.RequiredArgsConstructor;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.server.ServerWebExchange;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+@RestController
+@RequiredArgsConstructor
+public class AclsController extends AbstractController implements AclsApi {
+
+  private final AclsService aclsService;
+  private final AccessControlService accessControlService;
+
+  @Override
+  public Mono<ResponseEntity<Void>> createAcl(String clusterName, Mono<KafkaAclDTO> kafkaAclDto,
+                                              ServerWebExchange exchange) {
+    AccessContext context = AccessContext.builder()
+        .cluster(clusterName)
+        .aclActions(AclAction.EDIT)
+        .build();
+
+    return accessControlService.validateAccess(context)
+        .then(kafkaAclDto)
+        .map(ClusterMapper::toAclBinding)
+        .flatMap(binding -> aclsService.createAcl(getCluster(clusterName), binding))
+        .thenReturn(ResponseEntity.ok().build());
+  }
+
+  @Override
+  public Mono<ResponseEntity<Void>> deleteAcl(String clusterName, Mono<KafkaAclDTO> kafkaAclDto,
+                                              ServerWebExchange exchange) {
+    AccessContext context = AccessContext.builder()
+        .cluster(clusterName)
+        .aclActions(AclAction.EDIT)
+        .build();
+
+    return accessControlService.validateAccess(context)
+        .then(kafkaAclDto)
+        .map(ClusterMapper::toAclBinding)
+        .flatMap(binding -> aclsService.deleteAcl(getCluster(clusterName), binding))
+        .thenReturn(ResponseEntity.ok().build());
+  }
+
+  @Override
+  public Mono<ResponseEntity<Flux<KafkaAclDTO>>> listAcls(String clusterName, ServerWebExchange exchange) {
+    AccessContext context = AccessContext.builder()
+        .cluster(clusterName)
+        .aclActions(AclAction.VIEW)
+        .build();
+
+    return accessControlService.validateAccess(context).then(
+        Mono.just(
+            ResponseEntity.ok(
+                aclsService.listAcls(getCluster(clusterName)).map(ClusterMapper::toKafkaAclDto)))
+    );
+  }
+
+  @Override
+  public Mono<ResponseEntity<String>> getAclAsCsv(String clusterName, ServerWebExchange exchange) {
+    AccessContext context = AccessContext.builder()
+        .cluster(clusterName)
+        .aclActions(AclAction.VIEW)
+        .build();
+
+    return accessControlService.validateAccess(context).then(
+        aclsService.getAclAsCsvString(getCluster(clusterName))
+            .map(ResponseEntity::ok)
+            .flatMap(Mono::just)
+    );
+  }
+
+  @Override
+  public Mono<ResponseEntity<Void>> syncAclsCsv(String clusterName, Mono<String> csvMono, ServerWebExchange exchange) {
+    AccessContext context = AccessContext.builder()
+        .cluster(clusterName)
+        .aclActions(AclAction.EDIT)
+        .build();
+
+    return accessControlService.validateAccess(context)
+        .then(csvMono)
+        .flatMap(csv -> aclsService.syncAclWithAclCsv(getCluster(clusterName), csv))
+        .thenReturn(ResponseEntity.ok().build());
+  }
+}

kafka-ui-api/src/main/java/com/provectus/kafka/ui/mapper/ClusterMapper.java

@@ -20,19 +20,26 @@
 import com.provectus.kafka.ui.model.InternalReplica;
 import com.provectus.kafka.ui.model.InternalTopic;
 import com.provectus.kafka.ui.model.InternalTopicConfig;
+import com.provectus.kafka.ui.model.KafkaAclDTO;
 import com.provectus.kafka.ui.model.MetricDTO;
 import com.provectus.kafka.ui.model.Metrics;
 import com.provectus.kafka.ui.model.PartitionDTO;
 import com.provectus.kafka.ui.model.ReplicaDTO;
 import com.provectus.kafka.ui.model.TopicConfigDTO;
 import com.provectus.kafka.ui.model.TopicDTO;
 import com.provectus.kafka.ui.model.TopicDetailsDTO;
-import com.provectus.kafka.ui.service.masking.DataMasking;
 import com.provectus.kafka.ui.service.metrics.RawMetric;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
 import org.apache.kafka.clients.admin.ConfigEntry;
+import org.apache.kafka.common.acl.AccessControlEntry;
+import org.apache.kafka.common.acl.AclBinding;
+import org.apache.kafka.common.acl.AclOperation;
+import org.apache.kafka.common.acl.AclPermissionType;
+import org.apache.kafka.common.resource.PatternType;
+import org.apache.kafka.common.resource.ResourcePattern;
+import org.apache.kafka.common.resource.ResourceType;
 import org.mapstruct.Mapper;
 import org.mapstruct.Mapping;
 
@@ -109,8 +116,66 @@ default BrokerDiskUsageDTO map(Integer id, InternalBrokerDiskUsage internalBroke
     return brokerDiskUsage;
   }
 
-  default DataMasking map(List<ClustersProperties.Masking> maskingProperties) {
-    return DataMasking.create(maskingProperties);
+  static KafkaAclDTO.OperationEnum mapAclOperation(AclOperation operation) {
+    return switch (operation) {
+      case ALL -> KafkaAclDTO.OperationEnum.ALL;
+      case READ -> KafkaAclDTO.OperationEnum.READ;
+      case WRITE -> KafkaAclDTO.OperationEnum.WRITE;
+      case CREATE -> KafkaAclDTO.OperationEnum.CREATE;
+      case DELETE -> KafkaAclDTO.OperationEnum.DELETE;
+      case ALTER -> KafkaAclDTO.OperationEnum.ALTER;
+      case DESCRIBE -> KafkaAclDTO.OperationEnum.DESCRIBE;
+      case CLUSTER_ACTION -> KafkaAclDTO.OperationEnum.CLUSTER_ACTION;
+      case DESCRIBE_CONFIGS -> KafkaAclDTO.OperationEnum.DESCRIBE_CONFIGS;
+      case ALTER_CONFIGS -> KafkaAclDTO.OperationEnum.ALTER_CONFIGS;
+      case IDEMPOTENT_WRITE -> KafkaAclDTO.OperationEnum.IDEMPOTENT_WRITE;
+      case CREATE_TOKENS -> KafkaAclDTO.OperationEnum.CREATE_TOKENS;
+      case DESCRIBE_TOKENS -> KafkaAclDTO.OperationEnum.DESCRIBE_TOKENS;
+      case ANY -> throw new IllegalArgumentException("ANY operation can be only part of filter");
+      case UNKNOWN -> KafkaAclDTO.OperationEnum.UNKNOWN;
+    };
+  }
+
+  static KafkaAclDTO.ResourceTypeEnum mapAclResourceType(ResourceType resourceType) {
+    return switch (resourceType) {
+      case CLUSTER -> KafkaAclDTO.ResourceTypeEnum.CLUSTER;
+      case TOPIC -> KafkaAclDTO.ResourceTypeEnum.TOPIC;
+      case GROUP -> KafkaAclDTO.ResourceTypeEnum.GROUP;
+      case DELEGATION_TOKEN -> KafkaAclDTO.ResourceTypeEnum.DELEGATION_TOKEN;
+      case TRANSACTIONAL_ID -> KafkaAclDTO.ResourceTypeEnum.TRANSACTIONAL_ID;
+      case USER -> KafkaAclDTO.ResourceTypeEnum.USER;
+      case ANY -> throw new IllegalArgumentException("ANY type can be only part of filter");
+      case UNKNOWN -> KafkaAclDTO.ResourceTypeEnum.UNKNOWN;
+    };
+  }
+
+  static AclBinding toAclBinding(KafkaAclDTO dto) {
+    return new AclBinding(
+        new ResourcePattern(
+            ResourceType.valueOf(dto.getResourceType().name()),
+            dto.getResourceName(),
+            PatternType.valueOf(dto.getNamePatternType().name())
+        ),
+        new AccessControlEntry(
+            dto.getPrincipal(),
+            dto.getHost(),
+            AclOperation.valueOf(dto.getOperation().name()),
+            AclPermissionType.valueOf(dto.getPermission().name())
+        )
+    );
+  }
+
+  static KafkaAclDTO toKafkaAclDto(AclBinding binding) {
+    var pattern = binding.pattern();
+    var filter = binding.toFilter().entryFilter();
+    return new KafkaAclDTO()
+        .resourceType(mapAclResourceType(pattern.resourceType()))
+        .resourceName(pattern.name())
+        .namePatternType(KafkaAclDTO.NamePatternTypeEnum.fromValue(pattern.patternType().name()))
+        .principal(filter.principal())
+        .host(filter.host())
+        .operation(mapAclOperation(filter.operation()))
+        .permission(KafkaAclDTO.PermissionEnum.fromValue(filter.permissionType().name()));
   }
 
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/ClusterFeature.java

@@ -4,5 +4,7 @@ public enum ClusterFeature {
   KAFKA_CONNECT,
   KSQL_DB,
   SCHEMA_REGISTRY,
-  TOPIC_DELETION
+  TOPIC_DELETION,
+  KAFKA_ACL_VIEW,
+  KAFKA_ACL_EDIT
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/rbac/AccessContext.java

@@ -1,5 +1,6 @@
 package com.provectus.kafka.ui.model.rbac;
 
+import com.provectus.kafka.ui.model.rbac.permission.AclAction;
 import com.provectus.kafka.ui.model.rbac.permission.ApplicationConfigAction;
 import com.provectus.kafka.ui.model.rbac.permission.ClusterConfigAction;
 import com.provectus.kafka.ui.model.rbac.permission.ConnectAction;
@@ -37,6 +38,8 @@ public class AccessContext {
 
   Collection<KsqlAction> ksqlActions;
 
+  Collection<AclAction> aclActions;
+
   public static AccessContextBuilder builder() {
     return new AccessContextBuilder();
   }
@@ -55,6 +58,7 @@ public static final class AccessContextBuilder {
     private String schema;
     private Collection<SchemaAction> schemaActions = Collections.emptySet();
     private Collection<KsqlAction> ksqlActions = Collections.emptySet();
+    private Collection<AclAction> aclActions = Collections.emptySet();
 
     private AccessContextBuilder() {
     }
@@ -131,6 +135,12 @@ public AccessContextBuilder ksqlActions(KsqlAction... actions) {
       return this;
     }
 
+    public AccessContextBuilder aclActions(AclAction... actions) {
+      Assert.isTrue(actions.length > 0, "actions not present");
+      this.aclActions = List.of(actions);
+      return this;
+    }
+
     public AccessContext build() {
       return new AccessContext(
           applicationConfigActions,
@@ -140,7 +150,7 @@ public AccessContext build() {
           connect, connectActions,
           connector,
           schema, schemaActions,
-          ksqlActions);
+          ksqlActions, aclActions);
     }
   }
 }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/rbac/Permission.java

@@ -3,6 +3,7 @@
 import static com.provectus.kafka.ui.model.rbac.Resource.CLUSTERCONFIG;
 import static com.provectus.kafka.ui.model.rbac.Resource.KSQL;
 
+import com.provectus.kafka.ui.model.rbac.permission.AclAction;
 import com.provectus.kafka.ui.model.rbac.permission.ApplicationConfigAction;
 import com.provectus.kafka.ui.model.rbac.permission.ClusterConfigAction;
 import com.provectus.kafka.ui.model.rbac.permission.ConnectAction;
@@ -73,6 +74,7 @@ private List<String> getAllActionValues() {
       case SCHEMA -> Arrays.stream(SchemaAction.values()).map(Enum::toString).toList();
       case CONNECT -> Arrays.stream(ConnectAction.values()).map(Enum::toString).toList();
       case KSQL -> Arrays.stream(KsqlAction.values()).map(Enum::toString).toList();
+      case ACL -> Arrays.stream(AclAction.values()).map(Enum::toString).toList();
     };
   }
 

kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/rbac/Resource.java

@@ -11,7 +11,8 @@ public enum Resource {
   CONSUMER,
   SCHEMA,
   CONNECT,
-  KSQL;
+  KSQL,
+  ACL;
 
   @Nullable
   public static Resource fromString(String name) {

kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/rbac/permission/AclAction.java

@@ -0,0 +1,15 @@
+package com.provectus.kafka.ui.model.rbac.permission;
+
+import org.apache.commons.lang3.EnumUtils;
+import org.jetbrains.annotations.Nullable;
+
+public enum AclAction implements PermissibleAction {
+
+  VIEW,
+  EDIT;
+
+  @Nullable
+  public static AclAction fromString(String name) {
+    return EnumUtils.getEnum(AclAction.class, name);
+  }
+}