我們重新調整了說明文件的導覽結構，直接對應您的作業工作流程。詳情請參閱版本資訊和逐步解說影片。

Google 會運用 AI 技術將內容翻譯成你偏好的語言，但可能會出錯。

收集 Trellix Endpoint Security HX (舊稱 FireEye HX) 記錄檔

剖析器版本：30.0

支援的國家/地區：

Google SecOps SIEM

本指南說明如何使用 Google Cloud Storage V2，將 Trellix Endpoint Security HX (舊稱 FireEye HX) 記錄匯入 Google Security Operations。Trellix Endpoint Security HX 是一種端點偵測與回應 (EDR) 平台，可提供進階威脅偵測、調查和遏止功能。這項服務會運用情報主導的防護措施和即時入侵指標 (IOC)，偵測及回應端點上的複雜攻擊。

事前準備

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

Google SecOps 執行個體。
已啟用 Cloud Storage API 的 GCP 專案。
具備建立及管理 GCS 值區和 IAM 政策的權限。
建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限。
具備 Trellix Endpoint Security HX 主控台的特殊權限，且已啟用 API 存取權。
至少具備「分析師」角色的 Trellix HX API 使用者帳戶。

建立 Google Cloud Storage bucket

前往 Google Cloud 控制台。
選取專案。
在導覽選單中，依序前往「Cloud Storage」>「bucket」。
按一下「建立值區」。

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

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

點選「建立」。

收集 Trellix Endpoint Security HX API 憑證

取得 HX 控制台網址

登入 Trellix Endpoint Security HX 網頁控制台。
記下瀏覽器網址列中的控制台網址 (格式：https://<hx-hostname>:<port>)。

建立或驗證 API 使用者帳戶

使用管理員帳戶登入 Trellix HX 控制台。
依序前往「管理」>「使用者帳戶」。
建立新使用者或選取現有使用者，授予 API 存取權。
確認使用者具備 api_analyst 或 api_admin 角色。
請記下 API 存取權的使用者名稱和密碼。

驗證 API 存取權

請先測試憑證，再繼續進行整合：

HX_HOST="[https://hx-console.example.com:3000](https://hx-console.example.com:3000)"
HX_USER="api_user"
HX_PASS="api_password"

# Authenticate and get token
curl -k -X POST "$HX_HOST/hx/api/v3/token" \
    -u "$HX_USER:$HX_PASS"

為 Cloud Run 函式建立服務帳戶

Cloud Run 函式需要具備寫入 GCS bucket 權限的服務帳戶，並由 Pub/Sub 叫用。

建立服務帳戶

在 GCP 主控台中，依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。
按一下「Create Service Account」(建立服務帳戶)。
請提供下列設定詳細資料：
- 服務帳戶名稱：fireeye-hx-collector-sa
- 服務帳戶說明：Service account for Cloud Run function to collect Trellix HX logs
按一下「建立並繼續」。
在「將專案存取權授予這個服務帳戶」部分，新增下列角色：
1. Storage 物件管理員
2. Cloud Run 叫用者
3. Cloud Functions 叫用者
然後依序點選「繼續」和「完成」。

授予 GCS 值區的 IAM 權限

依序前往「Cloud Storage」>「Buckets」。
按一下 bucket 名稱 (fireeye-hx-logs)。
前往「權限」分頁標籤。
按一下「授予存取權」。
新增主體：輸入服務帳戶電子郵件地址。
指派角色：選取「Storage 物件管理員」。
按一下 [儲存]。

建立 Pub/Sub 主題

建立主題，供 Cloud Scheduler 發布訊息，以及 Cloud Run 函式訂閱。

在 GCP 控制台中，前往「Pub/Sub」>「主題」。
按一下「建立主題」。
主題 ID：fireeye-hx-logs-trigger。
點選「建立」。

建立 Cloud Run 函式來收集記錄

前往 GCP Console 的「Cloud Run」。
按一下「Create service」(建立服務)。
選取「函式」。
在「設定」部分，提供下列詳細資料：

設定值

服務名稱 fireeye-hx-collector

區域選取與 GCS bucket 相符的區域

執行階段 Python 3.12 以上版本
在「Trigger」(觸發條件) 專區：
1. 按一下「+ 新增觸發條件」。
2. 選取「Cloud Pub/Sub」。
3. 在「Select a Cloud Pub/Sub topic」(選取 Cloud Pub/Sub 主題) 中，選擇 fireeye-hx-logs-trigger。
4. 按一下 [儲存]。
在「驗證」部分中，選取「需要驗證」，然後勾選「身分與存取權管理 (IAM)」。
向下捲動至「Containers, Networking, Security」。
在「安全性」分頁中，選取服務帳戶 fireeye-hx-collector-sa。

設定	值
服務名稱	`fireeye-hx-collector`
區域	選取與 GCS bucket 相符的區域
執行階段	Python 3.12 以上版本

在「Containers」分頁中，按一下「Variables & Secrets」，然後新增下列項目：

變數名稱	範例值	說明
`GCS_BUCKET`	`fireeye-hx-logs`	GCS bucket 名稱
`GCS_PREFIX`	`hx-alerts`	記錄檔的前置字串
`STATE_KEY`	`hx-alerts/state.json`	狀態檔案路徑
`HX_HOST`	`https://hx-console.example.com:3000`	HX 控制台網址
`HX_USER`	`api_user`	API 使用者名稱
`HX_PASS`	`api_password`	API 密碼
`MAX_RECORDS`	`1000`	每次執行的記錄數上限
`PAGE_SIZE`	`100`	每頁記錄數
`LOOKBACK_HOURS`	`24`	初始回溯期

在「Requests」(要求) 中，將「Request timeout」(要求逾時) 設為 600 秒。
點按「Create」(建立)。服務建立完成後，系統會開啟內嵌程式碼編輯器。

新增函式程式碼

在「進入點」欄位中輸入「main」。

在內嵌編輯器中建立下列檔案：

main.py:

import functions_framework
from google.cloud import storage
import json
import os
import urllib3
from datetime import datetime, timezone, timedelta
import time

# Disable SSL warnings for self-signed certs
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

# Initialize HTTP client with timeouts
http = urllib3.PoolManager(
    timeout=urllib3.Timeout(connect=5.0, read=30.0),
    retries=False,
    cert_reqs='CERT_NONE',
)

# Initialize Storage client
storage_client = storage.Client()

# Environment variables
GCS_BUCKET = os.environ.get('GCS_BUCKET')
GCS_PREFIX = os.environ.get('GCS_PREFIX', 'hx-alerts')
STATE_KEY = os.environ.get('STATE_KEY', 'hx-alerts/state.json')
HX_HOST = os.environ.get('HX_HOST')
HX_USER = os.environ.get('HX_USER')
HX_PASS = os.environ.get('HX_PASS')
MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '1000'))
PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '100'))
LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))

