cista-storage/cista/serve.py

import os
import re
from pathlib import Path, PurePath

from sanic import Sanic

from cista import config, server80


def run(*, dev=False):
    """Run Sanic main process that spawns worker processes to serve HTTP requests."""
    from .app import app

    url, opts = parse_listen(config.config.listen)
    # Silence Sanic's warning about running in production rather than debug
    os.environ["SANIC_IGNORE_PRODUCTION_WARNING"] = "1"
    confdir = config.conffile.parent
    wwwroot = PurePath(__file__).parent / "wwwroot"
    if opts.get("ssl"):
        # Run plain HTTP redirect/acme server on port 80
        server80.app.prepare(port=80, motd=False)
        domain = opts["host"]
        check_cert(confdir / domain, domain)
        opts["ssl"] = str(confdir / domain)  # type: ignore
    app.prepare(
        **opts,
        motd=False,
        dev=dev,
        auto_reload=dev,
        reload_dir={confdir, wwwroot},
        access_log=True,
    )  # type: ignore
    Sanic.serve()


def check_cert(certdir, domain):
    if (certdir / "privkey.pem").exist() and (certdir / "fullchain.pem").exists():
        return
    # TODO: Use certbot to fetch a cert
    raise ValueError(
        f"TLS certificate files privkey.pem and fullchain.pem needed in {certdir}",
    )


def parse_listen(listen):
    if listen.startswith("/"):
        unix = Path(listen).resolve()
        if not unix.parent.exists():
            raise ValueError(
                f"Directory for unix socket does not exist: {unix.parent}/",
            )
        return "http://localhost", {"unix": unix}
    if re.fullmatch(r"(\w+(-\w+)*\.)+\w{2,}", listen, re.UNICODE):
        return f"https://{listen}", {"host": listen, "port": 443, "ssl": True}
    try:
        addr, _port = listen.split(":", 1)
        port = int(_port)
    except Exception:
        raise ValueError(f"Invalid listen address: {listen}") from None
    return f"http://localhost:{port}", {"host": addr, "port": port}