cyclone/backend/tests/test_security.py

"""SP19 — Security middleware + health probe tests.

Covers each middleware in isolation (using FastAPI's TestClient with a
minimal app) and the integration into the real ``cyclone.api`` app
(headers present on every response, /api/health returns the rich
snapshot).
"""
from __future__ import annotations

import asyncio
from typing import Any, Callable

import pytest
from fastapi import FastAPI, Request
from fastapi.testclient import TestClient

from cyclone.security import (
    DEFAULT_MAX_BODY_BYTES,
    DEFAULT_RATE_LIMIT_PER_MIN,
    BodySizeLimitMiddleware,
    RateLimitMiddleware,
    SecurityHeadersMiddleware,
    get_health_snapshot,
)


# ---------------------------------------------------------------------------
# Test apps — built inline per-test to avoid state pollution between tests.
# ---------------------------------------------------------------------------


def _echo_app_with(
    *middlewares: Callable[..., Any],
) -> FastAPI:
    """Build a tiny FastAPI app with the given middleware chain.

    ``middlewares`` are listed outermost-first (the first element
    runs first on the request). Starlette's ``add_middleware``
    *prepends*, so we add in reverse to preserve "outermost first"
    in the public API.
    """
    app = FastAPI()

    @app.post("/echo")
    async def echo(request: Request):
        body = await request.body()
        return {"received_bytes": len(body)}

    @app.get("/api/health")
    async def h():
        return {"status": "ok"}

    for mw in reversed(middlewares):
        app.add_middleware(mw)
    return app


# ---------------------------------------------------------------------------
# BodySizeLimitMiddleware
# ---------------------------------------------------------------------------


def test_body_size_accepts_under_limit():
    """500 bytes is under the 1024-byte limit; body passes through."""
    app = _echo_app_with(
        SecurityHeadersMiddleware,
        lambda a: BodySizeLimitMiddleware(a, max_bytes=1024),
    )
    client = TestClient(app)
    resp = client.post("/echo", content=b"x" * 500)
    assert resp.status_code == 200, resp.text
    assert resp.json() == {"received_bytes": 500}


def test_body_size_rejects_over_content_length():
    """A 200-byte body against a 100-byte cap returns 413."""
    app = _echo_app_with(
        SecurityHeadersMiddleware,
        lambda a: BodySizeLimitMiddleware(a, max_bytes=100),
    )
    client = TestClient(app)
    resp = client.post("/echo", content=b"x" * 200)
    assert resp.status_code == 413
    body = resp.json()
    assert body["error"] == "body_too_large"
    assert "100" in body["detail"]


def test_body_size_default_is_50mb():
    """The default cap is 50 MB so even large prodfiles fit."""
    assert DEFAULT_MAX_BODY_BYTES == 50 * 1024 * 1024


def test_body_size_rejects_bad_content_length():
    """A non-integer Content-Length is rejected as 400, not 500."""
    sent: list = []
    inner_app = FastAPI()

    @inner_app.post("/echo")
    async def echo(request: Request):
        return {"ok": True}

    app_instance = BodySizeLimitMiddleware(inner_app, max_bytes=100)

    async def fake_receive():
        return {"type": "http.request", "body": b"", "more_body": False}

    async def fake_send(msg):
        sent.append(msg)

    scope = {
        "type": "http",
        "method": "POST",
        "path": "/echo",
        "headers": [(b"content-length", b"not-a-number")],
        "query_string": b"",
    }
    asyncio.run(app_instance(scope, fake_receive, fake_send))
    start = next(m for m in sent if m["type"] == "http.response.start")
    assert start["status"] == 400


# ---------------------------------------------------------------------------
# RateLimitMiddleware
# ---------------------------------------------------------------------------


def test_rate_limit_default_is_300_per_min():
    assert DEFAULT_RATE_LIMIT_PER_MIN == 300


def test_rate_limit_allows_under_threshold():
    """A few requests under the per-minute limit are allowed."""
    app = _echo_app_with(
        SecurityHeadersMiddleware,
        lambda a: RateLimitMiddleware(a, per_minute=5),
    )
    client = TestClient(app)
    for _ in range(5):
        resp = client.post("/echo", content=b"x")
        assert resp.status_code == 200, resp.text


def test_rate_limit_blocks_over_threshold():
    """The 6th request within a 60s window is rate-limited."""
    app = _echo_app_with(
        SecurityHeadersMiddleware,
        lambda a: RateLimitMiddleware(a, per_minute=3),
    )
    client = TestClient(app)
    for _ in range(3):
        assert client.post("/echo", content=b"x").status_code == 200
    resp = client.post("/echo", content=b"x")
    assert resp.status_code == 429
    assert resp.json()["error"] == "rate_limited"


def test_rate_limit_exempts_health_probes():
    """A load balancer hammering /api/health should not trip the limiter."""
    app = _echo_app_with(
        SecurityHeadersMiddleware,
        lambda a: RateLimitMiddleware(a, per_minute=2),
    )
    client = TestClient(app)
    for _ in range(20):
        assert client.get("/api/health").status_code == 200
    assert client.post("/echo", content=b"x").status_code == 200


# ---------------------------------------------------------------------------
# SecurityHeadersMiddleware
# ---------------------------------------------------------------------------


def test_security_headers_present_on_200():
    app = _echo_app_with(SecurityHeadersMiddleware)
    client = TestClient(app)
    resp = client.get("/api/health")
    assert resp.headers["X-Content-Type-Options"] == "nosniff"
    assert resp.headers["X-Frame-Options"] == "DENY"
    assert resp.headers["Referrer-Policy"] == "same-origin"
    assert "default-src 'none'" in resp.headers["Content-Security-Policy"]


def test_security_headers_present_on_error_response():
    """413/429 still carry the security headers."""
    # Define a tiny factory so Starlette can introspect it as a class.
    class _BoundBodySize(BodySizeLimitMiddleware):
        def __init__(self, app):  # type: ignore[no-untyped-def]
            super().__init__(app, max_bytes=10)

    app = _echo_app_with(SecurityHeadersMiddleware, _BoundBodySize)
    client = TestClient(app)
    resp = client.post("/echo", content=b"x" * 100)
    assert resp.status_code == 413, resp.text
    assert resp.headers["X-Content-Type-Options"] == "nosniff"


# ---------------------------------------------------------------------------
# get_health_snapshot
# ---------------------------------------------------------------------------


def test_health_snapshot_basic_shape():
    snap = get_health_snapshot()
    d = snap.to_dict()
    assert "status" in d
    assert "version" in d
    assert "db" in d
    assert "scheduler" in d
    assert "pubsub" in d
    assert "batch" in d


def test_health_snapshot_db_ok_in_tests():
    """The conftest DB fixture is live; ``SELECT 1`` works."""
    snap = get_health_snapshot()
    assert snap.db.get("ok") is True


def test_health_snapshot_handles_no_scheduler():
    """Without a configured scheduler, the snapshot reports gracefully."""
    snap = get_health_snapshot()
    sched = snap.scheduler
    assert "running" in sched or "configured" in sched