86 lines
3 KiB
Python
86 lines
3 KiB
Python
"""Edge hardening: docs off, headers, body cap, run throttle, mail scrub."""
|
|
|
|
import pytest
|
|
|
|
from app.middleware import reset_throttle
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clean_throttle():
|
|
reset_throttle()
|
|
yield
|
|
reset_throttle()
|
|
|
|
|
|
async def test_docs_disabled_outside_debug(client):
|
|
assert (await client.get("/api/docs")).status_code == 404
|
|
assert (await client.get("/api/redoc")).status_code == 404
|
|
assert (await client.get("/api/openapi.json")).status_code == 404
|
|
|
|
|
|
async def test_security_headers_present(client):
|
|
response = await client.get("/health")
|
|
assert response.status_code == 200
|
|
assert response.headers["X-Content-Type-Options"] == "nosniff"
|
|
assert response.headers["X-Frame-Options"] == "SAMEORIGIN"
|
|
assert response.headers["Referrer-Policy"] == "strict-origin-when-cross-origin"
|
|
assert response.headers["Content-Security-Policy"] == "frame-ancestors 'self'"
|
|
|
|
|
|
async def test_oversize_body_rejected(client):
|
|
# Default cap is 2 MiB (settings.max_body_bytes); the middleware binds
|
|
# it at startup, so this test exceeds the real default.
|
|
big = "x" * (3 * 1024 * 1024)
|
|
response = await client.post("/api/auth/magic/consume", json={"token": big})
|
|
assert response.status_code == 413
|
|
|
|
|
|
async def test_run_throttle_returns_429(client, lfx_server, login, monkeypatch):
|
|
from app.settings import settings
|
|
|
|
limit = 2
|
|
monkeypatch.setattr(settings, "run_throttle_per_minute", limit)
|
|
await login()
|
|
|
|
flow = (await client.post("/api/flows", json={"name": "throttled"})).json()
|
|
await client.post(
|
|
f"/api/flows/{flow['id']}/versions",
|
|
json={
|
|
"langflow_json": {
|
|
"name": "t",
|
|
"data": {
|
|
"nodes": [
|
|
{
|
|
"id": "ChatInput-x",
|
|
"data": {
|
|
"id": "ChatInput-x",
|
|
"type": "ChatInput",
|
|
"node": {"template": {}, "outputs": []},
|
|
},
|
|
"type": "genericNode",
|
|
}
|
|
],
|
|
"edges": [],
|
|
},
|
|
}
|
|
},
|
|
)
|
|
# Undeployed: 422s still pass the throttle (only 429 matters here).
|
|
for _ in range(limit):
|
|
response = await client.post(f"/api/flows/{flow['id']}/run", json={"input": {}})
|
|
assert response.status_code == 422
|
|
response = await client.post(f"/api/flows/{flow['id']}/run", json={"input": {}})
|
|
assert response.status_code == 429
|
|
assert response.headers["Retry-After"]
|
|
|
|
|
|
def test_magic_link_never_logged(caplog, monkeypatch):
|
|
from app.auth import mail
|
|
from app.settings import settings
|
|
|
|
monkeypatch.setattr(settings, "smtp_host", "")
|
|
with caplog.at_level("INFO", logger="app.auth.mail"):
|
|
assert mail.send_magic_link("a@example.com", "http://x/login?token=SECRET") is False
|
|
assert "SECRET" not in caplog.text
|
|
assert "http://x/login" not in caplog.text
|
|
assert "a@example.com" in caplog.text
|