收集自定义安全数据分析日志

解析器版本:6.0

支持:

本文档介绍了如何使用多种注入方法将自定义安全分析数据注入到 Google Security Operations 中。本指南适用于没有预构建解析器或日志类型的自定义安全数据源。

自定义安全数据分析涵盖专有安全遥测数据、自定义应用日志、内部安全工具,或来自未与 Google SecOps 原生集成的来源的任何安全相关数据。您可以将这些数据作为非结构化日志注入,并可选择使用自定义解析器对其进行标准化处理。

准备工作

请确保满足以下前提条件:

  • Google SecOps 实例
  • 能够以 JSON、CSV、SYSLOG 或其他结构化格式导出日志的自定义安全数据源
  • 访问以下任一内容:
    • Google Cloud Console(用于创建 API 密钥和 GCS)
    • AWS 控制台(适用于 S3 或 Firehose)
    • Azure 门户(适用于 Azure Blob Storage)
    • 能够发送 Webhook 请求的 HTTP 客户端或应用
  • 在 Google SecOps 中创建和管理 Feed 的权限

选择注入方法

Google SecOps 支持多种自定义安全数据注入方法。选择最适合您数据源功能的方法:

注入方法 使用场景 延迟时间 设置复杂性
Webhook 应用实时推送 秒 低
Amazon S3 V2 批量导出到 S3 存储桶 几分钟到几小时 中
Google Cloud Storage V2 批量导出到 GCS 存储桶 几分钟到几小时 中
Azure Blob Storage V2 批量导出到 Azure 存储空间 几分钟到几小时 中
Amazon Data Firehose 从 AWS 实时流式传输 秒 高

选项 1:Webhook 注入(实时推送)

如果您的自定义安全应用可以向外部端点发送 HTTP POST 请求,请使用此方法。

在 Google SecOps 中创建 webhook Feed

创建 Feed

  1. 前往 SIEM 设置 > Feed。
  2. 点击添加新 Feed。
  3. 在下一页上,点击配置单个 Feed。
  4. 在 Feed 名称字段中,输入 Feed 的名称(例如 custom-security-analytics-webhook)。
  5. 选择 Webhook 作为来源类型。
  6. 选择自定义安全数据分析作为日志类型。
  7. 点击下一步。
  8. 为以下输入参数指定值:
    • 拆分定界符:可选:输入定界符以拆分多行事件。常见值:
      • \n - 换行符定界符(最常用于 NDJSON)
      • 如果每个请求都包含单个事件,请留空
    • 资源命名空间:资源命名空间
    • 注入标签:要应用于此 Feed 中事件的标签
  9. 点击下一步。
  10. 在最终确定界面中查看新的 Feed 配置,然后点击提交。

生成并保存密钥

创建 Feed 后,您必须生成用于身份验证的密钥:

  1. 在 Feed 详情页面上,点击生成密钥。
  2. 对话框会显示密钥。
  3. 复制并妥善保存密钥。

获取 Feed 端点网址

  1. 前往 Feed 的详细信息标签页。
  2. 在端点信息部分,复制 Feed 端点网址。
  3. 网址格式为:

    https://malachiteingestion-pa.googleapis.com/v2/unstructuredlogentries:batchCreate
    

    或

    https://<REGION>-malachiteingestion-pa.googleapis.com/v2/unstructuredlogentries:batchCreate
    
  4. 保存此网址以供后续步骤使用。

  5. 点击完成。

创建 Google Cloud API 密钥

Google SecOps 需要使用 API 密钥进行身份验证。在 Google Cloud Console 中创建受限 API 密钥。

创建 API 密钥

  1. 前往 Google Cloud 控制台的“凭据”页面。
  2. 选择您的项目(与您的 Chronicle 实例关联的项目)。
  3. 依次点击创建凭据 > API 密钥。
  4. 系统会创建 API 密钥,并在对话框中显示该密钥。
  5. 点击修改 API 密钥以限制密钥。

限制 API 密钥

  1. 在 API 密钥设置页面中:
    • 名称:输入一个描述性名称(例如 Chronicle Webhook API Key)。
  2. 在 API 限制下:
    1. 选择限制密钥。
    2. 在选择 API 下拉菜单中,搜索并选择 Google SecOps API(或 Chronicle API)。
  3. 点击保存。
  4. 从页面顶部的 API 密钥字段复制 API 密钥值。
  5. 安全地保存 API 密钥。

