n0mad1k
c20f189d49
masscan port discovery → nmap fingerprinting → probe engine → pattern/version matching Dynamic YAML inputs: countries.yaml (country codes + CIDR resolution) and targets.yaml (fingerprint probes) RIR delegated stats for authoritative country CIDR data (no external API) Supports tcp_banner, http, https, rtsp, udp probe types Version extraction and comparison operators (<=, >=, ==, etc.) SQLite results per engagement, CSV/JSON export format-spec/ docs for generating input files externally
119 lines
3.1 KiB
Python
119 lines
3.1 KiB
Python
import json
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import xml.etree.ElementTree as ET
|
|
from pathlib import Path
|
|
|
|
|
|
def check_deps() -> list[str]:
|
|
missing = []
|
|
for tool in ("masscan", "nmap"):
|
|
if not shutil.which(tool):
|
|
missing.append(tool)
|
|
return missing
|
|
|
|
|
|
def masscan_sweep(cidrs: list[str], ports: list[int], rate: int, output_dir: Path) -> list[dict]:
|
|
if not cidrs or not ports:
|
|
return []
|
|
|
|
port_str = ",".join(str(p) for p in sorted(set(ports)))
|
|
cidr_file = output_dir / "targets.txt"
|
|
cidr_file.write_text("\n".join(cidrs))
|
|
out_file = output_dir / "masscan.json"
|
|
|
|
cmd = [
|
|
"masscan",
|
|
f"--rate={rate}",
|
|
f"--ports={port_str}",
|
|
"-iL", str(cidr_file),
|
|
"-oJ", str(out_file),
|
|
"--output-format", "json",
|
|
"--wait", "3",
|
|
]
|
|
|
|
try:
|
|
subprocess.run(cmd, capture_output=True, timeout=3600)
|
|
except subprocess.TimeoutExpired:
|
|
pass
|
|
except FileNotFoundError:
|
|
raise RuntimeError("masscan not found — install it first")
|
|
|
|
if not out_file.exists():
|
|
return []
|
|
|
|
raw = out_file.read_text(errors="replace").strip()
|
|
if not raw or raw == "[]":
|
|
return []
|
|
|
|
raw = raw.rstrip(",\n")
|
|
if not raw.endswith("]"):
|
|
raw += "]"
|
|
if not raw.startswith("["):
|
|
raw = "[" + raw
|
|
|
|
try:
|
|
data = json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
return []
|
|
|
|
results = []
|
|
for entry in data:
|
|
ip = entry.get("ip")
|
|
for port_entry in entry.get("ports", []):
|
|
port = port_entry.get("port")
|
|
if ip and port:
|
|
results.append({"ip": ip, "port": port})
|
|
return results
|
|
|
|
|
|
def nmap_fingerprint(host: str, ports: list[int], output_dir: Path) -> dict:
|
|
if not ports:
|
|
return {}
|
|
|
|
port_str = ",".join(str(p) for p in sorted(set(ports)))
|
|
out_file = output_dir / f"nmap_{host.replace('.', '_')}.xml"
|
|
|
|
cmd = [
|
|
"nmap",
|
|
"-sV", "--version-intensity", "5",
|
|
"-p", port_str,
|
|
"-T4",
|
|
"--open",
|
|
"-oX", str(out_file),
|
|
host,
|
|
]
|
|
|
|
try:
|
|
subprocess.run(cmd, capture_output=True, timeout=60)
|
|
except (subprocess.TimeoutExpired, FileNotFoundError):
|
|
return {}
|
|
|
|
if not out_file.exists():
|
|
return {}
|
|
|
|
try:
|
|
tree = ET.parse(out_file)
|
|
except ET.ParseError:
|
|
return {}
|
|
|
|
result = {}
|
|
for port_el in tree.findall(".//port"):
|
|
portid = int(port_el.get("portid", 0))
|
|
state = port_el.find("state")
|
|
if state is None or state.get("state") != "open":
|
|
continue
|
|
service = port_el.find("service")
|
|
info = {}
|
|
if service is not None:
|
|
info["service"] = service.get("name", "")
|
|
info["product"] = service.get("product", "")
|
|
info["version"] = service.get("version", "")
|
|
info["extrainfo"] = service.get("extrainfo", "")
|
|
info["banner"] = " ".join(filter(None, [
|
|
info["product"], info["version"], info["extrainfo"]
|
|
]))
|
|
result[portid] = info
|
|
return result
|