mews.page/mews/app.py

"""The one dynamic page on mews.page.

The site is static apart from this: a form posts a page address to /result, the
page is fetched and checked, and the author gets the findings back. Nothing here
runs JavaScript or sets a cookie, because the response is itself a Mews page.
"""

from collections.abc import Iterable
import os
import threading
from urllib.parse import parse_qs

from mews import db, pages
from mews.check import submit
from mews.fetch import Fetcher

# Outbound checks are slow and this runs on a small machine, so only a couple
# happen at once; the rest of the queue is turned away rather than piled up.
_CHECKS = threading.BoundedSemaphore(2)
_REQUESTS = threading.BoundedSemaphore(8)
_SLOT_WAIT = 5.0

MAX_BODY = 4096

BUSY = "The checker is busy right now. Try again in a few minutes."


def directory_path() -> str:
    """Where the directory page is written."""
    return os.environ.get("MEWS_DIRECTORY", "directory.html")


def _start(start_response, status: str, body: bytes) -> list[bytes]:
    start_response(
        status,
        [
            ("Content-Type", "text/html; charset=utf-8"),
            ("Content-Length", str(len(body))),
            ("Cache-Control", "no-store"),
        ],
    )
    return [body]


def _client(environ: dict) -> str:
    """Return the reader's address, as the proxy reports it.

    Caddy sets X-Real-IP from the connection it accepted. The server binds to
    the loopback address only, so nothing else can set that header.
    """
    return environ.get("HTTP_X_REAL_IP") or environ.get("REMOTE_ADDR", "")


def _form(environ: dict) -> dict[str, list[str]]:
    try:
        length = min(int(environ.get("CONTENT_LENGTH") or 0), MAX_BODY)
    except ValueError:
        return {}
    if length <= 0:
        return {}
    raw = environ["wsgi.input"].read(length)
    return parse_qs(raw.decode("utf-8", "replace"), keep_blank_values=True)


def application(environ: dict, start_response) -> Iterable[bytes]:
    """Handle one request."""
    path = environ.get("PATH_INFO", "/")
    method = environ.get("REQUEST_METHOD", "GET")

    if path != "/result":
        return _start(
            start_response,
            "404 Not Found",
            pages.message(
                "Page not found",
                ["That address isn't part of this site."],
                [("/", "Mews"), ("/directory", "The directory")],
            ).encode(),
        )

    if method != "POST":
        return _start(
            start_response,
            "405 Method Not Allowed",
            pages.message(
                "Nothing to show",
                ["Results appear here after you check a page."],
                [("/check", "Check a page")],
            ).encode(),
        )

    form = _form(environ)
    raw_url = (form.get("url") or [""])[0].strip()
    listing = "list" in form

    if not _REQUESTS.acquire(blocking=False):
        return _start(start_response, "503 Service Unavailable", _busy())
    try:
        if not _CHECKS.acquire(timeout=_SLOT_WAIT):
            return _start(start_response, "503 Service Unavailable", _busy())
        try:
            body = _check(raw_url, _client(environ), listing)
        finally:
            _CHECKS.release()
    finally:
        _REQUESTS.release()
    return _start(start_response, "200 OK", body)


def _busy() -> bytes:
    return pages.message("Busy", [BUSY], [("/check", "Try again")]).encode()


def _check(raw_url: str, client: str, listing: bool) -> bytes:
    """Run one check and render the page the author gets back."""
    connection = db.connect()
    try:
        with Fetcher() as fetcher:
            outcome = submit(
                connection,
                raw_url,
                client=db.ip_hash(connection, client),
                listing=listing,
                fetcher=fetcher,
                directory=directory_path() if listing else None,
            )
    except ValueError as error:
        # The directory page refused to publish itself. The site is listed; the
        # page will be rebuilt by the next recheck pass.
        return pages.message(
            "Listed, but the directory didn't rebuild",
            [str(error), "Your site is listed and will appear shortly."],
            [("/directory", "The directory")],
        ).encode()
    finally:
        connection.close()

    if outcome.report is None:
        return pages.message(
            "We couldn't check that page",
            [outcome.message or "Something went wrong. Try again."],
            [("/check", "Try again"), ("/spec/0.1", "The spec")],
        ).encode()

    command = f"uv run mewslint.py {outcome.report.url or raw_url}"
    return pages.report(outcome.report, listed=outcome.listed, command=command).encode()


def serve() -> None:
    """Run the app for real, on the loopback address only."""
    # Imported here so the validator and the command line tools do not pull in
    # a web server.
    from waitress import serve as waitress_serve  # noqa: PLC0415

    listen = os.environ.get("MEWS_LISTEN", "127.0.0.1:8394")
    host = listen.rsplit(":", 1)[0]
    if host not in ("127.0.0.1", "::1", "localhost"):
        raise SystemExit(
            "mewsd serves on the loopback address only, behind a reverse proxy. "
            f"Set MEWS_LISTEN to 127.0.0.1 with a port, not {listen}."
        )
    waitress_serve(application, listen=listen, threads=4, ident="mews.page")


app = application