Python实时获取tcpdump输出及超时控制实现

陈默

发布时间：2026-10-14 | 2人浏览过

原创

本文介绍如何使用Python脚本实时获取tcpdump的网络抓包输出，并通过subprocess模块解析标准输出。针对长时间运行可能导致的进程挂起问题，文章提供了基于信号和超时的进程终止方案，实现自动清理资源并保存抓包数据的功能。

背景与需求分析

一、背景

近期有一个实际需求，需要确认客户端是否向服务端发送了UDP数据包。为了避免每次手动登录服务器执行tcpdump抓包命令所带来的繁琐工作，决定编写一个自动化脚本来替代人工操作，从而释放人力。

基础实时获取实现

二、代码实现

完整的脚本中包含了时间戳记录、发送端IP提取、数据包详细分析以及数据持久化存储等附加功能。为了清晰展示核心逻辑，此处暂时省略这些扩展功能，仅保留实现实时获取tcpdump输出的基础代码。具体代码如下：


# -*- coding: utf-8 -*-
# !/usr/bin/env python

# sudo tcpdump -tt -l -nn -c 5 -i enp4s0 udp port 514 or 51414

import subprocess

cmd = ['sudo', 'tcpdump', '-tt', '-l', '-nn', '-c', '5', '-i', 'enp4s0', 'udp', 'port', '514', 'or', '51414']
proc = subprocess.Popen(cmd, stdout=subprocess.PIPE)

while True:
  line = proc.stdout.readline()
  line = line.strip()
  if not line:
    print('tcpdump finished...')
    break
  print(line)

运行上述代码后，控制台将实时显示如下输出结果：


wenyuanblog@localhost:/home/test/script# python tcpdump_udp.py 
tcpdump: verbose output suppressed, use -v or -vv for full protocol decode
listening on enp4s0, link-type EN10MB (Ethernet), capture size 262144 bytes
1499774951.124178 IP 192.168.10.210.41974 > 192.168.10.251.514: UDP, length 139
1499774953.125664 IP 192.168.10.210.54995 > 192.168.10.251.51414: UDP, length 139
1499774956.128498 IP 192.168.10.210.56748 > 192.168.10.251.514: UDP, length 139
1499774958.129918 IP 192.168.10.210.53883 > 192.168.10.251.51414: UDP, length 139
1499774961.132921 IP 192.168.10.210.58803 > 192.168.10.251.514: UDP, length 139
5 packets captured
6 packets received by filter
0 packets dropped by kernel
tcpdump finished...

上述Python代码的功能等价于在终端手动执行 sudo tcpdump -tt -l -nn -c 5 -i enp4s0 udp port 514 or 51414 命令。需要特别注意的是，参数-l（即--line-buffered）非常关键，它确保tcpdump的输出是逐行缓冲的，从而使Python能够实时读取每一行数据，否则可能导致输出延迟或缓存问题。

增强版：添加超时保护机制

三、代码实现（更新）

前文提到的代码虽然实现了tcpdump的基本功能，但存在一个缺陷：缺乏超时保护机制。如果程序执行时间过长或陷入死循环，无法通过Ctrl+C等方式有效终止进程。为了解决这个问题，可以采用定时器结合多线程等多种方案。这里演示一种基于信号处理的实现方式，代码如下：


# -*- coding: utf-8 -*-
# !/usr/bin/env python

# sudo tcpdump -tt -l -nn -c 50 -i enp4s0 udp port 514 or 51414

import subprocess
import signal
import time
import os
import re
import json


class CmdServer:

  def __init__(self, cmd, timeout=120):
    '''
    :param cmd: 执行命令（列表形式）
    :param timeout: 任务超时时间（seconds，进程运行超过该时间，kill该进程）
    :param taskname: 任务名称（根据该任务名称记录命令输出信息）
    '''
    self.cmd = cmd
    self.timeout = timeout
    self.base_path = reduce(lambda x, y: os.path.dirname(x), range(1), os.path.abspath(__file__))
    self.output_path = os.path.join(self.base_path, 'data.json')
    self.udp_flow_list = []
    self.begin_time = int(time.time())

  # 执行tcpdump任务
  def run(self):
    if os.path.exists(self.output_path):
      with open(self.output_path, 'r') as f:
        self.udp_flow_list = json.load(f)

    proc = subprocess.Popen(self.cmd, stdout=subprocess.PIPE)
    stdout = ''

    while proc.poll() == None:
      current_time = int(time.time())
      if current_time - self.begin_time >= self.timeout:
        print('tcpdump timeout...')
        proc.send_signal(signal.SIGINT)
        stdout = proc.stdout.read()

    if proc.poll() is not None and not stdout:
      print('tcpdump finished...')
      stdout = proc.stdout.read()

    stdout_list = stdout.split('\n')
    if stdout_list:
      self._merge_data(stdout_list)
      self._save_data()

  # 数据合并（新增/更新）
  def _merge_data(self, stdout_list):
    for line in stdout_list:
      line = line.strip()
      if not line:
        continue
      timestamp = int(float(line.split('IP')[0].strip())) * 1000
      # 源
      src_ip_port_list = re.findall(r'IP(.+?)>', line)
      if not src_ip_port_list:
        continue
      src_ip_port_str = src_ip_port_list[0].strip()
      src_ip = '.'.join(src_ip_port_str.split('.')[0:4])
      # 目的
      dst_ip_port_list = re.findall(r'>(.+?):', line)
      if not dst_ip_port_list:
        continue
      dst_ip_port_str = dst_ip_port_list[0].strip()
      dst_port = dst_ip_port_str.split('.')[-1]

      # 新增/更新latest_timestamp
      src_item = filter(lambda x: src_ip == x['src_ip'], self.udp_flow_list)
      if src_item:
        src_item[0]['dst_port'] = dst_port
        src_item[0]['latest_timestamp'] = timestamp
      else:
        self.udp_flow_list.append(dict(
          src_ip=src_ip,
          dst_port=dst_port,
          latest_timestamp=timestamp
        ))

  # 保存数据
  def _save_data(self):
    # 写入文件
    with open(self.output_path, 'w') as f:
      json.dump(self.udp_flow_list, f, encoding="utf-8", ensure_ascii=False)


if __name__ == '__main__':
  cmd = ['sudo', 'tcpdump', '-tt', '-l', '-nn', '-c', '5', '-i', 'enp4s0', 'udp', 'port', '514', 'or', '51414']
  cmd_server = CmdServer(cmd, 10)
  cmd_server.run()