WHAT

code/iottb/models/capture_metadata_model.py

@@ -0,0 +1,102 @@
+import json
+import uuid
+from datetime import datetime
+from pathlib import Path
+from typing import Optional
+
+from iottb.definitions import ReturnCodes, CAPTURE_METADATA_FILE
+from iottb.models.device_metadata_model import DeviceMetadata
+from iottb.logger import logger
+
+
+class CaptureMetadata:
+    # Required Fields
+    device_metadata: DeviceMetadata
+    capture_id: str = lambda: str(uuid.uuid4())
+    device_id: str
+    capture_dir: Path
+    capture_file: str
+    capture_date: str = lambda: datetime.now().strftime('%d-%m-%YT%H:%M:%S').lower()
+
+    # Statistics
+    start_time: str
+    stop_time: str
+
+    # tcpdump
+    packet_count: Optional[int]
+    pcap_filter: str = ''
+    tcpdump_command: str = ''
+    interface: str = ''
+
+    # Optional Fields
+    device_ip_address: str = 'No IP Address set'
+    device_mac_address: Optional[str] = None
+
+    app: Optional[str] = None
+    app_version: Optional[str] = None
+    firmware_version: Optional[str] = None
+
+    def __init__(self, device_metadata: DeviceMetadata, capture_dir: Path):
+        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'
+
+    def build_capture_file_name(self):
+        logger.info(f'Building capture file name')
+        if self.app is None:
+            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}')
+        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}'
+        if file_path.is_file():
+            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:
+            json.dump(metadata, file)
+            return ReturnCodes.SUCCESS
+
+    def to_json(self, indent=2):
+        # TODO: Where to validate data?
+        logger.info(f'Converting CaptureMetadata to JSON')
+        data = {}
+
+        # List of fields from CaptureData class, if fields[key]==True, then it is a required field
+        fields = {
+            'capture_id': True,  #
+            'device_id': True,
+            'capture_dir': True,
+            'capture_file': False,
+            'capture_date': False,
+            'start_time': True,
+            'stop_time': True,
+            'packet_count': False,
+            'pcap_filter': False,
+            'tcpdump_command': False,
+            'interface': False,
+            'device_ip_address': False,
+            'device_mac_address': False,
+            'app': False,
+            'app_version': False,
+            'firmware_version': 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:
+                    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}')
+        return json.dumps(data, indent=indent)

code/iottb/models/device_metadata_model.py

@@ -0,0 +1,111 @@
+import json
+import uuid
+from datetime import datetime
+from pathlib import Path
+from typing import Optional, List
+
+# iottb modules
+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'}
+
+
+class DeviceMetadata:
+    # Required fields
+    device_name: str
+    device_short_name: str
+    device_id: str
+    date_created: str
+
+    device_root_path: Path
+    # Optional Fields
+    aliases: Optional[List[str]] = None
+    device_type: Optional[str] = None
+    device_serial_number: Optional[str] = None
+    device_firmware_version: Optional[str] = None
+    date_updated: Optional[str] = None
+
+    capture_files: Optional[List[str]] = []
+
+    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_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}')
+
+    @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}'
+        device_meta_filename = device_file_path
+
+        with device_meta_filename.open('r') as file:
+            metadata_json = json.load(file)
+        metadata_model_obj = cls.from_json(metadata_json)
+        return metadata_model_obj
+
+    def save_to_json(self, file_path: 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.')
+            return ReturnCodes.FILE_ALREADY_EXISTS
+        metadata = self.to_json(indent=2)
+        with file_path.open('w') as file:
+            json.dump(metadata, file)
+            return ReturnCodes.SUCCESS
+
+    @classmethod
+    def from_json(cls, metadata_json):
+        if isinstance(metadata_json, dict):
+            return DeviceMetadata(**metadata_json)
+
+    def to_json(self, indent=2):
+        # TODO: atm almost exact copy as in CaptureMetadata
+        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,
+        }
+
+        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.')
+                data[field] = str(value) if not isinstance(value, str) else value
+        logger.debug(f'Device metadata: {data}')
+        return json.dumps(data, indent=indent)
+
+
+def dir_contains_device_metadata(dir_path: Path):
+    if not dir_path.is_dir():
+        return False
+    else:
+        meta_file_path = dir_path / DEVICE_METADATA_FILE
+        print(f'Device metadata file path {str(meta_file_path)}')
+        if not meta_file_path.is_file():
+            return False
+        else:
+            return True