siftlode/backend/tests/test_body_limit.py

"""SEC-20 — the global request-body cap (`app/bodylimit.py`).

Two halves, deliberately kept apart:

* the MIDDLEWARE tests drive a toy ASGI app, so they pin the mechanism (declared length, counted
  bytes, the exemption match) with no router, no auth and no database in the way;
* the APP tests drive the real application, so they pin that the middleware is actually wired in,
  sits ahead of the auth dependencies, and refuses in the same shape `capped_body` does.

The pair that matters most is "an oversized body is refused" together with "a normal body still
gets through": the first alone would pass just as well if the cap refused *everything*.
"""

import json

import pytest
from fastapi.testclient import TestClient
from starlette.requests import Request

from app import bodylimit
from app.bodylimit import DEFAULT_MAX_BODY_BYTES, BodyLimitMiddleware
from app.routes import diagnostics as diagnostics_routes

SMALL = b"x" * 16
OVER = b"x" * (DEFAULT_MAX_BODY_BYTES + 1)


async def _reading_app(scope, receive, send):
    """A toy app that reads the whole body the way a route does, then reports its length.

    It uses a real Starlette `Request`, so the faked `http.disconnect` reaches it as the real
    ClientDisconnect a real handler would see — the middleware's exception path is exercised, not
    stubbed around.
    """
    body = await Request(scope, receive).body()
    payload = json.dumps({"read": len(body)}).encode()
    await send(
        {
            "type": "http.response.start",
            "status": 200,
            "headers": [(b"content-type", b"application/json")],
        }
    )
    await send({"type": "http.response.body", "body": payload})


def _client(**kwargs) -> TestClient:
    return TestClient(BodyLimitMiddleware(_reading_app, **kwargs))


def _chunks(total: int, size: int = 64 * 1024):
    """A body with NO Content-Length: httpx sends an iterator chunked, which is exactly the shape a
    declared-length check cannot see."""
    sent = 0
    while sent < total:
        n = min(size, total - sent)
        yield b"x" * n
        sent += n


# --- the mechanism ------------------------------------------------------------------------------


def test_a_declared_length_over_the_cap_is_refused_without_reading_the_body():
    r = _client().post("/anything", content=OVER)
    assert r.status_code == 413
    assert r.content == bodylimit.TOO_LARGE_BODY


def test_a_body_under_the_cap_is_passed_through_untouched():
    # The mutation guard for the test above: if the cap refused everything, this would fail.
    r = _client().post("/anything", content=SMALL)
    assert r.status_code == 200
    assert r.json() == {"read": len(SMALL)}


def test_a_chunked_body_is_refused_on_the_COUNTED_bytes():
    # No Content-Length at all, so the only thing that can stop this is counting as it arrives.
    r = _client().post("/anything", content=_chunks(DEFAULT_MAX_BODY_BYTES + 64 * 1024))
    assert r.status_code == 413
    assert r.content == bodylimit.TOO_LARGE_BODY


def test_a_chunked_body_under_the_cap_still_arrives_whole():
    size = DEFAULT_MAX_BODY_BYTES - 1
    r = _client().post("/anything", content=_chunks(size))
    assert r.status_code == 200
    assert r.json() == {"read": size}


def test_an_understated_content_length_does_not_buy_a_pass():
    # The declared-length check is the cheap half; a client that lies must still be counted out.
    r = _client().post("/anything", content=OVER, headers={"content-length": "10"})
    assert r.status_code == 413


def test_exactly_at_the_cap_is_allowed():
    size = DEFAULT_MAX_BODY_BYTES
    r = _client(limit=size).post("/anything", content=b"x" * size)
    assert r.status_code == 200
    assert r.json() == {"read": size}


# --- the exemption list -------------------------------------------------------------------------


def test_an_exempt_route_may_send_a_body_over_the_global_cap():
    client = _client(exempt=frozenset({("POST", "/api/things/{tid}/blob")}))
    r = client.post("/api/things/abc/blob", content=OVER)
    assert r.status_code == 200
    assert r.json() == {"read": len(OVER)}


def test_the_exemption_is_scoped_to_its_method_and_its_path():
    client = _client(exempt=frozenset({("POST", "/api/things/{tid}/blob")}))
    # Same path, different method.
    assert client.put("/api/things/abc/blob", content=OVER).status_code == 413
    # A neighbour that merely starts the same way.
    assert client.post("/api/things/abc/blob/extra", content=OVER).status_code == 413


def test_a_path_parameter_cannot_swallow_a_slash():
    # `{tid}` matching greedily across `/` would hand the exemption to routes that never got one.
    mw = BodyLimitMiddleware(_reading_app, exempt=frozenset({("POST", "/a/{tid}/b")}))
    assert mw.is_exempt("POST", "/a/one/b")
    assert not mw.is_exempt("POST", "/a/one/two/b")


