BeyondTrust Endpoint Privilege Management (EPM)-Logs erfassen

Unterstützt in:

In diesem Dokument wird beschrieben, wie Sie BeyondTrust Endpoint Privilege Management (EPM)-Logs mit Cloud Storage in Google Security Operations aufnehmen. Der Parser konzentriert sich auf die Umwandlung von JSON-Logrohdaten von BeyondTrust Endpoint in ein strukturiertes Format, das dem Google SecOps UDM entspricht. Zuerst werden Standardwerte für verschiedene Felder initialisiert. Anschließend wird die JSON-Nutzlast geparst und bestimmte Felder aus dem Rohlog werden den entsprechenden UDM-Feldern im event.idm.read_only_udm-Objekt zugeordnet.

Hinweis

Prüfen Sie, ob folgende Voraussetzungen erfüllt sind:

  • Eine Google SecOps-Instanz
  • Ein Google Cloud -Projekt mit aktivierter Cloud Storage API
  • Berechtigungen zum Erstellen und Verwalten von Cloud Storage-Buckets
  • Berechtigungen zum Verwalten von IAM-Richtlinien (Identity and Access Management) für Cloud Storage-Buckets
  • Berechtigungen zum Erstellen von Cloud Run-Funktionsdiensten, Pub/Sub-Themen und Cloud Scheduler-Jobs
  • Privilegierter Zugriff auf den BeyondTrust Endpoint Privilege Management-Mandanten oder die API

Cloud Storage-Bucket erstellen

  1. Rufen Sie die Google Cloud Console auf.
  2. Wählen Sie Ihr Projekt aus oder erstellen Sie ein neues.
  3. Rufen Sie im Navigationsmenü Cloud Storage > Buckets auf.
  4. Klicken Sie auf Bucket erstellen.

  5. Geben Sie die folgenden Konfigurationsdetails an:

    Einstellung Wert
    Bucket benennen Geben Sie einen global eindeutigen Namen ein, z. B. beyondtrust-epm-logs.
    Standorttyp Wählen Sie die Option aus, die am besten zu Ihren Anforderungen passt (Region, Dual-Region, Multi-Region).
    Standort Wählen Sie den Standort aus, z. B. us-central1.
    Speicherklasse Standard (empfohlen für Logs, auf die häufig zugegriffen wird)
    Zugriffskontrolle Einheitlich (empfohlen)
    Schutzmaßnahmen Optional: Objektversionsverwaltung oder Aufbewahrungsrichtlinie aktivieren

  6. Klicken Sie auf Erstellen.

BeyondTrust EPM API-Anmeldedaten erfassen

  1. Melden Sie sich als Administrator in der BeyondTrust Privilege Management-Webkonsole an.
  2. Rufen Sie Konfiguration > Einstellungen > API-Einstellungen auf.
  3. Klicken Sie auf API-Konto erstellen.
  4. Geben Sie die folgenden Konfigurationsdetails an:
    • Name: Geben Sie Google SecOps Collector ein.
    • API-Zugriff: Aktivieren Sie Berichterstellung (schreibgeschützt). Der von diesem Collector verwendete /management-api/v3/Events/FromStartDate-Endpunkt gehört zur Berechtigungskategorie Berichterstellung. Aktivieren Sie Audit (Read Only) nur, wenn Sie auch Audit-Logs für Webkonsolenaktivitäten erfassen möchten.
  5. Kopieren und speichern Sie die Client-ID und den Clientschlüssel.

  6. Kopieren Sie die API-Basis-URL, die oben auf der Seite API-Einstellungen angezeigt wird. Normalerweise ist dies https://<your-tenant>-services.pm.beyondtrustcloud.com. Sie verwenden dies als BPT_API_URL.

Dienstkonto für Cloud Run-Funktion erstellen

Die Cloud Run-Funktion benötigt ein Dienstkonto mit Berechtigungen zum Schreiben in den Cloud Storage-Bucket und zum Aufrufen durch Pub/Sub.

Erstellen Sie das Dienstkonto:

  1. Wechseln Sie in der Google Cloud Console zu IAM & Verwaltung > Dienstkonten.
  2. Klicken Sie auf Dienstkonto erstellen.
  3. Geben Sie die folgenden Konfigurationsdetails an:
    • Name des Dienstkontos: Geben Sie beyondtrust-epm-collector-sa ein.
    • Beschreibung des Dienstkontos: Geben Sie Service account for Cloud Run function to collect BeyondTrust EPM logs ein.
  4. Klicken Sie auf Erstellen und fortfahren.
  5. Fügen Sie im Abschnitt Diesem Dienstkonto Zugriff auf das Projekt erteilen die folgenden Rollen hinzu:
    1. Klicken Sie auf Rolle auswählen.
    2. Suchen Sie nach Storage-Objekt-Administrator und wählen Sie die Rolle aus.
    3. Klicken Sie auf + Weitere Rolle hinzufügen.
    4. Suchen Sie nach Cloud Run Invoker und wählen Sie die Rolle aus.
    5. Klicken Sie auf + Weitere Rolle hinzufügen.
    6. Suchen Sie nach Cloud Functions Invoker und wählen Sie die Rolle aus.
  6. Klicken Sie auf Weiter.
  7. Klicken Sie auf Fertig.

