hilt.py

#!/usr/bin/env python3
"""
LANCE - Recon Engine
External exposure detection and intelligence gathering module.
"""

import argparse
import subprocess
import json
import os
import sys
import ipaddress
import shutil
from datetime import datetime
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor, as_completed

# ─────────────────────────────────────────────
# CONFIG
# ─────────────────────────────────────────────

WORDLIST_PATHS = [
    "/usr/share/wordlists/dirb/common.txt",
    "/usr/share/seclists/Discovery/Web-Content/common.txt",
    "/usr/share/dirbuster/wordlists/directory-list-2.3-medium.txt",
]

RESULTS_DIR = Path("results")

BANNER = r"""
██╗      █████╗ ███╗   ██╗ ██████╗███████╗
██║     ██╔══██╗████╗  ██║██╔════╝██╔════╝
██║     ███████║██╔██╗ ██║██║     █████╗  
██║     ██╔══██║██║╚██╗██║██║     ██╔══╝  
███████╗██║  ██║██║ ╚████║╚██████╗███████╗
╚══════╝╚═╝  ╚═╝╚═╝  ╚═══╝ ╚═════╝╚══════╝
  RECON ENGINE  |  aeternus  |  LANCE v0.1
"""

# ─────────────────────────────────────────────
# UTILITIES
# ─────────────────────────────────────────────

def log(msg, level="INFO"):
    colors = {"INFO": "\033[94m", "OK": "\033[92m", "WARN": "\033[93m", "ERROR": "\033[91m", "PHASE": "\033[96m"}
    reset = "\033[0m"
    prefix = colors.get(level, "") + f"[{level}]" + reset
    print(f"{prefix} {msg}")

def run_cmd(cmd, timeout=300):
    """Run a shell command and return stdout, stderr, returncode."""
    try:
        result = subprocess.run(
            cmd, shell=True, capture_output=True, text=True, timeout=timeout
        )
        return result.stdout.strip(), result.stderr.strip(), result.returncode
    except subprocess.TimeoutExpired:
        return "", f"Command timed out after {timeout}s", 1
    except Exception as e:
        return "", str(e), 1

# Global timeout — set from CLI args, used by all phases
TOOL_TIMEOUT = 300

def tool_exists(tool):
    return shutil.which(tool) is not None

def check_tools():
    required = ["subfinder", "httpx", "nmap"]
    optional = ["amass", "gobuster"]
    missing_required = []
    missing_optional = []

    for t in required:
        if not tool_exists(t):
            missing_required.append(t)
    for t in optional:
        if not tool_exists(t):
            missing_optional.append(t)

    if missing_required:
        log(f"Missing required tools: {', '.join(missing_required)}", "ERROR")
        log("Install with: sudo apt install subfinder httpx-toolkit nmap", "ERROR")
        sys.exit(1)
    if missing_optional:
        log(f"Optional tools not found (skipping): {', '.join(missing_optional)}", "WARN")

    return {t: tool_exists(t) for t in required + optional}

def find_wordlist():
    for path in WORDLIST_PATHS:
        if os.path.exists(path):
            return path
    log("No wordlist found. Gobuster will be skipped.", "WARN")
    return None

def is_ip_or_cidr(target):
    try:
        ipaddress.ip_network(target, strict=False)
        return True
    except ValueError:
        return False

def ensure_results_dir(target):
    safe_name = target.replace("/", "_").replace(":", "_")
    output_dir = RESULTS_DIR / safe_name
    output_dir.mkdir(parents=True, exist_ok=True)
    return output_dir

# ─────────────────────────────────────────────
# RECON PHASES
# ─────────────────────────────────────────────

