Raccogli i log di BeyondTrust Endpoint Privilege Management (EPM)

Supportata in:

Questo documento spiega come importare i log di BeyondTrust Endpoint Privilege Management (EPM) in Google Security Operations utilizzando Cloud Storage. Il parser si concentra sulla trasformazione dei dati di log JSON non elaborati di BeyondTrust Endpoint in un formato strutturato conforme all'UDM di Google SecOps. Inizializza i valori predefiniti per vari campi e poi analizza il payload JSON, mappando successivamente campi specifici del log non elaborato nei campi UDM corrispondenti all'interno dell'oggetto event.idm.read_only_udm.

Prima di iniziare

Assicurati di soddisfare i seguenti prerequisiti:

  • Un'istanza Google SecOps
  • Un progetto Google Cloud con l'API Cloud Storage abilitata
  • Autorizzazioni per creare e gestire i bucket Cloud Storage
  • Autorizzazioni per gestire le policy di Identity and Access Management (IAM) sui bucket Cloud Storage
  • Autorizzazioni per creare servizi Cloud Run Functions, argomenti Pub/Sub e job Cloud Scheduler
  • Accesso con privilegi al tenant o all'API BeyondTrust Endpoint Privilege Management

Crea un bucket Cloud Storage

  1. Vai alla consoleGoogle Cloud .
  2. Seleziona il tuo progetto o creane uno nuovo.
  3. Nel menu di navigazione, vai a Cloud Storage > Bucket.
  4. Fai clic su Crea bucket.

  5. Fornisci i seguenti dettagli di configurazione:

    Impostazione Valore
    Assegna un nome al bucket Inserisci un nome globalmente univoco (ad esempio beyondtrust-epm-logs)
    Tipo di località Scegli in base alle tue esigenze (Regione, a due regioni, multiregionale)
    Località Seleziona la località (ad esempio us-central1)
    Classe di archiviazione Standard (consigliato per i log ad accesso frequente)
    Controllo degli accessi Uniforme (consigliato)
    Strumenti di protezione (Facoltativo) Abilita il controllo delle versioni degli oggetti o la policy di conservazione

  6. Fai clic su Crea.

Raccogli le credenziali API di BeyondTrust EPM

  1. Accedi alla console web di BeyondTrust Privilege Management come amministratore.
  2. Vai a Configuration (Configurazione) > Settings (Impostazioni) > API Settings (Impostazioni API).
  3. Fai clic su Crea un account API.
  4. Fornisci i seguenti dettagli di configurazione:
    • Nome: inserisci Google SecOps Collector.
    • Accesso API: attiva Reporting (sola lettura). L'endpoint /management-api/v3/Events/FromStartDate utilizzato da questo raccoglitore appartiene alla categoria di autorizzazioni Reporting. Attiva anche Audit (Read Only) (Controllo (sola lettura)) solo se intendi raccogliere anche i controlli delle attività della console web.
  5. Copia e salva l'ID client e il client secret.

  6. Copia l'URL di base dell'API mostrato nella parte superiore della pagina Impostazioni API. In genere si tratta di https://<your-tenant>-services.pm.beyondtrustcloud.com. Utilizzerai questo indirizzo come BPT_API_URL.

Crea un account di servizio per la funzione Cloud Run

La funzione Cloud Run ha bisogno di un account di servizio con le autorizzazioni per scrivere nel bucket Cloud Storage ed essere richiamata da Pub/Sub.

Crea il account di servizio

  1. Nella console Google Cloud , vai a IAM e amministrazione > Account di servizio.
  2. Fai clic su Crea account di servizio.
  3. Fornisci i seguenti dettagli di configurazione:
    • Nome service account: inserisci beyondtrust-epm-collector-sa.
    • Descrizione service account: inserisci Service account for Cloud Run function to collect BeyondTrust EPM logs.
  4. Fai clic su Crea e continua.
  5. Nella sezione Concedi a questo account di servizio l'accesso al progetto, aggiungi i seguenti ruoli:
    1. Fai clic su Seleziona un ruolo.
    2. Cerca e seleziona Storage Object Admin.
    3. Fai clic su + Aggiungi un altro ruolo.
    4. Cerca e seleziona Cloud Run Invoker.
    5. Fai clic su + Aggiungi un altro ruolo.
    6. Cerca e seleziona Cloud Functions Invoker.
  6. Fai clic su Continua.
  7. Fai clic su Fine.

