GravityAgent（三）：用 Rust 构建 IPv6 扫描执行引擎

用 Rust 构建 IPv6 扫描执行引擎

上一篇讲完算法适配层，这一篇我们下探到第一层——由 Rust 编写的 gravity-scanner。它是整个平台的基础设施，负责批次生命周期和扫描数据落库。

为什么扫描引擎要用 Rust

扫描引擎的核心诉求是高并发原始包收发。它需要：

• 直接构造并发送 TCP SYN / ICMP / UDP 原始包；
• 在极短时间内向海量地址发包；
• 同时接收并验证响应；
• 把结果高效写回数据库。

Rust 在这里的优势很直接：零成本抽象、无 GC、内存安全，配合 Tokio 异步运行时和 pnet 原始套接字库，可以在保证安全的前提下把发包性能拉满。

双角色架构

gravity-scanner 有两个子命令：

启动 Master 时，默认会自动拉起一个内置 Agent，连接本地 Master：

// main.rs：Master 启动后自动拉起内置 Agent
if !no_agent {
    let agent_server = format!("http://127.0.0.1:{}", listener.local_addr().unwrap().port());
    tokio::spawn(async move {
        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
        engine::run_agent(agent_server, concurrency, source_ip).await;
    });
}

如果只想跑 HTTP 服务，可以加 --no-agent：

# 基本启动（内置 Agent 自动随 Master 启动）
sudo ./target/release/gravity-scanner master \
  --bind 0.0.0.0:3000 \
  --db-url postgres://gravity:gravity@127.0.0.1:5432/gravity_scanner

# 仅运行 Master HTTP 服务（不启动内置 Agent）
sudo ./target/release/gravity-scanner master --no-agent

# 手动指定出口 IP（避免多网卡自动检测错误）
sudo ./target/release/gravity-scanner master --source-ip 203.0.113.1

双层存储设计

这是整个扫描系统最重要的架构决策：元数据与结果分离。

在代码里，Database 结构同时持有 PostgreSQL 连接池和按批次缓存的 SQLite 连接池：

pub struct Database {
    master_pool: PgPool,
    batch_pools: DashMap<String, SqlitePool>,  // 按 batch_id 缓存
}

为什么要这么设计？

1. PostgreSQL 负责高并发调度。任务租约、结果暂存这些高频写入操作放在 PostgreSQL，天然支持并发和事务。
2. SQLite 负责结果归档。每个批次一个文件，扫描完可以直接打包、迁移、删除，不会让主库无限膨胀。
3. 结果导出简单。导出某个批次的结果，只要读对应的 SQLite 文件即可。

任务租约模型

分布式扫描的核心问题是：如何把海量目标分发给多个 Agent，并保证不丢不重。

gravity-scanner 采用租约（lease）模型：

• Agent 通过长轮询 GET /api/agent/task 拉取任务；
• 租约有效期 LEASE_SECONDS = 600（10 分钟）；
• 结果积压时施加背压：pending + leased result_spool >= 8192 时 sleep 200ms 重试；
• 优先租约 preferred_batches（Agent 上次服务的批次），减少跨批次抖动。

一次租约如何填充任务

lease_from_batch 会在同一个 lease_id 下依次尝试五种来源，直到填满 max_targets：

① 租单 IP 任务（离散目标）
② 续租过期 IPv4 子区间（start_ip 前移，插入新子区间行）
③ 续租过期 IPv6 区间
④ 新取 IPv4 区间（推进 next_ip 游标）
⑤ 新取 IPv6 区间

这个顺序的设计意图是：优先复用已分配但未完成的任务，避免重复扫描；再推进新地址空间。

IPv6 游标机制

IPv6 地址以 32 字符十六进制字符串存储起止地址（完整 128 位），任务拉取时用游标分批推进：

任务范围：[start_hex, end_hex]
游标初始 = start_hex
每次拉取返回游标所指向的地址段（最多 agent_pull_max_targets 个）
游标推进到下一位置，直到到达 end_hex 或哨兵值 "zzzz...zzzz"

用十六进制字符串做游标，避免了 128 位整数在数据库和语言间的精度问题。

无状态 Cookie 探测

为了避免维护大量连接状态，引擎采用无状态 Cookie 验证：

tcp_cookie = hash(COOKIE_SALT ‖ nonce ‖ ip) & 0xFFFF_FFFF

发送时：
  TCP SYN 的 Sequence Number = tcp_cookie

收到 SYN-ACK 时验证：
  received_ack_number == tcp_cookie.wrapping_add(1)  ✓

ICMP 的 identifier/sequence 字段、UDP 的源端口均从同一 cookie 派生，确保无状态验证的一致性。

这意味着引擎不需要为每个探测保存一个”待响应表”，只需要在收到包时重新计算 cookie 并比对，内存占用与在途探测数无关。

多种探测协议

Agent 对每个目标 IP 并发发送多种探测包，任意一种有响应即判定存活：

存活原因有优先级，save_results_batch 按 IP 聚合时会择优记录：

ssh_banner (4) > tcp_syn (3) > icmp (2) > udp (1) > 其他 (0)

择优规则是：latency 最小 → 同 latency 比 reason 优先级 → 再同比端口小。端口级观测全部保留在 service_observations 表，供结果导出使用。

多线程并发包发送

                  ┌──────────────────────────────┐
                  │    run_task_loop              │
                  │  (拉取任务 → 切分 chunk)      │
                  └──────────────┬───────────────┘
                                 │ 每 chunk 最多 8192 个目标
                  ┌──────────────▼───────────────┐
                  │    burst sender 线程池        │
                  │  (默认 = CPU 核心数，最多 32) │
                  │  IPv4 TCP/UDP/ICMP sender     │
                  │  IPv6 TCP/UDP/ICMP sender     │
                  └──────────────────────────────┘

