security: rate limit admin authentication

This commit is contained in:
ik
2026-09-12 21:33:26 +07:00
parent ff8ea40d40
commit 7c28a0c36a
9 changed files with 158 additions and 13 deletions
+44 -1
View File
@@ -9,8 +9,51 @@ from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session
from app.database import Base
from app.admin_security import verify_admin
from app.main import _check_rate_limit, _is_trusted_proxy
from app.models import SubmissionAttempt
from app.models import AdminAuthAttempt, SubmissionAttempt
def _admin_config() -> MagicMock:
config = MagicMock()
config.admin_token = "correct-token"
config.admin_auth_attempt_limit = 3
config.admin_auth_window_seconds = 600
config.rate_limit_secret = "test-secret-for-testing"
config.trusted_proxy_cidrs = ["127.0.0.1/32"]
return config
def test_admin_auth_failures_are_hashed_and_limited() -> None:
engine = create_engine("sqlite://")
Base.metadata.create_all(engine)
request = MagicMock()
request.client.host = "203.0.113.42"
request.headers.get.return_value = None
with Session(engine) as db:
for _ in range(3):
with pytest.raises(HTTPException) as denied:
verify_admin(request, db, "Bearer wrong", _admin_config())
assert denied.value.status_code == 401
with pytest.raises(HTTPException) as limited:
verify_admin(request, db, "Bearer correct-token", _admin_config())
assert limited.value.status_code == 429
attempts = list(db.scalars(select(AdminAuthAttempt)))
assert len(attempts) == 3
assert all(item.client_hash != "203.0.113.42" and len(item.client_hash) == 64 for item in attempts)
def test_successful_admin_auth_clears_previous_failures() -> None:
engine = create_engine("sqlite://")
Base.metadata.create_all(engine)
request = MagicMock()
request.client.host = "203.0.113.43"
request.headers.get.return_value = None
with Session(engine) as db:
with pytest.raises(HTTPException):
verify_admin(request, db, "Bearer wrong", _admin_config())
assert verify_admin(request, db, "Bearer correct-token", _admin_config()) == "admin"
assert list(db.scalars(select(AdminAuthAttempt))) == []
def test_rate_limit_is_persistent_and_does_not_store_raw_client() -> None: