CFuzzer.py

import os
import uuid
from urllib.parse import quote

from playwright.sync_api import sync_playwright
from playwright.sync_api import TimeoutError as PWTimeoutError


LOG_DIRS = {
    "S4": "./log/client_S4_log",
    "S5": "./log/client_S5_log",
    "S6": "./log/client_S6_log",
}

SERVER_URL = "http://127.0.0.1:5000"


def _click_tag(page, selector):
    selector = selector.strip()
    try:
        page.wait_for_selector(selector, timeout=3000, state="attached")
        page.locator(selector).first.click(force=True)
        print(f"{selector} clicked")
    except (PWTimeoutError, Exception) as e:
        print(f"[EXC] {type(e).__name__}: {e}")
        return False


def _check_flag_cookie(context, initial_names=None):
    cookies = context.cookies()
    print(cookies)
    if initial_names is None:
        initial_names = set()
    return next(
        (c for c in cookies if c["name"] == "flag" and c["name"] not in initial_names),
        None,
    )


def _save_log(log_dir, html):
    log_path = os.path.abspath(os.path.join(log_dir, f"{uuid.uuid4()}.log"))
    with open(log_path, "w", encoding="utf-8") as f:
        f.write(html)
    print(f"Saved: {log_path}")


def _run_browser_test(url, log_key, click_selector=None):
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        context = browser.new_context()
        page = context.new_page()

        # Record cookies that exist before page load (should be empty for new context)
        initial_names = {c["name"] for c in context.cookies()}

        page.goto(url)
        page.wait_for_timeout(600)
        html = page.content()

        # want too see corpus
        # print(html)

        if click_selector:
            _click_tag(page, click_selector)
            page.wait_for_timeout(600)

        flag = _check_flag_cookie(context, initial_names)
        if flag:
            print(f"[+] Sandbox escape detected via {log_key} | cookie: {flag['name']}={flag['value']}")
            _save_log(LOG_DIRS[log_key], html)
        else:
            print(f"[-] No escape detected | url: {url}")

        browser.close()


def _load_xss_payloads(filepath="xss.txt"):
    """Parse xss.txt into (tag, attribute, payload, function, additional) tuples.
    Format per line: tag,attribute,payload,function[,additional]
    """
    payloads = []
    with open(filepath, encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            # split on first 4 commas so additional may contain commas safely
            parts = [p.strip() for p in line.split(",", 4)]
            if len(parts) < 4:
                continue
            tag = parts[0]
            # skip lines where tag and attribute got merged (e.g. "svg onResize")
            if " " in tag:
                continue
            payloads.append(parts)
    return payloads


def strategy4(tag, attribute, payload, function, additional):
    url = (
        f"{SERVER_URL}/4"
        f"?tag={quote(tag)}"
        f"&attribute={quote(attribute)}"
        f"&payload={quote(payload)}"
        f"&function={quote(function)}"
        f"&additional={quote(additional)}"
    )
    _run_browser_test(url, log_key="S4", click_selector=tag.strip())


def strategy5(tag, attribute, payload="javascript:document.cookie='flag=flag'", additional=""):
    url = (
        f"{SERVER_URL}/5"
        f"?tag={quote(tag)}"
        f"&attribute={quote(attribute)}"
        f"&payload={quote(payload)}"
        f"&additional={quote(additional)}"
    )
    _run_browser_test(url, log_key="S5", click_selector=tag.strip())


def strategy6(test, template_id):
    url = f"{SERVER_URL}/6?test={quote(test)}&tem={quote(template_id)}"
    _run_browser_test(url, log_key="S6")


if __name__ == "__main__":
    xss_payloads = _load_xss_payloads("xss.txt")

    # # Strategy 4: inject HTML elements via DOM APIs exposed to guest code
    for parts in xss_payloads:
        tag        = parts[0]
        attribute  = parts[1]
        payload    = parts[2]
        function   = parts[3]
        additional = parts[4] if len(parts) > 4 else ""
        strategy4(tag, attribute, payload, function, additional)

    # # Strategy 5: trigger scope chain — element created in host, accessed from guest
    # # 'function' field is not used (host template hardcodes appendChild)
    for parts in xss_payloads:
        tag        = parts[0]
        attribute  = parts[1]
        payload    = parts[2]
        additional = parts[4] if len(parts) > 4 else ""
        strategy5(tag, attribute, payload, additional)

    # Strategy 6: invoke policy-allowed APIs (endowment)
    DCC = ["eval", "Function", "setTimeout", "setInterval"]
    DCR = ["open", "location.replace"]
    MI  = ["import"]

    for fn in DCC:
        strategy6(fn, "1")
    for fn in DCR:
        strategy6(fn, "2")
    for fn in MI:
        strategy6(fn, "3")