Collecter les journaux BeyondTrust Endpoint Privilege Management (EPM)

Compatible avec :

Ce document explique comment ingérer les journaux BeyondTrust Endpoint Privilege Management (EPM) dans Google Security Operations à l'aide de Cloud Storage. L'analyseur se concentre sur la transformation des données de journaux JSON brutes de BeyondTrust Endpoint en un format structuré conforme à l'UDM Google SecOps. Il initialise d'abord les valeurs par défaut de différents champs, puis analyse la charge utile JSON. Il mappe ensuite des champs spécifiques du journal brut dans les champs UDM correspondants de l'objet event.idm.read_only_udm.

Avant de commencer

Assurez-vous de remplir les conditions suivantes :

  • Une instance Google SecOps
  • Un projet Google Cloud avec l'API Cloud Storage activée
  • Autorisations pour créer et gérer des buckets Cloud Storage
  • Autorisations permettant de gérer les stratégies Identity and Access Management (IAM) sur les buckets Cloud Storage
  • Autorisations pour créer des services de fonctions Cloud Run, des sujets Pub/Sub et des jobs Cloud Scheduler
  • Accès privilégié au locataire ou à l'API BeyondTrust Endpoint Privilege Management

Créer un bucket Cloud Storage

  1. Accédez à la consoleGoogle Cloud .
  2. Sélectionnez votre projet ou créez-en un.
  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, beyondtrust-epm-logs).
    Type d'emplacement Choisissez en fonction de vos besoins (région, birégion, multirégion).
    Emplacement Sélectionnez l'emplacement (par exemple, us-central1).
    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 BeyondTrust EPM

  1. Connectez-vous à la console Web BeyondTrust Privilege Management en tant qu'administrateur.
  2. Accédez à Configuration> Paramètres> Paramètres de l'API.
  3. Cliquez sur Create an API Account (Créer un compte d'API).
  4. Fournissez les informations de configuration suivantes :
    • Nom : saisissez Google SecOps Collector.
    • Accès API : activez Reporting (lecture seule). Le point de terminaison /management-api/v3/Events/FromStartDate utilisé par ce collecteur appartient à la catégorie d'autorisation Reporting. N'activez l'option Audit (lecture seule) que si vous souhaitez également collecter les journaux d'activité de la console Web.
  5. Copiez et enregistrez l'ID client et le code secret du client.

  6. Copiez l'URL de base de l'API affichée en haut de la page Paramètres de l'API. Il s'agit généralement de https://<your-tenant>-services.pm.beyondtrustcloud.com. Vous l'utiliserez comme BPT_API_URL.

Créer un compte de service pour Cloud Run Functions

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

Créer le compte de service

  1. Dans la console Google Cloud , accédez à IAM et administration > Comptes de service.
  2. Cliquez sur Créer un compte de service.
  3. Fournissez les informations de configuration suivantes :
    • Nom du compte de service : saisissez beyondtrust-epm-collector-sa.
    • Description du compte de service : saisissez Service account for Cloud Run function to collect BeyondTrust EPM 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. Cliquez sur Sélectionner un rôle.
    2. Recherchez et sélectionnez Administrateur des objets Storage.
    3. Cliquez sur + Ajouter un autre rôle.
    4. Recherchez et sélectionnez Demandeur Cloud Run.
    5. Cliquez sur + Ajouter un autre rôle.
    6. Recherchez et sélectionnez Demandeur Cloud Functions.
  6. Cliquez sur Continuer.
  7. Cliquez sur OK.

Ces rôles sont requis pour :

  • Administrateur des objets Storage : écrit des journaux dans un bucket Cloud Storage et gère les fichiers d'état
  • Demandeur Cloud Run : permet à Pub/Sub d'appeler la fonction
  • Demandeur Cloud Functions : permet d'appeler des fonctions

Accorder des autorisations IAM sur un bucket Cloud Storage

Accordez au compte de service des autorisations d'écriture sur le bucket Cloud Storage :

  1. Accédez à Cloud Storage > Buckets.
  2. Cliquez sur le nom de votre bucket.
  3. Accédez à l'onglet Autorisations.
  4. Cliquez sur Accorder l'accès.
  5. Fournissez les informations de configuration suivantes :
    • Ajouter des comptes principaux : saisissez l'adresse e-mail du compte de service. Exemple :beyondtrust-epm-collector-sa@PROJECT_ID.iam.gserviceaccount.com
    • Attribuer des rôles : sélectionnez Administrateur des objets Storage.
  6. Cliquez sur Enregistrer.

Créer un sujet Pub/Sub

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

  1. Dans la consoleGoogle Cloud , accédez à Pub/Sub > Sujets.
  2. Cliquez sur Créer un sujet.
  3. Fournissez les informations de configuration suivantes :
    • ID du sujet : saisissez beyondtrust-epm-trigger.
    • Conservez les valeurs par défaut des autres paramètres.
  4. Cliquez sur Créer.

Créer une fonction Cloud Run pour collecter les journaux

La fonction Cloud Run est déclenchée par des messages Pub/Sub provenant de Cloud Scheduler pour extraire les journaux de l'API BeyondTrust EPM et les écrire dans Cloud Storage.

  1. Dans la console Google Cloud , accédez à Cloud Run.
  2. Cliquez sur Créer un service.
  3. Sélectionnez Fonction. Utilisez un éditeur intégré pour créer une fonction.

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

    Paramètre Valeur
    Nom du service beyondtrust-epm-collector
    Région Sélectionnez la région correspondant à votre bucket Cloud Storage (par exemple, us-central1).
    Durée d'exécution Sélectionnez Python 3.12 ou version ultérieure.

  5. Dans la section Déclencheur (facultatif) :

    1. Cliquez sur + Ajouter un déclencheur.
    2. Sélectionnez Cloud Pub/Sub.
    3. Dans Sélectionner un sujet Cloud Pub/Sub, choisissez le sujet beyondtrust-epm-trigger.
    4. Cliquez sur Enregistrer.

  6. Dans la section Authentification :

    1. Sélectionnez Exiger l'authentification.
    2. Consultez Identity and Access Management (IAM).

  7. Accédez à Conteneurs, mise en réseau, sécurité et développez cette section.

  8. Accédez à l'onglet Sécurité :

    • Compte de service : sélectionnez le compte de service beyondtrust-epm-collector-sa.

  9. Accédez à l'onglet Conteneurs :

    1. Cliquez sur Variables et secrets.
    2. Cliquez sur + Ajouter une variable pour chaque variable d'environnement :
    Nom de la variable Exemple de valeur
    GCS_BUCKET beyondtrust-epm-logs
    GCS_PREFIX beyondtrust-epm/
    STATE_KEY beyondtrust-epm-state.json
    BPT_API_URL https://yourtenant-services.pm.beyondtrustcloud.com
    CLIENT_ID your-client-id
    CLIENT_SECRET your-client-secret
    RECORD_SIZE 1000
    MAX_BATCHES 50
    LOOKBACK_HOURS 24
    • RECORD_SIZE : enregistrements par requête. /Events/FromStartDate accepte les valeurs comprises entre 1 et 1 000. La fonction bloque les valeurs supérieures.
    • MAX_BATCHES : limite de sécurité sur le nombre de requêtes par exécution. Si une exécution l'atteint, le curseur avance quand même jusqu'à ce qui a été écrit, de sorte que l'exécution suivante se poursuit à partir de là au lieu de répéter le travail.
    • LOOKBACK_HOURS : période couverte par la première exécution. BeyondTrust conserve les événements passifs (codes 106, 107, 603, 706) pendant 30 jours et tous les autres événements pendant 90 jours. Par conséquent, une période d'analyse plus longue ne renvoie aucun résultat plus ancien.
    • LOOKBACK_HOURS : délai à remonter lors de la toute première exécution, avant qu'un curseur n'existe.

  10. Dans l'onglet Variables et secrets, accédez à Demandes :

    • Délai avant expiration de la requête : saisissez 600 secondes (10 minutes).

  11. Accédez à l'onglet Paramètres dans Conteneurs :

    • Dans la section Ressources :
      • Mémoire : sélectionnez 512 Mio ou plus.
      • CPU : sélectionnez 1.
    • Cliquez sur OK.

  12. Faites défiler la page jusqu'à Environnement d'exécution :

    • Sélectionnez Par défaut (recommandé).

  13. Dans la section Scaling de révision :

    • Nombre minimal d'instances : saisissez 0.
    • Nombre maximal d'instances : saisissez 100 (ou ajustez en fonction de la charge attendue).

  14. Cliquez sur Créer.

  15. Attendez que le service soit créé (une à deux minutes).

  16. Une fois le service créé, l'éditeur de code intégré s'ouvre automatiquement.

Ajouter un code de fonction

  1. Saisissez main dans Point d'entrée de la fonction.

  2. Dans l'éditeur de code intégré, créez deux fichiers :

    • Premier fichier : main.py:
    import hashlib
import json
import os
import re
import time
import urllib3
import uuid
from datetime import datetime, timedelta, timezone
from urllib.parse import urlencode

import functions_framework
from google.cloud import storage
from google.cloud.exceptions import NotFound

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

# Initialize Storage client
storage_client = storage.Client()

TOKEN_PATH = '/oauth/connect/token'
# Events/FromStartDate is the endpoint BeyondTrust documents for bulk extraction:
# "A new API is exposed to extract the events in bulk." It takes only StartDate and
# RecordSize. Events/search is not usable here: it requires an OperatingSystem value,
# it is a single string with no documented "all" value, so it can only ever return one
# operating system per call.
EVENTS_PATH = '/management-api/v3/Events/FromStartDate'
OAUTH_SCOPE = 'urn:management:api'
# RecordSize accepts 1 to 1000.
MAX_RECORD_SIZE = 1000

class FetchError(Exception):
    """Raised when the BeyondTrust API call fails.

    The cursor must never advance on a failed fetch, otherwise the events in the
    failed window are skipped permanently.
    """

def rfc3339(dt: datetime) -> str:
    """Render a datetime the way StartDate accepts it: milliseconds and a literal Z."""
    return dt.astimezone(timezone.utc).strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] + 'Z'

