index : cgit

#!/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
server so the interface can be previewed in a browser during development,
without configuring Apache/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.

Usage:
    python3 tools/serve.py --config path/to/cgitrc [--port 8080]

If --config is omitted, ./cgitrc in the current directory is used.
"""

import argparse
import mimetypes
import os
import subprocess
import sys
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parent.parent

# Bounds so a single request cannot exhaust the dev server. It only ever
# serves a browser on loopback, so these are generous.
MAX_BODY_BYTES = 8 * 1024 * 1024
CGI_TIMEOUT = 60


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"):
        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:
            headers.append((name, value))
    return status, reason, headers, body


class Handler(BaseHTTPRequestHandler):
    server_version = "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")

    def do_GET(self):
        self.handle_request()

    def do_HEAD(self):
        self.handle_request()

    def do_POST(self):
        self.handle_request()

    def static_path(self, url_path):
        name = url_path.lstrip("/")
        if not name or "/" in name:
            return None
        if not name.lower().endswith(self.static_suffixes):
            return None
        candidate = (self.server.data_dir / name).resolve()
        if candidate.parent != self.server.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"
        data = path.read_bytes()
        self.send_response(200)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        if self.command != "HEAD":
            self.wfile.write(data)

    def serve_cgi(self, url_path, query):
        body = b""
        try:
            length = int(self.headers.get("Content-Length", 0))
        except ValueError:
            length = 0
        if length < 0:
            length = 0
        if length > MAX_BODY_BYTES:
            self.send_error(413, "Request body too large")
            return
        if length:
            body = self.rfile.read(length)

        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),
            REQUEST_METHOD=self.command,
            SCRIPT_NAME="",
            PATH_INFO=url_path,
            QUERY_STRING=query,
            REQUEST_URI=self.path,
            HTTP_HOST=self.headers.get("Host", "localhost"),
            CGIT_CONFIG=str(self.server.config),
        )
        if length:
            env["CONTENT_LENGTH"] = str(length)
        if self.headers.get("Content-Type"):
            env["CONTENT_TYPE"] = self.headers["Content-Type"]

        try:
            result = subprocess.run(
                [str(self.server.cgit)], input=body, env=env,
                stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                timeout=CGI_TIMEOUT,
            )
        except subprocess.TimeoutExpired:
            self.send_error(504, "cgit timed out")
            return
        if result.returncode != 0:
            self.send_error(500, "cgit failed")
            sys.stderr.write(result.stderr.decode("latin-1", "replace"))
            return

        status, reason, headers, out = parse_cgi_output(result.stdout)
        self.send_response(status, reason)
        for name, value in 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)))
        self.end_headers()
        if self.command != "HEAD":
            self.wfile.write(out)

    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, fmt, *args):
        sys.stderr.write("  %s\n" % (fmt % args))


def main():
    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)):
        if not path.exists():
            sys.exit("error: %s not found: %s" % (label, path))

    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()

    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))
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        sys.stderr.write("\nstopped\n")


if __name__ == "__main__":
    main()