Крадем чужие эксплойты: практическое руководство по развертыванию honeypot-прокси для перехвата 0/1-day

Введение

Предположим вы увидели, что вышло описание нового CVE для какого-нибудь популярного корпоративного VPN. Соцсети взрываются, на GitHub тишина, а где-то на форуме уже появилась тема с продажей нового эксплоита. Что делать, если у вас нет $50 k, чтобы купить эксплоит, но есть желание его заполучить?

Ждать? Скучно и долго. Предлагаю хитрое решение — украсть его прямо из воздуха (ну почти). Сегодня я разберу концепцию и написание кода необычного honeypot (прокси), который маскируется под уязвимый сервер и перехватывает RAW — запросы атакующих, включая те самые заветные 0/1-day эксплойты.

Немного теории

Как всем известно, honeypot (он же горшок с медом) просто имитирует сервис и пассивно ждет атак. Наш подход слегка агрессивнее. Мы не имитируем, а пропускаем через себя трафик на настоящую уязвимую цель. То есть выступаем в роли прокси.

Алгоритм действий

1. Поиск жертвы: Находим через Zoomeye/Shodan реальный, желательно заброшенный, сервер с нужной версией софта (либо поднимаем свою виртуалку).
2. Развертывание нашего прокси: Поднимаем наш сервер на «видном месте» (VPS с белым IP).
3. Само проксирование: Все запросы к нашему серверу мы пересылаем на целевую уязвимую машину, а ее ответы — обратно атакующему.
4. Перехват и анализ данных: Мы остаемся невидимым человеком по середине (MITM), который детально логирует все, что через него проходит.

Для любого сервиса/сканера наш сервер выглядит как обычный ~~уязвимый~~ хост. А для нас — как открытая книга с его трафиком и полезной нагрузкой.

Практика

Шаг 1 — ищем подставную цель

Наша цель — найти в дикой природе реальный сервер, который подходит под наши нужды.
В нашем случае рассмотрим на примере сплоита для FortiGate SSL-VPN CVE-2023-27997. Ищем необходимую информацию и узнаем версии

Затем идем в Shodan (Zoomeye, Censys). Желательно нам нужны не просто любые FortiGate, а именно подходящие под выбранную уязвимость. Промпт для поиска:

"FortiGate" "6.4.0" "SSL VPN"

Либо более общий запрос, если поисковик не видит версию продукта:

"FortiGate" product:"FortiGate SSL-VPN"

Shodan выдаст список IP-адресов. Здесь смотрим на следующее:

Открытые порты 8443/10443 и т.д. для SSL-VPN, т.е. сервис работает.
В идеале — серверы, которые давно не обновлялись. В Shodan справа подписана дата информации.

Допустим, мы нашли подходящий IP: xxx.xxx.xxx.xxx:10443. Это и будет наш TARGET_HOST.

У нас есть IP-адрес реального, «живого» FortiGate, который играет роль приманки

Шаг 2: Развертывание ловушки — настройка прокси-сервера

Теперь нам нужно создать «зеркало» — наш прокси-сервер, который будет представляться атакующему таким же FortiGate.

Выбор платформы: Арендуем VPS с белым статическим IP-адресом, накатываем удобную ОС, для меня это kali.
Подготовка окружения: Ставим на VPS Python, копируем наш скрипт, генерируем самоподписанные сертификаты (чтобы поднять HTTPS):

openssl req -x509 -newkey rsa:4096 -keyout key.pem -out cert.pem -days 365 -nodes -subj "CN=xxx.xxx.xxx.xxx"

3. Конфигурация: В коде скрипта меняем всего две переменные:

TARGET_HOST = "xxx.xxx.xxx.xxx" # IP нашей "приманки" из  Шага 1
TARGET_PORT = 10443

4. Запускаем наш скрипт. Теперь на нашем VPS на порту 10443 висит HTTPS-сервис, который внешне неотличим от FortiGate.

Теперь перейдем к разбору кода

Динамический роутинг и обработка тела запроса

@app.route("/", defaults={"path": ""}, methods=["GET",  "POST",  "PUT",  "DELETE",  "PATCH",  "HEAD",  "OPTIONS"])
@app.route("/<path:path>",  methods = ["GET",  "POST",  "PUT",  "DELETE",  "PATCH",  "HEAD",  "OPTIONS"])
def proxy(path):
    client_ip = request.remote_addr
    target_url = f"{TARGET_BASE}/{path}"
    if request.query_string:
        target_url +=  f"?{request.query_string.decode('utf-8')}"

    # Самое важное: правильно прочитать и сохранить  тело запроса ДО отправки
    try:
        body =  request.get_data( cache= True )  # Кэшируем  данные для многократного чтения
        body_copy  = body
    except  Exception as e:
        body_copy =  None