def parse_iso(stamp: str) -> datetime:
    """Parse any ISO 8601 timestamp shape the API emits.

    event.ingested comes back with a +00:00 offset and up to seven fractional
    digits, e.g. 2026-08-03T13:29:55.1109163+00:00. fromisoformat's tolerance for
    long fractions varies across Python versions, so the fraction is trimmed to
    microseconds first.
    """
    text = stamp.strip()
    if text.endswith('Z'):
        text = text[:-1] + '+00:00'
    text = re.sub(r'\.(\d{1,6})\d*', r'.\1', text, count=1)
    parsed = datetime.fromisoformat(text)
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
    return parsed

def canonical(stamp: str) -> str:
    """Re-render an API timestamp into the one form StartDate accepts.

    The API returns event.ingested as +00:00-offset ISO 8601 but rejects that
    same shape as a StartDate value with 400 "Invalid Start date format": it
    accepts only Z-suffixed values. Every timestamp that came from the API must
    pass through here before being sent back or persisted.
    """
    return rfc3339(parse_iso(stamp))

def event_id(evt: dict) -> str:
    """Return the event's identity for deduplication.

    Falls back to a content hash when event.id is absent, so an id-less event
    still deduplicates instead of being re-ingested on every boundary re-read.
    """
    explicit = str((evt.get('event') or {}).get('id') or '')
    if explicit:
        return explicit
    digest = hashlib.sha256(
        json.dumps(evt, sort_keys=True, ensure_ascii=False).encode('utf-8')
    ).hexdigest()
    return f'sha256:{digest}'

