feat: tool normalizers, Exploit-DB grounding, orchestrator, report, triage, CLI

This commit is contained in:
2026-06-29 17:36:04 -04:00
parent 1f531e4948
commit aa4501f2fa
18 changed files with 1511 additions and 0 deletions
+123
View File
@@ -0,0 +1,123 @@
"""recon-triage CLI. Single entrypoint with `scan` and `replay` subcommands.
AUTHORIZED USE ONLY. This tool performs recon and triage only. It never runs,
generates, downloads, or executes exploits.
"""
from __future__ import annotations
import logging
from datetime import UTC
from pathlib import Path
import typer
from . import __version__
from .orchestrator import build_report_from_fixtures, run_scan
from .report import markdown
from .schema import ReconReport, export_json_schema
from .scope import Scope
from .triage.llm import run_triage
AUTH_NOTICE = (
"AUTHORIZED USE ONLY — recon & triage scope. This tool enumerates assets, "
"normalizes output, and references Exploit-DB candidates. It never runs or "
"generates exploits. You are responsible for ensuring all targets are in scope."
)
app = typer.Typer(
add_completion=False,
help=f"recon-triage v{__version__}\n\n{AUTH_NOTICE}",
no_args_is_help=True,
)
def _setup_logging(verbose: bool) -> None:
logging.basicConfig(
level=logging.DEBUG if verbose else logging.INFO,
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)
def _utcnow() -> str:
from datetime import datetime
return datetime.now(UTC).isoformat()
def _write_outputs(report: ReconReport, out_dir: Path) -> None:
out_dir.mkdir(parents=True, exist_ok=True)
(out_dir / "report.json").write_text(report.model_dump_json(indent=2) + "\n")
(out_dir / "report.md").write_text(markdown.render(report))
# Always export the schema alongside for inspection.
export_json_schema(out_dir / "schemas")
typer.echo(f"Wrote {out_dir/'report.json'} and {out_dir/'report.md'}")
@app.command()
def scan(
scope: Path = typer.Option(..., "--scope", help="Path to scope.yaml"),
target: str = typer.Option(..., "--target", help="In-scope root domain to enumerate"),
out: Path = typer.Option(Path("/data/out"), "--out", help="Output directory"),
enable_nuclei: bool = typer.Option(False, "--enable-nuclei", help="Run nuclei (needs network/time)"),
passive_only: bool = typer.Option(False, "--passive-only", help="Passive enumeration only"),
rate_limit: int | None = typer.Option(None, "--rate-limit", help="Per-tool rate limit"),
timeout: float = typer.Option(600.0, "--timeout", help="Per-tool timeout (seconds)"),
templates_dir: str | None = typer.Option(None, "--templates-dir", help="nuclei templates cache dir"),
verbose: bool = typer.Option(False, "--verbose", "-v"),
) -> None:
"""Run the live recon pipeline against an in-scope target."""
_setup_logging(verbose)
typer.echo(AUTH_NOTICE)
sc = Scope.load(scope)
run_id = f"scan-{_utcnow()}"
report = run_scan(
run_id=run_id,
target=target,
scope=sc,
enable_nuclei=enable_nuclei,
passive_only=passive_only,
rate_limit=rate_limit,
timeout=timeout,
templates_dir=templates_dir,
)
report.started_at = report.started_at or run_id.replace("scan-", "")
report.finished_at = _utcnow()
report.triage = run_triage(report)
_write_outputs(report, out)
@app.command()
def replay(
fixtures: Path = typer.Option(Path("tests/fixtures"), "--fixtures", help="Fixtures directory"),
out: Path = typer.Option(Path("/data/out"), "--out", help="Output directory"),
scope: Path | None = typer.Option(None, "--scope", help="Optional scope.yaml to gate fixtures"),
enable_nuclei: bool = typer.Option(True, "--enable-nuclei/--no-nuclei", help="Include nuclei fixture"),
verbose: bool = typer.Option(False, "--verbose", "-v"),
) -> None:
"""Run the full normalize->ground->report path on canned fixtures (no network)."""
_setup_logging(verbose)
typer.echo(AUTH_NOTICE)
sc = Scope.load(scope) if scope else None
report = build_report_from_fixtures(
fixtures, run_id="replay", scope=sc, enable_nuclei=enable_nuclei
)
report.started_at = "replay"
report.finished_at = "replay"
report.triage = run_triage(report)
_write_outputs(report, out)
@app.command()
def schema(out: Path = typer.Option(Path("schemas"), "--out")) -> None:
"""Export the JSON Schema for the report contract."""
p = export_json_schema(out)
typer.echo(f"Wrote schema to {p.parent}")
def main() -> None: # entrypoint for the console script
app()
if __name__ == "__main__": # pragma: no cover
main()
@@ -0,0 +1 @@
"""Exploit-DB grounding layer."""
@@ -0,0 +1,55 @@
"""Attach Exploit-DB candidates to every service with a product.
The matcher is injectable so it can be unit-tested fully offline: pass a
``search_fn`` that reads canned ``searchsploit --json`` fixtures instead of
shelling out. Identical query terms are cached within a run.
"""
from __future__ import annotations
import logging
from collections.abc import Callable
from ..schema import ExploitDBMatch, ReconReport
from ..tools import searchsploit
log = logging.getLogger("recon_triage.grounding")
# A search function maps a query term to the candidates searchsploit returned.
SearchFn = Callable[[str], list[ExploitDBMatch]]
def default_search_fn(timeout: float = 60.0) -> SearchFn:
def _search(term: str) -> list[ExploitDBMatch]:
matches, _ = searchsploit.run(term, timeout=timeout)
return matches
return _search
def ground_report(report: ReconReport, search_fn: SearchFn) -> int:
"""Populate ``service.exploitdb_candidates`` for every service with a product.
Returns the number of candidate matches attached. Never fabricates data — only
what ``search_fn`` returns is attached.
"""
cache: dict[str, list[ExploitDBMatch]] = {}
attached = 0
for host in report.hosts:
for port in host.ports:
svc = port.service
if svc is None:
continue
term = svc.query_term()
if not term:
continue
if term not in cache:
try:
cache[term] = search_fn(term)
except Exception as e: # fail soft per service
log.warning("searchsploit failed for %r: %s", term, e)
cache[term] = []
matches = cache[term]
svc.exploitdb_candidates = list(matches)
attached += len(matches)
return attached
@@ -0,0 +1,391 @@
"""Stage sequencing, host merging, failure isolation.
The orchestrator drives the deterministic pipeline (recon -> normalize -> ground ->
report). Every stage is fail-soft: a tool that errors, times out, or returns nothing
is recorded as failed/empty and the run continues. The ``replay`` path runs the full
normalize->ground->report chain on canned fixtures with zero network.
"""
from __future__ import annotations
import logging
from pathlib import Path
from .grounding.exploitdb import SearchFn, default_search_fn, ground_report
from .schema import (
Host,
HttpInfo,
Port,
ReconReport,
ScopeUsed,
Service,
StageResult,
StageStatus,
)
from .scope import Scope
from .tools import dnsx, httpx, naabu, nmap, nuclei, subfinder
from .tools.base import ToolRun
log = logging.getLogger("recon_triage.orchestrator")
class HostTable:
"""Accumulator that merges per-tool output into a coherent set of Hosts."""
def __init__(self) -> None:
self._by_host: dict[str, Host] = {}
self._ip_to_host: dict[str, str] = {}
# -- lookup helpers -----------------------------------------------------
def _ensure_host(self, hostname: str | None, ip: str | None = None) -> Host:
key = (hostname or "").lower() or None
if key is None and ip:
key = self._ip_to_host.get(ip) or ip
if key is None:
key = "unknown"
host = self._by_host.get(key)
if host is None:
host = Host(hostname=(hostname.lower() if hostname else None), ips=[])
self._by_host[key] = host
if ip and ip not in host.ips:
host.ips.append(ip)
self._ip_to_host[ip] = key
return host
def _get_port(self, host: Host, number: int, protocol: str = "tcp") -> Port:
for p in host.ports:
if p.number == number and p.protocol == protocol:
return p
p = Port(number=number, protocol=protocol)
host.ports.append(p)
return p
# -- ingest per tool ----------------------------------------------------
def add_subfinder(self, hosts: list[str]) -> None:
for h in hosts:
self._ensure_host(h)
def add_dnsx(self, records: list[dnsx.DnsRecord]) -> None:
for rec in records:
host = self._ensure_host(rec.host)
for ip in rec.ips:
if ip not in host.ips:
host.ips.append(ip)
self._ip_to_host[ip] = host.key()
for cname in rec.cname:
if cname not in host.cnames:
host.cnames.append(cname)
def add_naabu(self, hits: list[naabu.PortHit]) -> None:
for hit in hits:
host = self._ensure_host(hit.host, hit.ip)
self._get_port(host, hit.port, hit.protocol)
def add_nmap(self, nmap_hosts: list[nmap.NmapHost]) -> None:
for nh in nmap_hosts:
hostname = nh.hostnames[0] if nh.hostnames else None
host = self._ensure_host(hostname, nh.ip)
for np in nh.ports:
port = self._get_port(host, np.number, np.protocol)
port.state = np.state
if np.service is not None:
port.service = _merge_service(port.service, np.service)
def add_httpx(self, probes: list[httpx.HttpProbe]) -> None:
for probe in probes:
host = self._ensure_host(probe.host, probe.ip)
port_num = probe.port or _port_from_url(probe.info)
if port_num is None:
continue
port = self._get_port(host, port_num, "tcp")
if port.service is None:
port.service = Service(name="http")
port.service.http = _merge_http(port.service.http, probe.info)
def hosts(self) -> list[Host]:
# Stable ordering for reproducible reports.
return [self._by_host[k] for k in sorted(self._by_host.keys())]
def _merge_service(existing: Service | None, new: Service) -> Service:
if existing is None:
return new
existing.name = existing.name or new.name
existing.product = existing.product or new.product
existing.version = existing.version or new.version
existing.cpe = existing.cpe or new.cpe
return existing
def _merge_http(existing: HttpInfo | None, new: HttpInfo) -> HttpInfo:
if existing is None:
return new
existing.url = existing.url or new.url
existing.status = existing.status if existing.status is not None else new.status
existing.title = existing.title or new.title
existing.webserver = existing.webserver or new.webserver
existing.tls = existing.tls or new.tls
for t in new.technologies:
if t not in existing.technologies:
existing.technologies.append(t)
return existing
def _port_from_url(info: HttpInfo) -> int | None:
if not info.url:
return None
if info.url.startswith("https://"):
return 443
if info.url.startswith("http://"):
return 80
return None
def _stage(name: str, run: ToolRun | None, *, count: int) -> StageResult:
"""Translate a tool run + result count into a StageResult (fail-soft)."""
if run is None:
return StageResult(name=name, status=StageStatus.SKIPPED)
if run.not_found:
return StageResult(name=name, status=StageStatus.FAILED, detail="tool not found")
if run.timed_out:
return StageResult(name=name, status=StageStatus.FAILED, detail="timed out")
if not run.ok and count == 0:
return StageResult(name=name, status=StageStatus.FAILED, detail=run.stderr[:200])
if count == 0:
return StageResult(name=name, status=StageStatus.EMPTY)
return StageResult(name=name, status=StageStatus.OK, detail=f"{count} result(s)")
# ---------------------------------------------------------------------------
# Offline replay: full normalize -> ground -> report from canned fixtures.
# ---------------------------------------------------------------------------
FIXTURE_FILES = {
"subfinder": "subfinder.jsonl",
"dnsx": "dnsx.jsonl",
"naabu": "naabu.jsonl",
"nmap": "nmap.xml",
"httpx": "httpx.jsonl",
"nuclei": "nuclei.jsonl",
}
def build_report_from_fixtures(
fixtures_dir: str | Path,
*,
run_id: str,
scope: Scope | None = None,
search_fn: SearchFn | None = None,
enable_nuclei: bool = True,
) -> ReconReport:
"""Run the deterministic pipeline against committed fixtures — no network."""
fdir = Path(fixtures_dir)
table = HostTable()
stages: list[StageResult] = []
def read(name: str) -> str:
p = fdir / FIXTURE_FILES[name]
return p.read_text() if p.exists() else ""
sf = subfinder.normalize(read("subfinder"))
table.add_subfinder(sf)
stages.append(StageResult(name="subfinder", status=_status(sf)))
dx = dnsx.normalize(read("dnsx"))
table.add_dnsx(dx)
stages.append(StageResult(name="dnsx", status=_status(dx)))
nb = naabu.normalize(read("naabu"))
table.add_naabu(nb)
stages.append(StageResult(name="naabu", status=_status(nb)))
nm = nmap.normalize(read("nmap"))
table.add_nmap(nm)
stages.append(StageResult(name="nmap", status=_status(nm)))
hx = httpx.normalize(read("httpx"))
table.add_httpx(hx)
stages.append(StageResult(name="httpx", status=_status(hx)))
nuclei_findings = []
if enable_nuclei:
nuclei_findings = nuclei.normalize(read("nuclei"))
stages.append(StageResult(name="nuclei", status=_status(nuclei_findings)))
else:
stages.append(StageResult(name="nuclei", status=StageStatus.SKIPPED))
hosts = table.hosts()
# Apply scope gating as a final safety net even in replay.
if scope is not None:
hosts = _drop_out_of_scope_hosts(hosts, scope)
report = ReconReport(
run_id=run_id,
scope=scope.as_used() if scope else ScopeUsed(),
stages=stages,
hosts=hosts,
nuclei_findings=nuclei_findings,
)
search = search_fn or default_search_fn()
attached = ground_report(report, search)
report.stages.append(
StageResult(
name="searchsploit",
status=StageStatus.OK if attached else StageStatus.EMPTY,
detail=f"{attached} candidate(s)",
)
)
return report
def _status(items: list) -> StageStatus:
return StageStatus.OK if items else StageStatus.EMPTY
# ---------------------------------------------------------------------------
# Live scan: runs the real tools, scope-gated at every boundary.
# ---------------------------------------------------------------------------
def run_scan(
*,
run_id: str,
target: str,
scope: Scope,
enable_nuclei: bool = False,
passive_only: bool = False,
rate_limit: int | None = None,
timeout: float = 600.0,
templates_dir: str | None = None,
search_fn: SearchFn | None = None,
) -> ReconReport:
"""Run the full live pipeline. Scope is enforced before any active probing.
Each stage is wrapped so a tool failure is logged and recorded, never fatal.
"""
table = HostTable()
stages: list[StageResult] = []
# Stage 1: subfinder (passive). Refuse a target that is itself out of scope.
if not scope.is_in_scope(target):
log.warning("Target %s is not in scope; nothing to do.", target)
stages.append(StageResult(name="scope", status=StageStatus.FAILED, detail="target out of scope"))
return ReconReport(run_id=run_id, target=target, scope=scope.as_used(), stages=stages)
hosts_found, sf_run = subfinder.run(target, timeout=timeout, rate_limit=rate_limit)
hosts_found = scope.filter(hosts_found + [target])
table.add_subfinder(hosts_found)
stages.append(_stage("subfinder", sf_run, count=len(hosts_found)))
# Stage 2: dnsx resolve
dx, dx_run = dnsx.run(hosts_found, timeout=timeout)
# Drop any resolved IPs that fall out of scope before they are scanned.
dx = [_scope_dns(rec, scope) for rec in dx]
table.add_dnsx(dx)
stages.append(_stage("dnsx", dx_run, count=len(dx)))
if passive_only:
for name in ("naabu", "nmap", "httpx", "nuclei"):
stages.append(StageResult(name=name, status=StageStatus.SKIPPED, detail="passive-only"))
return _finalize(run_id, target, scope, table, stages, [], search_fn)
# Stage 3: naabu connect-scan (unprivileged)
scan_targets = _scope_targets(table, scope)
nb, nb_run = naabu.run(scan_targets, timeout=timeout, rate_limit=rate_limit)
nb = [h for h in nb if _hit_in_scope(h, scope)]
table.add_naabu(nb)
stages.append(_stage("naabu", nb_run, count=len(nb)))
# Stage 4: nmap -sV on discovered open ports
open_ports = sorted({h.port for h in nb})
nm_targets = _scope_targets(table, scope)
nm, nm_run = ([], None)
if nm_targets and open_ports:
nm, nm_run = nmap.run(nm_targets, ports=open_ports, timeout=timeout, unprivileged=True)
table.add_nmap(nm)
stages.append(_stage("nmap", nm_run, count=len(nm)))
# Stage 5: httpx probing
hx, hx_run = httpx.run(_scope_targets(table, scope), timeout=timeout, rate_limit=rate_limit)
table.add_httpx(hx)
stages.append(_stage("httpx", hx_run, count=len(hx)))
# Stage 6: nuclei (gated)
nuclei_findings: list = []
if enable_nuclei:
nuclei_findings, nu_run = nuclei.run(
_scope_targets(table, scope), timeout=timeout, rate_limit=rate_limit, templates_dir=templates_dir
)
stages.append(_stage("nuclei", nu_run, count=len(nuclei_findings)))
else:
stages.append(StageResult(name="nuclei", status=StageStatus.SKIPPED, detail="not enabled"))
return _finalize(run_id, target, scope, table, stages, nuclei_findings, search_fn)
def _finalize(run_id, target, scope, table, stages, nuclei_findings, search_fn):
hosts = _drop_out_of_scope_hosts(table.hosts(), scope)
report = ReconReport(
run_id=run_id,
target=target,
scope=scope.as_used(),
stages=stages,
hosts=hosts,
nuclei_findings=nuclei_findings,
)
search = search_fn or default_search_fn()
attached = ground_report(report, search)
report.stages.append(
StageResult(
name="searchsploit",
status=StageStatus.OK if attached else StageStatus.EMPTY,
detail=f"{attached} candidate(s)",
)
)
return report
def _scope_dns(rec: dnsx.DnsRecord, scope: Scope) -> dnsx.DnsRecord:
rec.a = [ip for ip in rec.a if scope.is_in_scope(rec.host) or scope.is_in_scope(ip)]
rec.aaaa = [ip for ip in rec.aaaa if scope.is_in_scope(rec.host) or scope.is_in_scope(ip)]
return rec
def _scope_targets(table: HostTable, scope: Scope) -> list[str]:
targets: list[str] = []
for host in table.hosts():
if host.hostname and scope.is_in_scope(host.hostname):
targets.append(host.hostname)
for ip in host.ips:
if scope.is_in_scope(ip) or (host.hostname and scope.is_in_scope(host.hostname)):
targets.append(ip)
# de-dupe, preserve order
seen: set[str] = set()
out = []
for t in targets:
if t not in seen:
seen.add(t)
out.append(t)
return out
def _hit_in_scope(hit: naabu.PortHit, scope: Scope) -> bool:
for v in (hit.host, hit.ip):
if v and scope.is_in_scope(v):
return True
return False
def _drop_out_of_scope_hosts(hosts: list[Host], scope: Scope) -> list[Host]:
kept: list[Host] = []
for h in hosts:
candidates = [h.hostname] if h.hostname else []
candidates += h.ips
if any(scope.is_in_scope(c) for c in candidates if c):
kept.append(h)
else:
log.warning("Dropping out-of-scope host from report: %s", h.key())
return kept
@@ -0,0 +1 @@
"""Human-readable report rendering."""
@@ -0,0 +1,134 @@
"""Render a ReconReport as the operator-facing Markdown summary.
Layout: hosts -> services (with versions) -> Exploit-DB candidates -> nuclei
findings -> triage priorities.
"""
from __future__ import annotations
from ..schema import Host, ReconReport, Service
_NOTICE = (
"> **Authorized use only.** This is a recon-and-triage report. It enumerates and "
"*references* findings (including unverified Exploit-DB candidates); it does **not** "
"run, generate, or download exploits. A human decides next steps."
)
def _service_line(svc: Service | None) -> str:
if svc is None:
return "_(no service info)_"
parts = [p for p in (svc.product, svc.version) if p]
label = " ".join(parts) if parts else (svc.name or "unknown")
bits = [f"**{label}**"]
if svc.name and (svc.product or svc.version):
bits.append(f"({svc.name})")
if svc.cpe:
bits.append(f"`{svc.cpe}`")
if svc.http:
h = svc.http
http_bits = []
if h.status is not None:
http_bits.append(f"HTTP {h.status}")
if h.title:
http_bits.append(f"{h.title}")
if h.webserver:
http_bits.append(h.webserver)
if h.technologies:
http_bits.append("tech: " + ", ".join(h.technologies))
if http_bits:
bits.append("" + " · ".join(http_bits))
return " ".join(bits)
def _host_section(host: Host) -> list[str]:
lines: list[str] = []
title = host.hostname or host.key()
ip_str = ", ".join(host.ips) if host.ips else "unresolved"
lines.append(f"### {title}")
lines.append(f"- IPs: {ip_str}")
if host.cnames:
lines.append(f"- CNAMEs: {', '.join(host.cnames)}")
if not host.ports:
lines.append("- _No open ports recorded._")
lines.append("")
return lines
for port in sorted(host.ports, key=lambda p: p.number):
lines.append(f"- **{port.number}/{port.protocol}** ({port.state}): {_service_line(port.service)}")
svc = port.service
if svc and svc.exploitdb_candidates:
lines.append(f" - Exploit-DB candidates ({len(svc.exploitdb_candidates)}, unverified):")
for m in svc.exploitdb_candidates:
url = f" <{m.url}>" if m.url else ""
lines.append(f" - `EDB-{m.edb_id}` {m.title}{url}")
lines.append("")
return lines
def render(report: ReconReport) -> str:
lines: list[str] = []
lines.append("# recon-triage report")
lines.append("")
lines.append(_NOTICE)
lines.append("")
lines.append(f"- Run ID: `{report.run_id}`")
if report.target:
lines.append(f"- Target: `{report.target}`")
if report.started_at:
lines.append(f"- Started: {report.started_at}")
if report.finished_at:
lines.append(f"- Finished: {report.finished_at}")
lines.append("")
# Stage status table
lines.append("## Pipeline stages")
lines.append("")
lines.append("| Stage | Status | Detail |")
lines.append("|---|---|---|")
for s in report.stages:
lines.append(f"| {s.name} | {s.status} | {s.detail or ''} |")
lines.append("")
# Hosts
lines.append("## Hosts & services")
lines.append("")
if not report.hosts:
lines.append("_No in-scope hosts discovered._")
lines.append("")
for host in report.hosts:
lines.extend(_host_section(host))
# Nuclei
lines.append("## Nuclei findings")
lines.append("")
if not report.nuclei_findings:
lines.append("_None (nuclei disabled or no matches)._")
lines.append("")
else:
lines.append("| Severity | Template | Matched at | References |")
lines.append("|---|---|---|---|")
for f in sorted(report.nuclei_findings, key=lambda x: x.template_id):
refs = ", ".join(f.reference[:3])
lines.append(f"| {f.severity} | `{f.template_id}` ({f.name or ''}) | {f.matched_at or ''} | {refs} |")
lines.append("")
# Triage
lines.append("## Triage priorities")
lines.append("")
if report.triage and report.triage.prioritized_findings:
lines.append(f"_Generated by: {report.triage.generated_by}"
+ (f" ({report.triage.model})" if report.triage.model else "") + "_")
lines.append("")
for i, item in enumerate(report.triage.prioritized_findings, 1):
lines.append(f"{i}. **[{item.severity}]** {item.summary} (confidence {item.confidence:.2f})")
lines.append(f" - Rationale: {item.rationale}")
if item.evidence_refs:
lines.append(f" - Evidence: {', '.join(item.evidence_refs)}")
if item.suggested_next_step:
lines.append(f" - Suggested next step: {item.suggested_next_step}")
lines.append("")
else:
lines.append("_No triage stage output._")
lines.append("")
return "\n".join(lines)
@@ -0,0 +1,2 @@
"""Tool wrappers. Each module runs one recon tool with its native structured-output
flag and normalizes the result into the unified schema."""
@@ -0,0 +1,95 @@
"""Common subprocess runner: explicit timeouts, output capture, fail-soft records.
Every tool wrapper goes through ``run_tool``. A tool that crashes, times out, or
returns nothing never aborts the run — the caller records a StageResult and moves on.
"""
from __future__ import annotations
import logging
import shutil
import subprocess
from dataclasses import dataclass
log = logging.getLogger("recon_triage.tools")
@dataclass
class ToolRun:
"""Result of one external tool invocation."""
cmd: list[str]
returncode: int | None
stdout: str
stderr: str
timed_out: bool = False
not_found: bool = False
@property
def ok(self) -> bool:
return not self.timed_out and not self.not_found and self.returncode == 0
@property
def empty(self) -> bool:
return not self.stdout.strip()
def tool_available(binary: str) -> bool:
return shutil.which(binary) is not None
def run_tool(
cmd: list[str],
*,
timeout: float = 120.0,
input_text: str | None = None,
) -> ToolRun:
"""Run ``cmd`` with a hard timeout, capturing stdout/stderr. Never raises for
process-level failures — they are reported on the returned ToolRun."""
binary = cmd[0]
if not tool_available(binary):
log.warning("Tool not found on PATH: %s", binary)
return ToolRun(cmd=cmd, returncode=None, stdout="", stderr="not found", not_found=True)
log.info("Running: %s", " ".join(cmd))
try:
proc = subprocess.run(
cmd,
input=input_text,
capture_output=True,
text=True,
timeout=timeout,
check=False,
)
return ToolRun(
cmd=cmd,
returncode=proc.returncode,
stdout=proc.stdout or "",
stderr=proc.stderr or "",
)
except subprocess.TimeoutExpired as e:
log.warning("Tool timed out after %ss: %s", timeout, binary)
out = e.stdout.decode() if isinstance(e.stdout, bytes) else (e.stdout or "")
err = e.stderr.decode() if isinstance(e.stderr, bytes) else (e.stderr or "")
return ToolRun(cmd=cmd, returncode=None, stdout=out, stderr=err, timed_out=True)
except OSError as e: # pragma: no cover - defensive
log.warning("Tool execution error for %s: %s", binary, e)
return ToolRun(cmd=cmd, returncode=None, stdout="", stderr=str(e), not_found=True)
def parse_jsonl(text: str) -> list[dict]:
"""Parse JSONL (one JSON object per line), skipping blank/garbage lines."""
import json
out: list[dict] = []
for line in text.splitlines():
line = line.strip()
if not line:
continue
try:
obj = json.loads(line)
except json.JSONDecodeError:
log.debug("Skipping non-JSON line: %s", line[:120])
continue
if isinstance(obj, dict):
out.append(obj)
return out
@@ -0,0 +1,45 @@
"""dnsx — resolve/validate hosts, record A/AAAA/CNAME. Native output: ``-json``."""
from __future__ import annotations
from dataclasses import dataclass, field
from .base import parse_jsonl, run_tool
BINARY = "dnsx"
@dataclass
class DnsRecord:
host: str
a: list[str] = field(default_factory=list)
aaaa: list[str] = field(default_factory=list)
cname: list[str] = field(default_factory=list)
@property
def ips(self) -> list[str]:
return list(self.a) + list(self.aaaa)
def normalize(raw: str) -> list[DnsRecord]:
"""Map dnsx -json lines to DnsRecord objects."""
out: list[DnsRecord] = []
for rec in parse_jsonl(raw):
host = (rec.get("host") or "").strip().lower()
if not host:
continue
out.append(
DnsRecord(
host=host,
a=[str(x) for x in (rec.get("a") or [])],
aaaa=[str(x) for x in (rec.get("aaaa") or [])],
cname=[str(x).lower() for x in (rec.get("cname") or [])],
)
)
return out
def run(hosts: list[str], *, timeout: float = 120.0) -> tuple[list[DnsRecord], object]:
cmd = [BINARY, "-json", "-a", "-aaaa", "-cname", "-resp"]
res = run_tool(cmd, timeout=timeout, input_text="\n".join(hosts) + "\n")
return normalize(res.stdout), res
@@ -0,0 +1,90 @@
"""httpx — HTTP probing: status, title, tech, webserver, TLS. Output: ``-json -td``."""
from __future__ import annotations
from dataclasses import dataclass
from ..schema import HttpInfo, TlsInfo
from .base import parse_jsonl, run_tool
BINARY = "httpx"
@dataclass
class HttpProbe:
host: str | None
ip: str | None
port: int | None
info: HttpInfo
def _port_of(rec: dict) -> int | None:
port = rec.get("port")
try:
return int(port)
except (TypeError, ValueError):
return None
def _tls_of(rec: dict) -> TlsInfo | None:
tls = rec.get("tls")
if not isinstance(tls, dict):
return None
issuer = tls.get("issuer_cn") or tls.get("issuer_common_name")
if not issuer:
org = tls.get("issuer_org")
if isinstance(org, list) and org:
issuer = org[0]
elif isinstance(org, str):
issuer = org
return TlsInfo(
subject_cn=tls.get("subject_cn") or tls.get("subject_common_name"),
issuer=issuer,
not_after=tls.get("not_after"),
)
def normalize(raw: str) -> list[HttpProbe]:
"""Map httpx -json lines into HttpProbe records keyed by host/ip/port."""
out: list[HttpProbe] = []
for rec in parse_jsonl(raw):
tech = rec.get("tech") or rec.get("technologies") or []
if isinstance(tech, str):
tech = [tech]
info = HttpInfo(
url=rec.get("url"),
status=rec.get("status_code") or rec.get("status-code"),
title=rec.get("title"),
webserver=rec.get("webserver"),
technologies=[str(t) for t in tech],
tls=_tls_of(rec),
)
out.append(
HttpProbe(
host=(rec.get("input") or rec.get("host") or "").lower() or None,
ip=rec.get("host") if _looks_like_ip(rec.get("host")) else rec.get("a", [None])[0] if isinstance(rec.get("a"), list) else None,
port=_port_of(rec),
info=info,
)
)
return out
def _looks_like_ip(v) -> bool:
import ipaddress
if not isinstance(v, str):
return False
try:
ipaddress.ip_address(v)
return True
except ValueError:
return False
def run(targets: list[str], *, timeout: float = 180.0, rate_limit: int | None = None) -> tuple[list[HttpProbe], object]:
cmd = [BINARY, "-json", "-td", "-silent", "-title", "-tech-detect", "-web-server", "-status-code"]
if rate_limit:
cmd += ["-rate-limit", str(rate_limit)]
res = run_tool(cmd, timeout=timeout, input_text="\n".join(targets) + "\n")
return normalize(res.stdout), res
@@ -0,0 +1,68 @@
"""naabu — port discovery. Native output: ``-json``.
Defaults to ``-scan-type connect`` (TCP connect) so it runs unprivileged with no
added capabilities.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from .base import parse_jsonl, run_tool
BINARY = "naabu"
@dataclass
class PortHit:
host: str | None
ip: str | None
port: int
protocol: str = "tcp"
@dataclass
class NaabuResult:
hits: list[PortHit] = field(default_factory=list)
def normalize(raw: str) -> list[PortHit]:
"""Map naabu -json lines to PortHit records."""
out: list[PortHit] = []
for rec in parse_jsonl(raw):
port = rec.get("port")
# naabu may emit port as int or nested object depending on version.
if isinstance(port, dict):
port_num = port.get("Port") or port.get("port")
else:
port_num = port
try:
port_num = int(port_num)
except (TypeError, ValueError):
continue
out.append(
PortHit(
host=(rec.get("host") or None),
ip=(rec.get("ip") or None),
port=port_num,
protocol=str(rec.get("protocol") or "tcp").lower(),
)
)
return out
def run(
targets: list[str],
*,
timeout: float = 300.0,
rate_limit: int | None = None,
ports: str | None = None,
) -> tuple[list[PortHit], object]:
# connect scan keeps the container unprivileged (no NET_RAW needed).
cmd = [BINARY, "-json", "-scan-type", "connect", "-silent"]
if ports:
cmd += ["-p", ports]
if rate_limit:
cmd += ["-rate", str(rate_limit)]
res = run_tool(cmd, timeout=timeout, input_text="\n".join(targets) + "\n")
return normalize(res.stdout), res
+113
View File
@@ -0,0 +1,113 @@
"""nmap — service + version detection. Native output: ``-sV -oX -`` (XML).
``product`` + ``version`` per service is the key signal feeding Exploit-DB matching,
so it is captured precisely. Uses ``-sT`` connect scan to stay unprivileged.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass, field
import xmltodict
from ..schema import Port, Service
from .base import run_tool
log = logging.getLogger("recon_triage.tools.nmap")
BINARY = "nmap"
@dataclass
class NmapHost:
ip: str | None = None
hostnames: list[str] = field(default_factory=list)
ports: list[Port] = field(default_factory=list)
def _as_list(value) -> list:
if value is None:
return []
if isinstance(value, list):
return value
return [value]
def _parse_service(svc: dict | None) -> Service | None:
if not svc:
return None
cpe = svc.get("cpe")
if isinstance(cpe, list):
cpe = cpe[0] if cpe else None
if isinstance(cpe, dict): # xmltodict text node
cpe = cpe.get("#text")
return Service(
name=svc.get("@name"),
product=svc.get("@product"),
version=svc.get("@version"),
cpe=cpe,
)
def normalize(raw_xml: str) -> list[NmapHost]:
"""Parse nmap XML into NmapHost records with precise product/version."""
if not raw_xml.strip():
return []
try:
doc = xmltodict.parse(raw_xml)
except Exception as e: # pragma: no cover - malformed XML
log.warning("Failed to parse nmap XML: %s", e)
return []
run = (doc or {}).get("nmaprun") or {}
hosts_out: list[NmapHost] = []
for host in _as_list(run.get("host")):
ip = None
for addr in _as_list(host.get("address")):
if addr.get("@addrtype") in ("ipv4", "ipv6"):
ip = addr.get("@addr")
break
hostnames = []
hn = host.get("hostnames")
if isinstance(hn, dict):
for h in _as_list(hn.get("hostname")):
if h.get("@name"):
hostnames.append(h["@name"].lower())
ports_out: list[Port] = []
ports_node = host.get("ports")
if isinstance(ports_node, dict):
for p in _as_list(ports_node.get("port")):
state = (p.get("state") or {}).get("@state", "open")
try:
number = int(p.get("@portid"))
except (TypeError, ValueError):
continue
ports_out.append(
Port(
number=number,
protocol=p.get("@protocol", "tcp"),
state=state,
service=_parse_service(p.get("service")),
)
)
hosts_out.append(NmapHost(ip=ip, hostnames=hostnames, ports=ports_out))
return hosts_out
def run(
targets: list[str],
ports: list[int] | None = None,
*,
timeout: float = 600.0,
unprivileged: bool = True,
) -> tuple[list[NmapHost], object]:
cmd = [BINARY, "-sV", "-oX", "-", "-Pn"]
if unprivileged:
cmd.append("-sT") # TCP connect scan; no raw sockets required
if ports:
cmd += ["-p", ",".join(str(p) for p in sorted(set(ports)))]
cmd += list(targets)
res = run_tool(cmd, timeout=timeout)
return normalize(res.stdout), res
@@ -0,0 +1,59 @@
"""nuclei — template-based exposure/misconfig/CVE checks. Output: ``-jsonl``.
Gated behind ``--enable-nuclei`` (off by default in V0): it needs network for
template fetch and meaningful runtime. Templates cache to a runtime volume.
"""
from __future__ import annotations
from ..schema import NucleiFinding, Severity
from .base import parse_jsonl, run_tool
BINARY = "nuclei"
_SEVERITIES = {s.value for s in Severity}
def _severity(value) -> Severity:
v = str(value or "").lower().strip()
return Severity(v) if v in _SEVERITIES else Severity.UNKNOWN
def normalize(raw: str) -> list[NucleiFinding]:
"""Map nuclei -jsonl lines into NucleiFinding records."""
out: list[NucleiFinding] = []
for rec in parse_jsonl(raw):
info = rec.get("info") or {}
ref = info.get("reference") or []
if isinstance(ref, str):
ref = [ref]
template_id = rec.get("template-id") or rec.get("templateID") or rec.get("template_id")
if not template_id:
continue
out.append(
NucleiFinding(
template_id=str(template_id),
name=info.get("name"),
severity=_severity(info.get("severity")),
matched_at=rec.get("matched-at") or rec.get("matched_at"),
host=rec.get("host"),
reference=[str(r) for r in ref if r],
)
)
return out
def run(
targets: list[str],
*,
timeout: float = 600.0,
rate_limit: int | None = None,
templates_dir: str | None = None,
) -> tuple[list[NucleiFinding], object]:
cmd = [BINARY, "-jsonl", "-silent", "-disable-update-check"]
if templates_dir:
cmd += ["-update-template-dir", templates_dir, "-templates", templates_dir]
if rate_limit:
cmd += ["-rate-limit", str(rate_limit)]
res = run_tool(cmd, timeout=timeout, input_text="\n".join(targets) + "\n")
return normalize(res.stdout), res
@@ -0,0 +1,59 @@
"""searchsploit — Exploit-DB grounding. Native output: ``--json``.
This is the anti-hallucination layer: we ONLY emit EDB-IDs/paths/titles that
searchsploit actually returned. Nothing is fabricated or inferred. The DB ships in
the image, so this requires no network at runtime.
"""
from __future__ import annotations
import json
import logging
from ..schema import ExploitDBMatch
from .base import run_tool
log = logging.getLogger("recon_triage.tools.searchsploit")
BINARY = "searchsploit"
def normalize(raw_json: str, query: str | None = None) -> list[ExploitDBMatch]:
"""Parse ``searchsploit --json`` output into ExploitDBMatch candidates.
Only ``RESULTS_EXPLOIT`` rows are emitted, each marked verified=False.
"""
if not raw_json.strip():
return []
try:
doc = json.loads(raw_json)
except json.JSONDecodeError:
log.warning("searchsploit returned non-JSON output; ignoring")
return []
rows = doc.get("RESULTS_EXPLOIT") or []
out: list[ExploitDBMatch] = []
for row in rows:
edb_id = row.get("EDB-ID") or row.get("Codes")
title = row.get("Title")
path = row.get("Path")
if not edb_id or not title:
continue
out.append(
ExploitDBMatch(
edb_id=str(edb_id),
title=str(title),
path=str(path or ""),
url=row.get("URL"),
query=query,
verified=False,
)
)
return out
def run(term: str, *, timeout: float = 60.0) -> tuple[list[ExploitDBMatch], object]:
# --json gives structured results; the DB is local so no network is used.
cmd = [BINARY, "--json", term]
res = run_tool(cmd, timeout=timeout)
return normalize(res.stdout, query=term), res
@@ -0,0 +1,27 @@
"""subfinder — passive subdomain enumeration. Native output: ``-silent -oJ`` (JSONL)."""
from __future__ import annotations
from .base import parse_jsonl, run_tool
BINARY = "subfinder"
def normalize(raw: str) -> list[str]:
"""Extract unique hostnames from subfinder JSONL output."""
hosts: list[str] = []
seen: set[str] = set()
for rec in parse_jsonl(raw):
host = (rec.get("host") or rec.get("input") or "").strip().lower()
if host and host not in seen:
seen.add(host)
hosts.append(host)
return hosts
def run(domain: str, *, timeout: float = 120.0, rate_limit: int | None = None) -> tuple[list[str], object]:
cmd = [BINARY, "-silent", "-oJ", "-d", domain]
if rate_limit:
cmd += ["-rate-limit", str(rate_limit)]
res = run_tool(cmd, timeout=timeout)
return normalize(res.stdout), res
@@ -0,0 +1 @@
"""Optional triage stage: deterministic ranking + optional LLM enrichment."""
+164
View File
@@ -0,0 +1,164 @@
"""Optional LLM triage stage (OpenAI-compatible). Strictly grounded, fully optional.
Behaviour contract:
- Configured purely by env: LLM_BASE_URL, LLM_MODEL, LLM_API_KEY.
- If LLM_BASE_URL is unset, this stage is skipped and deterministic ranking is used.
- The model is told to use ONLY the provided ReconReport, cite by id, and is
forbidden from inventing CVEs, EDB-IDs, paths, or tools.
- Output is validated against TriageReport. On invalid JSON, retry once, then fall
back to deterministic ranking.
- Acceptance gate: every suggested_next_step / evidence_ref must reference an
identifier present in the input report. Anything else is stripped.
"""
from __future__ import annotations
import json
import logging
import os
import re
from ..schema import ReconReport, TriageItem, TriageReport
from .ranking import collect_identifiers, deterministic_triage
log = logging.getLogger("recon_triage.triage.llm")
SYSTEM_PROMPT = (
"You are a defensive security triage assistant. You are given a ReconReport as "
"JSON describing hosts, services, versions, Exploit-DB candidate references, and "
"nuclei findings from an AUTHORIZED recon run.\n"
"Rules you MUST follow:\n"
"1. Use ONLY information present in the provided ReconReport JSON.\n"
"2. Cite every finding by an identifier that appears in the input "
"(host:port, EDB-<id>, or a nuclei template-id).\n"
"3. You are FORBIDDEN from inventing CVEs, EDB-IDs, file paths, tools, exploits, "
"or any reference not present in the input.\n"
"4. Do NOT provide exploit code, payloads, or step-by-step exploitation. Triage "
"and prioritization only.\n"
"5. Output STRICT JSON matching the TriageReport schema with a "
"'prioritized_findings' array. Each item has: summary, rationale, severity "
"(info|low|medium|high|critical|unknown), evidence_refs (array of ids from the "
"input), suggested_next_step (must reference a real id from the input), and "
"confidence (0..1). Output JSON only, no prose."
)
def is_configured() -> bool:
return bool(os.environ.get("LLM_BASE_URL"))
def _extract_json(text: str) -> dict | None:
text = text.strip()
# Strip markdown fences if present.
if text.startswith("```"):
text = re.sub(r"^```[a-zA-Z]*\n?", "", text)
text = re.sub(r"\n?```$", "", text).strip()
try:
obj = json.loads(text)
return obj if isinstance(obj, dict) else None
except json.JSONDecodeError:
# Try to find the first JSON object in the text.
m = re.search(r"\{.*\}", text, re.DOTALL)
if m:
try:
obj = json.loads(m.group(0))
return obj if isinstance(obj, dict) else None
except json.JSONDecodeError:
return None
return None
def _call_llm(report_json: str, *, base_url: str, model: str, api_key: str, timeout: float) -> str:
"""Single OpenAI-compatible chat completion call. Imported lazily so the rest of
the pipeline never depends on httpx being reachable."""
import httpx
url = base_url.rstrip("/") + "/chat/completions"
request_body = {
"model": model,
"messages": [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": "ReconReport JSON:\n" + report_json},
],
"temperature": 0.0,
"response_format": {"type": "json_object"},
}
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
resp = httpx.post(url, json=request_body, headers=headers, timeout=timeout)
resp.raise_for_status()
data = resp.json()
return data["choices"][0]["message"]["content"]
def _enforce_acceptance(triage: TriageReport, valid_ids: set[str]) -> TriageReport:
"""Strip evidence_refs / suggested_next_step that don't reference real ids.
An item whose suggested_next_step references no real id has it cleared; an item
with no valid evidence at all is dropped entirely.
"""
kept: list[TriageItem] = []
for item in triage.prioritized_findings:
refs = [r for r in item.evidence_refs if _ref_ok(r, valid_ids)]
step = item.suggested_next_step
if step and not _contains_valid_id(step, valid_ids):
log.warning("Stripping suggested_next_step with no valid id: %r", step)
step = None
if not refs:
log.warning("Dropping triage item with no valid evidence: %r", item.summary)
continue
item.evidence_refs = refs
item.suggested_next_step = step
kept.append(item)
return TriageReport(
prioritized_findings=kept,
model=triage.model,
generated_by=triage.generated_by,
)
def _ref_ok(ref: str, valid_ids: set[str]) -> bool:
return ref in valid_ids or _contains_valid_id(ref, valid_ids)
def _contains_valid_id(text: str, valid_ids: set[str]) -> bool:
return any(vid and vid in text for vid in valid_ids)
def run_triage(report: ReconReport, *, timeout: float = 60.0) -> TriageReport:
"""Produce a TriageReport. Uses the LLM if configured, else deterministic ranking.
Always returns a valid TriageReport; never raises for model/network problems.
"""
if not is_configured():
log.info("LLM_BASE_URL unset; using deterministic triage.")
return deterministic_triage(report)
base_url = os.environ["LLM_BASE_URL"]
model = os.environ.get("LLM_MODEL", "qwen2.5:7b-instruct")
api_key = os.environ.get("LLM_API_KEY", "not-needed")
valid_ids = collect_identifiers(report)
report_json = report.model_dump_json()
for attempt in (1, 2):
try:
content = _call_llm(
report_json, base_url=base_url, model=model, api_key=api_key, timeout=timeout
)
except Exception as e:
log.warning("LLM call failed (attempt %d): %s", attempt, e)
break
obj = _extract_json(content)
if obj is None:
log.warning("LLM returned invalid JSON (attempt %d).", attempt)
continue
try:
triage = TriageReport.model_validate(obj)
except Exception as e:
log.warning("LLM output failed schema validation (attempt %d): %s", attempt, e)
continue
triage.generated_by = "llm"
triage.model = model
return _enforce_acceptance(triage, valid_ids)
log.info("Falling back to deterministic triage.")
return deterministic_triage(report)
@@ -0,0 +1,83 @@
"""Deterministic, severity-based triage ranking.
This is the always-available fallback used when no LLM is configured (or when the
LLM output fails validation). It never invents data — every item references real
identifiers drawn from the ReconReport.
"""
from __future__ import annotations
from ..schema import (
SEVERITY_RANK,
ReconReport,
Severity,
TriageItem,
TriageReport,
)
def collect_identifiers(report: ReconReport) -> set[str]:
"""All valid reference ids in a report: host:port, EDB-<id>, nuclei template-ids."""
ids: set[str] = set()
for host in report.hosts:
key = host.key()
ids.add(key)
for port in host.ports:
ids.add(f"{key}:{port.number}")
svc = port.service
if svc:
for m in svc.exploitdb_candidates:
ids.add(f"EDB-{m.edb_id}")
ids.add(m.edb_id)
for f in report.nuclei_findings:
ids.add(f.template_id)
return ids
def deterministic_triage(report: ReconReport) -> TriageReport:
"""Rank findings without any model.
Priority signal: nuclei severity, then services carrying Exploit-DB candidates,
then plain open services.
"""
items: list[TriageItem] = []
# Nuclei findings ranked by severity.
for f in report.nuclei_findings:
items.append(
TriageItem(
summary=f"Nuclei: {f.name or f.template_id} on {f.host or f.matched_at or 'target'}",
rationale=f"Template {f.template_id} matched at {f.matched_at or 'n/a'}.",
severity=Severity(f.severity),
evidence_refs=[f.template_id] + ([f.host] if f.host else []),
suggested_next_step=f"Review nuclei template {f.template_id} and the matched endpoint.",
confidence=0.5,
)
)
# Services with Exploit-DB candidates.
for host in report.hosts:
key = host.key()
for port in host.ports:
svc = port.service
if not svc or not svc.exploitdb_candidates:
continue
product = svc.query_term() or svc.name or "service"
top = svc.exploitdb_candidates[0]
items.append(
TriageItem(
summary=f"{product} on {key}:{port.number} has {len(svc.exploitdb_candidates)} Exploit-DB candidate(s)",
rationale=(
f"Detected {product}; searchsploit returned candidate references. "
"Candidates are UNVERIFIED — confirm version applicability manually."
),
severity=Severity.MEDIUM,
evidence_refs=[f"{key}:{port.number}", f"EDB-{top.edb_id}"],
suggested_next_step=f"Manually verify EDB-{top.edb_id} against {product}.",
confidence=0.4,
)
)
# Sort: by severity rank desc, then confidence desc, stable for reproducibility.
items.sort(key=lambda it: (SEVERITY_RANK.get(str(it.severity), 0), it.confidence), reverse=True)
return TriageReport(prioritized_findings=items, generated_by="deterministic")