haproxy-openmanager/backend/tests/test_request_log_router_auth.py

"""v1.11.0: the request log API is gated, and its routes resolve.

Two distinct failure modes are pinned here.

**Auth.** The table holds redacted-but-real request and response bodies for
every user, so an unauthenticated or under-privileged caller must never get a
row. There is no database in this suite, so the behavioural checks assert only
that an anonymous call is rejected before any DB work — which is exactly the
property that matters — and a source scan covers the per-endpoint permission.

**Route order.** `/{log_id}` is a single-segment path and FastAPI matches in
declaration order, so declaring it before `/settings`, `/stats` or `/purge`
makes those three unreachable (they parse as a log id and 422). This is the
mirror image of the shadowing trap already present in routers/settings.py.
"""
import os
import re
import sys

import pytest

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

_BACKEND = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
_ROUTER = os.path.join(_BACKEND, "routers", "request_logs.py")
_MAIN = os.path.join(_BACKEND, "main.py")

REJECT = (401, 403, 422)


@pytest.fixture(scope="module")
def src():
    with open(_ROUTER, encoding="utf-8") as f:
        return f.read()


# --------------------------------------------------------------------------
# Behavioural: nothing is readable without credentials
# --------------------------------------------------------------------------

@pytest.mark.parametrize("method,path", [
    ("get", "/api/request-logs"),
    ("get", "/api/request-logs/1"),
    ("get", "/api/request-logs/stats"),
    ("get", "/api/request-logs/settings"),
    ("put", "/api/request-logs/settings"),
    ("post", "/api/request-logs/purge"),
])
def test_anonymous_access_is_rejected(client, method, path):
    res = getattr(client, method)(path) if method != "put" else client.put(path, json={})
    assert res.status_code in REJECT, (
        f"{method.upper()} {path} returned {res.status_code} without an Authorization "
        f"header — the request log contains captured bodies for every user"
    )


def test_a_garbage_token_is_rejected(client):
    res = client.get("/api/request-logs", headers={"authorization": "Bearer not-a-token"})
    assert res.status_code in REJECT


# --------------------------------------------------------------------------
# Source scan: per-endpoint permission
# --------------------------------------------------------------------------

def _handler_body(src, decorator):
    start = src.index(decorator)
    rest = src[start + len(decorator):]
    end = rest.find("\n@router.")
    return rest if end == -1 else rest[:end]


@pytest.mark.parametrize("decorator,action", [
    ('@router.get("/settings")', "manage"),
    ('@router.put("/settings")', "manage"),
    ('@router.get("/stats")', "read"),
    ('@router.post("/purge")', "manage"),
    ('@router.get("")', "read"),
    ('@router.get("/{log_id}")', "read"),
])
def test_every_endpoint_enforces_its_permission(src, decorator, action):
    body = _handler_body(src, decorator)
    assert f'_require(authorization, "{action}")' in body, (
        f"{decorator} does not enforce requestlog.{action}"
    )


def test_require_helper_raises_403_not_a_silent_pass(src):
    helper = src.split("async def _require", 1)[1].split("\nasync def ", 1)[0]
    assert "check_user_permission" in helper
    assert "status_code=403" in helper
    assert "current_user=current_user" in helper, (
        "the admin bypass is skipped, so every call pays an extra SELECT on users"
    )


# --------------------------------------------------------------------------
# Route declaration order
# --------------------------------------------------------------------------

@pytest.mark.parametrize("literal", ['@router.get("/settings")', '@router.put("/settings")',
                                     '@router.get("/stats")', '@router.post("/purge")'])
def test_literal_routes_are_declared_before_the_catch_all(src, literal):
    catch_all = src.index('@router.get("/{log_id}")')
    assert src.index(literal) < catch_all, (
        f"{literal} is declared after GET /{{log_id}}. FastAPI matches in declaration "
        f"order and /{{log_id}} is a single-segment path, so it would swallow this route "
        f"and the request would fail parsing 'settings' as an int."
    )


def test_list_route_is_declared_before_the_catch_all(src):
    assert src.index('@router.get("")') < src.index('@router.get("/{log_id}")')


# --------------------------------------------------------------------------
# Query construction
# --------------------------------------------------------------------------

def test_filters_are_bound_never_interpolated(src):
    """User-supplied filters reach the WHERE clause; they must arrive as $n
    parameters."""
    body = _handler_body(src, '@router.get("")')
    # The only f-string interpolation allowed into SQL is the placeholder index
    # and the assembled clause list, never a raw value.
    for match in re.findall(r'add\("([^"]+)"', body):
        assert "{n}" in match, f"filter clause {match!r} does not use a bound placeholder"


def test_list_endpoint_scopes_non_privileged_callers_to_themselves(src):
    body = _handler_body(src, '@router.get("")')
    assert "if not can_manage:" in body
    assert "direction = 'inbound' AND user_id =" in body, (
        "a caller with only requestlog.read can see every other user's captured request "
        "bodies"
    )


def test_detail_endpoint_applies_the_same_scoping(src):
    body = _handler_body(src, '@router.get("/{log_id}")')
    assert "can_manage" in body
    assert "404" in body, (
        "the detail endpoint should 404 rather than 403 for a row the caller may not see, "
        "so it does not confirm which ids exist"
    )


def test_list_response_omits_bodies(src):
    """A 200-row page carrying two 8 KB JSONB blobs per row is a multi-megabyte
    response; bodies belong to the detail endpoint."""
    columns = src.split("_LIST_COLUMNS = ", 1)[1].split('"""', 2)[1]
    assert "request_body," not in columns
    assert "response_body," not in columns
    assert "request_body_bytes" in columns, "the size is still useful in the list"


def test_count_is_bounded(src):
    body = _handler_body(src, '@router.get("")')
    assert "LIMIT {count_cap}" in body or "count_cap" in body, (
        "an unbounded COUNT(*) over request_logs is a sequential scan on every page change"
    )
    assert "total_is_estimate" in body


# --------------------------------------------------------------------------
# Registration
# --------------------------------------------------------------------------

def test_router_is_registered_in_main():
    with open(_MAIN, encoding="utf-8") as f:
        main_src = f.read()

    assert "from routers.request_logs import router as request_logs_router" in main_src
    assert "app.include_router(request_logs_router)" in main_src, (
        "the router is imported but never mounted, so every endpoint 404s"
    )