Add masscan+nuclei mode, operator tuning controls, and vars file support to webrunner

- New scan mode: masscan+nuclei — masscan finds open ip:port pairs,
  nuclei runs operator-supplied CVE template against discovered hosts
- Per-country vulnerable host count in merge output via CIDR→country lookup
- Tuning args on every scan: --nmap-timing, --nmap-timeout, --nmap-workers,
  --nuclei-rate, --nuclei-concurrency, --nuclei-timeout, --masscan-rate
- Vars file support: operator provides scan_vars.yaml to pre-fill all
  tuning settings without interactive prompts
- Templates copied to nodes at provision time — no live fetches (OPSEC)
- run_scan.yml passes all tuning args with Ansible defaults as fallback
- configure_node.yml installs nuclei binary + uploads template on demand
- Updated deployment summary shows mode-specific tuning params
- countries-format.md and targets-format.md updated for new capabilities
- scan_vars.yaml.example documents all configurable settings with comments
This commit is contained in:
n0mad1k
2026-05-02 14:49:24 -04:00
parent 9ffdad301a
commit ee1a6ff8d4
9 changed files with 406 additions and 13 deletions
+51
View File
@@ -5,16 +5,42 @@ Run by Ansible on the controller after all nodes complete.
"""
import argparse
import ipaddress
import json
import sys
from pathlib import Path
def _build_cidr_map(cc_cidrs: dict) -> list[tuple]:
result = []
for cc, cidrs in cc_cidrs.items():
for cidr in cidrs:
try:
net = ipaddress.ip_network(cidr, strict=False)
result.append((net, cc.upper()))
except ValueError:
pass
result.sort(key=lambda x: x[0].prefixlen, reverse=True)
return result
def _lookup_country(ip: str, cidr_map: list[tuple]) -> str:
try:
addr = ipaddress.ip_address(ip)
except ValueError:
return ""
for net, cc in cidr_map:
if addr in net:
return cc
return ""
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--results-dir", required=True)
parser.add_argument("--output", required=True)
parser.add_argument("--deployment-id", required=True)
parser.add_argument("--cidr-map", default="", help="JSON file mapping country codes to CIDR lists")
args = parser.parse_args()
results_dir = Path(args.results_dir)
@@ -24,6 +50,14 @@ def main():
print(f"No result files found in {results_dir}", file=sys.stderr)
sys.exit(1)
cidr_map: list[tuple] = []
if args.cidr_map:
try:
cc_cidrs = json.loads(Path(args.cidr_map).read_text())
cidr_map = _build_cidr_map(cc_cidrs)
except (OSError, json.JSONDecodeError):
pass
merged: dict = {
"deployment_id": args.deployment_id,
"nodes": [],
@@ -51,16 +85,33 @@ def main():
key = f"{entry.get('ip')}:{entry.get('port')}"
if key not in seen:
seen.add(key)
if cidr_map:
entry["country"] = _lookup_country(entry.get("ip", ""), cidr_map)
merged["results"].append(entry)
merged["total_results"] = len(merged["results"])
if cidr_map:
country_counts: dict[str, int] = {}
for entry in merged["results"]:
cc = entry.get("country", "")
country_counts[cc] = country_counts.get(cc, 0) + 1
merged["country_summary"] = dict(
sorted(country_counts.items(), key=lambda x: x[1], reverse=True)
)
Path(args.output).write_text(json.dumps(merged, indent=2))
print(f"Merged {len(node_files)} node(s) — {merged['total_results']} unique results")
for n in merged["nodes"]:
print(f" {n['node_name']}: {n['result_count']} results ({n['scan_mode']})")
if cidr_map and merged.get("country_summary"):
print("\nVulnerable hosts by country:")
for cc, count in list(merged["country_summary"].items())[:20]:
label = cc if cc else "unknown"
print(f" {label:<6} {count}")
if __name__ == "__main__":
main()