#!/usr/bin/env python3 """ cdp-input-invisible: 3-test spike validation. Tests: When Playwright connects over CDP and calls page.mouse.move(), does the page receive mousemove DOM events? 3 conditions: A. VANILLA (no CDP) — control: no input, expect 0 events B. CDP — primary test: no input, expect 0 events (proves CDP itself doesn't fire phantom events) C. VANILLA + INJECT — control: synthetic input, expect >=10 events (proves test framework delivers events) D. CDP + INJECT — primary test: synthetic input, expect 0 events (THE SIGNAL) If D produces 0 events while C produces events → CDP input invisibility is real. If D produces events → CDP routes input normally, the earlier apen-mouse result was an artifact. """ import asyncio, subprocess, time, os, glob, sys, uuid from playwright.async_api import async_playwright PORT = 9351 CHROME = "/usr/bin/google-chrome" if not os.path.exists(CHROME): cands = glob.glob("/home/huxley/.cache/ms-playwright/chromium_headless_shell-*/chrome-headless-shell-linux64/chrome-headless-shell") if cands: CHROME = cands[0] print(f"Using CHROME: {CHROME}") async def run(p, use_cdp, inject_input, run_id): """One test run. Returns the final counters + verdict.""" if use_cdp: udir = f"/tmp/cdp_inv_{uuid.uuid4().hex[:8]}" proc = subprocess.Popen([CHROME, "--headless=new", "--no-sandbox", "--disable-gpu", f"--remote-debugging-port={PORT}", f"--user-data-dir={udir}", "about:blank"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) time.sleep(3) async with async_playwright() as p_ctx: if use_cdp: browser = await p_ctx.chromium.connect_over_cdp(f"http://localhost:{PORT}") ctx = browser.contexts[0] if browser.contexts else await browser.new_context() page = ctx.pages[0] if ctx.pages else await ctx.new_page() else: browser = await p_ctx.chromium.launch(headless=True, args=["--no-sandbox", "--disable-setuid-sandbox", "--disable-dev-shm-usage"]) page = await (await browser.new_context()).new_page() try: await page.goto("http://127.0.0.1:3457/spikes/cdp-input-invisible/demo.html", wait_until="networkidle", timeout=15000) # The demo auto-runs for 8s and exposes window.__dbFinal at the end. # If injecting input, do 10 mouse moves + 5 keydowns in the first 6s. if inject_input: # Wait briefly for the page to be ready await page.wait_for_timeout(300) # Get the visible area center vp = await page.evaluate("() => ({ w: window.innerWidth, h: window.innerHeight })") cx, cy = vp['w'] // 2, vp['h'] // 2 # 10 mousemoves for i in range(10): x = cx + (i * 8) % 60 - 30 y = cy + (i * 5) % 40 - 20 await page.mouse.move(x, y) await page.wait_for_timeout(100) # 5 keydowns for k in "abcde": await page.keyboard.press(k) await page.wait_for_timeout(100) # Wait for the 8s mark + 250ms verdict render await page.wait_for_timeout(8500) final = await page.evaluate("() => window.__dbFinal || null") counters = await page.evaluate("() => ({...window.__counters})") await browser.close() finally: if use_cdp: proc.terminate() time.sleep(1) if proc.poll() is None: proc.kill() return {"final": final, "counters": counters} async def main(): results = {} async with async_playwright() as p: # A: vanilla, no input (control — expect 0) print("=" * 60) print("A. VANILLA, no input (control)") print("=" * 60) results['A'] = await run(p, False, False, 'A') print(f" Counters: {results['A']['counters']}") print(f" Final: {results['A']['final']}") # C: vanilla WITH input (control — expect events) print() print("=" * 60) print("C. VANILLA + synthetic input (control)") print("=" * 60) results['C'] = await run(p, False, True, 'C') print(f" Counters: {results['C']['counters']}") print(f" Final: {results['C']['final']}") # B: CDP, no input (sanity — expect 0) print() print("=" * 60) print("B. CDP, no input (sanity)") print("=" * 60) results['B'] = await run(p, True, False, 'B') print(f" Counters: {results['B']['counters']}") print(f" Final: {results['B']['final']}") # D: CDP WITH input (THE TEST — expect 0 if signal real) print() print("=" * 60) print("D. CDP + synthetic input (THE TEST)") print("=" * 60) results['D'] = await run(p, True, True, 'D') print(f" Counters: {results['D']['counters']}") print(f" Final: {results['D']['final']}") # Verdict print() print("=" * 60) print("SUMMARY") print("=" * 60) a = results['A']['final']['total'] if results['A']['final'] else None b = results['B']['final']['total'] if results['B']['final'] else None c = results['C']['final']['total'] if results['C']['final'] else None d = results['D']['final']['total'] if results['D']['final'] else None print(f" A. Vanilla, no input: total={a} (expect 0, control)") print(f" B. CDP, no input: total={b} (expect 0, sanity)") print(f" C. Vanilla + input: total={c} (expect >0, control)") print(f" D. CDP + input: total={d} (expect 0, SIGNAL if 0)") print() if c and c > 0 and d == 0: print(f" ✅ SIGNAL CONFIRMED. Vanilla delivers input (C={c}), CDP silently drops it (D={d}).") print(f" → Stealth browsers using connect_over_cdp() will have cdp_input_invisible=true") elif c and d and c > 0 and d > 0: ratio = d / c print(f" ⚠️ WEAK SIGNAL. CDP delivered {d}/{c} = {ratio:.1%} of expected events.") print(f" → Investigate threshold; partial delivery may still be useful.") elif d and d > 0: print(f" ❌ NO SIGNAL. CDP delivered {d} events. Earlier apen-mouse result was an artifact.") else: print(f" ❓ INCONCLUSIVE. Run again or check server.") asyncio.run(main())