使用pyshark捕获指定数量数据包时出现TypeError报错求助

问题场景 #

执行命令python python.py -i eth0 -c 10希望捕获10个数据包后退出，但运行时报错。

原代码 #

#!/usr/bin/python

import argparse
import pyshark
import time
import re as regex


parser = argparse.ArgumentParser()
parser.add_argument('-i', '--interface', metavar=" ", type=str, required = True, help = 'To specify the interface ')
parser.add_argument('-v', '--verbose', required = False, action = 'store_true', help = 'To print the all layer of packet')
parser.add_argument('-o', '--output', metavar=' ', help = 'To capture and save the pcap in a file')
parser.add_argument('-p', '--protocol', metavar=' ', help= 'To capture packet using ptotocl filter')
parser.add_argument('-u', '--udp', action = 'store_true', help = 'To capture udp packet only')
parser.add_argument('-t', '--tcp', action = 'store_true', help = 'To capture tcp packet only')
parser.add_argument('-c', '--count', metavar=' ',default=1,  help = 'To capture limited number of packet')

args = parser.parse_args()

if args.protocol:
   capture = pyshark.LiveCapture(interface=args.interface, display_filter=args.protocol)


elif args.udp:
   capture = pyshark.LiveCapture(interface=args.interface, bpf_filter='udp')

elif args.tcp:
   capture = pyshark.LiveCapture(interface=args.interface, bpf_filter='tcp')

else:
   capture = pyshark.LiveCapture(interface=args.interface, output_file=args.output)
   capture.sniff(packet_count = args.count)

packet_list = []

for packet in capture.sniff_continuously():
    if 'IP Layer' in str(packet.layers):
        protocol = regex.search(r'(Protocol:)(.*)',str(packet.ip))
        protocol_type = protocol.group(2).strip().split(' ')[0]
       # proto = protocol_type
    localtime = time.asctime(time.localtime(time.time())) 
    proto = protocol_type
    src_addr = packet.ip.src
    dst_addr = packet.ip.dst
    length = packet.length
    print (localtime, '\t' , proto, '\t' ,src_addr, '\t', dst_addr, '\t' , length)
    if args.verbose:
       print(packet.show())

报错信息 #

python python.py -i eth0 -c 10  

Traceback (most recent call last):
  File "/home/kali/python.py", line 32, in <module>
    capture.sniff(packet_count = args.count)
  File "/home/kali/.local/lib/python3.10/site-packages/pyshark/capture/capture.py", line 144, in load_packets
    self.apply_on_packets(
  File "/home/kali/.local/lib/python3.10/site-packages/pyshark/capture/capture.py", line 256, in apply_on_packets
    return self.eventloop.run_until_complete(coro)
  File "/usr/lib/python3.10/asyncio/base_events.py", line 646, in run_until_complete
    return future.result()
  File "/home/kali/.local/lib/python3.10/site-packages/pyshark/capture/capture.py", line 267, in packets_from_tshark
    await self._go_through_packets_from_fd(tshark_process.stdout, packet_callback, packet_count=packet_count)
  File "/home/kali/.local/lib/python3.10/site-packages/pyshark/capture/capture.py", line 299, in _go_through_packets_from_fd
    if packet_count and packets_captured >= packet_count:
TypeError: '>=' not supported between instances of 'int' and 'str'

问题原因与修复方案 #

1. 核心错误：参数类型不匹配 #

报错显示int和str无法比较，因为args.count默认是字符串类型，而sniff方法需要整数类型的packet_count。

修复：
在argparse定义-c/--count时添加type=int，确保输入的数字被解析为整数：

parser.add_argument('-c', '--count', metavar=' ', type=int, default=1,  help = '指定要捕获的数据包数量')

2. 逻辑漏洞：count参数仅在默认分支生效 #

原代码只有在未指定protocol、udp、tcp参数时才会调用capture.sniff(packet_count=args.count)，其他分支下sniff_continuously()会一直捕获数据包，不会停止。

修复：
将count参数的限制统一应用到所有分支，创建capture对象后统一调用capture.sniff(packet_count=args.count)加载指定数量的包。

3. 其他潜在问题修复 #

• 非IP包会导致protocol_type未定义：初始化protocol_type为默认值，避免报错；
• 拼写错误：将ptotocl filter修正为protocol filter；
• 增加非IP包的兼容处理，避免packet.ip属性不存在时报错。

修改后的完整代码 #

#!/usr/bin/python

import argparse
import pyshark
import time
import re as regex


parser = argparse.ArgumentParser()
parser.add_argument('-i', '--interface', metavar=" ", type=str, required=True, help='指定捕获接口')
parser.add_argument('-v', '--verbose', required=False, action='store_true', help='打印数据包所有层信息')
parser.add_argument('-o', '--output', metavar=' ', help='将捕获内容保存为pcap文件')
parser.add_argument('-p', '--protocol', metavar=' ', help='使用协议过滤器捕获数据包')
parser.add_argument('-u', '--udp', action='store_true', help='仅捕获UDP数据包')
parser.add_argument('-t', '--tcp', action='store_true', help='仅捕获TCP数据包')
parser.add_argument('-c', '--count', metavar=' ', type=int, default=1, help='指定要捕获的数据包数量')

args = parser.parse_args()

# 根据参数创建捕获对象
if args.protocol:
    capture = pyshark.LiveCapture(interface=args.interface, display_filter=args.protocol)
elif args.udp:
    capture = pyshark.LiveCapture(interface=args.interface, bpf_filter='udp')
elif args.tcp:
    capture = pyshark.LiveCapture(interface=args.interface, bpf_filter='tcp')
else:
    capture = pyshark.LiveCapture(interface=args.interface, output_file=args.output)

# 统一加载指定数量的数据包
capture.sniff(packet_count=args.count)

packet_list = []

for packet in capture:
    protocol_type = "Unknown"
    src_addr = "N/A"
    dst_addr = "N/A"
    
    if 'IP Layer' in str(packet.layers) and hasattr(packet, 'ip'):
        protocol = regex.search(r'(Protocol:)(.*)', str(packet.ip))
        if protocol:
            protocol_type = protocol.group(2).strip().split(' ')[0]
        src_addr = packet.ip.src
        dst_addr = packet.ip.dst
    
    localtime = time.asctime(time.localtime(time.time())) 
    length = packet.length
    
    print(localtime, '\t', protocol_type, '\t', src_addr, '\t', dst_addr, '\t', length)
    if args.verbose:
        print(packet.show())

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