Collecter les journaux Trellix Endpoint Security HX (anciennement FireEye HX)

Version de l'analyseur : 30.0

Compatible avec :

Ce guide explique comment ingérer les journaux Trellix Endpoint Security HX (anciennement FireEye HX) dans Google Security Operations à l'aide de Google Cloud Storage V2. Trellix Endpoint Security HX est une plate-forme de détection et de réponse au niveau des points de terminaison (EDR) qui offre des fonctionnalités avancées de détection, d'investigation et de confinement des menaces. Il utilise une protection basée sur l'intelligence et des indicateurs de compromission (IOC) en temps réel pour détecter les attaques sophistiquées sur les points de terminaison et y répondre.

Avant de commencer

Assurez-vous de remplir les conditions suivantes :

  • Une instance Google SecOps.
  • Un projet GCP avec l'API Cloud Storage activée.
  • Autorisations permettant de créer et de gérer des buckets GCS et des règles IAM.
  • Autorisations pour créer des services Cloud Run, des sujets Pub/Sub et des jobs Cloud Scheduler.
  • Accès privilégié à la console Trellix Endpoint Security HX avec l'accès à l'API activé.
  • Un compte utilisateur de l'API Trellix HX avec au moins le rôle d'analyste.

Créer un bucket Google Cloud Storage

  1. Accédez à la console Google Cloud.
  2. Sélectionnez votre projet.
  3. Dans le menu de navigation, accédez à Cloud Storage > Buckets.
  4. Cliquez sur Créer un bucket.

  5. Fournissez les informations de configuration suivantes :

    Paramètre Valeur
    Attribuez un nom au bucket Saisissez un nom unique (par exemple, fireeye-hx-logs).
    Type d'emplacement Choisissez en fonction de vos besoins (région, birégion, multirégion).
    Emplacement Sélectionnez l'emplacement le plus proche de votre instance Google SecOps.
    Classe de stockage Standard (recommandé pour les journaux auxquels vous accédez fréquemment)
    Access control (Contrôle des accès) Uniforme (recommandé)
    Outils de protection Facultatif : Activez la gestion des versions des objets ou la règle de conservation.

  6. Cliquez sur Créer.

Collecter les identifiants de l'API Trellix Endpoint Security HX

