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
91 lines
2.7 KiB
Python
91 lines
2.7 KiB
Python
import re
|
|
from typing import Optional
|
|
from .prober import ProbeResult
|
|
|
|
|
|
def _version_compare(extracted: str, operator: str, threshold: str) -> bool:
|
|
def to_tuple(v: str):
|
|
try:
|
|
return tuple(int(x) for x in v.split("."))
|
|
except ValueError:
|
|
return (v,)
|
|
|
|
ev = to_tuple(extracted)
|
|
tv = to_tuple(threshold)
|
|
if operator == "<=":
|
|
return ev <= tv
|
|
elif operator == ">=":
|
|
return ev >= tv
|
|
elif operator == "<":
|
|
return ev < tv
|
|
elif operator == ">":
|
|
return ev > tv
|
|
elif operator == "==":
|
|
return ev == tv
|
|
elif operator == "!=":
|
|
return ev != tv
|
|
return False
|
|
|
|
|
|
def _get_search_text(result: ProbeResult, match_in) -> str:
|
|
if isinstance(match_in, list):
|
|
parts = []
|
|
if "body" in match_in:
|
|
parts.append(result.body or result.banner)
|
|
if "headers" in match_in:
|
|
parts.append(" ".join(f"{k}: {v}" for k, v in result.headers.items()))
|
|
return "\n".join(parts)
|
|
if match_in == "headers":
|
|
return " ".join(f"{k}: {v}" for k, v in result.headers.items())
|
|
return result.body or result.banner
|
|
|
|
|
|
def match_probe(result: ProbeResult, probe: dict) -> tuple[bool, Optional[str]]:
|
|
if not result.success:
|
|
return False, None
|
|
|
|
match_in = probe.get("match_in", "body")
|
|
text = _get_search_text(result, match_in)
|
|
|
|
patterns = probe.get("patterns", [])
|
|
all_patterns = probe.get("all_patterns", [])
|
|
|
|
if patterns:
|
|
if not any(re.search(p, text) for p in patterns):
|
|
return False, None
|
|
|
|
if all_patterns:
|
|
if not all(re.search(p, text) for p in all_patterns):
|
|
return False, None
|
|
|
|
extracted_version = None
|
|
version_extract = probe.get("version_extract")
|
|
if version_extract:
|
|
m = re.search(version_extract, text)
|
|
if m:
|
|
extracted_version = m.group(1)
|
|
|
|
version_compare = probe.get("version_compare")
|
|
if version_compare and extracted_version:
|
|
op = version_compare.get("operator", "==")
|
|
val = str(version_compare.get("value", ""))
|
|
if not _version_compare(extracted_version, op, val):
|
|
return False, None
|
|
elif version_compare and not extracted_version:
|
|
return False, None
|
|
|
|
return True, extracted_version
|
|
|
|
|
|
def match_target(host: str, port: int, probe_results: list[tuple[dict, ProbeResult]]) -> list[dict]:
|
|
matches = []
|
|
for probe, result in probe_results:
|
|
matched, version = match_probe(result, probe)
|
|
if matched:
|
|
matches.append({
|
|
"probe_type": result.probe_type,
|
|
"confidence": probe.get("confidence", "medium"),
|
|
"version": version,
|
|
})
|
|
return matches
|