303 lines
9.4 KiB
Python
Executable File
303 lines
9.4 KiB
Python
Executable File
#!/usr/bin/python3
|
|
#coding:utf-8
|
|
#
|
|
# Copyright (c) 2021-2022 Amlogic, Inc. All rights reserved.
|
|
#
|
|
# SPDX-License-Identifier: MIT
|
|
#
|
|
|
|
import re
|
|
import sys
|
|
import getopt
|
|
import struct
|
|
import os
|
|
import time
|
|
|
|
VERSION = "20260302"
|
|
|
|
default_date = "2001-01-01"
|
|
default_hours = 12
|
|
packet_counter = 0
|
|
last_time = default_date + " " + str(default_hours) + ":00:00.000"
|
|
|
|
INTERFACE_MAIN = 0
|
|
INTERFACE_LOG = 1
|
|
|
|
LINK_TYPE_HCI = 201
|
|
LINK_TYPE_MAC = 1
|
|
LINK_TYPE_IP = 276
|
|
LINK_TYPE_LOG = 147
|
|
|
|
TYPE_HCI = "HCI"
|
|
TYPE_MAC = "MAC"
|
|
TYPE_IP = "IP"
|
|
|
|
def chop(line, prefix):
|
|
if line.startswith(prefix):
|
|
return line[len(prefix):]
|
|
return None
|
|
|
|
def str2hex(value):
|
|
if value:
|
|
return int(value, 16)
|
|
return None
|
|
|
|
def generateTimestamp(t):
|
|
global last_time
|
|
global packet_counter
|
|
|
|
if not t:
|
|
t = last_time
|
|
if t:
|
|
last_time = t
|
|
|
|
if t.isdigit():
|
|
tick_ms = int(t)
|
|
tv_sec = tick_ms // 1000
|
|
tv_usec = (tick_ms % 1000) * 1000
|
|
return (tv_sec, tv_usec)
|
|
|
|
parts = t.split(' ')
|
|
have_date = True
|
|
if len(parts) == 1:
|
|
have_date = False
|
|
t = "2000-01-01 " + t
|
|
|
|
try:
|
|
(t1, t2) = t.split('.')
|
|
if t1 and t2:
|
|
t_obj = time.strptime(t1, "%Y-%m-%d %H:%M:%S")
|
|
tv_sec = int(time.mktime(t_obj))
|
|
if not have_date:
|
|
tv_sec += 12*60*60
|
|
tv_usec = int(t2) * 1000
|
|
return (tv_sec, tv_usec)
|
|
except ValueError:
|
|
pass
|
|
|
|
packet_counter += 1
|
|
return (packet_counter, 0)
|
|
|
|
def write_pcapng_header(fout):
|
|
fout.write(b'\x0a\x0d\x0d\x0a')
|
|
block_length = 28
|
|
fout.write(struct.pack('<I', block_length))
|
|
fout.write(struct.pack('<I', 0x1a2b3c4d))
|
|
fout.write(struct.pack('<HH', 1, 0))
|
|
fout.write(struct.pack('<q', -1))
|
|
fout.write(struct.pack('<I', block_length))
|
|
|
|
def write_interface_description_block(fout, interface_id, link_type, snap_len=65535):
|
|
fout.write(struct.pack('<I', 0x00000001))
|
|
interface_desc_len = 8
|
|
block_length = 4 + 4 + interface_desc_len + 4
|
|
fout.write(struct.pack('<I', block_length))
|
|
fout.write(struct.pack('<HHI', link_type, 0, snap_len))
|
|
fout.write(struct.pack('<I', block_length))
|
|
|
|
def write_enhanced_packet_block(fout, timestamp, interface_id, data):
|
|
(tv_sec, tv_usec) = generateTimestamp(timestamp)
|
|
|
|
block_type = 0x00000006
|
|
original_len = len(data)
|
|
captured_len = len(data)
|
|
|
|
data_with_padding = data + b'\x00' * ((4 - len(data) % 4) % 4)
|
|
|
|
timestamp_ns = int(tv_sec) * 1000000 + int(tv_usec)
|
|
timestamp_hi = timestamp_ns >> 32
|
|
timestamp_lo = timestamp_ns & 0xFFFFFFFF
|
|
|
|
block_length = 4 + 4 + 4 + 8 + 4 + 4 + len(data_with_padding) + 4
|
|
|
|
fout.write(struct.pack('<I', block_type))
|
|
fout.write(struct.pack('<I', block_length))
|
|
fout.write(struct.pack('<I', interface_id))
|
|
fout.write(struct.pack('<II', timestamp_hi, timestamp_lo))
|
|
fout.write(struct.pack('<II', captured_len, original_len))
|
|
fout.write(data_with_padding)
|
|
fout.write(struct.pack('<I', block_length))
|
|
|
|
def handleHCIPacket(fout, timestamp, direction, hci_type, text):
|
|
try:
|
|
data = bytearray(list(map(str2hex, text.strip().split())))
|
|
# pack direction as 32-bit big-endian
|
|
dir_bytes = struct.pack('>I', int(direction))
|
|
full_packet = bytearray(dir_bytes) + bytearray([hci_type]) + data
|
|
write_enhanced_packet_block(fout, timestamp, INTERFACE_MAIN, full_packet)
|
|
except TypeError:
|
|
print('Cannot parse hexdump', text.strip())
|
|
|
|
def handleMACPacket(fout, timestamp, text):
|
|
try:
|
|
data = bytearray(list(map(str2hex, text.strip().split())))
|
|
write_enhanced_packet_block(fout, timestamp, INTERFACE_MAIN, data)
|
|
except TypeError:
|
|
print('Cannot parse hexdump', text.strip())
|
|
|
|
def handleIPPacket(fout, timestamp, text):
|
|
try:
|
|
data = bytearray(list(map(str2hex, text.strip().split())))
|
|
write_enhanced_packet_block(fout, timestamp, INTERFACE_MAIN, data)
|
|
except TypeError:
|
|
print('Cannot parse hexdump', text.strip())
|
|
|
|
def handleLogPacket(fout, timestamp, text):
|
|
try:
|
|
log_data = text.encode('utf-8', errors='replace')
|
|
write_enhanced_packet_block(fout, timestamp, INTERFACE_LOG, log_data)
|
|
except:
|
|
pass
|
|
|
|
def usage():
|
|
print("""
|
|
The script is used to generate PCAPNG files from HCI/MAC/IP log files.
|
|
Usage: python3 log_to_pcap.py -t <type> [-l] [-o <output>] <input-file>
|
|
|
|
Options:
|
|
-h, --help display help information
|
|
-t, --type=<type> specify log type: HCI, MAC, or IP
|
|
-l, --include-log Include LOG entries as separate interface
|
|
-o, --output=<file> specify output file (default: input_TYPE.pcapng)
|
|
-v, --version version information
|
|
|
|
Examples:
|
|
python3 log_to_pcap.py -t HCI -l -o output.pcapng input.txt
|
|
python3 log_to_pcap.py -t MAC input.txt
|
|
python3 log_to_pcap.py -t IP -l input.txt
|
|
""")
|
|
|
|
def main():
|
|
if len(sys.argv) == 1:
|
|
usage()
|
|
sys.exit(1)
|
|
|
|
try:
|
|
opts, args = getopt.getopt(sys.argv[1:], "ht:o:lv", ["help", "type=", "include-log", "version", "output="])
|
|
except getopt.GetoptError as err:
|
|
print(err)
|
|
usage()
|
|
sys.exit(2)
|
|
|
|
log_type = None
|
|
include_log = False
|
|
outfile = None
|
|
|
|
for opt, arg in opts:
|
|
if opt in ("-h", "--help"):
|
|
usage()
|
|
sys.exit()
|
|
elif opt in ("-t", "--type"):
|
|
log_type = arg.upper()
|
|
elif opt in ("-o", "--output"):
|
|
outfile = arg
|
|
elif opt in ("-l", "--include-log"):
|
|
include_log = True
|
|
elif opt in ("-v", "--version"):
|
|
print(VERSION)
|
|
sys.exit()
|
|
else:
|
|
print("Using the wrong way, please refer the help information!\n")
|
|
usage()
|
|
sys.exit(2)
|
|
|
|
if log_type not in [TYPE_HCI, TYPE_MAC, TYPE_IP]:
|
|
print("Error: -t/--type must be HCI, MAC, or IP\n")
|
|
usage()
|
|
sys.exit(2)
|
|
|
|
if len(args) < 1:
|
|
print("Error: input file is required\n")
|
|
usage()
|
|
sys.exit(2)
|
|
|
|
infile = args[0]
|
|
if outfile is None:
|
|
outfile = os.path.splitext(infile)[0] + "_" + log_type + ".pcapng"
|
|
if len(args) >= 2 and outfile is None:
|
|
outfile = args[1]
|
|
|
|
if log_type == TYPE_HCI:
|
|
link_type = LINK_TYPE_HCI
|
|
elif log_type == TYPE_MAC:
|
|
link_type = LINK_TYPE_MAC
|
|
elif log_type == TYPE_IP:
|
|
link_type = LINK_TYPE_IP
|
|
else:
|
|
link_type = LINK_TYPE_HCI
|
|
|
|
with open(outfile, 'wb') as fout:
|
|
write_pcapng_header(fout)
|
|
write_interface_description_block(fout, INTERFACE_MAIN, link_type=link_type, snap_len=65535)
|
|
|
|
if include_log:
|
|
write_interface_description_block(fout, INTERFACE_LOG, link_type=LINK_TYPE_LOG, snap_len=65535)
|
|
|
|
packet_counter = 0
|
|
line_counter = 0
|
|
|
|
with open(infile, 'rt', encoding='utf-8', errors='replace') as fin:
|
|
for line in fin:
|
|
try:
|
|
line = ''.join([c for c in line if c != '\0'])
|
|
line_counter += 1
|
|
timestamp = None
|
|
line = line.strip("\n\r")
|
|
if len(line) == 0:
|
|
continue
|
|
|
|
parts = re.match(r'\[(.*)\] (.*)', line)
|
|
if parts and len(parts.groups()) == 2:
|
|
(timestamp, line) = parts.groups()
|
|
|
|
if log_type == TYPE_HCI:
|
|
rest = chop(line, 'CMD=>')
|
|
if rest:
|
|
handleHCIPacket(fout, timestamp, 0, 1, rest)
|
|
continue
|
|
rest = chop(line, 'EVT<=')
|
|
if rest:
|
|
handleHCIPacket(fout, timestamp, 1, 4, rest)
|
|
continue
|
|
rest = chop(line, 'ACL=>')
|
|
if rest:
|
|
handleHCIPacket(fout, timestamp, 0, 2, rest)
|
|
continue
|
|
rest = chop(line, 'ACL<=')
|
|
if rest:
|
|
handleHCIPacket(fout, timestamp, 1, 2, rest)
|
|
continue
|
|
rest = chop(line, 'SCO=>')
|
|
if rest:
|
|
handleHCIPacket(fout, timestamp, 0, 3, rest)
|
|
continue
|
|
rest = chop(line, 'SCO<=')
|
|
if rest:
|
|
handleHCIPacket(fout, timestamp, 1, 3, rest)
|
|
continue
|
|
|
|
elif log_type == TYPE_MAC:
|
|
rest = chop(line, 'MAC--')
|
|
if rest:
|
|
handleMACPacket(fout, timestamp, rest)
|
|
continue
|
|
|
|
elif log_type == TYPE_IP:
|
|
rest = chop(line, 'IP--')
|
|
if rest:
|
|
handleIPPacket(fout, timestamp, rest)
|
|
continue
|
|
|
|
rest = chop(line, 'LOG--')
|
|
if rest:
|
|
if include_log:
|
|
handleLogPacket(fout, timestamp, rest)
|
|
continue
|
|
except:
|
|
print("Error in line %u: '%s'" % (line_counter, line))
|
|
|
|
print("\nPCAPNG file: %s" % outfile)
|
|
|
|
if __name__ == '__main__':
|
|
main()
|