【实战ES】实战 Elasticsearch：快速上手与深度实践-6.2.1记录用户操作轨迹

👉 点击关注不迷路
👉 点击关注不迷路
👉 点击关注不迷路

6.2.1 用户操作轨迹审计日志深度实践指南

• 审计日志全链路采集、存储与分析架构

1. 核心审计模型解析

1.1 审计事件分类矩阵

1.2 日志存储成本分析（按1亿事件/日）

2. 全链路配置实战

2.1 审计日志基础配置

# 启用 X-Pack 安全审计功能
# 当设置为 true 时，Elasticsearch 会开启安全审计机制，记录与安全相关的事件
# 这些事件可以帮助管理员监控和分析系统的安全状况，例如用户认证、访问控制等方面的操作
xpack.security.audit.enabled: true

# 指定需要包含在日志文件中的审计事件类型
# 这里列出了三种事件类型，分别是 authentication_granted（认证成功）、access_denied（访问被拒绝）和 anonymous_access_denied（匿名访问被拒绝）
# 只有这些类型的事件会被记录到审计日志中，方便管理员重点关注关键的安全事件
xpack.security.audit.logfile.events.include: authentication_granted,access_denied,anonymous_access_denied

# 指定需要从日志文件中排除的审计事件类型
# 这里排除了 authentication_success（认证成功）和 connection_granted（连接被允许）事件
# 因为这些事件可能会产生大量的日志记录，排除它们可以减少日志文件的大小，同时也避免无关信息干扰管理员对关键事件的分析
xpack.security.audit.logfile.events.exclude: authentication_success,connection_granted

# 配置是否在审计日志中记录请求体
# 当设置为 true 时，审计日志会包含请求的详细内容，这对于深入分析安全事件非常有帮助
# 但同时也会增加日志文件的大小，因为请求体可能包含大量的数据
xpack.security.audit.logfile.events.emit_request_body: true

# 设置审计日志文件的滚动间隔
# 这里设置为 1d，表示每天进行一次日志文件的滚动操作
# 滚动操作会创建一个新的日志文件，并将旧的日志文件进行归档，以避免单个日志文件过大
xpack.security.audit.logfile.rotation.interval: 1d

# 设置审计日志文件滚动的大小阈值
# 当日志文件的大小达到 1GB 时，会触发日志文件的滚动操作
# 结合滚动间隔和大小阈值，可以有效地管理审计日志文件的大小和数量
xpack.security.audit.logfile.rotation.size: 1GB

2.2 高级审计策略

// 向 Elasticsearch 发送 PUT 请求，用于更新集群的设置
PUT /_cluster/settings
{
    // 使用 "persistent" 表示这些设置会持久化存储，即使 Elasticsearch 重启后依然生效
    "persistent": {
        // 开启在审计日志文件中记录请求体的功能
        // 当设置为 true 时，审计日志将包含请求的详细内容，有助于深入分析安全事件
        // 不过，这可能会增加日志文件的大小，因为请求体可能包含大量数据
        "xpack.security.audit.logfile.events.emit_request_body": true,
        // 配置审计日志条目的过滤规则
        "xpack.security.audit.logfile.entries.filter": {
            // 定义一个过滤策略
            "policy": {
                // 规则列表，可根据不同条件设置不同的日志级别
                "rules": [
                    {
                        // 指定用户列表，这里包含 "admin" 和 "root" 用户
                        // 表示该规则适用于这些用户执行的操作
                        "users": ["admin", "root"],
                        // 指定操作类型，"*" 表示所有操作
                        // 即 "admin" 和 "root" 用户的所有操作都会应用此规则
                        "actions": ["*"],
                        // 为符合上述条件的操作设置日志级别为 "INFO"
                        // 这意味着 "admin" 和 "root" 用户的所有操作产生的审计日志将以 "INFO" 级别记录
                        "log_level": "INFO"
                    },
                    {
                        // 指定索引列表，使用通配符 "financial_*" 表示所有以 "financial_" 开头的索引
                        // 该规则适用于对这些索引的操作
                        "indices": ["financial_*"],
                        // 为涉及这些索引的操作设置日志级别为 "DEBUG"
                        // 即对以 "financial_" 开头的索引的操作产生的审计日志将以 "DEBUG" 级别记录，可提供更详细的信息
                        "log_level": "DEBUG"
                    }
                ]
            }
        }
    }
}