def event_ingested(evt: dict) -> str:
    """Return the Elastic ingestion timestamp, which is what StartDate filters on."""
    return str((evt.get('event') or {}).get('ingested') or '')

def next_millisecond(stamp: str) -> str:
    """Return the canonical timestamp one millisecond later.

    Used only when a full batch fits inside a single ingestion millisecond. Without
    this the cursor cannot move and the collector stalls on that timestamp forever.
    """
    return rfc3339(parse_iso(stamp) + timedelta(milliseconds=1))

@functions_framework.cloud_event
def main(cloud_event):
    """Fetch BeyondTrust EPM events and write them to Cloud Storage as NDJSON.

    Args:
            cloud_event: CloudEvent object containing the Pub/Sub message.
    """
    bucket_name = os.environ.get('GCS_BUCKET')
    prefix = os.environ.get('GCS_PREFIX', 'beyondtrust-epm/')
    state_key = os.environ.get('STATE_KEY', 'beyondtrust-epm-state.json')

    api_url = (os.environ.get('BPT_API_URL') or '').rstrip('/')
    client_id = os.environ.get('CLIENT_ID')
    client_secret = os.environ.get('CLIENT_SECRET')
    # Clamp both ends: the API rejects RecordSize outside 1 to 1000, and a
    # misconfigured 0 would otherwise crash-loop on a 400 every run.
    record_size = min(max(int(os.environ.get('RECORD_SIZE', '1000')), 1), MAX_RECORD_SIZE)
    max_batches = int(os.environ.get('MAX_BATCHES', '50'))
    lookback_hours = int(os.environ.get('LOOKBACK_HOURS', '24'))

    if not all([bucket_name, api_url, client_id, client_secret]):
        raise RuntimeError(
            'Missing required environment variables: '
            'GCS_BUCKET, BPT_API_URL, CLIENT_ID, CLIENT_SECRET'
        )

    bucket = storage_client.bucket(bucket_name)
    state = load_state(bucket, state_key)

    # StartDate is a rising cursor on the Elastic ingestion timestamp, not a closed
    # window. There is no EndDate on this endpoint.

    if state.get('last_ingested'):
        # canonical() also repairs state written by the previous script revision,
        # which persisted the API's raw +00:00 form that StartDate rejects.
        try:
            start_date = canonical(state['last_ingested'])
        except (ValueError, TypeError, AttributeError) as e:
            raise RuntimeError(
                f'Unparseable last_ingested in gs://{bucket_name}/{state_key}: '
                f'{state["last_ingested"]!r}. Fix or delete that object; deleting '
                f'restarts collection from the lookback window.'
            ) from e
    else:
        start_date = rfc3339(datetime.now(timezone.utc) - timedelta(hours=lookback_hours))

    seen_ids = set(state.get('seen_ids', []))
    print(f'Collecting events ingested from {start_date}')

    token = get_oauth_token(api_url, client_id, client_secret)
    fresh, cursor, drained = fetch_events(
        api_url, token, start_date, record_size, max_batches, seen_ids
    )

    if not fresh:
        print('No new events. Cursor left unchanged.')
        return

    # Everything that can fail is computed before the upload: a crash between the
    # upload and save_state replays the batch on the next run, so the window where
    # side effects exist without recorded state must stay minimal. The retained ids
    # share the cursor millisecond, because those are exactly the ones the
    # inclusive StartDate will return again.
    retained = {
        event_id(e) for e in fresh
        if event_ingested(e) and canonical(event_ingested(e)) == cursor
    }
    if cursor == start_date:
        # The cursor millisecond did not advance, so ids retained by earlier runs
        # are still on the boundary; dropping them would re-ingest their events.
        retained |= seen_ids

    timestamp = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')
    # The random suffix keeps concurrent executions (Pub/Sub delivers at least
    # once) from overwriting each other's object within the same second.
    filename = (
        f'{prefix}beyondtrust-epm-events-{timestamp}-'
        f'{uuid.uuid4().hex[:8]}.ndjson'
    )
    ndjson = '\n'.join(json.dumps(e, ensure_ascii=False) for e in fresh) + '\n'
    bucket.blob(filename).upload_from_string(
        ndjson, content_type='application/x-ndjson'
    )
    print(f'Wrote {len(fresh)} events to gs://{bucket_name}/{filename}')

    save_state(bucket, state_key, {
        'last_ingested': cursor,
        'seen_ids': sorted(retained),
        'updated_at': rfc3339(datetime.now(timezone.utc)),
    })

    if not drained:
        print(
            f'Stopped after {max_batches} batches with more events available. '
            'The cursor advanced, so the next run continues from here.'
        )

