CyberArk Endpoint Privilege Manager-Logs erfassen

Parserversion:2.0

Unterstützt in:

In diesem Dokument wird beschrieben, wie Sie CyberArk Endpoint Privilege Manager (EPM)-Logs mit Cloud Storage V2 in Google Security Operations aufnehmen.

CyberArk Endpoint Privilege Manager erzwingt das Prinzip der geringsten Berechtigung auf Windows-, macOS- und Linux-Endpunkten, indem lokale Administratorrechte entfernt, gesteuert wird, welche Anwendungen ausgeführt oder eskaliert werden können, und Anmeldedaten vor Diebstahl geschützt werden. EPM SaaS stellt seine Ereignisdaten über die EPM REST API bereit. Eine Cloud Run-Funktion fragt die Rohevents jedes EPM-Sets ab und schreibt sie in einen Cloud Storage-Bucket, aus dem Google SecOps sie aufnimmt.

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-Diensten, Pub/Sub-Themen und Cloud Scheduler-Jobs
  • Ein CyberArk EPM SaaS-Mandant
  • Ein EPM-Kontoadministrator, der den API-Nutzer und die Rolle erstellen kann

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. cyberark-epm-secops-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 Speicherort aus, z. B. us-central1.
    Speicherklasse Standard (empfohlen für Logs, auf die häufig zugegriffen wird)
    Zugriffskontrolle Einheitlich (empfohlen)
    Schutztools Optional: Objektversionierung oder Aufbewahrungsrichtlinie aktivieren

  6. Klicken Sie auf Erstellen.

CyberArk EPM-Anmeldedaten erfassen

Die Cloud Run-Funktion meldet sich mit einem EPM-Nutzer in der EPM REST API an. Erstellen Sie einen dedizierten Nutzer mit Leseberechtigung für die Sets, deren Ereignisse Sie erfassen möchten.

API-Nutzer erstellen

  1. Melden Sie sich als Kontoadministrator in der EPM-Verwaltungskonsole an.
  2. Klicken Sie auf die Kachel Verwaltung.
  3. Rufen Sie die Seite Nutzer auf und klicken Sie auf Nutzer hinzufügen.
  4. Geben Sie die folgenden Konfigurationsdetails an:
    • E-Mail-Adresse: Geben Sie den Anmeldenamen für den Integrationsnutzer ein (z. B. secops-collector@example.com).
    • Passwort: Geben Sie ein Passwort mit 12 bis 128 Zeichen ein.
  5. Klicken Sie auf Hinzufügen.
  6. Speichern Sie den Anmeldenamen und das Passwort an einem sicheren Ort. Sie verwenden sie als EPM_USERNAME- und EPM_PASSWORD-Werte der Cloud Run-Funktion.

Rolle mit Leseberechtigung erstellen und zuweisen

  1. Rufen Sie in der Verwaltung die Seite Rollen auf und klicken Sie auf Rolle erstellen.
  2. Geben Sie die folgenden Konfigurationsdetails an:
    • Name: Geben Sie einen Namen für die Rolle ein, z. B. SecOps view only.
    • Berechtigungsgruppe: Wählen Sie Lesemodus aus.
    • Sets: Wählen Sie alle Sets aus, deren Ereignisse Sie erfassen möchten.
  3. Klicken Sie auf Erstellen.
  4. Rufen Sie die Seite Rollenzuweisung auf.
  5. Klicken Sie in der Zeile der von Ihnen erstellten Rolle auf das Dreipunkt-Menü (...)> Nutzer zuweisen oder die Zuweisung aufheben.
  6. Klicken Sie auf Nutzer zuweisen, suchen Sie nach dem Integrationsnutzer, wählen Sie ihn aus und klicken Sie auf Zuweisen.

  7. Klicken Sie auf Speichern.

EPM-Anmelde-URL ermitteln

Die EPM-Authentifizierungs-API wird vom Dispatcher-Server Ihrer Region bereitgestellt. Die Funktion verwendet diese URL nur für die Anmeldung. Alle anderen Aufrufe erfolgen an die Manager-URL, die in der Anmeldeantwort zurückgegeben wird.