def get_api_token(host, username, password):
    url = f"{host}/hx/api/v3/token"
    auth_string = f"{username}:{password}"
    import base64
    auth_b64 = base64.b64encode(auth_string.encode('utf-8')).decode('utf-8')
    headers = {'Authorization': f'Basic {auth_b64}'}
    response = http.request('POST', url, headers=headers)
    if response.status == 204:
        return response.headers.get('X-FeApi-Token')
    else:
        raise Exception(f"Authentication failed: HTTP {response.status}")

@functions_framework.cloud_event
def main(cloud_event):
    if not all([GCS_BUCKET, HX_HOST, HX_USER, HX_PASS]):
        print('Error: Missing required environment variables')
        return
    try:
        bucket = storage_client.bucket(GCS_BUCKET)
        state = load_state(bucket, STATE_KEY)
        now = datetime.now(timezone.utc)
        last_time = None

        if isinstance(state, dict) and state.get("last_event_time"):
            try:
                last_time = parse_datetime(state["last_event_time"]) - timedelta(minutes=2)
            except Exception as e:
                print(f"Warning: Could not parse last_event_time: {e}")

        if last_time is None:
            last_time = now - timedelta(hours=LOOKBACK_HOURS)

        print(f"Fetching alerts from {last_time.isoformat()} to {now.isoformat()}")
        token = get_api_token(HX_HOST, HX_USER, HX_PASS)
        alerts, newest_time = fetch_alerts(HX_HOST, token, last_time, now, PAGE_SIZE, MAX_RECORDS)
        hosts = fetch_hosts(HX_HOST, token)
        all_records = alerts + hosts

        if not all_records:
            print("No new records found.")
            save_state(bucket, STATE_KEY, now.isoformat())
            return

        timestamp = now.strftime('%Y%m%d_%H%M%S')
        object_key = f"{GCS_PREFIX}/logs_{timestamp}.ndjson"
        ndjson = '\n'.join([json.dumps(record, ensure_ascii=False) for record in all_records]) + '\n'
        bucket.blob(object_key).upload_from_string(ndjson, content_type='application/x-ndjson')

        print(f"Wrote {len(all_records)} records to gs://{GCS_BUCKET}/{object_key}")
        save_state(bucket, STATE_KEY, newest_time if newest_time else now.isoformat())
        release_token(HX_HOST, token)
        print(f"Successfully processed {len(all_records)} records")
    except Exception as e:
        print(f'Error processing logs: {str(e)}')
        raise