def _declared_routes(root) -> set[tuple[str, str]]:
    """Every (method, path template) the app serves.

    It has to RECURSE: since FastAPI 0.141 / Starlette 1.6 an `include_router` leaves an opaque
    `_IncludedRouter` in `app.routes` and keeps the real routes under its `original_router`, instead
    of flattening them into the parent as older versions did. Reading `app.routes` alone finds 35
    entries and none of ours; walking the tree finds all 252.
    """
    found: set[tuple[str, str]] = set()
    stack = [root]
    while stack:
        node = stack.pop()
        for route in getattr(node, "routes", ()) or ():
            inner = getattr(route, "original_router", None)
            if inner is not None:
                stack.append(inner)
                continue
            for method in getattr(route, "methods", None) or ():
                found.add((method, route.path))
    return found


@pytest.mark.parametrize(("method", "template"), sorted(bodylimit.EXEMPT_ROUTES))
def test_every_exempt_template_still_resolves_to_a_real_route(method, template):
    """The drift guard. The exemption list names route templates that live in another file; a
    renamed route must fail the gate here, not lose its exemption silently in production."""
    from app.main import app

    assert (method, template) in _declared_routes(app)


def test_a_route_that_caps_itself_below_the_global_ceiling_is_NOT_exempt():
    """The admission rule, enforced rather than described: an exemption is for routes whose own cap
    is LARGER than the global one. The client log sink caps itself at 512 KiB, so the global ceiling
    never fires on it and exempting it would only widen what it accepts."""
    assert diagnostics_routes.BODY_MAX_BYTES <= DEFAULT_MAX_BODY_BYTES
    assert not any(path.startswith("/api/diagnostics") for _, path in bodylimit.EXEMPT_ROUTES)


def test_the_screenshot_exemption_is_justified_by_its_own_larger_cap():
    from app import debugsession

    assert debugsession.SCREENSHOT_MAX_BYTES > DEFAULT_MAX_BODY_BYTES
    assert ("POST", "/api/debug-sessions/{sid}/screenshots") in bodylimit.EXEMPT_ROUTES


# --- the real application -----------------------------------------------------------------------


def test_the_login_route_is_refused_before_a_single_dependency_runs(client, monkeypatch):
    """`/auth/password-login` is the reason SEC-20 exists: it is anonymous, it declares a body, and
    FastAPI reads that body in full before the first dependency. `_login_limiter.allow` is the very
    first statement of the handler, so it is the sharpest available proof that nothing ran."""
    from app.routes import auth as auth_routes

    reached = False

    def _record(*args, **kwargs):
        nonlocal reached
        reached = True
        return True

    monkeypatch.setattr(auth_routes._login_limiter, "allow", _record)

    r = client.post("/auth/password-login", content=OVER, headers={"content-type": "text/plain"})
    assert r.status_code == 413
    assert not reached, "the handler ran — the cap is inside the router instead of ahead of it"


def test_a_normal_login_body_still_reaches_the_handler(client, monkeypatch):
    """The mutation check for the test above. If the cap refused every body, the 413 test would
    pass while the app was broken; this one would not."""
    from app.routes import auth as auth_routes

    reached = False

    def _record(*args, **kwargs):
        nonlocal reached
        reached = True
        return True

    monkeypatch.setattr(auth_routes._login_limiter, "allow", _record)

    r = client.post("/auth/password-login", json={"email": "nobody@example.com", "password": "x"})
    assert r.status_code == 401
    assert reached


def test_the_global_refusal_is_byte_identical_to_the_per_route_one(db, as_user, make_user):
    """Two shapes for one fact is how the `cancelled`/`canceled` bug happened. The log sink caps
    itself with `capped_body`; its 413 and the middleware's must be the same bytes.

    Signed in on purpose: `current_user` resolves before the body dependency, so an anonymous call
    would be answered 401 and never reach the per-route cap this compares against. The body is over
    the SINK's 512 KiB but under the global 1 MiB, so it is `capped_body` that refuses it.
    """
    user = make_user()
    db.commit()
    over_the_sink = b"x" * (diagnostics_routes.BODY_MAX_BYTES + 1)
    assert len(over_the_sink) < DEFAULT_MAX_BODY_BYTES
    r = as_user(user).post(
        "/api/diagnostics/events",
        content=over_the_sink,
        headers={"content-type": "application/json"},
    )
    assert r.status_code == 413
    assert r.json() == json.loads(bodylimit.TOO_LARGE_BODY)