def phase_subdomain_enum(target, output_dir, tools):
    """Phase 1: Subdomain enumeration via subfinder + amass — run concurrently."""
    log(f"[Phase 1] Subdomain Enumeration → {target} (concurrent)", "PHASE")
    subdomains = set()
    lock = __import__("threading").Lock()

    def run_subfinder():
        log("Running subfinder...", "INFO")
        stdout, stderr, rc = run_cmd(
            f"subfinder -d {target} -silent -o {output_dir}/subfinder.txt",
            timeout=TOOL_TIMEOUT
        )
        if rc == 0 and (output_dir / "subfinder.txt").exists():
            with open(output_dir / "subfinder.txt") as f:
                found = [line.strip() for line in f if line.strip()]
            with lock:
                subdomains.update(found)
            log(f"subfinder → {len(found)} subdomains", "OK")
        else:
            log(f"subfinder error: {stderr[:150]}", "WARN")

    def run_amass():
        if not tools.get("amass"):
            log("amass not available, skipping.", "WARN")
            return
        log("Running amass (passive)...", "INFO")
        stdout, stderr, rc = run_cmd(
            f"amass enum -passive -d {target} -o {output_dir}/amass.txt",
            timeout=TOOL_TIMEOUT
        )
        if rc == 0 and (output_dir / "amass.txt").exists():
            with open(output_dir / "amass.txt") as f:
                found = [line.strip() for line in f if line.strip()]
            with lock:
                subdomains.update(found)
            log(f"amass → {len(found)} subdomains", "OK")
        else:
            log(f"amass error or no results: {stderr[:100]}", "WARN")

    # Fire both tools simultaneously
    with ThreadPoolExecutor(max_workers=2) as executor:
        futures = [executor.submit(run_subfinder), executor.submit(run_amass)]
        for f in as_completed(futures):
            f.result()  # surface any exceptions

    # Always include root domain
    subdomains.add(target)
    all_subs = sorted(subdomains)

    with open(output_dir / "all_subdomains.txt", "w") as f:
        f.write("\n".join(all_subs))

    log(f"Total unique subdomains: {len(all_subs)}", "OK")
    return all_subs


def phase_live_probe(targets, output_dir):
    """Phase 2: Probe live hosts and fingerprint web tech via httpx."""
    log(f"[Phase 2] Live Host Probing + Tech Fingerprinting ({len(targets)} targets)", "PHASE")

    input_file = output_dir / "all_subdomains.txt"
    httpx_out = output_dir / "httpx.json"

    stdout, stderr, rc = run_cmd(
        f"httpx -l {input_file} -json -silent "
        f"-title -status-code -tech-detect -web-server "
        f"-content-length -follow-redirects "
        f"-o {httpx_out}",
        timeout=TOOL_TIMEOUT
    )

    live_hosts = []
    if httpx_out.exists():
        with open(httpx_out) as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    entry = json.loads(line)
                    live_hosts.append(entry)
                except json.JSONDecodeError:
                    continue
        log(f"Live hosts detected: {len(live_hosts)}", "OK")
    else:
        log(f"httpx produced no output. stderr: {stderr[:200]}", "WARN")

    return live_hosts


def phase_port_scan(targets, output_dir):
    """Phase 3: Port scanning via nmap."""
    log(f"[Phase 3] Port Scanning ({len(targets)} targets)", "PHASE")

    scan_results = {}

    def scan_target(t):
        log(f"Scanning ports on {t}...", "INFO")
        nmap_out = output_dir / f"nmap_{t.replace('.', '_').replace('/', '_')}.xml"
        stdout, stderr, rc = run_cmd(
            f"nmap -sV -T4 --open -oX {nmap_out} {t}",
            timeout=TOOL_TIMEOUT
        )
        stdout_txt, _, _ = run_cmd(f"nmap -sV -T4 --open {t}", timeout=TOOL_TIMEOUT)
        return t, stdout_txt, str(nmap_out)

    # Run scans in parallel (max 5 concurrent)
    with ThreadPoolExecutor(max_workers=5) as executor:
        futures = {executor.submit(scan_target, t): t for t in targets[:20]}  # cap at 20
        for future in as_completed(futures):
            t, output, xml_path = future.result()
            scan_results[t] = {"raw": output, "xml": xml_path}
            # Extract open ports summary
            open_ports = [line for line in output.splitlines() if "/tcp" in line or "/udp" in line]
            log(f"{t} → {len(open_ports)} open ports", "OK" if open_ports else "INFO")

    # Save combined port scan results
    with open(output_dir / "port_scan_summary.json", "w") as f:
        json.dump(scan_results, f, indent=2)

    return scan_results


