129 lines
4.2 KiB
Python
129 lines
4.2 KiB
Python
"""Regression coverage for the unauthenticated Gateway health probes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.gateway import auth_middleware
|
|
from app.gateway.app import app as gateway_app
|
|
from app.gateway.auth.token_source import get_explicit_or_session_token
|
|
from app.gateway.auth_middleware import PUBLIC_HEALTH_PATHS, AuthMiddleware
|
|
|
|
|
|
def _health_only_app() -> FastAPI:
|
|
app = FastAPI()
|
|
|
|
async def health_check() -> dict[str, str]:
|
|
return {"status": "healthy", "service": "deer-flow-gateway"}
|
|
|
|
for path in PUBLIC_HEALTH_PATHS:
|
|
app.add_api_route(path, health_check, methods=["GET", "HEAD"])
|
|
app.add_middleware(AuthMiddleware)
|
|
return app
|
|
|
|
|
|
@pytest.mark.parametrize("path", sorted(PUBLIC_HEALTH_PATHS))
|
|
@pytest.mark.parametrize("method", ["GET", "HEAD"])
|
|
def test_health_probe_is_public_when_auth_is_enabled(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
path: str,
|
|
method: str,
|
|
) -> None:
|
|
monkeypatch.setattr(auth_middleware, "is_auth_disabled", lambda: False)
|
|
|
|
with TestClient(_health_only_app()) as client:
|
|
client.cookies.set("access_token", "invalid-token")
|
|
response = client.request(
|
|
method,
|
|
path,
|
|
headers={"Authorization": "Bearer invalid-token"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
if method == "GET":
|
|
assert response.json() == {"status": "healthy", "service": "deer-flow-gateway"}
|
|
else:
|
|
assert response.content == b""
|
|
|
|
|
|
@pytest.mark.parametrize("path", sorted(PUBLIC_HEALTH_PATHS))
|
|
def test_health_probe_allows_a_trailing_slash(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
path: str,
|
|
) -> None:
|
|
monkeypatch.setattr(auth_middleware, "is_auth_disabled", lambda: False)
|
|
|
|
with TestClient(_health_only_app(), follow_redirects=False) as client:
|
|
response = client.get(f"{path}/")
|
|
|
|
# FastAPI redirects to the canonical route; importantly, auth must not turn
|
|
# the request into a 401 before routing can perform that redirect.
|
|
assert response.status_code == 307
|
|
|
|
|
|
@pytest.mark.parametrize("path", sorted(PUBLIC_HEALTH_PATHS))
|
|
def test_health_probe_is_public_behind_a_reverse_proxy_prefix(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
path: str,
|
|
) -> None:
|
|
monkeypatch.setattr(auth_middleware, "is_auth_disabled", lambda: False)
|
|
|
|
with TestClient(_health_only_app(), root_path="/deerflow") as client:
|
|
response = client.get(f"/deerflow{path}")
|
|
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_gateway_registers_every_public_health_path_for_get_and_head() -> None:
|
|
methods_by_path = {
|
|
route.path: route.methods
|
|
for route in gateway_app.routes
|
|
if hasattr(route, "methods")
|
|
}
|
|
|
|
for path in PUBLIC_HEALTH_PATHS:
|
|
assert {"GET", "HEAD"} <= methods_by_path[path]
|
|
|
|
|
|
def test_non_health_route_still_requires_auth(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(auth_middleware, "is_auth_disabled", lambda: False)
|
|
app = _health_only_app()
|
|
|
|
@app.get("/private")
|
|
async def private() -> dict[str, bool]:
|
|
return {"ok": True}
|
|
|
|
with TestClient(app) as client:
|
|
response = client.get("/private")
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
def _auth_request(*, authorization: str = "", cookie: str = "") -> SimpleNamespace:
|
|
headers = {"authorization": authorization} if authorization else {}
|
|
cookies = {"access_token": cookie} if cookie else {}
|
|
return SimpleNamespace(headers=headers, cookies=cookies)
|
|
|
|
|
|
def test_explicit_bearer_token_wins_over_shared_cookie() -> None:
|
|
"""Another tab's cookie must not override this tab's explicit token."""
|
|
request = _auth_request(authorization="Bearer current-tab", cookie="other-tab")
|
|
|
|
assert get_explicit_or_session_token(request) == "current-tab"
|
|
|
|
|
|
def test_cookie_remains_the_fallback_for_cookie_only_clients() -> None:
|
|
request = _auth_request(cookie="browser-session")
|
|
|
|
assert get_explicit_or_session_token(request) == "browser-session"
|
|
|
|
|
|
def test_malformed_authorization_header_does_not_hide_valid_cookie() -> None:
|
|
request = _auth_request(authorization="Basic abc", cookie="browser-session")
|
|
|
|
assert get_explicit_or_session_token(request) == "browser-session"
|