#!/usr/bin/env python3
"""
Linux path-reachability verification harness.

This does NOT trigger the UAF on Linux (the nested message loop in
ExitFullscreenModeForTab is Windows-specific). It only confirms that the
web-content POC reaches the vulnerable assignment site
(digital_identity_request_impl.cc:651) and the synchronous
ExitFullscreen(true) call inside SetWebContentsBlocked.

Run this against an instrumented Chrome build (see chromium-instrumentation.patch).
"""
import json, os, signal, socket, subprocess, sys, threading, time, urllib.request
import websocket  # pip install websocket-client

HERE   = os.path.dirname(os.path.abspath(__file__))
CHROME = os.environ.get("CHROME", "/chromium/chromium/src/out/asan/chrome")
PORT   = 8080
CDP    = 9222

def wait_port(host, port, timeout=30):
    t0 = time.time()
    while time.time() - t0 < timeout:
        try:
            with socket.create_connection((host, port), 1):
                return True
        except OSError:
            time.sleep(0.2)
    return False

class CDPClient:
    def __init__(self, ws_url):
        self.ws = websocket.create_connection(ws_url, timeout=30)
        self.id = 0
        self.sessions = {}  # targetId -> sessionId
        self.pending = {}
        self.events = []
        self.lock = threading.Lock()
        self.reader = threading.Thread(target=self._read, daemon=True)
        self.reader.start()

    def _read(self):
        try:
            while True:
                msg = json.loads(self.ws.recv())
                with self.lock:
                    if 'id' in msg:
                        self.pending[msg['id']] = msg
                    else:
                        self.events.append(msg)
        except Exception:
            pass

    def send(self, method, params=None, session=None, timeout=15):
        with self.lock:
            self.id += 1
            mid = self.id
        m = {"id": mid, "method": method, "params": params or {}}
        if session:
            m["sessionId"] = session
        self.ws.send(json.dumps(m))
        t0 = time.time()
        while time.time() - t0 < timeout:
            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, name, pred=None, timeout=15):
        t0 = time.time()
        while time.time() - t0 < timeout:
            with self.lock:
                for i, e in enumerate(self.events):
                    if e.get("method") == name and (pred is None or pred(e)):
                        return self.events.pop(i)
            time.sleep(0.05)
        return None

def main():
    # 1. HTTP server
    srv = subprocess.Popen([sys.executable, os.path.join(HERE, "server.py"), str(PORT)],
                           stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    wait_port("127.0.0.1", PORT)

    # 2. Chrome
    profile = "/tmp/chrome-poc-profile"
    subprocess.run(["rm", "-rf", profile])
    log_path = os.path.join(HERE, "chrome-stderr.log")
    log_f = open(log_path, "wb")
    chrome = subprocess.Popen([
        CHROME,
        "--no-sandbox", "--no-first-run", "--no-default-browser-check",
        "--user-data-dir=" + profile,
        "--ozone-platform=headless",
        "--remote-debugging-port=%d" % CDP,
        "--remote-allow-origins=*",
        "--disable-popup-blocking",
        "--enable-logging=stderr",
        "about:blank",
    ], stdout=log_f, stderr=subprocess.STDOUT)
    if not wait_port("127.0.0.1", CDP, 60):
        print("CDP port never opened"); return 1
    time.sleep(1.0)

    ws_url = json.load(urllib.request.urlopen(
        f"http://127.0.0.1:{CDP}/json/version"))["webSocketDebuggerUrl"]
    c = CDPClient(ws_url)

    # 3. Find the initial page target
    c.send("Target.setDiscoverTargets", {"discover": True})
    targets = c.send("Target.getTargets")["result"]["targetInfos"]
    page = next(t for t in targets if t["type"] == "page")
    sid = c.send("Target.attachToTarget",
                 {"targetId": page["targetId"], "flatten": True})["result"]["sessionId"]

    c.send("Page.enable", session=sid)
    c.send("Runtime.enable", session=sid)
    c.send("Page.navigate", {"url": f"http://localhost:{PORT}/poc.html"}, session=sid)
    c.wait_event("Page.loadEventFired")
    time.sleep(0.5)

    def click(sid):
        # Runtime.evaluate with userGesture:true grants transient user
        # activation — equivalent to a real click for the purposes of
        # requestFullscreen() and navigator.credentials.get({digital}).
        c.send("Runtime.evaluate",
               {"expression": "document.getElementById('go').click()",
                "userGesture": True}, session=sid)

    def js(expr):
        r = c.send("Runtime.evaluate", {"expression": expr, "returnByValue": True},
                   session=sid)["result"]
        return r.get("result", {}).get("value"), r.get("exceptionDetails")

    print("[runner] page url:", js("location.href")[0])
    print("[runner] has digital:", js("'DigitalCredential' in window")[0])
    print("[runner] btn present:", js("!!document.getElementById('go')")[0])

    print("[runner] click 1: enter fullscreen")
    click(sid)
    time.sleep(2.0)
    print("[runner] log:", js("document.getElementById('log').textContent")[0])
    print("[runner] fullscreenElement:", js("!!document.fullscreenElement")[0])

    print("[runner] click 2: navigator.credentials.get({digital})")
    click(sid)
    time.sleep(4.0)
    print("[runner] log:", js("document.getElementById('log').textContent")[0])

    # 4. Cleanup
    chrome.send_signal(signal.SIGTERM)
    try: chrome.wait(10)
    except Exception: chrome.kill()
    srv.terminate()
    log_f.close()

    # 5. Show relevant browser logs
    print("\n=== [POC] markers from chrome stderr ===")
    with open(log_path, "r", errors="replace") as f:
        for line in f:
            if "[POC]" in line or "AddressSanitizer" in line or "heap-use-after-free" in line:
                print(line.rstrip())
    return 0

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