当前位置：首页 > news >正文

Golang实现分布式Masscan任务调度系统

news 来源：原创 2025/6/13 6:27:24

1. 系统架构概述

1.1 核心组件

1.任务生成器（Task Generator）：

功能：生成 Masscan 扫描任务，如指定目标 IP 范围、端口、扫描参数等。
输出：将任务发送到 Kafka 的任务队列（Topic）。

2.任务调度器（Task Scheduler）：

功能：从 Kafka 任务队列中消费任务，并将任务分配给可用的 Worker 节点。
实现：可以使用 Kafka Consumer 消费任务，并使用 ZooKeeper 或其他服务发现机制来管理 Worker 节点。

3.Worker 节点（Worker Nodes）：

功能：执行 Masscan 扫描任务。
实现：每个 Worker 节点作为一个独立的进程或服务，从 Kafka 接收任务，执行 Masscan 扫描，并将结果发送回 Kafka 的结果队列。

4.结果处理器（Result Processor）：

功能：从 Kafka 的结果队列中消费扫描结果，进行处理、分析或存储。
实现：可以使用 Kafka Consumer 消费结果，并将其存储到数据库或进行实时分析。

5.数据库（Database）：

功能：存储扫描任务和结果。
选择：如 PostgreSQL、MongoDB、Elasticsearch 等。

6.监控与日志（Monitoring & Logging）：

功能：监控系统的运行状态，记录日志以便故障排查。
实现：使用 Prometheus、Grafana、ELK（Elasticsearch, Logstash, Kibana）等工具。

1.2 工作流程

1.任务生成：任务生成器生成 Masscan 扫描任务，并将其发送到 Kafka 的任务队列。

2.任务调度：任务调度器从 Kafka 任务队列中消费任务，并将任务分配给可用的 Worker 节点。

3.任务执行：Worker 节点接收任务，执行 Masscan 扫描。

4.结果处理：Worker 节点将扫描结果发送回 Kafka 的结果队列。

5.结果存储与分析：结果处理器从 Kafka 结果队列中消费结果，并将其存储到数据库或进行实时分析。

2. 关键组件实现

2.1 任务生成器

任务生成器负责生成 Masscan 扫描任务，并将其发送到 Kafka 的任务队列。

go

package mainimport ("context""fmt""github.com/segmentio/kafka-go""time"
)func main() {// 配置 Kafka 连接writer := kafka.NewWriter(kafka.WriterConfig{Brokers:  []string{"kafka-broker1:9092", "kafka-broker2:9092"},Topic:    "masscan_tasks",Balancer: &kafka.LeastBytes{},})// 生成任务tasks := generateTasks()// 发送任务到 Kafkafor _, task := range tasks {msg := kafka.Message{Value: []byte(task),}err := writer.WriteMessages(context.Background(), msg)if err != nil {fmt.Println("Error sending message:", err)}}// 关闭连接writer.Close()
}func generateTasks() []string {// 示例：生成简单的 Masscan 命令tasks := []string{`{"command": "masscan 192.168.1.0/24 -p80,443 --rate=1000"}`,`{"command": "masscan 10.0.0.0/16 -p22,8080 --rate=500"}`,}return tasks
}

2.2 任务调度器

任务调度器从 Kafka 任务队列中消费任务，并将任务分配给 Worker 节点。

go

package mainimport ("context""fmt""github.com/segmentio/kafka-go""time"
)func main() {// 配置 Kafka 连接reader := kafka.NewReader(kafka.ReaderConfig{Brokers:  []string{"kafka-broker1:9092", "kafka-broker2:9092"},Topic:    "masscan_tasks",GroupID:  "masscan_scheduler",MinBytes: 1,    // 1BMaxBytes: 10e6, // 10MB})// 消费任务for {msg, err := reader.ReadMessage(context.Background())if err != nil {fmt.Println("Error reading message:", err)continue}// 解析任务var task Taskerr = json.Unmarshal(msg.Value, &task)if err != nil {fmt.Println("Error parsing message:", err)continue}// 分配任务给 Worker（这里简单打印，实际应发送到 Worker 的 Kafka 队列）fmt.Printf("Received task: %s\n", task.Command)// TODO: 实现任务分配逻辑}reader.Close()
}type Task struct {Command string `json:"command"`
}

2.3 Worker 节点

Worker 节点从 Kafka 接收任务，执行 Masscan 扫描，并将结果发送回 Kafka。

go

package mainimport ("context""encoding/json""fmt""github.com/segmentio/kafka-go""os/exec""time"
)type Task struct {Command string `json:"command"`
}type Result struct {TaskID     string `json:"task_id"`Output     string `json:"output"`Error      string `json:"error"`Timestamp  time.Time `json:"timestamp"`
}func main() {// 配置 Kafka 连接reader := kafka.NewReader(kafka.ReaderConfig{Brokers:  []string{"kafka-broker1:9092", "kafka-broker2:9092"},Topic:    "masscan_tasks",GroupID:  "masscan_worker",MinBytes: 1,    // 1BMaxBytes: 10e6, // 10MB})writer := kafka.NewWriter(kafka.WriterConfig{Brokers:  []string{"kafka-broker1:9092", "kafka-broker2:9092"},Topic:    "masscan_results",Balancer: &kafka.LeastBytes{},})for {msg, err := reader.ReadMessage(context.Background())if err != nil {fmt.Println("Error reading message:", err)continue}var task Taskerr = json.Unmarshal(msg.Value, &task)if err != nil {fmt.Println("Error parsing message:", err)continue}// 执行 Masscan 命令cmd := exec.Command("sh", "-c", task.Command)output, err := cmd.CombinedOutput()result := Result{TaskID:    fmt.Sprintf("%d", time.Now().UnixNano()),Output:    string(output),Error:     "",Timestamp: time.Now(),}if err != nil {result.Error = err.Error()}// 发送结果到 KafkaresultBytes, _ := json.Marshal(result)err = writer.WriteMessages(context.Background(), kafka.Message{Value: resultBytes,})if err != nil {fmt.Println("Error sending result:", err)}}reader.Close()writer.Close()
}

2.4 结果处理器

结果处理器从 Kafka 结果队列中消费结果，并将其存储到数据库或进行实时分析。

go

package mainimport ("context""encoding/json""fmt""github.com/segmentio/kafka-go""time"
)type Result struct {TaskID     string    `json:"task_id"`Output     string    `json:"output"`Error      string    `json:"error"`Timestamp  time.Time `json:"timestamp"`
}func main() {// 配置 Kafka 连接reader := kafka.NewReader(kafka.ReaderConfig{Brokers:  []string{"kafka-broker1:9092", "kafka-broker2:9092"},Topic:    "masscan_results",GroupID:  "masscan_result_processor",MinBytes: 1,    // 1BMaxBytes: 10e6, // 10MB})// 处理结果for {msg, err := reader.ReadMessage(context.Background())if err != nil {fmt.Println("Error reading message:", err)continue}var result Resulterr = json.Unmarshal(msg.Value, &result)if err != nil {fmt.Println("Error parsing message:", err)continue}// 处理结果，例如存储到数据库storeResult(result)}reader.Close()
}func storeResult(result Result) {// 示例：打印结果，实际应存储到数据库fmt.Printf("Result: %+v\n", result)
}