Devel
This commit is contained in:
0
code/kydcap/utils/__init__.py
Normal file
0
code/kydcap/utils/__init__.py
Normal file
40
code/kydcap/utils/capture_metadata_utils.py
Normal file
40
code/kydcap/utils/capture_metadata_utils.py
Normal file
@@ -0,0 +1,40 @@
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
from kydcap.config import ReturnCodes
|
||||
|
||||
|
||||
def set_device_ip_address(ip_addr: str, file_path: Path):
|
||||
assert ip_addr is not None
|
||||
assert file_path.is_file()
|
||||
with file_path.open('r') as f:
|
||||
data = json.load(f)
|
||||
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")
|
||||
return ReturnCodes.ABORTED
|
||||
with file_path.open('w') as f:
|
||||
json.dump(data, f)
|
||||
return ReturnCodes.SUCCESS
|
||||
|
||||
|
||||
def set_device_mac_address(mac_addr: str, file_path: Path):
|
||||
assert mac_addr is not None
|
||||
assert file_path.is_file()
|
||||
with file_path.open('r') as f:
|
||||
data = json.load(f)
|
||||
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")
|
||||
return ReturnCodes.ABORTED
|
||||
with file_path.open('w') as f:
|
||||
json.dump(data, f)
|
||||
return ReturnCodes.SUCCESS
|
||||
|
||||
# TODO finnish for other fields in capture metadata
|
||||
49
code/kydcap/utils/device_metadata_utils.py
Normal file
49
code/kydcap/utils/device_metadata_utils.py
Normal file
@@ -0,0 +1,49 @@
|
||||
import json
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
from kydcap.config import ReturnCodes
|
||||
|
||||
|
||||
def update_firmware_version(version: str, file_path: Path):
|
||||
assert file_path.is_file()
|
||||
with file_path.open('r') as file:
|
||||
metadata = json.load(file)
|
||||
metadata['device_firmware_version'] = version
|
||||
metadata['date_updated'] = datetime.now().strftime('%d-%m-%YT%H:%M:%S').lower()
|
||||
with file_path.open('w') as file:
|
||||
json.dump(metadata, file)
|
||||
return ReturnCodes.SUCCESS
|
||||
|
||||
|
||||
def add_capture_file_reference(capture_file_reference: str, file_path: Path):
|
||||
assert file_path.is_file()
|
||||
with file_path.open('r') as file:
|
||||
metadata = json.load(file)
|
||||
metadata['capture_files'] = capture_file_reference
|
||||
metadata['date_updated'] = datetime.now().strftime('%d-%m-%YT%H:%M:%S').lower()
|
||||
with file_path.open('w') as file:
|
||||
json.dump(metadata, file)
|
||||
return ReturnCodes.SUCCESS
|
||||
|
||||
|
||||
def update_device_serial_number(device_id: str, file_path: Path):
|
||||
assert file_path.is_file()
|
||||
with file_path.open('r') as file:
|
||||
metadata = json.load(file)
|
||||
metadata['device_id'] = device_id
|
||||
metadata['date_updated'] = datetime.now().strftime('%d-%m-%YT%H:%M:%S').lower()
|
||||
with file_path.open('w') as file:
|
||||
json.dump(metadata, file)
|
||||
return ReturnCodes.SUCCESS
|
||||
|
||||
|
||||
def update_device_type(device_type: str, file_path: Path):
|
||||
assert file_path.is_file()
|
||||
with file_path.open('r') as file:
|
||||
metadata = json.load(file)
|
||||
metadata['device_type'] = device_type
|
||||
metadata['date_updated'] = datetime.now().strftime('%d-%m-%YT%H:%M:%S').lower()
|
||||
with file_path.open('w') as file:
|
||||
json.dump(metadata, file)
|
||||
return ReturnCodes.SUCCESS
|
||||
28
code/kydcap/utils/tcpdump_utils.py
Normal file
28
code/kydcap/utils/tcpdump_utils.py
Normal file
@@ -0,0 +1,28 @@
|
||||
import shutil
|
||||
import subprocess
|
||||
|
||||
|
||||
def check_installed() -> bool:
|
||||
"""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."""
|
||||
if not check_installed():
|
||||
raise RuntimeError("tcpdump is not installed. Please install it to continue.")
|
||||
|
||||
|
||||
def list_interfaces() -> str:
|
||||
"""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 ""
|
||||
|
||||
|
||||
def start_tcpdump():
|
||||
return None
|
||||
Reference in New Issue
Block a user