Common: Add NetworkScanConfiguration

monkey/common/configuration/__init__.py

@@ -9,6 +9,8 @@ from .agent_configuration import (
     TCPScanConfigurationSchema,
     ScanTargetConfiguration,
     ScanTargetConfigurationSchema,
+    NetworkScanConfiguration,
+    NetworkScanConfigurationSchema,
     ExploitationOptionsConfiguration,
     ExploitationOptionsConfigurationSchema,
     ExploiterConfiguration,

monkey/common/configuration/agent_configuration.py

@@ -93,6 +93,25 @@ class TCPScanConfigurationSchema(Schema):
         return TCPScanConfiguration(**data)
 
 
+@dataclass(frozen=True)
+class NetworkScanConfiguration:
+    tcp: TCPScanConfiguration
+    icmp: ICMPScanConfiguration
+    fingerprinters: List[PluginConfiguration]
+    targets: ScanTargetConfiguration
+
+
+class NetworkScanConfigurationSchema(Schema):
+    tcp = fields.Nested(TCPScanConfigurationSchema)
+    icmp = fields.Nested(ICMPScanConfigurationSchema)
+    fingerprinters = fields.List(fields.Nested(PluginConfigurationSchema))
+    targets = fields.Nested(ScanTargetConfigurationSchema)
+
+    @post_load
+    def _make_network_scan_configuration(self, data, **kwargs):
+        return NetworkScanConfiguration(**data)
+
+
 class ExploitationOptionsConfigurationSchema(Schema):
     http_ports = fields.List(fields.Int())
 

monkey/tests/unit_tests/common/test_agent_configuration.py

@@ -6,6 +6,7 @@ from common.configuration import (
     ExploitationOptionsConfigurationSchema,
     ExploiterConfigurationSchema,
     ICMPScanConfigurationSchema,
+    NetworkScanConfigurationSchema,
     PluginConfigurationSchema,
     ScanTargetConfigurationSchema,
     TCPScanConfigurationSchema,
@@ -45,45 +46,77 @@ def test_custom_pba_configuration_schema():
     assert config.windows_filename == windows_filename
 
 
+BLOCKED_IPS = ["10.0.0.1", "192.168.1.1"]
+INACCESSIBLE_SUBNETS = ["172.0.0.0/24", "172.2.2.0/24", "192.168.56.0/24"]
+LOCAL_NETWORK_SCAN = True
+SUBNETS = ["10.0.0.2", "10.0.0.2/16"]
+SCAN_TARGET_CONFIGURATION = {
+    "blocked_ips": BLOCKED_IPS,
+    "inaccessible_subnets": INACCESSIBLE_SUBNETS,
+    "local_network_scan": LOCAL_NETWORK_SCAN,
+    "subnets": SUBNETS,
+}
+
+
 def test_scan_target_configuration():
-    blocked_ips = ["10.0.0.1", "192.168.1.1"]
-    inaccessible_subnets = ["172.0.0.0/24", "172.2.2.0/24", "192.168.56.0/24"]
-    local_network_scan = True
-    subnets = ["10.0.0.2", "10.0.0.2/16"]
-    scan_target_config = {
-        "blocked_ips": blocked_ips,
-        "inaccessible_subnets": inaccessible_subnets,
-        "local_network_scan": local_network_scan,
-        "subnets": subnets,
-    }
     schema = ScanTargetConfigurationSchema()
 
-    config = schema.load(scan_target_config)
+    config = schema.load(SCAN_TARGET_CONFIGURATION)
 
-    assert config.blocked_ips == blocked_ips
+    assert config.blocked_ips == BLOCKED_IPS
-    assert config.inaccessible_subnets == inaccessible_subnets
+    assert config.inaccessible_subnets == INACCESSIBLE_SUBNETS
-    assert config.local_network_scan == local_network_scan
+    assert config.local_network_scan == LOCAL_NETWORK_SCAN
-    assert config.subnets == subnets
+    assert config.subnets == SUBNETS
+
+
+TIMEOUT_MS = 2525
+ICMP_CONFIGURATION = {"timeout_ms": TIMEOUT_MS}
 
 
 def test_icmp_scan_configuration_schema():
-    timeout_ms = 2525
     schema = ICMPScanConfigurationSchema()
 
-    config = schema.load({"timeout_ms": timeout_ms})
+    config = schema.load(ICMP_CONFIGURATION)
 
-    assert config.timeout_ms == timeout_ms
+    assert config.timeout_ms == TIMEOUT_MS
+
+
+TIMEOUT_MS = 2525
+PORTS = [8080, 443]
+
+TCP_SCAN_CONFIGURATION = {"timeout_ms": TIMEOUT_MS, "ports": PORTS}
 
 
 def test_tcp_scan_configuration_schema():
-    timeout_ms = 2525
-    ports = [8080, 443]
     schema = TCPScanConfigurationSchema()
 
-    config = schema.load({"timeout_ms": timeout_ms, "ports": ports})
+    config = schema.load(TCP_SCAN_CONFIGURATION)
 
-    assert config.timeout_ms == timeout_ms
+    assert config.timeout_ms == TIMEOUT_MS
-    assert config.ports == ports
+    assert config.ports == PORTS
+
+
+def test_network_scan_configuration():
+    fingerprinters = [{"name": "mssql", "options": {}}]
+    network_scan_configuration = {
+        "tcp": TCP_SCAN_CONFIGURATION,
+        "icmp": ICMP_CONFIGURATION,
+        "fingerprinters": fingerprinters,
+        "targets": SCAN_TARGET_CONFIGURATION,
+    }
+    schema = NetworkScanConfigurationSchema()
+
+    config = schema.load(network_scan_configuration)
+
+    assert config.tcp.ports == TCP_SCAN_CONFIGURATION["ports"]
+    assert config.tcp.timeout_ms == TCP_SCAN_CONFIGURATION["timeout_ms"]
+    assert config.icmp.timeout_ms == ICMP_CONFIGURATION["timeout_ms"]
+    assert config.fingerprinters[0].name == fingerprinters[0]["name"]
+    assert config.fingerprinters[0].options == fingerprinters[0]["options"]
+    assert config.targets.blocked_ips == BLOCKED_IPS
+    assert config.targets.inaccessible_subnets == INACCESSIBLE_SUBNETS
+    assert config.targets.local_network_scan == LOCAL_NETWORK_SCAN
+    assert config.targets.subnets == SUBNETS
 
 
 def test_exploitation_options_configuration_schema():

vulture_allowlist.py

@@ -191,6 +191,7 @@ _make_exploitation_options_configuration  # unused method (monkey/common/configu
 _make_scan_target_configuration  # unused method (monkey/common/configuration/agent_configuration.py:105)
 _make_icmp_scan_configuration  # unused method (monkey/common/configuration/agent_configuration.py:107)
 _make_tcp_scan_configuration  # unused method (monkey/common/configuration/agent_configuration.py:122)
+_make_network_scan_configuration  # unused method (monkey/common/configuration/agent_configuration.py:110)
 LINUX  # unused variable (monkey/common/operating_systems.py:5)
 WINDOWS  # unused variable (monkey/common/operating_systems.py:6)