Questi ruoli sono obbligatori per:

  • Storage Object Admin: scrive i log in un bucket Cloud Storage e gestisce i file di stato
  • Cloud Run Invoker: consente a Pub/Sub di richiamare la funzione
  • Cloud Functions Invoker: consente la chiamata delle funzioni

Concedi le autorizzazioni IAM su un bucket Cloud Storage

Concedi al account di servizio le autorizzazioni in scrittura sul bucket Cloud Storage:

  1. Vai a Cloud Storage > Bucket.
  2. Fai clic sul nome del bucket.
  3. Vai alla scheda Autorizzazioni.
  4. Fai clic su Concedi l'accesso.
  5. Fornisci i seguenti dettagli di configurazione:
    • Aggiungi entità: inserisci l'indirizzo email del account di servizio. Ad esempio, beyondtrust-epm-collector-sa@PROJECT_ID.iam.gserviceaccount.com.
    • Assegna i ruoli: seleziona Storage Object Admin.
  6. Fai clic su Salva.

Crea un argomento Pub/Sub

Crea un argomento Pub/Sub in cui Cloud Scheduler pubblicherà e a cui la funzione Cloud Run si iscriverà.

  1. Nella consoleGoogle Cloud , vai a Pub/Sub > Argomenti.
  2. Fai clic su Crea argomento.
  3. Fornisci i seguenti dettagli di configurazione:
    • ID argomento: inserisci beyondtrust-epm-trigger.
    • Lascia invariate le altre impostazioni predefinite.
  4. Fai clic su Crea.

Crea una funzione Cloud Run per raccogliere i log

La funzione Cloud Run viene attivata dai messaggi Pub/Sub di Cloud Scheduler per recuperare i log dall'API BeyondTrust EPM e li scrive in Cloud Storage.

  1. Nella console Google Cloud , vai a Cloud Run.
  2. Fai clic su Crea servizio.
  3. Seleziona Funzione. Usa un editor per la modifica in linea per creare una funzione.

  4. Nella sezione Configura, fornisci i seguenti dettagli di configurazione:

    Impostazione Valore
    Nome servizio beyondtrust-epm-collector
    Regione Seleziona la regione corrispondente al tuo bucket Cloud Storage (ad esempio us-central1)
    Tempo di esecuzione Seleziona Python 3.12 o versioni successive

  5. Nella sezione Trigger (facoltativo):

    1. Fai clic su + Aggiungi trigger.
    2. Seleziona Cloud Pub/Sub.
    3. In Seleziona un argomento Cloud Pub/Sub, scegli l'argomento beyondtrust-epm-trigger.
    4. Fai clic su Salva.

  6. Nella sezione Autenticazione:

    1. Seleziona Richiedi autenticazione.
    2. Controlla Identity and Access Management (IAM).

  7. Vai a Container, networking, sicurezza ed espandi la sezione.

  8. Vai alla scheda Sicurezza:

    • Service account: seleziona il account di servizio beyondtrust-epm-collector-sa.

  9. Vai alla scheda Container:

    1. Fai clic su Variabili e secret.
    2. Fai clic su + Aggiungi variabile per ogni variabile di ambiente:
    Nome variabile Valore di esempio
    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: record per richiesta. /Events/FromStartDate accetta da 1 a 1000; la funzione blocca i valori più alti.
    • MAX_BATCHES: limite di sicurezza per le richieste per esecuzione. Se un'esecuzione raggiunge il limite, il cursore avanza comunque fino a ciò che è stato scritto, quindi l'esecuzione successiva continua da lì anziché ripetere il lavoro.
    • LOOKBACK_HOURS: l'intervallo di tempo coperto dalla prima esecuzione. BeyondTrust conserva gli eventi passivi (codici 106, 107, 603, 706) per 30 giorni e tutti gli altri eventi per 90 giorni, quindi un periodo di analisi più lungo non restituisce nulla di più vecchio.
    • LOOKBACK_HOURS: quanto indietro andare alla prima esecuzione, prima che esista un cursore.

  10. Nella scheda Variabili e secret, vai a Richieste:

    • Timeout della richiesta: inserisci 600 secondi (10 minuti).

  11. Vai alla scheda Impostazioni in Container:

    • Nella sezione Risorse:
      • Memoria: seleziona 512 MiB o un valore superiore.
      • CPU: seleziona 1.
    • Fai clic su Fine.

  12. Scorri fino ad Ambiente di esecuzione:

    • Seleziona Predefinita (opzione consigliata).

  13. Nella sezione Scalabilità della revisione:

    • Numero minimo di istanze: inserisci 0.
    • Numero massimo di istanze: inserisci 100 (o regola in base al carico previsto).

  14. Fai clic su Crea.

  15. Attendi la creazione del servizio (1-2 minuti).

  16. Dopo aver creato il servizio, l'editor di codice incorporato si apre automaticamente.