def fetch_events(api_url, token, start_date, record_size, max_batches, seen_ids):
    """Read forward from start_date until the API returns a short batch.

    The response envelope has no page count and no next-page token, so a batch
    shorter than record_size is the only documented end-of-data signal.

    StartDate is inclusive and the cursor lands on the newest event's timestamp, so
    every batch re-returns the events sharing it. Deduplication by event id therefore
    has to happen as batches arrive, not only between runs.

    Returns:
    Tuple of (new events, cursor to persist, whether the stream was drained).

    Raises:
    FetchError: on any API or transport failure, so the caller cannot mistake a
    failed fetch for an empty result and advance the cursor.
    """
    headers = {
        'Authorization': f'Bearer {token}',
        'Accept': 'application/json',
    }

    all_events = []
    seen = set(seen_ids)
    cursor = start_date

    for batch in range(1, max_batches + 1):
        query = urlencode({'StartDate': cursor, 'RecordSize': record_size})
        body = request_with_retry(f'{api_url}{EVENTS_PATH}?{query}', headers)

        events = body.get('events') or []
        new = [e for e in events if event_id(e) not in seen]
        seen.update(event_id(e) for e in events if event_id(e))
        all_events.extend(new)
        print(f'Batch {batch}: {len(events)} events, {len(new)} new')

        # Termination is judged on what the API returned, not on what survived
        # deduplication: a batch can be full and still be entirely duplicates.
        # The newest stamp in the batch, not events[-1]: ordering is a client-side
        # convention, not a documented guarantee, and a misordered tail would
        # regress the cursor and re-ingest events already written.
        stamps = [canonical(s) for s in (event_ingested(e) for e in events) if s]

        if len(events) < record_size:
            # Short batch: the stream is drained.
            if events and not stamps:
                print(
                    'Warning: no event in the final batch carries event.ingested; '
                    'cursor left unchanged.'
                )
            if stamps:
                cursor = max(max(stamps), cursor)
            return all_events, cursor, True

        if not stamps:
            raise FetchError(
                'No event in a full batch carries event.ingested, '
                'so the cursor cannot advance'
            )
        batch_max = max(stamps)
        if batch_max <= cursor:
            # A full batch fits inside one ingestion millisecond. Stepping past it is the only
            # way to make progress; holding the cursor here stalls collection forever.
            # Events beyond record_size at that exact millisecond are unreachable, which
            # needs more than 1000 events in one millisecond.
            print(
                f'Warning: a full batch shares ingestion timestamp {batch_max}; '
                'stepping past it. Events beyond RecordSize at that timestamp are skipped.'
            )
            cursor = next_millisecond(cursor)
        else:
            cursor = batch_max

    return all_events, cursor, False

