Python结合Nmap与MySQL实现自动化端口扫描系统

叶初

发布时间：2026-10-12| 1人浏览过

原创

本文详细介绍基于Python调用Nmap命令行工具，结合标准库解析扫描结果并将其持久化至MySQL数据库的完整流程。方案涵盖从接口获取IP网段、发起端口探测、过滤开放服务到数据入库的全链路操作，适用于企业内网或测试环境的资产盘点与基础安全检查。

Python自动化端口扫描系统的设计思路

该工具链旨在将分散的IP资产进行统一探测与登记。整体架构分为四层：配置管理层负责读取外部API地址与扫描引擎路径；数据采集层通过HTTP请求拉取服务器列表并使用专用库规范化格式；核心计算层依赖Nmap执行TCP连接检测并输出结构化报告；持久化层则利用DOM解析库提取关键状态字段，最终同步至关系型数据库供后续查询。

第1步：读取配置与初始化环境

程序启动时需加载独立配置文件，声明外部数据源URL与目标软件安装目录。建议使用绝对路径以避免跨平台解析差异，并在首次运行前确认目录权限与网络可达性。

第2步：通过API获取并清洗IP网段

访问预设的资产管理接口，按地域或集群标识分段拉取原始文本。收到响应后需过滤空行，替换特殊占位符为子网掩码分隔符，并交由IPy模块校验合法性并转换为标准的CIDR表示法（如10.0.0.0/24）。此步骤能有效避免无效路由导致的扫描报错。

第3步：调用Nmap执行批量扫描

遍历格式化后的IP列表，构建命令行指令。-PT参数指定使用TCP Ping模式，适合防火墙严格限制ICMP的环境；-oX导出XML格式报告便于程序解析。为避免文件名冲突，建议采用当前日期+主IP命名规则，并统一存放于隔离临时目录。

#coding:utf-8
import sys, os, time, subprocess
import MySQLdb
import re, urllib.request
import configparser
from IPy import IP
import xml.etree.ElementTree as ET

nowtime = time.strftime('%Y-%m-%d', time.localtime(time.time()))
configpath = r'C:\portscan\config.ini'

def getiplist(ipinf):
    serverarea = ['tj101', 'tj103', 'dh', 'dx']
    iplist = []
    for area in serverarea:
        try:
            url = f"{ipinf}{area}"
            ipapi = urllib.request.urlopen(url).read().decode('utf-8')
            for ip in ipapi.split('\n'):
                if ip:
                    ip = ip.replace('_', '/')
                    ip_obj = IP(ip)
                    iplist.append(str(ip_obj))
        except Exception as e:
            print(f"抓取失败: {e}")
    # 去重并触发扫描
    newiplist = list(set(iplist))
    nmapathx = config.get('nmap', 'nmapath')
    ipscan(newiplist, nmapathx)

if __name__ == "__main__":
    config = configparser.ConfigParser()
    config.read(configpath, encoding='utf-8')
    ipinf = config.get('ip', 'ipinf')
    getiplist(ipinf)

第4步：解析XML结果并入库

扫描完成后，遍历ElementTree节点树定位<host>区块。优先检查<state>的state属性是否等于open，仅对确认可达端口执行数据库写入。构造INSERT语句并绑定IP、端口号与状态字段，循环结束后务必刷新事务并提交。

def writeinmysql(scaniplist):
    base_path = r'C:\portscan\scanres'
    conn = MySQLdb.connect(
        host='10.5.10.57', user='nxadmin', passwd='nxadmin.com',
        port=3306, db='scandb', charset='utf8'
    )
    cur = conn.cursor()
    try:
        for ip in scaniplist:
            xmlfile = os.path.join(base_path, f"{ip}.xml")
            if not os.path.exists(xmlfile):
                continue
            root = ET.parse(xmlfile).getroot()
            for host in root.findall('host'):
                address_node = host.find('address')
                if address_node is None:
                    continue
                target_ip = address_node.attrib.get('addr', '')
                ports_node = host.find('ports')
                if ports_node is None:
                    continue
                for port in ports_node.findall('port'):
                    state_tag = port.find('state')
                    if state_tag is not None and state_tag.attrib.get('state') == 'open':
                        portid = port.attrib.get('portid', '')
                        sql = "INSERT INTO portscan (ip, port, state) VALUES(%s, %s, %s)"
                        cur.execute(sql, (target_ip, portid, 'open'))
        conn.commit()
    finally:
        cur.close()
        conn.close()