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 ジョブを作成する権限。
  • API アクセスが有効になっている Trellix Endpoint Security HX コンソールへの特権アクセス。
  • 少なくともアナリスト ロールを持つ Trellix HX API ユーザー アカウント。

Google Cloud Storage バケットを作成する

  1. Google Cloud コンソールに移動します。
  2. プロジェクトを選択します。
  3. ナビゲーション メニューで、[Cloud Storage] > [バケット] に移動します。
  4. [バケットを作成] をクリックします。

  5. 次の構成情報を提供してください。

    設定 値
    バケットに名前を付ける グローバルに一意の名前（fireeye-hx-logs など）を入力します。
    ロケーション タイプ ニーズに応じて選択します（リージョン、デュアルリージョン、マルチリージョン）。
    ロケーション Google SecOps インスタンスに最も近いロケーションを選択します。
    ストレージ クラス Standard（頻繁にアクセスされるログに推奨）
    アクセス制御 均一（推奨）
    保護ツール 省略可: オブジェクトのバージョニングまたは保持ポリシーを有効にする

  6. [作成] をクリックします。

Trellix Endpoint Security HX API 認証情報を収集する

HX コンソールの URL を取得する

  1. Trellix Endpoint Security HX ウェブコンソールにログインします。
  2. ブラウザのアドレスバーからコンソールの URL をメモします（形式: https://<hx-hostname>:<port>）。

API ユーザー アカウントを作成または確認する

  1. 管理者アカウントで Trellix HX コンソールにログインします。
  2. [Admin] > [User Accounts] に移動します。
  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 functions のサービス アカウントを作成する

Cloud Run functions の関数には、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. ストレージのオブジェクト管理者
    2. Cloud Run 起動元
    3. Cloud Functions 起動元
  6. [続行]、[完了] の順にクリックします。

GCS バケットに対する IAM 権限を付与する

  1. [Cloud Storage] > [バケット] に移動します。
  2. バケット名（fireeye-hx-logs）をクリックします。
  3. [権限] タブに移動します。
  4. [アクセス権を付与] をクリックします。
  5. プリンシパルを追加: サービス アカウントのメールアドレスを入力します。
  6. ロールを割り当てる: [Storage オブジェクト管理者] を選択します。
  7. [保存] をクリックします。

Pub/Sub トピックの作成

Cloud Scheduler がパブリッシュし、Cloud Run functions の関数がサブスクライブするトピックを作成します。

  1. GCP コンソールで、[Pub/Sub] > [トピック] に移動します。
  2. [トピックを作成] をクリックします。
  3. トピック ID: fireeye-hx-logs-trigger。
  4. [作成] をクリックします。

ログを収集する Cloud Run 関数を作成する

  1. GCP Console で Cloud Run に移動します。
  2. [サービスを作成] をクリックします。
  3. [関数] を選択します。

  4. [構成] セクションで、次の詳細を入力します。

    設定 値
    サービス名 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 コンソールの URL
    HX_USER api_user API ユーザー名
    HX_PASS api_password API パスワード
    MAX_RECORDS 1000 1 回の実行あたりの最大レコード数
    PAGE_SIZE 100 1 ページあたりのレコード数
    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 functions の関数と同じリージョンを選択します。
    周波数 0 * * * *（1 時間ごと、毎正時）
    タイムゾーン タイムゾーンを選択（UTC を推奨）
    ターゲット タイプ Pub/Sub
    トピック fireeye-hx-logs-trigger
    メッセージ本文 {}

  4. [作成] をクリックします。

統合をテストする

  1. Cloud Scheduler でジョブを見つけて、[強制実行] をクリックします。
  2. [Cloud Run] > [サービス] > [ログ] に移動して、成功したことを確認します。
  3. Cloud Storage を確認して、新しい .ndjson ファイルが hx-alerts/ フォルダに存在することを確認します。

Google SecOps でフィードを構成する

  1. [設定] > [データ管理] > [フィード] に移動します。SIEM のみのプラットフォームを使用している場合は、[設定] > [フィード] に移動します。
  2. [新しいフィードを追加] > [単一のフィードを構成する] をクリックします。
  3. フィード名: Trellix HX Logs。
  4. ソースタイプ: Google Cloud Storage V2。
  5. ログタイプ: FireEye HX。
  6. [サービス アカウントを取得する] をクリックし、メールアドレス（例: chronicle-12345678@...）をコピーします。
  7. [次へ] をクリックします。
  8. ストレージ バケットの URL: gs://fireeye-hx-logs/hx-alerts/（末尾のスラッシュを含む）。
  9. Source deletion option: 必要に応じて選択します。
  10. [次へ] をクリックして確認し、[送信] をクリックします。

サービス アカウントに IAM 権限を付与する

Google SecOps サービス アカウントには、バケットに対する Storage オブジェクト閲覧者のロールが必要です。

  1. [Cloud Storage] > [バケット] に移動します。
  2. バケットを選択し、[権限] タブに移動します。
  3. [アクセス権を付与] をクリックします。
  4. プリンシパルを追加: Google SecOps サービス アカウントのメールアドレスを貼り付けます。
  5. ロールを割り当てる: [Storage オブジェクト閲覧者] を選択します。
  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 のプロフェッショナルから回答を得ることができます。