Diese Rollen sind erforderlich für:

  • Storage-Objekt-Administrator: Schreibt Logs in einen Cloud Storage-Bucket und verwaltet Statusdateien.
  • Cloud Run-Aufrufer: Ermöglicht Pub/Sub, die Funktion aufzurufen.
  • Cloud Functions-Invoker: Ermöglicht den Funktionsaufruf

IAM-Berechtigungen für einen Cloud Storage-Bucket gewähren

Gewähren Sie dem Dienstkonto Schreibberechtigungen für den Cloud Storage-Bucket:

  1. Rufen Sie Cloud Storage > Buckets auf.
  2. Klicken Sie auf den Namen Ihres Buckets.
  3. Wechseln Sie zum Tab Berechtigungen.
  4. Klicken Sie auf Zugriff erlauben.
  5. Geben Sie die folgenden Konfigurationsdetails an:
    • Hauptkonten hinzufügen: Geben Sie die E-Mail-Adresse des Dienstkontos ein. Beispiel: beyondtrust-epm-collector-sa@PROJECT_ID.iam.gserviceaccount.com.
    • Rollen zuweisen: Wählen Sie Storage-Objekt-Administrator aus.
  6. Klicken Sie auf Speichern.

Pub/Sub-Thema erstellen

Erstellen Sie ein Pub/Sub-Thema, in dem Cloud Scheduler veröffentlicht und das von der Cloud Run-Funktion abonniert wird.

  1. Wechseln Sie in der Google Cloud Console zu Pub/Sub > Themen.
  2. Klicken Sie auf Thema erstellen.
  3. Geben Sie die folgenden Konfigurationsdetails an:
    • Themen-ID: Geben Sie beyondtrust-epm-trigger ein.
    • Übernehmen Sie die anderen Einstellungen.
  4. Klicken Sie auf Erstellen.

Cloud Run-Funktion zum Erfassen von Logs erstellen

Die Cloud Run-Funktion wird durch Pub/Sub-Nachrichten von Cloud Scheduler ausgelöst, um Logs von der BeyondTrust EPM API abzurufen und in Cloud Storage zu schreiben.

  1. Rufen Sie in der Google Cloud Console Cloud Run auf.
  2. Klicken Sie auf Dienst erstellen.
  3. Wählen Sie Funktion aus. Funktion mit einem Inline-Editor erstellen

  4. Geben Sie im Abschnitt Konfigurieren die folgenden Konfigurationsdetails an:

    Einstellung Wert
    Dienstname beyondtrust-epm-collector
    Region Wählen Sie die Region aus, die Ihrem Cloud Storage-Bucket entspricht (z. B. us-central1).
    Laufzeit Wählen Sie Python 3.12 oder höher aus.

  5. Im Abschnitt Trigger (optional):

    1. Klicken Sie auf + Trigger hinzufügen.
    2. Wählen Sie Cloud Pub/Sub aus.
    3. Wählen Sie unter Cloud Pub/Sub-Thema auswählen das Thema beyondtrust-epm-trigger aus.
    4. Klicken Sie auf Speichern.

  6. Im Abschnitt Authentifizierung:

    1. Wählen Sie Authentifizierung erforderlich aus.
    2. Identitäts- und Zugriffsverwaltung

  7. Rufen Sie Container, Netzwerk, Sicherheit auf und maximieren Sie den Bereich.

  8. Rufen Sie den Tab Sicherheit auf:

    • Dienstkonto: Wählen Sie das Dienstkonto beyondtrust-epm-collector-sa aus.

  9. Rufen Sie den Tab Container auf:

    1. Klicken Sie auf Variablen und Secrets.
    2. Klicken Sie für jede Umgebungsvariable auf + Variable hinzufügen:
    Variablenname Beispielwert
    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: Datensätze pro Anfrage. /Events/FromStartDate kann Werte zwischen 1 und 1.000 haben. Höhere Werte werden von der Funktion begrenzt.
    • MAX_BATCHES: Sicherheitsgrenzwert für Anfragen pro Ausführung. Wenn ein Lauf darauf trifft, wird der Cursor trotzdem auf das Geschriebene gesetzt, sodass der nächste Lauf dort fortgesetzt wird und die Arbeit nicht wiederholt wird.
    • LOOKBACK_HOURS: Wie weit der erste Lauf zurückreicht. BeyondTrust speichert passive Ereignisse (Codes 106, 107, 603, 706) 30 Tage lang und alle anderen Ereignisse 90 Tage lang. Bei einem längeren Lookback-Window werden also keine älteren Daten zurückgegeben.
    • LOOKBACK_HOURS: Wie weit zurückgegangen werden soll, wenn noch kein Cursor vorhanden ist.

  10. Rufen Sie auf dem Tab Variablen und Secrets die Anfragen auf:

    • Zeitlimit für Anfragen: Geben Sie 600 Sekunden (10 Minuten) ein.

  11. Rufen Sie den Tab Einstellungen unter Container auf:

    • Im Abschnitt Ressourcen:
      • Arbeitsspeicher: Wählen Sie 512 MiB oder höher aus.
      • CPU: Wählen Sie 1 aus.
    • Klicken Sie auf Fertig.

  12. Scrollen Sie zu Ausführungsumgebung:

    • Wählen Sie Standard aus (empfohlen).

  13. Im Abschnitt Versionsskalierung:

    • Mindestanzahl von Instanzen: Geben Sie 0 ein.
    • Maximale Anzahl von Instanzen: Geben Sie 100 ein (oder passen Sie den Wert an die erwartete Last an).

  14. Klicken Sie auf Erstellen.

  15. Warten Sie ein bis zwei Minuten, bis der Dienst erstellt wurde.

  16. Nachdem der Dienst erstellt wurde, wird automatisch der Inline-Code-Editor geöffnet.

