Package dpkt :: Module pcap
[hide private]
[frames] | no frames]

Source Code for Module dpkt.pcap

  1  # $Id: pcap.py 77 2011-01-06 15:59:38Z dugsong $ 
  2  # -*- coding: utf-8 -*- 
  3  """Libpcap file format.""" 
  4  from __future__ import print_function 
  5  from __future__ import absolute_import 
  6   
  7  import sys 
  8  import time 
  9  from decimal import Decimal 
 10   
 11  from . import dpkt 
 12   
 13  TCPDUMP_MAGIC = 0xa1b2c3d4 
 14  TCPDUMP_MAGIC_NANO = 0xa1b23c4d 
 15  PMUDPCT_MAGIC = 0xd4c3b2a1 
 16  PMUDPCT_MAGIC_NANO = 0x4d3cb2a1 
 17   
 18  PCAP_VERSION_MAJOR = 2 
 19  PCAP_VERSION_MINOR = 4 
 20   
 21  # see http://www.tcpdump.org/linktypes.html for explanations 
 22  DLT_NULL = 0 
 23  DLT_EN10MB = 1 
 24  DLT_EN3MB = 2 
 25  DLT_AX25 = 3 
 26  DLT_PRONET = 4 
 27  DLT_CHAOS = 5 
 28  DLT_IEEE802 = 6 
 29  DLT_ARCNET = 7 
 30  DLT_SLIP = 8 
 31  DLT_PPP = 9 
 32  DLT_FDDI = 10 
 33  DLT_PFSYNC = 18 
 34  DLT_PPP_SERIAL = 50 
 35  DLT_PPP_ETHER = 51 
 36  DLT_ATM_RFC1483 = 100 
 37  DLT_RAW = 101 
 38  DLT_C_HDLC = 104 
 39  DLT_IEEE802_11 = 105 
 40  DLT_FRELAY = 107 
 41  DLT_LOOP = 108 
 42  DLT_LINUX_SLL = 113 
 43  DLT_LTALK = 114 
 44  DLT_PFLOG = 117 
 45  DLT_PRISM_HEADER = 119 
 46  DLT_IP_OVER_FC = 122 
 47  DLT_SUNATM = 123 
 48  DLT_IEEE802_11_RADIO = 127 
 49  DLT_ARCNET_LINUX = 129 
 50  DLT_APPLE_IP_OVER_IEEE1394 = 138 
 51  DLT_MTP2_WITH_PHDR = 139 
 52  DLT_MTP2 = 140 
 53  DLT_MTP3 = 141 
 54  DLT_SCCP = 142 
 55  DLT_DOCSIS = 143 
 56  DLT_LINUX_IRDA = 144 
 57  DLT_USER0 = 147 
 58  DLT_USER1 = 148 
 59  DLT_USER2 = 149 
 60  DLT_USER3 = 150 
 61  DLT_USER4 = 151 
 62  DLT_USER5 = 152 
 63  DLT_USER6 = 153 
 64  DLT_USER7 = 154 
 65  DLT_USER8 = 155 
 66  DLT_USER9 = 156 
 67  DLT_USER10 = 157 
 68  DLT_USER11 = 158 
 69  DLT_USER12 = 159 
 70  DLT_USER13 = 160 
 71  DLT_USER14 = 161 
 72  DLT_USER15 = 162 
 73  DLT_IEEE802_11_RADIO_AVS = 163 
 74  DLT_BACNET_MS_TP = 165 
 75  DLT_PPP_PPPD = 166 
 76  DLT_GPRS_LLC = 169 
 77  DLT_GPF_T = 170 
 78  DLT_GPF_F = 171 
 79  DLT_LINUX_LAPD = 177 
 80  DLT_BLUETOOTH_HCI_H4 = 187 
 81  DLT_USB_LINUX = 189 
 82  DLT_PPI = 192 
 83  DLT_IEEE802_15_4 = 195 
 84  DLT_SITA = 196 
 85  DLT_ERF = 197 
 86  DLT_BLUETOOTH_HCI_H4_WITH_PHDR = 201 
 87  DLT_AX25_KISS = 202 
 88  DLT_LAPD = 203 
 89  DLT_PPP_WITH_DIR = 204 
 90  DLT_C_HDLC_WITH_DIR = 205 
 91  DLT_FRELAY_WITH_DIR = 206 
 92  DLT_IPMB_LINUX = 209 
 93  DLT_IEEE802_15_4_NONASK_PHY = 215 
 94  DLT_USB_LINUX_MMAPPED = 220 
 95  DLT_FC_2 = 224 
 96  DLT_FC_2_WITH_FRAME_DELIMS = 225 
 97  DLT_IPNET = 226 
 98  DLT_CAN_SOCKETCAN = 227 
 99  DLT_IPV4 = 228 
