quill/server.py

#!/usr/bin/env python3
"""Quill server: a read-only wiki over Git repositories on a Gitea server.

    python3 server.py [--config config.json]

Standard library only (Python 3.10+). People read; agents and people write by pushing to the
space's repository (branch -> pull request -> merge). Quill never writes anywhere.
"""
from __future__ import annotations

import argparse
import hashlib
import hmac
import html
import json
import os
import secrets
import sys
import urllib.error
import urllib.parse
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

import quill_core

HERE = Path(__file__).resolve().parent
CONFIG: dict = {}
COOKIE = "quill_view"
_SECRET = b""


def load_config(path: str | os.PathLike) -> dict:
    cfg = json.loads(Path(path).read_text(encoding="utf-8"))
    cfg.setdefault("name", "Quill")
    cfg.setdefault("host", "127.0.0.1")
    cfg.setdefault("port", 8782)
    cfg.setdefault("access", "public")
    cfg.setdefault("nav", [])
    if cfg["access"] not in ("public", "code"):
        raise ValueError('access must be "public" or "code"')
    if cfg["access"] == "code" and not cfg.get("view_code_sha256"):
        raise ValueError('access "code" needs view_code_sha256 (python3 server.py --hash-code CODE)')
    gitea = str(cfg.get("gitea_url") or "").rstrip("/")
    if not gitea.startswith(("http://", "https://")):
        raise ValueError("gitea_url must be an http(s) URL")
    cfg["gitea_url"] = gitea
    quill_core.WIKI_NAME = str(cfg["name"])
    quill_core.configure(cfg.get("spaces") or [], cfg.get("git_web_base") or gitea)
    return cfg


def _token() -> str:
    env = os.environ.get("QUILL_GITEA_TOKEN", "").strip()
    if env:
        return env
    f = CONFIG.get("gitea_token_file") or ""
    if f and Path(f).expanduser().is_file():
        return Path(f).expanduser().read_text(encoding="utf-8").strip()
    return ""


def gitea_get(path: str, raw: bool = False):
    """Read-only GET against the Gitea API; a token is optional (public repos need none)."""
    req = urllib.request.Request(CONFIG["gitea_url"] + "/api/v1" + path, headers={"User-Agent": "quill/1"})
    tok = _token()
    if tok:
        req.add_header("Authorization", "token " + tok)
    with urllib.request.urlopen(req, timeout=15) as res:
        body = res.read().decode("utf-8", "replace")
    return body if raw else json.loads(body)


def hash_code(code: str) -> str:
    return hashlib.sha256(code.encode("utf-8")).hexdigest()


def _view_cookie_value() -> str:
    return hmac.new(_SECRET, b"quill-view", hashlib.sha256).hexdigest()


def _page() -> bytes:
    nav = "".join(
        '    <a class="nav" href="' + html.escape(str(item.get("href") or "#"), quote=True) + '">'
        + html.escape(str(item.get("label") or "")) + "</a>\n"
        for item in CONFIG.get("nav") or [] if isinstance(item, dict)
    )
    page = (HERE / "quill.html").read_text(encoding="utf-8")
    return page.replace("__WIKI_NAME__", html.escape(CONFIG["name"])).replace("__NAV_LINKS__", nav).encode("utf-8")


VIEW_FORM = """<!doctype html><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
<title>%s</title><style>body{font:16px/1.5 system-ui,sans-serif;max-width:28rem;margin:12vh auto;padding:0 1rem}
input,button{font:inherit;padding:.6rem .8rem;border-radius:.5rem;border:1px solid #bbb}button{cursor:pointer}</style>
<h1>%s</h1><p>Enter the view code to read this wiki.</p>
<form method="post" action="/view"><input type="hidden" name="next" value="%s">
<input name="code" type="password" autocomplete="off" required autofocus> <button>Open</button></form>%s"""


