57 lines
2.3 KiB
Python
57 lines
2.3 KiB
Python
"""Rate-Limit je Key und Fehlversuche je IP (→ KONZEPT.md 16.3)."""
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from hugin_app.app import create_app
|
|
from hugin_app.rate_limit import Window
|
|
from hugin_app.settings import Settings
|
|
from hugin_app.hugin import Hugin
|
|
from hugin_app.migration import migrate
|
|
from test_app import Env, auth
|
|
|
|
|
|
@pytest.fixture
|
|
def brief(database_url):
|
|
migrate(database_url)
|
|
hugin = Hugin(Settings(database_url=database_url, admin_token="admin-geheim", rate_limit=1000,
|
|
failed_attempts=3))
|
|
u = Env(TestClient(create_app(hugin)), hugin)
|
|
return u
|
|
|
|
|
|
def test_window():
|
|
f = Window(2, 60)
|
|
assert f.allow("a") is None and f.allow("a") is None
|
|
assert f.allow("a") >= 1 and f.allow("b") is None
|
|
|
|
|
|
def test_requests_per_key(brief):
|
|
brief.hugin.rate_limit.requests.limit = 5
|
|
c = brief.client
|
|
codes = [c.get("/api/v1/packages", headers=brief.h("shop-lesen")).status_code for _ in range(7)]
|
|
assert codes[:5] == [200] * 5 and codes[5:] == [429, 429]
|
|
response = c.get("/api/v1/packages", headers=brief.h("shop-lesen"))
|
|
assert int(response.headers["retry-after"]) >= 1
|
|
# Andere Keys und die Health-Probes sind nicht betroffen
|
|
assert c.get("/api/v1/packages", headers=brief.h("vertrieb-lesen")).status_code == 200
|
|
assert c.get("/healthz").status_code == 200
|
|
|
|
|
|
def test_failed_attempts_lock_the_ip(brief):
|
|
c = brief.client
|
|
for _ in range(3):
|
|
assert c.get("/api/v1/packages", headers=auth("hk_FALSCH")).status_code == 401
|
|
# Jetzt kommt diese Adresse auch mit einem gültigen Key nicht weiter
|
|
assert c.get("/api/v1/packages", headers=brief.h("shop-lesen")).status_code == 429
|
|
assert c.post("/login", data={"key": brief.keys["shop-lesen"]}).status_code == 429
|
|
assert c.get("/healthz").status_code == 200
|
|
|
|
|
|
def test_failed_attempts_at_login_and_mcp(brief):
|
|
c = TestClient(brief.client.app, follow_redirects=False)
|
|
for _ in range(2):
|
|
assert c.post("/login", data={"key": "wrong"}).status_code == 401
|
|
assert c.post("/mcp", headers=auth("hk_FALSCH") | {"Accept": "application/json"},
|
|
json={"jsonrpc": "2.0", "id": 1, "method": "tools/list"}).status_code == 401
|
|
assert c.post("/login", data={"key": brief.keys["shop-lesen"]}).status_code == 429
|