19.14-审计日志

要点

审计日志（Audit Log）是安全体系的最后一道防线——记录所有安全相关事件，事后追溯和分析
审计日志 ≠ 应用日志：应用日志是调试用的，审计日志是安全合规用的，两者目的和记录内容不同
需要审计的事件：登录/登出、权限变更、敏感操作（删除、导出）、安全事件（攻击尝试、限流触发）
审计日志必须包含：谁（who）、什么时间（when）、什么操作（what）、什么结果（result）、从哪里来（where）
Workers 环境的审计日志存储：D1（结构化查询）、R2（大量日志存储）、日志平台（Loki/ELK/Datadog）
AI 场景的审计：记录 LLM 调用（prompt、输出、token 使用）、工具调用、安全过滤事件

1. 审计日志 vs 应用日志

维度	应用日志	审计日志
目的	调试、监控、排障	安全合规、事件追溯、审计
受众	开发者、运维	安全团队、合规审计、管理员
内容	代码执行细节、错误堆栈	用户操作、权限变更、安全事件
保留期限	通常 7-30 天	通常 1 年或更长（合规要求）
不可篡改	可以修改	必须不可篡改（写入后只追加）
敏感信息	可能包含（需要脱敏）	包含（本身就是记录敏感操作）

两者可以共存——应用日志写到 stdout，审计日志写到专门的存储。

2. 审计日志的数据模型

每条审计日志应该包含：

interface AuditLog {
  // 谁
  userId: string
  userEmail?: string
  userRole: string
  ipAddress: string
  userAgent: string
 
  // 什么时间
  timestamp: string  // ISO 8601
 
  // 什么操作
  action: string  // 例如：'user.login', 'data.export', 'permission.change'
  resourceType: string  // 例如：'user', 'order', 'document'
  resourceId?: string  // 操作的具体资源 ID
 
  // 什么结果
  success: boolean
  statusCode?: number
 
  // 详情
  details?: Record<string, any>
 
  // 从哪里来
  source: 'web' | 'api' | 'admin' | 'system'
}

3. 审计日志中间件

用中间件自动记录所有 API 请求：

// src/middleware/audit-log.ts
import { Context, Next } from 'hono'
 
export function auditLog() {
  return async (c: Context, next: Next) => {
    const startTime = Date.now()
 
    // 执行请求
    await next()
 
    // 记录审计日志
    const duration = Date.now() - startTime
    const user = c.get('user')
    const clientIP = c.req.header('CF-Connecting-IP') || 'unknown'
 
    const log: AuditLog = {
      userId: user?.id || 'anonymous',
      userEmail: user?.email,
      userRole: user?.role || 'anonymous',
      ipAddress: clientIP,
      userAgent: c.req.header('User-Agent') || '',
      timestamp: new Date().toISOString(),
      action: `${c.req.method} ${c.req.path}`,
      resourceType: c.req.path.split('/')[2] || 'unknown',
      resourceId: c.req.param('id'),
      success: c.res.status >= 200 && c.res.status < 400,
      statusCode: c.res.status,
      source: c.req.path.startsWith('/admin') ? 'admin' : 'api',
      details: {
        duration,
        method: c.req.method,
        path: c.req.path,
      },
    }
 
    // 异步写入，不阻塞响应
    c.env.waitUntil(writeAuditLog(c.env, log))
  }
}
 
async function writeAuditLog(env: Env, log: AuditLog) {
  // 写入 D1
  await env.DB.prepare(`
    INSERT INTO audit_logs (
      user_id, user_email, user_role, ip_address, user_agent,
      timestamp, action, resource_type, resource_id,
      success, status_code, source, details
    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
  `).bind(
    log.userId,
    log.userEmail || null,
    log.userRole,
    log.ipAddress,
    log.userAgent,
    log.timestamp,
    log.action,
    log.resourceType,
    log.resourceId || null,
    log.success ? 1 : 0,
    log.statusCode,
    log.source,
    JSON.stringify(log.details || {}),
  ).run()
}
 
// 使用
app.use('/api/*', auditLog())
app.use('/admin/*', auditLog())

4. 关键安全事件审计

4.1 登录/登出