class Handler(BaseHTTPRequestHandler):
    server_version = "Quill/1.0"

    def log_message(self, fmt, *args):
        sys.stderr.write("%s - %s %s\n" % (self.address_string(), self.command, urllib.parse.urlsplit(self.path).path))

    def _send(self, status: int, body: bytes, ctype: str, extra: dict | None = None) -> None:
        self.send_response(status)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("X-Content-Type-Options", "nosniff")
        self.send_header("Referrer-Policy", "same-origin")
        for k, v in (extra or {}).items():
            self.send_header(k, v)
        self.end_headers()
        if self.command != "HEAD":
            self.wfile.write(body)

    def _json(self, status: int, data: dict) -> None:
        self._send(status, json.dumps(data).encode("utf-8"), "application/json")

    def _can_read(self) -> bool:
        if CONFIG["access"] == "public":
            return True
        for part in (self.headers.get("Cookie") or "").split(";"):
            k, _, v = part.strip().partition("=")
            if k == COOKIE and hmac.compare_digest(v, _view_cookie_value()):
                return True
        code = self.headers.get("X-View-Code") or ""
        return bool(code) and hmac.compare_digest(hash_code(code), CONFIG["view_code_sha256"])

    def do_HEAD(self):  # noqa: N802
        self.do_GET()

    def do_GET(self):  # noqa: N802
        parts = urllib.parse.urlsplit(self.path)
        path = parts.path.rstrip("/") or "/"
        if path == "/health":
            return self._json(200, {"ok": True})
        if path == "/":
            return self._send(302, b"", "text/plain", {"Location": "/quill"})
        if path == "/view":
            nxt = urllib.parse.parse_qs(parts.query).get("next", ["/quill"])[0]
            if not nxt.startswith("/") or nxt.startswith("//"):
                nxt = "/quill"
            name = html.escape(CONFIG["name"])
            return self._send(200, (VIEW_FORM % (name, name, html.escape(nxt, quote=True), "")).encode(), "text/html; charset=utf-8")
        if quill_core.is_wiki_path(path):
            if not self._can_read():
                return self._send(302, b"", "text/plain", {"Location": "/view?next=" + urllib.parse.quote(path, safe="")})
            return self._send(200, _page(), "text/html; charset=utf-8")
        if path == "/v1/quill" or path.startswith("/v1/quill/"):
            if not self._can_read():
                return self._json(401, {"error": "view_code_required"})
            status, body = quill_core.api(path, gitea_get)
            return self._json(status, body)
        return self._json(404, {"error": "not_found"})

    def do_POST(self):  # noqa: N802
        path = urllib.parse.urlsplit(self.path).path
        if path != "/view" or CONFIG["access"] != "code":
            return self._json(405, {"error": "read_only", "message": "Quill is read-only. Write by pushing to the space's Git repo."})
        try:
            n = min(int(self.headers.get("Content-Length") or 0), 4096)
        except ValueError:
            n = 0
        form = urllib.parse.parse_qs(self.rfile.read(n).decode("utf-8", "replace"))
        code = (form.get("code") or [""])[0]
        nxt = (form.get("next") or ["/quill"])[0]
        if not nxt.startswith("/") or nxt.startswith("//"):
            nxt = "/quill"
        if code and hmac.compare_digest(hash_code(code), CONFIG["view_code_sha256"]):
            cookie = f"{COOKIE}={_view_cookie_value()}; HttpOnly; Path=/; SameSite=Lax; Max-Age=2592000"
            if CONFIG.get("secure_cookie", True) and not self.headers.get("Host", "").startswith(("127.0.0.1", "localhost")):
                cookie += "; Secure"
            return self._send(303, b"", "text/plain", {"Location": nxt, "Set-Cookie": cookie})
        name = html.escape(CONFIG["name"])
        return self._send(401, (VIEW_FORM % (name, name, html.escape(nxt, quote=True), "<p>Wrong code.</p>")).encode(), "text/html; charset=utf-8")

    def do_PUT(self):  # noqa: N802
        self._json(405, {"error": "read_only"})

    do_PATCH = do_DELETE = do_PUT


def _secret(data_dir: Path) -> bytes:
    data_dir.mkdir(parents=True, exist_ok=True)
    f = data_dir / "cookie-secret"
    if not f.is_file():
        f.write_text(secrets.token_hex(32), encoding="utf-8")
        f.chmod(0o600)
    return f.read_text(encoding="utf-8").strip().encode()


def main(argv=None) -> None:
    global CONFIG, _SECRET
    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    ap.add_argument("--config", default=os.environ.get("QUILL_CONFIG", str(HERE / "config.json")))
    ap.add_argument("--hash-code", metavar="CODE", help="print the sha256 for view_code_sha256 and exit")
    a = ap.parse_args(argv)
    if a.hash_code:
        print(hash_code(a.hash_code))
        return
    CONFIG = load_config(a.config)
    _SECRET = _secret(Path(CONFIG.get("data_dir") or HERE / "data").expanduser())
    srv = ThreadingHTTPServer((CONFIG["host"], int(CONFIG["port"])), Handler)
    print(f"{CONFIG['name']} on http://{CONFIG['host']}:{CONFIG['port']}/quill ({len(quill_core.SPACES)} spaces, access={CONFIG['access']})", flush=True)
    srv.serve_forever()


if __name__ == "__main__":
    main()