基于Python Scapy与PostgreSQL的网络扫描设备数据自动入库求助

1. 依赖安装 #

首先安装所需的Python库：

pip install scapy psycopg2-binary mac-vendor-lookup

• scapy：用于网络设备扫描和数据包捕获
• psycopg2-binary：PostgreSQL的Python驱动，负责数据库交互
• mac-vendor-lookup：可选，用于通过MAC地址查询设备厂商

2. PostgreSQL数据库准备 #

先创建数据库和存储设备数据的表，执行以下SQL语句：

-- 创建数据库（如果不存在）
CREATE DATABASE network_scans;

-- 切换到目标数据库
\c network_scans;

-- 创建设备扫描表
CREATE TABLE device_scans (
    id SERIAL PRIMARY KEY,
    ip_address VARCHAR(15) NOT NULL,
    mac_address VARCHAR(17) NOT NULL UNIQUE,
    vendor VARCHAR(100),
    scan_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);

设置mac_address为唯一约束，避免重复插入同一设备的重复数据。

3. 核心实现代码 #

以下代码支持两种模式：单次扫描网络并入库，以及持续监听ARP包实时捕获新设备入库。

import scapy.all as scapy
import psycopg2
from psycopg2 import OperationalError
from datetime import datetime
from mac_vendor_lookup import MacLookup

def create_db_connection(db_name, db_user, db_password, db_host, db_port):
    """创建并返回PostgreSQL数据库连接"""
    connection = None
    try:
        connection = psycopg2.connect(
            database=db_name,
            user=db_user,
            password=db_password,
            host=db_host,
            port=db_port,
        )
    except OperationalError as e:
        print(f"数据库连接失败: {e}")
    return connection

def insert_device_data(connection, ip, mac, vendor=None):
    """将设备数据插入数据库，自动跳过已存在的设备"""
    if not connection:
        return
    cursor = connection.cursor()
    try:
        # 检查设备是否已存在
        check_query = "SELECT id FROM device_scans WHERE mac_address = %s"
        cursor.execute(check_query, (mac,))
        if cursor.fetchone():
            print(f"设备已存在，跳过: MAC={mac}, IP={ip}")
            return
        
        # 插入新设备数据
        insert_query = """
        INSERT INTO device_scans (ip_address, mac_address, vendor)
        VALUES (%s, %s, %s)
        """
        cursor.execute(insert_query, (ip, mac, vendor))
        connection.commit()
        print(f"成功入库设备: IP={ip}, MAC={mac}, 厂商={vendor or 'Unknown'}")
    except OperationalError as e:
        print(f"数据插入失败: {e}")
        connection.rollback()
    finally:
        cursor.close()

def scan_network(ip_range):
    """单次扫描指定IP范围，返回设备列表"""
    arp_request = scapy.ARP(pdst=ip_range)
    broadcast = scapy.Ether(dst="ff:ff:ff:ff:ff:ff")
    arp_broadcast = broadcast / arp_request
    answered = scapy.srp(arp_broadcast, timeout=1, verbose=False)[0]
    
    devices = []
    for pkt in answered:
        ip = pkt[1].psrc
        mac = pkt[1].hwsrc
        try:
            vendor = MacLookup().lookup(mac)
        except:
            vendor = None
        devices.append({"ip": ip, "mac": mac, "vendor": vendor})
    return devices

def monitor_arp_traffic(connection):
    """持续监听网络ARP包，实时捕获新设备并入库"""
    def process_packet(packet):
        # 只处理ARP响应包（op=2）
        if packet.haslayer(scapy.ARP) and packet[scapy.ARP].op == 2:
            ip = packet[scapy.ARP].psrc
            mac = packet[scapy.ARP].hwsrc
            try:
                vendor = MacLookup().lookup(mac)
            except:
                vendor = None
            insert_device_data(connection, ip, mac, vendor)
    
    print("开始监听网络ARP流量...")
    scapy.sniff(prn=process_packet, filter="arp", store=0)

if __name__ == "__main__":
    # 替换为你的PostgreSQL配置
    DB_SETTINGS = {
        "db_name": "network_scans",
        "db_user": "postgres",
        "db_password": "your_password",
        "db_host": "localhost",
        "db_port": "5432"
    }
    # 替换为你的目标IP范围，比如192.168.0.0/24
    TARGET_IP_RANGE = "192.168.1.0/24"
    
    # 建立数据库连接
    db_conn = create_db_connection(**DB_SETTINGS)
    
    if db_conn:
        # 模式1：单次扫描并入库（取消注释启用）
        # scanned_devices = scan_network(TARGET_IP_RANGE)
        # for dev in scanned_devices:
        #     insert_device_data(db_conn, dev["ip"], dev["mac"], dev["vendor"])
        
        # 模式2：持续监听ARP流量实时入库（取消注释启用）
        monitor_arp_traffic(db_conn)
        
        db_conn.close()

4. 关键注意事项 #

• 运行脚本需要管理员/root权限，因为Scapy发送ARP包或监听网络需要系统权限
• 确保PostgreSQL服务正在运行，且配置的数据库用户有权限执行插入、查询操作
• 如果不需要MAC厂商信息，可以移除mac-vendor-lookup相关代码，减少依赖
• 可以根据需求扩展表结构，比如添加last_seen字段记录设备最后出现时间，或device_name字段存储设备名称
• 对于持续监听场景，建议添加数据库连接重连机制，避免因网络波动导致连接中断

内容的提问来源于stack exchange，提问作者Marcus Ref