3. 存储优化策略

3.1 索引生命周期管理

// 向 Elasticsearch 发送 PUT 请求，用于创建一个名为 audit_logs_policy 的索引生命周期管理（ILM）策略
PUT /_ilm/policy/audit_logs_policy
{
    // 定义 ILM 策略的主体内容
    "policy": {
        // 定义索引在不同阶段（phases）的操作
        "phases": {
            // 热阶段（hot）：索引处于活跃写入和频繁查询的阶段
            "hot": {
                // 热阶段要执行的操作
                "actions": {
                    // 滚动（rollover）操作：当索引满足一定条件时，创建一个新的索引来继续写入数据
                    "rollover": {
                        // 最大索引大小达到 50GB 时触发滚动操作
                        "max_size": "50GB",
                        // 索引的最大存活时间达到 1 天时触发滚动操作
                        "max_age": "1d"
                    },
                    // 设置索引的优先级
                    "set_priority": {
                        // 将索引优先级设置为 100，较高的优先级有助于在资源分配等场景中得到更优先的处理
                        "priority": 100
                    }
                }
            },
            // 温阶段（warm）：索引不再频繁写入，但仍需要保留以便查询
            "warm": {
                // 索引进入温阶段的最小年龄，即索引创建 7 天后进入该阶段
                "min_age": "7d",
                // 温阶段要执行的操作
                "actions": {
                    // 强制合并（forcemerge）操作：将索引的段（segments）合并成指定数量，以减少磁盘 I/O 和提高查询性能
                    "forcemerge": {
                        // 将索引段合并为最多 1 个
                        "max_num_segments": 1
                    }
                }
            },
            // 删除阶段（delete）：当索引达到一定年龄后，将其删除以释放磁盘空间
            "delete": {
                // 索引进入删除阶段的最小年龄，即索引创建 365 天后进入该阶段
                "min_age": "365d",
                // 删除阶段要执行的操作
                "actions": {
                    // 执行删除索引操作
                    "delete": {}
                }
            }
        }
    }
}

3.2 存储优化对比

• LZ4
  • 是一种无损数据压缩算法，同时也指基于该算法开发的快速数据压缩库。
  • 其核心原理是在输入数据中寻找重复出现的字符串（也称为 “匹配串”），并使用对这些匹配串的引用（偏移量和长度）来替代原始数据，从而实现数据的压缩。
  • 应用场景
    • 数据传输： 在网络传输中，如实时视频流、游戏数据传输等场景，需要快速地对数据进行压缩和解压缩，以减少传输带宽和延迟。LZ4 的高速特性使其非常适合这类应用。
    • 缓存系统： 在缓存系统中，为了节省存储空间和提高缓存读写效率，需要对缓存数据进行压缩。LZ4 可以在不显著影响性能的前提下，有效地减少缓存数据的大小。
    • 日志存储： 对于大量的日志数据，使用 LZ4 进行压缩可以快速存储和检索，同时减少磁盘空间的占用。
• ZSTD（Zstandard）
  • 是一种快速、高效的无损数据压缩算法，由 Facebook 开发并开源。它在速度、压缩比和内存使用之间取得了很好的平衡，适用于多种场景。
  • ZSTD 基于 LZ77 算法，结合了变长编码和 Huffman 编码，主要包含以下步骤：

• 应用场景
  • 数据存储： 日志文件、数据库备份、云存储（如 HDFS、Amazon S3）。
  • 数据传输： 实时流数据（如视频、游戏）、网络协议优化。
  • 大数据处理： Hadoop、Spark、Flink 等框架中的压缩。
  • 容器与虚拟化： Docker 镜像、虚拟机磁盘压缩。
  • Elasticsearch： 可替代 LZ4 以减少磁盘占用（需权衡压缩 / 解压性能）。
