bible/server.py

"""Static file server for the Bible site."""

from __future__ import annotations

import argparse
import hashlib
import io
import json
import os
import posixpath
import signal
import sys
from functools import partial
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer

NO_CACHE_NAMES = {"sw.js"}
NO_CACHE_EXTS = {".html", ".njk", ".json", ".webmanifest"}

BOOT_PAGE = "/boot.html"

WORKER = "sw.js"

VERSION_LINE = "const VERSION = 'dev';"
LIBRARY_VERSION_LINE = "const LIBRARY_VERSION = 'dev';"
LIBRARY_LINE = "const LIBRARY = [];"

def walk(roots):
    found = {}
    for root in roots:
        for dirpath, _dirs, names in os.walk(root):
            for name in sorted(names):
                path = os.path.join(dirpath, name)
                rel = os.path.relpath(path, root).replace(os.sep, "/")
                found.setdefault(rel, path)
    return found

def digest(files):
    out = hashlib.sha256()
    for rel in sorted(files):
        st = os.stat(files[rel])
        out.update(("%s\0%d\0%d\0" % (rel, st.st_size, st.st_mtime_ns)).encode())
    return out.hexdigest()[:12]

def page_url(rel):
    if rel == "index.html":
        return "./"
    return "./" + (rel[:-len(".html")] if rel.endswith(".html") else rel)

def stamp_worker(roots):
    chrome = walk(roots[:1])
    library = {rel: path for rel, path in walk(roots).items() if rel not in chrome}

    src = chrome.pop(WORKER, None)
    if src is None:
        return None
    with open(src, encoding="utf-8") as fh:
        body = fh.read()

    for line, value in (
        (VERSION_LINE, "const VERSION = '%s';" % digest(chrome)),
        (LIBRARY_VERSION_LINE, "const LIBRARY_VERSION = '%s';" % digest(library)),
        (LIBRARY_LINE, "const LIBRARY = %s;"
                       % json.dumps(sorted(page_url(rel) for rel in library), indent=2)),
    ):
        if line not in body:
            raise RuntimeError("%s: expected to find %r to stamp" % (WORKER, line))
        body = body.replace(line, value, 1)
    return body.encode("utf-8")

class Handler(SimpleHTTPRequestHandler):
    extensions_map = {
        **SimpleHTTPRequestHandler.extensions_map,
        ".html": "text/html; charset=utf-8",
        ".css": "text/css; charset=utf-8",
        ".js": "text/javascript; charset=utf-8",
        ".json": "application/json; charset=utf-8",
        ".njk": "text/plain; charset=utf-8",
        ".webmanifest": "application/manifest+json; charset=utf-8",
        ".png": "image/png",
        ".svg": "image/svg+xml",
    }
    server_version = "bible-static"
    sys_version = ""
    roots = ()

    def translate_path(self, path):
        for root in self.roots:
            self.directory = root
            candidate = super().translate_path(path)
            if os.path.isfile(candidate):
                return candidate
            if os.path.isfile(candidate + ".html"):
                return candidate + ".html"
            if os.path.isfile(os.path.join(candidate, "index.html")):
                return candidate
        self.directory = self.roots[0]
        return super().translate_path(path)

    def send_head(self):
        if self.headers.get("Sec-Fetch-Mode") == "navigate":
            self.path = BOOT_PAGE

        if posixpath.basename(self.path.split("?", 1)[0]) == WORKER:
            return self.send_worker()
        return super().send_head()

    def send_worker(self):
        try:
            body = stamp_worker(self.roots)
        except Exception as err:
            self.log_message("stamp failed: %s", err)
            self.send_error(500, "service worker unavailable")
            return None
        if body is None:
            self.send_error(404, "no service worker")
            return None
        self.send_response(200)
        self.send_header("Content-Type", "text/javascript; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        return io.BytesIO(body)

    def end_headers(self):
        path = self.path.split("?", 1)[0]
        name = posixpath.basename(path)
        ext = posixpath.splitext(name)[1].lower()
        # Pages have no extension, so testing the extension alone treats them
        # as files and caches them for a week.
        if name in NO_CACHE_NAMES or ext in NO_CACHE_EXTS or "." not in name:
            self.send_header("Cache-Control", "no-cache")
        else:
            self.send_header("Cache-Control", "public, max-age=604800")
        self.send_header("Vary", "Sec-Fetch-Mode")
        self.send_header("Service-Worker-Allowed", "/")
        self.send_header("X-Content-Type-Options", "nosniff")
        super().end_headers()

    def log_message(self, fmt, *args):
        sys.stdout.write("%s %s\n" % (self.address_string(), fmt % args))
        sys.stdout.flush()

def main(argv=None):
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--root", action="append", dest="roots", metavar="DIR",
                    help="directory to serve; repeatable, earlier wins "
                         "(default: assets then bible, or $SITE_ROOTS)")
    ap.add_argument("--host", default=os.environ.get("HOST", "0.0.0.0"))
    ap.add_argument("--port", type=int, default=int(os.environ.get("PORT", "8080")))
    args = ap.parse_args(argv)

    names = args.roots or os.environ.get("SITE_ROOTS", "assets:bible").split(":")
    roots = tuple(os.path.abspath(name) for name in names if name)
    for root in roots:
        if not os.path.isdir(root):
            sys.exit("site root not found: %s" % root)

    httpd = ThreadingHTTPServer((args.host, args.port),
                                partial(type("Rooted", (Handler,), {"roots": roots}),
                                        directory=roots[0]))
    httpd.daemon_threads = True

    def stop(_signum, _frame):
        httpd.shutdown()

    signal.signal(signal.SIGTERM, stop)
    signal.signal(signal.SIGINT, stop)

    print("serving %s at http://%s:%d" % (" + ".join(roots), args.host, args.port))
    sys.stdout.flush()
    try:
        httpd.serve_forever()
    finally:
        httpd.server_close()
    return 0

if __name__ == "__main__":
    sys.exit(main())