收集 HCL BigFix 日志

支持的平台：

本文档介绍了如何使用 Google Cloud 通过 Cloud Run 函数将 HCL BigFix 日志注入到 Google Security Operations 中。

HCL BigFix 是一种端点管理平台，可提供统一的端点管理、安全和合规性功能。它使组织能够发现、管理和修复其整个基础架构中的端点，包括补丁管理、软件分发、配置管理和安全合规性。Cloud Run 函数使用会话相关性查询轮询 HCL BigFix REST API，将结果以 NDJSON 格式写入 Cloud Storage 存储桶，然后 Google SecOps 通过 Cloud Storage V2 Feed 提取这些结果。

准备工作

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

  • Google SecOps 实例
  • 启用了结算功能的 Google Cloud 项目
  • 具有主操作员权限的 HCL BigFix 控制台访问权限
  • HCL BigFix Server 9.5 版或更高版本
  • 在 BigFix 服务器上运行的 WebReports 服务
  • 从 Cloud Run 函数到 BigFix 服务器的网络连接（通过端口 52311 [默认 REST API 端口]）
  • Google Cloud 已启用的 API：Cloud Run functions、Cloud Storage、Cloud Scheduler、Pub/Sub 和 Identity and Access Management (IAM)

创建 Google Cloud 存储桶

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

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

    设置 值
    为存储桶命名 输入一个全局唯一的名称（例如 hcl-bigfix-logs）
    位置类型 根据您的需求进行选择（区域级、双区域、多区域）
    位置 选择离您的 Google SecOps 实例最近的位置（例如 us-central1）
    存储类别 标准（建议用于经常访问的日志）
    访问权限控制 均匀（推荐）
    保护工具 可选：启用对象版本控制或保留政策

  6. 点击创建。

收集 HCL BigFix API 凭据

如需使 Cloud Run 函数能够检索端点数据，您需要创建一个具有 REST API 访问权限的专用 BigFix 控制台操作员。

为 API 访问创建专用操作员

  1. 登录 BigFix 控制台。
  2. 依次前往工具 > 创建运算符。

  3. 在添加用户对话框中，提供以下信息：

    • 用户名：输入一个描述性用户名（例如 chronicle_api）。
    • 密码：输入安全系数高的密码。
    • 确认密码：重新输入密码。

  4. 点击确定。

配置操作员权限

创建运算符后，系统会自动打开 Console Operator 窗口。

  1. 点击详情标签页。
  2. 在概览部分中：
    • 选择始终允许此用户登录。
  3. 在权限部分，设置以下权限：
    • Can Submit Queries（能否提交查询）：设置为 Yes（是）。
    • 自定义内容：设置为是。
  4. 在 Interface Login Privileges 部分：
    • 可以使用 REST API：设置为是。
  5. 点击受管理的计算机标签页。
  6. 选择所有计算机，以允许操作员查询所有端点；也可以根据需要选择特定的计算机组。
  7. 点击保存更改。

记录 API 凭据

保存以下信息，以便配置 Cloud Run 函数环境变量：

  • 用户名：您创建的运营商用户名（例如 chronicle_api）
  • 密码：运营商密码
  • BigFix 服务器主机名：BigFix 服务器的完全限定域名或 IP 地址（例如 bigfix.company.com）

  • API 端口：REST API 端口（默认值为 52311）

测试 API 连接

  • 通过从具有 BigFix 服务器网络访问权限的机器运行以下测试查询，验证 API 凭据是否有效：

    curl -k -u "chronicle_api:YOUR_PASSWORD" \
  -X POST \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "relevance=(name of it, id of it) of bes computers" \
  "https://bigfix-server:52311/api/query"

    成功的响应会返回一个包含计算机名称和 ID 的 XML 文档。如果您收到身份验证错误，请验证在上一步中配置的运营商权限。

为 Cloud Run 函数创建服务账号

  1. 在 Google Cloud 控制台中，依次前往 IAM 和管理 > 服务账号。
  2. 点击创建服务账号。

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

    • 服务账号名称：输入 bigfix-cloud-run（或一个描述性名称）。
    • 服务账号说明：输入 Service account for HCL BigFix Cloud Run function to write logs to Cloud Storage。

  4. 点击创建并继续。

  5. 在向此服务账号授予对项目的访问权限部分中，添加以下角色：

    • Storage Object Admin（用于读取/写入 Cloud Storage 存储桶中的对象）。
    • Cloud Run Invoker（允许 Cloud Scheduler 调用该函数）。

  6. 点击继续。

  7. 点击完成。

创建 Pub/Sub 主题

