"""The local server's front door: who may talk to it and how much they may send (#69).
A server bound to the loopback interface is reachable only from this
machine, but a web page in any browser on it can still send requests. A
page from another site is told apart by its ``Origin``. A site that points
its own name at 127.0.0.1 (DNS rebinding) shares the server's origin,
but cannot change the ``Host`` it sends. So a loopback server answers only
requests whose ``Host`` names this machine, and refuses state from pages
whose ``Origin`` is anything else. A server bound to another interface was
asked for by name; it takes any ``Host`` and leaves access to its network.
Every request body is also bounded (:data:`vibesketch.limits.MAX_REQUEST_BYTES`):
it is read up to the bound and refused with 413 past it, before the
application parses anything.
"""
from __future__ import annotations
import json
from collections.abc import Iterable
from urllib.parse import urlsplit
from starlette.types import ASGIApp, Message, Receive, Scope, Send
__all__ = ["LOOPBACK", "Guard", "is_loopback"]
#: The names this machine answers to on its loopback interface.
LOOPBACK = frozenset({"127.0.0.1", "localhost", "::1"})
[docs]
def is_loopback(host: str) -> bool:
return host.strip("[]").lower() in LOOPBACK or host.startswith("127.")
def _hostname(value: str) -> str:
"""The name in a ``Host`` header or an origin, without its port or brackets."""
netloc = urlsplit(value if "//" in value else f"//{value}").hostname
return (netloc or "").lower()
[docs]
class Guard:
"""ASGI middleware: the ``Host`` and ``Origin`` checks and the request body bound."""
def __init__(self, app: ASGIApp, *, hosts: Iterable[str] | None, max_body: int) -> None:
self.app = app
self.hosts = None if hosts is None else frozenset(h.strip("[]").lower() for h in hosts)
self.max_body = max_body
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
if scope["type"] != "http":
await self.app(scope, receive, send)
return
headers = {k.decode("latin-1").lower(): v.decode("latin-1") for k, v in scope["headers"]}
if self.hosts is not None:
host = _hostname(headers.get("host", ""))
if host not in self.hosts:
await _refuse(send, 421, f"this server does not answer for the host {host!r}")
return
origin = headers.get("origin")
if origin is not None and _hostname(origin) not in self.hosts:
await _refuse(send, 403, f"requests from {origin} are not accepted")
return
stated = headers.get("content-length")
if stated is not None and stated.isdigit() and int(stated) > self.max_body:
await _refuse(send, 413, self._too_large(int(stated)))
return
# Read the body up to the bound, then hand it to the application whole.
chunks: list[bytes] = []
size = 0
while True:
message = await receive()
if message["type"] != "http.request":
break # the client left; the application sees that too
chunk = message.get("body", b"")
size += len(chunk)
if size > self.max_body:
await _refuse(send, 413, self._too_large(size))
return
chunks.append(chunk)
if not message.get("more_body", False):
break
body = b"".join(chunks)
delivered = False
async def replay() -> Message:
nonlocal delivered
if delivered:
return await receive()
delivered = True
return {"type": "http.request", "body": body, "more_body": False}
await self.app(scope, replay, send)
def _too_large(self, size: int) -> str:
return (
f"the request is {size / 2**20:.0f} MiB, more than the "
f"{self.max_body / 2**20:.0f} MiB this server accepts"
)
async def _refuse(send: Send, status: int, reason: str) -> None:
body = json.dumps({"detail": reason}).encode("utf-8")
await send(
{
"type": "http.response.start",
"status": status,
"headers": [
(b"content-type", b"application/json"),
(b"content-length", str(len(body)).encode("ascii")),
],
}
)
await send({"type": "http.response.body", "body": body})