数据库审计与异常查询检测最佳实践

数据库审计与异常查询检测最佳实践概述通过采集查询事件并构建行为画像，对慢查询、权限异常与模式异常进行检测与告警，提升数据安全与可观测性。审计事件模型type QueryEvent = {

user: string

role: string

tenant?: string

sql: string

durationMs: number

rows: number

timestamp: string

clientIp: string

}

type Alert = { type: string; severity: 'low' | 'medium' | 'high'; message: string; timestamp: string }

慢查询与权限异常检测class QueryDetector {

slowThreshold = 1000

sensitiveTables = ['accounts', 'payments']

sensitiveRoles = ['readonly']

detect(e: QueryEvent): Alert[] {

const alerts: Alert[] = []

if (e.durationMs > this.slowThreshold) alerts.push(this.alert('slow_query', 'medium', `Slow query ${e.durationMs}ms`))

if (this.accessSensitive(e) && this.isSensitiveRole(e)) alerts.push(this.alert('permission_misuse', 'high', `Sensitive access by ${e.role}`))

return alerts

}

accessSensitive(e: QueryEvent): boolean {

return this.sensitiveTables.some(t => e.sql.toLowerCase().includes(t))

}

isSensitiveRole(e: QueryEvent): boolean { return this.sensitiveRoles.includes(e.role) }

alert(type: string, severity: Alert['severity'], message: string): Alert { return { type, severity, message, timestamp: new Date().toISOString() } }

}

行为画像与异常评分class BehaviorProfile {

profiles = new Map<string, { avgDuration: number; count: number }>()

update(e: QueryEvent) {

const p = this.profiles.get(e.user) || { avgDuration: 0, count: 0 }

p.avgDuration = (p.avgDuration * p.count + e.durationMs) / (p.count + 1)

p.count += 1

this.profiles.set(e.user, p)

}

anomalyScore(e: QueryEvent): number {

const p = this.profiles.get(e.user)

if (!p) return 0

const ratio = e.durationMs / Math.max(1, p.avgDuration)

return Math.min(100, Math.round((ratio - 1) * 20))

}

}

运维要点建立敏感对象清单与角色矩阵，检测权限滥用对慢查询设定阈值与告警，并驱动索引优化审计事件入库并按用户/租户进行画像与异常评分通过行为画像与异常检测，可在不影响性能的情况下提升数据库安全与可观测性。