Something went wrong. Try again.
Backend environment for match hosting for lance.blue
Something went wrong. Try again.
Python
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778#!/usr/bin/env python3"""Both ends of an upload, for tests/shell/test-upload.sh.
POST /sign/<name> mints a URL pointing at this server's /s3/<name>, or fails, per the first line of <dir>/mint-mode PUT /s3/<name> the signed destination PUT /api/<name> headquarters' own route, which arena no longer uses
Every request is appended to <dir>/requests as "METHOD path", which is what theshell test asserts on. /api/ is still served so the test can show that nothingarrives there, even from a manifest that offers it. The port is written to<dir>/port once bound, because the test asks for port 0 rather than guessing ata free one."""
import http.serverimport jsonimport pathlibimport sys
SIGN_PREFIX = "/sign/"
class Handler(http.server.BaseHTTPRequestHandler): directory = pathlib.Path()
def log_message(self, *args): """Quiet. The test reads the request log, not stderr."""
def record(self): with (self.directory / "requests").open("a") as log: log.write(f"{self.command} {self.path}\n")
def discard_body(self): length = int(self.headers.get("Content-Length") or 0) while length > 0: length -= len(self.rfile.read(min(length, 65536)))
def do_POST(self): self.discard_body() self.record() mode = (self.directory / "mint-mode").read_text().strip() if mode != "ok" or not self.path.startswith(SIGN_PREFIX): self.send_response(500) self.end_headers() return name = self.path[len(SIGN_PREFIX) :] body = json.dumps( { "url": f"http://{self.headers['Host']}/s3/{name}", "method": "PUT", "expiresIn": 900, } ).encode() self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body)
def do_PUT(self): self.discard_body() self.record() self.send_response(200) self.end_headers()
def main(): directory = pathlib.Path(sys.argv[1]) Handler.directory = directory server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) (directory / "port").write_text(str(server.server_address[1])) server.serve_forever()
if __name__ == "__main__": main()