#1706 - Started adding the core node software required by all nodes. Made some tweaks to the Frame to have send and receive timestamp.

src/primaite/simulator/system/processes/pcap.py

@@ -0,0 +1,61 @@
+import logging
+from pathlib import Path
+
+
+class _JSONFilter(logging.Filter):
+    def filter(self, record: logging.LogRecord) -> bool:
+        """Filter logs that start and end with '{' and '}' (JSON-like messages)."""
+        return record.getMessage().startswith("{") and record.getMessage().endswith("}")
+
+
+class PCAP:
+    """
+    A logger class for logging Frames as json strings.
+
+    This is essentially a PrimAITE simulated version of PCAP.
+
+    The PCAPs are logged to: <simulation output directory>/<hostname>/<hostname>_<ip address>_pcap.log
+    """
+
+    def __init__(self, hostname: str, ip_address: str):
+        """
+        Initialize the PCAP instance.
+
+        :param hostname: The hostname for which PCAP logs are being recorded.
+        :param ip_address: The IP address associated with the PCAP logs.
+        """
+        self.hostname = hostname
+        self.ip_address = str(ip_address)
+        self._setup_logger()
+
+    def _setup_logger(self):
+        """Set up the logger configuration."""
+        log_path = self._get_log_path()
+
+        file_handler = logging.FileHandler(filename=log_path)
+        file_handler.setLevel(60)  # Custom log level > CRITICAL to prevent any unwanted standard DEBUG-CRITICAL logs
+
+        log_format = "%(message)s"
+        file_handler.setFormatter(logging.Formatter(log_format))
+
+        logger_name = f"{self.hostname}_{self.ip_address}_pcap"
+        self.logger = logging.getLogger(logger_name)
+        self.logger.setLevel(60)  # Custom log level > CRITICAL to prevent any unwanted standard DEBUG-CRITICAL logs
+        self.logger.addHandler(file_handler)
+
+        self.logger.addFilter(_JSONFilter())
+
+    def _get_log_path(self) -> Path:
+        """Get the path for the log file."""
+        root = Path(__file__).parent.parent.parent.parent.parent.parent / "simulation_output" / self.hostname
+        root.mkdir(exist_ok=True, parents=True)
+        return root / f"{self.hostname}_{self.ip_address}_pcap.log"
+
+    def capture(self, frame):  # noqa Please don't make me, I'll have a circular import and cant use if TYPE_CHECKING ;(
+        """
+        Capture a Frame and log it.
+
+        :param frame: The PCAP frame to capture.
+        """
+        msg = frame.model_dump_json()
+        self.logger.log(level=60, msg=msg)  # Log at custom log level > CRITICAL

src/primaite/simulator/system/processes/sys_log.py

@@ -0,0 +1,87 @@
+import logging
+from pathlib import Path
+
+
+class _NotJSONFilter(logging.Filter):
+    def filter(self, record: logging.LogRecord) -> bool:
+        """Filter logs that do not start and end with '{' and '}'."""
+        return not record.getMessage().startswith("{") and not record.getMessage().endswith("}")
+
+
+class SysLog:
+    """
+    A simple logger class for writing the sys logs of a Node.
+
+    Logs are logged to: <simulation output directory>/<hostname>/<hostname>_sys.log
+    """
+
+    def __init__(self, hostname: str):
+        """
+        Initialize the SysLog instance.
+
+        :param hostname: The hostname for which logs are being recorded.
+        """
+        self.hostname = hostname
+        self._setup_logger()
+
+    def _setup_logger(self):
+        """Set up the logger configuration."""
+        log_path = self._get_log_path()
+
+        file_handler = logging.FileHandler(filename=log_path)
+        file_handler.setLevel(logging.DEBUG)
+
+        log_format = "%(asctime)s %(levelname)s: %(message)s"
+        file_handler.setFormatter(logging.Formatter(log_format))
+
+        self.logger = logging.getLogger(f"{self.hostname}_sys_log")
+        self.logger.setLevel(logging.DEBUG)
+        self.logger.addHandler(file_handler)
+
+        self.logger.addFilter(_NotJSONFilter())
+
+    def _get_log_path(self) -> Path:
+        """Get the path for the log file."""
+        root = Path(__file__).parent.parent.parent.parent.parent.parent / "simulation_output" / self.hostname
+        root.mkdir(exist_ok=True, parents=True)
+        return root / f"{self.hostname}_sys.log"
+
+    def debug(self, msg: str):
+        """
+        Log a debug message.
+
+        :param msg: The message to log.
+        """
+        self.logger.debug(msg)
+
+    def info(self, msg: str):
+        """
+        Log an info message.
+
+        :param msg: The message to log.
+        """
+        self.logger.info(msg)
+
+    def warning(self, msg: str):
+        """
+        Log a warning message.
+
+        :param msg: The message to log.
+        """
+        self.logger.warning(msg)
+
+    def error(self, msg: str):
+        """
+        Log an error message.
+
+        :param msg: The message to log.
+        """
+        self.logger.error(msg)
+
+    def critical(self, msg: str):
+        """
+        Log a critical message.
+
+        :param msg: The message to log.
+        """
+        self.logger.critical(msg)

