#!/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('> 32 timestamp_lo = timestamp_ns & 0xFFFFFFFF block_length = 4 + 4 + 4 + 8 + 4 + 4 + len(data_with_padding) + 4 fout.write(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 [-l] [-o ] Options: -h, --help display help information -t, --type= specify log type: HCI, MAC, or IP -l, --include-log Include LOG entries as separate interface -o, --output= 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()