PTah
d3a337992c
Path traversal fix in SPA fallback, admin-only host delete, login rate limit with 3 attempts and Telegram alert, JWT validated against active users in DB, and DOMPurify for agent report HTML. Co-authored-by: Cursor <cursoragent@cursor.com>
66 lines
1.9 KiB
Python
66 lines
1.9 KiB
Python
"""SAC UI login rate limit and Telegram alert on brute-force."""
|
|
|
|
|
|
def _failed_login(client, username: str = "test-admin") -> int:
|
|
return client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": username, "password": "wrong-password"},
|
|
).status_code
|
|
|
|
|
|
def test_login_blocked_after_max_failures(client, monkeypatch):
|
|
monkeypatch.setenv("SAC_LOGIN_MAX_FAILURES", "3")
|
|
from app.config import get_settings
|
|
|
|
get_settings.cache_clear()
|
|
|
|
assert _failed_login(client) == 401
|
|
assert _failed_login(client) == 401
|
|
assert _failed_login(client) == 401
|
|
assert _failed_login(client) == 429
|
|
|
|
ok = client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "test-admin", "password": "test-admin-password"},
|
|
)
|
|
assert ok.status_code == 429
|
|
|
|
get_settings.cache_clear()
|
|
|
|
|
|
def test_login_telegram_alert_on_threshold(client, monkeypatch):
|
|
monkeypatch.setenv("SAC_LOGIN_MAX_FAILURES", "3")
|
|
monkeypatch.setenv("SAC_LOGIN_ALERT_TELEGRAM", "true")
|
|
from app.config import get_settings
|
|
|
|
get_settings.cache_clear()
|
|
|
|
sent: list[str] = []
|
|
monkeypatch.setattr(
|
|
"app.services.login_rate_limit.send_telegram_text",
|
|
lambda text, **kwargs: sent.append(text),
|
|
)
|
|
|
|
for _ in range(3):
|
|
assert _failed_login(client, username="attacker") == 401
|
|
|
|
assert len(sent) == 1
|
|
assert "SAC: подбор пароля UI" in sent[0]
|
|
assert "attacker" in sent[0]
|
|
|
|
assert _failed_login(client, username="attacker") == 429
|
|
assert len(sent) == 1
|
|
|
|
get_settings.cache_clear()
|
|
|
|
|
|
def test_deactivated_user_token_rejected_on_me(client, jwt_monitor_headers, db_session):
|
|
from app.models.user import User
|
|
|
|
user = db_session.query(User).filter(User.username == "test-monitor").one()
|
|
user.is_active = False
|
|
db_session.commit()
|
|
|
|
response = client.get("/api/v1/auth/me", headers=jwt_monitor_headers)
|
|
assert response.status_code == 401
|