1+ import os
2+ import struct
3+ import time
4+ import math
5+
6+ from urh .util .Logger import logger
7+
8+ # Refer to PCAPNG spec
9+ # https://www.ietf.org/staging/draft-tuexen-opsawg-pcapng-02.html
10+
11+ class PCAPNG (object ):
12+ def __init__ (self ):
13+ self .__messages = []
14+ self .__messages_timestamps = []
15+ self .__filename = ""
16+
17+ def __build_pcapng_shb (self , shb_userappl : str = "" , shb_hardware : str = "" ) -> bytes :
18+ BLOCKTYPE = 0x0A0D0D0A
19+ HEADERS_BLOCK_LENGTH = 28
20+ MAGIC_NUMBER = 0x1A2B3C4D
21+ VERSION_MAJOR , VERSION_MINOR = 1 ,0
22+ SECTIONLENGTH = 0xFFFFFFFFFFFFFFFF # -1 => Not specified
23+
24+ shb_userappl_padded_len = math .ceil (len (shb_userappl )/ 4 ) * 4
25+ shb_hardware_padded_len = math .ceil (len (shb_hardware )/ 4 ) * 4
26+
27+ total_block_len = HEADERS_BLOCK_LENGTH
28+ if shb_userappl_padded_len > 0 :
29+ total_block_len += shb_userappl_padded_len + 4
30+
31+ if shb_hardware_padded_len > 0 :
32+ total_block_len += shb_hardware_padded_len + 4
33+
34+ shb = struct .pack (">IIIHHQ" ,
35+ BLOCKTYPE ,
36+ total_block_len ,
37+ MAGIC_NUMBER ,
38+ VERSION_MAJOR , VERSION_MINOR ,
39+ SECTIONLENGTH )
40+
41+ if shb_userappl != "" :
42+ SHB_USERAPPL = 4
43+ strpad = shb_userappl .ljust (shb_userappl_padded_len , "\0 " )
44+ shb += struct .pack (">HH" , SHB_USERAPPL , shb_userappl_padded_len )
45+ shb += bytes (strpad , 'ascii' )
46+
47+ if shb_hardware != "" :
48+ SHB_HARDWARE = 2
49+ strpad = shb_hardware .ljust (shb_hardware_padded_len , "\0 " )
50+ shb += struct .pack (">HH" , SHB_HARDWARE , shb_hardware_padded_len )
51+ shb += bytes (strpad , 'ascii' )
52+
53+ shb += struct .pack (">I" ,total_block_len )
54+ return shb
55+
56+ def __build_pcapng_idb (self , link_type ) -> bytes :
57+ BLOCKTYPE = 0x00000001
58+ BLOCKLENGTH = 20
59+ SNAP_LEN = 0
60+
61+ return struct .pack (">IIHHII" ,
62+ BLOCKTYPE ,
63+ BLOCKLENGTH ,
64+ link_type , 0 ,
65+ SNAP_LEN ,
66+ BLOCKLENGTH )
67+
68+ def __build_pcapng_epb (self , packet : bytes , timestamp : float ) -> bytes :
69+ BLOCKTYPE = 0x00000006
70+ BLOCKHEADERLEN = 32
71+ INTERFACE_ID = 0
72+
73+ captured_packet_len = len (packet )
74+ original_packet_len = captured_packet_len
75+ padded_packet_len = math .ceil (captured_packet_len / 4 ) * 4
76+ padding_len = padded_packet_len - original_packet_len
77+ padded_packet = packet + bytearray (padding_len )
78+ block_total_length = BLOCKHEADERLEN + padded_packet_len
79+ timestamp_int = int (timestamp * 1e6 ) # Set the proper resolution
80+ timestamp_high = timestamp_int >> 32
81+ timestamp_low = timestamp_int & 0x00000000FFFFFFFF
82+
83+ epb = struct .pack (">IIIIIII" ,
84+ BLOCKTYPE ,
85+ block_total_length ,
86+ INTERFACE_ID ,
87+ timestamp_high ,
88+ timestamp_low ,
89+ captured_packet_len ,
90+ original_packet_len )
91+ epb += padded_packet
92+ epb += struct .pack (">I" , block_total_length )
93+ return epb
94+
95+ ############################################################################################################
96+ # PUBLIC INTERFACES
97+ def create_pcapng_file (self , filename : str , shb_userappl : str = "" , shb_hardware : str = "" , link_type : int = 147 ) -> bytes :
98+ if (filename == "" ):
99+ return
100+ self .__filename = filename
101+
102+ shb_bytes = self .__build_pcapng_shb (shb_userappl ,shb_hardware )
103+ idb_bytes = self .__build_pcapng_idb (link_type )
104+
105+ if os .path .isfile (self .__filename ):
106+ logger .warning ("{0} already exists. Overwriting it" .format (filename ))
107+
108+ with open (self .__filename , "wb" ) as f :
109+ f .write (shb_bytes )
110+ f .write (idb_bytes )
111+
112+ def pcapng_file_append_packet (self , packet : bytes , timestamp : float ):
113+ with open (self .__filename , "ab" ) as f :
114+ f .write (self .__build_pcapng_epb (packet , timestamp ))
115+
116+ def pcapng_file_append_multi_packets (self , packets : list ):
117+ with open (self .__filename , "ab" ) as f :
118+ for packet , timestamp in packets :
119+ f .write (self .__build_pcapng_epb (packet , timestamp ))
120+
0 commit comments