app.post('/api/auth/login', async (c) => {
  const { email, password } = await c.req.json()
  const user = await verifyCredentials(email, password, c.env)
 
  if (!user) {
    // 登录失败也要记录
    await writeAuditLog(c.env, {
      userId: 'unknown',
      userRole: 'anonymous',
      ipAddress: c.req.header('CF-Connecting-IP') || 'unknown',
      userAgent: c.req.header('User-Agent') || '',
      timestamp: new Date().toISOString(),
      action: 'auth.login.failed',
      resourceType: 'user',
      success: false,
      statusCode: 401,
      source: 'web',
      details: { email, reason: 'invalid_credentials' },
    })
 
    return c.json({ error: 'Invalid credentials' }, 401)
  }
 
  // 登录成功
  const token = await generateJWT(user, c.env)
 
  await writeAuditLog(c.env, {
    userId: user.id,
    userEmail: user.email,
    userRole: user.role,
    ipAddress: c.req.header('CF-Connecting-IP') || 'unknown',
    userAgent: c.req.header('User-Agent') || '',
    timestamp: new Date().toISOString(),
    action: 'auth.login.success',
    resourceType: 'user',
    resourceId: user.id,
    success: true,
    statusCode: 200,
    source: 'web',
    details: { email },
  })
 
  return c.json({ token })
})

4.2 权限变更

app.put('/api/admin/users/:id/role', authMiddleware, requireRole('admin'), async (c) => {
  const targetUserId = c.req.param('id')
  const { role } = await c.req.json()
  const adminUser = c.get('user')
 
  // 记录权限变更
  await writeAuditLog(c.env, {
    userId: adminUser.id,
    userEmail: adminUser.email,
    userRole: adminUser.role,
    ipAddress: c.req.header('CF-Connecting-IP') || 'unknown',
    userAgent: c.req.header('User-Agent') || '',
    timestamp: new Date().toISOString(),
    action: 'user.role.changed',
    resourceType: 'user',
    resourceId: targetUserId,
    success: true,
    statusCode: 200,
    source: 'admin',
    details: { newRole: role, changedBy: adminUser.email },
  })
 
  // 执行变更
  await c.env.DB
    .prepare('UPDATE users SET role = ? WHERE id = ?')
    .bind(role, targetUserId)
    .run()
 
  return c.json({ ok: true })
})

4.3 敏感数据操作

// 数据导出
app.get('/api/data/export', authMiddleware, async (c) => {
  const user = c.get('user')
  const data = await exportData(c.env)
 
  // 记录敏感操作
  await writeAuditLog(c.env, {
    userId: user.id,
    userEmail: user.email,
    userRole: user.role,
    ipAddress: c.req.header('CF-Connecting-IP') || 'unknown',
    userAgent: c.req.header('User-Agent') || '',
    timestamp: new Date().toISOString(),
    action: 'data.export',
    resourceType: 'data',
    success: true,
    statusCode: 200,
    source: 'api',
    details: {
      recordCount: data.length,
      exportedFields: ['id', 'name', 'email', 'phone'],
    },
  })
 
  return c.json(data)
})

4.4 安全事件

// 限流触发
async function recordRateLimitHit(env: Env, ip: string, path: string) {
  await writeAuditLog(env, {
    userId: 'anonymous',
    userRole: 'anonymous',
    ipAddress: ip,
    userAgent: '',
    timestamp: new Date().toISOString(),
    action: 'security.rate_limit.hit',
    resourceType: 'endpoint',
    success: false,
    statusCode: 429,
    source: 'system',
    details: { ip, path },
  })
}
 
// CSRF 检测失败
async function recordCSRFViolation(env: Env, ip: string, origin: string) {
  await writeAuditLog(env, {
    userId: 'anonymous',
    userRole: 'anonymous',
    ipAddress: ip,
    userAgent: '',
    timestamp: new Date().toISOString(),
    action: 'security.csrf.violation',
    resourceType: 'endpoint',
    success: false,
    statusCode: 403,
    source: 'system',
    details: { ip, origin },
  })
}

5. AI 场景的审计

5.1 LLM 调用审计

// LLM 调用审计
async function auditLLMCall(
  env: Env,
  userId: string,
  model: string,
  promptTokens: number,
  completionTokens: number,
  duration: number,
  filtered: boolean
) {
  await writeAuditLog(env, {
    userId,
    userRole: '',  // 从其他地方获取
    ipAddress: '',
    userAgent: '',
    timestamp: new Date().toISOString(),
    action: 'ai.llm.call',
    resourceType: 'llm',
    success: true,
    source: 'api',
    details: {
      model,
      promptTokens,
      completionTokens,
      totalTokens: promptTokens + completionTokens,
      duration,
      filtered,
    },
  })
}
 