配置自定义应用以发送数据

将自定义安全应用或脚本配置为向 Chronicle Webhook 端点发送 HTTP POST 请求。

构建网络钩子网址:

  • 将 Chronicle 端点网址和 API 密钥组合在一起:

    <ENDPOINT_URL>?key=<API_KEY>
    

    示例:

    https://malachiteingestion-pa.googleapis.com/v2/unstructuredlogentries:batchCreate?key=AIzaSyD...
    

HTTP 请求格式:

  • 方法:POST

  • 网址:

    <ENDPOINT_URL>?key=<API_KEY>
    
  • 标头:

    Content-Type: application/json
    x-chronicle-auth: <SECRET_KEY>
    
  • 正文(单个活动):

    {
        "timestamp": "2025-01-15T10:30:00Z",
        "event_type": "authentication",
        "user": "john.doe@example.com",
        "action": "login",
        "result": "success",
        "source_ip": "203.0.113.45",
        "custom_field_1": "value1",
        "custom_field_2": "value2"
    }
    
  • 正文(以换行符定界符分隔的多个事件):

    {"timestamp": "2025-01-15T10:30:00Z", "event_type": "authentication", "action": "login"}
    {"timestamp": "2025-01-15T10:30:05Z", "event_type": "file_access", "action": "read"}
    {"timestamp": "2025-01-15T10:30:10Z", "event_type": "authentication", "action": "logout"}
    

示例:

  • 示例:Python 脚本:

    import requests
    import json
    from datetime import datetime
    
    # Configuration
    ENDPOINT_URL = "https://malachiteingestion-pa.googleapis.com/v2/unstructuredlogentries:batchCreate"
    API_KEY = "your-api-key-here"
    SECRET_KEY = "your-secret-key-here"
    
    # Construct full URL
    url = f"{ENDPOINT_URL}?key={API_KEY}"
    
    # Headers
    headers = {
        "Content-Type": "application/json",
        "x-chronicle-auth": SECRET_KEY
    }
    
    # Sample event
    event = {
        "timestamp": datetime.utcnow().isoformat() + "Z",
        "event_type": "custom_security_event",
        "severity": "high",
        "source": "custom_security_tool",
        "message": "Suspicious activity detected",
        "user": "admin@example.com",
        "ip_address": "192.168.1.100"
    }
    
    # Send request
    response = requests.post(url, headers=headers, data=json.dumps(event))
    
    if response.status_code == 200:
        print("Event sent successfully")
    else:
        print(f"Error: {response.status_code} - {response.text}")
    
  • 示例:cURL 命令:

    curl -X POST \
        "https://malachiteingestion-pa.googleapis.com/v2/unstructuredlogentries:batchCreate?key=YOUR_API_KEY" \
        -H "Content-Type: application/json" \
        -H "x-chronicle-auth: YOUR_SECRET_KEY" \
        -d '{
            "timestamp": "2025-01-15T10:30:00Z",
            "event_type": "security_alert",
            "severity": "critical",
            "source": "custom_tool",
            "message": "Threat detected"
        }'
    

身份验证方法参考

Google SecOps webhook Feed 支持多种身份验证方法。选择您的应用支持的方法。

如果您的应用支持自定义 HTTP 标头,请使用此方法来提高安全性。

  • 请求格式:

    POST <ENDPOINT_URL> HTTP/1.1
    Content-Type: application/json
    x-goog-chronicle-auth: <API_KEY>
    x-chronicle-auth: <SECRET_KEY>
    
    {
      "event": "data",
      "timestamp": "2025-01-15T10:30:00Z"
    }
    
  • 优势:

    • API 密钥和 Secret 不会显示在网址中。
    • 更安全,因为标头不会记录在 Web 服务器访问日志中。
    • 如果应用支持,则首选此方法。

方法 2:查询参数

如果您的应用不支持自定义标头,请将凭据附加到网址。

  • 网址格式:

    <ENDPOINT_URL>?key=<API_KEY>&secret=<SECRET_KEY>
    

    示例:

    https://malachiteingestion-pa.googleapis.com/v2/unstructuredlogentries:batchCreate?key=AIzaSyD...&secret=abcd1234...
    
  • 请求格式:

    POST <ENDPOINT_URL>?key=<API_KEY>&secret=<SECRET_KEY> HTTP/1.1
    Content-Type: application/json
    
    {
      "event": "data",
      "timestamp": "2025-01-15T10:30:00Z"
    }
    
  • 缺点:

    • 凭据在网址中可见。
    • 凭据可能会记录在 Web 服务器访问日志中。
    • 安全性低于标头。