def parse_datetime(value):
    if value.endswith("Z"):
        value = value[:-1] + "+00:00"
    return datetime.fromisoformat(value)

def load_state(bucket, key):
    try:
        blob = bucket.blob(key)
        if blob.exists():
            return json.loads(blob.download_as_text())
    except Exception as e:
        print(f"Warning: Could not load state: {e}")
    return {}

def save_state(bucket, key, last_event_time_iso):
    try:
        state = {'last_event_time': last_event_time_iso}
        bucket.blob(key).upload_from_string(json.dumps(state, indent=2), content_type='application/json')
    except Exception as e:
        print(f"Warning: Could not save state: {e}")

def fetch_alerts(host, token, start_time, end_time, page_size, max_records):
    endpoint = f"{host}/hx/api/v3/alerts"
    headers = {'X-FeApi-Token': token, 'Accept': 'application/json'}
    records = []; newest_time = None; offset = 0; backoff = 1.0

    while True:
        if len(records) >= max_records: break
        start_iso = start_time.strftime('%Y-%m-%dT%H:%M:%S.000Z')
        end_iso = end_time.strftime('%Y-%m-%dT%H:%M:%S.000Z')
        url = f"{endpoint}?sort=reported_at+asc&min_reported_at={start_iso}&max_reported_at={end_iso}&offset={offset}&limit={min(page_size, max_records - len(records))}"

        try:
            response = http.request('GET', url, headers=headers)
            if response.status == 429:
                time.sleep(int(response.headers.get('Retry-After', str(int(backoff))))); backoff = min(backoff * 2, 30.0); continue
            backoff = 1.0
            if response.status != 200: break
            data = json.loads(response.data.decode('utf-8'))
            page_results = data.get('data', {}).get('entries', [])
            if not page_results: break
            records.extend(page_results)
            for alert in page_results:
                event_time = alert.get('reported_at')
                if event_time and (newest_time is None or event_time > newest_time): newest_time = event_time
            offset += len(page_results)
            if offset >= data.get('data', {}).get('total', 0): break
        except Exception: break
    return records, newest_time

def fetch_hosts(host, token):
    endpoint = f"{host}/hx/api/v3/hosts"
    headers = {'X-FeApi-Token': token, 'Accept': 'application/json'}
    try:
        response = http.request('GET', f"{endpoint}?limit=100&offset=0", headers=headers)
        if response.status != 200: return []
        return json.loads(response.data.decode('utf-8')).get('data', {}).get('entries', [])
    except Exception: return []

def release_token(host, token):
    try:
        http.request('DELETE', f"{host}/hx/api/v3/token", headers={'X-FeApi-Token': token})
    except Exception: pass

requirements.txt:

functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0

點選「部署」即可儲存並部署函式。

建立 Cloud Scheduler 工作

前往 GCP 主控台的「Cloud Scheduler」。
點選「建立工作」。

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

設定	值
名稱	`fireeye-hx-collector-hourly`
區域	選取與 Cloud Run 函式相同的區域
頻率	`0 * * * *` (每小時整點)
時區	選取時區 (建議使用世界標準時間)
目標類型	Pub/Sub
主題	`fireeye-hx-logs-trigger`
郵件內文	`{}`

點選「建立」。

測試整合項目

在 Cloud Scheduler 中找到您的工作，然後按一下「強制執行」。
依序前往「Cloud Run」>「Services」>「Logs」，確認是否成功。
檢查 Cloud Storage，確認 hx-alerts/ 資料夾中是否有新的 .ndjson 檔案。

在 Google SecOps 中設定動態饋給

依序前往「SIEM 設定」>「動態消息」。
依序點選「新增動態消息」>「設定單一動態消息」。
動態饋給名稱：Trellix HX Logs。
來源類型：Google Cloud Storage V2。
記錄類型：FireEye HX。
按一下「取得服務帳戶」，然後複製電子郵件地址 (例如 chronicle-12345678@...)。
點選「下一步」。
儲存空間值區網址：gs://fireeye-hx-logs/hx-alerts/ (包含結尾斜線)。
來源刪除選項：根據偏好設定選取。
點選「下一步」，檢查內容，然後點選「提交」。

將 IAM 權限授予服務帳戶

Google SecOps 服務帳戶需要 bucket 的「Storage 物件檢視者」角色。

依序前往「Cloud Storage」>「Buckets」。
選取 bucket，然後前往「Permissions」(權限) 分頁。
按一下「授予存取權」。
新增主體：貼上 Google SecOps 服務帳戶電子郵件地址。
指派角色：選取「Storage 物件檢視者」。
按一下 [儲存]。

