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"]
 
@@ -104,18 +104,23 @@ def process(self, message: msg.Message) -> None:
         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")
@@ -175,11 +180,16 @@ def _scan_host(self, host: str, mask: int) -> Tuple[Dict[str, Any], str]:
             script_default=self.args.get("script_default", False),
             version_detection=self.args.get("version_info", False),
         )
-        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(

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
         },
     )