Aggiungi il codice per la funzione

  1. Inserisci main in Entry point della funzione

  2. Nell'editor di codice incorporato, crea due file:

    • Primo file - 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')}")

    • Secondo file: requirements.txt:
    functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0

  3. Fai clic su Esegui il deployment per salvare la funzione ed eseguirne il deployment.

  4. Attendi il completamento del deployment (2-3 minuti).

Crea un job Cloud Scheduler

Cloud Scheduler pubblica messaggi nell'argomento Pub/Sub a intervalli regolari, attivando la funzione Cloud Run.

  1. Nella console Google Cloud , vai a Cloud Scheduler.
  2. Fai clic su Crea job.

  3. Fornisci i seguenti dettagli di configurazione:

    Impostazione Valore
    Nome beyondtrust-epm-collector-hourly
    Regione Seleziona la stessa regione della funzione Cloud Run
    Frequenza 0 * * * * (ogni ora, allo scoccare dell'ora)
    Fuso orario Seleziona il fuso orario (consigliato UTC)
    Tipo di target Pub/Sub
    Argomento Seleziona l'argomento beyondtrust-epm-trigger
    Corpo del messaggio {} (oggetto JSON vuoto)

  4. Fai clic su Crea.

Opzioni di frequenza di pianificazione

Scegli la frequenza in base al volume dei log e ai requisiti di latenza:

Frequenza Espressione cron Caso d'uso
Ogni 5 minuti */5 * * * * Volume elevato, bassa latenza
Ogni 15 minuti */15 * * * * Volume medio
Ogni ora 0 * * * * Standard (consigliato)
Ogni 6 ore 0 */6 * * * Basso volume, elaborazione batch
Ogni giorno 0 0 * * * Raccolta dei dati storici

Testa il job dello scheduler

  1. Nella console Cloud Scheduler, trova il tuo job.
  2. Fai clic su Forza esecuzione per attivare manualmente.
  3. Attendi qualche secondo e vai a Cloud Run > Servizi > beyondtrust-epm-collector > Log.
  4. Verifica che la funzione sia stata eseguita correttamente.
  5. Controlla il bucket Cloud Storage per verificare che i log siano stati scritti.

Recupera il account di servizio Google SecOps

Google SecOps utilizza un account di servizio univoco per leggere i dati dal bucket Cloud Storage. Devi concedere a questo account di servizio l'accesso al tuo bucket.

Recupera l'indirizzo email del account di servizio

  1. Vai a Impostazioni > Gestione dei dati > Feed. Se utilizzi una piattaforma solo SIEM, vai a Impostazioni > Feed.
  2. Fai clic su Aggiungi nuovo feed.
  3. Fai clic su Configura un singolo feed.
  4. Nel campo Nome feed, inserisci un nome per il feed (ad es. BeyondTrust EPM logs).
  5. Seleziona Google Cloud Storage V2 come Tipo di origine.
  6. Seleziona BeyondTrust Endpoint Privilege Management come Tipo di log.

  7. Fai clic su Ottieni service account. Verrà visualizzato un indirizzo email univoco del account di servizio, ad esempio:

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

  8. Copia questo indirizzo email per utilizzarlo nel passaggio successivo.

Concedi le autorizzazioni IAM al account di servizio Google SecOps

Il account di servizio Google SecOps richiede due ruoli nel bucket Cloud Storage: Storage Object Viewer per leggere gli oggetti di log e un ruolo a livello di bucket per leggere i metadati del bucket.

  1. Vai a Cloud Storage > Bucket.
  2. Fai clic sul nome del bucket.
  3. Vai alla scheda Autorizzazioni.
  4. Fai clic su Concedi l'accesso.
  5. Fornisci i seguenti dettagli di configurazione:
    • Aggiungi entità: incolla l'indirizzo email del account di servizio Google SecOps.
    • Assegna ruoli: seleziona entrambe le seguenti opzioni:
    • Storage Object Viewer: legge gli oggetti di log.
    • Storage Legacy Bucket Reader: legge i metadati del bucket. Se hai selezionato l'opzione di eliminazione Elimina i file trasferiti e le directory vuote, seleziona Storage Legacy Bucket Writer, che concede anche l'autorizzazione di eliminazione.

  6. Fai clic su Salva.

Configura un feed in Google SecOps per importare i log di BeyondTrust EPM

  1. Vai a Impostazioni > Gestione dei dati > Feed. Se utilizzi una piattaforma solo SIEM, vai a Impostazioni > Feed.
  2. Fai clic su Aggiungi nuovo feed.
  3. Fai clic su Configura un singolo feed.
  4. Nel campo Nome feed, inserisci un nome per il feed (ad es. BeyondTrust EPM logs).
  5. Seleziona Google Cloud Storage V2 come Tipo di origine.
  6. Seleziona BeyondTrust Endpoint Privilege Management come Tipo di log.
  7. Fai clic su Avanti.

  8. Specifica i valori per i seguenti parametri di input:

    • URL bucket di archiviazione: inserisci l'URI del bucket Cloud Storage con il percorso del prefisso:

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

      • Sostituisci:

        • beyondtrust-epm-logs: il nome del tuo bucket Cloud Storage.
        • beyondtrust-epm/: prefisso/percorso della cartella facoltativo in cui sono archiviati i log (lascia vuoto per la directory principale).

      • Esempi:

        • Bucket principale: gs://beyondtrust-epm-logs/
        • Con prefisso: gs://beyondtrust-epm-logs/beyondtrust-epm/

    • Opzione di eliminazione dell'origine: seleziona l'opzione di eliminazione in base alle tue preferenze:

      • Non eliminare mai i file: non eliminare mai i file dall'origine. Questa opzione è consigliata per i test.

      • Elimina i file trasferiti e le directory vuote: elimina i file e le directory vuote dall'origine dopo il completamento di un recupero riuscito.

    • Età massima del file: includi i file modificati negli ultimi giorni. Il valore predefinito è 180 giorni.

    • Spazio dei nomi dell'asset: lo spazio dei nomi dell'asset.

    • Etichette di importazione: l'etichetta da applicare agli eventi di questo feed.

  9. Fai clic su Avanti.

  10. Controlla la nuova configurazione del feed di notizie nella schermata Finalizza e poi fai clic su Invia.

Tabella di mappatura UDM

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

Log delle modifiche

Visualizza il log delle modifiche per questo parser

Hai bisogno di ulteriore assistenza? Ricevi risposte dai membri della community e dai professionisti di Google SecOps.