• 每个 sender 线程对应独立的原始传输通道（pnet::transport_channel，16 MB 缓冲区）；
• 发送速率由 GRAVITY_BURST_SENDER_THREADS 控制；
• 发包后等待”结果稳定”（result_settle_idle_ms 内无新响应）再提交，避免丢失末尾响应。

自适应端口优先级

当启用 GRAVITY_ENABLE_APP_CONNECT_FALLBACK 时，引擎会基于历史响应率对每个组织（ASN）动态调整探测端口顺序：

fn score(&self, port: u16) -> f64 {
    let (hits, attempts) = ...; // 来自 OrgProbeStats
    (hits + 1.0) / (attempts + 3.0)  // Laplace 平滑
}

用 Laplace 平滑的好处是：即使某个端口还没有历史数据，也不会被”饿死”，而是以中性分数参与排序。随着数据积累，响应率高的端口会自然排到前面。

result_spool：异步缓冲写入

这是扫描系统里另一个关键设计。如果 Agent 每提交一条结果就直接写 SQLite，会造成严重的锁竞争（SQLite 是单写者模型）。

所以系统引入了双阶段写入：

Agent 提交结果
    │
    ▼
result_spool 表（PostgreSQL，高并发友好）
    │  每 500ms 轮询一次
    ▼
Result Worker（后台 Tokio 任务，N 个并行）
    │  批量租约 → ACK 任务 → 批量写入 SQLite → 释放租约
    ▼
batch_<id>.db / scan_results 表
    │
    ▼
reconcile_alive_targets（更新 batches 表 alive_targets 计数）

关键参数：

失败回滚时，spool 行会被置回 pending 并记录 last_error，带指数退避以抗 SQLite 锁。

HTTP API 设计

Master 暴露的 REST API 非常精简，核心路由只有 10 个：

.route("/", get(index_page))
.route("/api/batch", post(upload_batch).get(list_batches))
.route("/api/batch/:id/export", get(export_batch_results))
.route("/api/batch/:id/export/stream", get(stream_batch_results))
.route("/api/batch/:id/txt", get(download_batch_txt))
.route("/api/agent/task", get(pull_task))
.route("/api/agent/result", post(submit_result))
.route("/api/agent/heartbeat", post(heartbeat))
.route("/api/stats", get(get_stats))
.route("/api/stop", post(stop_all_tasks))

其中 /api/batch/:id/txt 特别有用——它直接导出批次的存活地址文本，正好是 Agent 做下一轮种子挖掘的输入格式。

认证与权限

所有需要认证的接口采用 HTTP Basic Auth，支持三类凭据：

管理员与普通用户的区别：

目标地址上传格式

POST /api/batch 接受混合格式，每行一条：

1.2.3.4                      # 单 IPv4 地址
2001:db8::1                  # 单 IPv6 地址
10.0.0.1-10.0.0.255          # IPv4 范围
2001:db8::1-2001:db8::ffff   # IPv6 范围
192.168.0.0/24               # IPv4 CIDR
2001:db8::/48                # IPv6 CIDR
# 注释行（以 # 开头）

支持的表单参数：

端口解析由 models.rs 中的 parse_scan_ports_spec 完成，支持去重、乱序区间和上限校验。

数据模型

批次（Batch）

pub struct Batch {
    pub id: String,              // UUID
    pub name: String,
    pub status: BatchStatus,     // Running | Completed | Stopped | Failed
    pub total_targets: i64,
    pub scanned_targets: i64,
    pub alive_targets: i64,
    pub created_at: NaiveDateTime,
    pub submitted_by: String,
    pub priority: i64,
    pub scan_ports: Option<String>,    // CSV 格式端口列表
    pub scan_icmp: bool,
    pub scan_parallel_probes: bool,
    pub pending_tasks: i64,
    pub leased_tasks: i64,
}

扫描结果（ScanResult）

pub struct ScanResult {
    pub id: i64,
    pub batch_id: String,
    pub agent_id: String,
    pub ip: String,
    pub port: u16,
    pub latency_ms: i64,
    pub alive_reason: String,    // "tcp_syn" | "icmp" | "udp" | "ssh_banner"
    pub org: String,             // 组织归属（启用 org_lookup 时填充）
    pub created_at: NaiveDateTime,
}

PostgreSQL 主库核心表

-- 批次元数据
batches (id, name, status, total_targets, scanned_targets, alive_targets,
         created_at, submitted_by, priority, scan_ports, scan_icmp, scan_parallel_probes)

-- 任务调度（IPv4 范围分段）
scan_tasks (...)

-- IPv4/IPv6 区间游标
scan_task_ranges (next_ip ...)
scan_task_ipv6_ranges (next_ip ...)

-- 区间租约
scan_range_leases (...)
scan_ipv6_range_leases (...)

-- 结果写入缓冲
result_spool (id, batch_id, agent_id, payload_json, lease_owner, lease_until, status)

Agent 心跳

Agent 会定期上报运行状态：

pub struct AgentHeartbeat {
    pub agent_id: String,
    pub hostname: String,
    pub public_ip: String,
    pub os: String,
    pub cpu_usage: f32,
    pub ram_usage: f32,
    pub pps: u64,          // 当前每秒发包数
    pub last_seen: i64,    // Unix 时间戳
}

Master 通过 GET /api/stats 汇总在线 Agent 数量、批次统计和结果队列水位，方便运维监控。

小结

gravity-scanner 的设计可以总结为几个关键词：

到这里，底层的”怎么扫”就讲清楚了。下一篇，我们回到 Python 侧，看看 os_identify 子系统是如何用原始套接字和 nmap 指纹库做操作系统识别的。