Python 3调用Shell网络工具的最优实现方案咨询

刚好之前做过类似的Python工具集成网络命令的需求，给你梳理一套实用的最优实现方案，覆盖你提到的所有痛点：

1. 搞定管道命令 & 输出解析的痛点 #

你说subprocess.check_output没法处理管道，还不好解析输出？其实换用subprocess.Popen就能解决，而且支持两种管道实现方式：

方式一：直接用shell（适合可信命令） #

如果你的命令是自己写的、没有用户输入的风险，可以直接把整个管道命令作为字符串传入，开启shell=True，这样grep/awk都能正常工作：

import subprocess

# 示例：nmap扫描后用grep过滤结果
cmd = "nmap -sP 192.168.1.0/24 | grep 'Nmap scan report for'"
process = subprocess.Popen(
    cmd,
    shell=True,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True  # 直接返回字符串，不用手动解码
)
stdout, stderr = process.communicate()

# 现在可以轻松解析输出了
for line in stdout.splitlines():
    print(f"找到目标：{line.split()[-1]}")

方式二：手动拼接管道（更安全，避免注入） #

如果命令里有用户输入，绝对不能用shell=True，可以用多个Popen对象手动连接管道：

# 第一步：运行nmap
nmap_proc = subprocess.Popen(
    ["nmap", "-sP", "192.168.1.0/24"],
    stdout=subprocess.PIPE,
    text=True
)
# 第二步：把nmap的输出传给grep
grep_proc = subprocess.Popen(
    ["grep", "Nmap scan report for"],
    stdin=nmap_proc.stdout,
    stdout=subprocess.PIPE,
    text=True
)
nmap_proc.stdout.close()  # 告诉nmap：输出已经被grep接手了

stdout, stderr = grep_proc.communicate()

实时处理持续输出 #

像arpspoof、SSLStrip这种持续输出的工具，别用communicate()一次性读，而是在循环里实时读取输出，方便存储或处理：

def read_continuous_output(process, output_list):
    for line in iter(process.stdout.readline, ''):
        if line:
            output_list.append(line.strip())  # 把输出存到列表里
    process.wait()

# 示例：启动arpspoof并实时记录输出
output = []
arp_proc = subprocess.Popen(
    ["arpspoof", "-i", "eth0", "-t", "192.168.1.1", "192.168.1.100"],
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    text=True
)
read_continuous_output(arp_proc, output)

2. 用线程独立运行每个命令 #

要让每个命令在独立线程跑，用Python自带的threading模块就行，同时用字典保存每个线程对应的进程实例，方便后续终止：

import threading
from queue import Queue

# 全局存储：key是命令ID，value是Popen进程对象
running_processes = {}
# 存储每个命令的输出：用Queue线程安全，或者用列表也可以
command_outputs = {}

def start_command(cmd_id, cmd_args):
    # 初始化输出存储
    output_queue = Queue()
    command_outputs[cmd_id] = output_queue

    def thread_task():
        # 启动进程
        process = subprocess.Popen(
            cmd_args,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True
        )
        running_processes[cmd_id] = process

        # 实时读取输出到队列
        for line in iter(process.stdout.readline, ''):
            if line:
                output_queue.put(line.strip())
        
        # 进程结束后清理
        process.wait()
        del running_processes[cmd_id]

    # 启动线程
    thread = threading.Thread(target=thread_task)
    thread.daemon = True  # 主线程退出时自动销毁子线程
    thread.start()
    return thread

# 示例：启动两个独立的命令
start_command("arpspoof", ["arpspoof", "-i", "eth0", "-t", "192.168.1.1", "192.168.1.100"])
start_command("nmap_scan", ["nmap", "-sV", "192.168.1.1"])

3. 优雅终止持续运行的工具 #

要终止SSLStrip、arpspoof这类工具，直接从running_processes里取出对应的进程对象，调用terminate()或发送信号即可：

import signal

def stop_command(cmd_id):
    if cmd_id not in running_processes:
        print(f"命令{cmd_id}没有在运行")
        return
    
    process = running_processes[cmd_id]
    # 先尝试优雅终止（比如发送Ctrl+C的SIGINT信号）
    process.send_signal(signal.SIGINT)
    # 如果5秒内没结束，强制终止
    try:
        process.wait(timeout=5)
    except subprocess.TimeoutExpired:
        process.kill()
    
    # 可选：清理输出存储
    if cmd_id in command_outputs:
        # 比如把队列里的输出写入文件
        with open(f"{cmd_id}_output.log", "w") as f:
            while not command_outputs[cmd_id].empty():
                f.write(command_outputs[cmd_id].get() + "\n")
        del command_outputs[cmd_id]

# 示例：终止arpspoof
stop_command("arpspoof")

额外实用建议 #

• 错误处理：把stderr=subprocess.STDOUT合并到stdout，方便统一处理所有输出；或者单独读取stderr排查问题。
• 安全第一：永远不要给不可信的用户输入用shell=True，避免shell注入攻击，尽量用列表形式传递命令参数。
• 输出持久化：如果需要长期存储输出，可以把队列里的内容定时写入文件，或者直接用列表存储后批量写入。

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