• 与其他算法的对比
  

4. 合规性集成方案

4.1 法规要求映射表

4.2 合规性检查脚本

• GDPR
  • GDPR 即《通用数据保护条例》（General Data Protection Regulation），它是欧盟在数据保护和隐私领域的一项重要法规
• HIPAA
  • HIPAA（Health Insurance Portability and Accountability Act）即《健康保险流通与责任法案》，是美国于 1996 年颁布的一项联邦法律，旨在保护医疗信息的隐私与安全。
  • 核心目标：
    • 保障患者医疗信息（PHI）的隐私与安全。
    • 简化医疗数据的电子传输流程。
    • 防止医疗欺诈和滥用行为。

def check_audit_compliance(log_index):
    """
    此函数用于对指定日志索引进行合规性检查，主要针对 GDPR 和 HIPAA 法规。
    :param log_index: 要进行合规性检查的日志索引名称
    :return: 调用 generate_compliance_report 函数生成的合规性报告
    """
    # GDPR合规检查：是否有数据删除记录
    # 构建一个 Elasticsearch 查询，用于检查最近 30 天内是否存在数据删除操作
    gdpr_check = {
        "query": {
            "bool": {
                # 使用 bool 查询的 must 子句，要求查询结果必须同时满足以下两个条件
                "must": [
                    # 条件 1：事件动作必须是 "delete"
                    {"term": {"event.action": "delete"}},
                    # 条件 2：事件时间戳必须在当前时间往前推 30 天内
                    {"range": {"@timestamp": {"gte": "now-30d"}}}
                ]
            }
        }
    }

    # HIPAA合规检查：是否记录PHI数据访问
    # 构建一个 Elasticsearch 查询，用于检查是否存在对医疗记录索引的访问记录
    # PHI（Protected Health Information）即受保护的健康信息，HIPAA 法规要求对其访问进行记录
    hipaa_check = {
        "query": {
            # 使用 term 查询，要求索引名称必须是 "medical_records"
            "term": {"indices": "medical_records"}
        }
    }

    # 执行检查并生成报告
    # 将 GDPR 和 HIPAA 的检查信息作为元组列表传递给 generate_compliance_report 函数
    # 元组的第一个元素是法规名称，第二个元素是对应的 Elasticsearch 查询
    return generate_compliance_report([
        ("GDPR", gdpr_check),
        ("HIPAA", hipaa_check)
    ])

5. 高级分析应用

5.1 异常检测规则

// 向 Elasticsearch 发送 PUT 请求，用于创建一个名为 audit_anomaly 的 Watcher 监控任务

PUT /_watcher/watch/audit_anomaly
{
    // 定义监控任务的触发条件
    "trigger": {
        // 设定按照固定的时间间隔触发，这里设置为每 5 分钟触发一次
        "schedule": {
            "interval": "5m"
        }
    },
    // 定义监控任务的数据输入来源，这里使用搜索查询的方式获取数据
    "input": {
        "search": {
            "request": {
                // 指定要搜索的索引，使用通配符 .security - audit* 表示匹配所有以 .security - audit 开头的索引
                "indices": [".security-audit*"],
                "body": {
                    // 设置返回的文档数量为 0，因为我们只关注聚合结果，不需要具体的文档内容
                    "size": 0,
                    "aggs": {
                        // 定义一个名为 unusual_activity 的聚合操作
                        "unusual_activity": {
                            // 使用 terms 聚合，根据 user 字段进行分组统计
                            "terms": {
                                // 指定分组的字段为 user
                                "field": "user",
                                // 只返回出现次数最多的前 10 个分组
                                "size": 10,
                                // 按照分组的文档数量降序排序
                                "order": {
                                    "_count": "desc"
                                }
                            }
                        }
                    }
                }
            }
        }
    },
    // 定义监控任务的条件判断逻辑，用于决定是否触发后续的操作
    "condition": {
        "compare": {
            // 比较搜索结果中 unusual_activity 聚合的第一个分组（即出现次数最多的分组）的文档数量
            "ctx.payload.aggregations.unusual_activity.buckets.0.doc_count": {
                // 判断该文档数量是否大于 100，如果大于 100 则满足条件，触发后续操作
                "gt": 100
            }
        }
    }
}

