feat: Force users to change password when using default password (#1282)

Add password change enforcement for users using default passwords or when
password_change_required flag is set:

- Add password_change_required field to EmailUser model with migration
- Check if user is using default password on login and force change
- Add /admin/change-password-required page with form for password change
- Add admin endpoint to force password change for specific users
- Set password_change_required=true for bootstrap admin user
- Return 403 with X-Password-Change-Required header for API login
- Add Force Password Change button to admin user management UI

Closes #1282

Signed-off-by: Mihai Criveti <[email protected]>

Author: crivetimihai

mcpgateway/admin.py

@@ -0,0 +1,27 @@
+"""Add password_change_required field to EmailUser
+
+Revision ID: z1a2b3c4d5e6
+Revises: 191a2def08d7
+Create Date: 2025-11-21 14:16:30.000000
+
+"""
+
+# Third-Party
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision = "z1a2b3c4d5e6"
+down_revision = "191a2def08d7"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    """Add password_change_required field to email_users table."""
+    op.add_column("email_users", sa.Column("password_change_required", sa.Boolean(), nullable=False, server_default="false"))
+
+
+def downgrade():
+    """Remove password_change_required field from email_users table."""
+    op.drop_column("email_users", "password_change_required")

mcpgateway/alembic/versions/z1a2b3c4d5e6_add_password_change_required.py

@@ -83,11 +83,12 @@ async def bootstrap_admin_user() -> None:
                 is_admin=True,
             )
 
-            # Mark admin user as email verified
+            # Mark admin user as email verified and require password change on first login
             # First-Party
             from mcpgateway.db import utc_now  # pylint: disable=import-outside-toplevel
 
             admin_user.email_verified_at = utc_now()
+            admin_user.password_change_required = True  # Force admin to change default password
             db.commit()
 
             # Personal team is automatically created during user creation if enabled

mcpgateway/bootstrap_db.py

@@ -290,6 +290,7 @@ class Settings(BaseSettings):
     email_auth_enabled: bool = Field(default=True, description="Enable email-based authentication")
     platform_admin_email: str = Field(default="[email protected]", description="Platform administrator email address")
     platform_admin_password: SecretStr = Field(default=SecretStr("changeme"), description="Platform administrator password")
+    default_user_password: SecretStr = Field(default=SecretStr("changeme"), description="Default password for new users")  # nosec B105
     platform_admin_full_name: str = Field(default="Platform Administrator", description="Platform administrator full name")
 
     # Argon2id Password Hashing Configuration

mcpgateway/config.py

@@ -482,6 +482,7 @@ class EmailUser(Base):
     password_hash_type: Mapped[str] = mapped_column(String(20), default="argon2id", nullable=False)
     failed_login_attempts: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
     locked_until: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True)
+    password_change_required: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
 
     # Timestamps
     created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utc_now, nullable=False)

mcpgateway/db.py

@@ -225,6 +225,28 @@ async def login(login_request: EmailLoginRequest, request: Request, db: Session
         if not user:
             raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid email or password")
 
+        # Check if password change is required OR if user is using default password
+        needs_password_change = user.password_change_required
+
+        # Also check if user is using the default password
+        if not needs_password_change:
+            # First-Party
+            from mcpgateway.services.argon2_service import Argon2PasswordService
+
+            password_service = Argon2PasswordService()
+            is_using_default_password = password_service.verify_password(settings.default_user_password.get_secret_value(), user.password_hash)  # nosec B105
+            if is_using_default_password:
+                needs_password_change = True
+                # Set the flag in database for future reference
+                user.password_change_required = True
+                db.commit()
+
+        if needs_password_change:
+            # For API login, return a specific error indicating password change is required
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN, detail="Password change required. Please change your password before continuing.", headers={"X-Password-Change-Required": "true"}
+            )
+
         # Create access token
         access_token, expires_in = await create_access_token(user)
 
@@ -502,6 +524,11 @@ async def create_user(user_request: EmailRegistrationRequest, current_user: Emai
             auth_provider="local",
         )
 
+        # If the user was created with the default password, force password change
+        if user_request.password == settings.default_user_password.get_secret_value():  # nosec B105
+            user.password_change_required = True
+            db.commit()
+
         logger.info(f"Admin {current_user.email} created user: {user.email}")
 
         return EmailUserResponse.from_email_user(user)
@@ -578,13 +605,19 @@ async def update_user(user_email: str, user_request: EmailRegistrationRequest, c
 
         # Update password if provided
         if user_request.password:
-            await auth_service.change_password(
-                email=user_email,
-                old_password=None,  # Admin can change without old password
-                new_password=user_request.password,
-                ip_address="admin_update",
-                user_agent="admin_panel",
-            )
+            # For admin updates, we need to directly update the password hash
+            # since we don't have the old password to verify
+            # First-Party
+            from mcpgateway.services.argon2_service import Argon2PasswordService
+
+            password_service = Argon2PasswordService()
+
+            # Validate the new password meets requirements
+            auth_service.validate_password(user_request.password)
+
+            # Update password hash directly
+            user.password_hash = password_service.hash_password(user_request.password)
+            user.password_change_required = False  # Clear password change requirement
 
         db.commit()
         db.refresh(user)

mcpgateway/routers/email_auth.py

@@ -4865,6 +4865,7 @@ class EmailUserResponse(BaseModel):
     created_at: datetime = Field(..., description="Account creation timestamp")
     last_login: Optional[datetime] = Field(None, description="Last successful login")
     email_verified: bool = Field(False, description="Whether email is verified")
+    password_change_required: bool = Field(False, description="Whether user must change password on next login")
 
     @classmethod
     def from_email_user(cls, user) -> "EmailUserResponse":
@@ -4885,6 +4886,7 @@ def from_email_user(cls, user) -> "EmailUserResponse":
             created_at=user.created_at,
             last_login=user.last_login,
             email_verified=user.is_email_verified(),
+            password_change_required=user.password_change_required,
         )