Region Anmelde-URL
USA https://login.epm.cyberark.com
Deutschland https://eu.epm.cyberark.com
Schweiz https://ch.epm.cyberark.com
Vereinigtes Königreich https://uk.epm.cyberark.com
Australien https://au.epm.cyberark.com
Kanada https://ca.epm.cyberark.com
Indien https://in.epm.cyberark.com
Japan https://jp.epm.cyberark.com
Singapur https://sg.epm.cyberark.com
Italien https://it.epm.cyberark.com
VAE https://ae.epm.cyberark.com
Federal https://login.epm.cyberarkgov.cloud

API-Zugriff testen

  • Testen Sie die Anmeldedaten, bevor Sie mit der Integration fortfahren:

    # Replace with your actual values
EPM_LOGIN_URL="https://login.epm.cyberark.com"
EPM_USERNAME="<your-username>"
EPM_PASSWORD="<your-password>"

# Sign in and get a session token and the manager URL
curl -s -X POST "$EPM_LOGIN_URL/EPM/API/Auth/EPM/Logon" \
    -H "Content-Type: application/json" \
    -d "{\"Username\":\"$EPM_USERNAME\",\"Password\":\"$EPM_PASSWORD\",\"ApplicationID\":\"GoogleSecOps\"}"

  • Eine erfolgreiche Antwort gibt EPMAuthenticationResult (das Sitzungstoken) und ManagerURL zurück. So bestätigen Sie die Rolle:

    # Replace with the values from the sign-in response
MANAGER_URL="<ManagerURL>"
TOKEN="<EPMAuthenticationResult>"

curl -s "$MANAGER_URL/EPM/API/Sets?Limit=1000" \
    -H "Authorization: basic $TOKEN"

In der Antwort wird jede Gruppe aufgeführt, die der Nutzer aufrufen kann. Wenn Sets leer ist, wird die Rolle nicht zugewiesen oder enthält die Sets nicht.

Dienstkonto für die 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.

Dienstkonto erstellen

  1. Rufen Sie in der Google Cloud Console die Seite IAM & Verwaltung > Dienstkonten auf.
  2. Klicken Sie auf Dienstkonto erstellen.
  3. Geben Sie die folgenden Konfigurationsdetails an:
    • Dienstkontoname: Geben Sie cyberark-epm-collector-sa ein.
    • Beschreibung des Dienstkontos: Geben Sie Service account for Cloud Run function to collect CyberArk 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 Object Admin 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 für Folgendes erforderlich:

  • Storage-Objekt-Administrator: Logs in den Cloud Storage-Bucket schreiben und Zustandsdateien verwalten
  • Cloud Run Invoker: Pub/Sub darf die Funktion aufrufen.
  • Cloud Functions-Invoker: Funktionsaufruf zulassen

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

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

  1. Rufen Sie in der Google Cloud Console 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 (z. B. cyberark-epm-collector-sa@PROJECT_ID.iam.gserviceaccount.com).
    • Rollen zuweisen: Wählen Sie Storage Object Admin 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. Rufen Sie in der Google Cloud Console die Option Pub/Sub > Themen auf.
  2. Klicken Sie auf Thema erstellen.
  3. Geben Sie die folgenden Konfigurationsdetails an:
    • Themen-ID: Geben Sie cyberark-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 Rohereignisse aus der CyberArk EPM REST API abzurufen und in Cloud Storage zu schreiben.

  1. Rufen Sie in der Google Cloud Console Cloud Run auf.
  2. Klicken Sie auf Funktion schreiben.

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

    Einstellung Wert
    Dienstname cyberark-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.

  4. Klicken Sie im Abschnitt Trigger auf Trigger hinzufügen und wählen Sie Pub/Sub-Trigger aus.

  5. Geben Sie im Bereich Eventarc-Trigger die folgenden Konfigurationsdetails an:

    • Triggername: Behalten Sie den generierten Namen bei oder geben Sie einen Namen für den Trigger ein.
    • Triggertyp: Wählen Sie Google-Quellen aus.
    • Ereignisanbieter: Wählen Sie Pub/Sub aus.
    • Ereignistyp: Wählen Sie google.cloud.pubsub.topic.v1.messagePublished aus.
    • Cloud Pub/Sub-Thema auswählen: Wählen Sie das Thema cyberark-epm-trigger aus.
    • Region: Wählen Sie dieselbe Region wie für die Funktion aus.
    • Dienstkonto: Wählen Sie das Dienstkonto cyberark-epm-collector-sa aus.

  6. Klicken Sie auf Trigger speichern.

  7. Im Abschnitt Authentifizierung:

    1. Wählen Sie Authentifizierung erforderlich aus.
    2. Wählen Sie Identity and Access Management (IAM) aus.

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

  9. Rufen Sie den Tab Sicherheit auf:

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

  10. 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 Beschreibung
    GCS_BUCKET cyberark-epm-secops-logs Name des Cloud Storage-Buckets
    GCS_PREFIX cyberark-epm Präfix für Logdateien
    STATE_KEY cyberark-epm-state.json Zustandspfad, außerhalb des Logpräfixes
    EPM_LOGIN_URL https://login.epm.cyberark.com EPM-Anmelde-URL für Ihre Region
    EPM_USERNAME secops-collector@example.com Anmeldename des EPM API-Nutzers
    EPM_PASSWORD <your-password> Passwort des EPM API-Nutzers
    EPM_APPLICATION_ID GoogleSecOps Bei der Anmeldung gemeldeter Anwendungsname
    EPM_SET_IDS (leer) Optionale, kommagetrennte Set-IDs; wenn leer, werden alle Sets erfasst, die der Nutzer aufrufen kann.
    EPM_API_VERSION (leer) Optionale API-Version (z. B. 24.12.0.1); bei leerem Wert wird die aktuelle Version verwendet.
    MAX_RECORDS 10000 Maximale Anzahl von Datensätzen pro Ausführung
    PAGE_SIZE 1000 Datensätze pro Seite (API-Maximum: 1.000)
    LOOKBACK_HOURS 24 Anfänglicher Rückschauzeitraum bei der ersten Ausführung
    OVERLAP_MINUTES 15 Lesen Sie das Zeitfenster vor dem Wasserzeichen noch einmal, um Ereignisse zu erfassen, die von Kundenservicemitarbeitern verspätet gemeldet werden.

  11. Rufen Sie im Bereich Variablen und Secrets die Anfragen auf:

    • Zeitüberschreitung bei Anfrage: Geben Sie 600 Sekunden (10 Minuten) ein.

  12. Gehen Sie zum Tab Einstellungen:

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

  13. Im Abschnitt Versionsskalierung:

    • Mindestanzahl von Instanzen: Geben Sie 0 ein.
    • Maximale Anzahl von Instanzen: Geben Sie 1 ein.

  14. Klicken Sie auf Erstellen.

  15. Warten Sie, bis der Dienst erstellt wurde (1–2 Minuten).

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