Cloud Scheduler 通过 Pub/Sub 主题触发 Cloud Run 函数。

  1. 在 Google Cloud 控制台中，前往 Pub/Sub > 主题。
  2. 点击创建主题。
  3. 在主题 ID 字段中，输入 bigfix-ingestion-trigger。
  4. 保留默认设置。

  5. 点击创建。

创建 Cloud Run 函数

准备函数源文件

为 Cloud Run 函数部署创建以下两个文件。

  • requirements.txt

    functions-framework==3.*
google-cloud-storage==2.*
requests>=2.31.0

  • main.py

    import base64
import json
import logging
import os
import xml.etree.ElementTree as ET
from datetime import datetime, timezone, timedelta

import requests
import urllib3
from google.cloud import storage

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

GCS_BUCKET = os.environ["GCS_BUCKET"]
GCS_PREFIX = os.environ.get("GCS_PREFIX", "bigfix_logs")
STATE_KEY = os.environ.get("STATE_KEY", "bigfix_state.json")
BIGFIX_SERVER = os.environ["BIGFIX_SERVER"]
BIGFIX_PORT = os.environ.get("BIGFIX_PORT", "52311")
BIGFIX_USERNAME = os.environ["BIGFIX_USERNAME"]
BIGFIX_PASSWORD = os.environ["BIGFIX_PASSWORD"]
MAX_RECORDS = int(os.environ.get("MAX_RECORDS", "10000"))
LOOKBACK_HOURS = int(os.environ.get("LOOKBACK_HOURS", "24"))

BIGFIX_BASE_URL = f"https://{BIGFIX_SERVER}:{BIGFIX_PORT}/api"

RELEVANCE_QUERIES = {
    "computers": (
        '(id of it, name of it, operating system of it, '
        'last report time of it as string, ip address of it as string, '
        'root server of it) of bes computers'
    ),
    "actions": (
        '(id of it, name of it, name of issuer of it, '
        'time issued of it as string, state of it) of bes actions'
    ),
    "fixlets": (
        '(id of it, name of it, source severity of it, '
        'source of it, category of it) of bes fixlets '
        'whose (source severity of it is not "")'
    ),
}

def _get_auth_header():
    """Build Basic Auth header from credentials."""
    credentials = f"{BIGFIX_USERNAME}:{BIGFIX_PASSWORD}"
    encoded = base64.b64encode(credentials.encode("utf-8")).decode("utf-8")
    return {"Authorization": f"Basic {encoded}"}

def _query_bigfix(relevance, session=None):
    """Execute a session relevance query against the BigFix REST API."""
    url = f"{BIGFIX_BASE_URL}/query"
    headers = _get_auth_header()
    headers["Content-Type"] = "application/x-www-form-urlencoded"

    data = {"relevance": relevance}
    http = session or requests
    resp = http.post(url, headers=headers, data=data, verify=False, timeout=120)
    resp.raise_for_status()
    return resp.text

def _parse_query_xml(xml_text, query_name):
    """Parse BigFix XML response into a list of dicts."""
    root = ET.fromstring(xml_text)
    results = []

    for result_elem in root.iter("Result"):
        tuples = result_elem.findall("Tuple")
        if tuples:
            for tup in tuples:
                answers = [a.text or "" for a in tup.findall("Answer")]
                record = _map_answers(query_name, answers)
                record["_query"] = query_name
                record["_collection_time"] = (
                    datetime.now(timezone.utc).isoformat()
                )
                results.append(record)
        else:
            answers_elems = result_elem.findall("Answer")
            if answers_elems:
                answers = [a.text or "" for a in answers_elems]
                record = _map_answers(query_name, answers)
                record["_query"] = query_name
                record["_collection_time"] = (
                    datetime.now(timezone.utc).isoformat()
                )
                results.append(record)

    logger.info(
        "Query '%s' returned %d results.", query_name, len(results)
    )
    return results

def _map_answers(query_name, answers):
    """Map positional answers to named fields based on query type."""
    field_maps = {
        "computers": [
            "client_id", "client_name", "OS",
            "last_report_time", "ip_address", "root_server",
        ],
        "actions": [
            "action_id", "action_name", "issuer",
            "time_issued", "state",
        ],
        "fixlets": [
            "fixlet_id", "fixlet_name", "source_severity",
            "source", "category",
        ],
    }
    fields = field_maps.get(query_name, [])
    record = {}
    for i, value in enumerate(answers):
        key = fields[i] if i < len(fields) else f"field_{i}"
        record[key] = value
    return record

