TelpoKafkaConsole/TelpoKafkaConsole.WebApi/Controllers/ScramAclsController.cs

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/<ScramAclsController>/Consumer
        [HttpPost("Consumer")] // 添加了路由
        public async Task<ApiResponse<string>> Consumer([FromBody] ScramAclsConsumerReq consumer)
        {

            // 创建 SASL 用户
            ScramCredentialsUser scramUser = new ScramCredentialsUser
            {
                Name = consumer.Name,
                Password = consumer.Password
            };
            await _servicekafkaAdmin.AlterUserScramCredentialsAsync(scramUser);

            // 检查并创建主题（如果不存在）
            var topics = await _servicekafkaAdmin.DescribeTopicsAsync(new List<string> { consumer.Topic });
            if (topics.Count == 0)
            {
                await _servicekafkaAdmin.CreateTopic(consumer.Topic, TimeSpan.FromDays(3), Math.Max(3, consumer.NumPartitions));
            }

            // 创建 ACLs
            List<AclBinding> aclBindings = new List<AclBinding>
            {
                // 允许用户读取特定主题
                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
                    }
                },
                // 允许用户读取特定消费组
                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.Group,
                        Name = consumer.Group,
                        ResourcePatternType = ResourcePatternType.Literal
                    },
                    Entry = new AccessControlEntry
                    {
                        Principal = $"User:{consumer.Name}",
                        Host = "*",
                        Operation = AclOperation.All,
                        PermissionType = AclPermissionType.Deny
                    }
                }
            };

            // 创建 ACLs
            await _servicekafkaAdmin.CreateAclsAsync(aclBindings);

            return ApiResponse<string>.Success($"创建消费者用户 {consumer.Name} 和 ACLs 规则成功");
            
          
        }

        /**public async Task<ApiResponse<string>> 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<string> { consumer.Topic });
            if (topics.Count.Equals(0))
            {
                await _servicekafkaAdmin.CreateTopic(consumer.Topic, TimeSpan.FromDays(3), consumer.NumPartitions>3 ? 3: consumer.NumPartitions);
            }
           
            // 创建 alcs
            List<AclBinding> 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.Broker,
            //            Name = "kafka-cluster",
            //            ResourcePatternType = ResourcePatternType.Literal
            //        },
            //        Entry = new AccessControlEntry
            //        {
            //            Principal = $"User:superuser",
            //            Host = "*",
            //            Operation = AclOperation.All,
            //            PermissionType = AclPermissionType.Allow
            //        }
            //},
            //// 禁止查看
            //  new AclBinding()
            //    {
            //        Pattern = new ResourcePattern
            //        {
            //            Type = ResourceType.Broker,
            //            Name = "kafka-cluster",
            //            ResourcePatternType = ResourcePatternType.Literal
            //        },
            //        Entry = new AccessControlEntry
            //        {
            //            Principal = $"User:{consumer.Name}",
            //            Host = "*",
            //            Operation = AclOperation.Describe,
            //            PermissionType = AclPermissionType.Deny
            //        }
            //    },
            //  // 禁止修改
            //   new AclBinding()
            //    {
            //        Pattern = new ResourcePattern
            //        {
            //            Type = ResourceType.Broker,
            //            Name = "kafka-cluster",
            //            ResourcePatternType = ResourcePatternType.Literal
            //        },
            //        Entry = new AccessControlEntry
            //        {
            //            Principal = $"User:{consumer.Name}",
            //            Host = "*",
            //            Operation = AclOperation.Alter,
            //            PermissionType = AclPermissionType.Deny
            //        }
            //    },
            //// 禁止写入
            //new AclBinding()
            //    {
            //        Pattern = new ResourcePattern
            //        {
            //            Type = ResourceType.Broker,
            //            Name = "kafka-cluster",
            //            ResourcePatternType = ResourcePatternType.Literal
            //        },
            //        Entry = new AccessControlEntry
            //        {
            //            Principal = $"User:{consumer.Name}",
            //            Host = "*",
            //            Operation = AclOperation.Write,
            //            PermissionType = AclPermissionType.Deny
            //        }
            //    },
            //// 禁止创建
            //new AclBinding()
            //    {
            //        Pattern = new ResourcePattern
            //        {
            //            Type = ResourceType.Broker,
            //            Name = "kafka-cluster",
            //            ResourcePatternType = ResourcePatternType.Literal
            //        },
            //        Entry = new AccessControlEntry
            //        {
            //            Principal = $"User:{consumer.Name}",
            //            Host = "*",
            //            Operation = AclOperation.Create,
            //            PermissionType = AclPermissionType.Deny
            //        }
            //    },
            //new AclBinding()
            //    {
            //        Pattern = new ResourcePattern
            //        {
            //            Type = ResourceType.Broker,
            //            Name = "kafka-cluster",
            //            ResourcePatternType = ResourcePatternType.Literal
            //        },
            //        Entry = new AccessControlEntry
            //        {
            //            Principal = $"User:{consumer.Name}",
            //            Host = "*",
            //            Operation = AclOperation.Unknown,
            //            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
                    }
                },
                new AclBinding()
                {
                    Pattern = new ResourcePattern
                    {
                        Type = ResourceType.Any,
                        Name = consumer.Group,
                        ResourcePatternType = ResourcePatternType.Literal
                    },
                    Entry = new AccessControlEntry
                    {
                        Principal = $"User:{consumer.Name}",
                        Host = "*",
                        Operation = AclOperation.Write,
                        PermissionType = AclPermissionType.Deny
                    }
                },
                new AclBinding()
                {
                    Pattern = new ResourcePattern
                    {
                        Type = ResourceType.Topic,
                        Name = consumer.Group,
                        ResourcePatternType = ResourcePatternType.Literal
                    },
                    Entry = new AccessControlEntry
                    {
                        Principal = $"User:{consumer.Name}",
                        Host = "*",
                        Operation = AclOperation.Write,
                        PermissionType = AclPermissionType.Deny
                    }
                },
                new AclBinding()
                {
                    Pattern = new ResourcePattern
                    {
                        Type = ResourceType.Topic,
                        Name = consumer.Group,
                        ResourcePatternType = ResourcePatternType.Literal
                    },
                    Entry = new AccessControlEntry
                    {
                        Principal = $"User:{consumer.Name}",
                        Host = "*",
                        Operation = AclOperation.Alter,
                        PermissionType = AclPermissionType.Deny
                    }
                },
                new AclBinding()
                {
                    Pattern = new ResourcePattern
                    {
                        Type = ResourceType.Topic,
                        Name = consumer.Group,
                        ResourcePatternType = ResourcePatternType.Literal
                    },
                    Entry = new AccessControlEntry
                    {
                        Principal = $"User:{consumer.Name}",
                        Host = "*",
                        Operation = AclOperation.Describe,
                        PermissionType = AclPermissionType.Deny
                    }
                },
                new AclBinding()
                {
                    Pattern = new ResourcePattern
                    {
                        Type = ResourceType.Topic,
                        Name = consumer.Group,
                        ResourcePatternType = ResourcePatternType.Literal
                    },
                    Entry = new AccessControlEntry
                    {
                        Principal = $"User:{consumer.Name}",
                        Host = "*",
                        Operation = AclOperation.Delete,
                        PermissionType = AclPermissionType.Deny
                    }
                }
            };
            await _servicekafkaAdmin.CreateAclsAsync(aclBindings);


            return ApiResponse<string>.Success($"创建 消费者用户 {consumer.Name} Acls 规则成功");
        }
        */

        // POST api/<ScramAclsController>/Producer
        [HttpPost("Producer")] // 添加了路由
        public async Task<ApiResponse<string>> 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<string> { producer.Topic });
            if (topics.Count.Equals(0))
            {
                await _servicekafkaAdmin.CreateTopic(producer.Topic, TimeSpan.FromDays(3), producer.NumPartitions);
            }
            // 创建 alcs
            List<AclBinding> 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<string>.Success($"创建 生产者用户 {producer.Name} Acls 规则成功");
        }

        // DELETE api/<ScramAclsController>/{username}
        [HttpDelete("{username}")]
        public async Task<ApiResponse<string>> Delete(string username)
        {
            // 删除用户
            var scramUsers = await _servicekafkaAdmin.DescribeUserScramCredentialsAsync(new List<string>
            {
                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<string>.Success($"删除用户 {username} 和 Acls 规则成功");

        }
    }
}