UDM 對應表

記錄欄位	UDM 對應	邏輯
`alert.event_type`	`metadata.product_event_type`	直接對應。
`alert.reported_at`	`metadata.event_timestamp`	剖析為時間戳記。
`alert.source`	`principal.hostname`	直接對應。
`alert.agent._id`	`principal.asset_id`	直接對應。
`alert.indicator.display_name`	`security_result.threat_name`	直接對應。
`alert.md5values`	`target.file.md5`	直接對應。
不適用	`metadata.vendor_name`	設為「Trellix」Trellix。
不適用	`metadata.product_name`	設為「Endpoint Security HX」。
不適用	`metadata.log_type`	設為 FIREEYE_HX。
`cs12`	`additional.fields`	從變更記錄對應
`cs9`	`target.process.file.md5`	從變更記錄對應
`account_name", "UUID", "Mitre", "host_details.data.sysinfo.url", "host_details.route", "host_details.data.reported_clone", and "host_details.data.timezone`	`security_result.detection_fields`	從變更記錄對應
`Desc`	`metadata.description`	從變更記錄對應
`Confidence`	`security_result.confidence`	從變更記錄對應
`alert.appliance._id`	`additional.fields`	從變更記錄對應
`host_details.data.stats.acqs", "host_details.data.stats.alerting_conditions", "host_details.data.stats.alerts", "host_details.data.stats.exploit_alerts", "host_details.data.stats.exploit_blocks", and "host_details.data.stats.false_positive_alerts`	`security_result.detection_fields`	從變更記錄對應
`categoryOutcome", "cs13`	`additional.fields`	從變更記錄對應
`cs6`	`target.process.file.sha1`	從變更記錄對應
`host_details.message`	`security_result.action_details`	從變更記錄對應
`alert.md5values", "alert.resolution", "alert.is_false_positive", and "alert.alert_type`	`additional.fields`	從變更記錄對應
`type.threat_type`	`security_result.threat_name`	從變更記錄對應
`ent.lms_event_id`	`metadata.product_log_id`	從變更記錄對應
`email.smtp.mail_from`	`network.email.from`	從變更記錄對應
`email.headers.subject`	`network.email.subject`	從變更記錄對應
`email.headers.to`	`network.email.to`	從變更記錄對應
`ent.type", "ent.id", "ent.name", "ent.object_source", "ent.binary", and "ent.attributes.scan_id`	`security_result.detection_fields`	從變更記錄對應
`cs11Label`	`additional_cs11Label.key`	從變更記錄對應
`cs11`	`additional_cs11.value`	從變更記錄對應
`alert.sysinfo.mac_address`	`principal.mac`	從變更記錄對應
`host_details.data.agent_version`	`metadata.product_version`	從變更記錄對應
`alert.url`	`metadata.url_back_to_product`	從變更記錄對應
`description`	`metadata.description`	從變更記錄對應
`alert.agent._id`	`principal.asset.asset_id`	從變更記錄對應
`alert.event_id`	`metadata.product_log_id`	從變更記錄對應
`deviceCustomDate1Label`	`additional_deviceCustomDate1.key`	從變更記錄對應
`deviceCustomDate1`	`additional_deviceCustomDate1.value`	從變更記錄對應
`deviceCustomDate2`	`additional_deviceCustomDate2.value`	從變更記錄對應
`client`	`principal.ip`	從變更記錄對應
`principal_ip`	`principal.ip`	從變更記錄對應
`remoteaddress`	`principal.ip`	從變更記錄對應
`host_`	`principal.hostname`	從變更記錄對應
`line`	`principal.application`	從變更記錄對應
`username`	`principal.user.userid`	從變更記錄對應
`client_app_type`	`principal.resource.attribute.labels`	從變更記錄對應
`upstream`	`target.url`	從變更記錄對應
`role`	`target.user.role_name`	從變更記錄對應
`server`	`target.resource.attribute.labels`	從變更記錄對應
`localusername`	`target.user.user_display_name`	從變更記錄對應
`request`	`additional.fields`	從變更記錄對應
`mlocked`	`additional.fields`	從變更記錄對應
`kernel_stack`	`additional.fields`	從變更記錄對應
`sessionID`	`network.session_id`	從變更記錄對應
`auth_mechanism`	`extensions.auth.mechanism`	從變更記錄對應
`authsubmethod`	`extensions.auth.auth_details`	從變更記錄對應

變更記錄

查看這個剖析器的變更記錄

還有其他問題嗎？向社群成員和 Google SecOps 專業人員尋求答案。