def request_with_retry(url, headers, attempts=4):
    """GET with backoff on 429 and 5xx.

    The documented rate limit is 1000 requests per 100 seconds.
    """
    backoff = 1.0
    for attempt in range(1, attempts + 1):
        try:
            response = http.request('GET', url, headers=headers)
        except Exception as e:
            raise FetchError(f'Request to {url} failed: {e}') from e

        if response.status in (429, 500, 502, 503, 504) and attempt < attempts:
            retry_after = response.headers.get('Retry-After')
            try:
                delay = int(retry_after) if retry_after else backoff
            except (TypeError, ValueError):
                delay = backoff
            # Retry-After is server-controlled input: bound it so a bogus value
            # cannot sleep past the function timeout or crash time.sleep.
            delay = min(max(delay, 1.0), 60.0)
            print(f'HTTP {response.status}. Retrying in {delay}s...')
            time.sleep(delay)
            backoff = min(backoff * 2, 30.0)
            continue

        if response.status != 200:
            raise FetchError(
                f'Request failed: {response.status} {response.data.decode("utf-8")}'
            )

        text = response.data.decode('utf-8')
        # BeyondTrust's sample folds "Owner" into "owner" before parsing, but that
        # workaround exists for PowerShell's case-insensitive ConvertFrom-Json.
        # Python parses case-sensitively, and the SecOps parser maps file.Owner.*
        # and file.owner to different UDM fields, so both keys must survive.
        try:
            return json.loads(text)
        except json.JSONDecodeError as e:
            raise FetchError(f'Malformed JSON response: {e}') from e

    raise FetchError(f'Giving up on {url} after {attempts} attempts')

def get_oauth_token(api_url, client_id, client_secret):
    """Get an access token using the OAuth client credentials flow.

    The token is valid for one hour.
    """
    body = urlencode({
        'grant_type': 'client_credentials',
        'client_id': client_id,
        'client_secret': client_secret,
        'scope': OAUTH_SCOPE,
    })
    response = http.request(
        'POST',
        f'{api_url}{TOKEN_PATH}',
        body=body,
        headers={'Content-Type': 'application/x-www-form-urlencoded'},
    )
    if response.status != 200:
        raise FetchError(
            f'Token request failed: {response.status} '
            f'{response.data.decode("utf-8")}'
        )
    return json.loads(response.data.decode('utf-8'))['access_token']

def load_state(bucket, key):
    """Read the collector state from Cloud Storage.

    Only a missing object is a cold start. Any other error is raised: swallowing it
    would reset collection to the lookback window and re-ingest that period.
    """
    blob = bucket.blob(key)
    try:
        return json.loads(blob.download_as_text())
    except NotFound:
        print('No state file found. Starting from the lookback window.')
        return {}

def save_state(bucket, key, state):
    """Write the collector state to Cloud Storage.

    Failures are raised, not logged. A run that cannot record its cursor must fail,
    otherwise the next run repeats the same window.
    """
    bucket.blob(key).upload_from_string(
        json.dumps(state, indent=2), content_type='application/json'
    )
    print(f"Saved state: last_ingested={state.get('last_ingested')}")

    • Deuxième fichier : 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.

  4. Attendez la fin du déploiement (deux à trois minutes).

Créer une tâche Cloud Scheduler

Cloud Scheduler publie des messages sur le sujet Pub/Sub à intervalles réguliers, ce qui déclenche la fonction Cloud Run.

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

  3. Fournissez les informations de configuration suivantes :

    Paramètre Valeur
    Nom beyondtrust-epm-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 Sélectionnez le thème beyondtrust-epm-trigger.
    Corps du message {} (objet JSON vide)

  4. Cliquez sur Créer.

