#2888: Add ConfigSchema to Services.

src/primaite/simulator/system/services/arp/arp.py

@@ -22,8 +22,15 @@ class ARP(Service):
     sends ARP requests and replies, and processes incoming ARP packets.
     """
 
+    config: "ARP.ConfigSchema"
+
     arp: Dict[IPV4Address, ARPEntry] = {}
 
+    class ConfigSchema(Service.ConfigSchema):
+        """ConfigSchema for ARP."""
+
+        type: str = "ARP"
+
     def __init__(self, **kwargs):
         kwargs["name"] = "ARP"
         kwargs["port"] = PORT_LOOKUP["ARP"]

src/primaite/simulator/system/services/database/database_service.py

@@ -24,6 +24,8 @@ class DatabaseService(Service):
     This class inherits from the `Service` class and provides methods to simulate a SQL database.
     """
 
+    config: "DatabaseService.ConfigSchema"
+
     password: Optional[str] = None
     """Password that needs to be provided by clients if they want to connect to the DatabaseService."""
 
@@ -36,6 +38,11 @@ class DatabaseService(Service):
     latest_backup_file_name: str = None
     """File name of latest backup."""
 
+    class ConfigSchema(Service.ConfigSchema):
+        """ConfigSchema for DatabaseService."""
+
+        type: str = "DATABASESERVICE"
+
     def __init__(self, **kwargs):
         kwargs["name"] = "DatabaseService"
         kwargs["port"] = PORT_LOOKUP["POSTGRES_SERVER"]

src/primaite/simulator/system/services/dns/dns_server.py

@@ -16,9 +16,16 @@ _LOGGER = getLogger(__name__)
 class DNSServer(Service):
     """Represents a DNS Server as a Service."""
 
+    config: "DNSServer.ConfigSchema"
+
     dns_table: Dict[str, IPv4Address] = {}
     "A dict of mappings between domain names and IPv4 addresses."
 
+    class ConfigSchema(Service.ConfigSchema):
+        """ConfigSchema for DNSServer."""
+
+        type: str = "DNSSERVER"
+
     def __init__(self, **kwargs):
         kwargs["name"] = "DNSServer"
         kwargs["port"] = PORT_LOOKUP["DNS"]

src/primaite/simulator/system/services/ftp/ftp_client.py

@@ -9,6 +9,7 @@ from primaite.simulator.file_system.file_system import File
 from primaite.simulator.network.protocols.ftp import FTPCommand, FTPPacket, FTPStatusCode
 from primaite.simulator.system.core.software_manager import SoftwareManager
 from primaite.simulator.system.services.ftp.ftp_service import FTPServiceABC
+from primaite.simulator.system.services.service import Service
 from primaite.utils.validation.ip_protocol import PROTOCOL_LOOKUP
 from primaite.utils.validation.port import Port, PORT_LOOKUP
 
@@ -19,10 +20,17 @@ class FTPClient(FTPServiceABC):
     """
     A class for simulating an FTP client service.
 
-    This class inherits from the `Service` class and provides methods to emulate FTP
+    This class inherits from the `FTPServiceABC` class and provides methods to emulate FTP
     RFC 959: https://datatracker.ietf.org/doc/html/rfc959
     """
 
+    config: "FTPClient.ConfigSchema"
+
+    class ConfigSchema(Service.ConfigSchema):
+        """ConfigSchema for FTPClient."""
+
+        type: str = "FTPCLIENT"
+
     def __init__(self, **kwargs):
         kwargs["name"] = "FTPClient"
         kwargs["port"] = PORT_LOOKUP["FTP"]

src/primaite/simulator/system/services/ftp/ftp_server.py

@@ -4,6 +4,7 @@ from typing import Any, Optional
 from primaite import getLogger
 from primaite.simulator.network.protocols.ftp import FTPCommand, FTPPacket, FTPStatusCode
 from primaite.simulator.system.services.ftp.ftp_service import FTPServiceABC
+from primaite.simulator.system.services.service import Service
 from primaite.utils.validation.ip_protocol import PROTOCOL_LOOKUP
 from primaite.utils.validation.port import is_valid_port, PORT_LOOKUP
 
@@ -14,13 +15,20 @@ class FTPServer(FTPServiceABC):
     """
     A class for simulating an FTP server service.
 
-    This class inherits from the `Service` class and provides methods to emulate FTP
+    This class inherits from the `FTPServiceABC` class and provides methods to emulate FTP
     RFC 959: https://datatracker.ietf.org/doc/html/rfc959
     """
 
+    config: "FTPServer.ConfigSchema"
+
     server_password: Optional[str] = None
     """Password needed to connect to FTP server. Default is None."""
 
+    class ConfigSchema(Service.ConfigSchema):
+        """ConfigSchema for FTPServer."""
+
+        type: str = "FTPServer"
+
     def __init__(self, **kwargs):
         kwargs["name"] = "FTPServer"
         kwargs["port"] = PORT_LOOKUP["FTP"]

src/primaite/simulator/system/services/icmp/icmp.py

