Support account lockout on repeated incorrect password entries

Author: azvoleff

gefapi/__init__.py

@@ -857,6 +857,7 @@ def check_if_token_in_blocklist(jwt_header, jwt_payload):
     return is_token_in_blocklist(jti)
 
 
+from gefapi.errors import AccountLockedError  # noqa:E402
 from gefapi.models import User  # noqa:E402
 from gefapi.services import UserService  # noqa:E402
 
@@ -878,6 +879,9 @@ def create_token():
 
     try:
         user = UserService.authenticate_user(email, password)
+    except AccountLockedError as e:
+        logger.warning(f"[JWT]: Account locked for {email}")
+        return jsonify(e.serialize), 401
     except Exception as e:
         logger.error(f"[JWT]: Error during authentication: {str(e)}")
         return jsonify({"msg": "Authentication failed"}), 500

gefapi/errors.py

@@ -52,3 +52,26 @@ class EmailError(Error):
 
 class PasswordValidationError(Error):
     pass
+
+
+class AccountLockedError(Error):
+    """Raised when a user account is locked due to too many failed login attempts."""
+
+    def __init__(
+        self,
+        message: str,
+        minutes_remaining: int | None = None,
+        requires_password_reset: bool = False,
+    ):
+        super().__init__(message)
+        self.minutes_remaining = minutes_remaining
+        self.requires_password_reset = requires_password_reset
+
+    @property
+    def serialize(self):
+        return {
+            "message": self.message,
+            "error_code": "account_locked",
+            "minutes_remaining": self.minutes_remaining,
+            "requires_password_reset": self.requires_password_reset,
+        }

gefapi/models/user.py

@@ -88,6 +88,12 @@ class User(db.Model):
     email_verified = db.Column(db.Boolean(), default=False, nullable=True)
     email_verified_at = db.Column(db.DateTime(), nullable=True)
 
+    # Account lockout fields for brute force protection
+    # failed_login_count: Number of consecutive failed login attempts
+    # locked_until: When the account will be auto-unlocked (NULL = not locked)
+    failed_login_count = db.Column(db.Integer(), default=0, nullable=False)
+    locked_until = db.Column(db.DateTime(), nullable=True, index=True)
+
     def __init__(self, email, password, name, country, institution, role="USER"):
         self.email = email
         self.password = self.set_password(password)
@@ -102,6 +108,9 @@ def __init__(self, email, password, name, country, institution, role="USER"):
         self.last_activity_at = None
         self.email_verified = False
         self.email_verified_at = None
+        # Initialize account lockout fields
+        self.failed_login_count = 0
+        self.locked_until = None
 
     def __repr__(self):
         return f"<User {self.email!r}>"
@@ -204,6 +213,81 @@ def get_token(self):
         """Generate JWT token"""
         return create_access_token(identity=self.id)
 
+    # -------------------------------------------------------------------------
+    # Account Lockout Methods
+    # -------------------------------------------------------------------------
+    # Lockout thresholds and durations
+    LOCKOUT_THRESHOLDS = [
+        (5, 15),  # After 5 failures: lock for 15 minutes
+        (10, 60),  # After 10 failures: lock for 60 minutes
+        (20, None),  # After 20 failures: lock until password reset
+    ]
+
+    def is_locked(self) -> bool:
+        """Check if the account is currently locked.
+
+        Returns:
+            True if account is locked, False otherwise
+        """
+        if self.locked_until is None:
+            return False
+        now = datetime.datetime.utcnow()
+        # Return True if lock hasn't expired yet
+        return self.locked_until > now
+
+    def get_lockout_minutes_remaining(self) -> int | None:
+        """Get minutes remaining until account unlocks.
+
+        Returns:
+            Minutes remaining, or None if not locked or permanently locked
+        """
+        if not self.is_locked():
+            return 0
+        if self.locked_until is None:
+            return None  # Shouldn't happen, but be safe
+        now = datetime.datetime.utcnow()
+        remaining = self.locked_until - now
+        return max(1, int(remaining.total_seconds() / 60))
+
+    def record_failed_login(self) -> tuple[bool, int | None]:
+        """Record a failed login attempt and apply lockout if needed.
+
+        Returns:
+            Tuple of (is_now_locked, lockout_minutes or None for permanent)
+        """
+        self.failed_login_count = (self.failed_login_count or 0) + 1
+        count = self.failed_login_count
+
+        # Determine lockout duration based on failure count
+        lockout_minutes = None
+        for threshold, minutes in self.LOCKOUT_THRESHOLDS:
+            if count >= threshold:
+                lockout_minutes = minutes
+
+        if lockout_minutes is not None:
+            self.locked_until = datetime.datetime.utcnow() + datetime.timedelta(
+                minutes=lockout_minutes
+            )
+            return True, lockout_minutes
+
+        if count >= self.LOCKOUT_THRESHOLDS[-1][0]:
+            # Permanent lock (until password reset)
+            # Set to far future date
+            self.locked_until = datetime.datetime.utcnow() + datetime.timedelta(
+                days=365 * 100
+            )
+            return True, None
+
+        return False, None
+
+    def clear_failed_logins(self) -> None:
+        """Clear failed login counter and unlock account.
+
+        Called on successful login or password reset.
+        """
+        self.failed_login_count = 0
+        self.locked_until = None
+
     @staticmethod
     @lru_cache(maxsize=1)
     def _get_encryption_key() -> bytes:

