CORS 预检取决于请求形态

import json
import sys
from http.server import BaseHTTPRequestHandler, HTTPServer
from threading import Event, Thread
from urllib.request import HTTPRedirectHandler, ProxyHandler, Request, build_opener

events = []


class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, *args, **kwargs):
        raise ValueError("This fixture does not follow redirects")


def open_owned_preflight(request):
    endpoint = api.socket.getsockname()
    if (endpoint != ("127.0.0.1", api.server_port)
            or type(endpoint[1]) is not int or not 1024 <= endpoint[1] <= 65535
            or request.full_url not in {f"http://127.0.0.1:{endpoint[1]}/{p}" for p in ("deny", "permit")}
            or request.get_method() != "OPTIONS" or request.data is not None
            or request.has_proxy()):
        raise ValueError("Only this fixture's owned loopback preflights are allowed")
    return direct_opener.open(request, timeout=5)


class API(BaseHTTPRequestHandler):
    def log_message(self, *args):
        pass

    def reply(self, preflight=False):
        events.append({"method": self.command, "path": self.path,
                       "requested_method": self.headers.get("Access-Control-Request-Method"),
                       "requested_headers": self.headers.get("Access-Control-Request-Headers")})
        body = b"" if preflight else b"readable result"
        self.send_response(204 if preflight else 200)
        if not preflight:
            self.send_header("Content-Length", str(len(body)))
        self.send_header("Access-Control-Allow-Origin", "*" if self.path == "/wildcard" else app_origin)
        self.send_header("Access-Control-Allow-Credentials", "true")
        self.send_header("Vary", "Origin")
        if preflight and self.path == "/permit":
            self.send_header("Access-Control-Allow-Methods", "PUT")
            self.send_header("Access-Control-Allow-Headers", "X-Lab")
            self.send_header("Access-Control-Max-Age", "0")
        self.end_headers()
        self.wfile.write(body)

    def do_OPTIONS(self):
        self.reply(preflight=True)

    def do_GET(self):
        self.reply()

    def do_PUT(self):
        self.rfile.read(int(self.headers.get("Content-Length", "0")))
        self.reply()


class App(BaseHTTPRequestHandler):
    def log_message(self, *args):
        pass

    def do_GET(self):
        if self.path == "/events":
            body = json.dumps(events).encode()
            kind = "application/json"
        else:
            body = ("""<!doctype html><meta charset="utf-8"><title>CORS preflight lab</title>
<h1>Credentials mode and preflight</h1><pre id="result">Running...</pre><script>
const api=""" + json.dumps(api_origin) + """;
async function run() {
  const cases=[
    ['wildcardInclude','/wildcard',{credentials:'include'}],
    ['wildcardOmit','/wildcard',{credentials:'omit'}],
    ['exactInclude','/exact',{credentials:'include'}],
    ['deniedPut','/deny',{method:'PUT',headers:{'X-Lab':'demo'},body:'synthetic'}],
    ['allowedPut','/permit',{method:'PUT',headers:{'X-Lab':'demo'},body:'synthetic'}]
  ];
  const results={};
  for(const [name,path,options] of cases) {
    try {const response=await fetch(api+path,options);results[name]={readable:true,status:response.status,body:await response.text()};}
    catch(error) {results[name]={readable:false,error:error.name};}
  }
  results.serverEvents=await (await fetch('/events')).json();
  window.labResult=results;
  document.getElementById('result').textContent=JSON.stringify(results,null,2);
}
run().catch(error=>{window.labError=String(error);});
</script>""").encode()
            kind = "text/html; charset=utf-8"
        self.send_response(200)
        self.send_header("Content-Type", kind)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)


api = HTTPServer(("127.0.0.1", 0), API)
app = HTTPServer(("127.0.0.1", 0), App)
api_origin = f"http://127.0.0.1:{api.server_port}"
app_origin = f"http://127.0.0.1:{app.server_port}"
# A process-local client for these two owned loopback requests only.
# It changes no system settings or global urllib opener.
direct_opener = build_opener(ProxyHandler({}), NoRedirect())
threads = [Thread(target=server.serve_forever, daemon=True) for server in (api, app)]
for thread in threads:
    thread.start()
try:
    headers = {"Origin": app_origin, "Access-Control-Request-Method": "PUT",
               "Access-Control-Request-Headers": "x-lab"}
    for path in ("deny", "permit"):
        with open_owned_preflight(Request(api_origin + "/" + path, method="OPTIONS", headers=headers)) as response:
            assert response.status == 204
            assert response.headers.get("Content-Length") is None
            assert response.headers.get("Access-Control-Allow-Methods") == ("PUT" if path == "permit" else None)
            assert response.headers.get("Access-Control-Allow-Headers") == ("X-Lab" if path == "permit" else None)
    events.clear()
    print(json.dumps({"app_url": app_origin, "wire_preflight_checks": "PASS",
                      "real_authentication_or_cookies": False}), flush=True)
    if "--serve" in sys.argv:
        Event().wait()
except KeyboardInterrupt:
    pass
finally:
    for server in (api, app):
        server.shutdown()
        server.server_close()
    for thread in threads:
        thread.join(timeout=5)