155 lines
4.9 KiB
Python
155 lines
4.9 KiB
Python
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", "/api/v1/repos/stackchain/project/issues/17/attachments"
|
|
)
|
|
== 3 * 1024 * 1024
|
|
)
|
|
assert main.request_body_limit("POST", "/unrelated/attachments") 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
|