Важно:
— defaults={"path": ""} и <path:path>: таким образом ловим все запросы, на любой путь, имитируя поведение настоящего сервера.
— request.get_data( cache= True ): Мы сохраняем тело запроса в переменную, чтобы иметь возможность его проанализировать и потом переслать. Без cache=True поток данных будет прочитан только один раз и мы не сможем его переслать дальше.

Логировать все подряд — скучно и ресурсоемко. Наша цель — моментально реагировать на попытки эксплуатации. Для этого мы внедрили простой, но эффективный механизм сигнатур.

Механизм обнаружения подозрительных пэйлоадов

# Наш лист опасных  паттернов
SUSPICIOUS_PATTERNS = [
    b"nc ", b"netcat", b"/bin/sh", b"/bin/bash", b"bash -i",
    b"wget http", b"curl http", b"python -c", b"php -r",
    b"system(", b"exec(", b"eval(", b"base64", b"/dev/tcp",
    b"powershell", b"cmd.exe", b"whoami", b"../../",
]

def contains_suspicious(data):
    if not data:
        return False
    data_lower = data.lower()
    for pattern in SUSPICIOUS_PATTERNS:
        if pattern.lower() in data_lower:
            return True
    return False

alert = False
alert_reasons = []

# Проверяем тело запроса
if body_copy and contains_suspicious(body_copy):
    alert =  True
    alert_reasons.append("suspicious  payload in body")

# Проверяем ЗАГОЛОВКИ! Часто именно там  скрывается шелл
for header_name, header_value in request.headers.items():
    if contains_suspicious(header_value.encode()):
        alert =  True
        alert_reasons.append(f"suspicious header: {header_name}")

Многие эксплойты, особенно для RCE, передают команды не в теле, а в заголовках (например, User-Agent, X-Forwarded-For, Cookie). Проверка только тела запроса — огромная слепая зона

Вся ценность — в данных. Мы не просто пишем логи в текстовый файл, а структурируем их в JSON для последующего анализа.

def log_request_response(client_ip, req, resp=None, direction="FORWARD", req_body=None, alert=False, alert_reasons=None):
    """Функция  для детального логирования всего и  вся"""
    timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]

    log_entry = {
        "timestamp": timestamp,
        "client_ip": client_ip,
        "method": req.method,
        "full_path": req.full_path,
        "headers": dict(req.headers),
        "body_preview": req_body.decode('utf-8', errors='replace')[:2000] if req_body  else None,
        "alert_triggered": alert,
        "alert_reasons":  alert_reasons,
    }

    if resp is not  None:
        log_entry["response_status"] = resp.status_code
        log_entry["response_headers"] = dict(resp.headers)
        log_entry["response_body_preview"] = getattr(resp, 'text', '')[:2000]

    log_filepath = os.path.join(LOGS_DIR,  f"{client_ip.replace(':', '_')}.jsonl")
    with open(log_filepath, "a", encoding='utf-8') as log_file:
        json.dump(log_entry, log_file, ensure_ascii=False)
        log_file.write("\n")

Весь код полностью будет в конце, т.к. он достаточно объемный.

Шаг 3: Магия в действии

Вот что происходит, когда атакующий находит наш сервер и отправляет на него запрос с эксплойтом.

Счастливый обладатель 0/1-day сканирует интернет. Его сканер находит наш VPS (наш-ip:10443). Он видит SSL-сертификат и баннер, идентичные настоящему FortiGate (потому что это ответ от реального сервера). Затем атакующий запускает свой скрипт, отправляет HTTP-запрос с эксплойтом на наш ip:10443

Наш прокси принимает этот запрос. Он моментально анализирует его на предмет подозрительных сигнатур. Не задерживая пакеты, полностью переслылает исходный, нетронутый запрос дальше, на реальную систему (xxx.xxx.xxx.xxx:10443).

Уязвимый сервер обрабатывает запрос так, как будто он пришел напрямую.

Если эксплойт валидный, он может даже выполниться на той машине (поэтому так важно использовать реальную цель). Затем сервер отправляет ответ прямиком на наш прокси, который логирует его и пересылает дальше атакующему