Funktionscode hinzufügen

  1. Geben Sie main unter Funktionseinstiegspunkt ein.

  2. Erstellen Sie im Inline-Codeeditor zwei Dateien:

    • Erste Datei – 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')}")

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

  3. Klicken Sie auf Bereitstellen, um die Funktion zu speichern und bereitzustellen.

  4. Warten Sie, bis die Bereitstellung abgeschlossen ist (2–3 Minuten).

Cloud Scheduler-Job erstellen

Cloud Scheduler veröffentlicht in regelmäßigen Abständen Nachrichten im Pub/Sub-Thema und löst so die Cloud Run-Funktion aus.

  1. Rufen Sie in der Google Cloud Console Cloud Scheduler auf.
  2. Klicken Sie auf Job erstellen.

  3. Geben Sie die folgenden Konfigurationsdetails an:

    Einstellung Wert
    Name beyondtrust-epm-collector-hourly
    Region Wählen Sie dieselbe Region wie die Cloud Run-Funktion aus.
    Frequenz 0 * * * * (jede Stunde, zur vollen Stunde)
    Zeitzone Zeitzone auswählen (UTC empfohlen)
    Zieltyp Pub/Sub
    Thema Wählen Sie das Thema beyondtrust-epm-trigger aus.
    Inhalt der Nachricht {} (leeres JSON-Objekt)

  4. Klicken Sie auf Erstellen.

Optionen für die Häufigkeit des Zeitplans

Wählen Sie die Häufigkeit basierend auf dem Logvolumen und den Latenzanforderungen aus:

Häufigkeit Cron-Ausdruck Anwendungsfall
Alle 5 Minuten */5 * * * * Hohes Volumen, niedrige Latenz
Alle 15 Minuten */15 * * * * Mittleres Suchvolumen
Stündlich 0 * * * * Standard (empfohlen)
Alle 6 Stunden 0 */6 * * * Geringes Volumen, Batchverarbeitung
Täglich 0 0 * * * Erhebung von Verlaufsdaten

Scheduler-Job testen

  1. Suchen Sie in der Cloud Scheduler-Konsole nach Ihrem Job.
  2. Klicken Sie auf Force run (Ausführung erzwingen), um den Trigger manuell auszulösen.
  3. Warten Sie einige Sekunden und rufen Sie Cloud Run > Dienste > beyondtrust-epm-collector > Logs auf.
  4. Prüfen Sie, ob die Funktion erfolgreich ausgeführt wurde.
  5. Prüfen Sie im Cloud Storage-Bucket, ob Logs geschrieben wurden.

Google SecOps-Dienstkonto abrufen

Google SecOps verwendet ein eindeutiges Dienstkonto, um Daten aus Ihrem Cloud Storage-Bucket zu lesen. Sie müssen diesem Dienstkonto Zugriff auf Ihren Bucket gewähren.