方法 3:混合(网址 + 标头)

某些配置在网址中使用 API 密钥,在标头中使用密钥。

  • 请求格式:

    POST <ENDPOINT_URL>?key=<API_KEY> HTTP/1.1
    Content-Type: application/json
    x-chronicle-auth: <SECRET_KEY>
    
    {
        "event": "data",
        "timestamp": "2025-01-15T10:30:00Z"
    }
    

Webhook 限制和最佳实践

请求限制

限制 值
请求大小上限 4 MB
最大 QPS(每秒查询次数) 15000
请求超时 30 秒
重试行为 自动(采用指数退避算法)

最佳做法

  • 批量事件:使用以换行符分隔的 JSON (NDJSON) 格式在单个请求中发送多个事件,以减少开销。
  • 添加时间戳:始终添加 ISO 8601 格式的时间戳字段,以确保事件顺序准确。
  • 使用结构化数据:以 JSON 格式发送数据,以便更轻松地进行解析和提取字段。
  • 实现重试逻辑:使用指数退避算法处理暂时性故障。
  • 监控响应代码:记录并提醒非 200 响应。

选项 2:Amazon S3 V2 注入(批量导出)

如果您的自定义安全应用可以将日志导出到 Amazon S3 存储桶,请使用此方法。

创建 Amazon S3 存储桶

  1. 打开 Amazon S3 控制台。
  2. 点击创建存储桶。
  3. 提供以下配置详细信息:
    • 存储桶名称:为存储桶输入一个有意义的名称(例如 custom-security-analytics-logs)。
    • 区域:选择您的首选 AWS 区域(例如 us-east-1)。
  4. 点击创建。

创建具有 S3 访问权限的 IAM 用户

  1. 打开 IAM 控制台。
  2. 依次点击用户 > 添加用户。
  3. 输入用户名(例如 chronicle-s3-reader)。
  4. 选择以编程方式使用。
  5. 点击 Next: Permissions。
  6. 选择 Attach existing policies directly。
  7. 搜索并选择 AmazonS3FullAccess。
  8. 点击 Next: Tags。
  9. 点击下一步:检查。
  10. 点击创建用户。
  11. 点击下载 .csv 文件以保存访问密钥 ID 和私有访问密钥。
  12. 点击关闭。

配置应用以导出到 S3

配置自定义安全应用,以将日志文件写入 S3 存储桶。应用应:

  • 以结构化格式(JSON、CSV 或纯文本)写入日志。
  • 使用一致的文件命名惯例。
  • 可选:按日期整理文件(例如 logs/2025/01/15/events.json)。
  • 写入完整的文件(避免部分写入)。

  • 文件结构示例:

    s3://custom-security-analytics-logs/
    ├── security-events/
    │   ├── 2025/01/15/
    │   │   ├── events-10-00.json
    │   │   ├── events-11-00.json
    │   │   └── events-12-00.json
    
  • 日志文件格式示例 (NDJSON):

    {"timestamp": "2025-01-15T10:00:00Z", "event_type": "login", "user": "alice@example.com", "result": "success"}
    {"timestamp": "2025-01-15T10:05:00Z", "event_type": "file_access", "user": "bob@example.com", "file": "/data/sensitive.txt"}
    {"timestamp": "2025-01-15T10:10:00Z", "event_type": "logout", "user": "alice@example.com"}
    

