Skip to content

Commit c4e689b

Browse files
yau-wdyau-wd
authored
Feature/Add Audit and Deletion (#5994)
* feat(audit): add emitEvent and file-based audit logging * fix(sanitize.util.ts): correct IPv6 sanitization for audit log IP masking * feat(account): add account deletion * fix(sanitize.util.ts): add recursive sanitization for nested metadata * fix(MainLayout/Header/Workspace): refresh workspace list when clicking dropdown * feat(views/account): require typed confirmation for account deletion
1 parent 1707688 commit c4e689b

28 files changed

Lines changed: 2156 additions & 27 deletions

‎packages/components/src/storage/AzureBlobStorageProvider.ts‎

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,9 +2,9 @@ import { BlobServiceClient, ContainerClient, StorageSharedKeyCredential } from '
22
import multer from 'multer'
33
import { MulterAzureStorage } from 'multer-azure-blob-storage'
44
import { v4 as uuidv4 } from 'uuid'
5+
import { winstonAzureBlob } from 'winston-azure-blob'
56
import { BaseStorageProvider } from './BaseStorageProvider'
67
import { FileInfo, StorageResult, StorageSizeResult } from './IStorageProvider'
7-
import { winstonAzureBlob } from 'winston-azure-blob'
88

99
/**
1010
* Extends MulterAzureStorage to set file.path from file.blobName after upload.
@@ -281,7 +281,7 @@ export class AzureBlobStorageProvider extends BaseStorageProvider {
281281
return multer({ storage: azureStorage })
282282
}
283283

284-
getLoggerTransports(logType: 'server' | 'error' | 'requests'): any[] {
284+
getLoggerTransports(logType: 'server' | 'error' | 'requests' | 'audit'): any[] {
285285
const connectionString = process.env.AZURE_BLOB_STORAGE_CONNECTION_STRING
286286
const accountName = process.env.AZURE_BLOB_STORAGE_ACCOUNT_NAME
287287
const accountKey = process.env.AZURE_BLOB_STORAGE_ACCOUNT_KEY
@@ -325,7 +325,18 @@ export class AzureBlobStorageProvider extends BaseStorageProvider {
325325
level: 'debug'
326326
})
327327
]
328+
} else if (logType === 'audit') {
329+
return [
330+
winstonAzureBlob({
331+
...baseConfig,
332+
blobName: 'logs/audit/audit',
333+
rotatePeriod: 'YYYY-MM-DD',
334+
extension: '.log.jsonl',
335+
level: 'info'
336+
})
337+
]
328338
}
339+
329340
return []
330341
}
331342
}

‎packages/components/src/storage/BaseStorageProvider.ts‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,9 @@
1+
import fs from 'node:fs'
12
import path from 'node:path'
2-
import { IStorageProvider, FileInfo, StorageResult, StorageSizeResult } from './IStorageProvider'
33
import sanitize from 'sanitize-filename'
44
import { getUserHome } from '../utils'
55
import { isPathTraversal, isUnsafeFilePath, isValidUUID } from '../validator'
6-
import fs from 'node:fs'
6+
import { FileInfo, IStorageProvider, StorageResult, StorageSizeResult } from './IStorageProvider'
77

88
export abstract class BaseStorageProvider implements IStorageProvider {
99
protected storagePath: string
@@ -38,7 +38,7 @@ export abstract class BaseStorageProvider implements IStorageProvider {
3838
abstract removeFolderFromStorage(...paths: string[]): Promise<StorageSizeResult>
3939
abstract getStorageSize(orgId: string): Promise<number>
4040
abstract getMulterStorage(): any
41-
abstract getLoggerTransports(logType: 'server' | 'error' | 'requests', config?: any): any[]
41+
abstract getLoggerTransports(logType: 'server' | 'error' | 'requests' | 'audit', config?: any): any[]
4242

4343
/**
4444
* Shared utility for sanitizing filenames to prevent path traversal and other issues

‎packages/components/src/storage/GCSStorageProvider.ts‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
1-
import { Storage, Bucket } from '@google-cloud/storage'
21
import { LoggingWinston } from '@google-cloud/logging-winston'
2+
import { Bucket, Storage } from '@google-cloud/storage'
33
import multer from 'multer'
44
import { v4 as uuidv4 } from 'uuid'
55
import { BaseStorageProvider } from './BaseStorageProvider'
@@ -336,7 +336,7 @@ export class GCSStorageProvider extends BaseStorageProvider {
336336
})
337337
}
338338

339-
getLoggerTransports(logType: 'server' | 'error' | 'requests'): any[] {
339+
getLoggerTransports(logType: 'server' | 'error' | 'requests' | 'audit'): any[] {
340340
const gcsConfig = {
341341
projectId: this.projectId,
342342
keyFilename: this.keyFilename,
@@ -368,7 +368,15 @@ export class GCSStorageProvider extends BaseStorageProvider {
368368
logName: 'requests'
369369
})
370370
]
371+
} else if (logType === 'audit') {
372+
return [
373+
new LoggingWinston({
374+
...gcsConfig,
375+
logName: 'audit'
376+
})
377+
]
371378
}
379+
372380
return []
373381
}
374382
}

‎packages/components/src/storage/IStorageProvider.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -94,5 +94,5 @@ export interface IStorageProvider {
9494
/**
9595
* Get the Winston logger transports for this provider
9696
*/
97-
getLoggerTransports(logType: 'server' | 'error' | 'requests', config?: any): any[]
97+
getLoggerTransports(logType: 'server' | 'error' | 'requests' | 'audit', config?: any): any[]
9898
}