Funktionscode hinzufügen

  1. Geben Sie main in das Feld Einstiegspunkt ein.

  2. Erstellen Sie im Inline-Code-Editor zwei Dateien:

    • Erste Datei – main.py::

      import functions_framework
from google.cloud import storage
from google.cloud.exceptions import NotFound
import hashlib
import json
import os
import urllib3
from urllib.parse import urlencode
from datetime import datetime, timezone, timedelta
import time

# 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()

# Environment variables
GCS_BUCKET = os.environ.get('GCS_BUCKET')
GCS_PREFIX = os.environ.get('GCS_PREFIX', 'cyberark-epm')
# STATE_KEY must stay OUTSIDE GCS_PREFIX. The feed ingests every object under
# its bucket URI and, with a deletion option selected, deletes what it
# transferred. A state file inside the prefix would be ingested as log data and
# then deleted, resetting collection and re-ingesting duplicates.
STATE_KEY = os.environ.get('STATE_KEY', 'cyberark-epm-state.json')
EPM_LOGIN_URL = os.environ.get('EPM_LOGIN_URL', 'https://login.epm.cyberark.com')
EPM_USERNAME = os.environ.get('EPM_USERNAME')
EPM_PASSWORD = os.environ.get('EPM_PASSWORD')
EPM_APPLICATION_ID = os.environ.get('EPM_APPLICATION_ID', 'GoogleSecOps')
# Comma-separated set IDs to collect. Empty collects every set the user can view.
EPM_SET_IDS = [s.strip() for s in os.environ.get('EPM_SET_IDS', '').split(',') if s.strip()]
# API version segment. Empty uses the latest version, which CyberArk recommends.
EPM_API_VERSION = os.environ.get('EPM_API_VERSION', '').strip()
MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '10000'))
# The Events/Search API returns at most 1,000 records per page.
PAGE_SIZE = min(int(os.environ.get('PAGE_SIZE', '1000')), 1000)
LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))
# The query re-reads this many minutes before the watermark. EPM agents report
# in batches, so an event can reach the service well after its event date.
# Re-read events are removed by content hash.
OVERLAP_MINUTES = int(os.environ.get('OVERLAP_MINUTES', '15'))
# The events APIs allow 5 calls per second; keep a small gap between calls.
CALL_INTERVAL_SECONDS = 0.25

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

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