5.2 可视化分析模板

// 向 Kibana 发送 POST 请求，用于创建一个名为 audit_monitor 的仪表盘

POST /kibana/dashboard/audit_monitor
{
    // 仪表盘的标题，这里设置为中文的“安全审计监控看板”
    "title": "安全审计监控看板",
    // 仪表盘上的面板列表，每个面板用于展示不同类型的数据可视化
    "panels": [
        {
            // 第一个面板的类型为时间序列图（timeseries）
            "type": "timeseries",
            // 该面板的标题，显示为“认证失败趋势”
            "title": "认证失败趋势",
            "params": {
                // 指定要从哪些索引中获取数据，使用通配符 .security - audit* 表示匹配所有以 .security - audit 开头的索引
                "index": ".security-audit*",
                // 指定时间字段，这里使用 @timestamp 字段来确定数据的时间顺序
                "time_field": "@timestamp",
                // 定义搜索查询条件，使用 term 查询，只筛选出 event.type 字段值为 "authentication_failed" 的文档，即认证失败的事件
                "query": {
                    "term": {
                        "event.type": "authentication_failed"
                    }
                },
                // 设置时间序列图的时间间隔为 1 小时，即按照每小时的时间粒度来统计数据
                "interval": "1h"
            }
        },
        {
            // 第二个面板的类型为饼图（pie）
            "type": "pie",
            // 该面板的标题，显示为“操作类型分布”
            "title": "操作类型分布",
            "params": {
                // 同样指定要从哪些索引中获取数据，匹配所有以 .security - audit 开头的索引
                "index": ".security-audit*",
                // 设置数据的分割模式为按照词项（terms）进行分割，也就是根据指定字段的值进行分组
                "split_mode": "terms",
                // 指定用于分组的字段为 event.action，即根据操作类型进行分组统计
                "terms_field": "event.action"
            }
        }
    ]
}

6. 企业级运维方案

6.1 金融行业案例

# 审计日志增强配置
# 配置 X-Pack 安全审计功能
xpack.security.audit:
  # 启用安全审计功能，设为 true 表示开启，开启后会对系统中的安全相关事件进行审计记录
  enabled: true
  # 配置审计日志文件相关设置
  logfile:
    events:
      # 配置需要包含在审计日志中的事件类型。这里设置为 '*' 表示包含所有类型的事件
      include: '*'
      # 配置需要从审计日志中排除的事件类型。设置为 '_all' 表示不记录所有事件，但由于上面 include 为 '*'，
      # 所以这里排除设置不生效，实际会记录所有事件
      exclude: '_all'
    formats:
      # 定义审计日志的输出格式，这里采用 JSON 格式
      - type: json
        filter:
          # 配置 JSON 格式输出时需要包含的字段过滤器
          - type: include
            # 指定需要包含在 JSON 输出中的字段，这些字段会出现在审计日志记录里
            fields: ['@timestamp', 'user', 'event', 'request']
  # 配置审计信息的输出目标
  outputs:
    # 第一个输出目标是日志文件，即会将审计信息记录到日志文件中
    - type: logfile
    # 第二个输出目标是 Webhook，会将审计信息发送到指定的 Webhook 地址
    - type: webhook
      # 指定 Webhook 的目标主机地址，这里是一个列表，目前列表中只有一个地址
      # 当有审计信息时，会将信息发送到 https://splunk.company.com 这个地址
      host: ["https://splunk.company.com"]

• 实施效果：

6.2 运维监控指标

附录：审计日志工具链

实施规范：

1. 敏感操作日志保留至少6年
2. 审计日志存储必须加密
3. 定期执行日志完整性校验
4. 建立日志访问审批制度