using Confluent.Kafka.Admin; using Microsoft.AspNetCore.Mvc; using TelpoKafkaConsole.Model; using TelpoKafkaConsole.Service; using TelpoKafkaConsole.WebApi.Controllers.Api; using TelpoKafkaConsole.WebApi.Model.Request; using static Confluent.Kafka.ConfigPropertyNames; namespace TelpoKafkaConsole.WebApi.Controllers { [Route("api/[controller]")] [ApiController] public class ScramAclsController : ControllerBase { private readonly KafkaAdminService _servicekafkaAdmin; public ScramAclsController(KafkaAdminService kafkaAdminService) { _servicekafkaAdmin = kafkaAdminService; } // POST api//Consumer [HttpPost("Consumer")] // 添加了路由 public async Task> Consumer([FromBody] ScramAclsConsumerReq consumer) { // 创建用户 ScramCredentialsUser scramUser = new() { Name = consumer.Name, Password = consumer.Password, }; await _servicekafkaAdmin.AlterUserScramCredentialsAsync(scramUser); // 创建 topic var topics = await _servicekafkaAdmin.DescribeTopicsAsync(new List { consumer.Topic }); if (topics.Count.Equals(0)) { await _servicekafkaAdmin.CreateTopic(consumer.Topic, TimeSpan.FromDays(3), consumer.NumPartitions); } // 创建 alcs List aclBindings = new() { new AclBinding() { Pattern = new ResourcePattern { Type = ResourceType.Broker, Name = "kafka-cluster", ResourcePatternType = ResourcePatternType.Literal }, Entry = new AccessControlEntry { Principal = $"User:{consumer.Name}", Host = "*", Operation = AclOperation.All, PermissionType = AclPermissionType.Deny } }, new AclBinding() { Pattern = new ResourcePattern { Type = ResourceType.Group, Name = consumer.Group, ResourcePatternType = ResourcePatternType.Literal }, Entry = new AccessControlEntry { Principal = $"User:{consumer.Name}", Host = "*", Operation = AclOperation.Read, PermissionType = AclPermissionType.Allow } }, new AclBinding() { Pattern = new ResourcePattern { Type = ResourceType.Topic, Name = consumer.Topic, ResourcePatternType = ResourcePatternType.Literal }, Entry = new AccessControlEntry { Principal = $"User:{consumer.Name}", Host = "*", Operation = AclOperation.Read, PermissionType = AclPermissionType.Allow } } }; await _servicekafkaAdmin.CreateAclsAsync(aclBindings); return ApiResponse.Success($"创建 消费者用户 {consumer.Name} Acls 规则成功"); } // POST api//Producer [HttpPost("Producer")] // 添加了路由 public async Task> Producer([FromBody] ScramAclsProducerReq producer) { // 创建用户 ScramCredentialsUser scramUser = new() { Name = producer.Name, Password = producer.Password, }; await _servicekafkaAdmin.AlterUserScramCredentialsAsync(scramUser); // 创建 topic var topics = await _servicekafkaAdmin.DescribeTopicsAsync(new List { producer.Topic }); if (topics.Count.Equals(0)) { await _servicekafkaAdmin.CreateTopic(producer.Topic, TimeSpan.FromDays(3), producer.NumPartitions); } // 创建 alcs List aclBindings = new() { new AclBinding() { Pattern = new ResourcePattern { Type = ResourceType.Broker, Name = "kafka-cluster", ResourcePatternType = ResourcePatternType.Literal }, Entry = new AccessControlEntry { Principal = $"User:{producer.Name}", Host = "*", Operation = AclOperation.All, PermissionType = AclPermissionType.Deny } }, new AclBinding() { Pattern = new ResourcePattern { Type = ResourceType.Topic, Name = producer.Topic, ResourcePatternType = ResourcePatternType.Literal }, Entry = new AccessControlEntry { Principal = $"User:{producer.Name}", Host = "*", Operation = AclOperation.Write, PermissionType = AclPermissionType.Allow } } }; await _servicekafkaAdmin.CreateAclsAsync(aclBindings); return ApiResponse.Success($"创建 生产者用户 {producer.Name} Acls 规则成功"); } // DELETE api//{username} [HttpDelete("{username}")] public async Task> Delete(string username) { // 删除用户 var scramUsers = await _servicekafkaAdmin.DescribeUserScramCredentialsAsync(new List { username }); if (scramUsers.Count==1) { ScramCredentialsUser scramUser = new() { Name = username }; await _servicekafkaAdmin.AlterUserScramCredentialsAsync(scramUser, "DELETE"); } // 删除alcs var acls = await _servicekafkaAdmin.DescribeAclsAsync(); var userAclsBinding = acls.Where(i => i.Entry.Principal.EndsWith(username)).ToList(); if (userAclsBinding.Count>0) { await _servicekafkaAdmin.DeleteAclsAsync(userAclsBinding); } return ApiResponse.Success($"删除用户 {username} 和 Acls 规则成功"); } } }