def parse_datetime(value: str) -> datetime:
        """Parse an ISO 8601 datetime string into an aware datetime."""
        if value.endswith('Z'):
                value = value[:-1] + '+00:00'
        dt = datetime.fromisoformat(value)
        if dt.tzinfo is None:
                dt = dt.replace(tzinfo=timezone.utc)
        return dt

def epm_time(dt: datetime) -> str:
        """Render a datetime in the YYYY-MM-DDThh:mm:ssZ form the EPM filter accepts."""
        return dt.astimezone(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')

def api_path(*parts: str) -> str:
        """Build an /EPM/API path, inserting the version segment when one is set."""
        segments = ['EPM', 'API']
        if EPM_API_VERSION:
                segments.append(EPM_API_VERSION)
        segments.extend(parts)
        return '/' + '/'.join(segments)

def event_keys(event: dict) -> set:
        """Return the identity this event can be recognised by.

        Raw events carry no unique event ID, so a record counts as already
        collected when its full JSON content was seen before. The overlap window
        returns the same JSON again, which this catches without any knowledge of
        the payload schema.
        """
        return {'sha256:' + hashlib.sha256(
                json.dumps(event, sort_keys=True, ensure_ascii=False).encode('utf-8')
        ).hexdigest()}

def event_time(event: dict) -> str:
        """Return the event timestamp as an ISO 8601 string, or an empty string.

        Raw events carry the time of the most recent occurrence in `lastEventDate`.
        """
        value = event.get('lastEventDate')
        return str(value) if value else ''

@functions_framework.cloud_event
def main(cloud_event):
        """Fetch CyberArk EPM raw events and write them to Cloud Storage.

        Args:
                cloud_event: CloudEvent object containing the Pub/Sub message.
        """
        if not all([GCS_BUCKET, EPM_LOGIN_URL, EPM_USERNAME, EPM_PASSWORD]):
                # Raise rather than return: a bare return acks the Pub/Sub message and
                # reports the run as successful, silently discarding the schedule tick.
                raise RuntimeError('Missing required environment variables')

        bucket = storage_client.bucket(GCS_BUCKET)
        state = load_state(bucket, STATE_KEY)
        set_states = state.get('sets', {})

        now = datetime.now(timezone.utc)

        manager_url, token = login()
        headers = {
                'Authorization': f'basic {token}',
                'Content-Type': 'application/json',
                'Accept': 'application/json',
        }

        sets = list_sets(manager_url, headers)
        if EPM_SET_IDS:
                sets = [s for s in sets if s['Id'] in EPM_SET_IDS]
                missing = set(EPM_SET_IDS) - {s['Id'] for s in sets}
                if missing:
                        raise FetchError(f'Configured set IDs not visible to this user: {sorted(missing)}')
        if not sets:
                raise FetchError('No EPM sets are visible to this user; check the assigned role')

        fresh_all = []
        new_set_states = {}
        budget = MAX_RECORDS

        for epm_set in sets:
                set_id = epm_set['Id']
                set_state = set_states.get(set_id, {})
                watermark = None
                if set_state.get('last_event_time'):
                        watermark = parse_datetime(set_state['last_event_time'])
                seen_keys = set(set_state.get('seen_keys', []))

                if watermark is None:
                        start_time = now - timedelta(hours=LOOKBACK_HOURS)
                else:
                        start_time = watermark - timedelta(minutes=OVERLAP_MINUTES)

                print(f"Set {set_id} ({epm_set.get('Name', '')}): fetching events from {start_time.isoformat()} to {now.isoformat()}")

                # A FetchError here propagates: the run fails, the watermark is untouched,
                # and the next invocation retries the same window.
                records, newest_event_time = fetch_events(
                        manager_url=manager_url,
                        headers=headers,
                        set_id=set_id,
                        start_time=start_time,
                        end_time=now,
                        page_size=PAGE_SIZE,
                        max_records=budget,
                )
                budget -= len(records)

                # Remove events already written by an earlier run. Without this, the overlap
                # window re-emits its events on every invocation.
                fresh = [r for r in records if not (event_keys(r) & seen_keys)]
                print(f"Set {set_id}: fetched {len(records)} records, {len(fresh)} new after deduplication")

                if not fresh:
                        new_set_states[set_id] = set_state
                        continue

                if not newest_event_time:
                        # A watermark that cannot be derived from the data is not a watermark.
                        # Advancing on wall-clock here would skip every late-reported event.
                        raise FetchError(f'Set {set_id}: records were returned but no event timestamp could be parsed')

                # The parser reads the set from the event itself. Annotate copies so the
                # deduplication hashes keep matching what the API returns next time.
                fresh_all.extend({**record, 'setID': set_id} for record in fresh)

                new_watermark = parse_datetime(newest_event_time)
                if watermark and new_watermark < watermark:
                        new_watermark = watermark

                # Retain only the hashes still inside the overlap window, so the state file
                # stays small while covering every event the next query can re-read.
                cutoff = new_watermark - timedelta(minutes=OVERLAP_MINUTES)
                retained = set()
                for record in records:
                        stamp = event_time(record)
                        if not stamp:
                                continue
                        try:
                                if parse_datetime(stamp) >= cutoff:
                                        retained.update(event_keys(record))
                        except ValueError:
                                continue

                new_set_states[set_id] = {
                        'last_event_time': new_watermark.isoformat(),
                        'seen_keys': sorted(retained),
                }

                if budget <= 0:
                        print(f"Reached max_records limit ({MAX_RECORDS}); remaining sets are collected by the next run")
                        break

        # Keep the state of sets not visited in this run.
        for set_id, set_state in set_states.items():
                new_set_states.setdefault(set_id, set_state)

        if not fresh_all:
                print("No new events found. Watermarks left unchanged.")
                return

        timestamp = now.strftime('%Y%m%dT%H%M%SZ')
        object_key = f"{GCS_PREFIX}/events_{timestamp}.ndjson"
        blob = bucket.blob(object_key)

        ndjson = '\n'.join(json.dumps(record, ensure_ascii=False) for record in fresh_all) + '\n'
        blob.upload_from_string(ndjson, content_type='application/x-ndjson')

        print(f"Wrote {len(fresh_all)} records to gs://{GCS_BUCKET}/{object_key}")

        # Advance the watermarks only after the data is durably written.
        save_state(bucket, STATE_KEY, {'sets': new_set_states})

        print(f"Successfully processed {len(fresh_all)} records")

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

        Only a missing object is treated as a cold start. Any other error is raised:
        swallowing it would silently reset collection to the full lookback window
        and re-ingest that entire 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: dict):
        """Write the collector state to Cloud Storage.

        Failures are raised, not logged. If the state write fails after the data was
        uploaded, the next run repeats the same window and duplicates it.
        """
        blob = bucket.blob(key)
        blob.upload_from_string(
                json.dumps(state, indent=2),
                content_type='application/json',
        )
        print(f"Saved state for {len(state.get('sets', {}))} sets")

