收集文件扫描框架日志

解析器版本：3.0

支持的平台：

Google SecOps SIEM

本文档介绍了如何使用 Google Cloud Storage V2 将文件扫描框架日志注入到 Google Security Operations。

文件扫描框架 (FSF) 是 Emerson Electric Co. 开发的一款开源模块化递归文件扫描解决方案。FSF 使用客户端-服务器架构来分析文件并生成详细的 JSON 扫描结果，包括文件元数据、YARA 签名匹配项、提取的子对象和模块特定的元数据。

准备工作

请确保您满足以下前提条件：

• Google SecOps 实例
• 已启用 Cloud Storage API 的 GCP 项目
• 创建和管理 GCS 存储分区的权限
• 管理 GCS 存储分区的 IAM 政策的权限
• 已部署的 FSF 服务器实例，对日志目录具有写入权限
• 对 FSF 服务器主机的根访问权限或 sudo 访问权限

创建 Google Cloud Storage 存储桶

1. 前往 Google Cloud 控制台。
2. 选择您的项目或创建新项目。
3. 在导航菜单中，依次前往 Cloud Storage > 存储分区。
4. 点击创建存储分区。

5. 提供以下配置详细信息：

   设置	值
   为存储桶命名	输入一个全局唯一的名称（例如 fsf-logs-secops）。
   位置类型	请根据您的需求（区域级、双区域级、多区域级）进行选择。
   位置	选择相应营业地点（例如 us-central1）。
   存储类别	标准（建议用于经常访问的日志）。
   访问权限控制	统一（推荐）。
   保护工具	可选：启用对象版本控制或保留政策。

6. 点击创建。

配置 FSF 日志输出目录

FSF 将 JSON 扫描结果写入可配置的日志目录。为 Google SecOps 注入配置专用目录。

1. 通过 SSH 连接到 FSF 服务器主机。

2. 打开 FSF 服务器配置文件：

   sudo nano /opt/fsf/fsf-server/conf/config.py
   

3. 找到 SCANNER_CONFIG 字典。

4. 将 LOG_PATH 参数更新为专用目录：

   SCANNER_CONFIG = {
       'LOG_PATH': '/var/log/fsf',
       'YARA_PATH': '/opt/fsf/fsf-server/yara/rules.yara',
       'PID_PATH': '/tmp/scanner.pid',
       'EXPORT_PATH': '/tmp',
       'TIMEOUT': 60,
       'MAX_DEPTH': 10
   }
   

5. 保存并关闭文件。

6. 创建具有适当权限的日志目录：

   sudo mkdir -p /var/log/fsf
   sudo chown -R fsf:fsf /var/log/fsf
   sudo chmod 755 /var/log/fsf
   

7. 重启 FSF 服务器以应用更改：

   sudo systemctl restart fsf
   

8. 验证 FSF 是否正在将日志写入新目录：

   ls -lh /var/log/fsf/
   

注意： FSF 会将每个 JSON 对象写入日志文件中的一行。每次扫描结果都会作为新行附加到日志文件中。

安装和配置 Fluentd

Fluentd 将跟踪 FSF 日志文件，并将它们发送到 Google Cloud Storage。

安装 Fluentd

1. 在 FSF 服务器主机上，安装 Fluentd (td-agent)：

   curl -fsSL https://toolbelt.treasuredata.com/sh/install-ubuntu-jammy-td-agent4.sh | sh
   

   注意： 对于其他操作系统，请参阅 Fluentd 安装指南。

2. 安装 GCS 输出插件：

   sudo td-agent-gem install fluent-plugin-gcs
   

3. 验证插件安装：

   td-agent-gem list | grep fluent-plugin-gcs
   

为 Fluentd 创建 GCP 服务账号

1. 在 GCP 控制台中，依次前往 IAM 和管理 > 服务账号。
2. 点击创建服务账号。
3. 提供以下配置详细信息：
   • 服务账号名称：输入 fsf-fluentd-shipper。
   • 服务账号说明：输入 Service account for Fluentd to ship FSF logs to GCS。