def _load_state(gcs_client):
    """Load the last run state from GCS."""
    bucket = gcs_client.bucket(GCS_BUCKET)
    blob = bucket.blob(f"{GCS_PREFIX}/{STATE_KEY}")
    if blob.exists():
        data = blob.download_as_text()
        return json.loads(data)
    return {}

def _save_state(gcs_client, state):
    """Persist run state to GCS."""
    bucket = gcs_client.bucket(GCS_BUCKET)
    blob = bucket.blob(f"{GCS_PREFIX}/{STATE_KEY}")
    blob.upload_from_string(
        json.dumps(state), content_type="application/json"
    )
    logger.info("State saved to gs://%s/%s/%s", GCS_BUCKET, GCS_PREFIX, STATE_KEY)

def _write_ndjson(gcs_client, records, timestamp_str):
    """Write records as NDJSON to GCS."""
    if not records:
        logger.info("No records to write.")
        return

    records = records[:MAX_RECORDS]
    ndjson_lines = "\n".join(json.dumps(r) for r in records)
    blob_name = f"{GCS_PREFIX}/{timestamp_str}/bigfix_events.ndjson"
    bucket = gcs_client.bucket(GCS_BUCKET)
    blob = bucket.blob(blob_name)
    blob.upload_from_string(ndjson_lines, content_type="application/x-ndjson")
    logger.info(
        "Wrote %d records to gs://%s/%s",
        len(records), GCS_BUCKET, blob_name,
    )

def main(event, context):
    """Cloud Run function entry point triggered by Pub/Sub."""
    logger.info("HCL BigFix ingestion function started.")

    gcs_client = storage.Client()
    state = _load_state(gcs_client)

    last_run = state.get("last_run")
    if last_run:
        logger.info("Last successful run: %s", last_run)

    now = datetime.now(timezone.utc)
    timestamp_str = now.strftime("%Y/%m/%d/%H%M%S")

    all_records = []
    session = requests.Session()

    for query_name, relevance in RELEVANCE_QUERIES.items():
        try:
            xml_text = _query_bigfix(relevance, session=session)
            records = _parse_query_xml(xml_text, query_name)
            all_records.extend(records)
        except requests.exceptions.RequestException as e:
            logger.error(
                "Failed to execute query '%s': %s", query_name, e
            )
        except ET.ParseError as e:
            logger.error(
                "Failed to parse XML for query '%s': %s", query_name, e
            )

    session.close()

    _write_ndjson(gcs_client, all_records, timestamp_str)

    state["last_run"] = now.isoformat()
    state["records_written"] = len(all_records)
    _save_state(gcs_client, state)

    logger.info(
        "HCL BigFix ingestion complete. %d total records written.",
        len(all_records),
    )
    return "OK"

部署 Cloud Run 函数

  1. 将这两个文件（main.py 和 requirements.txt）保存到本地目录（例如 bigfix-function/）中。
  2. 打开 Cloud Shell 或安装了 gcloud CLI 的终端。

  3. 运行以下命令以部署函数：

    gcloud functions deploy bigfix-to-gcs \
  --gen2 \
  --region=us-central1 \
  --runtime=python312 \
  --trigger-topic=bigfix-ingestion-trigger \
  --entry-point=main \
  --memory=512MB \
  --timeout=540s \
  --service-account=bigfix-cloud-run@PROJECT_ID.iam.gserviceaccount.com \
  --set-env-vars="GCS_BUCKET=hcl-bigfix-logs,GCS_PREFIX=bigfix_logs,STATE_KEY=bigfix_state.json,BIGFIX_SERVER=bigfix.company.com,BIGFIX_PORT=52311,BIGFIX_USERNAME=chronicle_api,BIGFIX_PASSWORD=YOUR_PASSWORD,MAX_RECORDS=10000,LOOKBACK_HOURS=24"

  4. 替换以下占位值：

    • PROJECT_ID：您的 Google Cloud 项目 ID。
    • hcl-bigfix-logs：您的 Cloud Storage 存储桶名称。
    • bigfix.company.com：BigFix 服务器的主机名或 IP 地址。
    • chronicle_api：您的 BigFix 操作员用户名。
    • YOUR_PASSWORD：您的 BigFix 操作员密码。

  5. 通过检查函数状态来验证部署：

    gcloud functions describe bigfix-to-gcs --region=us-central1 --gen2

环境变量参考文档