def phase_dir_bruteforce(live_hosts, output_dir, wordlist, tools):
    """Phase 4: Directory/content bruteforcing via gobuster."""
    if not tools.get("gobuster"):
        log("gobuster not available, skipping dir bruteforce.", "WARN")
        return {}

    if not wordlist:
        log("No wordlist available, skipping dir bruteforce.", "WARN")
        return {}

    log(f"[Phase 4] Directory Bruteforcing ({len(live_hosts)} live hosts)", "PHASE")
    log(f"Wordlist: {wordlist}", "INFO")

    dir_results = {}

    def bruteforce_host(host_entry):
        url = host_entry.get("url") or host_entry.get("input")
        if not url:
            return None, None

        safe_name = url.replace("://", "_").replace("/", "_").replace(".", "_")
        out_file = output_dir / f"gobuster_{safe_name}.txt"

        log(f"Bruteforcing {url}...", "INFO")
        stdout, stderr, rc = run_cmd(
            f"gobuster dir -u {url} -w {wordlist} -q -o {out_file} "
            f"--timeout 10s -t 30 --no-error",
            timeout=TOOL_TIMEOUT
        )

        findings = []
        if out_file.exists():
            with open(out_file) as f:
                findings = [line.strip() for line in f if line.strip()]

        log(f"{url} → {len(findings)} paths found", "OK" if findings else "INFO")
        return url, findings

    # Run in parallel (max 3 concurrent to avoid being too noisy)
    with ThreadPoolExecutor(max_workers=3) as executor:
        futures = {executor.submit(bruteforce_host, h): h for h in live_hosts[:10]}  # cap at 10
        for future in as_completed(futures):
            url, findings = future.result()
            if url:
                dir_results[url] = findings or []

    with open(output_dir / "dir_bruteforce.json", "w") as f:
        json.dump(dir_results, f, indent=2)

    return dir_results


# ─────────────────────────────────────────────
# REPORT BUILDER
# ─────────────────────────────────────────────

def build_report(target, subdomains, live_hosts, port_scan, dir_results, output_dir):
    """Assemble unified JSON report for consumption by vuln scanner."""
    log("[Report] Assembling unified recon report...", "PHASE")

    report = {
        "meta": {
            "target": target,
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "engine": "LANCE Recon Engine v0.1",
            "operator": "aeternus"
        },
        "summary": {
            "total_subdomains": len(subdomains),
            "live_hosts": len(live_hosts),
            "scanned_targets": len(port_scan),
            "bruteforced_hosts": len(dir_results),
        },
        "subdomains": subdomains,
        "live_hosts": live_hosts,
        "port_scan": port_scan,
        "directory_bruteforce": dir_results,
        "attack_surface": []
    }

    # Build attack surface entries — normalized for vuln scanner
    for host in live_hosts:
        url = host.get("url") or host.get("input", "")
        entry = {
            "url": url,
            "ip": host.get("host", ""),
            "status_code": host.get("status_code"),
            "title": host.get("title", ""),
            "webserver": host.get("webserver", ""),
            "technologies": host.get("tech", []),
            "content_length": host.get("content_length"),
            "open_paths": dir_results.get(url, []),
            "ports": []
        }

        # Pull port data for this host
        hostname = url.replace("https://", "").replace("http://", "").split("/")[0]
        if hostname in port_scan:
            raw = port_scan[hostname].get("raw", "")
            open_ports = [line.strip() for line in raw.splitlines() if "/tcp" in line or "/udp" in line]
            entry["ports"] = open_ports

        report["attack_surface"].append(entry)

    report_path = output_dir / "recon_report.json"
    with open(report_path, "w") as f:
        json.dump(report, f, indent=2)

    log(f"Report saved → {report_path}", "OK")
    return report, report_path