‎packages/components/src/storage/LocalStorageProvider.ts‎

Lines changed: 12 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,8 @@
1-
import path from 'node:path'
21
import fs from 'fs'
32
import multer from 'multer'
4-
import DailyRotateFile from 'winston-daily-rotate-file'
3+
import path from 'node:path'
54
import { transports } from 'winston'
5+
import DailyRotateFile from 'winston-daily-rotate-file'
66
import { BaseStorageProvider } from './BaseStorageProvider'
77
import { FileInfo, StorageResult, StorageSizeResult } from './IStorageProvider'
88

@@ -350,7 +350,7 @@ export class LocalStorageProvider extends BaseStorageProvider {
350350
return process.env.HOME || process.env.USERPROFILE || process.env.HOMEPATH || ''
351351
}
352352

353-
getLoggerTransports(logType: 'server' | 'error' | 'requests', config?: any): any[] {
353+
getLoggerTransports(logType: 'server' | 'error' | 'requests' | 'audit', config?: any): any[] {
354354
const logDir = config?.logging?.dir || path.join(this.getUserHome(), '.flowise', 'logs')
355355

356356
if (!fs.existsSync(logDir)) {
@@ -373,6 +373,15 @@ export class LocalStorageProvider extends BaseStorageProvider {
373373
level: config?.logging?.express?.level ?? 'debug'
374374
})
375375
]
376+
} else if (logType === 'audit') {
377+
return [
378+
new DailyRotateFile({
379+
filename: path.join(logDir, 'audit-%DATE%.log.jsonl'),
380+
datePattern: 'YYYY-MM-DD-HH',
381+
maxSize: '20m',
382+
level: 'info'
383+
})
384+
]
376385
}
377386

378387
// For 'error' type, return empty array (handled by exceptionHandlers in logger.ts)

‎packages/components/src/storage/S3StorageProvider.ts‎

Lines changed: 13 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -7,11 +7,11 @@ import {
77
S3Client,
88
S3ClientConfig
99
} from '@aws-sdk/client-s3'
10-
import { Readable } from 'node:stream'
1110
import multer from 'multer'
1211
import multerS3 from 'multer-s3'
13-
import { transports } from 'winston'
12+
import { Readable } from 'node:stream'
1413
import { v4 as uuidv4 } from 'uuid'
14+
import { transports } from 'winston'
1515
import { BaseStorageProvider } from './BaseStorageProvider'
1616
import { FileInfo, StorageResult, StorageSizeResult } from './IStorageProvider'
1717

@@ -507,7 +507,7 @@ export class S3StorageProvider extends BaseStorageProvider {
507507
})
508508
}
509509

