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 ast
import json
import pathlib

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

from app import bodylimit, debugsession, sysconfig
from app.config import settings
from app.models import SystemConfig
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 per-route ceilings ---------------------------------------------------------------------

BIG = {("POST", "/api/things/{tid}/blob"): DEFAULT_MAX_BODY_BYTES * 4}


def test_a_route_with_a_raised_ceiling_may_send_a_body_over_the_global_cap():
    r = _client(routes=BIG).post("/api/things/abc/blob", content=OVER)
    assert r.status_code == 200
    assert r.json() == {"read": len(OVER)}


def test_a_raised_ceiling_is_still_a_ceiling():
    """The difference between this and an exemption, and the reason the list is a mapping: a route
    that needs more bytes does not thereby get unlimited bytes. Left uncounted, the upload route
    would let an ANONYMOUS caller stream without bound — FastAPI parses a declared body before the
    first dependency, so `require_human` has not run yet when the bytes land."""
    over = b"x" * (BIG[("POST", "/api/things/{tid}/blob")] + 1)
    r = _client(routes=BIG).post("/api/things/abc/blob", content=over)
    assert r.status_code == 413
    assert r.content == bodylimit.TOO_LARGE_BODY


def test_a_raised_ceiling_counts_a_chunked_body_too():
    cap = BIG[("POST", "/api/things/{tid}/blob")]
    r = _client(routes=BIG).post("/api/things/abc/blob", content=_chunks(cap + 64 * 1024))
    assert r.status_code == 413


def test_the_raised_ceiling_is_scoped_to_its_method_and_its_path():
    client = _client(routes=BIG)
    # 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 raised ceiling to routes without one.
    mw = BodyLimitMiddleware(_reading_app, routes={("POST", "/a/{tid}/b"): 99})
    assert mw.limit_for("POST", "/a/one/b") == 99
    assert mw.limit_for("POST", "/a/one/two/b") == DEFAULT_MAX_BODY_BYTES


def test_every_request_gets_a_finite_ceiling():
    """There is no pass-through path. A route that is not listed gets the default; one that is
    listed gets its own; nothing is simply not counted."""
    mw = BodyLimitMiddleware(_reading_app)
    assert mw.limit_for("GET", "/api/anything") == DEFAULT_MAX_BODY_BYTES
    for (method, template), cap in bodylimit.ROUTE_MAX_BODY_BYTES.items():
        assert cap > DEFAULT_MAX_BODY_BYTES  # the admission rule: raised, never lowered
        assert mw.limit_for(method, template.replace("{sid}", "abc")) == cap


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.ROUTE_MAX_BODY_BYTES))
def test_every_listed_template_still_resolves_to_a_real_route(method, template):
    """The drift guard. The list names route templates that live in another file; a renamed route
    must fail the gate here, not lose its allowance silently in production."""
    from app.main import app

    assert (method, template) in _declared_routes(app)


def test_the_default_ceiling_clears_the_largest_body_a_default_route_takes():
    """The headroom the module's docblock claims, held as a RATIO rather than as a figure in prose.
    The client log sink caps itself at 512 KiB — below the default, which is why it is not listed:
    the global ceiling never fires on it and raising it could only widen what it accepts.

    `<=` alone would not hold the claim. The docblock calls the 2x margin load-bearing precisely
    because a ceiling that refuses a legitimate request is a worse bug than the one this module
    fixes, and a margin of 1x is no margin at all.
    """
    assert diagnostics_routes.BODY_MAX_BYTES * 2 <= DEFAULT_MAX_BODY_BYTES
    assert not any(
        path.startswith("/api/diagnostics") for _, path in bodylimit.ROUTE_MAX_BODY_BYTES
    )


# Every `capped_body` call site in the routers, as (module, limit constant). A route that reads its
# own body declares the size it MEANS to accept; if that is above the global default and its route
# is not in ROUTE_MAX_BODY_BYTES, this middleware refuses a request the route would have taken —
# silently, in production, on a path nobody tested. The point of the list is the third entry.
CAPPED_BODY_SITES = {
    ("diagnostics.py", "BODY_MAX_BYTES"),
    ("debug_sessions.py", "debugsession.SCREENSHOT_MAX_BYTES"),
}