为 S3 配置 Google SecOps Feed

  1. 前往 SIEM 设置 > Feed。
  2. 点击添加新 Feed。
  3. 在下一页上,点击配置单个 Feed。
  4. 在 Feed 名称字段中,输入 Feed 的名称(例如 custom-security-analytics-s3)。
  5. 选择 Amazon S3 V2 作为来源类型。
  6. 选择自定义 Security 数据分析 作为日志类型。
  7. 点击下一步。
  8. 为以下输入参数指定值:

    • S3 URI:存储桶 URI,格式为:s3://custom-security-analytics-logs/security-events/

    • 来源删除选项:根据您的偏好选择删除选项:

      • 永不:转移后永不删除任何文件(建议用于测试)。
      • 删除已转移的文件:在成功转移后删除文件。
      • 删除已转移的文件和空目录:成功转移后删除文件和空目录。
    • 文件存在时间上限:包含在过去指定天数内修改过的文件(默认值为 180 天)。

    • 访问密钥 ID:输入 IAM 用户的访问密钥。

    • 私有访问密钥:输入 IAM 用户的私有密钥。

    • 资源命名空间:资源命名空间。

    • 注入标签:要应用于此 Feed 中事件的标签。

  9. 点击下一步。

  10. 在最终确定界面中查看新的 Feed 配置,然后点击提交。

选项 3:Google Cloud Storage V2 提取(批量导出)

如果您的自定义安全应用可以将日志导出到 Google Cloud Storage 存储桶,请使用此方法。

创建 GCS 存储桶

  1. 前往 Google Cloud 控制台。
  2. 选择您的项目或创建一个新项目。
  3. 在导航菜单中,前往 Cloud Storage > 存储桶。
  4. 点击创建存储分区。
  5. 提供以下配置详细信息:

    设置 值
    为存储桶命名 输入一个全局唯一的名称(例如 custom-security-analytics-logs)
    位置类型 根据您的需求进行选择(区域级、双区域级、多区域级)
    位置 选择位置(例如 us-central1)
    存储类别 标准(建议用于频繁访问的日志)
    访问权限控制 均匀(推荐)
    保护工具 可选:启用对象版本控制或保留政策
  6. 点击创建。

配置应用以导出到 GCS

使用以下方法之一,将自定义安全应用的日志文件写入 GCS 存储桶:

  • Google Cloud SDK:使用 gsutil 或客户端库
  • 服务账号:创建具有 Storage Object Creator 角色的服务账号
  • 签名网址:生成签名网址,以提供临时写入权限

  • 使用 gsutil 的示例:

    gsutil cp /path/to/logs/events.json gs://custom-security-analytics-logs/security-events/
    
  • 使用 Python 客户端库的示例:

    from google.cloud import storage
    import json
    
    # Initialize client
    client = storage.Client()
    bucket = client.bucket('custom-security-analytics-logs')
    
    # Upload log file
    blob = bucket.blob('security-events/2025/01/15/events.json')
    
    # Write NDJSON data
    events = [
        {"timestamp": "2025-01-15T10:00:00Z", "event_type": "login"},
        {"timestamp": "2025-01-15T10:05:00Z", "event_type": "logout"}
    ]
    
    ndjson_data = '\n'.join([json.dumps(event) for event in events]) + '\n'
    blob.upload_from_string(ndjson_data, content_type='application/x-ndjson')
    

获取 Google SecOps 服务账号

  1. 前往 SIEM 设置 > Feed。
  2. 点击添加新 Feed。
  3. 点击配置单个 Feed。
  4. 在 Feed 名称字段中,输入 Feed 的名称(例如 custom-security-analytics-gcs)。
  5. 选择 Google Cloud Storage V2 作为来源类型。
  6. 选择自定义 Security 数据分析 作为日志类型。
  7. 点击获取服务账号。
  8. 系统会显示一个唯一的服务账号电子邮件地址,例如:

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
    
  9. 复制此邮箱,以便在下一步中使用。

授予 IAM 权限

  1. 前往 Cloud Storage > 存储桶。
  2. 点击您的存储桶名称。
  3. 前往权限标签页。
  4. 点击授予访问权限。
  5. 提供以下配置详细信息:
    • 添加主账号:粘贴 Google SecOps 服务账号电子邮件地址。
    • 分配角色:选择 Storage Object Viewer。
  6. 点击保存。

为 GCS 配置 Google SecOps Feed

  1. 从 Feed 创建页面继续操作(或前往 SIEM 设置 > Feed > 添加新 Feed)。
  2. 点击下一步。
  3. 为以下输入参数指定值:

    • 存储桶网址:输入带有前缀路径的 GCS 存储桶 URI:

      gs://custom-security-analytics-logs/security-events/
      
    • 来源删除选项:根据您的偏好选择删除选项:

      • 永不:转移后永不删除任何文件(建议用于测试)。
      • 删除已转移的文件:在成功转移后删除文件。
      • 删除已转移的文件和空目录:成功转移后删除文件和空目录。
    • 文件存在时间上限:包含在过去指定天数内修改过的文件(默认值为 180 天)。

    • 资源命名空间:资源命名空间。

    • 注入标签:要应用于此 Feed 中事件的标签。

  4. 点击下一步。

  5. 在最终确定界面中查看新的 Feed 配置,然后点击提交。

