119 lines
4.5 KiB
Python
119 lines
4.5 KiB
Python
import click
|
|
import subprocess
|
|
import json
|
|
from pathlib import Path
|
|
import logging
|
|
import re
|
|
from datetime import datetime
|
|
from iottb.definitions import APP_NAME, CFG_FILE_PATH
|
|
from iottb.models.iottb_config import IottbConfig
|
|
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
|
|
|
|
|
|
@click.command('sniff', help='Sniff packets with tcpdump')
|
|
@click.argument('device')
|
|
@click.option('-i', '--interface', callback=validate_sniff, help='Network interface to capture on',
|
|
envvar='IOTTB_CAPTURE_INTERFACE')
|
|
@click.option('-a', '--address', callback=validate_sniff, help='IP or MAC address to filter packets by',
|
|
envvar='IOTTB_CAPTURE_ADDRESS')
|
|
@click.option('--db', '--database', type=click.Path(exists=True, file_okay=False), envvar='IOTTB_DB',
|
|
help='Database of device. Only needed if not current default.')
|
|
@click.option('--unsafe', is_flag=True, default=False, envvar='IOTTB_UNSAFE', is_eager=True,
|
|
help='Disable checks for otherwise required options')
|
|
@click.option('--guided', is_flag=True, default=False)
|
|
def sniff(device, interface, address, db, unsafe, guided):
|
|
""" Sniff packets from a device """
|
|
logger.info('sniff command invoked')
|
|
|
|
# Step1: Load Config
|
|
config = IottbConfig(Path(CFG_FILE_PATH))
|
|
logger.debug(f'Config loaded: {config}')
|
|
|
|
# Step2: determine relevant database
|
|
database = db if db else config.default_database
|
|
path = config.default_db_location[database]
|
|
full_db_path = Path(path) / database
|
|
logger.debug(f'Full db path is {str(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')
|
|
|
|
# Generate filter
|
|
if not unsafe:
|
|
if is_ip_address(address):
|
|
packet_filter = f"host {address}"
|
|
elif is_mac_address(address):
|
|
packet_filter = f"ether host {address}"
|
|
else:
|
|
logger.error('Invalid address format')
|
|
click.echo('Invalid address format')
|
|
return
|
|
else:
|
|
packet_filter = None
|
|
|
|
|
|
|
|
|
|
@click.command('sniff', help='Sniff packets with tcpdump')
|
|
@click.argument('device')
|
|
@click.option('-i', '--interface', required=False, help='Network interface to capture on', envvar='IOTTB_CAPTURE_INTERFACE')
|
|
@click.option('-a', '--address', required=True, help='IP or MAC address to filter packets by', envvar='IOTTB_CAPTURE_ADDRESS')
|
|
@click.option('--db', '--database', type=click.Path(exists=True, file_okay=False), envvar='IOTTB_DB',
|
|
help='Database of device. Only needed if not current default.')
|
|
@click.option('--unsafe', is_flag=True, default=False, envvar='IOTTB_UNSAFE',
|
|
help='Disable checks for otherwise required options')
|
|
@click.option('--guided', is_flag=True)
|
|
def sniff2(device, interface, address, cfg_file):
|
|
""" Sniff packets from a device """
|
|
logger.info('sniff command invoked')
|
|
# Step 1: Load Config
|
|
# Dependency: Config file must exist
|
|
config = IottbConfig(Path(CFG_FILE_PATH))
|
|
logger.debug(f'Config loaded: {config}')
|