Updates to change/reset password functionality (#5294)

* feat: require old password when changing password

* update account settings page - require old password for changing passwords

* update profile dropdown - go to /account route for updating account details

* Remove all session based on user id after password change

* fix: run lint-fix

* remove unnecessary error page on account

* fix: prevent logout if user provides wrong current password

* fix: remove unused user profile page

* fix: import

---------

Co-authored-by: Henry <hzj94@hotmail.com>

packages/server/src/database/entities/index.ts

@@ -25,6 +25,7 @@ import { OrganizationUser } from '../../enterprise/database/entities/organizatio
 import { Workspace } from '../../enterprise/database/entities/workspace.entity'
 import { WorkspaceUser } from '../../enterprise/database/entities/workspace-user.entity'
 import { LoginMethod } from '../../enterprise/database/entities/login-method.entity'
+import { LoginSession } from '../../enterprise/database/entities/login-session.entity'
 
 export const entities = {
     ChatFlow,
@@ -55,5 +56,6 @@ export const entities = {
     OrganizationUser,
     Workspace,
     WorkspaceUser,
-    LoginMethod
+    LoginMethod,
+    LoginSession
 }

packages/server/src/enterprise/database/entities/login-session.entity.ts

@@ -0,0 +1,13 @@
+import { Column, Entity, PrimaryColumn } from 'typeorm'
+
+@Entity({ name: 'login_sessions' })
+export class LoginSession {
+    @PrimaryColumn({ type: 'varchar' })
+    sid: string
+
+    @Column({ type: 'text' })
+    sess: string
+
+    @Column({ type: 'bigint', nullable: true })
+    expire?: number
+}

packages/server/src/enterprise/middleware/passport/SessionPersistance.ts

@@ -3,9 +3,13 @@ import { RedisStore } from 'connect-redis'
 import { getDatabaseSSLFromEnv } from '../../../DataSource'
 import path from 'path'
 import { getUserHome } from '../../../utils'
+import type { Store } from 'express-session'
+import { LoginSession } from '../../database/entities/login-session.entity'
+import { getRunningExpressApp } from '../../../utils/getRunningExpressApp'
 
 let redisClient: Redis | null = null
 let redisStore: RedisStore | null = null
+let dbStore: Store | null = null
 
 export const initializeRedisClientAndStore = (): RedisStore => {
     if (!redisClient) {
@@ -35,6 +39,8 @@ export const initializeRedisClientAndStore = (): RedisStore => {
 }
 
 export const initializeDBClientAndStore: any = () => {
+    if (dbStore) return dbStore
+
     const databaseType = process.env.DATABASE_TYPE || 'sqlite'
     switch (databaseType) {
         case 'mysql': {
@@ -51,7 +57,8 @@ export const initializeDBClientAndStore: any = () => {
                     tableName: 'login_sessions'
                 }
             }
-            return new MySQLStore(options)
+            dbStore = new MySQLStore(options)
+            return dbStore
         }
         case 'mariadb':
             /* TODO: Implement MariaDB session store */
@@ -70,12 +77,13 @@ export const initializeDBClientAndStore: any = () => {
                 database: process.env.DATABASE_NAME,
                 ssl: getDatabaseSSLFromEnv()
             })
-            return new pgSession({
+            dbStore = new pgSession({
                 pool: pgPool, // Connection pool
                 tableName: 'login_sessions',
                 schemaName: 'public',
                 createTableIfMissing: true
             })
+            return dbStore
         }
         case 'default':
         case 'sqlite': {
@@ -83,11 +91,93 @@ export const initializeDBClientAndStore: any = () => {
             const sqlSession = require('connect-sqlite3')(expressSession)
             let flowisePath = path.join(getUserHome(), '.flowise')
             const homePath = process.env.DATABASE_PATH ?? flowisePath
-            return new sqlSession({
+            dbStore = new sqlSession({
                 db: 'database.sqlite',
                 table: 'login_sessions',
                 dir: homePath
             })
+            return dbStore
         }
     }
 }
+
+const getUserIdFromSession = (session: any): string | undefined => {
+    try {
+        const data = typeof session === 'string' ? JSON.parse(session) : session
+        return data?.passport?.user?.id
+    } catch {
+        return undefined
+    }
+}
+
+export const destroyAllSessionsForUser = async (userId: string): Promise<void> => {
+    try {
+        if (redisStore && redisClient) {
+            const prefix = (redisStore as any)?.prefix ?? 'sess:'
+            const pattern = `${prefix}*`
+            const keysToDelete: string[] = []
+            const batchSize = 1000
+
+            const stream = redisClient.scanStream({
+                match: pattern,
+                count: batchSize
+            })
+
+            for await (const keysBatch of stream) {
+                if (keysBatch.length === 0) continue
+
+                const sessions = await redisClient.mget(...keysBatch)
+                for (let i = 0; i < sessions.length; i++) {
+                    if (getUserIdFromSession(sessions[i]) === userId) {
+                        keysToDelete.push(keysBatch[i])
+                    }
+                }
+
+                if (keysToDelete.length >= batchSize) {
+                    const pipeline = redisClient.pipeline()
+                    keysToDelete.splice(0, batchSize).forEach((key) => pipeline.del(key))
+                    await pipeline.exec()
+                }
+            }
+
+            if (keysToDelete.length > 0) {
+                const pipeline = redisClient.pipeline()
+                keysToDelete.forEach((key) => pipeline.del(key))
+                await pipeline.exec()
+            }
+        } else if (dbStore) {
+            const appServer = getRunningExpressApp()
+            const dataSource = appServer.AppDataSource
+            const repository = dataSource.getRepository(LoginSession)
+
+            const databaseType = process.env.DATABASE_TYPE || 'sqlite'
+            switch (databaseType) {
+                case 'sqlite':
+                    await repository
+                        .createQueryBuilder()
+                        .delete()
+                        .where(`json_extract(sess, '$.passport.user.id') = :userId`, { userId })
+                        .execute()
+                    break
+                case 'mysql':
+                    await repository
+                        .createQueryBuilder()
+                        .delete()
+                        .where(`JSON_EXTRACT(sess, '$.passport.user.id') = :userId`, { userId })
+                        .execute()
+                    break
+                case 'postgres':
+                    await repository.createQueryBuilder().delete().where(`sess->'passport'->'user'->>'id' = :userId`, { userId }).execute()
+                    break
+                default:
+                    console.warn('Unsupported database type:', databaseType)
+                    break
+            }
+        } else {
+            console.warn('Session store not available, skipping session invalidation')
+        }
+    } catch (error) {
+        console.error('Error destroying sessions for user:', error)
+        throw error
+    }
+}

packages/server/src/enterprise/middleware/passport/index.ts

@@ -81,6 +81,11 @@ const _initializePassportMiddleware = async (app: express.Application) => {
     app.use(passport.initialize())
     app.use(passport.session())
 
+    if (options.store) {
+        const appServer = getRunningExpressApp()
+        appServer.sessionStore = options.store
+    }
+
     passport.serializeUser((user: any, done) => {
         done(null, user)
     })

packages/server/src/enterprise/services/account.service.ts

@@ -26,6 +26,7 @@ import { UserErrorMessage, UserService } from './user.service'
 import { WorkspaceUserErrorMessage, WorkspaceUserService } from './workspace-user.service'
 import { WorkspaceErrorMessage, WorkspaceService } from './workspace.service'
 import { sanitizeUser } from '../../utils/sanitize.util'
+import { destroyAllSessionsForUser } from '../middleware/passport/SessionPersistance'
 
 type AccountDTO = {
     user: Partial<User>
@@ -576,6 +577,9 @@ export class AccountService {
             await queryRunner.startTransaction()
             data.user = await this.userService.saveUser(data.user, queryRunner)
             await queryRunner.commitTransaction()
+
+            // Invalidate all sessions for this user after password reset
+            await destroyAllSessionsForUser(user.id as string)
         } catch (error) {
             await queryRunner.rollbackTransaction()
             throw error

packages/server/src/enterprise/services/user.service.ts

@@ -1,5 +1,4 @@
 import { StatusCodes } from 'http-status-codes'
-import bcrypt from 'bcryptjs'
 import { InternalFlowiseError } from '../../errors/internalFlowiseError'
 import { getRunningExpressApp } from '../../utils/getRunningExpressApp'
 import { Telemetry, TelemetryEventType } from '../../utils/telemetry'
@@ -8,8 +7,9 @@ import { isInvalidEmail, isInvalidName, isInvalidPassword, isInvalidUUID } from
 import { DataSource, ILike, QueryRunner } from 'typeorm'
 import { generateId } from '../../utils'
 import { GeneralErrorMessage } from '../../utils/constants'
-import { getHash } from '../utils/encryption.util'
+import { compareHash, getHash } from '../utils/encryption.util'
 import { sanitizeUser } from '../../utils/sanitize.util'
+import { destroyAllSessionsForUser } from '../middleware/passport/SessionPersistance'
 
 export const enum UserErrorMessage {
     EXPIRED_TEMP_TOKEN = 'Expired Temporary Token',
@@ -24,7 +24,8 @@ export const enum UserErrorMessage {
     USER_EMAIL_UNVERIFIED = 'User Email Unverified',
     USER_NOT_FOUND = 'User Not Found',
     USER_FOUND_MULTIPLE = 'User Found Multiple',
-    INCORRECT_USER_EMAIL_OR_CREDENTIALS = 'Incorrect Email or Password'
+    INCORRECT_USER_EMAIL_OR_CREDENTIALS = 'Incorrect Email or Password',
+    PASSWORDS_DO_NOT_MATCH = 'Passwords do not match'
 }
 export class UserService {
     private telemetry: Telemetry
@@ -134,7 +135,7 @@ export class UserService {
         return newUser
     }
 
-    public async updateUser(newUserData: Partial<User> & { password?: string }) {
+    public async updateUser(newUserData: Partial<User> & { oldPassword?: string; newPassword?: string; confirmPassword?: string }) {
         let queryRunner: QueryRunner | undefined
         let updatedUser: Partial<User>
         try {
@@ -158,10 +159,18 @@ export class UserService {
                 this.validateUserStatus(newUserData.status)
             }
 
-            if (newUserData.password) {
-                const salt = bcrypt.genSaltSync(parseInt(process.env.PASSWORD_SALT_HASH_ROUNDS || '5'))
-                // @ts-ignore
-                const hash = bcrypt.hashSync(newUserData.password, salt)
+            if (newUserData.oldPassword && newUserData.newPassword && newUserData.confirmPassword) {
+                if (!oldUserData.credential) {
+                    throw new InternalFlowiseError(StatusCodes.BAD_REQUEST, UserErrorMessage.INVALID_USER_CREDENTIAL)
+                }
+                // verify old password
+                if (!compareHash(newUserData.oldPassword, oldUserData.credential)) {
+                    throw new InternalFlowiseError(StatusCodes.BAD_REQUEST, UserErrorMessage.INVALID_USER_CREDENTIAL)
+                }
+                if (newUserData.newPassword !== newUserData.confirmPassword) {
+                    throw new InternalFlowiseError(StatusCodes.BAD_REQUEST, UserErrorMessage.PASSWORDS_DO_NOT_MATCH)
+                }
+                const hash = getHash(newUserData.newPassword)
                 newUserData.credential = hash
                 newUserData.tempToken = ''
                 newUserData.tokenExpiry = undefined
@@ -171,6 +180,11 @@ export class UserService {
             await queryRunner.startTransaction()
             await this.saveUser(updatedUser, queryRunner)
             await queryRunner.commitTransaction()
+
+            // Invalidate all sessions for this user if password was changed
+            if (newUserData.oldPassword && newUserData.newPassword && newUserData.confirmPassword) {
+                await destroyAllSessionsForUser(updatedUser.id as string)
+            }
         } catch (error) {
             if (queryRunner && queryRunner.isTransactionActive) await queryRunner.rollbackTransaction()
             throw error

packages/server/src/index.ts

@@ -73,6 +73,7 @@ export class App {
     queueManager: QueueManager
     redisSubscriber: RedisEventSubscriber
     usageCacheManager: UsageCacheManager
+    sessionStore: any
 
     constructor() {
         this.app = express()