4. 点击创建并继续。
5. 在向此服务账号授予对项目的访问权限部分：
   1. 点击选择角色。
   2. 搜索并选择 Storage Object Admin。
6. 点击继续。
7. 点击完成。

创建服务账号密钥

1. 在服务账号列表中，点击相应服务账号 (fsf-fluentd-shipper)。
2. 前往密钥标签页。
3. 点击添加密钥 > 创建新密钥。
4. 选择 JSON 作为密钥类型。
5. 点击创建。
6. JSON 密钥文件将下载到您的计算机。

7. 将密钥文件转移到 FSF 服务器主机：

   scp /path/to/downloaded-key.json user@fsf-server:/etc/td-agent/gcp-key.json
   

8. 为密钥文件设置适当的权限：

   sudo chown td-agent:td-agent /etc/td-agent/gcp-key.json
   sudo chmod 600 /etc/td-agent/gcp-key.json
   

授予对 GCS 存储桶的 IAM 权限

1. 前往 Cloud Storage > 存储分区。
2. 点击相应存储桶的名称 (fsf-logs-secops)。
3. 前往权限标签页。
4. 点击授予访问权限。
5. 提供以下配置详细信息：
   • 添加主账号：输入服务账号电子邮件地址（例如 fsf-fluentd-shipper@PROJECT_ID.iam.gserviceaccount.com）。
   • 分配角色：选择 Storage Object Admin。
6. 点击保存。

配置 Fluentd

1. 在 FSF 服务器主机上，创建 Fluentd 配置文件：

   sudo nano /etc/td-agent/td-agent.conf
   