选项 4:Azure Blob Storage V2 提取(批量导出)

如果您的自定义安全应用可以将日志导出到 Azure Blob Storage,请使用此方法。

创建 Azure 存储账号

  1. 在 Azure 门户中,搜索存储账号。
  2. 点击 + 创建。
  3. 提供以下配置详细信息:

    设置 值
    订阅 选择您的 Azure 订阅
    资源组 选择现有组或创建新组
    存储账号名称 输入一个唯一名称(例如 customsecuritylogs)
    区域 选择区域(例如 East US)
    性能 标准(推荐)
    冗余 GRS(地理位置冗余存储)或 LRS(本地冗余存储)
  4. 点击检查 + 创建。

  5. 查看账号概览,然后点击创建。

  6. 等待部署完成。

获取存储账号凭据

  1. 前往您刚刚创建的存储账号。
  2. 在左侧导航栏中,选择安全性 + 网络下的访问密钥。
  3. 点击显示密钥。
  4. 复制并保存以下内容以供日后使用:
    • 存储账号名称:customsecuritylogs
    • 密钥 1 或密钥 2:共享访问密钥

创建 Blob 容器

  1. 在同一存储账号中,从左侧导航栏中选择容器。
  2. 点击 + Container。
  3. 提供以下配置详细信息:
    • 名称:输入容器名称(例如 security-events)
    • 公开访问权限级别:选择私密(不允许匿名访问)
  4. 点击创建。

配置应用以导出到 Azure Blob

使用以下方法之一,将自定义安全应用配置为将日志文件写入 Azure Blob 容器:

  • Azure CLI:使用 az storage blob upload
  • Azure SDK:使用适用于您编程语言的客户端库
  • AzCopy:使用 AzCopy 命令行工具

示例:

  • 使用 Azure CLI 的示例:

    az storage blob upload \
        --account-name customsecuritylogs \
        --container-name security-events \
        --name logs/2025/01/15/events.json \
        --file /path/to/events.json \
        --account-key <YOUR_ACCESS_KEY>
    
  • 使用 Python SDK 的示例:

    from azure.storage.blob import BlobServiceClient
    import json
    
    # Initialize client
    connection_string = "DefaultEndpointsProtocol=https;AccountName=customsecuritylogs;AccountKey=<YOUR_KEY>;EndpointSuffix=core.windows.net"
    blob_service_client = BlobServiceClient.from_connection_string(connection_string)
    
    # Get container client
    container_client = blob_service_client.get_container_client("security-events")
    
    # Upload log file
    blob_client = container_client.get_blob_client("logs/2025/01/15/events.json")
    
    # Write NDJSON data
    events = [
        {"timestamp": "2025-01-15T10:00:00Z", "event_type": "login"},
        {"timestamp": "2025-01-15T10:05:00Z", "event_type": "logout"}
    ]
    
    ndjson_data = '\n'.join([json.dumps(event) for event in events]) + '\n'
    blob_client.upload_blob(ndjson_data, overwrite=True)
    

为 Azure Blob 配置 Google SecOps Feed

  1. 前往 SIEM 设置 > Feed。
  2. 点击添加新 Feed。
  3. 在下一页上,点击配置单个 Feed。
  4. 在 Feed 名称字段中,输入 Feed 的名称(例如 custom-security-analytics-azure)。
  5. 选择 Microsoft Azure Blob Storage V2 作为来源类型。
  6. 选择自定义安全数据分析作为日志类型。
  7. 点击下一步。
  8. 为以下输入参数指定值:

    • Azure URI:输入包含容器路径的 Blob 服务端点网址:

      https://customsecuritylogs.blob.core.windows.net/security-events/
      
    • 来源删除选项:根据您的偏好选择删除选项:

      • 永不:转移后永不删除任何文件。
      • 删除已转移的文件:在成功转移后删除文件。
      • 删除已转移的文件和空目录:成功转移后删除文件和空目录。
    • 文件存在时间上限:包含在过去指定天数内修改过的文件(默认值为 180 天)。

    • 共享密钥:输入存储账号中的共享密钥值(访问密钥)。

    • 资源命名空间:资源命名空间。

    • 注入标签:要应用于此 Feed 中事件的标签。

  9. 点击下一步。

  10. 在最终确定界面中查看新的新闻信息流配置,然后点击提交。

