dashboard/auth_server.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import secrets
 import time
 from urllib.parse import parse_qs, urlsplit
@@ -91,13 +92,15 @@ async def login(request: Request) -> Response:
     if content_type != "application/x-www-form-urlencoded":
         authentication.log_auth_event("login_request_rejected", correlation_id)
         return PlainTextResponse("Bad Request", status_code=400)
-    body = await request.body()
+    body = bytearray()
-    if len(body) > _MAX_LOGIN_BODY_BYTES:
+    async for chunk in request.stream():
-        authentication.log_auth_event("login_request_rejected", correlation_id)
+        if len(body) + len(chunk) > _MAX_LOGIN_BODY_BYTES:
-        return PlainTextResponse("Bad Request", status_code=400)
+            authentication.log_auth_event("login_request_rejected", correlation_id)
+            return PlainTextResponse("Bad Request", status_code=400)
+        body.extend(chunk)
     try:
         values = parse_qs(
-            body.decode("utf-8"),
+            bytes(body).decode("utf-8"),
             keep_blank_values=True,
             strict_parsing=True,
             max_num_fields=3,
@@ -121,7 +124,9 @@ async def login(request: Request) -> Response:
     if not authentication.verify_credentials(
         settings.credentials, submitted_id, submitted_password
     ):
-        authentication.enforce_failed_login_delay(started_at=started_at)
+        await asyncio.to_thread(
+            authentication.enforce_failed_login_delay, started_at=started_at
+        )
         authentication.log_auth_event("login_failed", correlation_id)
         return _redirect("/?auth_error=invalid")
 

tests/test_dashboard_auth_server.py

@@ -1,18 +1,22 @@
 from __future__ import annotations
 
+import asyncio
 import sys
+import threading
+from collections.abc import Callable
 from http.cookies import SimpleCookie
 from pathlib import Path
 
 import pytest
 from starlette.testclient import TestClient
+from starlette.requests import Request
 
 ROOT_DIR = Path(__file__).resolve().parents[1]
 if str(ROOT_DIR) not in sys.path:
     sys.path.insert(0, str(ROOT_DIR))
 
 from dashboard import authentication
-from dashboard.auth_server import create_auth_routes
+from dashboard.auth_server import create_auth_routes, login
 
 
 ADMIN_ID = "admin"
@@ -218,3 +222,104 @@ def test_cookie_token_round_trip(client: TestClient) -> None:
     assert authentication.verify_persistent_token(
         authentication.load_auth_settings(), token
     ) is not None
+
+
+def _streaming_login_request(
+    chunks: list[bytes],
+) -> tuple[Request, Callable[[], int]]:
+    messages = [
+        {
+            "type": "http.request",
+            "body": chunk,
+            "more_body": index < len(chunks) - 1,
+        }
+        for index, chunk in enumerate(chunks)
+    ]
+    calls = 0
+
+    async def receive():
+        nonlocal calls
+        calls += 1
+        return messages.pop(0)
+
+    scope = {
+        "type": "http",
+        "method": "POST",
+        "scheme": "https",
+        "path": "/api/auth/login",
+        "raw_path": b"/api/auth/login",
+        "query_string": b"",
+        "headers": [
+            (b"host", b"testserver"),
+            (b"origin", b"https://testserver"),
+            (b"content-type", b"application/x-www-form-urlencoded"),
+        ],
+        "client": ("testclient", 50000),
+        "server": ("testserver", 443),
+    }
+    return Request(scope, receive), lambda: calls
+
+
+def test_login_offloads_failed_login_delay_from_event_loop(monkeypatch) -> None:
+    entered = threading.Event()
+    release = threading.Event()
+    probe_ran = threading.Event()
+    probe_before_release = threading.Event()
+
+    def blocking_delay(*, started_at):
+        entered.set()
+        if probe_ran.wait(timeout=1):
+            probe_before_release.set()
+        release.set()
+
+    monkeypatch.setattr(authentication, "enforce_failed_login_delay", blocking_delay)
+
+    async def exercise() -> None:
+        request, _ = _streaming_login_request([b"admin_id=admin&admin_password=wrong"])
+        task = asyncio.create_task(login(request))
+
+        async def probe() -> None:
+            probe_ran.set()
+
+        asyncio.create_task(probe())
+        assert await asyncio.to_thread(entered.wait, 1)
+        release.set()
+        response = await task
+        assert response.status_code == 303
+        assert response.headers["location"] == "/?auth_error=invalid"
+        assert "set-cookie" not in response.headers
+
+    asyncio.run(exercise())
+    assert probe_before_release.is_set()
+
+
+def test_login_rejects_exactly_one_byte_over_streaming_limit(monkeypatch) -> None:
+    def unexpected_verification(*_args, **_kwargs):
+        raise AssertionError("oversized body reached credential verification")
+
+    monkeypatch.setattr(authentication, "verify_credentials", unexpected_verification)
+    request, calls = _streaming_login_request([b"x" * 4097])
+    response = asyncio.run(login(request))
+
+    assert response.status_code == 400
+    assert "set-cookie" not in response.headers
+    assert calls() == 1
+
+
+def test_login_accepts_body_at_streaming_limit() -> None:
+    prefix = b"admin_id=admin&admin_password="
+    body = prefix + (b"x" * (4096 - len(prefix)))
+    assert len(body) == 4096
+    request, calls = _streaming_login_request([body])
+    response = asyncio.run(login(request))
+
+    assert response.status_code == 303
+    assert calls() == 1
+
+
+def test_login_stops_reading_stream_after_limit_is_exceeded() -> None:
+    request, calls = _streaming_login_request([b"x" * 4096, b"y", b"z"])
+    response = asyncio.run(login(request))
+
+    assert response.status_code == 400
+    assert calls() == 2