100  DLT_IPV6 = 229 
101  DLT_IEEE802_15_4_NOFCS = 230 
102  DLT_DBUS = 231 
103  DLT_DVB_CI = 235 
104  DLT_MUX27010 = 236 
105  DLT_STANAG_5066_D_PDU = 237 
106  DLT_NFLOG = 239 
107  DLT_NETANALYZER = 240 
108  DLT_NETANALYZER_TRANSPARENT = 241 
109  DLT_IPOIB = 242 
110  DLT_MPEG_2_TS = 243 
111  DLT_NG40 = 244 
112  DLT_NFC_LLCP = 245 
113  DLT_INFINIBAND = 247 
114  DLT_SCTP = 248 
115  DLT_USBPCAP = 249 
116  DLT_RTAC_SERIAL = 250 
117  DLT_BLUETOOTH_LE_LL = 251 
118  DLT_NETLINK = 253 
119  DLT_BLUETOOTH_LINUX_MONITOR = 253 
120  DLT_BLUETOOTH_BREDR_BB = 255 
121  DLT_BLUETOOTH_LE_LL_WITH_PHDR = 256 
122  DLT_PROFIBUS_DL = 257 
123  DLT_PKTAP = 258 
124  DLT_EPON = 259 
125  DLT_IPMI_HPM_2 = 260 
126  DLT_ZWAVE_R1_R2 = 261 
127  DLT_ZWAVE_R3 = 262 
128  DLT_WATTSTOPPER_DLM = 263 
129  DLT_ISO_14443 = 264 
130   
131  if sys.platform.find('openbsd') != -1: 
132      DLT_LOOP = 12 
133      DLT_RAW = 14 
134  else: 
135      DLT_LOOP = 108 
136      DLT_RAW = 12 
137   
138  dltoff = {DLT_NULL: 4, DLT_EN10MB: 14, DLT_IEEE802: 22, DLT_ARCNET: 6, 
139            DLT_SLIP: 16, DLT_PPP: 4, DLT_FDDI: 21, DLT_PFLOG: 48, DLT_PFSYNC: 4, 
140            DLT_LOOP: 4, DLT_LINUX_SLL: 16} 
141 142 143 -class PktHdr(dpkt.Packet):
144 """pcap packet header. 145 146 TODO: Longer class information.... 147 148 Attributes: 149 __hdr__: Header fields of pcap header. 150 TODO. 151 """ 152 __hdr__ = ( 153 ('tv_sec', 'I', 0), 154 ('tv_usec', 'I', 0), 155 ('caplen', 'I', 0), 156 ('len', 'I', 0), 157 )
158
159 160 -class LEPktHdr(PktHdr):
161 __byte_order__ = '<'
162
163 164 -class FileHdr(dpkt.Packet):
165 """pcap file header. 166 167 TODO: Longer class information.... 168 169 Attributes: 170 __hdr__: Header fields of pcap file header. 171 TODO. 172 """ 173 174 __hdr__ = ( 175 ('magic', 'I', TCPDUMP_MAGIC), 176 ('v_major', 'H', PCAP_VERSION_MAJOR), 177 ('v_minor', 'H', PCAP_VERSION_MINOR), 178 ('thiszone', 'I', 0), 179 ('sigfigs', 'I', 0), 180 ('snaplen', 'I', 1500), 181 ('linktype', 'I', 1), 182 )
183
184 185 -class LEFileHdr(FileHdr):
186 __byte_order__ = '<'
187
188 189 -class Writer(object):
190 """Simple pcap dumpfile writer. 191 192 TODO: Longer class information.... 193 194 Attributes: 195 __hdr__: Header fields of simple pcap dumpfile writer. 196 TODO. 197 """ 198
199 - def __init__(self, fileobj, snaplen=1500, linktype=DLT_EN10MB, nano=False):
200 self.__f = fileobj 201 self._precision = 9 if nano else 6 202 magic = TCPDUMP_MAGIC_NANO if nano else TCPDUMP_MAGIC 203 if sys.byteorder == 'little': 204 fh = LEFileHdr(snaplen=snaplen, linktype=linktype, magic=magic) 205 else: 206 fh = FileHdr(snaplen=snaplen, linktype=linktype, magic=magic) 207 self.__f.write(bytes(fh))
208
209 - def writepkt(self, pkt, ts=None):
210 if ts is None: 211 ts = time.time() 212 s = bytes(pkt) 213 n = len(s) 214 sec = int(ts) 215 usec = int(round(ts % 1 * 10 ** self._precision)) 216 if sys.byteorder == 'little': 217 ph = LEPktHdr(tv_sec=sec, 218 tv_usec=usec, 219 caplen=n, len=n) 220 else: 221 ph = PktHdr(tv_sec=sec, 222 tv_usec=usec, 223 caplen=n, len=n) 224 self.__f.write(bytes(ph)) 225 self.__f.write(s)
226
227 - def close(self):
228 self.__f.close()
229
230 231 -class Reader(object):
232 """Simple pypcap-compatible pcap file reader. 233 234 TODO: Longer class information.... 235 236 Attributes: 237 __hdr__: Header fields of simple pypcap-compatible pcap file reader. 238 TODO. 239 """ 240
241 - def __init__(self, fileobj):
242 self.name = getattr(fileobj, 'name', '<%s>' % fileobj.__class__.__name__) 243 self.__f = fileobj 244 buf = self.__f.read(FileHdr.__hdr_len__) 245 self.__fh = FileHdr(buf) 246 self.__ph = PktHdr 247 if self.__fh.magic in (PMUDPCT_MAGIC, PMUDPCT_MAGIC_NANO): 248 self.__fh = LEFileHdr(buf) 249 self.__ph = LEPktHdr 250 elif self.__fh.magic not in (TCPDUMP_MAGIC, TCPDUMP_MAGIC_NANO): 251 raise ValueError('invalid tcpdump header') 252 if self.__fh.linktype in dltoff: 253 self.dloff = dltoff[self.__fh.linktype] 254 else: 255 self.dloff = 0 256 self._divisor = 1E6 if self.__fh.magic in (TCPDUMP_MAGIC, PMUDPCT_MAGIC) else Decimal('1E9') 257 self.snaplen = self.__fh.snaplen 258 self.filter = '' 259 self.__iter = iter(self)
260 261 @property
262 - def fd(self):
263 return self.__f.fileno()
264
265 - def fileno(self):
266 return self.fd
267 270
271 - def setfilter(self, value, optimize=1):
272 return NotImplementedError
273
274 - def readpkts(self):
275 return list(self)
276
277 - def __next__(self):
278 return next(self.__iter)
279
280 - def dispatch(self, cnt, callback, *args):
281 """Collect and process packets with a user callback. 282 283 Return the number of packets processed, or 0 for a savefile. 284 285 Arguments: 286 287 cnt -- number of packets to process; 288 or 0 to process all packets until EOF 289 callback -- function with (timestamp, pkt, *args) prototype 290 *args -- optional arguments passed to callback on execution 291 """ 292 processed = 0 293 if cnt > 0: 294 for _ in range(cnt): 295 try: 296 ts, pkt = next(iter(self)) 297 except StopIteration: 298 break 299 callback(ts, pkt, *args) 300 processed += 1 301 else: 302 for ts, pkt in self: 303 callback(ts, pkt, *args) 304 processed += 1 305 return processed
306
307 - def loop(self, callback, *args):
308 self.dispatch(0, callback, *args)
309
310 - def __iter__(self):
311 while 1: 312 buf = self.__f.read(PktHdr.__hdr_len__) 313 if not buf: 314 break 315 hdr = self.__ph(buf) 316 buf = self.__f.read(hdr.caplen) 317 yield (hdr.tv_sec + (hdr.tv_usec / self._divisor), buf)
318
319 320 -def test_pcap_endian():
321 be = b'\xa1\xb2\xc3\xd4\x00\x02\x00\x04\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x60\x00\x00\x00\x01' 322 le = b'\xd4\xc3\xb2\xa1\x02\x00\x04\x00\x00\x00\x00\x00\x00\x00\x00\x00\x60\x00\x00\x00\x01\x00\x00\x00' 323 befh = FileHdr(be) 324 lefh = LEFileHdr(le) 325 assert (befh.linktype == lefh.linktype)
326
327 328 -def test_reader():
329 data = ( # full libpcap file with one packet 330 b'\xd4\xc3\xb2\xa1\x02\x00\x04\x00\x00\x00\x00\x00\x00\x00\x00\x00\xff\xff\x00\x00\x01\x00\x00\x00' 331 b'\xb2\x67\x4a\x42\xae\x91\x07\x00\x46\x00\x00\x00\x46\x00\x00\x00\x00\xc0\x9f\x32\x41\x8c\x00\xe0' 332 b'\x18\xb1\x0c\xad\x08\x00\x45\x00\x00\x38\x00\x00\x40\x00\x40\x11\x65\x47\xc0\xa8\xaa\x08\xc0\xa8' 333 b'\xaa\x14\x80\x1b\x00\x35\x00\x24\x85\xed' 334 ) 335 336 # --- BytesIO tests --- 337 from .compat import BytesIO 338 339 # BytesIO 340 fobj = BytesIO(data) 341 reader = Reader(fobj) 342 assert reader.name == '<BytesIO>' 343 _, buf1 = next(iter(reader)) 344 assert buf1 == data[FileHdr.__hdr_len__ + PktHdr.__hdr_len__:] 345 346 # --- dispatch() tests --- 347 348 # test count = 0 349 fobj.seek(0) 350 reader = Reader(fobj) 351 assert reader.dispatch(0, lambda ts, pkt: None) == 1 352 353 # test count > 0 354 fobj.seek(0) 355 reader = Reader(fobj) 356 assert reader.dispatch(4, lambda ts, pkt: None) == 1 357 358 # test iterative dispatch 359 fobj.seek(0) 360 reader = Reader(fobj) 361 assert reader.dispatch(1, lambda ts, pkt: None) == 1 362 assert reader.dispatch(1, lambda ts, pkt: None) == 0
363
364 365 -def test_writer_precision():
366 data = b'foo' 367 from .compat import BytesIO 368 369 # default precision 370 fobj = BytesIO() 371 writer = Writer(fobj) 372 writer.writepkt(data, ts=1454725786.526401) 373 fobj.flush() 374 fobj.seek(0) 375 376 reader = Reader(fobj) 377 ts, buf1 = next(iter(reader)) 378 assert ts == 1454725786.526401 379 assert buf1 == b'foo' 380 381 # nano precision 382 from decimal import Decimal 383 384 fobj = BytesIO() 385 writer = Writer(fobj, nano=True) 386 writer.writepkt(data, ts=Decimal('1454725786.010203045')) 387 fobj.flush() 388 fobj.seek(0) 389 390 reader = Reader(fobj) 391 ts, buf1 = next(iter(reader)) 392 assert ts == Decimal('1454725786.010203045') 393 assert buf1 == b'foo'
394 395 396 if __name__ == '__main__': 397 test_pcap_endian() 398 test_reader() 399 test_writer_precision() 400 401 print('Tests Successful...') 402