n0mad1k
159 lines
4.1 KiB
Python
159 lines
4.1 KiB
Python
import json
|
|
import shutil
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
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 _fmt_elapsed(seconds: int) -> str:
|
|
m, s = divmod(seconds, 60)
|
|
h, m = divmod(m, 60)
|
|
return f"{h}h{m:02d}m{s:02d}s" if h else f"{m}m{s:02d}s"
|
|
|
|
|
|
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),
|
|
"--wait", "3",
|
|
]
|
|
|
|
stop = threading.Event()
|
|
start = time.time()
|
|
|
|
def _progress():
|
|
while not stop.is_set():
|
|
found = 0
|
|
if out_file.exists():
|
|
try:
|
|
found = out_file.read_text(errors="replace").count('"ip"')
|
|
except Exception:
|
|
pass
|
|
elapsed = _fmt_elapsed(int(time.time() - start))
|
|
print(f"\r [{elapsed}] masscan running — {found} hosts found so far ", end="", flush=True)
|
|
stop.wait(timeout=4)
|
|
|
|
t = threading.Thread(target=_progress, daemon=True)
|
|
t.start()
|
|
|
|
try:
|
|
proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
proc.wait(timeout=3600)
|
|
except subprocess.TimeoutExpired:
|
|
proc.kill()
|
|
proc.wait()
|
|
except KeyboardInterrupt:
|
|
proc.kill()
|
|
proc.wait()
|
|
stop.set()
|
|
t.join(timeout=2)
|
|
print()
|
|
raise
|
|
except FileNotFoundError:
|
|
stop.set()
|
|
t.join(timeout=2)
|
|
print()
|
|
raise RuntimeError("masscan not found — install it first")
|
|
|
|
stop.set()
|
|
t.join(timeout=2)
|
|
print()
|
|
|
|
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
|