方法 5:Amazon Data Firehose 提取(实时流式传输)

当您的自定义安全应用将日志写入 Amazon CloudWatch Logs,并且您需要将日志实时流式传输到 Google SecOps 时,请使用此方法。

创建 Google SecOps Firehose Feed

  1. 前往 SIEM 设置 > Feed。
  2. 点击添加新 Feed。
  3. 在 Feed 名称字段中,输入 Feed 的名称(例如 custom-security-analytics-firehose)。
  4. 选择 Amazon Data Firehose 作为来源类型。
  5. 选择自定义安全数据分析作为日志类型。
  6. 点击下一步。
  7. 为以下输入参数指定值:
    • 拆分定界符:可选:输入 \n 以拆分以换行符分隔的日志。
    • 资源命名空间:资源命名空间。
    • 注入标签:要应用于此 Feed 中事件的标签。
  8. 点击下一步。
  9. 检查 Feed 配置,然后点击提交。
  10. 点击生成密钥,生成用于对此 Feed 进行身份验证的密钥。
  11. 复制并保存密钥,因为您无法再次查看此密钥。
  12. 点击详细信息标签页。
  13. 从端点信息字段中复制 Feed 端点网址。
  14. 点击完成。

创建 Google Cloud API 密钥

  1. 前往 Google Cloud 控制台的凭据页面:https://console.cloud.google.com/apis/credentials
  2. 点击创建凭据,然后选择 API 密钥。
  3. 点击修改 API 密钥以限制密钥。
  4. 在 API 限制下,选择限制密钥。
  5. 搜索并选择 Google SecOps API。
  6. 点击保存。
  7. 复制并保存 API 密钥。

构建端点网址

  1. 以以下格式将 API 密钥附加到 Feed 端点网址:

    <FEED_ENDPOINT_URL>?key=<API_KEY>
    

    示例:

    https://malachiteingestion-pa.googleapis.com/v2/unstructuredlogentries:batchCreate?key=AIzaSyD...
    
  2. 保存此完整网址,以在下一步中使用。

为 Firehose 创建 IAM 政策

  1. 在 AWS 控制台中,依次前往 IAM > 政策 > 创建政策 > JSON 标签页。
  2. 粘贴以下政策 JSON:

    {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Effect": "Allow",
                "Action": [
                    "firehose:PutRecord",
                    "firehose:PutRecordBatch"
                ],
                "Resource": "arn:aws:firehose:<REGION>:<ACCOUNT_ID>:deliverystream/CustomSecurityToChronicle"
            }
        ]
    }
    
  3. 替换以下内容:

    • <REGION>:您的 AWS 区域(例如 us-east-1)。
    • <ACCOUNT_ID>:您的 AWS 账号 ID(12 位数字)。
  4. 将政策命名为 CloudWatchLogsToFirehosePolicy。

  5. 点击创建政策。

为 CloudWatch Logs 创建 IAM 角色

  1. 依次前往 IAM > 角色 > 创建角色。
  2. 选择 Custom trust policy(自定义信任政策),然后粘贴:

    {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Effect": "Allow",
                "Principal": {
                    "Service": "logs.<REGION>.amazonaws.com"
                },
                "Action": "sts:AssumeRole"
            }
        ]
    }
    
  3. 将 <REGION> 替换为您的 AWS 区域。

  4. 点击下一步。

  5. 搜索并选择您在上一步中创建的政策 CloudWatchLogsToFirehosePolicy。

  6. 点击下一步。

  7. 将角色命名为 CloudWatchLogsToFirehoseRole。

  8. 点击 Create role。