510-
getLoggerTransports(logType: 'server' | 'error' | 'requests'): any[] {
510+
getLoggerTransports(logType: 'server' | 'error' | 'requests' | 'audit'): any[] {
511511
if (logType === 'server') {
512512
const s3ServerStream = new S3StreamLogger({
513513
bucket: this.bucket,
@@ -532,7 +532,17 @@ export class S3StorageProvider extends BaseStorageProvider {
532532
config: this.s3Config
533533
})
534534
return [new transports.Stream({ stream: s3ServerReqStream })]
535+
} else if (logType === 'audit') {
536+
const instance = process.env.HOSTNAME || process.env.POD_NAME || String(process.pid)
537+
const s3AuditStream = new S3StreamLogger({
538+
bucket: this.bucket,
539+
folder: 'logs/audit',
540+
name_format: `audit-%Y-%m-%d-%H-%M-%S-%L-${instance}.log.jsonl`,
541+
config: this.s3Config
542+
})
543+
return [new transports.Stream({ stream: s3AuditStream })]
535544
}
545+
536546
return []
537547
}
538548
}

‎packages/server/package.json‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -109,6 +109,7 @@
109109
"flowise-components": "workspace:^",
110110
"flowise-nim-container-manager": "^1.0.11",
111111
"flowise-ui": "workspace:^",
112+
"geoip-lite": "^1.4.10",
112113
"global-agent": "^3.0.0",
113114
"gulp": "^4.0.2",
114115
"handlebars": "^4.7.8",
@@ -158,6 +159,7 @@
158159
"@types/cors": "^2.8.12",
159160
"@types/crypto-js": "^4.1.1",
160161
"@types/express-session": "^1.18.0",
162+
"@types/geoip-lite": "^1.4.4",
161163
"@types/jest": "^29.5.14",
162164
"@types/jsonwebtoken": "^9.0.6",
163165
"@types/multer": "^1.4.7",

‎packages/server/src/IdentityManager.ts‎

Lines changed: 21 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ import * as fs from 'fs'
1616
import { StatusCodes } from 'http-status-codes'
1717
import jwt from 'jsonwebtoken'
1818
import path from 'path'
19+
import Stripe from 'stripe'
1920
import { LoginMethodStatus } from './enterprise/database/entities/login-method.entity'
2021
import { ErrorMessage, LoggedInUser } from './enterprise/Interface.Enterprise'
2122
import { Permissions } from './enterprise/rbac/Permissions'
@@ -33,7 +34,6 @@ import { UsageCacheManager } from './UsageCacheManager'
3334
import { GeneralErrorMessage, LICENSE_QUOTAS } from './utils/constants'
3435
import { getRunningExpressApp } from './utils/getRunningExpressApp'
3536
import { ENTERPRISE_FEATURE_FLAGS } from './utils/quotaUsage'
36-
import Stripe from 'stripe'
3737

3838
const allSSOProviders = ['azure', 'google', 'auth0', 'github']
3939
export class IdentityManager {
@@ -526,4 +526,24 @@ export class IdentityManager {
526526
throw error
527527
}
528528
}
529+
530+
/**
531+
* Cancels a Stripe subscription and syncs the result to the usage cache.
532+
*
533+
* Requests cancellation via Stripe, then updates the subscription data in cache
534+
* so usage and billing state stay consistent. Throws if the Stripe manager
535+
* is not initialized.
536+
*
537+
* @param subscriptionId - The Stripe subscription ID to cancel
538+
* @returns The cancelled Stripe subscription object
539+
*/
540+
public async cancelSubscription(subscriptionId: string) {
541+
if (!this.stripeManager) throw new Error('Stripe manager is not initialized')
542+
const subscription = await this.stripeManager.cancelSubscription(subscriptionId)
543+
const cacheManager = await UsageCacheManager.getInstance()
544+
await cacheManager.updateSubscriptionDataToCache(subscriptionId, {
545+
subsriptionDetails: this.stripeManager.getSubscriptionObject(subscription)
546+
})
547+
return subscription
548+
}
529549
}

‎packages/server/src/StripeManager.ts‎

