如何用多线程调用Nmap子进程扫描子网并实时输出（附错误排查）

核心错误解析 #

你遇到的AttributeError: '_io.TextIOWrapper' object has no attribute 'communicate'是误用subprocess API导致的：communicate()是subprocess.Popen实例的方法，不能直接调用在stdout文件句柄上。除此之外，代码还有多处逻辑和语法错误，以下是完整修复方案。

关键问题列表 #

1. Nmap参数传递错误：原代码用逗号分隔参数，导致命令执行失败；且shell=True存在安全风险。
2. IP合法性判断失效：text.upper().isupper() is False完全无法验证IP/子网有效性。
3. 输出处理逻辑混乱：先调用communicate()读取全部输出，又试图循环读stderr，逻辑矛盾。
4. 变量引用错误：sys.scanout_q不存在，混淆了全局与局部变量。
5. Awk命令参数错误：正则表达式格式错误，跨进程调用增加复杂度。

修正后的完整代码 #

import queue
import threading
import subprocess
from ipaddress import IPv4Network, AddressValueError, NetmaskValueError

def Scan_IP(self):
    num_threads = 50
    scanin_q = queue.Queue()
    scanout_q = queue.Queue()

    text = self.textbox.text().strip()

    # 验证并处理IP/子网
    try:
        # 验证输入为合法IPv4网络（strict=False允许主机位不为0的子网）
        network = IPv4Network(text, strict=False)
        
        def scan_worker(q_in, q_out):
            while True:
                ip = q_in.get()
                if ip is None:  # 线程终止信号
                    break
                
                # 构建Nmap命令参数列表（安全且避免shell注入）
                nmap_args = [
                    'nmap', '-Pn', '--open', '-T5',
                    str(ip)
                ]
                
                # 启动Nmap进程，合并 stdout 和 stderr
                nmap_proc = subprocess.Popen(
                    nmap_args,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.STDOUT,
                    encoding='utf-8',
                    shell=False
                )
                
                # 实时读取输出，过滤并立即显示
                for line in nmap_proc.stdout:
                    line = line.strip()
                    # 过滤需要的输出行（替代原Awk逻辑）
                    target_keywords = [
                        'Nmap scan report', 'STATE', 'tcp', 'type:',
                        'Running', 'details:', 'Service Info:'
                    ]
                    if any(keyword in line for keyword in target_keywords):
                        print(line)  # 立即打印结果
                        q_out.put(line)  # 存入结果队列
                
                # 等待进程结束，释放资源
                nmap_proc.wait()
                q_in.task_done()

        # 启动线程池
        threads = []
        for _ in range(num_threads):
            thread = threading.Thread(target=scan_worker, args=(scanin_q, scanout_q))
            thread.daemon = True
            thread.start()
            threads.append(thread)

        # 将子网内所有主机IP加入任务队列
        for ip in network.hosts():
            scanin_q.put(ip)

        # 等待所有扫描任务完成
        scanin_q.join()

        # 向每个线程发送终止信号，确保线程正常退出
        for _ in range(num_threads):
            scanin_q.put(None)
        for thread in threads:
            thread.join()

    except (AddressValueError, NetmaskValueError):
        print('\n无效的IP地址/子网，请重试！')
    except Exception as e:
        print(f"扫描出错: {str(e)}")

修复说明 #

• Subprocess错误修复：移除错误的stdout.communicate()调用，改为逐行读取Nmap输出，实现实时结果展示。
• IP验证逻辑修正：使用ipaddress模块的异常捕获来判断输入合法性，替代原无效的字符串判断。
• 安全参数传递：使用参数列表而非shell命令字符串，避免shell注入风险，同时确保参数传递准确。
• 输出过滤优化：用Python内置逻辑替代Awk调用，降低跨进程复杂度，提高代码可维护性。
• 线程管理改进：添加线程终止信号，确保所有线程可以正常退出，避免资源泄漏。

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