Для атакующего весь цикл выглядит абсолютно нормально. Он получил ожидаемый ответ от «уязвимого сервера» и даже не подозревает, что его трафик был полностью перехвачен и проанализирован.

Шаг 4: Перехват и анализ — Сбор урожая

Пока атакующий радуется успешному выполнению команды, мы уже владеем его сплоитом.

Логи: Все сырые данные (заголовки, тело запроса, ответ) аккуратно складываются в JSON-логи, сгруппированные по IP-адресу источника.
Алерты: Если в запросе были подозрительные паттерны, скрипт моментально выводит их в консоль: ALERT from 94.130.12.77: suspicious header: User-Agent. При желании можно настроить на эти алерты уведомления в Telegram.
Анализ: После получения алерта мы открываем файл logs/94.130.12_77.log и видим полный, готовый к использованию HTTP-запрос, который применил атакующий. Это может быть curl-команда или готовый скрипт на Python. Мы буквально воруем эксплойт из потока данных.

Итоги

Проделанная работа наглядно демонстрирует простую, но мощную мысль: в гонке за 0-day не всегда нужно быть первым покупателем. Иногда достаточно быть самым хитрым наблюдателем

Понятно, что код далеко не идеален, но каждый может модернизировать и доработать его под себя, спасибо, что прочитали

Обещанный код:

#!/usr/bin/env python3 
from flask import Flask,  request, Response
import requests
import os
import json
from datetime import datetime
from urllib3.exceptions import InsecureRequestWarning

requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning)

app = Flask(__name__)

TARGET_HOST = "xxxxxxxxxxxxx"  #  надо заменить
TARGET_PORT = 10443
TARGET_BASE = f"https://{TARGET_HOST}:{TARGET_PORT}" #  надо заменить
LOGS_DIR = "logs"
os.makedirs(LOGS_DIR,  exist_ok=True)

SUSPICIOUS_PATTERNS = [
    b"nc ",
    b"netcat",
    b"/bin/sh",
    b"/bin/bash",
    b"bash -i",
    b"rm -rf",
    b"wget http",
    b"curl http",
    b"python -c",
    b"perl -e",
    b"php -r",
    b"system(",
    b"exec(",
    b"passthru(",
    b"shell_exec(",
    b"popen(",
    b"eval(",
    b"base64",
    b"reverse shell",
    b"bash -c",
    b"mkfifo",
    b"/dev/tcp",
    b"telnet ",
    b"powershell",
    b"cmd.exe",
    b"whoami",
    b"../../",
    b"..\\",
    b"cmd /c",
    b"start ",
]

def get_log_path(client_ip):
    safe_ip = client_ip.replace(":",  "_") 
    return os.path.join(LOGS_DIR, f"{safe_ip}.log" )


def contains_suspicious(data ):
    if not  data:
        return  False
    data_lower  = data.lower()
    for pattern in SUSPICIOUS_PATTERNS:
        if pattern.lower()  in  data_lower:
            return  True
    return  False

def log_request_response(client_ip,  req,  resp= None,  direction = "FORWARD",  req_body = None, alert=False, alert_reasons=None):
    timestamp = datetime.now().strftime( '%Y-%m-%d %H:%M:%S.%f')[:-3]
    log_entry = {
        "timestamp": timestamp,
        "direction": direction,
        "client_ip": client_ip,
        "method": req.method,
        "path": req.path,
        "query_string": req.query_string.decode('utf-8') if req.query_string else "",
        "headers": dict(req.headers),
        "body_preview": req_body.decode('utf-8', errors='replace')[:2000] if req_body else "<no body>",
        "alert": alert,
        "alert_reasons": alert_reasons or [] }

    if resp is not  None:
        log_entry["response_status"] = resp.status_code
        log_entry["response_headers"]=dict(resp.headers)
        try:
            resp_text = resp.text
            log_entry["response_body_preview"] = resp_text[:2000]
        except:
            log_entry["response_body_preview"] = "<binary  or decсode error>"

    log_path = get_log_path(client_ip)
    with open(log_path, "a", encoding="utf-8") as f:
        f.write(json.dumps(log_entry, ensure_ascii=False, indent=2) +  "\n")
        f.write("-"  * 80 + "\n")

    if  alert:
        print(f" AlERT  from {client_ip}: {', '.join(alert_reasons)}")

