340 lines
13 KiB
Python
340 lines
13 KiB
Python
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import uuid
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from time import time
|
|
|
|
import click
|
|
from click_option_group import optgroup
|
|
|
|
from iottb.utils.string_processing import make_canonical_name
|
|
|
|
# Setup logger
|
|
logger = logging.getLogger('iottb.sniff')
|
|
|
|
|
|
def is_ip_address(address):
|
|
ip_pattern = re.compile(r"^(?:[0-9]{1,3}\.){3}[0-9]{1,3}$")
|
|
return ip_pattern.match(address) is not None
|
|
|
|
|
|
def is_mac_address(address):
|
|
mac_pattern = re.compile(r"^([0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2}$")
|
|
return mac_pattern.match(address) is not None
|
|
|
|
|
|
def load_config(cfg_file):
|
|
"""Loads configuration from the given file path."""
|
|
with open(cfg_file, 'r') as config_file:
|
|
return json.load(config_file)
|
|
|
|
|
|
def validate_sniff(ctx, param, value):
|
|
logger.info('Validating sniff...')
|
|
if ctx.params.get('unsafe') and not value:
|
|
return None
|
|
if not ctx.params.get('unsafe') and not value:
|
|
raise click.BadParameter('Address is required unless --unsafe is set.')
|
|
if not is_ip_address(value) and not is_mac_address(value):
|
|
raise click.BadParameter('Address must be a valid IP address or MAC address.')
|
|
return value
|
|
|
|
|
|
def run_pre(pre):
|
|
pass
|
|
|
|
|
|
def run_post(post):
|
|
pass
|
|
|
|
|
|
@click.command('sniff', help='Sniff packets with tcpdump')
|
|
@optgroup.group('Testbed sources')
|
|
@optgroup.option('--db', '--database', type=str, envvar='IOTTB_DB', show_envvar=True,
|
|
help='Database of device. Only needed if not current default.')
|
|
@optgroup.option('--app', type=str, help='Companion app being used during capture', required=False)
|
|
@optgroup.group('Runtime behaviour')
|
|
@optgroup.option('--unsafe', is_flag=True, default=False, envvar='IOTTB_UNSAFE', is_eager=True,
|
|
help='Disable checks for otherwise required options.\n', show_envvar=True)
|
|
@optgroup.option('--guided', is_flag=True, default=False, envvar='IOTTB_GUIDED', show_envvar=True)
|
|
@optgroup.option('--pre', type=click.Path(exists=True, executable=True), help='Script to be executed before main '
|
|
'command'
|
|
'is started.')
|
|
@optgroup.group('Tcpdump options')
|
|
@optgroup.option('-i', '--interface',
|
|
help='Network interface to capture on.' +
|
|
'If not specified tcpdump tries to find and appropriate one.\n', show_envvar=True,
|
|
envvar='IOTTB_CAPTURE_INTERFACE')
|
|
@optgroup.option('-a', '--address', callback=validate_sniff,
|
|
help='IP or MAC address to filter packets by.\n', show_envvar=True,
|
|
envvar='IOTTB_CAPTURE_ADDRESS')
|
|
@optgroup.option('-I', '--monitor-mode', help='Put interface into monitor mode.', is_flag=True)
|
|
@optgroup.option('--ff', type=str, envvar='IOTTB_CAPTURE_FILTER', show_envvar=True,
|
|
help='tcpdump filter as string or file path.')
|
|
@optgroup.option('-#', '--print-pacno', is_flag=True, default=True,
|
|
help='Print packet number at beginning of line. True by default.')
|
|
@optgroup.option('-e', '--print-ll', is_flag=True, default=False,
|
|
help='Print link layer headers. True by default.')
|
|
@optgroup.option('-c', '--count', type=int, help='Number of packets to capture.', default=1000)
|
|
# @optgroup.option('--mins', type=int, help='Time in minutes to capture.', default=1)
|
|
@click.argument('tcpdump-args', nargs=-1, required=False, metavar='[TCPDUMP-ARGS]')
|
|
@click.argument('device', required=False)
|
|
@click.pass_context
|
|
def sniff(ctx, device, interface, print_pacno, ff, count, monitor_mode, print_ll, address, db, unsafe, guided,
|
|
app, tcpdump_args, pre, post, **params):
|
|
""" Sniff packets from a device """
|
|
logger.info('sniff command invoked')
|
|
# Step 0: run pre script:
|
|
if pre:
|
|
click.echo(f'Running pre command {pre}')
|
|
run_pre(pre)
|
|
# Step1: Load Config
|
|
config = ctx.obj['CONFIG']
|
|
logger.debug(f'Config loaded: {config}')
|
|
|
|
# Step2: determine relevant database
|
|
database = db if db else config.default_database
|
|
path = config.db_path_dict[database]
|
|
full_db_path = Path(path) / database
|
|
logger.debug(f'Full db path is {str(full_db_path)}')
|
|
|
|
# 2.2: Check if it exists
|
|
if not full_db_path.is_dir():
|
|
logger.error('DB unexpectedly missing')
|
|
click.echo('DB unexpectedly missing')
|
|
return
|
|
|
|
canonical_name, aliases = make_canonical_name(device)
|
|
click.echo(f'Using canonical device name {canonical_name}')
|
|
device_path = full_db_path / canonical_name
|
|
|
|
# Step 3: now the device
|
|
if not device_path.exists():
|
|
if not unsafe:
|
|
logger.error(f'Device path {device_path} does not exist')
|
|
click.echo(f'Device path {device_path} does not exist')
|
|
return
|
|
else:
|
|
device_path.mkdir(parents=True, exist_ok=True)
|
|
logger.info(f'Device path {device_path} created')
|
|
|
|
click.echo(f'Found device at path {device_path}')
|
|
# Step 4: Generate filter
|
|
generic_filter = None
|
|
cap_filter = None
|
|
if ff:
|
|
logger.debug(f'ff: {ff}')
|
|
if Path(ff).is_file():
|
|
logger.info('Given filter option is a file')
|
|
with open(ff, 'r') as f:
|
|
cap_filter = f.read().strip()
|
|
else:
|
|
logger.info('Given filter option is an expression')
|
|
cap_filter = ff
|
|
else:
|
|
if address is not None:
|
|
if is_ip_address(address):
|
|
generic_filter = 'net'
|
|
cap_filter = f'{generic_filter} {address}'
|
|
elif is_mac_address(address):
|
|
generic_filter = 'ether net'
|
|
cap_filter = f'{generic_filter} {address}'
|
|
elif not unsafe:
|
|
logger.error('Invalid address format')
|
|
click.echo('Invalid address format')
|
|
return
|
|
|
|
logger.info(f'Generic filter {generic_filter}')
|
|
click.echo(f'Using filter {cap_filter}')
|
|
|
|
# Step 5: prep capture directory
|
|
capture_date = datetime.now().strftime('%Y-%m-%d')
|
|
capture_base_dir = device_path / f'sniffs/{capture_date}'
|
|
capture_base_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
logger.debug(f'Previous captures {capture_base_dir.glob('cap*')}')
|
|
capture_count = sum(1 for _ in capture_base_dir.glob('cap*'))
|
|
logger.debug(f'Capture count is {capture_count}')
|
|
|
|
capture_dir = f'cap{capture_count:04d}-{datetime.now().strftime('%H%M')}'
|
|
logger.debug(f'capture_dir: {capture_dir}')
|
|
|
|
# Full path
|
|
capture_dir_full_path = capture_base_dir / capture_dir
|
|
capture_dir_full_path.mkdir(parents=True, exist_ok=True)
|
|
|
|
click.echo(f'Files will be placed in {str(capture_dir_full_path)}')
|
|
logger.debug(f'successfully created capture directory')
|
|
|
|
# Step 6: Prepare capture file names
|
|
# Generate UUID for filenames
|
|
capture_uuid = str(uuid.uuid4())
|
|
click.echo(f'Capture has id {capture_uuid}')
|
|
|
|
pcap_file = f"{canonical_name}_{capture_uuid}.pcap"
|
|
pcap_file_full_path = capture_dir_full_path / pcap_file
|
|
stdout_log_file = f'stdout_{capture_uuid}.log'
|
|
stderr_log_file = f'stderr_{capture_uuid}.log'
|
|
|
|
logger.debug(f'Full pcap file path is {pcap_file_full_path}')
|
|
logger.info(f'pcap file name is {pcap_file}')
|
|
logger.info(f'stdout log file is {stdout_log_file}')
|
|
logger.info(f'stderr log file is {stderr_log_file}')
|
|
|
|
# Step 7: Build tcpdump command
|
|
logger.debug(f'pgid {os.getpgrp()}')
|
|
logger.debug(f'ppid {os.getppid()}')
|
|
logger.debug(f'(real, effective, saved) user id: {os.getresuid()}')
|
|
logger.debug(f'(real, effective, saved) group id: {os.getresgid()}')
|
|
|
|
cmd = ['sudo', 'tcpdump']
|
|
|
|
# 7.1 process flags
|
|
flags = []
|
|
if print_pacno:
|
|
flags.append('-#')
|
|
if print_ll:
|
|
flags.append('-e')
|
|
if monitor_mode:
|
|
flags.append('-I')
|
|
flags.append('-n') # TODO: Integrate, in case name resolution is wanted!
|
|
cmd.extend(flags)
|
|
flags_string = " ".join(flags)
|
|
logger.debug(f'Flags: {flags_string}')
|
|
|
|
# debug interlude
|
|
verbosity = ctx.obj['VERBOSITY']
|
|
if verbosity > 0:
|
|
verbosity_flag = '-'
|
|
for i in range(0, verbosity):
|
|
verbosity_flag = verbosity_flag + 'v'
|
|
logger.debug(f'verbosity string to pass to tcpdump: {verbosity_flag}')
|
|
cmd.append(verbosity_flag)
|
|
|
|
# 7.2 generic (i.e. reusable) kw args
|
|
generic_kw_args = []
|
|
if count:
|
|
generic_kw_args.extend(['-c', str(count)])
|
|
# if mins:
|
|
# generic_kw_args.extend(['-G', str(mins * 60)]) TODO: this currently loads to errors with sudo
|
|
cmd.extend(generic_kw_args)
|
|
generic_kw_args_string = " ".join(generic_kw_args)
|
|
logger.debug(f'KW args: {generic_kw_args_string}')
|
|
|
|
# 7.3 special kw args (not a priori reusable)
|
|
non_generic_kw_args = []
|
|
if interface:
|
|
non_generic_kw_args.extend(['-i', interface])
|
|
non_generic_kw_args.extend(['-w', str(pcap_file_full_path)])
|
|
cmd.extend(non_generic_kw_args)
|
|
non_generic_kw_args_string = " ".join(non_generic_kw_args)
|
|
logger.debug(f'Non transferable (special) kw args: {non_generic_kw_args_string}')
|
|
|
|
# 7.4 add filter expression
|
|
if cap_filter:
|
|
logger.debug(f'cap_filter (not generic): {cap_filter}')
|
|
cmd.append(cap_filter)
|
|
|
|
full_cmd_string = " ".join(cmd)
|
|
|
|
logger.info(f'tcpdump command: {"".join(full_cmd_string)}')
|
|
click.echo('Capture setup complete!')
|
|
# Step 8: Execute tcpdump command
|
|
start_time = datetime.now().strftime("%H:%M:%S")
|
|
start = time()
|
|
try:
|
|
if guided:
|
|
click.confirm(f'Execute following command: {full_cmd_string}')
|
|
stdout_log_file_abs_path = capture_dir_full_path / stdout_log_file
|
|
stderr_log_file_abs_path = capture_dir_full_path / stderr_log_file
|
|
stdout_log_file_abs_path.touch(mode=0o777)
|
|
stderr_log_file_abs_path.touch(mode=0o777)
|
|
with open(stdout_log_file_abs_path, 'w') as out, open(stderr_log_file_abs_path, 'w') as err:
|
|
logger.debug(f'\nstdout: {out}.\nstderr: {err}.\n')
|
|
|
|
tcp_complete = subprocess.run(cmd, check=True, capture_output=True, text=True)
|
|
|
|
out.write(tcp_complete.stdout)
|
|
err.write(tcp_complete.stderr)
|
|
|
|
# click.echo(f'Mock sniff execution')
|
|
click.echo(f"Capture complete. Saved to {pcap_file}")
|
|
except subprocess.CalledProcessError as e:
|
|
logger.error(f'Failed to capture packets: {e}')
|
|
click.echo(f'Failed to capture packets: {e}')
|
|
click.echo(f'Check {stderr_log_file} for more info.')
|
|
if ctx.obj['DEBUG']:
|
|
msg = [f'STDERR log {stderr_log_file} contents:\n']
|
|
with open(capture_dir_full_path / stderr_log_file) as log:
|
|
for line in log:
|
|
msg.append(line)
|
|
|
|
click.echo("\t".join(msg), lvl='e')
|
|
# print('DEBUG ACTIVE')
|
|
if guided:
|
|
click.prompt('Create metadata anyway?')
|
|
else:
|
|
click.echo('Aborting capture...')
|
|
exit()
|
|
end_time = datetime.now().strftime("%H:%M:%S")
|
|
end = time()
|
|
delta = end - start
|
|
click.echo(f'tcpdump took {delta:.2f} seconds.')
|
|
# Step 9: Register metadata
|
|
metadata = {
|
|
'device': canonical_name,
|
|
'device_id': device,
|
|
'capture_id': capture_uuid,
|
|
'capture_date_iso': datetime.now().isoformat(),
|
|
'invoked_command': " ".join(map(str, cmd)),
|
|
'capture_duration': delta,
|
|
'generic_parameters': {
|
|
'flags': flags_string,
|
|
'kwargs': generic_kw_args_string,
|
|
'filter': generic_filter
|
|
},
|
|
'non_generic_parameters': {
|
|
'kwargs': non_generic_kw_args_string,
|
|
'filter': cap_filter
|
|
},
|
|
'features': {
|
|
'interface': interface,
|
|
'address': address
|
|
},
|
|
'resources': {
|
|
'pcap_file': str(pcap_file),
|
|
'stdout_log': str(stdout_log_file),
|
|
'stderr_log': str(stderr_log_file)
|
|
},
|
|
'environment': {
|
|
'capture_dir': capture_dir,
|
|
'database': database,
|
|
'capture_base_dir': str(capture_base_dir),
|
|
'capture_dir_abs_path': str(capture_dir_full_path)
|
|
}
|
|
}
|
|
|
|
click.echo('Ensuring correct ownership of created files.')
|
|
username = os.getlogin()
|
|
gid = os.getgid()
|
|
|
|
# Else there are issues when running with sudo:
|
|
try:
|
|
subprocess.run(f'sudo chown -R {username}:{username} {device_path}', shell=True)
|
|
except OSError as e:
|
|
click.echo(f'Some error {e}')
|
|
|
|
click.echo(f'Saving metadata.')
|
|
metadata_abs_path = capture_dir_full_path / 'capture_metadata.json'
|
|
with open(metadata_abs_path, 'w') as f:
|
|
json.dump(metadata, f, indent=4)
|
|
|
|
click.echo(f'END SNIFF SUBCOMMAND')
|
|
if post:
|
|
click.echo(f'Running post command {post}')
|
|
run_post(post)
|