变量 必需 默认 说明
GCS_BUCKET 是 用于存储 NDJSON 输出的 Cloud Storage 存储桶名称
GCS_PREFIX 否 bigfix_logs 存储分区内的对象前缀（文件夹路径）
STATE_KEY 否 bigfix_state.json 前缀内状态文件的 blob 名称
BIGFIX_SERVER 是 BigFix 服务器主机名或 IP 地址
BIGFIX_PORT 否 52311 BigFix REST API 端口
BIGFIX_USERNAME 是 BigFix 操作员用户名
BIGFIX_PASSWORD 是 BigFix 操作员密码
MAX_RECORDS 否 10000 每次执行要写入的记录数上限
LOOKBACK_HOURS 否 24 初始收集时回溯的小时数

创建 Cloud Scheduler 作业

Cloud Scheduler 通过 Pub/Sub 主题定期触发 Cloud Run 函数。

  1. 在 Google Cloud 控制台中，前往 Cloud Scheduler。
  2. 点击创建作业。
  3. 提供以下配置详细信息：
    • 名称：输入 bigfix-ingestion-schedule。
    • 区域：选择与 Cloud Run 函数相同的区域（例如 us-central1）。
    • 频率：输入 0 */4 * * *（每 4 小时运行一次）。根据数据量和新鲜度要求调整时间表。
    • 时区：选择您偏好的时区。
  4. 点击继续。
  5. 在配置执行部分中：
    • 目标类型：选择 Pub/Sub。
    • 主题：选择 bigfix-ingestion-trigger。
    • 消息正文：输入 {"run": true}。
  6. 点击继续。
  7. 在配置可选设置部分中：
    • 重试次数上限：输入 3。
    • 退避时长下限：输入 5s。
    • 退避时长上限：输入 60s。
  8. 点击创建。

  9. 如需立即运行测试，请点击作业名称旁边的三点状图标 (...)，然后选择强制运行。

检索 Google SecOps 服务账号并配置 Feed

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

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

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

  7. 点击获取服务账号。

    系统会显示一个唯一的服务账号电子邮件地址，例如：

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com

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

  9. 点击下一步。

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

    • 存储桶网址：输入 Cloud Storage 存储桶 URI：

      gs://hcl-bigfix-logs/bigfix_logs/
      • 将 hcl-bigfix-logs 替换为您的 Cloud Storage 存储桶名称。
      • 将 bigfix_logs 替换为您配置的 GCS_PREFIX 值。

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

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

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

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

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

    • 提取标签：要应用于相应 Feed 中事件的标签。

  11. 点击下一步。

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

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

Google SecOps 服务账号需要对您的 Cloud Storage 存储桶具有 Storage Object Viewer 角色。

  1. 前往 Cloud Storage > 存储分区。
  2. 点击您的存储桶名称（例如 hcl-bigfix-logs）。
  3. 前往权限标签页。
  4. 点击授予访问权限。
  5. 提供以下配置详细信息：
    • 添加主账号：粘贴 Google SecOps 服务账号电子邮件地址（例如 chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com）。
    • 分配角色：选择 Storage Object Viewer。

  6. 点击保存。

UDM 映射表

日志字段 UDM 映射 逻辑
WMI_Asset_ID entity.asset.asset_id 如果不为空且不为“N/A”，则设置为“Asset Id:”+ 值
Computer_Serial_Number、Computer_Manufacturer、Computer_Model entity.asset.hardware 包含 serial_number、manufacturer、model 的对象
client_name entity.asset.hostname 直接复制值
MAC_Addresses entity.asset.mac 格式化后从数组合并
OperatingSystemType entity.asset.platform_software.platform 如果值不为空且不是“Unix”，则为大写值
操作系统 entity.asset.platform_software.platform_version 直接复制值
SerialNumber entity.asset.product_object_id 值（如果不为空）
WindowsOperatingSystem entity.asset.software 如果名称不为空，则将名称设置为值的对象
OperatingSystemRole entity.asset.type 如果值不为空，则为大写值
DNS_Name entity.domain.name 值（如果不为空）
CloudInfo entity.labels 如果非空，则为具有键“CloudInfo”和值（不含引号）的标签对象
网域角色 entity.user.attribute.roles 如果名称不为空，则将名称设置为值的对象
Active_Directory_Path entity.user.group_identifiers 合并后的值（如果不为空且不为“”）
WMI_Service_Tag entity.user.product_object_id 值（如果不为空）
User_Name entity.user.user_display_name 如果值不为空且不为“”，则为相应值
client_id entity.user.userid 直接复制值
metadata.entity_type 如果 SerialNumber 不为空，则设置为“ASSET”，否则设置为“USER”
metadata.product_name 设置为“HCL Big Fix”
metadata.vendor_name 设置为“HCL Software”
CloudInfo event.idm.entity.entity.labels 从变更日志映射

更新日志

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

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