Lines changed: 16 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
1-
import Stripe from 'stripe'
21
import { Request } from 'express'
3-
import { UsageCacheManager } from './UsageCacheManager'
2+
import Stripe from 'stripe'
43
import { UserPlan } from './Interface'
4+
import { UsageCacheManager } from './UsageCacheManager'
55
import { LICENSE_QUOTAS } from './utils/constants'
66

77
export class StripeManager {
@@ -611,4 +611,18 @@ export class StripeManager {
611611
throw error
612612
}
613613
}
614+
615+
/**
616+
* Cancels a Stripe subscription immediately without proration or an immediate invoice.
617+
*
618+
* Calls the Stripe API to cancel the subscription with `prorate: false` and
619+
* `invoice_now: false`. Throws if the Stripe client is not initialized.
620+
*
621+
* @param subscriptionId - The Stripe subscription ID to cancel
622+
* @returns A promise resolving to the Stripe API response containing the cancelled subscription
623+
*/
624+
public async cancelSubscription(subscriptionId: string): Promise<Stripe.Response<Stripe.Subscription>> {
625+
if (!this.stripe) throw new Error('Stripe is not initialized')
626+
return await this.stripe.subscriptions.cancel(subscriptionId, { prorate: false, invoice_now: false })
627+
}
614628
}

‎packages/server/src/enterprise/controllers/account.controller.ts‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,10 @@
11
import { NextFunction, Request, Response } from 'express'
22
import { StatusCodes } from 'http-status-codes'
3+
import { QueryRunner } from 'typeorm'
4+
import { InternalFlowiseError } from '../../errors/internalFlowiseError'
5+
import { GeneralErrorMessage } from '../../utils/constants'
36
import { getRunningExpressApp } from '../../utils/getRunningExpressApp'
7+
import { emitEvent, TelemetryEventCategory, TelemetryEventResult } from '../../utils/telemetry'
48
import { Organization } from '../database/entities/organization.entity'
59
import { User } from '../database/entities/user.entity'
610
import { AccountDTO, AccountService } from '../services/account.service'
@@ -146,6 +150,45 @@ export class AccountController {
146150
return res.json({ message: 'Authentication failed' })
147151
}
148152
}
153+
154+
public async delete(req: Request, res: Response, next: NextFunction) {
155+
let queryRunner: QueryRunner | undefined
156+
try {
157+
const { confirmationText } = req.body
158+
if (confirmationText !== 'permanently delete') {
159+
throw new InternalFlowiseError(StatusCodes.BAD_REQUEST, 'Confirmation text must match "permanently delete"')
160+
}
161+
162+
queryRunner = getRunningExpressApp().AppDataSource.createQueryRunner()
163+
await queryRunner.connect()
164+
if (!req.user || !req.ip) throw new InternalFlowiseError(StatusCodes.UNAUTHORIZED, GeneralErrorMessage.UNAUTHORIZED)
165+
166+
const accountService = new AccountService()
167+
await accountService.delete(queryRunner, req.user, req.ip)
168+
169+
return res.status(StatusCodes.OK).json({ message: 'Account deleted' })
170+
} catch (error) {
171+
if (queryRunner && queryRunner.isTransactionActive) await queryRunner.rollbackTransaction()
172+
173+
await emitEvent({
174+
category: TelemetryEventCategory.AUDIT,
175+
eventType: 'account-deleted',
176+
actionType: 'delete',
177+
userId: req.user?.id ?? 'unknown',
178+
orgId: req.user?.activeOrganizationId ?? 'unknown',
179+
resourceId: req.user?.id ?? 'unknown',
180+
ipAddress: req.ip,
181+
result: TelemetryEventResult.FAILED,
182+
metadata: {
183+
failureReason: error instanceof InternalFlowiseError ? error.message : 'internal_error'
184+
}
185+
})
186+
187+
next(error)
188+
} finally {
189+
if (queryRunner && !queryRunner.isReleased) await queryRunner.release()
190+
}
191+
}
149192
}
150193

151194
function sanitizeRegistrationDTO(data: AccountDTO): AccountDTO {

0 commit comments

Comments
 (0)