exploit_search.py

#!/usr/bin/env python3
"""
Multi-Source Exploit Search Tool
Searches: searchsploit (local), exploit-db.com (online), ippsec.rocks (HTB walkthroughs)
"""

import argparse
import subprocess
import requests
import json
import re
import sys
from typing import List, Optional
from dataclasses import dataclass, field
from colorama import Fore, Back, Style, init

init(autoreset=True)

W = 80   # total output width

# ─── Result dataclass ────────────────────────────────────────────────────────

@dataclass
class ExploitResult:
    title:      str
    source:     str
    path:       Optional[str] = None
    url:        Optional[str] = None
    etype:      Optional[str] = None
    platform:   Optional[str] = None
    date:       Optional[str] = None
    video_url:  Optional[str] = None
    timestamps: List[tuple]   = field(default_factory=list)
    edb_id:     Optional[str] = None

# ─── Searcher ─────────────────────────────────────────────────────────────────

class ExploitSearcher:

    IPPSEC_DATASET = (
        "https://raw.githubusercontent.com/IppSec/ippsec.github.io/master/dataset.json"
    )
    EDB_API = "https://www.exploit-db.com/search"

    def __init__(self, query: str, verbose: bool = False, no_color: bool = False):
        self.query   = query
        self.verbose = verbose
        if no_color:
            for obj in (Fore, Back, Style):
                for attr in [a for a in dir(obj) if not a.startswith("_")]:
                    try:
                        setattr(obj, attr, "")
                    except AttributeError:
                        pass

    # ── Banner ────────────────────────────────────────────────────────────────

    def banner(self):
        inner = W - 2
        title = "Multi-Source Exploit Search Tool"
        print()
        print(f"{Fore.CYAN}╔{'═' * inner}╗")
        print(f"║{title:^{inner}}║")
        print(f"╚{'═' * inner}╝{Style.RESET_ALL}")
        print(f"  {Fore.WHITE}Query  :{Style.RESET_ALL} {Fore.YELLOW}{self.query}{Style.RESET_ALL}")
        print(f"  {Fore.WHITE}Sources:{Style.RESET_ALL} searchsploit  •  exploit-db.com  •  ippsec.rocks")
        print()

    # ── SearchSploit ──────────────────────────────────────────────────────────

    def search_searchsploit(self) -> List[ExploitResult]:
        results = []
        proc = None
        print(f"{Fore.GREEN}[*]{Style.RESET_ALL} Searching {Fore.WHITE}searchsploit{Style.RESET_ALL} (local ExploitDB)...")
        try:
            proc = subprocess.run(
                ["searchsploit", "-j", self.query],
                capture_output=True, text=True, timeout=30
            )
            data = json.loads(proc.stdout)
            for e in data.get("RESULTS_EXPLOIT", []) + data.get("RESULTS_SHELLCODE", []):
                results.append(ExploitResult(
                    title    = e.get("Title",            "N/A"),
                    source   = "searchsploit",
                    path     = e.get("Path",             "N/A"),
                    etype    = e.get("Type",             "N/A"),
                    platform = e.get("Platform",         "N/A"),
                    date     = e.get("Date_Published",   "N/A"),
                    edb_id   = e.get("EDB-ID",           "N/A"),
                ))
            print(f"    {Fore.GREEN}+{Style.RESET_ALL}  {len(results)} result(s) found\n")
        except FileNotFoundError:
            print(f"    {Fore.RED}!{Style.RESET_ALL}  Not found -- install with: sudo apt install exploitdb\n")
        except subprocess.TimeoutExpired:
            print(f"    {Fore.RED}!{Style.RESET_ALL}  Timed out\n")
        except json.JSONDecodeError as exc:
            print(f"    {Fore.RED}!{Style.RESET_ALL}  JSON parse error: {exc}\n")
            if self.verbose and proc:
                print(f"       raw: {proc.stdout[:200]}\n")
        except Exception as exc:
            print(f"    {Fore.RED}!{Style.RESET_ALL}  {exc}\n")
        return results

    # ── ExploitDB online ──────────────────────────────────────────────────────

    def search_exploitdb(self) -> List[ExploitResult]:
        results = []
        print(f"{Fore.GREEN}[*]{Style.RESET_ALL} Searching {Fore.WHITE}exploit-db.com{Style.RESET_ALL} (online database)...")
        try:
            resp = requests.get(
                self.EDB_API,
                params={"q": self.query, "draw": "1", "start": "0",
                        "length": "50", "action": "search"},
                headers={
                    "User-Agent":       "Mozilla/5.0 (X11; Linux x86_64)",
                    "X-Requested-With": "XMLHttpRequest",
                    "Accept":           "application/json, text/javascript, */*",
                    "Referer":          "https://www.exploit-db.com/",
                },
                timeout=20,
            )
            if resp.status_code == 200:
                try:
                    data = resp.json()
                    for rec in data.get("data", []):
                        edb_id      = str(rec.get("id", "")).strip()
                        raw_title   = rec.get("description") or rec.get("title") or "N/A"
                        clean_title = re.sub(r"<[^>]+>", "", str(raw_title)).strip()
                        def _label(val):
                            return val.get("label", "N/A") if isinstance(val, dict) else str(val or "N/A")
                        results.append(ExploitResult(
                            title    = clean_title,
                            source   = "exploit-db.com",
                            url      = f"https://www.exploit-db.com/exploits/{edb_id}" if edb_id else None,
                            etype    = _label(rec.get("type")),
                            platform = _label(rec.get("platform")),
                            date     = rec.get("date_published", rec.get("date", "N/A")),
                            edb_id   = edb_id,
                        ))
                    print(f"    {Fore.GREEN}+{Style.RESET_ALL}  {len(results)} result(s) found\n")
                except (ValueError, KeyError):
                    search_url = f"https://www.exploit-db.com/search?q={requests.utils.quote(self.query)}"
                    results.append(ExploitResult(
                        title="(dynamic page -- open link in browser)",
                        source="exploit-db.com", url=search_url, etype="web-link"))
                    print(f"    {Fore.YELLOW}~{Style.RESET_ALL}  Dynamic response; direct link provided\n")
            else:
                print(f"    {Fore.RED}!{Style.RESET_ALL}  HTTP {resp.status_code}\n")
        except requests.Timeout:
            print(f"    {Fore.RED}!{Style.RESET_ALL}  Timed out\n")
        except Exception as exc:
            print(f"    {Fore.RED}!{Style.RESET_ALL}  {exc}\n")
        return results

    # ── IppSec ────────────────────────────────────────────────────────────────

    def search_ippsec(self) -> List[ExploitResult]:
        """
        dataset.json is a flat array -- one record per subtitle line.
        Schema: {machine, videoId, timestamp:{minutes,seconds}, line, tag}
        Academy entries use {machine, academy, line, tag} with no videoId.
        We group matching lines by machine name for display.
        """
        results = []
        print(f"{Fore.GREEN}[*]{Style.RESET_ALL} Searching {Fore.WHITE}ippsec.rocks{Style.RESET_ALL} (HTB video walkthroughs)...")
        try:
            resp = requests.get(
                self.IPPSEC_DATASET,
                headers={"User-Agent": "Mozilla/5.0 (X11; Linux x86_64)"},
                timeout=25,
            )
            if resp.status_code != 200:
                print(f"    {Fore.RED}!{Style.RESET_ALL}  HTTP {resp.status_code}\n")
                return results

            query_lower = self.query.lower()
            grouped = {}  # machine -> {video_url, hits[]}

            for entry in resp.json():
                line = entry.get("line", "")
                if query_lower not in line.lower():
                    continue

                machine  = entry.get("machine", "Unknown")
                video_id = entry.get("videoId", "")
                ts_obj   = entry.get("timestamp", {})

                if video_id and isinstance(ts_obj, dict):
                    mins       = int(ts_obj.get("minutes", 0))
                    secs       = int(ts_obj.get("seconds", 0))
                    total_secs = mins * 60 + secs
                    hh, r      = divmod(total_secs, 3600)
                    mm, ss     = divmod(r, 60)
                    ts_str     = f"{hh:02d}:{mm:02d}:{ss:02d}" if hh else f"{mm:02d}:{ss:02d}"
                    yt_url     = f"https://www.youtube.com/watch?v={video_id}"
                    deep       = f"{yt_url}&t={total_secs}s"
                else:
                    # Academy entry -- link to HTB Academy module
                    academy_id = entry.get("academy", "")
                    ts_str     = "academy"
                    yt_url     = f"https://academy.hackthebox.com/module/details/{academy_id}"
                    deep       = yt_url

                if machine not in grouped:
                    grouped[machine] = {"video_url": yt_url, "hits": []}
                grouped[machine]["hits"].append((ts_str, line.strip(), deep))

            for machine, data in grouped.items():
                results.append(ExploitResult(
                    title      = machine,
                    source     = "ippsec.rocks",
                    video_url  = data["video_url"],
                    etype      = "HTB Walkthrough",
                    timestamps = data["hits"],
                ))

            print(f"    {Fore.GREEN}+{Style.RESET_ALL}  {len(results)} video(s) matched\n")
        except requests.Timeout:
            print(f"    {Fore.RED}!{Style.RESET_ALL}  Timed out\n")
        except Exception as exc:
            print(f"    {Fore.RED}!{Style.RESET_ALL}  {exc}\n")
        return results

    # ── Display ───────────────────────────────────────────────────────────────

    def display(self, ss_res, edb_res, ips_res):
        divider = f"{Fore.CYAN}{'=' * W}{Style.RESET_ALL}"
        print(divider)
        print(f"  {Fore.WHITE}Results for:{Style.RESET_ALL} {Fore.YELLOW}{self.query}{Style.RESET_ALL}")
        print(divider)

        self._render_searchsploit(ss_res)
        self._render_exploitdb(edb_res)
        self._render_ippsec(ips_res)

        print(f"{Fore.CYAN}{'-' * W}{Style.RESET_ALL}")
        print(f"  {Fore.GREEN}Total: {len(ss_res)+len(edb_res)+len(ips_res)}{Style.RESET_ALL}"
              f"   {Fore.WHITE}searchsploit:{Style.RESET_ALL} {len(ss_res)}"
              f"   {Fore.WHITE}exploit-db.com:{Style.RESET_ALL} {len(edb_res)}"
              f"   {Fore.WHITE}ippsec.rocks:{Style.RESET_ALL} {len(ips_res)}")
        print(f"{Fore.CYAN}{'-' * W}{Style.RESET_ALL}\n")

    # ── Section renderers ─────────────────────────────────────────────────────

    def _render_searchsploit(self, results):
        if not results:
            return
        self._section_head("SEARCHSPLOIT", len(results))
        for i, r in enumerate(results, 1):
            badge = self._badge(r.etype)
            plat  = f"{Fore.BLUE}[{r.platform}]{Style.RESET_ALL} " if r.platform and r.platform != "N/A" else ""
            date  = f"  {Fore.WHITE}{r.date}{Style.RESET_ALL}"      if r.date     and r.date     != "N/A" else ""
            print(f"  {Fore.YELLOW}{i:>3}.{Style.RESET_ALL}  {plat}{badge}{r.title}{date}")
            if r.edb_id and r.edb_id != "N/A":
                print(f"        {Fore.WHITE}EDB :{Style.RESET_ALL} {Fore.CYAN}https://www.exploit-db.com/exploits/{r.edb_id}{Style.RESET_ALL}")
            if r.path and r.path != "N/A":
                print(f"        {Fore.WHITE}File:{Style.RESET_ALL} {Fore.CYAN}{r.path}{Style.RESET_ALL}")
            print()

    def _render_exploitdb(self, results):
        if not results:
            return
        self._section_head("EXPLOIT-DB.COM", len(results))
        for i, r in enumerate(results, 1):
            if r.etype == "web-link":
                print(f"        {Fore.CYAN}{r.url}{Style.RESET_ALL}\n")
                continue
            badge = self._badge(r.etype)
            plat  = f"{Fore.BLUE}[{r.platform}]{Style.RESET_ALL} " if r.platform and r.platform != "N/A" else ""
            date  = f"  {Fore.WHITE}{r.date}{Style.RESET_ALL}"      if r.date     and r.date     != "N/A" else ""
            print(f"  {Fore.YELLOW}{i:>3}.{Style.RESET_ALL}  {plat}{badge}{r.title}{date}")
            if r.url:
                print(f"        {Fore.WHITE}URL :{Style.RESET_ALL} {Fore.CYAN}{r.url}{Style.RESET_ALL}")
            print()

    def _render_ippsec(self, results):
        if not results:
            return
        self._section_head("IPPSEC.ROCKS", len(results))
        for i, r in enumerate(results, 1):
            print(f"  {Fore.YELLOW}{i:>3}.{Style.RESET_ALL}  {Fore.WHITE}{r.title}{Style.RESET_ALL}")
            print(f"        {Fore.WHITE}Video:{Style.RESET_ALL} {Fore.CYAN}{r.video_url}{Style.RESET_ALL}")
            if r.timestamps:
                print(f"        {Fore.WHITE}Hits :{Style.RESET_ALL}")
                for ts_str, text, link in r.timestamps:
                    short = (text[:60] + "...") if len(text) > 60 else text
                    print(f"          {Fore.GREEN}{ts_str:<9}{Style.RESET_ALL} {short}")
                    print(f"          {' ' * 9} {Fore.CYAN}{link}{Style.RESET_ALL}")
            print()

    # ── Helpers ───────────────────────────────────────────────────────────────

    def _section_head(self, name: str, count: int):
        label = f" {name}  ({count} result{'s' if count != 1 else ''}) "
        bar   = "-" * (W - len(label) - 4)
        print(f"{Fore.MAGENTA}+--{label}{bar}+{Style.RESET_ALL}")
        print()

    def _badge(self, etype: Optional[str]) -> str:
        if not etype or etype in ("N/A", "web-link"):
            return ""
        palette = {
            "remote":  Fore.RED,
            "local":   Fore.YELLOW,
            "webapps": Fore.BLUE,
            "web":     Fore.BLUE,
            "dos":     Fore.MAGENTA,
        }
        color = next((v for k, v in palette.items() if k in etype.lower()), Fore.WHITE)
        return f"{color}[{etype}]{Style.RESET_ALL} "

    # ── Entry point ───────────────────────────────────────────────────────────

    def run(self):
        self.banner()
        ss_res  = self.search_searchsploit()
        edb_res = self.search_exploitdb()
        ips_res = self.search_ippsec()
        self.display(ss_res, edb_res, ips_res)


# ─── CLI ──────────────────────────────────────────────────────────────────────

def main():
    parser = argparse.ArgumentParser(
        description="Search multiple exploit databases",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""examples:
  %(prog)s "apache 2.4"
  %(prog)s "eternalblue" -v
  %(prog)s "sql injection" --no-color
        """,
    )
    parser.add_argument("query",           help="Search term(s)")
    parser.add_argument("-v", "--verbose", action="store_true", help="Verbose error output")
    parser.add_argument("--no-color",      action="store_true", help="Disable ANSI colours")
    args = parser.parse_args()
    ExploitSearcher(args.query, args.verbose, args.no_color).run()


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print(f"\n{Fore.RED}[!] Aborted{Style.RESET_ALL}")
        sys.exit(130)