diff --git a/Plugins/services/kafka/connector.go b/Plugins/services/kafka/connector.go
new file mode 100644
index 0000000..6229d73
--- /dev/null
+++ b/Plugins/services/kafka/connector.go
@@ -0,0 +1,114 @@
+package kafka
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/IBM/sarama"
+	"github.com/shadow1ng/fscan/common"
+	"github.com/shadow1ng/fscan/plugins/base"
+)
+
+// KafkaConnection Kafka连接包装器
+type KafkaConnection struct {
+	client sarama.Client
+	target string
+}
+
+// KafkaConnector Kafka连接器实现
+type KafkaConnector struct {
+	host string
+	port string
+}
+
+// NewKafkaConnector 创建Kafka连接器
+func NewKafkaConnector() *KafkaConnector {
+	return &KafkaConnector{}
+}
+
+// Connect 连接到Kafka服务
+func (c *KafkaConnector) Connect(ctx context.Context, info *common.HostInfo) (interface{}, error) {
+	c.host = info.Host
+	c.port = info.Ports
+	
+	target := fmt.Sprintf("%s:%s", c.host, c.port)
+	
+	// 返回连接信息，实际连接在Authenticate时建立
+	return &KafkaConnection{
+		client: nil, // 延迟连接
+		target: target,
+	}, nil
+}
+
+// Authenticate 认证
+func (c *KafkaConnector) Authenticate(ctx context.Context, conn interface{}, cred *base.Credential) error {
+	kafkaConn, ok := conn.(*KafkaConnection)
+	if !ok {
+		return fmt.Errorf("无效的连接类型")
+	}
+	
+	// 关闭之前的连接（如果有）
+	if kafkaConn.client != nil {
+		kafkaConn.client.Close()
+	}
+	
+	// 创建新的认证配置
+	timeout := time.Duration(common.Timeout) * time.Second
+	config := sarama.NewConfig()
+	config.Net.DialTimeout = timeout
+	config.Net.ReadTimeout = timeout
+	config.Net.WriteTimeout = timeout
+	config.Net.TLS.Enable = false
+	config.Version = sarama.V2_0_0_0
+
+	// 如果提供了用户名密码，设置SASL认证
+	if cred.Username != "" || cred.Password != "" {
+		config.Net.SASL.Enable = true
+		config.Net.SASL.Mechanism = sarama.SASLTypePlaintext
+		config.Net.SASL.User = cred.Username
+		config.Net.SASL.Password = cred.Password
+		config.Net.SASL.Handshake = true
+	}
+
+	brokers := []string{kafkaConn.target}
+
+	// 尝试作为消费者连接测试
+	consumer, err := sarama.NewConsumer(brokers, config)
+	if err == nil {
+		consumer.Close()
+		
+		// 创建认证后的客户端
+		client, clientErr := sarama.NewClient(brokers, config)
+		if clientErr != nil {
+			return fmt.Errorf("创建认证客户端失败: %v", clientErr)
+		}
+		kafkaConn.client = client
+		return nil
+	}
+
+	// 如果消费者连接失败，尝试作为客户端连接
+	client, clientErr := sarama.NewClient(brokers, config)
+	if clientErr == nil {
+		kafkaConn.client = client
+		return nil
+	}
+
+	// 检查认证相关错误
+	if strings.Contains(err.Error(), "SASL") ||
+		strings.Contains(err.Error(), "authentication") ||
+		strings.Contains(err.Error(), "credentials") {
+		return fmt.Errorf("Kafka认证失败")
+	}
+
+	return fmt.Errorf("Kafka连接失败: %v", err)
+}
+
+// Close 关闭连接
+func (c *KafkaConnector) Close(conn interface{}) error {
+	if kafkaConn, ok := conn.(*KafkaConnection); ok && kafkaConn.client != nil {
+		return kafkaConn.client.Close()
+	}
+	return nil
+}
\ No newline at end of file
diff --git a/Plugins/services/kafka/exploiter.go b/Plugins/services/kafka/exploiter.go
new file mode 100644
index 0000000..911ab71
--- /dev/null
+++ b/Plugins/services/kafka/exploiter.go
@@ -0,0 +1,36 @@
+package kafka
+
+import (
+	"context"
+
+	"github.com/shadow1ng/fscan/common"
+	"github.com/shadow1ng/fscan/plugins/base"
+)
+
+// KafkaExploiter Kafka利用器实现 - 最小化版本，不提供利用功能
+type KafkaExploiter struct {
+	*base.BaseExploiter
+}
+
+// NewKafkaExploiter 创建Kafka利用器
+func NewKafkaExploiter() *KafkaExploiter {
+	exploiter := &KafkaExploiter{
+		BaseExploiter: base.NewBaseExploiter("kafka"),
+	}
+	
+	// Kafka插件不提供利用功能
+	exploiter.setupExploitMethods()
+	
+	return exploiter
+}
+
+// setupExploitMethods 设置利用方法
+func (e *KafkaExploiter) setupExploitMethods() {
+	// Kafka插件不提供利用功能，仅进行弱密码扫描和未授权访问检测
+}
+
+// Exploit 利用接口实现 - 空实现
+func (e *KafkaExploiter) Exploit(ctx context.Context, info *common.HostInfo, creds *base.Credential) (*base.ExploitResult, error) {
+	// Kafka插件不提供利用功能
+	return nil, nil
+}
\ No newline at end of file
diff --git a/Plugins/services/kafka/plugin.go b/Plugins/services/kafka/plugin.go
new file mode 100644
index 0000000..192b83c
--- /dev/null
+++ b/Plugins/services/kafka/plugin.go
@@ -0,0 +1,216 @@
+package kafka
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/IBM/sarama"
+	"github.com/shadow1ng/fscan/common"
+	"github.com/shadow1ng/fscan/common/i18n"
+	"github.com/shadow1ng/fscan/plugins/base"
+)
+
+// KafkaPlugin Kafka插件实现
+type KafkaPlugin struct {
+	*base.ServicePlugin
+	exploiter *KafkaExploiter
+}
+
+// NewKafkaPlugin 创建Kafka插件
+func NewKafkaPlugin() *KafkaPlugin {
+	// 插件元数据
+	metadata := &base.PluginMetadata{
+		Name:        "kafka",
+		Version:     "2.0.0",
+		Author:      "fscan-team",
+		Description: "Apache Kafka消息队列扫描和利用插件",
+		Category:    "service",
+		Ports:       []int{9092, 9093, 9094}, // Kafka常用端口
+		Protocols:   []string{"tcp"},
+		Tags:        []string{"kafka", "message-queue", "bruteforce", "unauthorized"},
+	}
+	
+	// 创建连接器和服务插件
+	connector := NewKafkaConnector()
+	servicePlugin := base.NewServicePlugin(metadata, connector)
+	
+	// 创建Kafka插件
+	plugin := &KafkaPlugin{
+		ServicePlugin: servicePlugin,
+		exploiter:     NewKafkaExploiter(),
+	}
+	
+	// 设置能力
+	plugin.SetCapabilities([]base.Capability{
+		base.CapWeakPassword,
+		base.CapDataExtraction,
+	})
+	
+	return plugin
+}
+
+// Scan 重写扫描方法，先检测无认证访问
+func (p *KafkaPlugin) Scan(ctx context.Context, info *common.HostInfo) (*base.ScanResult, error) {
+	// 如果禁用了暴力破解，只进行服务识别
+	if common.DisableBrute {
+		return p.performServiceIdentification(ctx, info)
+	}
+	
+	target := fmt.Sprintf("%s:%s", info.Host, info.Ports)
+	
+	// 先尝试无认证访问
+	unauthCred := &base.Credential{Username: "", Password: ""}
+	unauthResult, err := p.ScanCredential(ctx, info, unauthCred)
+	if err == nil && unauthResult.Success {
+		// 无认证访问成功
+		common.LogSuccess(i18n.GetText("kafka_unauth_access", target))
+		
+		return &base.ScanResult{
+			Success:     true,
+			Service:     "Kafka",
+			Credentials: []*base.Credential{unauthCred},
+			Extra: map[string]interface{}{
+				"service":       "Kafka",
+				"port":          info.Ports,
+				"unauthorized":  true,
+				"access_type":   "no_authentication",
+			},
+		}, nil
+	}
+	
+	// 执行基础的密码扫描
+	result, err := p.ServicePlugin.Scan(ctx, info)
+	if err != nil || !result.Success {
+		return result, err
+	}
+	
+	// 记录成功的弱密码发现
+	cred := result.Credentials[0]
+	common.LogSuccess(i18n.GetText("kafka_weak_pwd_success", target, cred.Username, cred.Password))
+	
+	return result, nil
+}
+
+// generateCredentials 重写凭据生成方法
+func (p *KafkaPlugin) generateCredentials() []*base.Credential {
+	// 获取Kafka专用的用户名字典
+	usernames := common.Userdict["kafka"]
+	if len(usernames) == 0 {
+		// 默认Kafka用户名
+		usernames = []string{"admin", "kafka", "test", "user", "root"}
+	}
+	
+	return base.GenerateCredentials(usernames, common.Passwords)
+}
+
+// Exploit 使用exploiter执行利用
+func (p *KafkaPlugin) Exploit(ctx context.Context, info *common.HostInfo, creds *base.Credential) (*base.ExploitResult, error) {
+	return p.exploiter.Exploit(ctx, info, creds)
+}
+
+// GetExploitMethods 获取利用方法
+func (p *KafkaPlugin) GetExploitMethods() []base.ExploitMethod {
+	return p.exploiter.GetExploitMethods()
+}
+
+// IsExploitSupported 检查利用支持
+func (p *KafkaPlugin) IsExploitSupported(method base.ExploitType) bool {
+	return p.exploiter.IsExploitSupported(method)
+}
+
+// performServiceIdentification 执行Kafka服务识别（-nobr模式）
+func (p *KafkaPlugin) performServiceIdentification(ctx context.Context, info *common.HostInfo) (*base.ScanResult, error) {
+	target := fmt.Sprintf("%s:%s", info.Host, info.Ports)
+	
+	// 尝试连接Kafka获取版本信息
+	kafkaInfo, isKafka := p.identifyKafkaService(ctx, info)
+	if isKafka {
+		// 记录服务识别成功
+		common.LogSuccess(i18n.GetText("kafka_service_identified", target, kafkaInfo))
+		
+		return &base.ScanResult{
+			Success: true,
+			Service: "Kafka",
+			Banner:  kafkaInfo,
+			Extra: map[string]interface{}{
+				"service": "Kafka",
+				"port":    info.Ports,
+				"info":    kafkaInfo,
+			},
+		}, nil
+	}
+	
+	// 如果无法识别为Kafka，返回失败
+	return &base.ScanResult{
+		Success: false,
+		Error:   fmt.Errorf("无法识别为Kafka服务"),
+	}, nil
+}
+
+// identifyKafkaService 通过连接识别Kafka服务
+func (p *KafkaPlugin) identifyKafkaService(ctx context.Context, info *common.HostInfo) (string, bool) {
+	target := fmt.Sprintf("%s:%s", info.Host, info.Ports)
+	timeout := time.Duration(common.Timeout) * time.Second
+
+	config := sarama.NewConfig()
+	config.Net.DialTimeout = timeout
+	config.Net.ReadTimeout = timeout
+	config.Net.WriteTimeout = timeout
+	config.Net.TLS.Enable = false
+	config.Version = sarama.V2_0_0_0
+
+	brokers := []string{target}
+	
+	// 尝试创建客户端连接
+	client, err := sarama.NewClient(brokers, config)
+	if err != nil {
+		// 检查错误是否表明这是Kafka服务但认证失败
+		if strings.Contains(strings.ToLower(err.Error()), "kafka") ||
+			strings.Contains(strings.ToLower(err.Error()), "sasl") ||
+			strings.Contains(strings.ToLower(err.Error()), "authentication") {
+			return fmt.Sprintf("Kafka服务 (需要认证): %v", err), true
+		}
+		return "", false
+	}
+	defer client.Close()
+	
+	// 获取集群信息
+	brokerList := client.Brokers()
+	if len(brokerList) > 0 {
+		return fmt.Sprintf("Kafka集群 (Brokers: %d)", len(brokerList)), true
+	}
+	
+	return "Kafka服务", true
+}
+
+// =============================================================================
+// 插件注册
+// =============================================================================
+
+// RegisterKafkaPlugin 注册Kafka插件
+func RegisterKafkaPlugin() {
+	factory := base.NewSimplePluginFactory(
+		&base.PluginMetadata{
+			Name:        "kafka",
+			Version:     "2.0.0",
+			Author:      "fscan-team",
+			Description: "Apache Kafka消息队列扫描和利用插件",
+			Category:    "service",
+			Ports:       []int{9092, 9093, 9094}, // Kafka常用端口
+			Protocols:   []string{"tcp"},
+			Tags:        []string{"kafka", "message-queue", "bruteforce", "unauthorized"},
+		},
+		func() base.Plugin {
+			return NewKafkaPlugin()
+		},
+	)
+	
+	base.GlobalPluginRegistry.Register("kafka", factory)
+}
+
+// 自动注册
+func init() {
+	RegisterKafkaPlugin()
+}
\ No newline at end of file