收集 Trellix Endpoint Security HX（以前称为 FireEye HX）日志

解析器版本：30.0

支持的平台：

本指南介绍了如何使用 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 访问权限。
  • 至少具有 Analyst 角色的 Trellix HX API 用户账号。

创建 Google Cloud Storage 存储桶

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

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

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

  6. 点击创建。

收集 Trellix Endpoint Security HX API 凭据

获取 HX 控制台网址

  1. 登录 Trellix Endpoint Security HX Web 控制台。
  2. 记下浏览器地址栏中的控制台网址（格式：https://<hx-hostname>:<port>）。

创建或验证 API 用户账号

  1. 使用管理员账号登录 Trellix HX 控制台。
  2. 依次前往管理 > 用户账号。
  3. 创建新用户或选择现有用户以授予 API 访问权限。
  4. 确保用户拥有 api_analyst 或 api_admin 角色。
  5. 请记好用于 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 存储桶写入内容以及被 Pub/Sub 调用的权限。

创建服务账号

  1. 在 GCP 控制台中，依次前往 IAM 和管理 > 服务账号。
  2. 点击创建服务账号。
  3. 提供以下配置详细信息：
    • 服务账号名称：fireeye-hx-collector-sa
    • 服务账号说明：Service account for Cloud Run function to collect Trellix HX logs
  4. 点击创建并继续。
  5. 在向此服务账号授予对项目的访问权限部分中，添加以下角色：
    1. Storage Object Admin
    2. Cloud Run Invoker
    3. Cloud Functions Invoker
  6. 依次点击继续和完成。

授予对 GCS 存储桶的 IAM 权限

  1. 依次前往 Cloud Storage > 存储分区。
  2. 点击相应存储桶的名称 (fireeye-hx-logs)。
  3. 前往权限标签页。
  4. 点击授予访问权限。
  5. 添加主账号：输入服务账号电子邮件地址。
  6. 分配角色：选择 Storage Object Admin。
  7. 点击保存。

创建 Pub/Sub 主题

创建一个主题，Cloud Scheduler 将向该主题发布消息，而 Cloud Run 函数将订阅该主题。

  1. 在 GCP 控制台中，依次前往 Pub/Sub > 主题。
  2. 点击创建主题。
  3. 主题 ID：fireeye-hx-logs-trigger。
  4. 点击创建。

创建 Cloud Run 函数以收集日志

  1. 在 GCP 控制台中，前往 Cloud Run。
  2. 点击创建服务。
  3. 选择函数。

  4. 在配置部分中，提供以下详细信息：

    设置 值
    Service 名称 fireeye-hx-collector
    区域 选择与您的 GCS 存储桶匹配的区域
    运行时 Python 3.12 或更高版本

  5. 在触发器部分中：

    1. 点击 + 添加触发器。
    2. 选择 Cloud Pub/Sub。
    3. 在选择 Cloud Pub/Sub 主题中，选择 fireeye-hx-logs-trigger。
    4. 点击保存。

  6. 在身份验证部分中，选择需要身份验证，然后选中 Identity and Access Management (IAM)。

  7. 向下滚动到容器、网络、安全性。

  8. 在安全性标签页中，选择服务账号 fireeye-hx-collector-sa。

  9. 在容器标签页中，点击变量和密钥，然后添加以下内容：

    变量名称 示例值 说明
    GCS_BUCKET fireeye-hx-logs GCS 存储桶名称
    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 初始回溯期

  10. 在请求中，将请求超时设置为 600 秒。

  11. 点击创建。创建服务后，系统会打开内嵌代码编辑器。

添加函数代码

  1. 在入口点字段中输入 main。

  2. 在内嵌编辑器中创建以下文件：

    • 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

  3. 点击部署以保存并部署该函数。

创建 Cloud Scheduler 作业

  1. 在 GCP Console 中，前往 Cloud Scheduler。
  2. 点击创建作业。

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

    设置 值
    名称 fireeye-hx-collector-hourly
    区域 选择与 Cloud Run 函数相同的区域
    频率 0 * * * *（每小时一次，整点时）
    时区 选择时区（建议选择 UTC）
    目标类型 Pub/Sub
    主题 fireeye-hx-logs-trigger
    消息正文 {}

  4. 点击创建。

测试集成

  1. 在 Cloud Scheduler 中，找到您的作业，然后点击强制运行。
  2. 前往 Cloud Run > 服务 > 日志，验证是否成功。
  3. 检查 Cloud Storage，确认 hx-alerts/ 文件夹中是否存在新的 .ndjson 文件。

在 Google SecOps 中配置 Feed

  1. 依次前往 SIEM 设置 > Feed。
  2. 依次点击添加新 Feed > 配置单个 Feed。
  3. Feed 名称：Trellix HX Logs。
  4. 来源类型：Google Cloud Storage V2。
  5. 日志类型：FireEye HX。
  6. 点击 Get Service Account，然后复制电子邮件地址（例如 chronicle-12345678@...）。
  7. 点击下一步。
  8. 存储桶网址：gs://fireeye-hx-logs/hx-alerts/（包含结尾斜杠）。
  9. 来源删除选项：根据您的偏好选择。
  10. 点击下一步，查看，然后点击提交。

向服务账号授予 IAM 权限

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

  1. 依次前往 Cloud Storage > 存储分区。
  2. 选择您的存储桶，然后前往权限标签页。
  3. 点击授予访问权限。
  4. 添加主账号：粘贴 Google SecOps 服务账号电子邮件地址。
  5. 分配角色：选择 Storage Object Viewer。
  6. 点击保存。

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。
不适用 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 专业人士的解答。