def request_json(method: str, url: str, headers: dict, body: dict = None):
        """Call the EPM API and return the decoded JSON body.

        Raises FetchError on any transport or HTTP failure, so that the caller does
        not mistake a failed call for an empty result and advance the watermark.
        """
        backoff = 1.0
        for attempt in range(5):
                time.sleep(CALL_INTERVAL_SECONDS)
                try:
                        response = http.request(
                                method,
                                url,
                                body=json.dumps(body).encode('utf-8') if body is not None else None,
                                headers=headers,
                        )
                except Exception as e:
                        raise FetchError(f'Request to {url} failed: {e}') from e

                if response.status == 429:
                        raw_retry_after = response.headers.get('Retry-After')
                        try:
                                # Retry-After may also be an HTTP date, which int() cannot parse.
                                wait = int(raw_retry_after) if raw_retry_after else int(backoff)
                        except (TypeError, ValueError):
                                wait = int(backoff)
                        print(f"Rate limited (429). Retrying after {wait}s...")
                        time.sleep(wait)
                        backoff = min(backoff * 2, 30.0)
                        continue

                if response.status != 200:
                        raise FetchError(f'HTTP {response.status} from {url}: {response.data.decode("utf-8", "replace")}')

                try:
                        return json.loads(response.data.decode('utf-8'))
                except json.JSONDecodeError as e:
                        raise FetchError(f'Malformed JSON response from {url}: {e}') from e

        raise FetchError('Rate limited repeatedly; giving up without advancing the watermark')

