diff --git a/shelfmark/core/auth_routes.py b/shelfmark/core/auth_routes.py index c313a825..e226524b 100644 --- a/shelfmark/core/auth_routes.py +++ b/shelfmark/core/auth_routes.py @@ -46,7 +46,8 @@ def cleanup_old_lockouts() -> None: expired_users.append(username) for username in expired_users: logger.info("Lockout expired for user: %s", username) - del failed_login_attempts[username] + # A login can clear the entry while this runs, so it may be gone already. + failed_login_attempts.pop(username, None) def _get_lockout_until(username: str, *, repair_if_locked: bool = False) -> datetime | None: @@ -123,8 +124,7 @@ def record_failed_login(username: str, ip_address: str) -> bool: def clear_failed_logins(username: str) -> None: """Clear failed login attempts for a user after successful login.""" - if username in failed_login_attempts: - del failed_login_attempts[username] + if failed_login_attempts.pop(username, None) is not None: logger.debug("Cleared failed login attempts for user: %s", username) @@ -255,29 +255,26 @@ def register_auth_routes( return _failed_login_response(username, ip_address) # Authenticate against DB user - if db_user: - if not db_user.get("password_hash") or not check_password( - db_user["password_hash"], password - ): - return _failed_login_response(username, ip_address) + if not db_user.get("password_hash") or not check_password( + db_user["password_hash"], password + ): + return _failed_login_response(username, ip_address) - is_admin = db_user["role"] == "admin" - session["user_id"] = username - session["db_user_id"] = db_user["id"] - session["is_admin"] = is_admin - session.permanent = remember_me - clear_failed_logins(username) - logger.info( - "Login successful for user '%s' from IP %s (%s auth, is_admin=%s, remember_me=%s)", - username, - ip_address, - auth_mode, - is_admin, - remember_me, - ) - return jsonify({"success": True}) - - return _failed_login_response(username, ip_address) + is_admin = db_user["role"] == "admin" + session["user_id"] = username + session["db_user_id"] = db_user["id"] + session["is_admin"] = is_admin + session.permanent = remember_me + clear_failed_logins(username) + logger.info( + "Login successful for user '%s' from IP %s (%s auth, is_admin=%s, remember_me=%s)", + username, + ip_address, + auth_mode, + is_admin, + remember_me, + ) + return jsonify({"success": True}) except operational_errors as e: logger.error_trace(f"Built-in auth error: {e}") diff --git a/tests/core/test_auth_routes.py b/tests/core/test_auth_routes.py index 27612a6d..e3ef2779 100644 --- a/tests/core/test_auth_routes.py +++ b/tests/core/test_auth_routes.py @@ -672,3 +672,21 @@ class TestAuthCheck: def test_only_get_is_allowed(self, main_module): rules = [r for r in main_module.app.url_map.iter_rules() if r.rule == "/api/auth/check"] assert [sorted(r.methods - {"OPTIONS"}) for r in rules] == [["GET", "HEAD"]] + + +class TestLockoutCleanupRace: + def test_cleanup_survives_an_entry_cleared_while_it_runs(self, main_module, client): + # A successful login can clear an expired entry between the cleanup collecting it and + # removing it (the log write in between is a yield point under gevent). + from shelfmark.core import auth_routes + + expired = datetime.now(UTC) - timedelta(minutes=1) + main_module.failed_login_attempts["alice"] = {"count": 10, "lockout_until": expired} + + def cleared_meanwhile(*_args, **_kwargs): + auth_routes.clear_failed_logins("alice") + + with patch.object(auth_routes.logger, "info", side_effect=cleared_meanwhile): + auth_routes.cleanup_old_lockouts() + + assert "alice" not in main_module.failed_login_attempts