@app.route("/", defaults={"path": ""}, methods=["GET" , "POST",  "PUT",  "DELETE",  "PATCH", "HEAD",  "OPTIONS"])
@app.route("/<path:path>", methods=["GET", "POST",  "PUT",  "DELETE",  "PATCH",  "HEAD",  "OPTIONS"])
def proxy(path):
    client_ip = request.remote_addr
    target_url = f"{TARGET_BASE}/{path}"
    if request.query_string:
        target_url += f"?{request.query_string.decode('utf-8')}"

    try:
        body = request.get_data(cache=True)
        body_copy = body
    except Exception as e:
        body_copy = None

    headers = {key: value for key, value in request.headers if key.lower() != "host"}

    alert  = False
    alert_reasons =  []

    if body_copy and contains_suspicious(body_copy):
        alert = True
        alert_reasons.append("suspicious payload in body")

    for k, v in request.headers.items():
        if contains_suspicious(v.encode('utf-8', errors='ignore')):
            alert = True
            alert_reasons.append(f"suspicious header: {k}")

    if alert:
        log_entry = {
            "timestamp": datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3],
            "direction": "ALERT",
            "client_ip": client_ip,
            "method": request.method,
            "url": request.url,
            "headers": dict(request.headers),
            "body_preview": body_copy.decode('utf-8', errors='replace') if body_copy else "<no body>",
            "alert": True,
            "alert_reasons": alert_reasons
        }
        log_path = get_log_path(client_ip)
        with open(log_path, "a", encoding="utf-8") as f:
            f.write(json.dumps(log_entry, ensure_ascii=False, indent=2) + "\n")
            f.write("-" * 80 + "\n")
        print(f"  ALERT from  {client_ip}: {', '.join(alert_reasons)}")

    try:
        resp = requests.request(
            method=request.method,
            url=target_url,
            headers=headers,
            data=body_copy,
            verify=False,
            allow_redirects=False,
            timeout=30
        )

        log_request_response(
            client_ip, request, resp,
            direction="FORWARD",
            req_body=body_copy,
            alert=alert,
            alert_reasons=alert_reasons
        )

        response_headers = [(k, v) for k, v in resp.headers.items() if k.lower() != "transfer-encoding"]
        return Response(
            resp.content,
            status=resp.status_code,
            headers=response_headers
        )

    except requests.exceptions.RequestException  as e:
        error_msg = str(e)

        log_entry =  {
            "timestamp": datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3],
            "direction": "ERROR_WITH_ALERT" if alert else "ERROR",
            "client_ip": client_ip,
            "method": request.method,
            "url": request.url,
            "headers": dict(request.headers),
            "body_preview": body_copy.decode('utf-8', errors='replace') if body_copy else "<no body>",
            "error": error_msg,
            "alert": alert,
            "alert_reasons": alert_reasons
        }
        log_path =  get_log_path(client_ip)
        with open(log_path, "a", encoding="utf-8")  as f:
            f.write(json.dumps(log_entry, ensure_ascii=False, indent=2) + "\n")
            f.write("-" * 80 + "\n")

        if alert:
            print(f"ALERT  (error path) from  {client_ip}: {', '.join(alert_reasons)}")

        return Response("Target request failed", status=502)

if __name__  ==  "__main__":
    CERT_FILE =  "cert.pem"
    KEY_FILE="key.pem"

    if not (os.path.exists(CERT_FILE) and  os.path.exists(KEY_FILE)):
        print("Сертификаты не найдены.  Нужно создать cert.pem и key.pem...")
        print("Выполни: openssl req -x509 -newkey rsa:4096 -keyout key.pem -out cert.pem -days 365 -nodes -subj '/CN=0.0.0.0'")
        print("Или нажми Ctrl+C  и создай их вручную.")
        exit(1)

    print(f"Прокси запущен на  0.0.0.0:10443")
    print(f"Целевой  хост: {TARGET_BASE}")
    print(f"Логи по  IP: папка '{LOGS_DIR}/'")

    app.run(host="0.0.0.0", port=10443, ssl_context=(CERT_FILE, KEY_FILE))

Присоединяйся к обсуждению

Крадем чужие эксплойты: практическое руководство по развертыванию honeypot-прокси для перехвата 0/1-day

Введение

Немного теории

Алгоритм действий

Практика

Шаг 1 — ищем подставную цель​

Шаг 2: Развертывание ловушки — настройка прокси-сервера​

Теперь перейдем к разбору кода​

Шаг 3: Магия в действии​

Шаг 4: Перехват и анализ — Сбор урожая​

More posts