def login():
        """Authenticate to EPM and return the manager URL and session token.

        EPM allows one API login per minute per user, so the function signs in once
        per run and reuses the token for every call in that run.
        """
        url = f"{EPM_LOGIN_URL.rstrip('/')}{api_path('Auth', 'EPM', 'Logon')}"
        body = {
                'Username': EPM_USERNAME,
                'Password': EPM_PASSWORD,
                'ApplicationID': EPM_APPLICATION_ID,
        }
        data = request_json('POST', url, {'Content-Type': 'application/json'}, body)
        token = data.get('EPMAuthenticationResult')
        manager_url = data.get('ManagerURL')
        if not token or not manager_url:
                raise FetchError(f'Login response did not contain a token and manager URL: {data}')
        if data.get('IsPasswordExpired'):
                raise FetchError('The EPM user password has expired; set a new password and update EPM_PASSWORD')
        return manager_url.rstrip('/'), token

def list_sets(manager_url: str, headers: dict):
        """Return the sets visible to the API user."""
        url = f"{manager_url}{api_path('Sets')}?{urlencode({'Offset': 0, 'Limit': 1000})}"
        data = request_json('GET', url, headers)
        sets = data.get('Sets') or []
        print(f"Found {len(sets)} sets")
        return sets

def fetch_events(manager_url: str, headers: dict, set_id: str, start_time: datetime, end_time: datetime, page_size: int, max_records: int):
        """Fetch raw events for one set from the EPM Events/Search API.

        Pages with nextCursor, which returns events in chronological order. A time
        watermark is only correct over an ascending scan: with newest-first ordering
        a truncated run would advance the watermark past unread events.

        Returns:
                Tuple of (records list, newest_event_time ISO string).
        """
        endpoint = f"{manager_url}{api_path('Sets', set_id, 'Events', 'Search')}"
        body = {
                'filter': f'eventDate GE {epm_time(start_time)} AND eventDate LE {epm_time(end_time)}',
        }

        records = []
        newest_time = None
        cursor = 'start'
        page_num = 0

        while True:
                page_num += 1

                if len(records) >= max_records:
                        # Stop cleanly. The remaining events stay ahead of the watermark and
                        # are collected by the next run.
                        print(f"Set {set_id}: reached max_records limit ({max_records})")
                        break

                # Rebuild the full query on every page; the time window must never be dropped.
                url = f"{endpoint}?{urlencode({'nextCursor': cursor, 'limit': page_size})}"
                data = request_json('POST', url, headers, body)

                page_results = data.get('events') or []
                if not page_results:
                        print(f"Set {set_id}: no more results (empty page)")
                        break

                print(f"Set {set_id}: page {page_num} retrieved {len(page_results)} events")
                records.extend(page_results)

                for event in page_results:
                        stamp = event_time(event)
                        if not stamp:
                                continue
                        try:
                                if newest_time is None or parse_datetime(stamp) > parse_datetime(newest_time):
                                        newest_time = stamp
                        except ValueError as e:
                                print(f"Warning: Could not parse event time {stamp!r}: {e}")

                cursor = data.get('nextCursor')
                if not cursor:
                        break

        print(f"Set {set_id}: retrieved {len(records)} total events from {page_num} pages")
        return records, newest_time

    • 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).

So werden Ereignisse von der Funktion erfasst

  • Die Funktion meldet sich einmal pro Ausführung an, listet die Sets auf, die der Nutzer aufrufen kann, und ruft die API Get detailed raw events (Detaillierte Rohereignisse abrufen) jedes Sets mit einem Filter für eventDate und Paging mit nextCursor in chronologischer Reihenfolge auf.
  • Jedes Ereignis wird mit dem zusätzlichen Feld setID geschrieben, damit der Satz im Log erhalten bleibt.
  • In der Statusdatei werden pro Set die neuesten lastEventDate und die Hashes der Ereignisse im Überschneidungszeitraum gespeichert. Bei jedem Lauf werden die Daten OVERLAP_MINUTES vor diesem Wasserzeichen noch einmal gelesen und die bereits geschriebenen Ereignisse werden gelöscht. So werden auch spät gemeldete Ereignisse ohne Duplikate erfasst.
  • Bei API-, Anmelde- oder Speicherfehlern schlägt der Lauf fehl und das Wasserzeichen bleibt unverändert. Beim nächsten Lauf wird dasselbe Zeitfenster noch einmal versucht.

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 cyberark-epm-collector-hourly
    Region Dieselbe Region wie die Cloud Run-Funktion auswählen
    Frequenz 0 * * * * (stündlich zur vollen Stunde)
    Zeitzone Zeitzone auswählen (UTC empfohlen)
    Zieltyp Pub/Sub
    Thema Wählen Sie das Pub/Sub-Thema cyberark-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 15 Minuten */15 * * * * Niedrige Latenz