Obtenir l'URL de la console HX

  1. Connectez-vous à la console Web Trellix Endpoint Security HX.
  2. Notez l'URL de la console dans la barre d'adresse du navigateur (format : https://<hx-hostname>:<port>).

Créer ou valider un compte utilisateur de l'API

  1. Connectez-vous à la console Trellix HX avec un compte administrateur.
  2. Accédez à Admin > User Accounts (Administrateur > Comptes utilisateur).
  3. Créez un nouvel utilisateur ou sélectionnez-en un existant pour accéder à l'API.
  4. Assurez-vous que l'utilisateur dispose du rôle api_analyst ou api_admin.
  5. Notez le nom d'utilisateur et le mot de passe pour l'accès à l'API.

Vérifier l'accès à l'API

  • Testez vos identifiants avant de procéder à l'intégration :

    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"

Créer un compte de service pour la fonction Cloud Run

La fonction Cloud Run a besoin d'un compte de service disposant des autorisations nécessaires pour écrire dans le bucket GCS et être appelée par Pub/Sub.

Créer un compte de service

  1. Dans la console GCP, accédez à IAM et administration > Comptes de service.
  2. Cliquez sur Créer un compte de service.
  3. Fournissez les informations de configuration suivantes :
    • Service account name (Nom du compte de service) : fireeye-hx-collector-sa
    • Description du compte de service : Service account for Cloud Run function to collect Trellix HX logs
  4. Cliquez sur Créer et continuer.
  5. Dans la section Autoriser ce compte de service à accéder au projet, ajoutez les rôles suivants :
    1. Administrateur des objets Storage
    2. Demandeur Cloud Run
    3. Demandeur Cloud Functions
  6. Cliquez sur Continuer, puis sur OK.

Accorder des autorisations IAM sur le bucket GCS

  1. Accédez à Cloud Storage > Buckets.
  2. Cliquez sur le nom du bucket (fireeye-hx-logs).
  3. Accédez à l'onglet Autorisations.
  4. Cliquez sur Accorder l'accès.
  5. Ajouter des comptes principaux : saisissez l'adresse e-mail du compte de service.
  6. Attribuer des rôles : sélectionnez Administrateur des objets Storage.
  7. Cliquez sur Enregistrer.

Créer un sujet Pub/Sub

Créez un sujet auquel Cloud Scheduler publiera des messages et auquel la fonction Cloud Run s'abonnera.

  1. Dans la console GCP, accédez à Pub/Sub > Sujets.
  2. Cliquez sur Créer un sujet.
  3. ID du thème : fireeye-hx-logs-trigger.
  4. Cliquez sur Créer.

Créer une fonction Cloud Run pour collecter les journaux

  1. Dans la console GCP, accédez à Cloud Run.
  2. Cliquez sur Créer un service.
  3. Sélectionnez Fonction.

  4. Dans la section Configurer, fournissez les informations suivantes :

    Paramètre Valeur
    Nom du service fireeye-hx-collector
    Région Sélectionnez la région correspondant à votre bucket GCS.
    Durée d'exécution Python 3.12 ou version ultérieure

  5. Dans la section Déclencheur :

    1. Cliquez sur + Ajouter un déclencheur.
    2. Sélectionnez Cloud Pub/Sub.
    3. Dans Sélectionner un sujet Cloud Pub/Sub, sélectionnez fireeye-hx-logs-trigger.
    4. Cliquez sur Enregistrer.

  6. Dans la section Authentification, sélectionnez Exiger l'authentification et cochez Identity and Access Management (IAM).

  7. Faites défiler la page jusqu'à Conteneurs, mise en réseau, sécurité.

  8. Dans l'onglet Sécurité, sélectionnez le compte de service fireeye-hx-collector-sa.

  9. Dans l'onglet Conteneurs, cliquez sur Variables et secrets, puis ajoutez les éléments suivants :

    Nom de la variable Exemple de valeur Description
    GCS_BUCKET fireeye-hx-logs Nom du bucket GCS
    GCS_PREFIX hx-alerts Préfixe des fichiers journaux
    STATE_KEY hx-alerts/state.json Chemin d'accès au fichier d'état
    HX_HOST https://hx-console.example.com:3000 URL de la console HX
    HX_USER api_user Nom d'utilisateur de l'API
    HX_PASS api_password Mot de passe de l'API
    MAX_RECORDS 1000 Nombre maximal d'enregistrements par exécution
    PAGE_SIZE 100 Enregistrements par page
    LOOKBACK_HOURS 24 Période d'analyse initiale

  10. Dans Requests (Requêtes), définissez Request timeout (Délai avant expiration de la requête) sur 600 secondes.

  11. Cliquez sur Créer. Une fois le service créé, l'éditeur de code intégré s'ouvre.

Ajouter un code de fonction

  1. Saisissez main dans le champ Point d'entrée.

  2. Créez les fichiers suivants dans l'éditeur intégré :

    • 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. Cliquez sur Déployer pour enregistrer et déployer la fonction.

Créer une tâche Cloud Scheduler

  1. Dans la console GCP, accédez à Cloud Scheduler.
  2. Cliquez sur Créer une tâche.

  3. Fournissez les informations de configuration suivantes :

    Paramètre Valeur
    Nom fireeye-hx-collector-hourly
    Région Sélectionnez la même région que la fonction Cloud Run.
    Fréquence 0 * * * * (toutes les heures, à l'heure pile)
    Fuseau horaire Sélectionnez un fuseau horaire (UTC recommandé).
    Type de cible Pub/Sub
    Sujet fireeye-hx-logs-trigger
    Corps du message {}

  4. Cliquez sur Créer.

Tester l'intégration

  1. Dans Cloud Scheduler, recherchez votre job et cliquez sur Forcer l'exécution.
  2. Accédez à Cloud Run > Services > Journaux pour vérifier que l'opération a réussi.
  3. Cochez Cloud Storage pour vérifier qu'un fichier .ndjson existe dans le dossier hx-alerts/.

Configurer un flux dans Google SecOps

  1. Accédez à Paramètres SIEM > Flux.
  2. Cliquez sur Ajouter un flux > Configurer un seul flux.
  3. Nom du flux : Trellix HX Logs.
  4. Type de source : Google Cloud Storage V2.
  5. Type de journal : FireEye HX.
  6. Cliquez sur Get Service Account (Obtenir un compte de service), puis copiez l'adresse e-mail (par exemple, chronicle-12345678@...).
  7. Cliquez sur Suivant.
  8. URL du bucket de stockage : gs://fireeye-hx-logs/hx-alerts/ (incluez la barre oblique de fin).
  9. Option de suppression de la source : sélectionnez l'option de votre choix.
  10. Cliquez sur Suivant, vérifiez, puis cliquez sur Envoyer.

Accorder des autorisations IAM au compte de service

Le compte de service Google SecOps a besoin du rôle Lecteur des objets Storage sur votre bucket.

  1. Accédez à Cloud Storage > Buckets.
  2. Sélectionnez votre bucket, puis accédez à l'onglet Autorisations.
  3. Cliquez sur Accorder l'accès.
  4. Ajouter des comptes principaux : collez l'adresse e-mail du compte de service Google SecOps.
  5. Attribuer des rôles : sélectionnez Lecteur d'objets Storage.
  6. Cliquez sur Enregistrer.

Table de mappage UDM

Champ du journal Mappage UDM Logique
alert.event_type metadata.product_event_type Mappage direct.
alert.reported_at metadata.event_timestamp Analysé en tant que code temporel.
alert.source principal.hostname Mappage direct.
alert.agent._id principal.asset_id Mappage direct.
alert.indicator.display_name security_result.threat_name Mappage direct.
alert.md5values target.file.md5 Mappage direct.
N/A metadata.vendor_name Définissez-le sur Trellix.
N/A metadata.product_name Définissez la valeur sur Endpoint Security HX.
N/A metadata.log_type Définissez-le sur FIREEYE_HX.
cs12 additional.fields Mappé à partir du journal des modifications
cs9 target.process.file.md5 Mappé à partir du journal des modifications
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 Mappé à partir du journal des modifications
Desc metadata.description Mappé à partir du journal des modifications
Confidence security_result.confidence Mappé à partir du journal des modifications
alert.appliance._id additional.fields Mappé à partir du journal des modifications
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 Mappé à partir du journal des modifications
categoryOutcome", "cs13 additional.fields Mappé à partir du journal des modifications
cs6 target.process.file.sha1 Mappé à partir du journal des modifications
host_details.message security_result.action_details Mappé à partir du journal des modifications
alert.md5values", "alert.resolution", "alert.is_false_positive", and "alert.alert_type additional.fields Mappé à partir du journal des modifications
type.threat_type security_result.threat_name Mappé à partir du journal des modifications
ent.lms_event_id metadata.product_log_id Mappé à partir du journal des modifications
email.smtp.mail_from network.email.from Mappé à partir du journal des modifications
email.headers.subject network.email.subject Mappé à partir du journal des modifications
email.headers.to network.email.to Mappé à partir du journal des modifications
ent.type", "ent.id", "ent.name", "ent.object_source", "ent.binary", and "ent.attributes.scan_id security_result.detection_fields Mappé à partir du journal des modifications
cs11Label additional_cs11Label.key Mappé à partir du journal des modifications
cs11 additional_cs11.value Mappé à partir du journal des modifications
alert.sysinfo.mac_address principal.mac Mappé à partir du journal des modifications
host_details.data.agent_version metadata.product_version Mappé à partir du journal des modifications
alert.url metadata.url_back_to_product Mappé à partir du journal des modifications
description metadata.description Mappé à partir du journal des modifications
alert.agent._id principal.asset.asset_id Mappé à partir du journal des modifications
alert.event_id metadata.product_log_id Mappé à partir du journal des modifications
deviceCustomDate1Label additional_deviceCustomDate1.key Mappé à partir du journal des modifications
deviceCustomDate1 additional_deviceCustomDate1.value Mappé à partir du journal des modifications
deviceCustomDate2 additional_deviceCustomDate2.value Mappé à partir du journal des modifications
client principal.ip Mappé à partir du journal des modifications
principal_ip principal.ip Mappé à partir du journal des modifications
remoteaddress principal.ip Mappé à partir du journal des modifications
host_ principal.hostname Mappé à partir du journal des modifications
line principal.application Mappé à partir du journal des modifications
username principal.user.userid Mappé à partir du journal des modifications
client_app_type principal.resource.attribute.labels Mappé à partir du journal des modifications
upstream target.url Mappé à partir du journal des modifications
role target.user.role_name Mappé à partir du journal des modifications
server target.resource.attribute.labels Mappé à partir du journal des modifications
localusername target.user.user_display_name Mappé à partir du journal des modifications
request additional.fields Mappé à partir du journal des modifications
mlocked additional.fields Mappé à partir du journal des modifications
kernel_stack additional.fields Mappé à partir du journal des modifications
sessionID network.session_id Mappé à partir du journal des modifications
auth_mechanism extensions.auth.mechanism Mappé à partir du journal des modifications
authsubmethod extensions.auth.auth_details Mappé à partir du journal des modifications

Journal des modifications

Afficher le journal des modifications pour ce parseur

Vous avez encore besoin d'aide ? Obtenez des réponses de membres de la communauté et de professionnels Google SecOps.