Improved IPv6 Handling for Nmap Agent

agent/nmap_agent.py

@@ -43,8 +43,8 @@
 DNS_RESOLV_CONFIG_PATH = "/etc/resolv.conf"
 
 DEFAULT_MASK_IPV6 = 128
-# scan up to 65536 host
 IPV6_CIDR_LIMIT = 112
+IPV6_MIN_PREFIX = 8  # Minimum safe prefix length for IPv6
 
 BLACKLISTED_SERVICES = ["tcpwrapped"]
 
@@ -74,6 +74,15 @@
         self._scope_domain_regex: Optional[str] = self.args.get("scope_domain_regex")
         self._vpn_config: Optional[str] = self.args.get("vpn_config")
         self._dns_config: Optional[str] = self.args.get("dns_config")
+        self._ipv6_cidr_limit: int = int(
+            self.args.get("ipv6_cidr_limit", IPV6_CIDR_LIMIT)
+        )
+
+    def _validate_ipv6_settings(self) -> None:
+        """Validate IPv6-specific settings."""
+        max_mask = int(self.args.get("max_network_mask_ipv6", "128"))
+        if max_mask < IPV6_MIN_PREFIX or max_mask > 128:
+            raise ValueError(f"IPv6 prefix must be between {IPV6_MIN_PREFIX} and 128")
 
     def start(self) -> None:
         if self._vpn_config is not None and self._dns_config is not None:
@@ -104,18 +113,23 @@
         elif "v6" in message.selector:
             mask = int(message.data.get("mask", DEFAULT_MASK_IPV6))
             if mask < IPV6_CIDR_LIMIT:
-                raise ValueError(
-                    f"Subnet mask below {IPV6_CIDR_LIMIT} is not supported"
+                logger.error(
+                    "IPv6 subnet mask below %s is not supported", IPV6_CIDR_LIMIT
                 )
+                return
+
+            # Normalize IPv6 address
+            ip = ipaddress.IPv6Address(host)
+            normalized_host = str(ip.exploded)
 
             max_mask = int(self.args.get("max_network_mask_ipv6", "128"))
             if mask < max_mask:
                 for subnet in ipaddress.ip_network(
-                    f"{host}/{mask}", strict=False
+                    f"{normalized_host}/{mask}", strict=False
                 ).subnets(new_prefix=max_mask):
                     hosts.append((str(subnet.network_address), max_mask))
             else:
-                hosts = [(host, mask)]
+                hosts = [(normalized_host, mask)]
 
         domain_name = self._prepare_domain_name(
             message.data.get("name"), message.data.get("url")
@@ -163,6 +177,8 @@
             logger.error("Neither host or domain are set.")
 
     def _scan_host(self, host: str, mask: int) -> Tuple[Dict[str, Any], str]:
+        is_ipv6 = ":" in host  # Simple check for IPv6 address
+
         options = nmap_options.NmapOptions(
             dns_resolution=False,
             ports=self.args.get("ports"),
@@ -174,12 +190,18 @@
             scripts=self.args.get("scripts"),
             script_default=self.args.get("script_default", False),
             version_detection=self.args.get("version_info", False),
+            ipv6_enabled=is_ipv6,
         )
-        client = nmap_wrapper.NmapWrapper(options)
 
+        client = nmap_wrapper.NmapWrapper(options)
         logger.info("scanning target %s/%s with options %s", host, mask, options)
-        scan_results, normal_results = client.scan_hosts(hosts=host, mask=mask)
-        return scan_results, normal_results
+
+        try:
+            scan_results, normal_results = client.scan_hosts(hosts=host, mask=mask)
+            return scan_results, normal_results
+        except subprocess.CalledProcessError as e:
+            logger.error("Nmap scan failed for IPv6 host %s: %s", host, e)
+            raise
 
     def _scan_domain(self, domain_name: str) -> Tuple[Dict[str, Any], str]:
         options = nmap_options.NmapOptions(

agent/nmap_options.py

@@ -62,6 +62,7 @@ class NmapOptions:
     )
     no_ping: bool = True
     privileged: Optional[bool] = None
+    ipv6_enabled: bool = False
 
     def _set_os_detection_option(self) -> List[str]:
         """Appends the os detection option to the list of nmap options."""
@@ -155,6 +156,8 @@ def _run_scripts_command(self, scripts: List[str]) -> List[str]:
     def command_options(self) -> List[str]:
         """Computes the list of nmap options."""
         command_options = []
+        if self.ipv6_enabled is True:
+            command_options.append("-6")
         command_options.extend(self._set_os_detection_option())
         command_options.extend(self._set_version_detection_option())
         command_options.extend(self._set_dns_resolution_option())

tests/conftest.py

@@ -290,15 +290,41 @@ def ipv6_msg_without_mask() -> message.Message:
     )
 
 
+@pytest.fixture
+def invalid_ipv6_msg() -> message.Message:
+    """Creates an invalid IPv6 message for testing error handling."""
+    return message.Message.from_data(
+        selector="v3.asset.ip.v6",
+        data={
+            "version": 6,
+            "host": "invalid_ipv6",
+            "mask": "112",
+        },
+    )
+
+
+@pytest.fixture
+def large_subnet_ipv6_msg() -> message.Message:
+    """Creates a message with a large IPv6 subnet."""
+    return message.Message.from_data(
+        selector="v3.asset.ip.v6",
+        data={
+            "version": 6,
+            "host": "2600:3c01:224a:6e00::",
+            "mask": "112",
+        },
+    )
+
+
 @pytest.fixture
 def ipv6_msg_above_limit() -> message.Message:
-    """Creates a dummy message of type v3.asset.ip.v6 for testing purposes."""
+    """Creates a message with an IPv6 subnet above the allowed limit (below mask 112)."""
     return message.Message.from_data(
         selector="v3.asset.ip.v6",
         data={
             "version": 6,
-            "host": "2600:3c01:224a:6e00:f03c:91ff:fe18:bb2f",
-            "mask": "64",
+            "host": "2600:3c01:224a:6e00::",
+            "mask": "96",  # Below IPV6_CIDR_LIMIT of 112
         },
     )