Stündlich 0 * * * * Standard (empfohlen)
Alle 6 Stunden 0 */6 * * * Geringes Volumen, Batchverarbeitung

Integration testen

  1. Suchen Sie in der Cloud Scheduler-Konsole nach Ihrem Job.
  2. Klicken Sie auf Ausführung erzwingen, um den Job manuell auszulösen.
  3. Warten Sie einige Sekunden.
  4. Rufen Sie Cloud Run > Dienste auf.
  5. Klicken Sie auf den Namen Ihrer Funktion (cyberark-epm-collector).
  6. Klicken Sie auf den Tab Logs.

  7. Prüfen Sie, ob die Funktion erfolgreich ausgeführt wurde. Darauf sollten Sie achten:

    Found N sets
Set <set-id> (<set-name>): fetching events from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00
Set <set-id>: page 1 retrieved X events
Wrote X records to gs://cyberark-epm-secops-logs/cyberark-epm/events_YYYYMMDDTHHMMSSZ.ndjson
Successfully processed X records

  8. Rufen Sie Cloud Storage > Buckets auf.

  9. Klicken Sie auf den Namen Ihres Buckets.

  10. Rufen Sie den Präfixordner (cyberark-epm/) auf.

  11. Prüfen Sie, ob eine neue .ndjson-Datei mit dem aktuellen Zeitstempel erstellt wurde.

Wenn in den Logs Fehler angezeigt werden:

  • HTTP 401: Prüfen Sie EPM_USERNAME und EPM_PASSWORD oder ob die Anmeldung innerhalb einer Minute wiederholt wurde.
  • HTTP 403: Dem Nutzer fehlt eine Rolle für die Gruppe oder das API-Nutzungslimit des Mandanten wurde erreicht.
  • HTTP 404: Falsche Set-ID in EPM_SET_IDS
  • HTTP 429: Ratenbegrenzung – Funktionswiederholungen mit Backoff
  • Fehlende Umgebungsvariablen: Prüfen Sie, ob alle erforderlichen Variablen festgelegt sind.

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. CyberArk EPM logs.
  5. Wählen Sie Google Cloud Storage V2 als Quelltyp aus.
  6. Wählen Sie CyberArk Endpoint Privilege Manager (EPM) als Log-Typ aus.
  7. Klicken Sie auf Dienstkonto abrufen.

  8. Es wird eine eindeutige E-Mail-Adresse für das Dienstkonto angezeigt, z. B.:

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

  9. Kopieren Sie diese E-Mail-Adresse, um sie im nächsten Schritt zu verwenden.

  10. Klicken Sie auf Weiter.

  11. 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://cyberark-epm-secops-logs/cyberark-epm/
      • Ersetzen Sie cyberark-epm-secops-logs durch den Namen des Cloud Storage-Buckets.

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

      • Dateien nie löschen: Dateien werden nie aus der Quelle gelöscht (empfohlen für Tests).

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

    • Höchstalter der Datei: Dateien einbeziehen, die in den letzten Tagen geändert wurden (Standardwert: 180 Tage)

    • Asset-Namespace: Der Asset-Namespace

    • Labels für Datenaufnahme: Das Label, das auf die Ereignisse aus diesem Feed angewendet werden soll

  12. Klicken Sie auf Weiter.

  13. Prüfen Sie die neue Nachrichten-Feed-Konfiguration auf dem Bildschirm Abschließen und klicken Sie dann auf Senden.

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 folgenden beiden Optionen aus:
      • Storage Object Viewer: Liest die Logobjekte.
      • Leser alter Storage-Buckets: Berechtigt zum Lesen der 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. Dadurch wird auch die Löschberechtigung gewährt.
  6. Klicken Sie auf Speichern.

UDM-Zuordnungstabelle