src/primaite/simulator/system/software.py

@@ -0,0 +1,94 @@
+from enum import Enum
+
+from primaite.simulator.core import SimComponent
+
+
+class SoftwareHealthState(Enum):
+    """Enumeration of the Software Health States."""
+
+    GOOD = 1
+    "The software is in a good and healthy condition."
+    COMPROMISED = 2
+    "The software's security has been compromised."
+    OVERWHELMED = 3
+    "he software is overwhelmed and not functioning properly."
+    PATCHING = 4
+    "The software is undergoing patching or updates."
+
+
+class ApplicationOperatingState(Enum):
+    """Enumeration of Application Operating States."""
+
+    CLOSED = 0
+    "The application is closed or not running."
+    RUNNING = 1
+    "The application is running."
+    INSTALLING = 3
+    "The application is being installed or updated."
+
+
+class ServiceOperatingState(Enum):
+    """Enumeration of Service Operating States."""
+
+    STOPPED = 0
+    "The service is not running."
+    RUNNING = 1
+    "The service is currently running."
+    RESTARTING = 2
+    "The service is in the process of restarting."
+    INSTALLING = 3
+    "The service is being installed or updated."
+    PAUSED = 4
+    "The service is temporarily paused."
+    DISABLED = 5
+    "The service is disabled and cannot be started."
+
+
+class ProcessOperatingState(Enum):
+    """Enumeration of Process Operating States."""
+
+    RUNNING = 1
+    "The process is running."
+    PAUSED = 2
+    "The process is temporarily paused."
+
+
+class SoftwareCriticality(Enum):
+    """Enumeration of Software Criticality Levels."""
+
+    LOWEST = 1
+    "The lowest level of criticality."
+    LOW = 2
+    "A low level of criticality."
+    MEDIUM = 3
+    "A medium level of criticality."
+    HIGH = 4
+    "A high level of criticality."
+    HIGHEST = 5
+    "The highest level of criticality."
+
+
+class Software(SimComponent):
+    """
+    Represents software information along with its health, criticality, and status.
+
+    This class inherits from the Pydantic BaseModel and provides a structured way to store
+    information about software entities.
+
+    Attributes:
+        name (str): The name of the software.
+        health_state_actual (SoftwareHealthState): The actual health state of the software.
+        health_state_visible (SoftwareHealthState): The health state of the software visible to users.
+        criticality (SoftwareCriticality): The criticality level of the software.
+        patching_count (int, optional): The count of patches applied to the software. Default is 0.
+        scanning_count (int, optional): The count of times the software has been scanned. Default is 0.
+        revealed_to_red (bool, optional): Indicates if the software has been revealed to red team. Default is False.
+    """
+
+    name: str
+    health_state_actual: SoftwareHealthState
+    health_state_visible: SoftwareHealthState
+    criticality: SoftwareCriticality
+    patching_count: int = 0
+    scanning_count: int = 0
+    revealed_to_red: bool = False