4. 网络资产发现自动化

网络资产发现是安全自动化流水线的第一道工序。在漏洞扫描、弱口令检测、基线核查之前，必须先回答一个基础问题：目标网段里到底有哪些主机在线、开放了哪些端口。如果资产清单本身不准确，后续所有检测都会建立在错误的前提上。本节围绕主机存活探测的三种典型手段——ARP扫描、ICMP探测、TCP探测——展开，并给出把探测结果结构化落库的工程做法。

一、为什么不能只用一种探测方式

主机存活探测的本质是"发送一个探测包，等待一个响应"。不同协议层面对应的探测方式，适用场景和可靠性差异很大：

• ARP扫描：工作在链路层，向本网段广播ARP请求，任何在线主机都会回应自己的MAC地址。它只适用于同一广播域（同一VLAN/子网），但可靠性极高，几乎不会漏报，也无法被普通主机防火墙屏蔽。
• ICMP探测：即常见的ping，工作在网络层。跨网段探测时最常用，但很多主机或边界设备会禁用ICMP回显，导致"在线但ping不通"的漏报。
• TCP探测：向目标端口发起TCP连接（SYN或全连接），根据是否收到SYN-ACK或RST判断主机存活。即使ICMP被禁，只要目标有任意一个端口可达，就能确认存活，是跨网段探测中最可靠的补充手段。

工程实践中通常采用"分层组合"策略：同网段优先ARP，跨网段用ICMP加TCP多端口探测，三者结果取并集，最大限度降低漏报。

二、ARP扫描的实现要点

ARP扫描依赖原始套接字或第三方库。Python中常用 scapy 构造ARP请求并捕获应答。核心思路是：构造 ARP(pdst="192.168.1.0/24") 的广播包，用 srp() 在二层收发，遍历返回的应答对，提取IP与MAC。

from scapy.all import ARP, Ether, srp

def arp_scan(cidr, timeout=2):
    pkt = Ether(dst="ff:ff:ff:ff:ff:ff") / ARP(pdst=cidr)
    ans, _ = srp(pkt, timeout=timeout, verbose=False)
    hosts = []
    for _, rcv in ans:
        hosts.append({"ip": rcv.psrc, "mac": rcv.hwsrc, "method": "arp"})
    return hosts

# 示例：扫描 192.168.1.0/24，超时2秒
# 假设返回 [{'ip':'192.168.1.1','mac':'00:1a:2b:...','method':'arp'}, ...]

需要注意：srp() 需要管理员/root权限；timeout 决定等待应答的时长，局域网内一般1~2秒足够；扫描大网段时应分批，避免广播风暴。

三、ICMP与TCP探测

ICMP探测可以直接调用系统ping，也可以用scapy构造。用系统命令更简单，但解析输出较繁琐；用scapy则能精确控制超时和包数量。下面用scapy实现ICMP探测：

from scapy.all import IP, ICMP, sr1

def icmp_probe(ip, timeout=1):
    resp = sr1(IP(dst=ip)/ICMP(), timeout=timeout, verbose=False)
    if resp is not None:
        return {"ip": ip, "method": "icmp", "alive": True}
    return {"ip": ip, "method": "icmp", "alive": False}

TCP探测更实用。对每个目标IP，选择若干常见端口（如22、80、443、445、3389）发起连接。若端口开放，会收到SYN-ACK；若端口关闭但主机在线，会收到RST。两者都说明主机存活。用socket实现全连接探测：

import socket

def tcp_probe(ip, ports=(22, 80, 443, 445, 3389), timeout=1):
    for port in ports:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(timeout)
        try:
            s.connect((ip, port))
            return {"ip": ip, "method": "tcp", "alive": True, "port": port}
        except ConnectionRefusedError:
            return {"ip": ip, "method": "tcp", "alive": True, "port": port}
        except (socket.timeout, OSError):
            continue
        finally:
            s.close()
    return {"ip": ip, "method": "tcp", "alive": False}

这里的关键判断是：connect 成功说明端口开放且主机在线；抛出 ConnectionRefusedError 说明主机在线但端口关闭（收到了RST）；只有超时才说明可能不在线。很多初学者会忽略"拒绝连接也是存活信号"这一点，导致漏报。

四、结果结构化存储

三种探测方式返回的字段不完全一致，需要统一成一张资产表。建议字段包括：ip、mac（ARP可得，其他为空）、alive、methods（命中的探测方式列表）、open_ports、first_seen、last_seen。用SQLite存储便于后续查询和增量更新：

import sqlite3, json, datetime

def init_db(path="assets.db"):
    conn = sqlite3.connect(path)
    conn.execute("""CREATE TABLE IF NOT EXISTS hosts(
        ip TEXT PRIMARY KEY,
        mac TEXT,
        alive INTEGER,
        methods TEXT,
        open_ports TEXT,
        first_seen TEXT,
        last_seen TEXT)""")
    return conn

def upsert_host(conn, rec):
    now = datetime.datetime.now().isoformat(timespec="seconds")
    cur = conn.execute("SELECT methods, open_ports FROM hosts WHERE ip=?", (rec["ip"],))
    row = cur.fetchone()
    if row:
        methods = set(json.loads(row[0])) | {rec["method"]}
        ports = set(json.loads(row[1])) | set(rec.get("open_ports", []))
        conn.execute("UPDATE hosts SET alive=1, methods=?, open_ports=?, last_seen=? WHERE ip=?",
                     (json.dumps(sorted(methods)), json.dumps(sorted(ports)), now, rec["ip"]))
    else:
        conn.execute("INSERT INTO hosts VALUES(?,?,?,?,?,?,?)",
                     (rec["ip"], rec.get("mac"), 1, json.dumps([rec["method"]]),
                      json.dumps(rec.get("open_ports", [])), now, now))
    conn.commit()

这样，同一IP被多种方式命中时，methods 会累积为 ["arp","icmp","tcp"]，open_ports 也会合并，形成一份可追溯、可增量更新的资产清单。后续漏洞扫描只需读取 alive=1 的记录即可。

五、组合策略与判断依据小结

把三种方式串起来，一个实用的探测流程是：先判断目标是否与探测机同网段，同网段走ARP；跨网段先ICMP，ICMP无响应再对常见端口做TCP探测。任一方式确认存活即标记 alive=1，并记录命中的方式。判断依据可以归纳为：ARP有应答、ICMP有Echo Reply、TCP有SYN-ACK或RST，三者满足其一即判定在线。最终结果统一写入SQLite，并保留JSON快照用于变更对比，就完成了从"探测"到"可用资产数据"的闭环。