def print_summary(report):
    s = report["summary"]
    m = report["meta"]
    print("\n" + "─" * 50)
    print(f"  RECON COMPLETE  |  {m['timestamp']}")
    print("─" * 50)
    print(f"  Target          : {m['target']}")
    print(f"  Subdomains      : {s['total_subdomains']}")
    print(f"  Live Hosts      : {s['live_hosts']}")
    print(f"  Scanned Targets : {s['scanned_targets']}")
    print(f"  Bruteforced     : {s['bruteforced_hosts']}")
    print("─" * 50)
    print(f"  Full report     : results/{m['target'].replace('/', '_')}/recon_report.json")
    print("─" * 50 + "\n")


# ─────────────────────────────────────────────
# MAIN
# ─────────────────────────────────────────────

def run_recon(target, is_ip=False, timeout=300):
    global TOOL_TIMEOUT
    TOOL_TIMEOUT = timeout

    output_dir = ensure_results_dir(target)
    tools = check_tools()
    wordlist = find_wordlist()

    log(f"Target: {target}", "INFO")
    log(f"Output: {output_dir}", "INFO")
    log(f"Timeout per tool: {TOOL_TIMEOUT}s", "INFO")

    if is_ip:
        # IP/CIDR: skip subdomain enum, go straight to port scan
        targets = [str(ip) for ip in ipaddress.ip_network(target, strict=False).hosts()]
        targets = targets[:50]  # safety cap
        subdomains = targets
        live_hosts = phase_live_probe(targets, output_dir)
    else:
        subdomains = phase_subdomain_enum(target, output_dir, tools)
        live_hosts = phase_live_probe(subdomains, output_dir)

    # Extract hostnames for port scan
    hostnames = list(set(
        h.get("host") or h.get("input", "").replace("https://", "").replace("http://", "").split("/")[0]
        for h in live_hosts
    ))
    hostnames = [h for h in hostnames if h]

    port_scan = phase_port_scan(hostnames if hostnames else [target], output_dir)
    dir_results = phase_dir_bruteforce(live_hosts, output_dir, wordlist, tools)

    report, report_path = build_report(target, subdomains, live_hosts, port_scan, dir_results, output_dir)
    print_summary(report)

    return report_path


def main():
    print(BANNER)
    parser = argparse.ArgumentParser(
        description="LANCE Recon Engine — External exposure detection",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  python3 recon.py -d example.com
  python3 recon.py -l domains.txt
  python3 recon.py -i 192.168.1.0/24
  python3 recon.py -d example.com --timeout 120
        """
    )

    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("-d", "--domain", help="Single target domain")
    group.add_argument("-l", "--list", help="File containing list of domains")
    group.add_argument("-i", "--ip", help="IP address or CIDR range")

    parser.add_argument(
        "--timeout", type=int, default=300, metavar="SECONDS",
        help="Timeout in seconds for each tool (default: 300)"
    )

    args = parser.parse_args()

    if args.timeout < 10:
        log("Timeout must be at least 10 seconds.", "ERROR")
        sys.exit(1)

    if args.domain:
        run_recon(args.domain, timeout=args.timeout)

    elif args.list:
        if not os.path.exists(args.list):
            log(f"File not found: {args.list}", "ERROR")
            sys.exit(1)
        with open(args.list) as f:
            domains = [line.strip() for line in f if line.strip() and not line.startswith("#")]
        log(f"Loaded {len(domains)} domains from {args.list}", "INFO")
        for domain in domains:
            log(f"\n{'='*40}\nProcessing: {domain}\n{'='*40}", "PHASE")
            run_recon(domain, timeout=args.timeout)

    elif args.ip:
        if not is_ip_or_cidr(args.ip):
            log(f"Invalid IP/CIDR: {args.ip}", "ERROR")
            sys.exit(1)
        run_recon(args.ip, is_ip=True, timeout=args.timeout)


if __name__ == "__main__":
    main()