gefapi/services/user_service.py

@@ -490,6 +490,8 @@ def change_password(user, old_password, new_password):
             )
         _validate_password_strength(new_password)
         user.password = user.set_password(new_password)
+        # Clear any account lockout on password change
+        user.clear_failed_logins()
         try:
             db.session.add(user)
             db.session.commit()
@@ -529,6 +531,8 @@ def admin_change_password(user, new_password):
         logger.info(f"[SERVICE]: Admin changing password for user {user.email}")
         _validate_password_strength(new_password)
         user.password = user.set_password(new_password)
+        # Clear any account lockout on admin password change
+        user.clear_failed_logins()
         try:
             db.session.add(user)
             db.session.commit()
@@ -619,6 +623,8 @@ def _recover_password_legacy(user):
         """
         password = _generate_secure_password()
         user.password = user.set_password(password=password)
+        # Clear any account lockout on password recovery
+        user.clear_failed_logins()
         try:
             logger.info("[DB]: ADD")
             db.session.add(user)
@@ -744,6 +750,9 @@ def reset_password_with_token(token_string, new_password):
             # Set new password
             user.password = user.set_password(password=new_password)
 
+            # Clear any account lockout - password reset unlocks the account
+            user.clear_failed_logins()
+
             # Mark token as used
             reset_token.mark_used()
 
@@ -967,6 +976,17 @@ def delete_user(
 
     @staticmethod
     def authenticate_user(email, password):
+        """Authenticate a user by email and password.
+
+        Returns:
+            User object on successful authentication, None if user not found
+            or password is invalid.
+
+        Raises:
+            AccountLockedError: If the account is locked due to failed attempts.
+        """
+        from gefapi.errors import AccountLockedError
+
         logger.info(f"[AUTH]: Authentication attempt for {email}")
         user = User.query.filter_by(email=email).first()
 
@@ -975,14 +995,98 @@ def authenticate_user(email, password):
             log_authentication_event(False, email, "user_not_found")
             return None
 
+        # Check if account is locked
+        if user.is_locked():
+            minutes_remaining = user.get_lockout_minutes_remaining()
+            if minutes_remaining is None:
+                logger.warning(f"[AUTH]: Account locked until password reset: {email}")
+                log_authentication_event(False, email, "account_locked_permanent")
+                raise AccountLockedError(
+                    "Your account is locked due to too many failed login attempts. "
+                    "Please reset your password to regain access.",
+                    minutes_remaining=None,
+                    requires_password_reset=True,
+                )
+            logger.warning(
+                f"[AUTH]: Account temporarily locked for {email}, "
+                f"{minutes_remaining} minutes remaining"
+            )
+            log_authentication_event(
+                False, email, f"account_locked_{minutes_remaining}m"
+            )
+            raise AccountLockedError(
+                f"Your account is temporarily locked. "
+                f"Please try again in {minutes_remaining} minute(s).",
+                minutes_remaining=minutes_remaining,
+                requires_password_reset=False,
+            )
+
         if not user.check_password(password):
             logger.warning(f"[AUTH]: Failed login - invalid password: {email}")
-            log_authentication_event(False, email, "invalid_password")
+
+            # Record failed attempt and apply lockout if threshold reached
+            try:
+                is_locked, lockout_minutes = user.record_failed_login()
+                db.session.add(user)
+                db.session.commit()
+
+                if is_locked:
+                    if lockout_minutes is None:
+                        logger.warning(
+                            f"[AUTH]: Account locked until password reset after "
+                            f"{user.failed_login_count} failures: {email}"
+                        )
+                        log_authentication_event(
+                            False, email, "account_locked_permanent"
+                        )
+                        # Log security event for account lockout
+                        from gefapi.utils.security_events import log_security_event
+
+                        log_security_event(
+                            "ACCOUNT_LOCKED",
+                            user_email=email,
+                            details={
+                                "reason": "max_failed_attempts",
+                                "failed_count": user.failed_login_count,
+                                "unlock_method": "password_reset_required",
+                            },
+                            level="warning",
+                        )
+                        raise AccountLockedError(
+                            "Your account has been locked due to too many failed "
+                            "login attempts. Please reset your password to unlock.",
+                            minutes_remaining=None,
+                            requires_password_reset=True,
+                        )
+                    logger.warning(
+                        f"[AUTH]: Account locked for {lockout_minutes} minutes "
+                        f"after {user.failed_login_count} failures: {email}"
+                    )
+                    log_authentication_event(
+                        False, email, f"account_locked_{lockout_minutes}m"
+                    )
+                    raise AccountLockedError(
+                        f"Your account has been temporarily locked after "
+                        f"{user.failed_login_count} failed login attempts. "
+                        f"Please try again in {lockout_minutes} minute(s).",
+                        minutes_remaining=lockout_minutes,
+                        requires_password_reset=False,
+                    )
+                log_authentication_event(False, email, "invalid_password")
+            except AccountLockedError:
+                # Re-raise AccountLockedError so it propagates to the route handler
+                raise
+            except Exception as e:
+                logger.warning(f"[AUTH]: Failed to record failed login: {e}")
+                db.session.rollback()
+                log_authentication_event(False, email, "invalid_password")
+
             return None
 
-        # Successful authentication - update login and activity timestamps
+        # Successful authentication - clear failed login count and update timestamps
         try:
             now = datetime.datetime.utcnow()
+            user.clear_failed_logins()  # Reset lockout state
             user.last_login_at = now
             user.last_activity_at = now
             db.session.add(user)

migrations/versions/e3f4a5b6c7d8_add_account_lockout_fields.py

@@ -0,0 +1,61 @@
+"""add account lockout fields to user
+
+Revision ID: e3f4a5b6c7d8
+Revises: d2e3f4a5b6c7
+Create Date: 2026-02-05 10:00:00.000000
+
+This migration adds fields to support account lockout after failed login attempts:
+- failed_login_count: Number of consecutive failed login attempts
+- locked_until: When the account will be automatically unlocked (NULL = not locked)
+
+SECURITY BENEFITS:
+1. Prevents slow brute force attacks that bypass rate limiting
+2. Forces users with wrong credentials to reset password
+3. Reduces Rollbar noise from repeated failed logins
+
+LOCKOUT POLICY:
+- After 5 failed attempts: Lock for 15 minutes
+- After 10 failed attempts: Lock for 1 hour
+- After 20 failed attempts: Lock until password reset
+- Successful login or password reset clears the counter
+
+EXISTING USER HANDLING:
+All existing users start with failed_login_count = 0 and locked_until = NULL
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision = "e3f4a5b6c7d8"
+down_revision = "d2e3f4a5b6c7"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    with op.batch_alter_table("user", schema=None) as batch_op:
+        batch_op.add_column(
+            sa.Column(
+                "failed_login_count",
+                sa.Integer(),
+                nullable=False,
+                server_default="0",
+            )
+        )
+        batch_op.add_column(
+            sa.Column("locked_until", sa.DateTime(), nullable=True)
+        )
+        # Index for efficient queries on locked accounts
+        batch_op.create_index(
+            "ix_user_locked_until",
+            ["locked_until"],
+            unique=False,
+        )
+
+
+def downgrade():
+    with op.batch_alter_table("user", schema=None) as batch_op:
+        batch_op.drop_index("ix_user_locked_until")
+        batch_op.drop_column("locked_until")
+        batch_op.drop_column("failed_login_count")