From 1f0ce89ac75f07bc5ca32e8c1ad64813d0a68c60 Mon Sep 17 00:00:00 2001 From: Bryce Kwon Date: Sun, 2 Aug 2026 09:11:40 -1000 Subject: Tidy the preview server and pass cgit the request --- tools/serve.py | 309 +++++++++++++++++++++++++++++++++++++-------------------- 1 file changed, 200 insertions(+), 109 deletions(-) diff --git a/tools/serve.py b/tools/serve.py index 62e29d1..99ff02f 100755 --- a/tools/serve.py +++ b/tools/serve.py @@ -1,14 +1,15 @@ #!/usr/bin/env python3 """Local preview server for cgit. -cgit is a CGI program: it reads the request from environment variables and -writes an HTTP response to stdout. This wraps it in a tiny stdlib-only HTTP +cgit is a CGI program. It reads the request from environment variables and +writes an HTTP response to stdout. This wraps it in a small stdlib-only HTTP server so the interface can be previewed in a browser during development, -without configuring Apache/nginx. It is a development aid, not a production +without configuring Apache or nginx. It is a development aid, not a production server. -Static assets (cgit.css, cgit.js, images) are served straight from disk; -every other request is handed to the cgit binary as CGI. +Static assets (cgit.css, cgit.js, images) are served straight from disk. Every +other request is handed to the cgit binary as CGI, with the same environment +the web server configs in custom/servers/ set up. Usage: python3 tools/serve.py --config path/to/cgitrc [--port 8080] @@ -16,6 +17,8 @@ Usage: If --config is omitted, ./cgitrc in the current directory is used. """ +from __future__ import annotations + import argparse import mimetypes import os @@ -23,6 +26,8 @@ import subprocess import sys from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path +from typing import NamedTuple, cast +from urllib.parse import unquote REPO_ROOT = Path(__file__).resolve().parent.parent @@ -31,174 +36,260 @@ REPO_ROOT = Path(__file__).resolve().parent.parent MAX_BODY_BYTES = 8 * 1024 * 1024 CGI_TIMEOUT = 60 +# Suffixes eligible to be served off disk. Everything else is a cgit URL. +STATIC_SUFFIXES = (".css", ".js", ".png", ".ico", ".gif", ".jpg", ".jpeg", + ".svg", ".webp", ".txt", ".woff", ".woff2") + +# Request headers cgit reads, and the CGI variable each arrives as. +PASSED_HEADERS = ( + ("Cookie", "HTTP_COOKIE"), + ("Referer", "HTTP_REFERER"), + ("Content-Type", "CONTENT_TYPE"), +) + + +class CgiResponse(NamedTuple): + status: int + reason: str + headers: list[tuple[str, str]] + body: bytes -def parse_cgi_output(raw): - """Split a CGI response into (status, headers, body).""" - header_blob, _, body = raw.partition(b"\r\n\r\n") - if not body and b"\n\n" in raw: - header_blob, _, body = raw.partition(b"\n\n") - status = 200 - reason = "OK" - headers = [] - for line in header_blob.replace(b"\r\n", b"\n").split(b"\n"): +def split_cgi_output(raw: bytes) -> CgiResponse: + """Split a raw CGI response into its status, headers and body. + + cgit ends its header block with a blank CRLF line, but a lua filter that + writes its own headers may use a bare LF, so both terminators are + accepted. Splitting on the separator rather than on a non-empty body keeps + a legitimately empty body, such as a 304, from being read as headers. + """ + blob, separator, body = raw.partition(b"\r\n\r\n") + if not separator: + blob, separator, body = raw.partition(b"\n\n") + + status, reason = 200, "OK" + headers: list[tuple[str, str]] = [] + for line in blob.replace(b"\r\n", b"\n").split(b"\n"): if not line.strip(): continue - name, _, value = line.partition(b":") - name = name.strip().decode("latin-1") - value = value.strip().decode("latin-1") - if name.lower() == "status": - parts = value.split(None, 1) - try: - status = int(parts[0]) - except (IndexError, ValueError): - status = 200 - reason = parts[1] if len(parts) > 1 else "" - else: + raw_name, _, raw_value = line.partition(b":") + name = raw_name.strip().decode("latin-1") + value = raw_value.strip().decode("latin-1") + if name.lower() != "status": headers.append((name, value)) - return status, reason, headers, body + continue + # "Status: 404 Not Found", where the reason phrase is optional. A + # value that will not parse leaves the 200 OK default whole rather + # than pairing a stale code with a new phrase. + code, _, phrase = value.partition(" ") + try: + status = int(code) + except ValueError: + continue + reason = phrase.strip() + return CgiResponse(status, reason, headers, body) -class Handler(BaseHTTPRequestHandler): - server_version = "cgit-preview" +class CgitHandler(BaseHTTPRequestHandler): + server_version: str = "cgit-preview" - # Only these are served straight from disk; everything else is CGI. - static_suffixes = (".css", ".js", ".png", ".ico", ".gif", ".jpg", - ".jpeg", ".svg", ".webp", ".txt", ".woff", ".woff2") + @property + def preview(self) -> CgitServer: + """The owning server, narrowed from BaseServer for the paths it holds.""" + return cast("CgitServer", self.server) - def do_GET(self): - self.handle_request() + def do_GET(self) -> None: + self.respond() - def do_HEAD(self): - self.handle_request() + def do_HEAD(self) -> None: + self.respond() - def do_POST(self): - self.handle_request() + def do_POST(self) -> None: + self.respond() - def static_path(self, url_path): - name = url_path.lstrip("/") + def respond(self) -> None: + raw_path, _, query = self.path.partition("?") + # A real web server hands the CGI a decoded PATH_INFO, so decode here + # too and keep a repository whose name needs escaping working. + path = unquote(raw_path) + asset = self.locate_asset(path) + if asset is None: + self.run_cgit(path, query) + else: + self.send_asset(asset) + + def locate_asset(self, path: str) -> Path | None: + """Return the file backing a root-level asset request, or None. + + Only bare names at the root qualify, which is what the configs in + custom/servers/ allow as well, so a repository file such as + /myrepo/tree/cgit.css still reaches cgit rather than 404ing on disk. + """ + name = path.lstrip("/") if not name or "/" in name: return None - if not name.lower().endswith(self.static_suffixes): + if not name.lower().endswith(STATIC_SUFFIXES): return None - candidate = (self.server.data_dir / name).resolve() - if candidate.parent != self.server.data_dir or not candidate.is_file(): + candidate = (self.preview.data_dir / name).resolve() + if candidate.parent != self.preview.data_dir or not candidate.is_file(): return None return candidate - def serve_static(self, path): - ctype = mimetypes.guess_type(str(path))[0] or "application/octet-stream" + def send_asset(self, path: Path) -> None: data = path.read_bytes() + content_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream" self.send_response(200) - self.send_header("Content-Type", ctype) + self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(data))) self.end_headers() if self.command != "HEAD": - self.wfile.write(data) + _ = self.wfile.write(data) - def serve_cgi(self, url_path, query): - body = b"" + def read_body(self) -> bytes | None: + """Return the request body, or None if it exceeds the cap.""" try: length = int(self.headers.get("Content-Length", 0)) except ValueError: - length = 0 - if length < 0: - length = 0 + return b"" + if length <= 0: + return b"" if length > MAX_BODY_BYTES: - self.send_error(413, "Request body too large") - return - if length: - body = self.rfile.read(length) + return None + return self.rfile.read(length) + def cgi_environ(self, path: str, query: str, body: bytes) -> dict[str, str]: env = dict(os.environ) env.update( GATEWAY_INTERFACE="CGI/1.1", SERVER_PROTOCOL="HTTP/1.1", SERVER_SOFTWARE=self.server_version, - SERVER_NAME=self.server.server_name, - SERVER_PORT=str(self.server.server_port), + SERVER_NAME=self.preview.server_name, + SERVER_PORT=str(self.preview.server_port), REQUEST_METHOD=self.command, + REQUEST_URI=self.path, + # An empty SCRIPT_NAME puts cgit at the root of the URL space, the + # same as the server configs in custom/servers/ do. SCRIPT_NAME="", - PATH_INFO=url_path, + PATH_INFO=path, QUERY_STRING=query, - REQUEST_URI=self.path, HTTP_HOST=self.headers.get("Host", "localhost"), - CGIT_CONFIG=str(self.server.config), + CGIT_CONFIG=str(self.preview.config), ) - if length: - env["CONTENT_LENGTH"] = str(length) - if self.headers.get("Content-Type"): - env["CONTENT_TYPE"] = self.headers["Content-Type"] + for header, variable in PASSED_HEADERS: + value = self.headers.get(header) + if value: + env[variable] = value + if body: + env["CONTENT_LENGTH"] = str(len(body)) + return env + + def run_cgit(self, path: str, query: str) -> None: + body = self.read_body() + if body is None: + self.send_error(413, "Request body too large") + return try: result = subprocess.run( - [str(self.server.cgit)], input=body, env=env, - stdout=subprocess.PIPE, stderr=subprocess.PIPE, + [str(self.preview.cgit)], + input=body, + env=self.cgi_environ(path, query, body), + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, timeout=CGI_TIMEOUT, ) except subprocess.TimeoutExpired: - self.send_error(504, "cgit timed out") + self.send_error(504, f"cgit did not finish within {CGI_TIMEOUT}s") return + + # cgit reports config problems on stderr while still exiting 0, so + # relay it either way rather than only on failure. + if result.stderr: + _ = sys.stderr.write(result.stderr.decode("latin-1", "replace")) if result.returncode != 0: - self.send_error(500, "cgit failed") - sys.stderr.write(result.stderr.decode("latin-1", "replace")) + self.send_error(500, f"cgit exited with status {result.returncode}") return - status, reason, headers, out = parse_cgi_output(result.stdout) - self.send_response(status, reason) - for name, value in headers: + response = split_cgi_output(result.stdout) + self.send_response(response.status, response.reason) + for name, value in response.headers: self.send_header(name, value) - if not any(n.lower() == "content-length" for n, _ in headers): - self.send_header("Content-Length", str(len(out))) + if not any(name.lower() == "content-length" for name, _ in response.headers): + self.send_header("Content-Length", str(len(response.body))) self.end_headers() if self.command != "HEAD": - self.wfile.write(out) + _ = self.wfile.write(response.body) - def handle_request(self): - url_path, _, query = self.path.partition("?") - static = self.static_path(url_path) - if static: - self.serve_static(static) - else: - self.serve_cgi(url_path, query) + def log_message(self, format: str, *args: object) -> None: + # Named to match BaseHTTPRequestHandler, shadowing the builtin. + _ = sys.stderr.write(f" {format % args}\n") - def log_message(self, fmt, *args): - sys.stderr.write(" %s\n" % (fmt % args)) +class CgitServer(ThreadingHTTPServer): + """Holds the paths the handler needs, so none are attached after the fact.""" -def main(): + def __init__(self, address: tuple[str, int], config: Path, cgit: Path, + data_dir: Path) -> None: + self.config: Path = config + self.cgit: Path = cgit + self.data_dir: Path = data_dir + super().__init__(address, CgitHandler) + + +class Options(argparse.Namespace): + """Typed view of the command line, since Namespace is otherwise untyped. + + The defaults live here and are handed to add_argument below, so the two + cannot drift apart. + """ + + host: str = "127.0.0.1" + port: int = 8080 + config: str = "cgitrc" + cgit: str = str(REPO_ROOT / "build" / "cgit") + data: str = str(REPO_ROOT / "assets") + + +def parse_args(argv: list[str] | None = None) -> Options: parser = argparse.ArgumentParser(description="Preview cgit locally.") - parser.add_argument("--config", default="cgitrc", - help="path to cgitrc (default: ./cgitrc)") - parser.add_argument("--port", type=int, default=8080) - parser.add_argument("--host", default="127.0.0.1") - parser.add_argument("--cgit", default=str(REPO_ROOT / "build" / "cgit"), - help="path to the cgit binary") - parser.add_argument("--data", default=str(REPO_ROOT / "assets"), - help="directory holding cgit.css, cgit.js, images") - args = parser.parse_args() - - config = Path(args.config).resolve() - cgit = Path(args.cgit).resolve() - for label, path in (("config", config), ("cgit binary", cgit)): + _ = parser.add_argument("--config", default=Options.config, + help="path to cgitrc (default: ./cgitrc)") + _ = parser.add_argument("--port", type=int, default=Options.port) + _ = parser.add_argument("--host", default=Options.host) + _ = parser.add_argument("--cgit", default=Options.cgit, + help="path to the cgit binary") + _ = parser.add_argument("--data", default=Options.data, + help="directory holding cgit.css, cgit.js, images") + return parser.parse_args(argv, namespace=Options()) + + +def main() -> None: + opts = parse_args() + + config = Path(opts.config).resolve() + cgit = Path(opts.cgit).resolve() + data_dir = Path(opts.data).resolve() + for label, path in (("config", config), ("cgit binary", cgit), + ("data directory", data_dir)): if not path.exists(): - sys.exit("error: %s not found: %s" % (label, path)) + sys.exit(f"error: {label} not found: {path}") + # Pin the two types the page depends on, so the preview matches what the + # server configs send rather than whatever is in this machine's mime table. mimetypes.add_type("text/css", ".css") - mimetypes.add_type("application/javascript", ".js") - - httpd = ThreadingHTTPServer((args.host, args.port), Handler) - httpd.config = config - httpd.cgit = cgit - httpd.data_dir = Path(args.data).resolve() + mimetypes.add_type("text/javascript", ".js") - url = "http://%s:%d/" % (args.host, args.port) - sys.stderr.write("cgit preview serving %s\n config: %s\n press Ctrl-C to stop\n" - % (url, config)) + httpd = CgitServer((opts.host, opts.port), config, cgit, data_dir) + banner = (f"cgit preview serving http://{opts.host}:{opts.port}/\n" + f" config: {config}\n" + f" press Ctrl-C to stop\n") + _ = sys.stderr.write(banner) try: httpd.serve_forever() except KeyboardInterrupt: - sys.stderr.write("\nstopped\n") + _ = sys.stderr.write("\nstopped\n") + finally: + httpd.server_close() if __name__ == "__main__": -- cgit v2.8.0