python-nmap实现python利用nmap扫描分析

技术分享 3天前 0 999+

前言

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, C901     self, 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 nmap  nm = 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: CFQ002     self,     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_output      scan_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 = None      if callback is not None:         callback(host, scan_data)     return  class PortScannerAsync(object):      def __init__(self):         self._process = None         self._nm = PortScanner()   #创建一个PortScanner实例，用于执行扫描操作         return      def 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 = True         self._process.start()          return

__scan_progressive__函数，它是执行扫描任务和调用回调函数的核心部分，调用 self._nm.scan 方法进行扫描，调用 callback 函数以处理扫描结果。

scan函数会创建一个新的进程并指定 _scan_progressive_ 函数作为目标函数，将自身实例、扫描参数和回调函数等传递给该函数。

在目标函数中，针对每个发现的主机，调用PortScanner实例的scan方法进行扫描，并在扫描完成后调用用户提供的回调函数，将主机信息和扫描结果传递给回调函数进行处理。

PortScannerAsync还提供了一些管理进程的函数，用于扩展。