// 在 LLM 调用中集成
app.post('/api/ai/chat', authMiddleware, async (c) => {
  const user = c.get('user')
  const { messages, model } = await c.req.json()
 
  const startTime = Date.now()
  const result = await callLLM(messages, model, c.env)
  const duration = Date.now() - startTime
 
  // 记录 AI 调用
  await auditLLMCall(
    c.env,
    user.id,
    model,
    result.usage.promptTokens,
    result.usage.completionTokens,
    duration,
    result.filtered
  )
 
  return c.json({ content: result.content })
})

5.2 工具调用审计

// Agent 工具调用审计
async function auditToolCall(
  env: Env,
  userId: string,
  toolName: string,
  params: any,
  result: any,
  duration: number,
  allowed: boolean
) {
  await writeAuditLog(env, {
    userId,
    userRole: '',
    ipAddress: '',
    userAgent: '',
    timestamp: new Date().toISOString(),
    action: `ai.tool.${allowed ? 'call' : 'blocked'}`,
    resourceType: 'tool',
    resourceId: toolName,
    success: allowed,
    source: 'api',
    details: {
      toolName,
      params: sanitizeParams(params),  // 脱敏参数
      resultSize: JSON.stringify(result).length,
      duration,
    },
  })
}

5.3 安全过滤审计

// 内容被过滤时记录
async function auditContentFilter(
  env: Env,
  userId: string,
  contentType: 'input' | 'output',
  reason: string,
  content: string
) {
  await writeAuditLog(env, {
    userId,
    userRole: '',
    ipAddress: '',
    userAgent: '',
    timestamp: new Date().toISOString(),
    action: `ai.content.filtered.${contentType}`,
    resourceType: 'content',
    success: false,
    source: 'system',
    details: {
      contentType,
      reason,
      contentPreview: content.slice(0, 200),  // 只记录前 200 字符
    },
  })
}

6. 审计日志存储

6.1 D1 存储（结构化查询）

适合中小规模（百万条以内），支持 SQL 查询：

-- 审计日志表
CREATE TABLE audit_logs (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  user_id TEXT NOT NULL,
  user_email TEXT,
  user_role TEXT NOT NULL,
  ip_address TEXT NOT NULL,
  user_agent TEXT,
  timestamp TEXT NOT NULL,
  action TEXT NOT NULL,
  resource_type TEXT,
  resource_id TEXT,
  success INTEGER NOT NULL,
  status_code INTEGER,
  source TEXT NOT NULL,
  details TEXT,  -- JSON
  created_at TEXT DEFAULT (datetime('now'))
);
 
-- 索引
CREATE INDEX idx_audit_user_id ON audit_logs(user_id);
CREATE INDEX idx_audit_action ON audit_logs(action);
CREATE INDEX idx_audit_timestamp ON audit_logs(timestamp);
CREATE INDEX idx_audit_resource ON audit_logs(resource_type, resource_id);

// 查询审计日志
app.get('/api/admin/audit-logs', authMiddleware, requireRole('admin'), async (c) => {
  const { userId, action, startDate, endDate, page = '1', pageSize = '50' } = c.req.query()
 
  let sql = 'SELECT * FROM audit_logs WHERE 1=1'
  const params: any[] = []
 
  if (userId) {
    sql += ' AND user_id = ?'
    params.push(userId)
  }
  if (action) {
    sql += ' AND action LIKE ?'
    params.push(`${action}%`)
  }
  if (startDate) {
    sql += ' AND timestamp >= ?'
    params.push(startDate)
  }
  if (endDate) {
    sql += ' AND timestamp <= ?'
    params.push(endDate)
  }
 
  sql += ' ORDER BY timestamp DESC LIMIT ? OFFSET ?'
  params.push(parseInt(pageSize), (parseInt(page) - 1) * parseInt(pageSize))
 
  const result = await c.env.DB.prepare(sql).bind(...params).all()
  return c.json(result.results)
})

6.2 日志平台

大规模场景（百万条以上）推荐把审计日志发送到专门的日志平台：