创建 Kinesis Data Firehose 传送流

  1. 在 AWS 控制台中,依次前往 Kinesis > Data Firehose > Create delivery stream(创建传送流)。
  2. 提供以下配置详细信息:

    • 来源和目的地:

      • 来源:选择直接 PUT 或其他来源
      • 目的地:选择 HTTP 端点
    • 交付数据流名称:

      • 交付数据流名称:输入 CustomSecurityToChronicle
    • HTTP 端点目标:

      • HTTP 端点网址:输入您之前构建的完整端点网址(Feed 端点 + API 密钥)
      • 内容编码:选择 GZIP(建议选择此项以节省带宽)
    • 自定义 HTTP 标头:

      • 点击添加自定义 HTTP 标头
      • 标头名称:输入 X-Goog-Chronicle-Auth
      • 标头值:输入您在上一步中保存的 Secret 密钥
    • 备份设置:

      • Amazon S3 中的源记录备份:选择仅备份失败的数据(推荐)
      • S3 存储桶:选择现有存储桶或创建新存储桶来存储失败的记录
    • 缓冲提示:

      • 缓冲区空间:输入 1 MiB(HTTP 端点的最小值)
      • 缓冲间隔:输入 60 秒
    • 重试时长:

      • 重试时长:输入 300 秒(5 分钟)
  3. 点击创建传送数据流。

  4. 等待传送数据流状态变为有效(1-2 分钟)。

将 CloudWatch 日志组订阅到 Firehose

  1. 在 AWS 控制台中,依次前往 CloudWatch > 日志 > 日志组。
  2. 选择包含自定义安全分析日志的目标日志组。
  3. 点击订阅过滤条件标签页。
  4. 点击创建 > 创建 Amazon Kinesis Data Firehose 订阅过滤器。
  5. 提供以下配置详细信息:
    • 目标:选择交付数据流 CustomSecurityToChronicle。
    • 授予权限:选择角色 CloudWatchLogsToFirehoseRole。
    • 订阅过滤条件名称:输入 CustomSecurityToChronicle。
    • 日志格式:选择其他(Google SecOps 会处理解析)。
    • 订阅过滤模式:留空可发送所有事件,也可输入过滤模式以仅发送特定事件。
  6. 点击开始直播。
  7. 日志现在将通过 Firehose 实时流式传输到 Google SecOps。

创建自定义解析器(可选)

以非结构化日志的形式注入自定义安全数据后,您可以创建自定义解析器,将数据标准化为 UDM 格式,以便更好地进行搜索和检测。

何时创建自定义解析器

在以下情况下,请创建自定义解析器:

  • 您需要从自定义日志格式中提取特定字段
  • 您想在自定义数据上启用 UDM 搜索
  • 您需要将自定义字段映射到标准 UDM 字段,以用于检测规则
  • 您希望通过为特定字段编入索引来提高搜索性能

创建自定义日志类型

  1. 前往 SIEM 设置 > 可用的日志类型。
  2. 点击请求日志类型。
  3. 在创建自定义日志类型或请求预建日志类型部分下,选择创建自定义日志类型。
  4. 请提供以下信息:
    • 日志类型名称:输入描述性名称(例如 CUSTOM_SECURITY_ANALYTICS)。
    • 说明:输入日志类型的说明。
    • 日志示例:以原始格式粘贴 5-10 个日志条目示例。
  5. 点击提交。
  6. 自定义日志类型将在大约 10 分钟后可用。

创建自定义解析器

  1. 前往 SIEM 设置 > 解析器。
  2. 点击创建解析器。
  3. 选择自定义解析器。
  4. 请提供以下信息:
    • 解析器名称:输入描述性名称
    • 日志类型:选择您的自定义日志类型(例如 CUSTOM_SECURITY_ANALYTICS)
    • 解析器代码:使用 Google SecOps 的解析器配置语言输入解析器配置
  5. 使用示例日志测试解析器。
  6. 点击提交以激活解析器。

解析器配置示例

  • 对于自定义 JSON 日志格式:

    {
        "timestamp": "2025-01-15T10:30:00Z",
        "event_type": "authentication",
        "user": "john.doe@example.com",
        "action": "login",
        "result": "success",
        "source_ip": "203.0.113.45"
    }
    

    解析器配置示例:

    filter {
        json {
            fields {
                timestamp: timestamp
                event_type: event_type
                user: user
                action: action
                result: result
                source_ip: source_ip
            }
        }
    }
    
    event {
        $e.metadata.event_timestamp.seconds = parseTimestamp(timestamp, "yyyy-MM-dd'T'HH:mm:ss'Z'")
        $e.metadata.event_type = "USER_LOGIN"
        $e.principal.user.email_addresses = user
        $e.target.ip = source_ip
        $e.security_result.action = if(result == "success", "ALLOW", "BLOCK")
    }
    

    如需详细了解如何创建自定义解析器,请参阅管理预建解析器和自定义解析器。

