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 인스턴스에 가장 가까운 위치를 선택합니다.
    스토리지 클래스 표준(자주 액세스하는 로그에 권장)
    액세스 제어 균일(권장)
    보호 도구 선택사항: 객체 버전 관리 또는 보관 정책 사용 설정

  6. 만들기를 클릭합니다.

Trellix Endpoint Security HX API 사용자 인증 정보 수집

HX 콘솔 URL 가져오기

  1. Trellix Endpoint Security HX 웹 콘솔에 로그인합니다.
  2. 브라우저 주소 표시줄에서 콘솔 URL을 확인합니다(형식: 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 Functions의 서비스 계정 만들기

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. 스토리지 객체 관리자
    2. Cloud Run 호출자
    3. Cloud Functions 호출자
  6. 계속을 클릭한 다음 완료를 클릭합니다.

GCS 버킷에 IAM 권한 부여

  1. Cloud Storage > 버킷으로 이동합니다.
  2. 버킷 이름(fireeye-hx-logs)을 클릭합니다.
  3. 권한 탭으로 이동합니다.
  4. 액세스 권한 부여를 클릭합니다.
  5. 주 구성원 추가: 서비스 계정 이메일을 입력합니다.
  6. 역할 할당: 스토리지 객체 관리자를 선택합니다.
  7. 저장을 클릭합니다.

Pub/Sub 주제 만들기

Cloud Scheduler가 게시하고 Cloud Run 함수가 구독할 주제를 만듭니다.

  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 실행당 최대 레코드 수
    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에서 피드 구성

  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. 소스 삭제 옵션: 원하는 대로 선택합니다.
  10. 다음을 클릭하고 검토한 후 제출을 클릭합니다.

서비스 계정에 IAM 권한 부여

Google SecOps 서비스 계정에는 버킷에 대한 스토리지 객체 뷰어 역할이 필요합니다.

  1. Cloud Storage > 버킷으로 이동합니다.
  2. 버킷을 선택하고 권한 탭으로 이동합니다.
  3. 액세스 권한 부여를 클릭합니다.
  4. 주 구성원 추가: Google SecOps 서비스 계정 이메일을 붙여넣습니다.
  5. 역할 할당: 스토리지 객체 뷰어를 선택합니다.
  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 전문가에게 문의하여 답변을 받으세요.