Replace all double quotes strings with single quoted strings.
This commit is contained in:
Sebastian Lenzlinger
parent
266a669e5e
commit
e569eb3e5b
@ -2,31 +2,31 @@ def setup_sniff_tcpdump_parser(parser_sniff):
|
||||
# arguments which will be passed to tcpdump
|
||||
parser_sniff_tcpdump = parser_sniff.add_argument_group('tcpdump arguments')
|
||||
# TODO: tcpdump_parser.add_argument('-c', '--count', re)
|
||||
parser_sniff_tcpdump.add_argument("-a", "--ip-address=", help="IP address of the device to sniff", dest="device_ip")
|
||||
parser_sniff_tcpdump.add_argument("-i", "--interface=", help="Interface of the capture device.", dest="capture_interface",default="")
|
||||
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.",
|
||||
action="store_true")
|
||||
parser_sniff_tcpdump.add_argument("-#", "--number",
|
||||
help="Print packet number at beginning of line. Set by default.",
|
||||
action="store_true")
|
||||
parser_sniff_tcpdump.add_argument("-e", help="Print link layer headers. Option is set by default.",
|
||||
action="store_true")
|
||||
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('-a', '--ip-address=', help='IP address of the device to sniff', dest='device_ip')
|
||||
parser_sniff_tcpdump.add_argument('-i', '--interface=', help='Interface of the capture device.', dest='capture_interface',default='')
|
||||
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.',
|
||||
action='store_true')
|
||||
parser_sniff_tcpdump.add_argument('-#', '--number',
|
||||
help='Print packet number at beginning of line. Set by default.',
|
||||
action='store_true')
|
||||
parser_sniff_tcpdump.add_argument('-e', help='Print link layer headers. Option is set by default.',
|
||||
action='store_true')
|
||||
parser_sniff_tcpdump.add_argument('-t', action='count', default=0,
|
||||
help='Please see tcpdump manual for details. Unused by default.')
|
||||
|
||||
|
||||
def setup_sniff_parser(subparsers):
|
||||
# create parser for "sniff" command
|
||||
parser_sniff = subparsers.add_parser("sniff", help="Start tcpdump capture.")
|
||||
# create parser for 'sniff' command
|
||||
parser_sniff = subparsers.add_parser('sniff', help='Start tcpdump capture.')
|
||||
setup_sniff_tcpdump_parser(parser_sniff)
|
||||
setup_pcap_filter_parser(parser_sniff)
|
||||
cap_size_group = parser_sniff.add_mutually_exclusive_group(required=True)
|
||||
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=0)
|
||||
cap_size_group.add_argument('--mins', type=int, help='Time in minutes to capture.', default=60)
|
||||
|
||||
|
||||
def setup_pcap_filter_parser(parser_sniff):
|
||||
parser_pcap_filter = parser_sniff.add_argument_parser("pcap-filter expression")
|
||||
parser_pcap_filter = parser_sniff.add_argument_parser('pcap-filter expression')
|
||||
pass
|
||||
|
||||
@ -15,5 +15,5 @@ class Metadata:
|
||||
|
||||
|
||||
def create_metadata(filename, unique_id, device_details):
|
||||
date_string = datetime.datetime.now().strftime("%Y-%m-%d-%H-%M-%S")
|
||||
meta_filename = f"meta_{date_string}_{unique_id}.json"
|
||||
date_string = datetime.datetime.now().strftime('%Y-%m-%d-%H-%M-%S')
|
||||
meta_filename = f'meta_{date_string}_{unique_id}.json'
|
||||
|
||||
@ -8,33 +8,33 @@ from iottb.definitions import DEVICE_METADATA_FILE
|
||||
|
||||
|
||||
def write_device_metadata_to_file(metadata: DeviceMetadata, device_path: Path):
|
||||
"""Write the device metadata to a JSON file in the specified directory."""
|
||||
meta_file_path = device_path / "meta.json"
|
||||
'''Write the device metadata to a JSON file in the specified directory.'''
|
||||
meta_file_path = device_path / 'meta.json'
|
||||
meta_file_path.write_text(metadata.json(indent=2))
|
||||
|
||||
|
||||
def confirm_device_metadata(metadata: DeviceMetadata) -> bool:
|
||||
"""Display device metadata for user confirmation."""
|
||||
'''Display device metadata for user confirmation.'''
|
||||
print(metadata.json(indent=2))
|
||||
return input("Confirm device metadata? (y/n): ").strip().lower() == 'y'
|
||||
return input('Confirm device metadata? (y/n): ').strip().lower() == 'y'
|
||||
|
||||
|
||||
def get_device_metadata_from_user() -> DeviceMetadata:
|
||||
"""Prompt the user to enter device details and return a populated DeviceMetadata object."""
|
||||
device_name = input("Device name: ")
|
||||
device_short_name = device_name.lower().replace(" ", "-")
|
||||
'''Prompt the user to enter device details and return a populated DeviceMetadata object.'''
|
||||
device_name = input('Device name: ')
|
||||
device_short_name = device_name.lower().replace(' ', '-')
|
||||
return DeviceMetadata(device_name=device_name, device_short_name=device_short_name)
|
||||
|
||||
|
||||
def initialize_device_root_dir(device_name: str) -> Path:
|
||||
"""Create and return the path for the device directory."""
|
||||
'''Create and return the path for the device directory.'''
|
||||
device_path = Path.cwd() / device_name
|
||||
device_path.mkdir(exist_ok=True)
|
||||
return device_path
|
||||
|
||||
|
||||
def write_metadata(metadata: BaseModel, device_name: str):
|
||||
"""Write device metadata to a JSON file."""
|
||||
'''Write device metadata to a JSON file.'''
|
||||
meta_path = Path.cwd() / device_name / DEVICE_METADATA_FILE
|
||||
meta_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with meta_path.open('w') as f:
|
||||
@ -42,7 +42,7 @@ def write_metadata(metadata: BaseModel, device_name: str):
|
||||
|
||||
|
||||
def get_device_metadata(file_path: Path) -> DeviceMetadata | None:
|
||||
"""Fetch device metadata from a JSON file."""
|
||||
'''Fetch device metadata from a JSON file.'''
|
||||
|
||||
if dev_metadata_exists(file_path):
|
||||
with file_path.open('r') as f:
|
||||
@ -51,10 +51,10 @@ def get_device_metadata(file_path: Path) -> DeviceMetadata | None:
|
||||
device_metadata = DeviceMetadata.from_json(device_metadata_json)
|
||||
return device_metadata
|
||||
except ValueError as e:
|
||||
print(f"Validation error for device metadata: {e}")
|
||||
print(f'Validation error for device metadata: {e}')
|
||||
else:
|
||||
# TODO Decide what to do (e.g. search for file etc)
|
||||
print(f"No device metadata at {file_path}")
|
||||
print(f'No device metadata at {file_path}')
|
||||
return None
|
||||
|
||||
|
||||
|
||||
@ -10,8 +10,8 @@ from iottb.subcommands.add_device import setup_init_device_root_parser
|
||||
######################
|
||||
def setup_argparse():
|
||||
# create top level parser
|
||||
root_parser = argparse.ArgumentParser(prog="iottb")
|
||||
subparsers = root_parser.add_subparsers(title="subcommands", required=True, dest="command")
|
||||
root_parser = argparse.ArgumentParser(prog='iottb')
|
||||
subparsers = root_parser.add_subparsers(title='subcommands', required=True, dest='command')
|
||||
|
||||
setup_capture_parser(subparsers)
|
||||
setup_init_device_root_parser(subparsers)
|
||||
@ -27,12 +27,12 @@ def main():
|
||||
try:
|
||||
args.func(args)
|
||||
except KeyboardInterrupt:
|
||||
print("Received keyboard interrupt. Exiting...")
|
||||
print('Received keyboard interrupt. Exiting...')
|
||||
exit(1)
|
||||
except Exception as e:
|
||||
print(f"Error: {e}")
|
||||
print(f'Error: {e}')
|
||||
# create_capture_directory(args.device_name)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
|
||||
@ -1,16 +1,16 @@
|
||||
from datetime import datetime
|
||||
from enum import Flag, unique, global_enum
|
||||
|
||||
DEVICE_METADATA_FILE = "device_metadata.json"
|
||||
CAPTURE_METADATA_FILE = "capture_metadata.json"
|
||||
TODAY_DATE_STRING = datetime.now().strftime("%d%b%Y").lower() # TODO convert to function in utils or so
|
||||
DEVICE_METADATA_FILE = 'device_metadata.json'
|
||||
CAPTURE_METADATA_FILE = 'capture_metadata.json'
|
||||
TODAY_DATE_STRING = datetime.now().strftime('%d%b%Y').lower() # TODO convert to function in utils or so
|
||||
|
||||
CAPTURE_FOLDER_BASENAME = "capture_###"
|
||||
CAPTURE_FOLDER_BASENAME = 'capture_###'
|
||||
|
||||
AFFIRMATIVE_USER_RESPONSE = {"yes", "y", "true", "Y", "Yes", "YES"}
|
||||
NEGATIVE_USER_RESPONSE = {"no", "n", "N", "No"}
|
||||
YES_DEFAULT = AFFIRMATIVE_USER_RESPONSE.union({"", " "})
|
||||
NO_DEFAULT = NEGATIVE_USER_RESPONSE.union({"", " "})
|
||||
AFFIRMATIVE_USER_RESPONSE = {'yes', 'y', 'true', 'Y', 'Yes', 'YES'}
|
||||
NEGATIVE_USER_RESPONSE = {'no', 'n', 'N', 'No'}
|
||||
YES_DEFAULT = AFFIRMATIVE_USER_RESPONSE.union({'', ' '})
|
||||
NO_DEFAULT = NEGATIVE_USER_RESPONSE.union({'', ' '})
|
||||
|
||||
|
||||
@unique
|
||||
|
||||
@ -24,12 +24,12 @@ class CaptureMetadata:
|
||||
|
||||
# tcpdump
|
||||
packet_count: Optional[int]
|
||||
pcap_filter: str = ""
|
||||
tcpdump_command: str = ""
|
||||
interface: str = ""
|
||||
pcap_filter: str = ''
|
||||
tcpdump_command: str = ''
|
||||
interface: str = ''
|
||||
|
||||
# Optional Fields
|
||||
device_ip_address: str = "No IP Address set"
|
||||
device_ip_address: str = 'No IP Address set'
|
||||
device_mac_address: Optional[str] = None
|
||||
|
||||
app: Optional[str] = None
|
||||
@ -37,30 +37,30 @@ class CaptureMetadata:
|
||||
firmware_version: Optional[str] = None
|
||||
|
||||
def __init__(self, device_metadata: DeviceMetadata, capture_dir: Path, /, **data: Any):
|
||||
logger.info(f"Creating CaptureMetadata model from DeviceMetadata: {device_metadata}")
|
||||
logger.info(f'Creating CaptureMetadata model from DeviceMetadata: {device_metadata}')
|
||||
self.device_metadata = device_metadata
|
||||
|
||||
self.capture_dir = capture_dir
|
||||
assert capture_dir.is_dir(), f"Capture directory {capture_dir} does not exist"
|
||||
assert capture_dir.is_dir(), f'Capture directory {capture_dir} does not exist'
|
||||
|
||||
def build_capture_file_name(self):
|
||||
logger.info(f"Building capture file name")
|
||||
logger.info(f'Building capture file name')
|
||||
if self.app is None:
|
||||
logger.debug(f"No app specified")
|
||||
logger.debug(f'No app specified')
|
||||
prefix = self.device_metadata.device_short_name
|
||||
else:
|
||||
logger.debug(f"App specified: {self.app}")
|
||||
assert str(self.app).strip() not in {"", " "}, f"app is not a valid name: {self.app}"
|
||||
prefix = self.app.lower().replace(" ", "_")
|
||||
# assert self.capture_dir is not None, f"{self.capture_dir} does not exist"
|
||||
filename = f"{prefix}_{str(self.capture_id)}.pcap"
|
||||
logger.debug(f"Capture file name: {filename}")
|
||||
logger.debug(f'App specified: {self.app}')
|
||||
assert str(self.app).strip() not in {'', ' '}, f'app is not a valid name: {self.app}'
|
||||
prefix = self.app.lower().replace(' ', '_')
|
||||
# assert self.capture_dir is not None, f'{self.capture_dir} does not exist'
|
||||
filename = f'{prefix}_{str(self.capture_id)}.pcap'
|
||||
logger.debug(f'Capture file name: {filename}')
|
||||
self.capture_file = filename
|
||||
|
||||
def save_capture_metadata_to_json(self, file_path: Path = Path(CAPTURE_METADATA_FILE)):
|
||||
assert self.capture_dir.is_dir(), f"capture_dir is not a directory: {self.capture_dir}"
|
||||
assert self.capture_dir.is_dir(), f'capture_dir is not a directory: {self.capture_dir}'
|
||||
if file_path.is_file():
|
||||
print(f"File {file_path} already exists, update instead.")
|
||||
print(f'File {file_path} already exists, update instead.')
|
||||
return ReturnCodes.FILE_ALREADY_EXISTS
|
||||
metadata = self.to_json(indent=2)
|
||||
with file_path.open('w') as file:
|
||||
@ -69,7 +69,7 @@ class CaptureMetadata:
|
||||
|
||||
def to_json(self, indent=2):
|
||||
# TODO: Where to validate data?
|
||||
logger.info(f"Converting CaptureMetadata to JSON")
|
||||
logger.info(f'Converting CaptureMetadata to JSON')
|
||||
data = {}
|
||||
|
||||
# List of fields from CaptureData class, if fields[key]==True, then it is a required field
|
||||
@ -94,9 +94,9 @@ class CaptureMetadata:
|
||||
|
||||
for field, is_mandatory in fields.items():
|
||||
value = getattr(self, field, None)
|
||||
if value not in [None, ""] or is_mandatory:
|
||||
if value in [None, ""] and is_mandatory:
|
||||
raise ValueError(f"Field {field} is required and cannot be empty.")
|
||||
if value not in [None, ''] or is_mandatory:
|
||||
if value in [None, ''] and is_mandatory:
|
||||
raise ValueError(f'Field {field} is required and cannot be empty.')
|
||||
data[field] = str(value) if not isinstance(value, str) else value
|
||||
logger.debug(f"Capture metadata: {data}")
|
||||
logger.debug(f'Capture metadata: {data}')
|
||||
return json.dumps(data, indent=indent)
|
||||
|
||||
@ -9,7 +9,7 @@ from iottb.definitions import ReturnCodes, DEVICE_METADATA_FILE
|
||||
from iottb.logger import logger
|
||||
# 3rd party libs
|
||||
|
||||
IMMUTABLE_FIELDS = {"device_name", "device_short_name", "device_id", "date_created"}
|
||||
IMMUTABLE_FIELDS = {'device_name', 'device_short_name', 'device_id', 'date_created'}
|
||||
|
||||
|
||||
class DeviceMetadata:
|
||||
@ -31,23 +31,23 @@ class DeviceMetadata:
|
||||
|
||||
def __init__(self, device_name: str, device_root_path: Path):
|
||||
self.device_name = device_name
|
||||
self.device_short_name = device_name.lower().replace(" ", "_")
|
||||
self.device_short_name = device_name.lower().replace(' ', '_')
|
||||
self.device_id = str(uuid.uuid4())
|
||||
self.date_created = datetime.now().strftime('%d-%m-%YT%H:%M:%S').lower()
|
||||
self.device_root_path = device_root_path
|
||||
if not self.device_root_path or not self.device_root_path.is_dir():
|
||||
logger.error(f"Invalid device root path: {device_root_path}")
|
||||
raise ValueError(f"Invalid device root path: {device_root_path}")
|
||||
logger.debug(f"Device name: {device_name}")
|
||||
logger.debug(f"Device short_name: {self.device_short_name}")
|
||||
logger.debug(f"Device root dir: {device_root_path}")
|
||||
logger.info(f"Initialized DeviceMetadata model: {device_name}")
|
||||
logger.error(f'Invalid device root path: {device_root_path}')
|
||||
raise ValueError(f'Invalid device root path: {device_root_path}')
|
||||
logger.debug(f'Device name: {device_name}')
|
||||
logger.debug(f'Device short_name: {self.device_short_name}')
|
||||
logger.debug(f'Device root dir: {device_root_path}')
|
||||
logger.info(f'Initialized DeviceMetadata model: {device_name}')
|
||||
|
||||
@classmethod
|
||||
def load_from_json(cls, device_file_path: Path):
|
||||
logger.info(f"Loading DeviceMetadata from JSON file: {device_file_path}")
|
||||
assert device_file_path.is_file(), f"{device_file_path} is not a file"
|
||||
assert device_file_path.name == DEVICE_METADATA_FILE, f"{device_file_path} is not a {DEVICE_METADATA_FILE}"
|
||||
logger.info(f'Loading DeviceMetadata from JSON file: {device_file_path}')
|
||||
assert device_file_path.is_file(), f'{device_file_path} is not a file'
|
||||
assert device_file_path.name == DEVICE_METADATA_FILE, f'{device_file_path} is not a {DEVICE_METADATA_FILE}'
|
||||
device_meta_filename = device_file_path
|
||||
|
||||
with device_meta_filename.open('r') as file:
|
||||
@ -56,9 +56,9 @@ class DeviceMetadata:
|
||||
return metadata_model_obj
|
||||
|
||||
def save_to_json(self, file_path: Path):
|
||||
logger.info(f"Saving DeviceMetadata to JSON file: {file_path}")
|
||||
logger.info(f'Saving DeviceMetadata to JSON file: {file_path}')
|
||||
if file_path.is_file():
|
||||
print(f"File {file_path} already exists, update instead.")
|
||||
print(f'File {file_path} already exists, update instead.')
|
||||
return ReturnCodes.FILE_ALREADY_EXISTS
|
||||
metadata = self.to_json(indent=2)
|
||||
with file_path.open('w') as file:
|
||||
@ -75,27 +75,27 @@ class DeviceMetadata:
|
||||
data = {}
|
||||
|
||||
fields = {
|
||||
"device_name": True,
|
||||
"device_short_name": True,
|
||||
"device_id": True,
|
||||
"date_created": True,
|
||||
"device_root_path": True,
|
||||
"aliases": False,
|
||||
"device_type": False,
|
||||
"device_serial_number": False,
|
||||
"device_firmware_version": False,
|
||||
"date_updated": False,
|
||||
"capture_files": False,
|
||||
'device_name': True,
|
||||
'device_short_name': True,
|
||||
'device_id': True,
|
||||
'date_created': True,
|
||||
'device_root_path': True,
|
||||
'aliases': False,
|
||||
'device_type': False,
|
||||
'device_serial_number': False,
|
||||
'device_firmware_version': False,
|
||||
'date_updated': False,
|
||||
'capture_files': False,
|
||||
}
|
||||
|
||||
for field, is_mandatory in fields.items():
|
||||
value = getattr(self, field, None)
|
||||
if value not in [None, ""] or is_mandatory:
|
||||
if value in [None, ""] and is_mandatory:
|
||||
logger.debug(f"Mandatory field {field}: {value}")
|
||||
raise ValueError(f"Field {field} is required and cannot be empty.")
|
||||
if value not in [None, ''] or is_mandatory:
|
||||
if value in [None, ''] and is_mandatory:
|
||||
logger.debug(f'Mandatory field {field}: {value}')
|
||||
raise ValueError(f'Field {field} is required and cannot be empty.')
|
||||
data[field] = str(value) if not isinstance(value, str) else value
|
||||
logger.debug(f"Device metadata: {data}")
|
||||
logger.debug(f'Device metadata: {data}')
|
||||
return json.dumps(data, indent=indent)
|
||||
|
||||
|
||||
@ -104,7 +104,7 @@ def dir_contains_device_metadata(dir_path: Path):
|
||||
return False
|
||||
else:
|
||||
meta_file_path = dir_path / DEVICE_METADATA_FILE
|
||||
print(f"Device metadata file path {str(meta_file_path)}")
|
||||
print(f'Device metadata file path {str(meta_file_path)}')
|
||||
if not meta_file_path.is_file():
|
||||
return False
|
||||
else:
|
||||
|
||||
@ -8,47 +8,47 @@ from iottb.utils.device_metadata_utils import *
|
||||
|
||||
|
||||
def setup_init_device_root_parser(subparsers):
|
||||
parser = subparsers.add_parser("add-device", aliases=["add-device-root", "add"])
|
||||
parser.add_argument("--root_dir", type=pathlib.Path, default=pathlib.Path.cwd())
|
||||
parser = subparsers.add_parser('add-device', aliases=['add-device-root', 'add'])
|
||||
parser.add_argument('--root_dir', type=pathlib.Path, default=pathlib.Path.cwd())
|
||||
group = parser.add_mutually_exclusive_group()
|
||||
group.add_argument("--guided", action="store_true", help="Guided setup", default=False)
|
||||
group.add_argument("--name", action="store", type=str, help="name of device")
|
||||
group.add_argument('--guided', action='store_true', help='Guided setup', default=False)
|
||||
group.add_argument('--name', action='store', type=str, help='name of device')
|
||||
parser.set_defaults(func=handle_add)
|
||||
|
||||
|
||||
def handle_add(args):
|
||||
logger.info(f"Add device handler called with args {args}")
|
||||
logger.info(f'Add device handler called with args {args}')
|
||||
|
||||
args.root_dir.mkdir(parents=True, exist_ok=True) # else metadata.save_to_file will fail TODO: unclear what to assume
|
||||
|
||||
if args.guided:
|
||||
logger.debug("begin guided setup")
|
||||
logger.debug('begin guided setup')
|
||||
metadata = guided_setup(args.root_dir)
|
||||
logger.debug("guided setup complete")
|
||||
logger.debug('guided setup complete')
|
||||
else:
|
||||
logger.debug("Setup through passed args: setup")
|
||||
logger.debug('Setup through passed args: setup')
|
||||
if not args.name:
|
||||
logger.error("No device name specified with unguided setup.")
|
||||
logger.error('No device name specified with unguided setup.')
|
||||
return ReturnCodes.ERROR
|
||||
metadata = DeviceMetadata(args.name, args.root_dir)
|
||||
|
||||
file_path = args.root_dir / DEVICE_METADATA_FILE
|
||||
if file_path.exists():
|
||||
print("Directory already contains a metadata file. Aborting.")
|
||||
print('Directory already contains a metadata file. Aborting.')
|
||||
return ReturnCodes.ABORTED
|
||||
serialized_metadata = metadata.to_json()
|
||||
response = input(f"Confirm device metadata: {serialized_metadata} [y/N]")
|
||||
logger.debug(f"response: {response}")
|
||||
response = input(f'Confirm device metadata: {serialized_metadata} [y/N]')
|
||||
logger.debug(f'response: {response}')
|
||||
if response not in definitions.AFFIRMATIVE_USER_RESPONSE:
|
||||
print("Adding device aborted by user.")
|
||||
print('Adding device aborted by user.')
|
||||
return ReturnCodes.ABORTED
|
||||
|
||||
logger.debug(f"Device metadata file {file_path}")
|
||||
logger.debug(f'Device metadata file {file_path}')
|
||||
if metadata.save_to_json(file_path) == ReturnCodes.FILE_ALREADY_EXISTS:
|
||||
logger.error("File exists after checking, which should not happen.")
|
||||
logger.error('File exists after checking, which should not happen.')
|
||||
return ReturnCodes.ABORTED
|
||||
|
||||
print("Device metadata successfully created.")
|
||||
print('Device metadata successfully created.')
|
||||
return ReturnCodes.SUCCESS
|
||||
|
||||
|
||||
@ -57,17 +57,17 @@ def configure_metadata():
|
||||
|
||||
|
||||
def guided_setup(device_root) -> DeviceMetadata:
|
||||
logger.info("Guided setup")
|
||||
response = "N"
|
||||
device_name = ""
|
||||
while response.upper() == "N":
|
||||
device_name = input("Please enter name of device: ")
|
||||
response = input(f"Confirm device name: {device_name} [y/N] ")
|
||||
if device_name == "" or device_name is None:
|
||||
print("Name cannot be empty")
|
||||
logger.warning("Name cannot be empty")
|
||||
logger.debug(f"Response is {response}")
|
||||
logger.debug(f"Device name is {device_name}")
|
||||
logger.info('Guided setup')
|
||||
response = 'N'
|
||||
device_name = ''
|
||||
while response.upper() == 'N':
|
||||
device_name = input('Please enter name of device: ')
|
||||
response = input(f'Confirm device name: {device_name} [y/N] ')
|
||||
if device_name == '' or device_name is None:
|
||||
print('Name cannot be empty')
|
||||
logger.warning('Name cannot be empty')
|
||||
logger.debug(f'Response is {response}')
|
||||
logger.debug(f'Device name is {device_name}')
|
||||
|
||||
return DeviceMetadata(device_name, device_root)
|
||||
|
||||
|
||||
@ -10,31 +10,31 @@ from iottb.utils.capture_utils import get_capture_src_folder, make_capture_src_f
|
||||
def setup_capture_parser(subparsers):
|
||||
parser = subparsers.add_parser('sniff', help='Sniff packets with tcpdump')
|
||||
# metadata args
|
||||
parser.add_argument("-a", "--ip-address", help="IP address of the device to sniff", dest="device_ip")
|
||||
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",
|
||||
parser.add_argument('device_root', help='Root folder for device to sniff',
|
||||
type=Path, 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.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",
|
||||
parser_sniff_tcpdump.add_argument('-i', '--interface', help='Interface to capture on.', dest='capture_interface',
|
||||
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. True by default.",
|
||||
action="store_true", dest="no_name_resolution")
|
||||
parser_sniff_tcpdump.add_argument("-#", "--number",
|
||||
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. 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('-I', '--monitor-mode', help='Put interface into monitor mode',
|
||||
action='store_true')
|
||||
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. True by default.',
|
||||
action='store_true')
|
||||
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.')
|
||||
|
||||
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=1000)
|
||||
cap_size_group.add_argument("--mins", type=int, help="Time in minutes to capture.", default=1)
|
||||
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)
|
||||
|
||||
@ -45,25 +45,25 @@ def cwd_is_device_root_dir() -> bool:
|
||||
|
||||
|
||||
def start_guided_device_root_dir_setup():
|
||||
assert False, "Not implemented"
|
||||
assert False, 'Not implemented'
|
||||
|
||||
|
||||
def handle_metadata():
|
||||
assert not cwd_is_device_root_dir()
|
||||
print(f"Unable to find {DEVICE_METADATA_FILE} in current working directory")
|
||||
print("You need to setup a device root directory before using this command")
|
||||
response = input("Would you like to be guided through the setup? [y/n]")
|
||||
if response.lower() == "y":
|
||||
print(f'Unable to find {DEVICE_METADATA_FILE} in current working directory')
|
||||
print('You need to setup a device root directory before using this command')
|
||||
response = input('Would you like to be guided through the setup? [y/n]')
|
||||
if response.lower() == 'y':
|
||||
start_guided_device_root_dir_setup()
|
||||
else:
|
||||
print("'iottb init-device-root --help' for more information.")
|
||||
print(''iottb init-device-root --help' for more information.')
|
||||
exit(ReturnCodes.ABORTED)
|
||||
# device_id = handle_capture_metadata()
|
||||
return ReturnCodes.SUCCESS
|
||||
|
||||
|
||||
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"
|
||||
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)
|
||||
return device_metadata
|
||||
|
||||
@ -73,26 +73,26 @@ def run_tcpdump(cmd):
|
||||
try:
|
||||
p = subprocess.run(cmd, capture_output=True, text=True, check=True)
|
||||
if p.returncode != 0:
|
||||
print(f"Error running tcpdump {p.stderr}")
|
||||
print(f'Error running tcpdump {p.stderr}')
|
||||
else:
|
||||
print(f"tcpdump run successfully\n: {p.stdout}")
|
||||
print(f'tcpdump run successfully\n: {p.stdout}')
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
|
||||
|
||||
def handle_capture(args):
|
||||
assert args.device_root is not None, f"Device root directory is required"
|
||||
assert dir_contains_device_metadata(args.device_root), f"Device metadata file '{args.device_root}' does not exist"
|
||||
assert args.device_root is not None, f'Device root directory is required'
|
||||
assert dir_contains_device_metadata(args.device_root), f'Device metadata file '{args.device_root}' does not exist'
|
||||
# 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")
|
||||
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: ")
|
||||
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
|
||||
@ -115,13 +115,13 @@ def handle_capture(args):
|
||||
capture_metadata.start_time = start_time
|
||||
capture_metadata.stop_time = stop_time
|
||||
except KeyboardInterrupt:
|
||||
print("Received keyboard interrupt.")
|
||||
print('Received keyboard interrupt.')
|
||||
exit(ReturnCodes.ABORTED)
|
||||
except subprocess.CalledProcessError as e:
|
||||
print(f"Failed to capture packet: {e}")
|
||||
print(f'Failed to capture packet: {e}')
|
||||
exit(ReturnCodes.FAILURE)
|
||||
except Exception as e:
|
||||
print(f"Failed to capture packet: {e}")
|
||||
print(f'Failed to capture packet: {e}')
|
||||
exit(ReturnCodes.FAILURE)
|
||||
|
||||
return ReturnCodes.SUCCESS
|
||||
@ -141,7 +141,7 @@ def build_tcpdump_args(args, cmd, capture_metadata: CaptureMetadata):
|
||||
cmd.append('-c')
|
||||
cmd.append(str(args.count))
|
||||
elif args.mins:
|
||||
assert False, "Unimplemented option"
|
||||
assert False, 'Unimplemented option'
|
||||
|
||||
if args.app_name is not None:
|
||||
capture_metadata.app = args.app_name
|
||||
@ -162,7 +162,7 @@ def build_tcpdump_args(args, cmd, capture_metadata: CaptureMetadata):
|
||||
# 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"
|
||||
# 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)
|
||||
|
||||
@ -9,12 +9,12 @@ def set_device_ip_address(ip_addr: str, file_path: Path):
|
||||
assert file_path.is_file()
|
||||
with file_path.open('r') as f:
|
||||
data = json.load(f)
|
||||
current_ip = data["device_ip_address"]
|
||||
current_ip = data['device_ip_address']
|
||||
if current_ip is not None:
|
||||
print(f"Device IP Address is set to {current_ip}")
|
||||
response = input(f"Do you want to change the recorded IP address to {ip_addr}? [Y/N] ")
|
||||
if response.upper() == "N":
|
||||
print("Aborting change to device IP address")
|
||||
print(f'Device IP Address is set to {current_ip}')
|
||||
response = input(f'Do you want to change the recorded IP address to {ip_addr}? [Y/N] ')
|
||||
if response.upper() == 'N':
|
||||
print('Aborting change to device IP address')
|
||||
return ReturnCodes.ABORTED
|
||||
with file_path.open('w') as f:
|
||||
json.dump(data, f)
|
||||
@ -26,12 +26,12 @@ def set_device_mac_address(mac_addr: str, file_path: Path):
|
||||
assert file_path.is_file()
|
||||
with file_path.open('r') as f:
|
||||
data = json.load(f)
|
||||
current_mac = data["device_mac_address"]
|
||||
current_mac = data['device_mac_address']
|
||||
if current_mac is not None:
|
||||
print(f"Device MAC Address is set to {current_mac}")
|
||||
response = input(f"Do you want to change the recorded MAC address to {mac_addr}? [Y/N] ")
|
||||
if response.upper() == "N":
|
||||
print("Aborting change to device MAC address")
|
||||
print(f'Device MAC Address is set to {current_mac}')
|
||||
response = input(f'Do you want to change the recorded MAC address to {mac_addr}? [Y/N] ')
|
||||
if response.upper() == 'N':
|
||||
print('Aborting change to device MAC address')
|
||||
return ReturnCodes.ABORTED
|
||||
with file_path.open('w') as f:
|
||||
json.dump(data, f)
|
||||
|
||||
@ -16,29 +16,29 @@ def get_capture_date_folder(device_root: Path):
|
||||
try:
|
||||
today_folder.mkdir()
|
||||
except FileExistsError:
|
||||
print(f"Folder {today_folder} already exists")
|
||||
print(f'Folder {today_folder} already exists')
|
||||
return today_folder
|
||||
raise FileNotFoundError(f"Given path {device_root} is not a device root directory")
|
||||
raise FileNotFoundError(f'Given path {device_root} is not a device root directory')
|
||||
|
||||
|
||||
def get_capture_src_folder(device_folder: Path):
|
||||
assert device_folder.is_dir(), f"Given path {device_folder} is not a folder"
|
||||
assert device_folder.is_dir(), f'Given path {device_folder} is not a folder'
|
||||
today_iso = get_iso_date()
|
||||
max_sequence_number = 1
|
||||
for d in device_folder.iterdir():
|
||||
if d.is_dir() and d.name.startswith(f'{today_iso}_capture_'):
|
||||
name = d.name
|
||||
num = int(name.split("_")[2])
|
||||
num = int(name.split('_')[2])
|
||||
max_sequence_number = max(max_sequence_number, num)
|
||||
|
||||
next_sequence_number = max_sequence_number + 1
|
||||
return device_folder.joinpath(f"{today_iso}_capture_{next_sequence_number:03}")
|
||||
return device_folder.joinpath(f'{today_iso}_capture_{next_sequence_number:03}')
|
||||
|
||||
|
||||
def make_capture_src_folder(capture_src_folder: Path):
|
||||
try:
|
||||
capture_src_folder.mkdir()
|
||||
except FileExistsError:
|
||||
print(f"Folder {capture_src_folder} already exists")
|
||||
print(f'Folder {capture_src_folder} already exists')
|
||||
finally:
|
||||
return capture_src_folder
|
||||
|
||||
@ -5,25 +5,25 @@ from typing import Optional
|
||||
|
||||
|
||||
def check_installed() -> bool:
|
||||
"""Check if tcpdump is installed and available on the system path."""
|
||||
'''Check if tcpdump is installed and available on the system path.'''
|
||||
return shutil.which('tcpdump') is not None
|
||||
|
||||
|
||||
def ensure_installed():
|
||||
"""Ensure that tcpdump is installed, raise an error if not."""
|
||||
'''Ensure that tcpdump is installed, raise an error if not.'''
|
||||
if not check_installed():
|
||||
raise RuntimeError("tcpdump is not installed. Please install it to continue.")
|
||||
raise RuntimeError('tcpdump is not installed. Please install it to continue.')
|
||||
|
||||
|
||||
def list_interfaces() -> str:
|
||||
"""List available network interfaces using tcpdump."""
|
||||
'''List available network interfaces using tcpdump.'''
|
||||
ensure_installed()
|
||||
try:
|
||||
result = subprocess.run(['tcpdump', '--list-interfaces'], capture_output=True, text=True, check=True)
|
||||
return result.stdout
|
||||
except subprocess.CalledProcessError as e:
|
||||
print(f"Failed to list interfaces: {e}")
|
||||
return ""
|
||||
print(f'Failed to list interfaces: {e}')
|
||||
return ''
|
||||
|
||||
|
||||
def is_valid_ipv4(ip: str) -> bool:
|
||||
|
||||
@ -13,6 +13,6 @@ def subfolder_exists(parent: Path, child: str):
|
||||
|
||||
|
||||
def generate_unique_string_with_prefix(prefix: str):
|
||||
return prefix + "_" + str(uuid.uuid4())
|
||||
return prefix + '_' + str(uuid.uuid4())
|
||||
|
||||
|
||||
|
||||
@ -10,7 +10,7 @@ from iottb.__main__ import main
|
||||
|
||||
class TestDeviceSetup(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.test_dir = Path("/tmp/iottbtest/test_add_device")
|
||||
self.test_dir = Path('/tmp/iottbtest/test_add_device')
|
||||
self.test_dir.mkdir(parents=True, exist_ok=True)
|
||||
# self.captured_output = StringIO()
|
||||
# sys.stdout = self.captured_output
|
||||
@ -25,12 +25,12 @@ class TestDeviceSetup(unittest.TestCase):
|
||||
self.test_dir.rmdir()
|
||||
# sys.stdout = sys.__stdout__
|
||||
|
||||
@patch("builtins.input", side_effect=["iPhone 14", "y", "y"])
|
||||
@patch('builtins.input', side_effect=['iPhone 14', 'y', 'y'])
|
||||
def test_guided_device_setup(self, mock_input):
|
||||
sys.argv = ['__main__.py', 'add', '--root_dir', str(self.test_dir), '--guided']
|
||||
main()
|
||||
expected_file = self.test_dir / DEVICE_METADATA_FILE
|
||||
self.assertTrue(expected_file.exists()), f"Expected file not created: {expected_file}"
|
||||
self.assertTrue(expected_file.exists()), f'Expected file not created: {expected_file}'
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user