seb/2024-bsc-sebastian-lenzlinger
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

UNTESTED REFACTORING:

Browse Source 
Move more functionality into Metadata Model classes to ensure data is available and better passable between functions.
This commit is contained in:
Sebastian Lenzlinger
2024-05-07 18:58:54 +02:00
parent ec08bf71ab
commit 577ac9e5cf
13 changed files with 402 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files

137
code/iottb/subcommands/capture.py
View File

@@ -1,10 +1,10 @@
import subprocess
import uuid
from pathlib import Path
from iottb.definitions import *
from iottb.models.device_metadata_model import DeviceMetadata
from iottb.models.capture_metadata_model import CaptureMetadata
from iottb.models.device_metadata_model import DeviceMetadata, dir_contains_device_metadata
from iottb.utils.capture_utils import get_capture_src_folder, make_capture_src_folder
def setup_capture_parser(subparsers):
@@ -12,29 +12,31 @@ def setup_capture_parser(subparsers):
# metadata args
parser.add_argument("-a", "--ip-address", help="IP address of the device to sniff", dest="device_ip")
# tcpdump args
parser.add_argument("device-root", help="Root folder for device to sniff", dest="device_folder",
type=Path, required=True, default=Path.cwd())
parser.add_argument("-s", "--safe", help="Ensure correct device root folder before sniffing", action="store_true")
parser.add_argument("--app", help="Application name to sniff", dest="app_name", default=None)
parser_sniff_tcpdump = parser.add_argument_group('tcpdump arguments')
parser_sniff_tcpdump.add_argument("-i", "--interface", help="Interface to capture on.", dest="capture_interface",
default="any")
required=True)
parser_sniff_tcpdump.add_argument("-I", "--monitor-mode", help="Put interface into monitor mode",
action="store_true")
parser_sniff_tcpdump.add_argument("-n", help="Deactivate name resolution. Option is set by default.",
parser_sniff_tcpdump.add_argument("-n", help="Deactivate name resolution. True by default.",
action="store_true", dest="no_name_resolution")
parser_sniff_tcpdump.add_argument("-#", "--number",
help="Print packet number at beginning of line. Set by default.",
help="Print packet number at beginning of line. True by default.",
action="store_true")
parser_sniff_tcpdump.add_argument("-e", help="Print link layer headers. Option is set by default.",
parser_sniff_tcpdump.add_argument("-e", help="Print link layer headers. True by default.",
action="store_true", dest="print_link_layer")
parser_sniff_tcpdump.add_argument("-t", action="count", default=0,
help="Please see tcpdump manual for details. Unused by default.")
# parser_sniff_tcpdump.add_argument("--filter",type=str,default="ip help=f"pcap filter expression. \
# Defaults is '{default}'")
# shared args
cap_size_group = parser.add_mutually_exclusive_group(required=False)
cap_size_group.add_argument("-c", "--count", type=int, help="Number of packets to capture.", default=0)
cap_size_group.add_argument("--mins", type=int, help="Time in minutes to capture.", default=60)
cap_size_group.add_argument("-c", "--count", type=int, help="Number of packets to capture.", default=1000)
cap_size_group.add_argument("--mins", type=int, help="Time in minutes to capture.", default=1)
parser.set_defaults(func=handle_capture)
# return parser
# parser.add_default(func=handle_sniff(args=sniff_args))
def cwd_is_device_root_dir() -> bool:
@@ -42,9 +44,6 @@ def cwd_is_device_root_dir() -> bool:
return device_metadata_file.is_file()
def get_user_capture_config():
pass
def start_guided_device_root_dir_setup():
assert False, "Not implemented"
@@ -63,52 +62,58 @@ def handle_metadata():
return ReturnCodes.SUCCESS
def get_device_id_from_file(device_metadata_filename: Path) -> str:
def get_device_metadata_from_file(device_metadata_filename: Path) -> str:
assert device_metadata_filename.is_file(), f"Device metadata file '{device_metadata_filename} does not exist"
device_metadata = DeviceMetadata.load_from_json(device_metadata_filename)
device_id = device_metadata.device_id
return device_id
return device_metadata
def setup_capture_folder():
pass
def run_tcpdum(cmd):
def run_tcpdump(cmd):
# TODO: Maybe specify files for stout and stderr
try:
p = subprocess.run(cmd)
p = subprocess.run(cmd, capture_output=True, text=True, check=True)
if p.returncode != 0:
print(p.stderr)
else:
print(p.stdout)
except KeyboardInterrupt:
pass
def generate_capfile_name():
name = datetime.now().strftime("%Y%m%d_%H%M%S") + uuid.uuid4().hex
def handle_capture(args):
assert args.device_root is not None, f"Device root directory is required"
device_metadata_file = Path.cwd() / DEVICE_METADATA_FILE
device_id = get_device_id_from_file(device_metadata_file)
assert device_metadata_file.is_file(), f"Device metadata file '{device_metadata_file} does not exist"
capture_dir = setup_capture_folder(args.dev_root)
assert dir_contains_device_metadata(args.device_root)
# get device metadata
if args.safe and not dir_contains_device_metadata(args.device_root):
print(f"Supplied folder contains no device metadata. "
f"Please setup a device root directory before using this command")
exit(ReturnCodes.ABORTED)
elif dir_contains_device_metadata(args.device_root):
device_metadata_filename = args.device_root / DEVICE_METADATA_FILE
device_data = DeviceMetadata.load_from_json(device_metadata_filename)
else:
name = input("Please enter a device name: ")
args.device_root.mkdir(parents=True, exist_ok=True)
device_data = DeviceMetadata(name, args.device_root)
# start constructing environment for capture
capture_dir = get_capture_src_folder(args.device_root)
make_capture_src_folder(capture_dir)
capture_metadata = CaptureMetadata(device_data, capture_dir)
capture_metadata.set_interface(args.capture_interface)
cmd = ['sudo', 'tcpdump', '-i', args.capture_interface]
if args.monitor_mode:
cmd.append('-I')
if args.no_name_resolution:
cmd.append('-n')
if args.number:
cmd.append('-#')
if args.print_link_layer:
cmd.append('-e')
if args.count:
cmd.append('-c')
cmd.append(str(args.count))
elif args.mins:
pass
cmd = build_tcpdump_args(args, cmd, capture_metadata)
capture_metadata.set_tcpdump_command(cmd)
print('Executing: ' + ' '.join(cmd))
# TODO maybe dump this into file -> put into device metadata
# run capture
try:
start_time = datetime.now().strftime('%H:%M:%S')
run_tcpdum(cmd)
run_tcpdump(cmd)
stop_time = datetime.now().strftime('%H:%M:%S')
capture_metadata.set_start_time(start_time)
capture_metadata.set_stop_time(stop_time)
except KeyboardInterrupt:
print("Received keyboard interrupt.")
exit(ReturnCodes.ABORTED)
@@ -122,4 +127,44 @@ def handle_capture(args):
return ReturnCodes.SUCCESS
def build_tcpdump_args(args, cmd, capture_metadata: CaptureMetadata):
if args.monitor_mode:
cmd.append('-I')
if args.no_name_resolution:
cmd.append('-n')
if args.number:
cmd.append('-#')
if args.print_link_layer:
cmd.append('-e')
if args.count:
cmd.append('-c')
cmd.append(str(args.count))
elif args.mins:
assert False, "Unimplemented option"
if args.app_name is not None:
capture_metadata.set_app_name(args.app_name)
capture_metadata.build_capture_file_name()
cmd.append('-w')
cmd.append(capture_metadata.get_capfile_name())
if args.safe:
cmd.append(f'host {args.device_ip}') # if not specified, filter 'any' implied by tcpdump
capture_metadata.set_device_ip_address(args.device_ip)
return cmd
# def capture_file_cmd(args, cmd, capture_dir, capture_metadata: CaptureMetadata):
# capture_file_prefix = capture_metadata.get_device_metadata().get_device_short_name()
# if args.app_name is not None:
# capture_file_prefix = args.app_name
# capture_metadata.set_app(args.app_name)
# capfile_name = capture_file_prefix + "_" + str(capture_metadata.get_capture_id()) + ".pcap"
# capture_metadata.set_capture_file(capfile_name)
# capfile_abs_path = capture_dir / capfile_name
# capture_metadata.set_capture_file(capfile_name)
# cmd.append('-w')
# cmd.append(str(capfile_abs_path))

0
code/iottb/subcommands/config.py
View File