- 前言
- python-nmap的基本使用
- PortScanner扫描
- PortScannerAsync异步扫描
- python-nmap的源码分析
前言
Nmap是一个非常用的网络/端口扫描工具，如果想将nmap集成进你的工具里。可以使用python-nmap这个python库，它提供了一个简单的接口来使用nmap进行扫描。
python-nmap的基本使用
在安装这个模块之前，请提前安装好nmap工具，python-nmap模块自身不提供任何扫描功能，只是提供一个接口来使用namp。
pip install python-nmap
目前最新版本是0.7.1支持python3，python2的版本详细参考：https://pypi.org/project/python-nmap/
PortScanner扫描
python-nmap其中的一个核心类是PortScanner，它负责与nmap扫描器进行交互，用于执行扫描并管理扫描结果。
首先需要创建一个PortScanner实例：
import nmap
nm = nmap.PortScanner()
执行扫描：
使用 scan() 方法执行网络扫描。可以指定目标主机、端口范围、扫描类型等参数。
def scan( # NOQA: CFQ001, C901self, hosts="127.0.0.1", ports=None, arguments="-sV", sudo=False, timeout=0
):""":param hosts: 主机字符串，如 nmap 使用的 'scanme.nmap.org' 或 '198.116.0-255.1-127' 或 '216.163.128.20/20':param ports: 端口字符串，如 nmap 使用的 '22,53,110,143-4564':param arguments: nmap 参数字符串 '-sU -sX -sC':param sudo: 如果为 True，则使用 sudo 启动 nmap:param timeout: 整数，如果大于零，将在指定秒数后终止扫描，否则将无限期等待:returns: 扫描结果作为字典"""
例如扫描主机和端口，scan() 返回一个字典作为扫描结果。
scan_result = nm.scan("192.168.88.150", "22")
获取扫描结果：
PortScanner封装了一系列方法，可以方便的获取扫描结果中我们想要的数据，而不需要去手动的解析上面返回的这一长串字典数据。
1、all_hosts() 获取所有扫描的主机
all_hosts = nm.all_hosts()
返回一个排序后的列表，包含所有扫描的ip地址。
['192.168.88.150']
2、command_line() 获取当前用于扫描的nmap命令
command_line = nm.command_line()
返回当前的扫描命令，这里的参数列表中的 -oX - 它会让nmap的把xml格式作为标准输出。
nmap -oX - -p 22 -sV 192.168.88.150
3、scaninfo() 获取当前扫描信息
scaninfo = nm.scaninfo()
返回一个当前扫描信息的字典。
{'tcp': {'method': 'syn', 'services': '22'}}
4、scanstats() 获取扫描统计信息
scan_stats = nm.scanstats()
返回一个当前描统计信息的字典，包括扫描时间等。
{'timestr': 'Mon Dec 30 17:25:17 2024', 'elapsed': '6.59', 'uphosts': '1', 'downhosts': '0', 'totalhosts': '1'}
5、has_host() 检查特定主机是否被扫描
has_host = nm.has_host("192.168.88.150")
如果有扫描结果返回True，否则返回False。
6、以 CSV 格式获取扫描结果
csv_result = nm.csv()
返回csv格式的文本输出。
host;hostname;hostname_type;protocol;port;name;state;product;extrainfo;reason;version;conf;cpe
192.168.88.150;;;tcp;22;ssh;open;OpenSSH;"Ubuntu Linux; protocol 2.0";syn-ack;8.9p1 Ubuntu 3ubuntu0.10;10;cpe:/o:linux:linux_kernel
写一个小案例，扫描一个网段内的所有存活主机。
import nmapnm = nmap.PortScanner()
nm.scan(hosts='192.168.88.0/24', arguments='-sn')
hosts_list = [(x, nm[x]['status']['state']) for x in nm.all_hosts()]for host, status in hosts_list:if status == 'up':print(f'{host} status: {status}')
除了上述PortScanner类提供的几个基本方法，其实还可以更加灵活的运用。
PortScannerAsync异步扫描
对于需要同时扫描多个主机或端口范围的情况，使用PortScanner同步扫描，并不是一个好办法。好在python-nmap提供了一个异步扫描的方案，PortScannerAsync使用多进程技术异步扫描，避免同步扫描可能导致的阻塞，提高了扫描效率。
首先需要创建一个PortScannerAsync实例：
import nmap
nm_async = nmap.PortScannerAsync()
同样是使用 scan() 方法执行扫描：
def scan( # NOQA: CFQ002self,hosts="127.0.0.1",ports=None,arguments="-sV",callback=None,sudo=False,timeout=0,
):""":param hosts: 主机字符串，格式与 nmap 使用的格式相同，例如'scanme.nmap.org' 或 '198.116.0-255.1-127' 或 '216.163.128.20/20'。:param ports: 端口字符串，格式与 nmap 使用的格式相同，例如 '22,53,110,143-4564'。:param arguments: nmap 的参数字符串，例如 '-sU -sX -sC'。:param callback: 回调函数，该函数以（主机，扫描数据）作为参数。:param sudo: 如果为真，则使用 sudo 启动 nmap。:param timeout: 整数，如果大于零，将会在指定秒数之后终止扫描，否则将无限期等待。"""
和PortScanner的scan()很类似，但是多了一个callback参数，需要传一个回调函数，用于扫描结束后的结果处理。
import nmap# 定义回调函数（扫描结果处理）
def scan_callback(host, data):""":param host: 扫描完主机ip地址:param data：扫描结果"""print(host, data)nm_async = nmap.PortScannerAsync()
hosts = '192.168.88.0/24'
ports = '1-1000'
arguments = '-sS'if __name__ == '__main__':# 创建一个新的进程扫描，避免主进程阻塞nm_async.scan(hosts, ports, arguments, callback=scan_callback)while nm_async.still_scanning():# still_scanning判断是否还在扫描print("scanning...")nm_async.wait(1)
使用异步扫描，类似于放到后台扫描，避免了一直阻塞主进程。当然使用Process和PortScanner也可以实现一样的效果，PortScannerAsync是python-nmap封装好了，开箱即用。
python-nmap的源码分析
python-nmap其实已经比较完善了，但是如果想用做一些二次开发，不妨来看看源码，分析分析它的的工作流程。
这里主要看一下PortScanner和PortScannerAsync这两类。其中PortScanner是python-nmap的核心类，scan方法又是PortScanner的核心方法，其实只要了解了scan方法就知道python-nmap的整个逻辑了。
PortScanner的全貌：
后面6个函数前面介绍过，前面的几个函数主要也是为scan函数服务的，重点看scan函数。
scan函数的逻辑并不复杂，简单说，接收用户输入，构建完整的nmap命令行参数列表，交给nmap扫描， 获取nmap扫描结果，解析扫描结果并返回。
scan函数简化后的代码逻辑：
def scan(self, hosts="127.0.0.1", ports=None, arguments="-sV", sudo=False, timeout=0):# 对输入参数（主机、端口、扫描参数等）进行类型检查和合法性验证if sys.version_info[0] == 2:......else:......# shlex模块对主机和扫描参数进行分割处理h_args = shlex.split(hosts)f_args = shlex.split(arguments)# 构建完整的nmap命令行参数列表args = ([self._nmap_path, "-oX", "-"] + h_args + ["-p", ports] * (ports is not None) + f_args)# 启动nmap进程p = subprocess.Popen(args, bufsize=100000, stdin=subprocess.PIPE, stdout=subprocess.PIPE,stderr=subprocess.PIPE,)# 超时处理if timeout == 0:(self._nmap_last_output, nmap_err) = p.communicate()else:......return self.analyse_nmap_xml_scan(nmap_xml_output=self._nmap_last_output, nmap_err=nmap_err)-
它首先对输入参数（主机、端口、扫描参数等）进行类型检查和合法性验证，确保符合nmap命令的要求。
-
然后，使用shlex模块对主机和扫描参数进行分割处理，构建完整的nmap命令行参数列表。
-
在执行扫描过程中，通过subprocess.Popen启动nmap进程，并根据设置的超时时间等待扫描完成或进行超时处理。
-
构建nmap命令行参数列把XML格式作为标准输出。所以定义一个analyse_nmap_xml_scan函数来解析nmap的xml输出，该函数会将nmap生成的xml扫描结果解析为一个结构化的Python字典。
扫描结果的解析与存储也是重要一环，它在analyse_nmap_xml_scan函数实现。
analyse_nmap_xml_scan函数简化后的代码逻辑：
def analyse_nmap_xml_scan( self, nmap_xml_output=None, nmap_err=nmap_err):if nmap_xml_output is not None:self._nmap_last_output = nmap_xml_outputscan_result = {}# 将XML字符串转换为元素树try:dom = ET.fromstring(self._nmap_last_output)except Exception:pass# 扫描结果存储结构, get等方法拿到树中的数据scan_result["nmap"] = {"command_line": dom.get("args"),"scaninfo": {"timestr": dom.find("runstats/finished").get("timestr"),......},}return scan_result在analyse_nmap_xml_scan函数中，它会从nmap_xml_output拿到nmap的xml结果输出，再使用xml.etree.ElementTree模块对nmap扫描生成的xml输出进行解析。使用fromstring方法将xml字符串转换为可操作的元素树，然后遍历树的各个元素，提取关键信息。
通过遍历xml树结构，提取诸如扫描命令行信息、扫描统计数据、主机详细信息。对于每个主机，分别解析其地址信息、主机名信息、端口信息以及脚本输出信息，并将这些信息按照层次结构存储在scan_result字典中，也就是scan函数最后返回的内容。
其实nmap提供5种不同的输出格式，默认的方式是interactive output发送给标准输出。但-oX -会让nmap输出xml到标准输出stdout，而xml输出对于程序处理非常方便的。
PortScannerAsync类也很清晰，它使用PortScanner类进行端口扫描，multiprocessing库的Process类用于创建进程，以此实现异步扫描。
PortScannerAsync类简化后的代码逻辑：
def __scan_progressive__(self, hosts, ports, arguments, callback, timeout):# 此函数在一个单独的进程中执行扫描操作，并调用回调函数处理结果try:scan_data = self._nm.scan(host, ports, arguments, timeout)except Exception:scan_data = Noneif callback is not None:callback(host, scan_data)returnclass PortScannerAsync(object):def __init__(self):self._process = Noneself._nm = PortScanner() #创建一个PortScanner实例，用于执行扫描操作returndef scan(self, hosts="127.0.0.1", ports=None, arguments="-sV", callback=None, timeout=0):if sys.version_info[0] == 2:......else:......self._process = Process( # 创建一个新的进程对象，将 __scan_progressive__ 函数作为目标函数target=__scan_progressive__,args=(self, hosts, ports, arguments, callback, timeout),)self._process.daemon = Trueself._process.start()return__scan_progressive__函数，它是执行扫描任务和调用回调函数的核心部分，调用 self._nm.scan 方法进行扫描，调用 callback 函数以处理扫描结果。
scan函数会创建一个新的进程并指定 _scan_progressive_ 函数作为目标函数，将自身实例、扫描参数和回调函数等传递给该函数。
在目标函数中，针对每个发现的主机，调用PortScanner实例的scan方法进行扫描，并在扫描完成后调用用户提供的回调函数，将主机信息和扫描结果传递给回调函数进行处理。
PortScannerAsync还提供了一些管理进程的函数，用于扩展。
以上就是一个简单的分析，但愿有点用吧。
参考文章
https://pypi.org/project/python-nmap/
若有错误，欢迎指正！o(￣▽￣)ブ
