security: bound and redact inbound requests (#327)

tests/test_request_boundary.py

@@ -0,0 +1,147 @@
+import httpx
+import pytest
+
+from src import main
+from src.request_boundary import RequestBodyLimitMiddleware
+
+
+@pytest.fixture
+def access_control(monkeypatch, tmp_path):
+    monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "operator")
+    monkeypatch.setenv(
+        "STACKCHAIN_DASHBOARD_ACCESS_TOKEN", "correct horse battery staple"
+    )
+    monkeypatch.setenv(
+        "STACKCHAIN_DASHBOARD_SESSION_SECRET",
+        "a-separate-session-signing-secret-with-enough-entropy",
+    )
+    monkeypatch.setenv("STACKCHAIN_SESSION_DB", str(tmp_path / "sessions.sqlite3"))
+    monkeypatch.setenv(
+        "STACKCHAIN_LOGIN_ATTEMPT_DB", str(tmp_path / "login-attempts.sqlite3")
+    )
+
+
+@pytest.mark.anyio
+async def test_declared_oversized_sign_in_is_rejected_before_throttle(
+    access_control, monkeypatch
+):
+    throttle_touched = False
+
+    def attempt_store():
+        nonlocal throttle_touched
+        throttle_touched = True
+        raise AssertionError("oversized request reached sign-in throttling")
+
+    monkeypatch.setattr(main, "_login_attempt_store", attempt_store)
+    body = b'{"access_token":"' + (b"x" * 20_000) + b'"}'
+    transport = httpx.ASGITransport(app=main.app)
+
+    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
+        response = await client.post(
+            "/api/v1/session",
+            content=body,
+            headers={"Content-Type": "application/json"},
+        )
+
+    assert response.status_code == 413
+    assert response.json() == {"detail": "Request body too large"}
+    assert response.headers["cache-control"] == "no-store"
+    assert response.headers["x-content-type-options"] == "nosniff"
+    assert len(response.content) < 128
+    assert throttle_touched is False
+
+
+@pytest.mark.anyio
+async def test_streamed_oversized_sign_in_without_content_length_is_rejected(
+    access_control, monkeypatch
+):
+    throttle_touched = False
+
+    def attempt_store():
+        nonlocal throttle_touched
+        throttle_touched = True
+        raise AssertionError("oversized stream reached sign-in throttling")
+
+    async def chunks():
+        yield b'{"access_token":"'
+        yield b"x" * 10_000
+        yield b"y" * 10_000
+        yield b'"}'
+
+    monkeypatch.setattr(main, "_login_attempt_store", attempt_store)
+    request = httpx.Request(
+        "POST",
+        "https://test/api/v1/session",
+        content=chunks(),
+        headers={"Content-Type": "application/json"},
+    )
+    request.headers.pop("transfer-encoding", None)
+    transport = httpx.ASGITransport(app=main.app)
+
+    response = await transport.handle_async_request(request)
+    await response.aread()
+
+    assert response.status_code == 413
+    assert response.json() == {"detail": "Request body too large"}
+    assert response.headers["cache-control"] == "no-store"
+    assert throttle_touched is False
+
+
+@pytest.mark.anyio
+async def test_validation_error_never_reflects_submitted_values(access_control):
+    canary = "DO-NOT-REFLECT-ACCESS-TOKEN-CANARY"
+    transport = httpx.ASGITransport(app=main.app)
+
+    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
+        response = await client.post(
+            "/api/v1/session",
+            json={"access_token": canary, "device_label": canary * 3},
+        )
+
+    assert response.status_code == 422
+    assert canary not in response.text
+    details = response.json()["detail"]
+    assert details
+    assert all(set(item) <= {"type", "loc", "msg"} for item in details)
+
+
+def test_request_limits_are_route_specific_and_cover_api_mutations():
+    assert main.request_body_limit("POST", "/api/v1/session") == 16 * 1024
+    assert (
+        main.request_body_limit("POST", "/api/v1/repos/stackchain/project/issues")
+        == 64 * 1024
+    )
+    assert main.request_body_limit("GET", "/api/v1/context") is None
+    assert main.request_body_limit("POST", "/unrelated") is None
+
+
+@pytest.mark.anyio
+async def test_request_limit_uses_application_path_under_domain_subpath():
+    downstream_called = False
+    sent = []
+
+    async def downstream(_scope, _receive, _send):
+        nonlocal downstream_called
+        downstream_called = True
+
+    async def receive():
+        return {"type": "http.request", "body": b"", "more_body": False}
+
+    async def send(message):
+        sent.append(message)
+
+    middleware = RequestBodyLimitMiddleware(downstream)
+    await middleware(
+        {
+            "type": "http",
+            "method": "POST",
+            "root_path": "/dashboard",
+            "path": "/api/v1/session",
+            "headers": [(b"content-length", b"20000")],
+        },
+        receive,
+        send,
+    )
+
+    assert sent[0]["status"] == 413
+    assert downstream_called is False