90 lines
2.6 KiB
Python
90 lines
2.6 KiB
Python
"""Input bounds of the unauthenticated OAuth routes (SEC-01, SEC-09)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from aegis_gitea_mcp.oauth.callback import MAX_CONSENT_BODY_BYTES
|
|
from tests.security.helpers import asgi_request
|
|
from tests.test_oauth_helpers import BASE, build_oauth_app
|
|
|
|
CHUNK = 64 * 1024
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _env(mock_env: None) -> None:
|
|
"""The audit logger and the rate limiter read the process-wide settings."""
|
|
|
|
|
|
def chunks_of(total: int) -> list[bytes]:
|
|
return [b"a" * CHUNK for _ in range(total // CHUNK)]
|
|
|
|
|
|
async def test_consent_post_over_the_cap_is_413_without_reading_the_body() -> None:
|
|
app = build_oauth_app().app
|
|
result = await asgi_request(
|
|
app,
|
|
"POST",
|
|
"/oauth/consent",
|
|
headers={
|
|
"content-type": "application/x-www-form-urlencoded",
|
|
"content-length": str(1024 * 1024),
|
|
"origin": BASE,
|
|
},
|
|
chunks=chunks_of(1024 * 1024),
|
|
)
|
|
assert result.status == 413
|
|
assert result.bytes_read == 0
|
|
|
|
|
|
async def test_consent_post_without_declared_length_stops_reading_at_the_cap() -> None:
|
|
app = build_oauth_app().app
|
|
result = await asgi_request(
|
|
app,
|
|
"POST",
|
|
"/oauth/consent",
|
|
headers={"content-type": "application/x-www-form-urlencoded", "origin": BASE},
|
|
chunks=chunks_of(1024 * 1024),
|
|
)
|
|
assert result.status == 413
|
|
assert result.bytes_read <= MAX_CONSENT_BODY_BYTES + CHUNK
|
|
|
|
|
|
async def test_consent_multipart_is_415_without_reading_any_body() -> None:
|
|
app = build_oauth_app().app
|
|
result = await asgi_request(
|
|
app,
|
|
"POST",
|
|
"/oauth/consent",
|
|
headers={"content-type": "multipart/form-data; boundary=x", "origin": BASE},
|
|
chunks=chunks_of(512 * 1024),
|
|
)
|
|
assert result.status == 415
|
|
assert result.bytes_read == 0
|
|
|
|
|
|
async def test_consent_json_body_is_415() -> None:
|
|
app = build_oauth_app().app
|
|
result = await asgi_request(
|
|
app,
|
|
"POST",
|
|
"/oauth/consent",
|
|
headers={"content-type": "application/json", "origin": BASE},
|
|
chunks=[b"{}"],
|
|
)
|
|
assert result.status == 415
|
|
|
|
|
|
async def test_small_urlencoded_consent_post_still_reaches_the_handler() -> None:
|
|
app = build_oauth_app().app
|
|
result = await asgi_request(
|
|
app,
|
|
"POST",
|
|
"/oauth/consent",
|
|
headers={"content-type": "application/x-www-form-urlencoded", "origin": BASE},
|
|
chunks=[b"request=unknown&csrf=x&decision=approve"],
|
|
)
|
|
# Reaches the handler, which answers its fixed 400 page for an unknown request.
|
|
assert result.status == 400
|
|
assert b"invalid or has expired" in result.body
|