@@ -22,8 +22,15 @@ class ICMP(Service):
     network diagnostics, notably the ping command.
     """
 
+    config: "ICMP.ConfigSchema"
+
     request_replies: Dict = {}
 
+    class ConfigSchema(Service.ConfigSchema):
+        """ConfigSchema for ICMP."""
+
+        type: str = "ICMP"
+
     def __init__(self, **kwargs):
         kwargs["name"] = "ICMP"
         kwargs["port"] = PORT_LOOKUP["NONE"]

src/primaite/simulator/system/services/ntp/ntp_client.py

@@ -15,10 +15,17 @@ _LOGGER = getLogger(__name__)
 class NTPClient(Service):
     """Represents a NTP client as a service."""
 
+    config: "NTPClient.ConfigSchema"
+
     ntp_server: Optional[IPv4Address] = None
     "The NTP server the client sends requests to."
     time: Optional[datetime] = None
 
+    class ConfigSchema(Service.ConfigSchema):
+        """ConfigSchema for NTPClient."""
+
+        type: str = "NTPCLIENT"
+
     def __init__(self, **kwargs):
         kwargs["name"] = "NTPClient"
         kwargs["port"] = PORT_LOOKUP["NTP"]

src/primaite/simulator/system/services/ntp/ntp_server.py

@@ -14,6 +14,13 @@ _LOGGER = getLogger(__name__)
 class NTPServer(Service):
     """Represents a NTP server as a service."""
 
+    config: "NTPServer.ConfigSchema"
+
+    class ConfigSchema(Service.ConfigSchema):
+        """ConfigSchema for NTPServer."""
+
+        type: str = "NTPSERVER"
+
     def __init__(self, **kwargs):
         kwargs["name"] = "NTPServer"
         kwargs["port"] = PORT_LOOKUP["NTP"]

src/primaite/simulator/system/services/service.py

@@ -1,10 +1,12 @@
 # © Crown-owned copyright 2024, Defence Science and Technology Laboratory UK
 from __future__ import annotations
 
-from abc import abstractmethod
+from abc import ABC, abstractmethod
 from enum import Enum
 from typing import Any, ClassVar, Dict, Optional, Type
 
+from pydantic import BaseModel
+
 from primaite import getLogger
 from primaite.interface.request import RequestFormat, RequestResponse
 from primaite.simulator.core import RequestManager, RequestPermissionValidator, RequestType
@@ -37,6 +39,8 @@ class Service(IOSoftware):
     Services are programs that run in the background and may perform input/output operations.
     """
 
+    config: "Service.ConfigSchema"
+
     operating_state: ServiceOperatingState = ServiceOperatingState.STOPPED
     "The current operating state of the Service."
 
@@ -49,6 +53,11 @@ class Service(IOSoftware):
     _registry: ClassVar[Dict[str, Type["Service"]]] = {}
     """Registry of service types. Automatically populated when subclasses are defined."""
 
+    class ConfigSchema(BaseModel, ABC):
+        """Config Schema for Service class."""
+
+        type: str
+
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
 
@@ -69,6 +78,21 @@ class Service(IOSoftware):
             raise ValueError(f"Tried to define new hostnode {identifier}, but this name is already reserved.")
         cls._registry[identifier] = cls
 
+    @classmethod
+    def from_config(cls, config: Dict) -> "Service":
+        """Create a service from a config dictionary.
+
+        :param config: dict of options for service components constructor
+        :type config: dict
+        :return: The service component.
+        :rtype: Service
+        """
+        if config["type"] not in cls._registry:
+            raise ValueError(f"Invalid service type {config['type']}")
+        service_class = cls._registry[config["type"]]
+        service_object = service_class(config=service_class.ConfigSchema(**config))
+        return service_object
+
     def _can_perform_action(self) -> bool:
         """
         Checks if the service can perform actions.
@@ -232,14 +256,14 @@ class Service(IOSoftware):
 
     def disable(self) -> bool:
         """Disable the service."""
-        self.sys_log.info(f"Disabling Application {self.name}")
+        self.sys_log.info(f"Disabling Service {self.name}")
         self.operating_state = ServiceOperatingState.DISABLED
         return True
 
     def enable(self) -> bool:
         """Enable the disabled service."""
         if self.operating_state == ServiceOperatingState.DISABLED:
-            self.sys_log.info(f"Enabling Application {self.name}")
+            self.sys_log.info(f"Enabling Service {self.name}")
             self.operating_state = ServiceOperatingState.STOPPED
             return True
         return False

src/primaite/simulator/system/services/terminal/terminal.py

@@ -132,9 +132,16 @@ class RemoteTerminalConnection(TerminalClientConnection):
 class Terminal(Service):
     """Class used to simulate a generic terminal service. Can be interacted with by other terminals via SSH."""
 
+    config: "Terminal.ConfigSchema"
+
     _client_connection_requests: Dict[str, Optional[Union[str, TerminalClientConnection]]] = {}
     """Dictionary of connect requests made to remote nodes."""
 
+    class ConfigSchema(Service.ConfigSchema):
+        """ConfigSchema for Terminal."""
+
+        type: str = "TERMINAL"
+
     def __init__(self, **kwargs):
         kwargs["name"] = "Terminal"
         kwargs["port"] = PORT_LOOKUP["SSH"]

src/primaite/simulator/system/services/web_server/web_server.py

@@ -22,8 +22,15 @@ _LOGGER = getLogger(__name__)
 class WebServer(Service):
     """Class used to represent a Web Server Service in simulation."""
 
+    config: "WebServer.ConfigSchema"
+
     response_codes_this_timestep: List[HttpStatusCode] = []
 
+    class ConfigSchema(Service.ConfigSchema):
+        """ConfigSchema for WebServer."""
+
+        type: str = "WEBSERVER"
+
     def describe_state(self) -> Dict:
         """
         Produce a dictionary describing the current state of this object.