E‑Mail-Adresse des Dienstkontos abrufen

  1. Rufen Sie die SIEM-Einstellungen > Feeds auf.
  2. Klicken Sie auf Neuen Feed hinzufügen.
  3. Klicken Sie auf Einzelnen Feed konfigurieren.
  4. Geben Sie im Feld Feedname einen Namen für den Feed ein, z. B. BeyondTrust EPM logs.
  5. Wählen Sie Google Cloud Storage V2 als Quelltyp aus.
  6. Wählen Sie BeyondTrust Endpoint Privilege Management als Logtyp aus.

  7. Klicken Sie auf Dienstkonto abrufen. Es wird eine eindeutige E-Mail-Adresse für das Dienstkonto angezeigt, z. B.:

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

  8. Kopieren Sie diese E-Mail-Adresse für den nächsten Schritt.

Dem Google SecOps-Dienstkonto IAM-Berechtigungen gewähren

Das Google SecOps-Dienstkonto benötigt zwei Rollen für Ihren Cloud Storage-Bucket: Storage-Objekt-Betrachter zum Lesen der Logobjekte und eine Rolle auf Bucket-Ebene zum Lesen der Bucket-Metadaten.

  1. Rufen Sie Cloud Storage > Buckets auf.
  2. Klicken Sie auf den Namen Ihres Buckets.
  3. Wechseln Sie zum Tab Berechtigungen.
  4. Klicken Sie auf Zugriff erlauben.
  5. Geben Sie die folgenden Konfigurationsdetails an:
    • Hauptkonten hinzufügen: Fügen Sie die E‑Mail-Adresse des Google SecOps-Dienstkontos ein.
    • Rollen zuweisen: Wählen Sie die beiden folgenden Optionen aus:
    • Storage-Objekt-Betrachter: liest die Logobjekte.
    • Leser alter Storage-Buckets: Liest die Bucket-Metadaten. Wenn Sie die Löschoption Übertragene Dateien und leere Verzeichnisse löschen ausgewählt haben, wählen Sie stattdessen Storage Legacy Bucket Writer aus. Damit wird auch die Löschberechtigung erteilt.

  6. Klicken Sie auf Speichern.

Feed in Google SecOps konfigurieren, um BeyondTrust EPM-Logs aufzunehmen

  1. Rufen Sie die SIEM-Einstellungen > Feeds auf.
  2. Klicken Sie auf Neuen Feed hinzufügen.
  3. Klicken Sie auf Einzelnen Feed konfigurieren.
  4. Geben Sie im Feld Feedname einen Namen für den Feed ein, z. B. BeyondTrust EPM logs.
  5. Wählen Sie Google Cloud Storage V2 als Quelltyp aus.
  6. Wählen Sie BeyondTrust Endpoint Privilege Management als Logtyp aus.
  7. Klicken Sie auf Weiter.

  8. Geben Sie Werte für die folgenden Eingabeparameter an:

    • Storage-Bucket-URL: Geben Sie den Cloud Storage-Bucket-URI mit dem Präfixpfad ein:

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

      • Ersetzen Sie:

        • beyondtrust-epm-logs: Name Ihres Cloud Storage-Bucket.
        • beyondtrust-epm/: Optionales Präfix/Ordnerpfad, in dem Logs gespeichert werden (für den Stamm leer lassen).

      • Beispiele:

        • Root-Bucket: gs://beyondtrust-epm-logs/
        • Mit Präfix: gs://beyondtrust-epm-logs/beyondtrust-epm/

    • Option zum Löschen der Quelle: Wählen Sie die gewünschte Löschoption aus:

      • Dateien nie löschen: Dateien werden nie aus der Quelle gelöscht. Dies wird zu Testzwecken empfohlen.

      • Übertragene Dateien und leere Verzeichnisse löschen: Dateien und leere Verzeichnisse werden nach einem erfolgreichen Abruf aus der Quelle gelöscht.

    • Maximales Dateialter: Dateien einschließen, die in den letzten Tagen geändert wurden. Der Standardwert ist 180 Tage.

    • Asset-Namespace: Der Asset-Namespace.

    • Aufnahmelabels: Das Label, das auf die Ereignisse aus diesem Feed angewendet werden soll.

  9. Klicken Sie auf Weiter.

  10. Prüfen Sie die neue Feedkonfiguration auf dem Bildschirm Abschließen und klicken Sie dann auf Senden.

UDM-Zuordnungstabelle

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

Änderungsprotokoll

Änderungsprotokoll für diesen Parser ansehen

Benötigen Sie weitere Hilfe? Antworten von Community-Mitgliedern und Google SecOps-Experten erhalten