让可复用访问令牌远离请求地址

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

parser = argparse.ArgumentParser()
parser.add_argument("--serve", action="store_true")
parser.add_argument("--serve-seconds", type=int, default=120)
args = parser.parse_args()
if not 1 <= args.serve_seconds <= 300:
    parser.error("--serve-seconds must be between 1 and 300")

def require(condition, message):
    if not condition:
        raise AssertionError(message)

# Synthetic demonstration value only; this server is not an authentication service.
demo = "DEMO-NOT-A-SECRET"
events = []


class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        raise RuntimeError("Unexpected redirect in local fixture")


opener = build_opener(ProxyHandler({}), NoRedirect())


class LocalServer(HTTPServer):
    def get_request(self):
        connection, address = super().get_request()
        connection.settimeout(2)
        return connection, address


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

    def do_GET(self):
        path = urlsplit(self.path).path
        if path == "/events":
            body = json.dumps(events).encode()
            kind = "application/json"
        elif path == "/echo":
            entry = {"request_target": self.path,
                     "authorization_present": self.headers.get("Authorization") == "Bearer " + demo,
                     "referer": self.headers.get("Referer")}
            events.append(entry)
            body = json.dumps(entry).encode()
            kind = "application/json"
        elif path == "/":
            events.append({"request_target": self.path, "document": True})
            body = ("""<!doctype html><meta charset="utf-8"><title>Token placement lab</title>
<h1>A clean referrer does not erase the request URL</h1><pre id="result">Running...</pre><script>
const demo=""" + json.dumps(demo) + """;
function require(condition, message) {
  if (!condition) throw new Error(message);
}
function verifyObservation(result, origin) {
  const queryTarget='/echo?access_token='+demo;
  require(new URL(result.page).origin===origin &&
          new URL(result.page).pathname==='/' &&
          new URL(result.page).search==='?access_token='+demo, 'page URL mismatch');
  require(result.query.request_target===queryTarget &&
          result.query.authorization_present===false, 'query placement mismatch');
  require(result.header.request_target==='/echo' &&
          result.header.authorization_present===true, 'header placement mismatch');
  require(result.query.referer===null && result.header.referer===null,
          'Referer was not suppressed');
  require(result.resourceUrls.includes(origin+queryTarget) &&
          result.resourceUrls.includes(origin+'/echo'), 'resource URLs missing');
  require(result.events.some(e=>e.document===true &&
          e.request_target==='/?access_token='+demo), 'document event missing');
  for (const expected of [result.query,result.header]) {
    require(result.events.some(e=>e.request_target===expected.request_target &&
            e.authorization_present===expected.authorization_present &&
            e.referer===null), 'matching server event missing');
  }
}
async function readJson(path, options={}) {
  const response=await fetch(path,{...options,redirect:'error',
                                  signal:AbortSignal.timeout(5000)});
  require(response.ok, 'HTTP status '+response.status);
  return response.json();
}
async function run() {
  require(location.protocol==='http:' && location.hostname==='127.0.0.1',
          'Use the printed loopback URL');
  performance.clearResourceTimings();
  const query=await readJson('/echo?access_token='+demo);
  const header=await readJson('/echo',{headers:{Authorization:'Bearer '+demo}});
  const deadline=performance.now()+5000;
  let resourceUrls=[];
  do {
    resourceUrls=performance.getEntriesByType('resource').map(e=>e.name)
      .filter(name=>new URL(name).pathname==='/echo');
    if (resourceUrls.length>=2) break;
    await new Promise(resolve=>setTimeout(resolve,25));
  } while (performance.now()<deadline);
  const result={page:location.href,query,header,resourceUrls,
                events:await readJson('/events')};
  verifyObservation(result,location.origin);
  window.labResult={status:'PASS',...result};
  document.getElementById('result').textContent=JSON.stringify(window.labResult,null,2);
}
run().catch(error=>{
  window.labError=String(error);
  window.labResult={status:'FAIL',error:window.labError};
  document.getElementById('result').textContent=JSON.stringify(window.labResult,null,2);
});
</script>""").encode()
            kind = "text/html; charset=utf-8"
        else:
            self.send_error(404)
            return
        self.send_response(200)
        self.send_header("Content-Type", kind)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Referrer-Policy", "no-referrer")
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)


def close_server(server, thread):
    stopper = None
    try:
        if thread.is_alive():
            stopper = Thread(target=server.shutdown, daemon=True)
            stopper.start()
            stopper.join(timeout=3)
        if thread.ident is not None:
            thread.join(timeout=3)
        require((stopper is None or not stopper.is_alive()) and not thread.is_alive(),
                "Server shutdown not verified")
    finally:
        server.server_close()
    require(server.socket.fileno() == -1, "Listener close not verified")
    print(json.dumps({"cleanup": "server thread stopped and listener closed"}), flush=True)


server = LocalServer(("127.0.0.1", 0), Handler)
thread = Thread(target=lambda: server.serve_forever(poll_interval=0.1), daemon=True)
origin = f"http://127.0.0.1:{server.server_port}"
try:
    thread.start()
    with opener.open(origin + "/echo?access_token=" + demo, timeout=5) as response:
        query = json.load(response)
    with opener.open(Request(origin + "/echo", headers={"Authorization": "Bearer " + demo}), timeout=5) as response:
        header = json.load(response)
    require(query["request_target"] == "/echo?access_token=" + demo and
            query["authorization_present"] is False, "query placement mismatch")
    require(header["request_target"] == "/echo" and
            header["authorization_present"] is True, "header placement mismatch")
    events.clear()
    print(json.dumps({"app_url": origin + "/?access_token=" + demo,
                      "wire_placement_checks": "PASS", "browser_checks": "NOT RUN BY PYTHON",
                      "serve_seconds": args.serve_seconds if args.serve else 0,
                      "real_credentials_used": False}), flush=True)
    if args.serve:
        Event().wait(args.serve_seconds)
except KeyboardInterrupt:
    pass
finally:
    close_server(server, thread)