Logfeld UDM-Zuordnung Logik
accessTargetType_label additional.fields Zusammengeführt
accessaction_label additional.fields Zusammengeführt
agentEventCount_label additional.fields Zusammengeführt
arrivalTime_label additional.fields Zusammengeführt
deceptionType_label additional.fields Zusammengeführt
eventCount_label additional.fields Zusammengeführt
logonAttemptTypeId_label additional.fields Zusammengeführt
logonStatusId_label additional.fields Zusammengeführt
modificationTime_label additional.fields Zusammengeführt
originalFileName_label additional.fields Zusammengeführt
owner_label additional.fields Zusammengeführt
packageName_label additional.fields Zusammengeführt
publisher_field additional.fields Zusammengeführt
publisher_label additional.fields Zusammengeführt
setID_label additional.fields Zusammengeführt
skippedCount_label additional.fields Zusammengeführt
sourceProcessSigner_field additional.fields Zusammengeführt
sourceProcessSigner_label additional.fields Zusammengeführt
sourceType_label additional.fields Zusammengeführt
threatProtectionActionId_label additional.fields Zusammengeführt
threatProtectionAction_label additional.fields Zusammengeführt
user_is_admin_label additional.fields Zusammengeführt
winEventRecordId_label additional.fields Zusammengeführt
winEventType_label additional.fields Zusammengeführt
workingDirectory_label additional.fields Zusammengeführt
displayName metadata.description Direkt zugeordnet
eachevent.displayName metadata.description Direkt zugeordnet
lastEventDate metadata.event_timestamp Geparst als ISO8601
has_principal metadata.event_type Zugeordnet: true → STATUS_UPDATE
has_principal_user metadata.event_type Zugeordnet: true → USER_UNCATEGORIZED
eachevent.eventType metadata.product_event_type Direkt zugeordnet
eventType metadata.product_event_type Direkt zugeordnet
productName metadata.product_version Direkt zugeordnet
productVersion metadata.product_version Direkt zugeordnet
url network.http.referral_url Direkt zugeordnet
agentId principal.asset.asset_id Direkt zugeordnet
eachevent.agentId principal.asset.asset_id Direkt zugeordnet
computerName principal.asset.hostname Direkt zugeordnet
sourceName principal.file.full_path Direkt zugeordnet
computerName principal.hostname Direkt zugeordnet
eachevent.computerName principal.hostname Direkt zugeordnet
operatingSystemType principal.platform Direkt zugeordnet
operating_system_type principal.platform Zugeordnet: Windows → WINDOWS
company principal.user.company_name Direkt zugeordnet
eachevent.userName principal.user.userid Direkt zugeordnet
userName principal.user.userid Direkt zugeordnet
sec_result security_result Zusammengeführt
threat_protection_action security_result.action_details Direkt zugeordnet
fileDescription security_result.description Direkt zugeordnet
eachevent.policyName security_result.rule_name Direkt zugeordnet
policyName security_result.rule_name Direkt zugeordnet
justification security_result.summary Direkt zugeordnet
fileQualifier target.asset.asset_id Direkt zugeordnet
software target.asset.software Zusammengeführt
eachevent.filePath target.file.full_path Direkt zugeordnet
filePath target.file.full_path Direkt zugeordnet
eachevent.hash target.file.sha1 Direkt zugeordnet
hash target.file.sha1 Direkt zugeordnet
fileSize target.file.size Direkt zugeordnet
eachevent.processCommandLine target.process.command_line Umbenannt/zugeordnet
processCommandLine target.process.command_line Direkt zugeordnet
fileName target.process.file.names Zusammengeführt
eachevent.sourceProcessCommandLine target.process.parent_process.command_line Umbenannt/zugeordnet
sourceProcessCommandLine target.process.parent_process.command_line Direkt zugeordnet
each_event.sourceProcessHash target.process.parent_process.file.sha1 Umbenannt/zugeordnet
sourceProcessHash target.process.parent_process.file.sha1 Direkt zugeordnet
sourceProcessPublisher_label target.resource.attribute.labels Zusammengeführt
exposed_accountName target.user.attribute.labels Zusammengeführt
exposed_domain target.user.attribute.labels Zusammengeführt
exposed_source target.user.attribute.labels Zusammengeführt
exposed_username target.user.attribute.labels Zusammengeführt
sourceProcessUsername target.user.userid Direkt zugeordnet
– metadata.event_type Konstante: STATUS_UPDATE
– metadata.product_name Konstante: EPM
– metadata.vendor_name Konstante: CYBERARK
– principal.platform Konstante: WINDOWS
– security_result.alert_state Konstante: ALERTING

Änderungsprotokoll

Änderungsprotokoll für diesen Parser ansehen

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