收集 HCL BigFix 記錄

支援的國家/地區：

本文說明如何透過 Cloud Run 函式，使用 Google Cloud Storage V2 將 HCL BigFix 記錄檔擷取至 Google Security Operations。

HCL BigFix 是端點管理平台，提供統一的端點管理、安全性和法規遵循功能。可協助機構發掘、管理及修復基礎架構中的端點，包括修補程式管理、軟體發布、設定管理和安全合規。Cloud Run 函式會使用工作階段相關性查詢輪詢 HCL BigFix REST API，將結果以 NDJSON 格式寫入 Cloud Storage 值區，然後 Google SecOps 會透過 Cloud Storage V2 資訊提供擷取這些結果。

事前準備

請確認您已完成下列事前準備事項：

  • Google SecOps 執行個體
  • 已啟用計費功能的 Google Cloud 專案
  • 具備主控台操作員權限的 HCL BigFix 主控台存取權
  • HCL BigFix Server 9.5 以上版本
  • 在 BigFix 伺服器上執行的 WebReports 服務
  • 從 Cloud Run 函式到 BigFix 伺服器的網路連線 (通訊埠 52311，預設 REST API 通訊埠)
  • Google Cloud 已啟用 API：Cloud Run 函式、Cloud Storage、Cloud Scheduler、Pub/Sub 和 Identity and Access Management (IAM)

建立 Google Cloud 儲存空間 bucket

  1. 前往 Google Cloud 控制台。
  2. 選取專案或建立新專案。
  3. 在導覽選單中，依序前往「Cloud Storage」>「Bucket」。
  4. 按一下「建立值區」。

  5. 請提供下列設定詳細資料：

    設定 值
    為 bucket 命名 輸入全域不重複的名稱 (例如 hcl-bigfix-logs)
    位置類型 根據需求選擇 (區域、雙區域、多區域)
    位置 選取最接近 Google SecOps 執行個體的位置 (例如 us-central1)
    儲存空間級別 標準 (建議用於經常存取的記錄)
    存取控管 統一 (建議)
    保護工具 選用：啟用物件版本管理或資料保留政策

  6. 點按「Create」(建立)。

收集 HCL BigFix API 憑證

如要讓 Cloud Run 函式擷取端點資料，請建立專屬的 BigFix Console 運算子，並授予 REST API 存取權。

建立專用的 API 存取權運算子

  1. 登入 BigFix 控制台。
  2. 依序前往「工具」>「建立作業員」。

  3. 在「新增使用者」對話方塊中，提供下列資訊：

    • 使用者名稱：輸入描述性使用者名稱 (例如 chronicle_api)。
    • 密碼：輸入高強度密碼。
    • 確認密碼：再次輸入密碼。

  4. 按一下 [確定]。

設定操作者權限

建立運算子後，系統會自動開啟「Console Operator」(控制台運算子) 視窗。

  1. 點按「Details」(詳細資料) 分頁標籤。
  2. 在「總覽」部分：
    • 選取「一律允許這位使用者登入」。
  3. 在「權限」部分中，設定下列權限：
    • 可以提交查詢：設為「是」。
    • 自訂內容：設為「是」。
  4. 在「介面登入權限」部分：
    • 可使用 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"

    成功的回應會傳回 XML 文件，其中包含電腦名稱和 ID。如果收到驗證錯誤訊息，請確認您在先前步驟中設定的運算子權限。

為 Cloud Run 函式建立服務帳戶

  1. 在 Google Cloud 控制台中，依序前往「IAM 與管理」>「服務帳戶」。
  2. 按一下「Create Service Account」(建立服務帳戶)。

  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 bucket 中的物件)。
    • Cloud Run 叫用者 (允許 Cloud Scheduler 叫用函式)。

  6. 按一下「繼續」。

  7. 按一下 [完成]。

建立 Pub/Sub 主題

Cloud Scheduler 會透過 Pub/Sub 主題觸發 Cloud Run 函式。

  1. 在 Google Cloud 控制台中，依序前往「Pub/Sub」>「主題」。
  2. 按一下「建立主題」。
  3. 在「Topic ID」(主題 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 bucket 名稱。
    • 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 bucket 名稱
GCS_PREFIX 否 bigfix_logs bucket 中的物件前置字串 (資料夾路徑)
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. 請提供下列設定詳細資料：
    • 「Name」(名稱)：輸入 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 服務帳戶並設定動態饋給

Google SecOps 會使用專屬服務帳戶，從 Cloud Storage 值區讀取資料。您必須授予這個服務帳戶值區存取權。

取得服務帳戶電子郵件地址

  1. 依序前往「SIEM 設定」>「動態饋給」。
  2. 按一下「新增動態消息」。
  3. 按一下「設定單一動態饋給」。
  4. 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 HCL BigFix Logs)。
  5. 選取「Google Cloud Storage V2」做為「來源類型」。
  6. 選取「HCL BigFix」做為「記錄類型」。

  7. 按一下「取得服務帳戶」。

    系統會顯示專屬服務帳戶電子郵件地址，例如：

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

  8. 複製這個電子郵件地址，以便在下一步中使用。

  9. 點選「下一步」。

  10. 指定下列輸入參數的值：

    • 儲存空間 bucket 網址：輸入 Cloud Storage bucket URI：

      gs://hcl-bigfix-logs/bigfix_logs/
      • 將 hcl-bigfix-logs 替換為 Cloud Storage 值區名稱。
      • 將 bigfix_logs 替換為您設定的 GCS_PREFIX 值。

    • 來源刪除選項：根據偏好設定選取刪除選項：

      • 永不：移轉後一律不刪除任何檔案 (建議用於測試)。
      • 刪除已轉移的檔案：成功轉移檔案後刪除檔案。

      • 刪除已轉移的檔案和空白目錄：成功轉移後刪除檔案和空白目錄。

    • 檔案存在時間上限：包含在過去天數內修改的檔案 (預設為 180 天)。

    • 資產命名空間：資產命名空間。

    • 擷取標籤：要套用至這個動態饋給事件的標籤。

  11. 點選「下一步」。

  12. 在「Finalize」(完成) 畫面中檢查新的動態饋給設定，然後按一下「Submit」(提交)。

將 IAM 權限授予 Google SecOps 服務帳戶

Google SecOps 服務帳戶必須具備 Cloud Storage bucket 的「Storage 物件檢視者」角色。

  1. 依序前往「Cloud Storage」>「Buckets」。
  2. 按一下 bucket 名稱 (例如 hcl-bigfix-logs)。
  3. 前往「權限」分頁標籤。
  4. 按一下「授予存取權」。
  5. 請提供下列設定詳細資料：
    • 新增主體：貼上 Google SecOps 服務帳戶電子郵件地址 (例如 chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com)。
    • 指派角色：選取「Storage 物件檢視者」。

  6. 按一下 [儲存]。

UDM 對應表

記錄欄位 UDM 對應 邏輯
WMI_Asset_ID entity.asset.asset_id 如不為空白且不為「N/A」，請設為「資產 ID：」+ 值
Computer_Serial_Number、Computer_Manufacturer、Computer_Model entity.asset.hardware 具有 serial_number、製造商、型號的物件
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 專業人員尋求答案。