Feature: validate targets before executing modules (-C)

nettacker/config.py

@@ -175,6 +175,7 @@ class DefaultSettings(ConfigBase):
     show_all_profiles = False
     show_help_menu = False
     show_version = False
+    validate_before_scan = False
     skip_service_discovery = False
     socks_proxy = None
     targets = None

nettacker/core/app.py

@@ -4,6 +4,7 @@
 import shutil
 import socket
 import sys
+from concurrent.futures import ThreadPoolExecutor, as_completed
 from threading import Thread
 
 import multiprocess
@@ -12,7 +13,8 @@
 from nettacker.config import Config, version_info
 from nettacker.core.arg_parser import ArgParser
 from nettacker.core.die import die_failure
-from nettacker.core.graph import create_compare_report, create_report
+from nettacker.core.graph import create_report, create_compare_report
+from nettacker.core.hostcheck import resolve_quick
 from nettacker.core.ip import (
     generate_ip_range,
     get_ip_range,
@@ -218,7 +220,6 @@ def run(self):
         if self.arguments.scan_compare_id is not None:
             create_compare_report(self.arguments, scan_id)
         log.info("ScanID: {0} ".format(scan_id) + _("done"))
-
         return exit_code
 
     def start_scan(self, scan_id):
@@ -287,7 +288,64 @@ def scan_target(
 
         return os.EX_OK
 
+    def filter_valid_targets(self, targets, timeout_per_target=2.0, max_threads=None, dedupe=True):
+        """
+        Parallel validation of targets via resolve_quick(target, timeout_sec).
+        Returns a list of canonical targets (order preserved, invalids removed).
+        """
+        # Ensure it's a concrete list (len, indexing OK)
+        try:
+            targets = list(targets)
+        except TypeError:
+            raise TypeError(f"`targets` must be iterable, got {type(targets).__name__}")
+
+        if not targets:
+            return []
+
+        if max_threads is None:
+            max_threads = min(len(targets), 10)  # cap threads
+
+        # Preserve order
+        validated_target = [None] * len(targets)  # Invalid targets will be replaced by "None"
+
+        def _task(idx, t):
+            ok, canon = resolve_quick(t, timeout_sec=timeout_per_target)
+            return idx, t, (canon if ok and canon else None)
+
+        with ThreadPoolExecutor(max_workers=max_threads) as ex:
+            futures = [ex.submit(_task, i, t) for i, t in enumerate(targets)]
+            for fut in as_completed(futures):
+                try:
+                    idx, orig_target, canon = fut.result()
+                except (OSError, socket.gaierror) as exc:
+                    log.error(f"Invalid target (resolver error): {exc!s}")
+                    continue
+
+                if canon:
+                    validated_target[idx] = canon
+                else:
+                    log.info(f"Invalid target -> dropping: {orig_target}")
+
+        # Keep order, drop Nones
+        filtered = [c for c in validated_target if c is not None]
+
+        if dedupe:
+            seen, unique = set(), []
+            for _c in filtered:
+                if _c not in seen:
+                    seen.add(_c)
+                    unique.append(_c)
+            return unique
+        return filtered
+
     def scan_target_group(self, targets, scan_id, process_number):
+        if not self.arguments.socks_proxy and self.arguments.validate_before_scan:
+            targets = self.filter_valid_targets(
+                targets,
+                timeout_per_target=2.0,
+                max_threads=self.arguments.parallel_module_scan or None,
+                dedupe=True,
+            )
         active_threads = []
         log.verbose_event_info(_("single_process_started").format(process_number))
         total_number_of_modules = len(targets) * len(self.arguments.selected_modules)

nettacker/core/arg_parser.py

@@ -362,6 +362,14 @@ def add_arguments(self):
             dest="skip_service_discovery",
             help=_("skip_service_discovery"),
         )
+        method_options.add_argument(
+            "-C",
+            "--validate-before-scan",
+            action="store_true",
+            default=Config.settings.validate_before_scan,
+            dest="validate_before_scan",
+            help=_("validate_before_scan"),
+        )
         method_options.add_argument(
             "-t",
             "--thread-per-host",

nettacker/core/hostcheck.py

@@ -0,0 +1,92 @@
+# nettacker/core/hostcheck.py
+from __future__ import annotations
+
+import concurrent.futures
+import re
+import socket
+
+from nettacker import logger
+from nettacker.core.ip import is_single_ipv4, is_single_ipv6
+
+log = logger.get_logger()
+
+_LABEL = re.compile(r"^(?!-)[A-Za-z0-9-]{1,63}(?<!-)$")
+
+
+def valid_hostname(host: str, allow_single_label: bool = True) -> bool:
+    """
+    Validate hostname syntax per RFC 1123.
+    Args:
+        host: Hostname to validate.
+        allow_single_label: If True, accept single-label names (e.g., "localhost").
+
+    Returns:
+        True if the hostname is syntactically valid.
+    """
+    if host.endswith(
+        "."
+    ):  # From RFC 1123 ,the number of characters can be 250 at max (without dots) and 253 with dots
+        host = host[:-1]
+    if len(host) > 253:
+        return False
+    parts = host.split(".")
+    if len(parts) < 2 and not allow_single_label:
+        return False
+    return all(_LABEL.match(p) for p in parts)
+
+
+def _gai_once(name: str, use_ai_addrconfig: bool, port):
+    flags = getattr(socket, "AI_ADDRCONFIG", 0) if use_ai_addrconfig else 0
+    return socket.getaddrinfo(name, port, socket.AF_UNSPEC, socket.SOCK_STREAM, 0, flags)
+
+
+def _clean_host(s: str) -> str:
+    # remove surrounding quotes and whitespaces
+    s = s.strip().strip('"').strip("'")
+    s = s.strip()  # again, after quote strip
+    # drop trailing commas that often sneak in from CSV-like inputs
+    if s.endswith(","):
+        s = s[:-1].rstrip()
+    # collapse accidental spaces inside
+    return s
+
+
+def resolve_quick(
+    host: str, timeout_sec: float = 2.0, allow_single_label: bool = True
+) -> tuple[bool, str | None]:
+    """
+    Perform fast DNS resolution with timeout.
+    Args:
+        host: Hostname or IP literal to resolve.
+        timeout_sec: Maximum time to wait for resolution.
+        allow_single_label: If True, allow single-label hostnames (e.g., "intranet").
+
+    Returns:
+        (True, host_name) on success, (False, None) on failure/timeout.
+    """
+    host = _clean_host(host)
+    if is_single_ipv4(host) or is_single_ipv6(host):  # IP literal, no resolution needed
+        return True, host
+
+    if host.endswith("."):
+        host = host[:-1]
+
+    if not valid_hostname(host, allow_single_label=allow_single_label):
+        return False, None
+
+    def _call(use_ai_addrconfig: bool):
+        return _gai_once(host, use_ai_addrconfig, None)
+
+    for use_ai in (True, False):
+        try:
+            # Run getaddrinfo in a thread so we can enforce timeout
+            with concurrent.futures.ThreadPoolExecutor(max_workers=1) as ex:
+                fut = ex.submit(_call, use_ai)
+                fut.result(timeout=timeout_sec)  # raises on timeout or error
+            return True, host.lower()
+        except concurrent.futures.TimeoutError:
+            continue
+        except (OSError, socket.gaierror):
+            # DNS resolution failed for this candidate, try next
+            continue
+    return False, None

nettacker/locale/en.yaml

@@ -11,6 +11,7 @@ API_options: API options
 API_port: API port number
 Method: Method
 skip_service_discovery: skip service discovery before scan and enforce all modules to scan anyway
+validate_before_scan: Pre-validate targets/connectivity before scanning; unreachable targets are skipped.
 no_live_service_found: no any live service found to scan.
 icmp_need_root_access: to use icmp_scan module or --ping-before-scan you need to run the script as root!
 available_graph: "build a graph of all activities and information, you must use HTML output. available graphs: {0}"