Loki（Grafana 生态）：轻量级，适合日志搜索
ELK（Elasticsearch + Logstash + Kibana）：功能全面，资源消耗大
Datadog：商业方案，集成完善

// 发送到日志平台
async function sendToLogPlatform(env: Env, log: AuditLog) {
  await fetch(env.LOG_PLATFORM_URL, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'Authorization': `Bearer ${env.LOG_PLATFORM_TOKEN}`,
    },
    body: JSON.stringify(log),
  })
}

6.3 日志保留策略

审计日志需要保留足够长的时间（合规要求通常 1 年以上），但不能无限增长：

// 清理超过 1 年的日志
app.post('/api/admin/audit-logs/cleanup', authMiddleware, requireRole('admin'), async (c) => {
  const oneYearAgo = new Date()
  oneYearAgo.setFullYear(oneYearAgo.getFullYear() - 1)
 
  await c.env.DB.prepare(
    'DELETE FROM audit_logs WHERE timestamp < ?'
  ).bind(oneYearAgo.toISOString()).run()
 
  return c.json({ ok: true })
})

7. 审计日志分析

7.1 异常检测

// 检测同一 IP 短时间内多次登录失败
async function detectBruteForce(env: Env, ip: string) {
  const fiveMinutesAgo = new Date(Date.now() - 5 * 60 * 1000).toISOString()
 
  const result = await env.DB.prepare(`
    SELECT COUNT(*) as count
    FROM audit_logs
    WHERE ip_address = ?
      AND action = 'auth.login.failed'
      AND timestamp > ?
  `).bind(ip, fiveMinutesAgo).first()
 
  if ((result?.count || 0) > 10) {
    // 触发告警
    await sendAlert(env, {
      type: 'brute_force',
      ip,
      attempts: result.count,
    })
  }
}
 
// 检测异常的数据导出量
async function detectAnomalousExport(env: Env, userId: string) {
  const oneHourAgo = new Date(Date.now() - 60 * 60 * 1000).toISOString()
 
  const result = await env.DB.prepare(`
    SELECT SUM(JSON_EXTRACT(details, '$.recordCount')) as totalRecords
    FROM audit_logs
    WHERE user_id = ?
      AND action = 'data.export'
      AND timestamp > ?
  `).bind(userId, oneHourAgo).first()
 
  if ((result?.totalRecords || 0) > 10000) {
    await sendAlert(env, {
      type: 'anomalous_export',
      userId,
      totalRecords: result.totalRecords,
    })
  }
}

7.2 安全报告

// 每日安全报告
async function generateDailySecurityReport(env: Env) {
  const today = new Date().toISOString().slice(0, 10)
  const yesterday = new Date(Date.now() - 86400000).toISOString().slice(0, 10)
 
  const stats = await env.DB.prepare(`
    SELECT
      action,
      COUNT(*) as count,
      SUM(CASE WHEN success = 0 THEN 1 ELSE 0 END) as failures
    FROM audit_logs
    WHERE timestamp >= ?
    GROUP BY action
    ORDER BY count DESC
  `).bind(yesterday).all()
 
  return {
    date: yesterday,
    summary: stats.results,
  }
}

8. 审计日志检查清单

检查项	说明
登录/登出事件	成功和失败都要记录
权限变更	谁改了谁的权限
敏感操作	删除、导出、批量修改
安全事件	限流触发、CSRF 违规、攻击尝试
AI 调用	模型、token 使用、过滤事件
工具调用	Agent 调用了什么工具
不可篡改	日志写入后只能追加，不能修改删除
保留策略	至少 1 年，按合规要求
异常检测	自动检测暴力破解、异常操作
脱敏处理	日志里的密码等敏感字段要脱敏

9. 小结

审计日志是安全体系的最后一道防线——记录所有安全相关事件，事后追溯和分析。

审计日志 ≠ 应用日志。审计日志记录「谁在什么时间做了什么操作、结果如何」，保留期限更长，必须不可篡改。

需要审计的事件：登录/登出、权限变更、敏感操作（删除、导出）、安全事件（攻击尝试、限流触发）、AI 调用和工具调用。

Workers 环境下审计日志存 D1（结构化查询）或日志平台（大规模）。日志需要保留至少 1 年。

异常检测：同一 IP 多次登录失败、异常的数据导出量、可疑的工具调用。

下一篇是本系列最后一篇——安全测试清单，把所有安全措施汇总成一个可执行的检查表。