def test_every_capped_body_site_is_known_and_fits_under_its_ceiling():
    """Parsed, not grepped. A regex over the call text has to guess how the arguments are spelled,
    and the shapes it guesses wrong — a bare `4 * 1024 * 1024`, a parameter not named `request` —
    are the natural ways the NEXT site gets written, so a miss reads as "no new sites" rather than
    as a failure. `ast` sees every call whatever it looks like."""
    routes_dir = pathlib.Path(__file__).resolve().parents[1] / "app" / "routes"
    found: set[tuple[str, str]] = set()
    for py in routes_dir.rglob("*.py"):
        tree = ast.parse(py.read_text("utf-8"), filename=str(py))
        for node in ast.walk(tree):
            if not isinstance(node, ast.Call):
                continue
            fn = node.func
            name = fn.attr if isinstance(fn, ast.Attribute) else getattr(fn, "id", None)
            if name != "capped_body":
                continue
            limit = next(
                (kw.value for kw in node.keywords if kw.arg == "limit"),
                node.args[1] if len(node.args) > 1 else None,
            )
            found.add((py.name, ast.unparse(limit) if limit is not None else "<none>"))
    assert found == CAPPED_BODY_SITES, (
        f"a capped_body call site appeared, moved or changed shape: {found ^ CAPPED_BODY_SITES}. "
        "Each one declares a body size; check it against bodylimit.DEFAULT_MAX_BODY_BYTES and, if "
        "it is larger, add its route to bodylimit.ROUTE_MAX_BODY_BYTES — then update this list."
    )
    # And the two known ones really do satisfy the rule.
    assert diagnostics_routes.BODY_MAX_BYTES <= DEFAULT_MAX_BODY_BYTES
    assert debugsession.SCREENSHOT_MAX_BYTES > DEFAULT_MAX_BODY_BYTES
    assert ("POST", "/api/debug-sessions/{sid}/screenshots") in bodylimit.ROUTE_MAX_BODY_BYTES


def test_the_upload_ceiling_and_the_admin_knob_cannot_drift_apart():
    """The upload route's in-handler cap is a `sysconfig` value an admin can change; the edge is a
    constant. If the knob could be set above the edge, an admin raising it would silently start
    getting 413s from a layer the admin page never mentions.

    Since P1 the two count the SAME bytes — the body is the file, with no multipart envelope
    wrapped round it — so the old allowance is gone and the edge is the knob's own maximum,
    exactly. A gap reappearing means someone re-wrapped the upload in an envelope without saying so.
    """
    spec = next(s for s in sysconfig.SPECS if s.key == "download_upload_max_bytes")
    edge = bodylimit.ROUTE_MAX_BODY_BYTES[("POST", "/api/downloads/upload")]
    assert spec.max == edge


def _route_objects(root):
    """`_declared_routes`' sibling, keeping the route OBJECT so its FastAPI dependant can be read."""
    found: dict[tuple[str, str], object] = {}
    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[(method, route.path)] = route
    return found


@pytest.mark.parametrize(("method", "template"), sorted(bodylimit.ROUTE_MAX_BODY_BYTES))
def test_no_route_with_a_raised_ceiling_declares_a_body_parameter(method, template):
    """P1's invariant, and the reason this is a test rather than a comment.

    A raised ceiling is only honest when the route's own cap applies to the bytes AS THEY ARRIVE.
    FastAPI reads and spools a DECLARED body (`= File(...)`, `= Form(...)`, a Pydantic model)
    BEFORE the first dependency resolves — so on such a route this middleware's ceiling is the only
    thing between an anonymous caller and the whole allowance, `require_human` never gets a say, and
    the route's in-handler cap measures bytes that have already landed. That is exactly what P1
    closed on the upload route, which sat in `ROUTE_MAX_BODY_BYTES` with a 500 MiB ceiling and a
    `file: UploadFile = File(...)` body for one release.

    Asked of FastAPI itself (`route.dependant.body_params`) rather than parsed or grepped: the
    router already knows which parameters it will read the body for, and a shape a regex or an AST
    walk guesses wrong is precisely how the next route slips through."""
    from app.main import app

    route = _route_objects(app).get((method, template))
    assert route is not None, f"{method} {template} is not a route any more"
    body_params = [p.name for p in getattr(route.dependant, "body_params", ())]
    assert not body_params, (
        f"{method} {template} declares a body parameter ({', '.join(body_params)}), so FastAPI "
        "spools it before the first dependency runs and the raised ceiling is all that stands "
        "between an anonymous caller and the whole allowance. Read the body yourself, from a "
        "dependency declared AFTER the auth one — see routes/downloads.py:spool_upload."
    )