Options de fréquence de programmation

Choisissez la fréquence en fonction du volume de journaux et des exigences de latence :

Fréquence Expression Cron Cas d'utilisation
Toutes les 5 minutes */5 * * * * Volume élevé, faible latence
Toutes les 15 minutes */15 * * * * Volume moyen
Toutes les heures 0 * * * * Standard (recommandé)
Toutes les 6 heures 0 */6 * * * Traitement par lot à faible volume
Tous les jours 0 0 * * * Collecte de données historiques

Tester le job du planificateur

  1. Dans la console Cloud Scheduler, recherchez votre job.
  2. Cliquez sur Exécuter de force pour déclencher manuellement.
  3. Patientez quelques secondes, puis accédez à Cloud Run > Services > beyondtrust-epm-collector > Journaux.
  4. Vérifiez que la fonction s'est exécutée correctement.
  5. Vérifiez le bucket Cloud Storage pour confirmer que les journaux ont été écrits.

Récupérer le compte de service Google SecOps

Google SecOps utilise un compte de service unique pour lire les données de votre bucket Cloud Storage. Vous devez accorder à ce compte de service l'accès à votre bucket.

Obtenir l'adresse e-mail du compte de service

  1. Accédez à Paramètres SIEM> Flux.
  2. Cliquez sur Ajouter un flux.
  3. Cliquez sur Configurer un flux unique.
  4. Dans le champ Nom du flux, saisissez un nom pour le flux (par exemple, BeyondTrust EPM logs).
  5. Sélectionnez Google Cloud Storage V2 comme Type de source.
  6. Sélectionnez BeyondTrust Endpoint Privilege Management comme Type de journal.

  7. Cliquez sur Obtenir un compte de service. Une adresse e-mail de compte de service unique s'affiche, par exemple :

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com

  8. Copiez cette adresse e-mail pour l'utiliser à l'étape suivante.

Accorder des autorisations IAM au compte de service Google SecOps

Le compte de service Google SecOps a besoin de deux rôles sur votre bucket Cloud Storage : Lecteur des objets Storage pour lire les objets journaux et un rôle au niveau du bucket pour lire les métadonnées du bucket.

  1. Accédez à Cloud Storage > Buckets.
  2. Cliquez sur le nom de votre bucket.
  3. Accédez à l'onglet Autorisations.
  4. Cliquez sur Accorder l'accès.
  5. Fournissez les informations de configuration suivantes :
    • Ajouter des comptes principaux : collez l'adresse e-mail du compte de service Google SecOps.
    • Attribuer des rôles : sélectionnez les deux options suivantes :
    • Lecteur des objets Storage : lit les objets journaux.
    • Lecteur des anciens buckets Storage : lit les métadonnées du bucket. Si vous avez sélectionné l'option de suppression Supprimer les fichiers transférés et les répertoires vides, sélectionnez plutôt Rédacteur de bucket Storage hérité, qui accorde également l'autorisation de suppression.

  6. Cliquez sur Enregistrer.

Configurer un flux dans Google SecOps pour ingérer les journaux BeyondTrust EPM

  1. Accédez à Paramètres SIEM> Flux.
  2. Cliquez sur Ajouter un flux.
  3. Cliquez sur Configurer un flux unique.
  4. Dans le champ Nom du flux, saisissez un nom pour le flux (par exemple, BeyondTrust EPM logs).
  5. Sélectionnez Google Cloud Storage V2 comme Type de source.
  6. Sélectionnez BeyondTrust Endpoint Privilege Management comme Type de journal.
  7. Cliquez sur Suivant.

  8. Spécifiez les valeurs des paramètres d'entrée suivants :

    • URL du bucket de stockage : saisissez l'URI du bucket Cloud Storage avec le chemin d'accès au préfixe :

      gs://beyondtrust-epm-logs/beyondtrust-epm/

      • Remplacer :

        • beyondtrust-epm-logs : nom de votre bucket Cloud Storage.
        • beyondtrust-epm/ : préfixe/chemin d'accès au dossier facultatif où les journaux sont stockés (laisser vide pour la racine).

      • Exemples :

        • Bucket racine : gs://beyondtrust-epm-logs/
        • Avec le préfixe : gs://beyondtrust-epm-logs/beyondtrust-epm/

    • Option de suppression de la source : sélectionnez l'option de suppression de votre choix :

      • Ne jamais supprimer les fichiers : ne supprime jamais les fichiers de la source. Cette option est recommandée à des fins de test.

      • Supprimer les fichiers transférés et les répertoires vides : supprime les fichiers et les répertoires vides de la source une fois l'extraction réussie.

    • Âge maximal du fichier : incluez les fichiers modifiés au cours des derniers jours. La valeur par défaut est de 180 jours.

    • Espace de noms de l'élément : espace de noms de l'élément.

    • Libellés d'ingestion : libellé à appliquer aux événements de ce flux.

  9. Cliquez sur Suivant.

  10. Vérifiez la configuration de votre nouveau flux sur l'écran Finaliser, puis cliquez sur Envoyer.

