Introduce logger.py module.

code/iottb/logger.py

@@ -0,0 +1,28 @@
+import logging
+import sys
+from logging.handlers import RotatingFileHandler
+
+
+def setup_logging():
+    logger_obj = logging.getLogger('iottbLogger')
+    logger_obj.setLevel(logging.INFO)
+
+    file_handler = RotatingFileHandler('iottb.log')
+    console_handler = logging.StreamHandler(sys.stdout)
+
+    file_handler.setLevel(logging.DEBUG)
+    console_handler.setLevel(logging.INFO)
+
+    file_fmt = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+    console_fmt = logging.Formatter('%(name)s - %(levelname)s - %(message)s')
+
+    file_handler.setFormatter(file_fmt)
+    console_handler.setFormatter(console_fmt)
+
+    logger_obj.addHandler(file_handler)
+    logger_obj.addHandler(console_handler)
+
+    return logger_obj
+
+
+logger = setup_logging()

code/iottb/models/capture_metadata_model.py

@@ -13,7 +13,7 @@ from iottb.models.device_metadata_model import DeviceMetadata
 
 class CaptureMetadata(BaseModel):
     # Required Fields
-    device_metadata = Field(DeviceMetadata, exclude=True)
+    device_metadata: DeviceMetadata = Field(exclude=True)
     capture_id: uuid.UUID = Field(default_factory=lambda: str(uuid.uuid4()))
     capture_dir: Path
     capture_file: str
@@ -27,7 +27,7 @@ class CaptureMetadata(BaseModel):
     packet_count: Optional[int]
     pcap_filter: str = ""
     tcpdump_command: str = ""
-    interface = Field(str, default="")
+    interface: str = ""
 
     # Optional Fields
     device_ip_address: Optional[str] = "No IP Address set"

code/iottb/models/device_metadata_model.py

@@ -33,8 +33,8 @@ class DeviceMetadata(BaseModel):
         super().__init__(**data)
         self.device_name = device_name
         self.device_short_name = device_name.lower().replace(" ", "_")
-        assert dir_contains_device_metadata(device_root_dir), \
-            f"Directory {device_root_dir} is missing a {DEVICE_METADATA_FILE} file"
+        # assert dir_contains_device_metadata(device_root_dir), \
+        #     f"Directory {device_root_dir} is missing a {DEVICE_METADATA_FILE} file"
         self.device_root_dir = device_root_dir
 
     def get_device_id(self) -> str:
@@ -82,11 +82,10 @@ class DeviceMetadata(BaseModel):
         self.date_updated = datetime.now().strftime('%d-%m-%YT%H:%M:%S')
 
     @classmethod
-    def load_from_json(cls, root_path: Path):
-        assert root_path.is_file()
-        assert root_path.name == DEVICE_METADATA_FILE
-        assert dir_contains_device_metadata(root_path)
-        device_meta_filename = root_path / DEVICE_METADATA_FILE
+    def load_from_json(cls, device_file_path: 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.model_validate_json(metadata_json)
@@ -122,6 +121,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)}")
         if not meta_file_path.is_file():
             return False
         else:

code/iottb/subcommands/add_device.py

@@ -8,23 +8,23 @@ 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", type=pathlib.Path, default=pathlib.Path.cwd())
+    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("--name", action="store", type=str, help="name of device")
     parser.set_defaults(func=handle_add)
 
 
 def handle_add(args):
     print("Entered add-device-root")
-    metadata = None
+
     if args.guided:
         metadata = guided_setup(args.root_dir)
     else:
         device_name = args.name
         args.root_dir.mkdir(parents=True, exist_ok=True)
         args.root_dir.chdir()
-        metadata = DeviceMetadata(device_name)
+        metadata = DeviceMetadata(device_name, args.root_dir)
 
     file_path = args.root_dir / DEVICE_METADATA_FILE
     response = input(f"Confirm device metadata: {metadata.model_dump()} [y/N]")
@@ -50,10 +50,10 @@ def guided_setup(device_root) -> DeviceMetadata:
         device_name = input("Please enter name of device: ")
         if device_name == "" or device_name is None:
             print("Name cannot be empty")
-        response = input(f"Confirm device name: {device_name} [y/N]")
+        response = input(f"Confirm device name: {device_name} [y/N] ")
 
     assert response.lower() in definitions.AFFIRMATIVE_USER_RESPONSE.add(""), f"{response.upper()} not supported"
     assert device_name != ""
     assert device_name is not None
-    return DeviceMetadata(device_name)
+    return DeviceMetadata(device_name, device_root)
 

code/iottb/subcommands/capture.py

@@ -12,8 +12,8 @@ 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("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)
 
@@ -73,16 +73,16 @@ def run_tcpdump(cmd):
     try:
         p = subprocess.run(cmd, capture_output=True, text=True, check=True)
         if p.returncode != 0:
-            print(p.stderr)
+            print(f"Error running tcpdump {p.stderr}")
         else:
-            print(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)
+    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. "