def test_a_stored_override_outside_the_bounds_falls_back_on_READ(db):
    """`set_value` refuses an out-of-range int, but neither bound has always been what it is now —
    the upload knob had no `max` before SEC-20 and its `min` was 0 before E7 — so a value that was
    legal when it was written can be sitting in the database. Served as-is it is exactly the drift
    the test above forbids: the route would accept bytes the edge refuses.

    The fallback is the env default and a warning, NOT a clamp. A clamp is a silent rewrite, and on
    the min side it would be a wrong one: a stored 0 meant "no per-file cap" to the route, and
    clamped to 1 it becomes a one-byte cap that kills uploads with nothing in the log.
    """
    spec = next(s for s in sysconfig.SPECS if s.key == "download_upload_max_bytes")
    assert spec.max is not None and spec.min is not None
    default = settings.download_upload_max_bytes

    row = SystemConfig(key="download_upload_max_bytes", value=str(spec.max * 4))
    db.add(row)
    db.commit()
    assert sysconfig.get_int(db, "download_upload_max_bytes") == default

    row.value = "0"  # the legacy "no per-file cap", now below min
    db.commit()
    assert sysconfig.get_int(db, "download_upload_max_bytes") == default

    # An override that IS in range must still be honoured — otherwise a bounds predicate written
    # backwards would make every override inert and the two assertions above could not tell.
    # Deliberately not `spec.max`: that equals the env default for this key, so the assertion would
    # be satisfied by the fallback it is meant to rule out.
    row.value = str(spec.max - 1)
    db.commit()
    assert spec.max - 1 != default
    assert sysconfig.get_int(db, "download_upload_max_bytes") == spec.max - 1


def test_the_cap_sits_inside_the_response_decorators_and_outside_the_setup_gate():
    """The registration order, held in code rather than in the comment beside it. `add_middleware`
    inverts the reading order (last registered = OUTERMOST), which is exactly the kind of thing a
    prose description gets backwards — it already did once.

    The layers are named, not counted by class: all four `@app.middleware("http")` decorators —
    including the setup gate, which must be INSIDE — register as the same `BaseHTTPMiddleware`, so a
    class-shaped assertion passes for every arrangement anyone would plausibly get wrong.

    Outside the cap: only response decorators, so the 413 is dressed like any other answer. Inside
    it: everything that does work on the caller's behalf — the setup gate opens a DB session per
    request while an instance is unconfigured, which is the anonymous pre-auth work SEC-20 exists to
    remove.
    """
    from app.main import app

    def layer(m) -> str:
        dispatch = m.kwargs.get("dispatch") if m.kwargs else None
        return dispatch.__name__ if dispatch is not None else getattr(m.cls, "__name__", str(m.cls))

    names = [layer(m) for m in app.user_middleware]  # outer → inner
    i = names.index("BodyLimitMiddleware")
    assert set(names[:i]) <= {
        "log_requests",
        "security_headers",
        "static_cache_headers",
        "CORSMiddleware",
    }, f"something that is not a response decorator sits outside the cap: {names[:i]}"
    assert "log_requests" in names[:i], "the access log must stay outside, so a 413 is logged"
    assert "setup_gate" in names[i + 1 :], "the setup gate must not run for an oversized body"
    assert "SessionMiddleware" in names[i + 1 :]


def test_the_screenshot_ceiling_matches_the_cap_the_route_itself_enforces():
    assert debugsession.SCREENSHOT_MAX_BYTES > DEFAULT_MAX_BODY_BYTES
    assert (
        bodylimit.ROUTE_MAX_BODY_BYTES[("POST", "/api/debug-sessions/{sid}/screenshots")]
        == debugsession.SCREENSHOT_MAX_BYTES
    )


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


def test_the_global_refusal_carries_the_baseline_security_headers():
    """The 413 is a response like any other, so it has to be dressed like one. The middleware sits
    INSIDE the security-header and CORS layers for exactly this reason: a refusal a browser drops
    for a missing header is indistinguishable from a network failure, and the user would be shown a
    generic error instead of the translated `too_large`.

    A bare TestClient on purpose, not the `client` fixture: this is the only test that exercises the
    middleware ORDER, and that fixture is DB-gated — so on a host without Postgres the one assertion
    holding the move would have skipped while the file still reported green. The refusal never
    reaches a database; it is answered above the router.
    """
    from app.main import app

    r = TestClient(app).post(
        "/auth/password-login", content=OVER, headers={"content-type": "text/plain"}
    )
    assert r.status_code == 413
    assert r.headers.get("x-content-type-options") == "nosniff"
    assert "referrer-policy" in r.headers


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
    # RAW BYTES, not parsed JSON: the two used to differ by the two spaces `json.dumps` puts in by
    # default and `JSONResponse` does not, and a structural comparison could not see it — which is
    # the only kind of drift the docblock's promise was ever at risk from.
    assert r.content == bodylimit.TOO_LARGE_BODY