2. 添加以下配置：

   # Tail FSF JSON logs
   <source>
       @type tail
       path /var/log/fsf/*.log
       pos_file /var/log/td-agent/fsf.log.pos
       tag fsf.scan
       read_from_head true
       <parse>
           @type json
           time_key timestamp
           time_format %Y-%m-%dT%H:%M:%S.%L%z
       </parse>
   </source>
   
   # Ship to Google Cloud Storage
   <match fsf.scan>
       @type gcs
       project YOUR_GCP_PROJECT_ID
       keyfile /etc/td-agent/gcp-key.json
       bucket fsf-logs-secops
       object_key_format %{path}%{time_slice}_%{index}.%{file_extension}
       path fsf-logs/
       <buffer tag,time>
           @type file
           path /var/log/td-agent/buffer/gcs
           timekey 3600
           timekey_wait 10m
           timekey_use_utc true
           chunk_limit_size 10MB
       </buffer>
       <format>
           @type json
       </format>
       store_as json
       auto_create_bucket false
   </match>
   

3. 将 YOUR_GCP_PROJECT_ID 替换为您的实际 GCP 项目 ID。

4. 保存并关闭文件。

5. 创建缓冲区目录：

   sudo mkdir -p /var/log/td-agent/buffer/gcs
   sudo chown -R td-agent:td-agent /var/log/td-agent/buffer
   

6. 重启 Fluentd 以应用配置：

   sudo systemctl restart td-agent
   

7. 启用 Fluentd 以在启动时启动：

   sudo systemctl enable td-agent
   

8. 验证 Fluentd 是否正在运行：

   sudo systemctl status td-agent
   

验证日志传送

1. 检查 Fluentd 日志是否存在错误：

   sudo tail -f /var/log/td-agent/td-agent.log
   

2. 触发测试 FSF 扫描以生成日志：

   echo "test content" > /tmp/test.txt
   /opt/fsf/fsf-client/fsf_client.py /tmp/test.txt --suppress-report
   

3. 等待 1-2 分钟，让 Fluentd 处理并发送日志。

4. 在 GCP 控制台中，前往 Cloud Storage > 存储分区。

5. 点击相应存储桶的名称 (fsf-logs-secops)。

6. 找到 fsf-logs/ 前缀。

7. 验证是否正在创建带有时间戳的 JSON 文件。

8. 下载并检查文件，确认其中包含 JSON 格式的 FSF 扫描结果。

注意： 默认情况下，Fluentd 会缓冲日志，并每小时将日志发送到 GCS。调整配置中的 timekey 参数，即可更改发货频率。

检索 Google SecOps 服务账号

Google SecOps 使用唯一的服务账号从您的 GCS 存储桶中读取数据。您必须授予此服务账号对您的存储桶的访问权限。

获取服务账号电子邮件地址

1. 依次前往 SIEM 设置 > Feed。
2. 点击添加新 Feed。
3. 点击配置单个 Feed。
4. 在 Feed 名称字段中，输入 Feed 的名称（例如 FSF File Scanning Logs）。
5. 选择 Google Cloud Storage V2 作为来源类型。
6. 选择文件扫描框架作为日志类型。

7. 点击获取服务账号。系统会显示一个唯一的服务账号电子邮件地址，例如：

   [e-mail]
   

8. 复制该电子邮件地址，以便在下一步中使用。

   注意： 每个 Google SecOps 实例都有一个唯一的服务账号。请勿使用其他文档或示例中的服务账号。

9. 点击下一步。

10. 为以下输入参数指定值：

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

      gs://fsf-logs-secops/fsf-logs/
      

      注意： 请务必在 URI 末尾添加尾随斜杠 (/)。

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

      • 永不：转移后永不删除任何文件（建议用于测试）。
      • 删除已转移的文件：在成功转移后删除文件。

      • 删除已转移的文件和空目录：成功转移后删除文件和空目录。

        注意 ：如果您选择删除选项，服务账号必须具有 Storage Object Admin 角色，而不是 Storage Object Viewer 角色。相应地更新 IAM 权限。

    • 文件存在时间上限：包含在过去指定天数内修改的文件（默认值为 180 天）。

    • 资产命名空间：资产命名空间。

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

11. 点击下一步。

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

向 Google SecOps 服务账号授予 IAM 权限

Google SecOps 服务账号需要您的 GCS 存储桶的 Storage Object Viewer 角色。

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

注意： 如果您打算使用“删除已转移的文件”或“删除已转移的文件和空目录”删除选项，请授予存储对象管理员角色，而不是存储对象查看者角色。

验证注入

1. 等待 10-15 分钟，直到初始提取完成。
2. 在 Google SecOps 中，依次前往 SIEM 设置 > Feed。
3. 找到相应 Feed (FSF File Scanning Logs)。
4. 验证状态是否显示为有效。
5. 点击 Feed 名称可查看提取指标。
6. 验证已提取的事件数量是否在增加。
7. 前往 Google SecOps 中的搜索。

8. 运行搜索查询，验证 FSF 日志是否正在被注入：

   metadata.log_type = "FILE_SCANNING_FRAMEWORK"
   

9. 验证 FSF 扫描结果是否显示在搜索结果中。

问题排查

GCS 中未显示任何日志

• 验证 FSF 是否正在将日志写入 /var/log/fsf/：

  ls -lh /var/log/fsf/
  tail -f /var/log/fsf/*.log
  

• 检查 Fluentd 日志是否存在错误：

  sudo tail -f /var/log/td-agent/td-agent.log
  

• 验证 GCP 服务账号密钥是否有效且具有正确的权限。

• 检查 Fluentd 配置中的存储桶名称是否与实际存储桶名称一致。

Fluentd 权限错误

• 验证服务账号 (fsf-fluentd-shipper) 是否具有对相应存储桶的 Storage Object Admin 角色。
• 检查 Fluentd 配置中的密钥路径是否正确。

• 验证密钥文件是否具有正确的所有权和权限：

  ls -l /etc/td-agent/gcp-key.json
  

Google SecOps 未提取日志

• 验证 Google SecOps 服务账号是否对相应存储桶具有 Storage Object Viewer 角色。
• 检查 Feed 配置中的存储桶 URI 是否正确，并包含结尾斜杠。
• 验证 GCS 存储桶中是否存在指定前缀路径下的文件。
• 在 SIEM 设置 > Feed 中查看 Feed 状态，看看是否有错误消息。

FSF 日志格式不正确

• 验证 FSF 是否已配置为写入 JSON 输出（默认行为）。
• 检查 Fluentd <parse> 部分是否已配置为 @type json。

• 手动检查日志文件，验证其是否包含有效的 JSON：

  head -n 1 /var/log/fsf/*.log | jq .
  

UDM 映射表

日志字段	UDM 映射	逻辑
CompressType_label、compressed_parents	about.labels	从 CompressType_label（键为“压缩类型”，如果消息包含“压缩类型”，则值为 Object.EXTRACT_ZIP.Object_0.Compress Type）和 compressed_parents（键为“压缩的父文件”，串联自 Object.EXTRACT_ZIP.Object_0.META_VT_CACHE.vt_data.additional_info.compressed_parents）合并
Object.EXTRACT_EMBEDDED.Object_0.META_BASIC_INFO.MD5、Object.EXTRACT_ZIP.Object_0.META_BASIC_INFO.MD5、Object.EXTRACT_SWF.META_BASIC_INFO.MD5、Object.EXTRACT_GZIP.META_BASIC_INFO.MD5、Object.EXTRACT_CAB.Object_0.META_BASIC_INFO.MD5	intermediary.file.md5	如果存在 EXTRACT_EMBEDDED，则为 Object.EXTRACT_EMBEDDED.Object_0.META_BASIC_INFO.MD5 中的值；如果存在 EXTRACT_ZIP，则为 Object.EXTRACT_ZIP.Object_0.META_BASIC_INFO.MD5 中的值；如果存在 EXTRACT_SWF，则为 Object.EXTRACT_SWF.META_BASIC_INFO.MD5 中的值；如果存在 EXTRACT_GZIP，则为 Object.EXTRACT_GZIP.META_BASIC_INFO.MD5 中的值；否则为 Object.EXTRACT_CAB.Object_0.META_BASIC_INFO.MD5 中的值
Object.EXTRACT_EMBEDDED.Object_0.Description	intermediary.file.mime_type	直接复制值
Object.EXTRACT_EMBEDDED.Object_0.META_BASIC_INFO.SHA1、Object.EXTRACT_ZIP.Object_0.META_BASIC_INFO.SHA1、Object.EXTRACT_SWF.META_BASIC_INFO.SHA1、Object.EXTRACT_GZIP.META_BASIC_INFO.SHA1、Object.EXTRACT_CAB.Object_0.META_BASIC_INFO.SHA1	intermediary.file.sha1	如果存在 EXTRACT_EMBEDDED，则为 Object.EXTRACT_EMBEDDED.Object_0.META_BASIC_INFO.SHA1 中的值；如果存在 EXTRACT_ZIP，则为 Object.EXTRACT_ZIP.Object_0.META_BASIC_INFO.SHA1 中的值；如果存在 EXTRACT_SWF，则为 Object.EXTRACT_SWF.META_BASIC_INFO.SHA1 中的值；如果存在 EXTRACT_GZIP，则为 Object.EXTRACT_GZIP.META_BASIC_INFO.SHA1 中的值；否则为 Object.EXTRACT_CAB.Object_0.META_BASIC_INFO.SHA1 中的值
Object.EXTRACT_EMBEDDED.Object_0.META_BASIC_INFO.SHA256、Object.EXTRACT_ZIP.Object_0.META_BASIC_INFO.SHA256、Object.EXTRACT_SWF.META_BASIC_INFO.SHA256、Object.EXTRACT_GZIP.META_BASIC_INFO.SHA256、Object.EXTRACT_CAB.Object_0.META_BASIC_INFO.SHA256	intermediary.file.sha256	如果存在 EXTRACT_EMBEDDED，则为 Object.EXTRACT_EMBEDDED.Object_0.META_BASIC_INFO.SHA256 中的值；否则，如果存在 EXTRACT_ZIP，则为 Object.EXTRACT_ZIP.Object_0.META_BASIC_INFO.SHA256 中的值；否则，如果存在 EXTRACT_SWF，则为 Object.EXTRACT_SWF.META_BASIC_INFO.SHA256 中的值；否则，如果存在 EXTRACT_GZIP，则为 Object.EXTRACT_GZIP.META_BASIC_INFO.SHA256 中的值；否则，为 Object.EXTRACT_CAB.Object_0.META_BASIC_INFO.SHA256 中的值
Object.EXTRACT_EMBEDDED.Object_0.META_BASIC_INFO.Size、Object.EXTRACT_ZIP.Object_0.META_BASIC_INFO.Size、Object.EXTRACT_SWF.META_BASIC_INFO.Size、Object.EXTRACT_GZIP.META_BASIC_INFO.Size、Object.EXTRACT_CAB.Object_0.META_BASIC_INFO.Size	intermediary.file.size	如果存在 EXTRACT_EMBEDDED，则为 Object.EXTRACT_EMBEDDED.Object_0.META_BASIC_INFO.Size 中的值；否则，如果存在 EXTRACT_ZIP，则为 Object.EXTRACT_ZIP.Object_0.META_BASIC_INFO.Size 中的值；否则，如果存在 EXTRACT_SWF，则为 Object.EXTRACT_SWF.META_BASIC_INFO.Size 中的值；否则，如果存在 EXTRACT_GZIP，则为 Object.EXTRACT_GZIP.META_BASIC_INFO.Size 中的值；否则，为 Object.EXTRACT_CAB.Object_0.META_BASIC_INFO.Size 中的值；已去除末尾的“.*”并转换为无符号整数
Object.EXTRACT_ZIP.Object_0.META_VT_CACHE.vt_data.scan_id	intermediary.resource.id	直接复制值
Object.EXTRACT_ZIP.Object_0.META_VT_CACHE.vt_data.permalink	intermediary.url	直接复制值
Object.META_EMERSON_INFO.results	intermediary.user.email_addresses	从结果数组中的 matched_email 合并
Summary.Observations	metadata.description	从数组中串联，使用“,”分隔符，移除了开头的逗号
扫描时间	metadata.event_timestamp	使用日期过滤器转换，格式为 yyyy-MM-dd HH:mm:ss
来源	metadata.event_type	如果 Source 不为空，则设置为“SCAN_FILE”，否则设置为“GENERIC_EVENT”
Object.META_VT_CACHE._id	metadata.product_log_id	直接复制值
result.ad_data.message	network.http.response_code	使用 grok 模式 INT 从 result.ad_data.message 中提取为整数
来源	principal.hostname	直接复制值
Object.META_EMERSON_INFO.result_summary, Object.EXTRACT_ZIP.Object_0.META_VT_CACHE.vt_data.verbose_msg	security_result.summary	如果存在，则设置为 Object.META_EMERSON_INFO.result_summary；否则设置为 Object.EXTRACT_ZIP.Object_0.META_VT_CACHE.vt_data.verbose_msg
文件名	target.file.full_path	直接复制值
Object.META_BASIC_INFO.MD5	target.file.md5	直接复制值
Summary.Yara	target.file.mime_type	从 Summary.Yara 的第一个索引中提取，转换为大写，如果存在 Yara，则移除“FT_”；否则，如果存在 EXTRACT_ZIP，则设置为“ZIP”；如果存在 EXTRACT_SWF，则设置为“SWF”；如果存在 EXTRACT_GZIP，则设置为“GZIP”；如果存在 EXTRACT_CAB，则设置为“CAB”
Object.META_BASIC_INFO.SHA1、Object.META_VT_CACHE.SHA1	target.file.sha1	如果 Object.META_BASIC_INFO.SHA1 不为空，则为该值；否则为 Object.META_VT_CACHE.SHA1
Object.META_BASIC_INFO.SHA256	target.file.sha256	直接复制值
Object.META_BASIC_INFO.Size	target.file.size	剥离了尾随的“.*”，并转换为无符号整数
	metadata.vendor_name	设置为“EMERSON”
	metadata.product_name	设置为“文件扫描框架”

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