From f7d9b962265555d617eda69fd1cd901117620a85 Mon Sep 17 00:00:00 2001 From: joelsernamoreno Date: Tue, 1 Apr 2025 23:22:32 +0200 Subject: [PATCH 1/3] Named pipe and tshark support --- src/main.py | 10 +++++- src/modules/pcap_dump.py | 75 +++++++++++++++++++++++++--------------- 2 files changed, 57 insertions(+), 28 deletions(-) diff --git a/src/main.py b/src/main.py index 959b2da..631fef8 100644 --- a/src/main.py +++ b/src/main.py @@ -65,6 +65,8 @@ def main(): modules.add_argument('--info', action = 'store_true', help = 'Read generic information about the baseband device.') modules.add_argument('--pcap-dump', metavar = 'PCAP_FILE', type = FileType('ab'), help = 'Generate a PCAP file containing GSMTAP frames for 2G/3G/4G, to be loaded using Wireshark.') + modules.add_argument('--tshark', action = 'store_true', help = 'Same as --pcap-dump, but directly spawn tshark.') + modules.add_argument('--analysis', action = 'store_true', help = 'Same as --pcap-dump, but send the output to a named pipe.') modules.add_argument('--wireshark-live', action = 'store_true', help = 'Same as --pcap-dump, but directly spawn a Wireshark instance.') # modules.add_argument('--efs-dump', metavar = 'OUTPUT_DIR', help = 'Dump the internal EFS filesystem of the device.') modules.add_argument('--memory-dump', metavar = 'OUTPUT_DIR', help = 'Dump the memory of the device (may not or partially work with recent devices).') @@ -73,7 +75,7 @@ def main(): 'To be used in combination with --adb.') modules.add_argument('--decoded-sibs-dump', action = 'store_true', help = 'Print decoded SIBs to stdout (experimental, requires pycrate).') - pcap_options = parser.add_argument_group(title = 'PCAP generation options', description = 'To be used along with --pcap-dump or --wireshark-live.') + pcap_options = parser.add_argument_group(title = 'PCAP generation options', description = 'To be used along with --pcap-dump, --wireshark-live, --tshark or --analysis.') pcap_options.add_argument('--reassemble-sibs', action = 'store_true', help = 'Include reassembled UMTS SIBs as supplementary frames, also embedded fragmented in RRC frames.') pcap_options.add_argument('--decrypt-nas', action = 'store_true', help = 'Include unencrypted LTE NAS as supplementary frames, also embedded ciphered in RRC frames.') @@ -154,6 +156,12 @@ def parse_modules_args(args): if args.wireshark_live: from .modules.pcap_dump import WiresharkLive diag_input.add_module(WiresharkLive(diag_input, args.reassemble_sibs, args.decrypt_nas, args.include_ip_traffic)) + if args.analysis and not args.tshark: + from .modules.pcap_dump import ExternalAnalysis + diag_input.add_module(ExternalAnalysis(diag_input, args.reassemble_sibs, args.decrypt_nas, args.include_ip_traffic)) + if args.tshark and not args.analysis: + from .modules.pcap_dump import TsharkLive + diag_input.add_module(TsharkLive(diag_input, args.reassemble_sibs, args.decrypt_nas, args.include_ip_traffic)) if args.json_geo_dump: diag_input.add_module(JsonGeoDumper(diag_input, args.json_geo_dump)) if args.decoded_sibs_dump: diff --git a/src/modules/pcap_dump.py b/src/modules/pcap_dump.py index f41f214..19746fb 100644 --- a/src/modules/pcap_dump.py +++ b/src/modules/pcap_dump.py @@ -9,6 +9,7 @@ from logging import warning from sys import platform import gzip +import os from ..modules._enable_log_mixin import EnableLogMixin, TYPES_FOR_RAW_PACKET_LOGGING from ..modules.decoded_sibs_dump import DecodedSibsDumper @@ -506,68 +507,88 @@ def __del__(self): if getattr(self, 'pcap_file', None): self.pcap_file.close() +class ExternalAnalysis(PcapDumper): + def __init__(self, diag_input, reassemble_sibs, decrypt_nas, include_ip_traffic): + tshark = which('tshark') + if not tshark: + raise Exception('Could not find Tshark in $PATH') -""" - This is the same module, except that il will launch directly a FIFO to - Wireshark rather than write the PCAP to a file -""" + pipe_path = '/tmp/tshark_pipe' + if not os.path.exists(pipe_path): + os.mkfifo(pipe_path) -class WiresharkLive(PcapDumper): + tshark_pipe = Popen([tshark, '-i', '-'], stdin=PIPE, stdout=open(pipe_path, 'w')) + tshark_pipe.stdin.appending_to_file = False + + super().__init__(diag_input, tshark_pipe.stdin, reassemble_sibs, decrypt_nas, include_ip_traffic) + + def __del__(self): + self.tshark_pipe.stdin.close() + self.tshark_pipe.wait() +class TsharkLive(PcapDumper): def __init__(self, diag_input, reassemble_sibs, decrypt_nas, include_ip_traffic): - + tshark = which('tshark') + if not tshark: + raise Exception('Could not find Tshark in $PATH') + + tshark_pipe = Popen([tshark, '-i', '-'], stdin=PIPE) + tshark_pipe.stdin.appending_to_file = False + + super().__init__(diag_input, tshark_pipe.stdin, reassemble_sibs, decrypt_nas, include_ip_traffic) + + def __del__(self): + self.tshark_pipe.stdin.close() + self.tshark_pipe.wait() + +class WiresharkLive(PcapDumper): + def __init__(self, diag_input, reassemble_sibs, decrypt_nas, include_ip_traffic): + wireshark = ( which('C:\\Program Files\\Wireshark\\Wireshark.exe') or which('C:\\Program Files (x86)\\Wireshark\\Wireshark.exe') or which('wireshark') or which('wireshark-gtk') ) - + if not wireshark: - + raise Exception('Could not find Wireshark in $PATH') - + if not IS_UNIX: - + self.detach_process = None - + wireshark_pipe = Popen([wireshark, '-k', '-i', '-'], stdin = PIPE, stdout = DEVNULL, stderr = STDOUT, preexec_fn = self.detach_process, bufsize = 0 ).stdin - + wireshark_pipe.appending_to_file = False - + super().__init__(diag_input, wireshark_pipe, reassemble_sibs, decrypt_nas, include_ip_traffic) - + """ Executed when we launch a Wireshark process, after fork() """ - def detach_process(self): - + try: - + # Don't be hit by CTRL+C - setpgrp() - + # Drop privileges if needed - uid, gid = getenv('SUDO_UID'), getenv('SUDO_GID') - + if uid and gid: - - uid, gid = int(uid), int(gid) + uid, gid = int(uid), int(gid) setgroups(getgrouplist(getpwuid(uid).pw_name, gid)) - setresgid(gid, gid, -1) - setresuid(uid, uid, -1) - + except Exception: - print_exc() From 636d8c5420e6335edbc3ee9c47a482c232b12001 Mon Sep 17 00:00:00 2001 From: joelsernamoreno Date: Tue, 1 Apr 2025 23:39:58 +0200 Subject: [PATCH 2/3] Named pipe and tshark support --- processor.py | 18 ++++++++++++++++++ src/main.py | 12 ++++++------ src/modules/pcap_dump.py | 2 +- 3 files changed, 25 insertions(+), 7 deletions(-) create mode 100644 processor.py diff --git a/processor.py b/processor.py new file mode 100644 index 0000000..6f0b564 --- /dev/null +++ b/processor.py @@ -0,0 +1,18 @@ +import os + +def read_from_pipe(pipe_path='/tmp/tshark_pipe'): + if not os.path.exists(pipe_path): + print(f"Error: The named pipe {pipe_path} does not exist. Make sure that TsharkLive has created it.") + return + + try: + with open(pipe_path, 'r') as pipe: + for line in pipe: + print(f"{line.strip()}") + except KeyboardInterrupt: + print("\n") + except Exception as e: + print(f"Error: {e}") + +if __name__ == "__main__": + read_from_pipe() diff --git a/src/main.py b/src/main.py index 631fef8..4b0ad43 100644 --- a/src/main.py +++ b/src/main.py @@ -66,7 +66,7 @@ def main(): modules.add_argument('--info', action = 'store_true', help = 'Read generic information about the baseband device.') modules.add_argument('--pcap-dump', metavar = 'PCAP_FILE', type = FileType('ab'), help = 'Generate a PCAP file containing GSMTAP frames for 2G/3G/4G, to be loaded using Wireshark.') modules.add_argument('--tshark', action = 'store_true', help = 'Same as --pcap-dump, but directly spawn tshark.') - modules.add_argument('--analysis', action = 'store_true', help = 'Same as --pcap-dump, but send the output to a named pipe.') + modules.add_argument('--analyze', action = 'store_true', help = 'Same as --pcap-dump, but send the output to a named pipe.') modules.add_argument('--wireshark-live', action = 'store_true', help = 'Same as --pcap-dump, but directly spawn a Wireshark instance.') # modules.add_argument('--efs-dump', metavar = 'OUTPUT_DIR', help = 'Dump the internal EFS filesystem of the device.') modules.add_argument('--memory-dump', metavar = 'OUTPUT_DIR', help = 'Dump the memory of the device (may not or partially work with recent devices).') @@ -75,7 +75,7 @@ def main(): 'To be used in combination with --adb.') modules.add_argument('--decoded-sibs-dump', action = 'store_true', help = 'Print decoded SIBs to stdout (experimental, requires pycrate).') - pcap_options = parser.add_argument_group(title = 'PCAP generation options', description = 'To be used along with --pcap-dump, --wireshark-live, --tshark or --analysis.') + pcap_options = parser.add_argument_group(title = 'PCAP generation options', description = 'To be used along with --pcap-dump, --wireshark-live, --tshark or --analyze.') pcap_options.add_argument('--reassemble-sibs', action = 'store_true', help = 'Include reassembled UMTS SIBs as supplementary frames, also embedded fragmented in RRC frames.') pcap_options.add_argument('--decrypt-nas', action = 'store_true', help = 'Include unencrypted LTE NAS as supplementary frames, also embedded ciphered in RRC frames.') @@ -156,10 +156,10 @@ def parse_modules_args(args): if args.wireshark_live: from .modules.pcap_dump import WiresharkLive diag_input.add_module(WiresharkLive(diag_input, args.reassemble_sibs, args.decrypt_nas, args.include_ip_traffic)) - if args.analysis and not args.tshark: - from .modules.pcap_dump import ExternalAnalysis - diag_input.add_module(ExternalAnalysis(diag_input, args.reassemble_sibs, args.decrypt_nas, args.include_ip_traffic)) - if args.tshark and not args.analysis: + if args.analyze and not args.tshark: + from .modules.pcap_dump import ExternalAnalyze + diag_input.add_module(ExternalAnalyze(diag_input, args.reassemble_sibs, args.decrypt_nas, args.include_ip_traffic)) + if args.tshark and not args.analyze: from .modules.pcap_dump import TsharkLive diag_input.add_module(TsharkLive(diag_input, args.reassemble_sibs, args.decrypt_nas, args.include_ip_traffic)) if args.json_geo_dump: diff --git a/src/modules/pcap_dump.py b/src/modules/pcap_dump.py index 19746fb..0873b8d 100644 --- a/src/modules/pcap_dump.py +++ b/src/modules/pcap_dump.py @@ -507,7 +507,7 @@ def __del__(self): if getattr(self, 'pcap_file', None): self.pcap_file.close() -class ExternalAnalysis(PcapDumper): +class ExternalAnalyze(PcapDumper): def __init__(self, diag_input, reassemble_sibs, decrypt_nas, include_ip_traffic): tshark = which('tshark') if not tshark: From eccdd2b4a40f75edafa8ed62ca023dbd0a0eb4f9 Mon Sep 17 00:00:00 2001 From: joelsernamoreno Date: Sun, 6 Apr 2025 16:55:20 +0200 Subject: [PATCH 3/3] Add analyze --- src/modules/pcap_dump.py | 55 +++++++++++++++++++--------------------- 1 file changed, 26 insertions(+), 29 deletions(-) diff --git a/src/modules/pcap_dump.py b/src/modules/pcap_dump.py index 0873b8d..c17748f 100644 --- a/src/modules/pcap_dump.py +++ b/src/modules/pcap_dump.py @@ -9,15 +9,14 @@ from logging import warning from sys import platform import gzip +import sys import os - from ..modules._enable_log_mixin import EnableLogMixin, TYPES_FOR_RAW_PACKET_LOGGING from ..modules.decoded_sibs_dump import DecodedSibsDumper MODULES_DIR = realpath(dirname(__file__)) SRC_WIRESHARK_PLUGIN_DIR = realpath(MODULES_DIR + '/wireshark_plugin') - try: from os import setpgrp, getenv, setresgid, setresuid, setgroups, getgrouplist from pwd import getpwuid @@ -38,17 +37,17 @@ """ class PcapDumper(DecodedSibsDumper): - + def __init__(self, diag_input, pcap_file, reassemble_sibs, decrypt_nas, include_ip_traffic): - + self.pcap_file = pcap_file - + """ Write a PCAP file header - https://wiki.wireshark.org/Development/LibpcapFileFormat#File_Format """ - + if not self.pcap_file.appending_to_file: - + self.pcap_file.write(pack('