Table de mappage UDM

Champ de journal Mappage UDM Logique
DomainIdentifier_label additional.fields Fusionné
WatsonEventType_label additional.fields Fusionné
action_label additional.fields Fusionné
activity_id_label additional.fields Fusionné
app_id_label additional.fields Fusionné
app_insights_instrumentation_key_label additional.fields Fusionné
asimov_instrumentation_key_label additional.fields Fusionné
authorization_request_control_authorization_label additional.fields Fusionné
bundle_name_label additional.fields Fusionné
changedBy_label additional.fields Fusionné
code_signature_subject_name_label additional.fields Fusionné
collector_api_key_label additional.fields Fusionné
command_label additional.fields Fusionné
configuration_application_group_description_label additional.fields Fusionné
configuration_application_group_identifier_label additional.fields Fusionné
configuration_application_group_name_label additional.fields Fusionné
configuration_application_identifier_label additional.fields Fusionné
configuration_application_type_label additional.fields Fusionné
configuration_identifier_label additional.fields Fusionné
configuration_message_identifier_label additional.fields Fusionné
configuration_message_name_label additional.fields Fusionné
configuration_message_type_label additional.fields Fusionné
configuration_name_label additional.fields Fusionné
configuration_revision_number_label additional.fields Fusionné
configuration_rule_identifier_label additional.fields Fusionné
configuration_rule_on_demand_label additional.fields Fusionné
configuration_rule_script_outcome_rule_affected_label additional.fields Fusionné
configuration_token_identifier_label additional.fields Fusionné
configuration_token_name_label additional.fields Fusionné
configuration_workstyle_description_label additional.fields Fusionné
configuration_workstyle_identifier_label additional.fields Fusionné
configuration_workstyle_name_label additional.fields Fusionné
content_length_label additional.fields Fusionné
content_type_label additional.fields Fusionné
domainNetBIOSName_label additional.fields Fusionné
domain_label additional.fields Fusionné
entity_label additional.fields Fusionné
entity_name_label additional.fields Fusionné
event_action_label additional.fields Fusionné
exitstatus_label additional.fields Fusionné
file_hash_md5_label additional.fields Fusionné
file_hash_sha1_label additional.fields Fusionné
file_hash_sha256_label additional.fields Fusionné
file_version_label additional.fields Fusionné
gid_label additional.fields Fusionné
group_data_id_label additional.fields Fusionné
group_id_label additional.fields Fusionné
group_label additional.fields Fusionné
handle_label additional.fields Fusionné
host_name_label additional.fields Fusionné
host_uptime_labels additional.fields Fusionné
http_host_label additional.fields Fusionné
id_label additional.fields Fusionné
iolog_label additional.fields Fusionné
is_opted_in_label additional.fields Fusionné
linenum_label additional.fields Fusionné
local_identifier_label additional.fields Fusionné
locale_label additional.fields Fusionné
master_utcoffset_label additional.fields Fusionné
masterlocale_label additional.fields Fusionné
owner_identifier_label additional.fields Fusionné
parent_entity_id_label additional.fields Fusionné
parent_process_exec_label additional.fields Fusionné
parent_process_label additional.fields Fusionné
pbmasterdnodename_label additional.fields Fusionné
pipeName_label additional.fields Fusionné
process_entity_id_label additional.fields Fusionné
process_hash_label additional.fields Fusionné
process_name_label additional.fields Fusionné
process_parent_name_label additional.fields Fusionné
process_start_time_label additional.fields Fusionné
processexe_label additional.fields Fusionné
product_label additional.fields Fusionné
product_type_label additional.fields Fusionné
product_version_label additional.fields Fusionné
requestuser_label additional.fields Fusionné
runargv_label additional.fields Fusionné
runcwd_label additional.fields Fusionné
runeffectivegroup_label additional.fields Fusionné
runeffectiveuser_label additional.fields Fusionné
runhost_label additional.fields Fusionné
schema_version_label additional.fields Fusionné
sku_name_label additional.fields Fusionné
telemetry_level_label additional.fields Fusionné
tenant_id_label additional.fields Fusionné
type_label additional.fields Fusionné
uid_label additional.fields Fusionné
user_id_label additional.fields Fusionné
user_name_label additional.fields Fusionné
vs_exe_version_label additional.fields Fusionné
inter_host intermediary.hostname Mappé directement
Processes.description metadata.description Mappé directement
details metadata.description Mappé directement
event_data.reason metadata.description Mappé directement
file.pe.description metadata.description Mappé directement
created metadata.event_timestamp Analysé en tant que ISO8601
datetime metadata.event_timestamp Analysé en tant que MMM dd HH:mm:ss
has_principal metadata.event_type Mappé : true → STATUS_UPDATE
has_user metadata.event_type Mappé : true → USER_UNCATEGORIZED
parent_working_directory_label metadata.ingestion_labels Fusionné
working_directory_label metadata.ingestion_labels Fusionné
auditType metadata.product_event_type Mappé directement
event_datas.ActionId metadata.product_log_id Mappé directement
labels.related_item_id metadata.product_log_id Mappé directement
uniqueid metadata.product_log_id Mappé directement
masterdversion metadata.product_version Mappé directement
headers.http_version network.application_protocol_version Mappé directement
headers.request_method network.http.method Mappé directement
host.os.platform principal.administrative_domain Mappé directement
Processes.process principal.application Mappé directement
agent_ephemeral_id_label principal.asset.attribute.labels Fusionné
agent_id_label principal.asset.attribute.labels Fusionné
agent_version_label principal.asset.attribute.labels Fusionné
ecs_version_label principal.asset.attribute.labels Fusionné
_hardware principal.asset.hardware Fusionné
host.hostname principal.asset.hostname Mappé directement
submithost principal.asset.hostname Mappé directement
ip_address principal.asset.ip Fusionné
masterhostip principal.asset.ip Fusionné
submithostip principal.asset.ip Fusionné
file.path principal.file.full_path Mappé directement
lineinfile principal.file.full_path Mappé directement
host.hostname principal.hostname Mappé directement
submithost principal.hostname Mappé directement
ip_address principal.ip Fusionné
masterhostip principal.ip Fusionné
submithostip principal.ip Fusionné
mac principal.mac Fusionné
host.os.name principal.platform_version Mappé directement
host.os.version principal.platform_version Mappé directement
process.command_line principal.process.command_line Mappé directement
runcommand principal.process.command_line Mappé directement
process.executable principal.process.file.full_path Mappé directement
cmd principal.process.parent_process.command_line Mappé directement
Processes.process_path principal.process.parent_process.file.full_path Mappé directement
process.parent.executable principal.process.parent_process.file.full_path Mappé directement
Processes.parent_process_id principal.process.parent_process.pid Mappé directement
Processes.process_id principal.process.pid Mappé directement
logpid principal.process.pid Mappé directement
file.Owner.DomainName principal.user.company_name Mappé directement
file.Owner.Name principal.user.user_display_name Mappé directement
runuser principal.user.user_display_name Mappé directement
userName principal.user.user_display_name Mappé directement
Processes.user principal.user.userid Mappé directement
userId principal.user.userid Mappé directement
userid principal.user.userid Mappé directement
Processes.user_id principal.user.windows_sid Mappé directement
EPMWinMac.Configuration.Rule.Action security_result.action Fusionné
security_result_action security_result.action Fusionné
event_data.outcome security_result.category_details Fusionné
host.os.version security_result.category_details Fusionné
EPMWinMac.Configuration.Application.Description security_result.description Mappé directement
EPMWinMac.Configuration.Message.Description security_result.description Mappé directement
host.os.type src.administrative_domain Mappé directement
file.name src.file.names Fusionné
host.os.full src.platform_version Mappé directement
host.os.family target.administrative_domain Mappé directement
Processes.dest target.asset.hostname Mappé directement
file.extension target.file.mime_type Mappé directement
Processes.dest target.hostname Mappé directement
host.domain target.hostname Mappé directement
file_DriveType_label target.resource.attribute.labels Fusionné
file_drive_letter_label target.resource.attribute.labels Fusionné
owner_label target.resource.attribute.labels Fusionné
N/A metadata.event_type Constante : USER_UNCATEGORIZED
N/A metadata.product_name Constante : Beyondtrust Privilege Management
N/A metadata.vendor_name Constante : Beyondtrust Privilege Management
N/A network.application_protocol Constante : HTTP
N/A principal.platform Constante : MAC

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.