一、背景

今天有個小需求，要確認客戶端有沒有往服務端發送udp包，但為了減輕工作量，不想每次到機器上手動執行tcpdump抓包命令。于是就寫了個腳本來釋放人力。

二、代碼實現

整個腳本我還加了一些其他功能：時間戳、發送端IP提取，數據包分析，數據持久化等。這里都先去掉，僅記錄下簡單的實時獲取tcpdump輸出功能。代碼如下：

# -*- coding: utf-8 -*-# !/usr/bin/env python# sudo tcpdump -tt -l -nn -c 5 -i enp4s0 udp port 514 or 51414import subprocesscmd = [’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 decodelistening on enp4s0, link-type EN10MB (Ethernet), capture size 262144 bytes1499774951.124178 IP 192.168.10.210.41974 > 192.168.10.251.514: UDP, length 1391499774953.125664 IP 192.168.10.210.54995 > 192.168.10.251.51414: UDP, length 1391499774956.128498 IP 192.168.10.210.56748 > 192.168.10.251.514: UDP, length 1391499774958.129918 IP 192.168.10.210.53883 > 192.168.10.251.51414: UDP, length 1391499774961.132921 IP 192.168.10.210.58803 > 192.168.10.251.514: UDP, length 1395 packets captured6 packets received by filter0 packets dropped by kerneltcpdump finished...

以上代碼相當于手動執行了 sudo tcpdump -tt -l -nn -c 5 -i enp4s0 udp port 514 or 51414 這條命令。注意參數-l很重要（行顯）。

三、代碼實現（更新）

上面的代碼能實現tcpdump的功能，但是有一個問題：沒有做超時保護。即當程序執行時間過長時kill該進程（這里使用ctrl+c的方式）。要實現這個功能有很多種方案，例如定時器+多線程等，這里僅演示一種方案，代碼如下：

# -*- coding: utf-8 -*-# !/usr/bin/env python# sudo tcpdump -tt -l -nn -c 50 -i enp4s0 udp port 514 or 51414import subprocessimport signalimport timeimport osimport reimport jsonclass 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_portsrc_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()

四、總結

比較簡單，僅僅是記錄下。

以上就是python如何實時獲取tcpdump輸出的詳細內容，更多關于python獲取tcpdump輸出的資料請關注好吧啦網其它相關文章！