From 4b5db31b6bf4fbc7d552cb48ace0e963bb2ff794 Mon Sep 17 00:00:00 2001 From: n0mad1k Date: Fri, 1 May 2026 11:47:12 -0400 Subject: [PATCH] Parallelize nmap phase in webrunner node scanner MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit scan_masscan_nmap() and scan_geo_scout() were running nmap serially — one host at a time with a 60s timeout, turning a 1h estimate into 24h+ for large scans. Both modes now use ThreadPoolExecutor with 10 workers (configurable via WEBRUNNER_NMAP_WORKERS env var). geo-scout uses an inner scan_one() helper to batch nmap + probe_service() per host within the same future. Also fixed probe_service() bytes format antipattern and removed dead targets_arg variable from scan_nmap_only(). run_scan.yml converted from command: > folded scalar to command: argv: list form to prevent argument splitting on space-containing values. --- modules/webrunner/tasks/node_scanner.py | 52 ++++++++++++++++--------- modules/webrunner/tasks/run_scan.yml | 40 ++++++++++++------- 2 files changed, 60 insertions(+), 32 deletions(-) diff --git a/modules/webrunner/tasks/node_scanner.py b/modules/webrunner/tasks/node_scanner.py index 0948aec..b0743e9 100644 --- a/modules/webrunner/tasks/node_scanner.py +++ b/modules/webrunner/tasks/node_scanner.py @@ -6,13 +6,16 @@ Reads cidrs.txt, runs scan pipeline, writes results.json. import argparse import json +import os import subprocess import sys import time import xml.etree.ElementTree as ET +from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path WORKDIR = Path("/root/webrunner") +NMAP_WORKERS = int(os.environ.get("WEBRUNNER_NMAP_WORKERS", "10")) def log(msg: str): @@ -120,7 +123,7 @@ def probe_service(ip: str, port: int) -> str: with socket.create_connection((ip, port), timeout=3) as s: s.settimeout(3) try: - s.send(b"HEAD / HTTP/1.0\r\nHost: %b\r\n\r\n" % ip.encode()) + s.send(b"HEAD / HTTP/1.0\r\nHost: " + ip.encode() + b"\r\n\r\n") banner = s.recv(512).decode(errors="replace").strip() return banner[:200] except Exception: @@ -145,7 +148,6 @@ def scan_nmap_only(ports: str, node_name: str) -> list[dict]: port_str = ports out_file = WORKDIR / "nmap_sweep.xml" - targets_arg = " ".join(cidrs[:50]) # nmap handles CIDRs natively cmd = [ "nmap", "-sV", "--version-intensity", "3", "-p", port_str, "-T4", "--open", @@ -196,17 +198,22 @@ def scan_masscan_nmap(ports: str, rate: int, node_name: str) -> list[dict]: return [] by_ip = group_hits_by_ip(hits) - log(f"nmap fingerprinting {len(by_ip)} hosts...") + log(f"nmap fingerprinting {len(by_ip)} hosts (workers={NMAP_WORKERS})...") results = [] - for idx, (ip, ip_ports) in enumerate(by_ip.items()): - nmap_info = run_nmap(ip, ip_ports) - for port in ip_ports: - entry: dict = {"ip": ip, "port": port} - if port in nmap_info: - entry.update(nmap_info[port]) - results.append(entry) - if (idx + 1) % 50 == 0: - log(f" nmap: {idx + 1}/{len(by_ip)} hosts done") + done = 0 + with ThreadPoolExecutor(max_workers=NMAP_WORKERS) as pool: + futures = {pool.submit(run_nmap, ip, ip_ports): (ip, ip_ports) for ip, ip_ports in by_ip.items()} + for future in as_completed(futures): + ip, ip_ports = futures[future] + nmap_info = future.result() + for port in ip_ports: + entry: dict = {"ip": ip, "port": port} + if port in nmap_info: + entry.update(nmap_info[port]) + results.append(entry) + done += 1 + if done % 50 == 0: + log(f" nmap: {done}/{len(by_ip)} hosts done") return results @@ -231,25 +238,32 @@ def scan_geo_scout(ports: str, rate: int, node_name: str) -> list[dict]: except Exception: pass - results = [] - for idx, (ip, ip_ports) in enumerate(by_ip.items()): + def scan_one(ip: str, ip_ports: list[int]) -> list[dict]: nmap_info = run_nmap(ip, ip_ports) + entries = [] for port in ip_ports: entry: dict = {"ip": ip, "port": port} if port in nmap_info: entry.update(nmap_info[port]) - # Banner grab / probe banner = probe_service(ip, port) if banner: entry["probe_banner"] = banner - # Match against target profile if port in target_ports: t = target_ports[port] entry["target_name"] = t.get("name", "") entry["target_desc"] = t.get("description", "") - results.append(entry) - if (idx + 1) % 50 == 0: - log(f" geo-scout: {idx + 1}/{len(by_ip)} hosts done") + entries.append(entry) + return entries + + results = [] + done = 0 + with ThreadPoolExecutor(max_workers=NMAP_WORKERS) as pool: + futures = {pool.submit(scan_one, ip, ip_ports): ip for ip, ip_ports in by_ip.items()} + for future in as_completed(futures): + results.extend(future.result()) + done += 1 + if done % 50 == 0: + log(f" geo-scout: {done}/{len(by_ip)} hosts done") return results diff --git a/modules/webrunner/tasks/run_scan.yml b/modules/webrunner/tasks/run_scan.yml index 1709a6b..b4ca498 100644 --- a/modules/webrunner/tasks/run_scan.yml +++ b/modules/webrunner/tasks/run_scan.yml @@ -10,12 +10,18 @@ mode: '0644' - name: Run scan on {{ node_name }} ({{ node_ip_count }} IPs, mode={{ scan_mode }}) - command: > - python3 /root/webrunner/node_scanner.py - --mode {{ scan_mode }} - --ports {{ ports_str }} - --rate {{ masscan_rate }} - --node-name {{ node_name }} + command: + argv: + - python3 + - /root/webrunner/node_scanner.py + - --mode + - "{{ scan_mode }}" + - --ports + - "{{ ports_str }}" + - --rate + - "{{ masscan_rate }}" + - --node-name + - "{{ node_name }}" args: chdir: /root/webrunner register: scan_output @@ -25,13 +31,21 @@ when: not (use_tor | default(false) | bool) - name: Run scan via Tor on {{ node_name }} ({{ node_ip_count }} IPs, mode={{ scan_mode }}) - command: > - proxychains4 -f /etc/proxychains4.conf - python3 /root/webrunner/node_scanner.py - --mode {{ scan_mode }} - --ports {{ ports_str }} - --rate {{ masscan_rate }} - --node-name {{ node_name }} + command: + argv: + - proxychains4 + - -f + - /etc/proxychains4.conf + - python3 + - /root/webrunner/node_scanner.py + - --mode + - "{{ scan_mode }}" + - --ports + - "{{ ports_str }}" + - --rate + - "{{ masscan_rate }}" + - --node-name + - "{{ node_name }}" args: chdir: /root/webrunner register: scan_output