fix(auth): don't fail the lockout cleanup when a login clears the entry first (#1658)

cleanup_old_lockouts collects the expired usernames, logs each one and
then deletes it. Under gevent the log write can yield, and a successful
login for the same user in that gap clears the entry first, so the del
raises KeyError in whatever request ran the cleanup. Both removals now
use pop(username, None), and the new test reproduces the KeyError on
main.

While in there I dropped the if db_user check in the builtin login path.
It is always true after the not db_user return just above it, and the
failed response after it could never be reached. The diff looks bigger
than it is because the block moves out one indent.


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

<!-- coderabbit-ai-generated:v1 -->

## Summary by CodeRabbit

* **Bug Fixes**
* Login failures now follow a consistent response path when an account
is unavailable or its password data is invalid.
* Expired login lockouts can be cleaned up safely even if they are
cleared at the same time.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

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}")

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