验证数据注入

配置 Feed 后,请验证数据是否已成功提取。

查看 Feed 状态

  1. 前往 SIEM 设置 > Feed。
  2. 在列表中找到您的 Feed。
  3. 查看状态列:
    • 有效:Feed 正在运行并注入数据
    • 错误:Feed 遇到错误(点击可查看详情)
    • 已暂停:Feed 已暂停

搜索注入的日志

  1. 依次前往搜索 > 原始日志扫描。
  2. 输入搜索查询内容以查找自定义日志:

    metadata.log_type = "CUSTOM_SECURITY_DATA_ANALYTICS"
    
  3. 如有需要,请调整时间范围。

  4. 点击搜索。

  5. 验证日志是否显示在结果中。

监控 Feed 指标

  1. 前往 SIEM 设置 > Feed。
  2. 点击您的 Feed 名称。
  3. 前往指标标签页。
  4. 查看以下指标:
    • 注入的事件数:注入的事件总数
    • 注入的字节数:注入的数据总量
    • 注入速率:每秒事件数
    • 错误:提取错误数量

问题排查

Webhook 注入问题

  • 问题:HTTP 401 未经授权

    • 原因:API 密钥或密钥无效
    • 解决方案:验证 API 密钥和密钥是否正确且未过期
  • 问题:HTTP 403 Forbidden

    • 原因:API 密钥没有 Chronicle API 权限
    • 解决方案:修改 API 密钥,并确保在“API 限制”下选择了 Chronicle API
  • 问题:HTTP 400 错误请求

    • 原因:请求格式或载荷无效
    • 解决方案:验证 Content-Type 标头是否已设置为 application/json,以及载荷是否为有效的 JSON

S3/GCS/Azure Blob 注入问题

  • 问题:未提取任何数据

    • 原因:存储桶 URI 不正确或缺少权限
    • 解决方案:验证存储桶 URI 是否包含尾部斜杠,以及服务账号是否具有 Storage Object Viewer 角色
  • 问题:文件在提取后未被删除

    • 原因:服务账号没有删除权限
    • 解决方案:授予 Storage Object Admin 角色,而不是 Storage Object Viewer 角色
  • 问题:旧文件未提取

    • 原因:“文件存在时间上限”设置排除了旧文件
    • 解决方案:增加 Feed 配置中的“最长文件存在时间”值

Firehose 提取问题

  • 问题:传送流显示错误

    • 原因:端点网址或身份验证无效
    • 解决方案:验证端点网址是否包含 API 密钥参数,以及 X-Goog-Chronicle-Auth 标头是否包含正确的密钥
  • 问题:数据新鲜度指标较高

    • 原因:Firehose 受到限制或出现递送失败的情况
    • 解决方案:检查 CloudWatch 指标,了解 ThrottledRecords 和 DeliveryToHTTP.Success 速率
  • 问题:没有从 CloudWatch 流式传输的日志

    • 原因:未配置订阅过滤条件或 IAM 角色缺少权限
    • 解决方案:验证订阅过滤条件是否有效,以及 IAM 角色是否具有 firehose:PutRecord 权限

UDM 映射表

自定义安全数据分析日志以非结构化数据的形式提取。如需启用 UDM 字段映射,请按照上文“创建自定义解析器”部分中的说明创建自定义解析器。

创建自定义解析器后,UDM 字段将根据您的解析器配置进行填充。安全分析数据的常见 UDM 字段包括:

UDM 字段 说明
metadata.event_timestamp 事件时间戳
metadata.event_type 事件类型(例如 USER_LOGIN、FILE_ACCESS)
principal.user.email_addresses 用户电子邮件地址
principal.ip 来源 IP 地址
target.resource.name 目标资源名称
security_result.action 安全操作(ALLOW、BLOCK 等)
security_result.severity 事件严重程度

如需查看 UDM 字段的完整列表,请参阅 UDM 字段参考。

更新日志

查看相应解析器的更新日志

需要更多帮助?获得社区成员和 Google SecOps 专业人士的解答。