exploitdb.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# exploitdb.py - Search exploits from exploit-db.com
#
# Version: 1.0
# Author: Mathieu Deous
# License: FreeBSD
#

import cmd
import csv
import os
import re
import shutil
import sys
from collections import defaultdict
from copy import copy
from urllib.request import urlopen
from zipfile import ZipFile

CURRENT_DIR = os.path.realpath(os.path.dirname("."))
EXPLOITS_DIR_ORIG = os.path.join(CURRENT_DIR, "exploitdb-main")
EXPLOITS_DIR = os.path.join(CURRENT_DIR, "exploits")
EXPLOITS_CSV = os.path.join(EXPLOITS_DIR, "files_exploits.csv")
ARCHIVE_PATH = os.path.join(CURRENT_DIR, "master.zip")
LATEST_PATH = os.path.join(CURRENT_DIR, ".latest")
ARCHIVE_URL = (
    "https://gitlab.com/exploit-database/exploitdb/-/archive/main/exploitdb-main.zip"
)


class ExploitSearch(cmd.Cmd):
    prompt: str = "\033[1;31mexploitdb\033[0m\033[1;32m>\033[0m "
    intro: str = (
        "\n \033[40m\033[1;37m"
        "-=[ exploitdb.py - Search exploits from exploit-db.com ]="
        "-\033[0m\n"
    )
    fields: tuple[str] = (
        "id",
        "file",
        "description",
        "date",
        "author",
        "platform",
        "type",
        "port",
    )
    search_regexes: dict[str, re.Pattern] = {
        "plain": [
            re.compile(r'(\w+)?:(?!r[\'"])([^ \'"]+)?'),
            re.compile(r'(\w+)?:(?:\'|")([^"\']+)?'),
        ],
        "regex": [re.compile(r'(\w+):r(?:\'|")([^"\']+)')],
    }
    highlighted_fields_map: dict[str, list[str]] = {
        "id": ["id"],
        "description": ["description"],
        "file": ["type", "platform"],
    }
    csv_file: str
    exploits: list[dict[str, str]]
    fields_value_completion: dict[str, set[str]]

    def __init__(self, csv_file: str = EXPLOITS_CSV):
        self.csv_file = csv_file
        self.exploits = []
        self.load_csv()
        self.fields_value_completion = {
            "platform": {e["platform"] for e in self.exploits},
            "type": {e["type"] for e in self.exploits},
            "port": {e["port"] for e in self.exploits},
        }
        super().__init__()

    @staticmethod
    def get_archive_etag() -> str:
        resp = urlopen(ARCHIVE_URL)
        etag = resp.info().get("etag").strip('"')
        return etag

    @staticmethod
    def download_archive():
        downloaded_size = 0
        block_size = 4096
        with open(ARCHIVE_PATH, "wb") as outfile:
            resp = urlopen(ARCHIVE_URL)
            buff = resp.read(block_size)
            while buff:
                outfile.write(buff)
                downloaded_size += len(buff)
                readable_size, unit = format_bytes(downloaded_size)
                status = (
                    f"Downloading exploits archive... {readable_size:.2f} {unit}"
                    + "   "
                )
                sys.stdout.write(status)
                sys.stdout.write("\b" * (len(status) + 1))
                buff = resp.read(block_size)
            sys.stdout.write("\n")

    def parse_args(self, args: str) -> dict:
        search_args = {"plain": [], "regex": []}
        if ":" not in args:
            search_args["plain"].append(("description", args))
        else:
            for search_type, regexes in self.search_regexes.items():
                for regex in regexes:
                    result = regex.findall(args)
                    if result:
                        for field_name, pattern in result:
                            if field_name not in self.fields:
                                pattern = f"{field_name}:{pattern}"
                                search_args["plain"].append(("description", pattern))
                            else:
                                search_args[search_type].append((field_name, pattern))
        return search_args

    def load_csv(self, startup: bool = True):
        if not os.path.exists(self.csv_file):
            print("Database not found, updating now")
            etag = self.get_archive_etag()
            self.updatedb(etag)
        else:
            if startup:
                print("Checking for new database version... ", end="")
                with open(LATEST_PATH) as infile:
                    current_etag = infile.read().strip()
                latest_etag = self.get_archive_etag()
                if latest_etag != current_etag:
                    print("UPDATE FOUND")
                    self.updatedb(latest_etag)
                else:
                    print("OK")
        with open(self.csv_file) as infile:
            reader = csv.reader(infile)
            for index, entry in enumerate(reader):
                if index == 0:
                    header = entry
                    continue
                exploit = dict(zip(header, entry))
                if exploit["port"] == "0":
                    exploit["port"] = "n/a"
                if not exploit["platform"]:
                    exploit["platform"] = "n/a"
                if "//" in exploit["file"]:
                    exploit["file"] = exploit["file"].replace("//", "/")
                self.exploits.append(exploit)

    def search(self, search_params: str) -> list[tuple[dict, dict]]:
        matches = []
        args = self.parse_args(search_params)
        for exploit in self.exploits:
            matching = True
            for search_type, search_args in args.items():
                if search_type == "plain":
                    for field_name, pattern in search_args:
                        if pattern.lower() not in exploit[field_name].lower():
                            matching = False
                            break
                    if not matching:
                        break
                elif search_type == "regex":
                    for field_name, pattern in search_args:
                        if re.search(pattern, exploit[field_name], flags=re.I) is None:
                            matching = False
                            break
                    if not matching:
                        break
            if matching:
                matches.append((exploit, args))
        return matches

    def do_search(self, line: str):
        """
        search - search database for exploits
        Usage: search field:pattern [field:pattern, ...]
        """
        results = self.search(line)
        for result, args in results:
            flattened_args = defaultdict(list)
            for search_type in ("plain", "regex"):
                for k, v in args[search_type]:
                    flattened_args[k].append(v)
            result = copy(result)
            for field_name, search_vals in self.highlighted_fields_map.items():
                for search_val in search_vals:
                    if search_val in flattened_args:
                        for pattern in flattened_args[search_val]:
                            re_pattern = pattern
                            if field_name not in args["regex"]:
                                re_pattern = re.escape(pattern)
                            result[field_name] = re.sub(
                                re_pattern,
                                lambda matchobj: "\033[1;33m"
                                + matchobj.group(0)
                                + "\033[0m",
                                result[field_name],
                                flags=re.I,
                                count=1,
                            )
            result_str = f"[{result['id']}] {result['description']} - {result['file']}"
            print(result_str)
        print("")

    def complete_search(
        self, _text: str, line: str, _begidx: int, _endidx: int
    ) -> list[str]:
        last_arg = line.split()[-1]
        if last_arg:
            if ":" in last_arg:
                field_name, pattern = last_arg.split(":")
                if pattern:
                    return [
                        f
                        for f in self.fields_value_completion[field_name]
                        if f.startswith(pattern)
                    ]
                return [f for f in self.fields_value_completion[field_name]]
            return [f + ":" for f in self.fields if f.startswith(last_arg)]
        return [f + ":" for f in self.fields]

    def info(self, exploit_id: str) -> dict:
        for exploit in self.exploits:
            if exploit["id"] == exploit_id:
                return exploit
        return None

    def do_info(self, line: str):
        """
        info - get details about given exploit
        Usage: info exploit_id
        """
        result = self.info(line)
        if result is None:
            print(f"No exploit with this ID: {line}\n")
            return
        desc_len = len(result["description"])
        fstring = "{{:<13}} | {{:<{}}}".format(desc_len)
        print(result)
        print(("{{:=^{}}}".format(desc_len + 17)).format(" #%s " % result["id"]))
        print(fstring.format(" Filename ", result["file"] + " "))
        print(fstring.format(" Description ", result["description"] + " "))
        print(fstring.format(" Published ", result["date_published"] + " "))
        print(fstring.format(" Updated ", result["date_updated"] + " "))
        print(fstring.format(" Author ", result["author"] + " "))
        print(fstring.format(" Platform ", result["platform"] + " "))
        print(fstring.format(" Type ", result["type"] + " "))
        print(fstring.format(" Port ", result["port"] + " "))
        print((desc_len + 17) * "=" + "\n")

    def complete_info(self, text: str, _line: str, _begidx: int, _endidx: int):
        if not text:
            return [e["id"] for e in self.exploits]
        else:
            return [e["id"] for e in self.exploits if e["id"].startswith(text)]

    def updatedb(self, etag: str):
        self.download_archive()
        print("Extracting files...")
        if os.path.exists(EXPLOITS_DIR):
            shutil.rmtree(EXPLOITS_DIR)
            os.mkdir(EXPLOITS_DIR)
        with ZipFile(ARCHIVE_PATH) as infile:
            infile.extractall(path=CURRENT_DIR)
        os.remove(ARCHIVE_PATH)
        os.rename(EXPLOITS_DIR_ORIG, EXPLOITS_DIR)
        os.chmod(EXPLOITS_CSV, 0o644)
        self.exploits = []
        with open(LATEST_PATH, "w") as outfile:
            outfile.write(etag)
        self.load_csv(startup=False)
        print("OK\n")

    def do_updatedb(self, _line: str):
        """
        updatedb - update local exploits database
        Usage: updatedb
        """
        etag = self.get_archive_etag()
        self.updatedb(etag)

    def do_show(self, line: str):
        """
        show - display the content of an exploit file
        Usage: show exploit_id|exploit_path
        """
        if line.isdigit():
            field = "id"
        else:
            field = "file"
        found = False
        for exploit in self.exploits:
            if exploit[field] == line:
                found = True
                break
        if not found:
            print(f"Exploit not found: {line}\n")
        else:
            sploit_path = os.path.join(EXPLOITS_DIR, exploit["file"])
            with open(sploit_path) as infile:
                print(infile.read())
                print("")

    def complete_show(
        self, text: str, _line: str, _begidx: str, _endidx: str
    ) -> list[str]:
        completions = []
        if not text:
            completions.extend([s["id"] for s in self.exploits])
            completions.extend([s["file"] for s in self.exploits])
        else:
            completions.extend(
                [s["id"] for s in self.exploits if s["id"].startswith(text)]
            )
            completions.extend(
                [s["file"] for s in self.exploits if s["file"].startswith(text)]
            )
        return completions

    @staticmethod
    def do_EOF(self):
        return True


def format_bytes(size: int | float) -> tuple[float, str]:
    multiple_units = ["KB", "MB", "GB"]
    unit = "B"
    for u in multiple_units:
        if size >= 1024.0:
            size /= 1024.0
            unit = u
    return size, unit


def main(restarted: bool = False):
    es = ExploitSearch()
    if restarted:
        es.intro = "\n"
    try:
        es.cmdloop()
    except KeyboardInterrupt:
        main(True)


if __name__ == "__main__":
    main()