Raccogli i log di CyberArk Endpoint Privilege Manager

Versione parser: 2.0

Supportato in:

Questo documento spiega come importare i log di CyberArk Endpoint Privilege Manager (EPM) in Google Security Operations utilizzando Cloud Storage V2.

CyberArk Endpoint Privilege Manager applica il principio del privilegio minimo sugli endpoint Windows, macOS e Linux rimuovendo i diritti di amministratore locale, controllando quali applicazioni possono essere eseguite o elevate e proteggendo le credenziali dal furto. EPM SaaS espone i dati sugli eventi tramite l'API REST EPM. Una funzione Cloud Run esegue il polling degli eventi non elaborati di ogni set di EPM e li scrive in un bucket Cloud Storage, da cui Google SecOps li importa.

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 i criteri IAM (Identity and Access Management) nei bucket Cloud Storage
  • Autorizzazioni per creare servizi Cloud Run, argomenti Pub/Sub e job Cloud Scheduler
  • Un tenant SaaS CyberArk EPM
  • Un amministratore dell'account EPM per creare l'utente e il ruolo API

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 univoco globale (ad esempio cyberark-epm-secops-logs).
    Tipo di località Scegli in base alle tue esigenze (regione singola, a due regioni, multiregionale)
    Località Seleziona la località (ad esempio us-central1)
    Classe di archiviazione Standard (consigliato per i log ad accesso frequente)
    Controllo dell'accesso 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 di CyberArk EPM

La funzione Cloud Run accede all'API REST EPM con un utente EPM. Crea un utente dedicato con accesso di sola visualizzazione ai set di cui vuoi raccogliere gli eventi.

Crea l'utente API

  1. Accedi alla console di gestione EPM come amministratore dell'account.
  2. Fai clic sul riquadro Amministrazione.
  3. Vai alla pagina Utenti e fai clic su Aggiungi utente.
  4. Fornisci i seguenti dettagli di configurazione:
    • Indirizzo email: inserisci il nome di accesso dell'utente dell'integrazione (ad esempio, secops-collector@example.com).
    • Password: inserisci una password composta da 12-128 caratteri.
  5. Fai clic su Aggiungi.
  6. Salva il nome e la password per l'accesso in modo sicuro. Li utilizzerai come valori EPM_USERNAME e EPM_PASSWORD della funzione Cloud Run.

Crea un ruolo di sola visualizzazione e assegnalo

  1. In Amministrazione, vai alla pagina Ruoli e fai clic su Crea ruolo.
  2. Fornisci i seguenti dettagli di configurazione:
    • Nome: inserisci un nome per il ruolo (ad esempio, SecOps view only).
    • Gruppo di autorizzazioni: seleziona Sola visualizzazione.
    • Set: seleziona ogni set di cui vuoi raccogliere gli eventi.
  3. Fai clic su Crea.
  4. Vai alla pagina Assegnazione dei ruoli.
  5. Nella riga del ruolo che hai creato, fai clic su Altre azioni (…) > Assegna o annulla l'assegnazione di utenti.
  6. Fai clic su Assegna utenti, cerca l'utente di integrazione, selezionalo e fai clic su Assegna.

  7. Fai clic su Salva.

Determina l'URL di accesso a EPM

L'API di autenticazione EPM viene gestita dal server dispatcher della tua regione. La funzione utilizza questo URL solo per accedere; ogni altra chiamata va all'URL del gestore restituito dalla risposta di accesso.

Regione URL di accesso
Stati Uniti https://login.epm.cyberark.com
Germania https://eu.epm.cyberark.com
Svizzera https://ch.epm.cyberark.com
Regno Unito https://uk.epm.cyberark.com
Australia https://au.epm.cyberark.com
Canada https://ca.epm.cyberark.com
India https://in.epm.cyberark.com
Giappone https://jp.epm.cyberark.com
Singapore https://sg.epm.cyberark.com
Italia https://it.epm.cyberark.com
EAU https://ae.epm.cyberark.com
Federale https://login.epm.cyberarkgov.cloud

Testare l'accesso API

  • Testa le credenziali prima di procedere con l'integrazione nel seguente modo:

    # 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\"}"

  • Una risposta positiva restituisce EPMAuthenticationResult (il token di sessione) e ManagerURL. Per confermare il ruolo, elenca i set con il token nel seguente modo:

    # 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"

La risposta elenca ogni insieme che l'utente può visualizzare. Se Sets è vuoto, il ruolo non è assegnato o non include i set.

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 un account di servizio

  1. Nella Google Cloud console, vai a IAM e amministrazione > Service Account.
  2. Fai clic su Crea account di servizio.
  3. Fornisci i seguenti dettagli di configurazione:
    • Nome del service account: inserisci cyberark-epm-collector-sa
    • Descrizione service account: inserisci Service account for Cloud Run function to collect CyberArk 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 necessari per:

  • Storage Object Admin: scrive i log nel 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 della funzione

Concedere le autorizzazioni IAM sul bucket Cloud Storage

Concedi all'account di servizio le autorizzazioni di scrittura sul bucket Cloud Storage:

  1. Nella Google Cloud console, 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 cyberark-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 a cui Cloud Scheduler pubblicherà e a cui la funzione Cloud Run si iscriverà.

  1. Nella Google Cloud console, vai a Pub/Sub > Argomenti.
  2. Fai clic su Crea argomento.
  3. Fornisci i seguenti dettagli di configurazione:
    • ID argomento: inserisci cyberark-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 verrà attivata dai messaggi Pub/Sub di Cloud Scheduler per recuperare gli eventi non elaborati dall'API REST CyberArk EPM e scriverli in Cloud Storage.

  1. Nella Google Cloud console, vai a Cloud Run.
  2. Fai clic su Scrivi una funzione.

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

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

  4. Nella sezione Trigger, fai clic su Aggiungi trigger e seleziona Trigger Pub/Sub.

  5. Nel riquadro Trigger Eventarc, fornisci i seguenti dettagli di configurazione:

    • Nome attivatore: mantieni il nome generato o inserisci un nome per l'attivatore.
    • Tipo di trigger: seleziona Origini Google.
    • Provider di eventi: seleziona Pub/Sub.
    • Tipo di evento: seleziona google.cloud.pubsub.topic.v1.messagePublished.
    • Seleziona un argomento Cloud Pub/Sub: scegli l'argomento cyberark-epm-trigger.
    • Regione: seleziona la stessa regione della funzione.
    • Service account: seleziona il account di servizio cyberark-epm-collector-sa.

  6. Fai clic su Salva trigger.

  7. Nella sezione Autenticazione:

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

  8. Vai a Container, networking, sicurezza ed espandilo.

  9. Vai alla scheda Sicurezza:

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

  10. 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 Descrizione
    GCS_BUCKET cyberark-epm-secops-logs Nome del bucket Cloud Storage
    GCS_PREFIX cyberark-epm Prefisso per i file di log
    STATE_KEY cyberark-epm-state.json Percorso dello stato, al di fuori del prefisso del log
    EPM_LOGIN_URL https://login.epm.cyberark.com URL di accesso a EPM per la tua regione
    EPM_USERNAME secops-collector@example.com Nome di accesso dell'utente dell'API EPM
    EPM_PASSWORD <your-password> Password dell'utente API EPM
    EPM_APPLICATION_ID GoogleSecOps Nome dell'applicazione segnalato al momento dell'accesso
    EPM_SET_IDS (vuoto) (Facoltativo) ID set separati da virgole; se vuoto, vengono raccolti tutti i set che l'utente può visualizzare
    EPM_API_VERSION (vuoto) Versione dell'API facoltativa (ad esempio, 24.12.0.1); se vuoto, viene utilizzata l'ultima versione
    MAX_RECORDS 10000 Numero massimo di record per esecuzione
    PAGE_SIZE 1000 Record per pagina (il massimo per l'API è 1000)
    LOOKBACK_HOURS 24 Periodo di riferimento iniziale alla prima esecuzione
    OVERLAP_MINUTES 15 Rileggi la finestra prima della filigrana per rilevare gli eventi segnalati in ritardo dagli agenti

  11. Nella sezione Variabili e secret, vai a Richieste:

    • Timeout richiesta: inserisci 600 secondi (10 minuti)

  12. Vai alla scheda Impostazioni:

    • Nella sezione Risorse:
      • Memoria: seleziona 512 MiB o un valore superiore
      • CPU: seleziona 1

  13. Nella sezione Scalabilità della revisione:

    • Numero minimo di istanze: inserisci 0
    • Numero massimo di istanze: inserisci 1

  14. Fai clic su Crea.

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

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

Aggiungi il codice per la funzione

  1. Inserisci main nel campo Entry point (Punto di ingresso).

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

    • Primo file - 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

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

Come la funzione raccoglie gli eventi

  • La funzione esegue l'accesso una volta per esecuzione, elenca i set che l'utente può visualizzare e chiama l'API Get detailed raw events di ogni set con un filtro su eventDate, paginazione con nextCursor in ordine cronologico.
  • Ogni evento viene scritto con il campo setID aggiuntivo in modo che il set venga conservato nel log.
  • Il file di stato conserva, per ogni insieme, l'ultimo lastEventDate visto e gli hash degli eventi all'interno della finestra di sovrapposizione. Ogni esecuzione rilegge OVERLAP_MINUTES prima del watermark ed elimina gli eventi già scritti, in modo che gli agenti che inviano report in ritardo vengano comunque raccolti senza duplicati.
  • Qualsiasi errore di API, accesso o archiviazione non riesce a eseguire l'esecuzione e lascia invariata la filigrana, quindi la successiva esecuzione riprova la stessa finestra.

Crea un job Cloud Scheduler

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

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

  3. Fornisci i seguenti dettagli di configurazione:

    Impostazione Valore
    Nome cyberark-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 (UTC consigliato)
    Tipo di target Pub/Sub
    Argomento Seleziona l'argomento Pub/Sub cyberark-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 15 minuti */15 * * * * Bassa latenza
Ogni ora 0 * * * * Standard (consigliato)
Ogni 6 ore 0 */6 * * * Basso volume, elaborazione batch

Testare l'integrazione

  1. Nella console Cloud Scheduler, trova il tuo job.
  2. Fai clic su Forza esecuzione per attivare il job manualmente.
  3. Attendi alcuni secondi.
  4. Vai a Cloud Run > Servizi.
  5. Fai clic sul nome della funzione (cyberark-epm-collector).
  6. Fai clic sulla scheda Log.

  7. Verifica che la funzione sia stata eseguita correttamente. Cerca:

    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. Vai a Cloud Storage > Bucket.

  9. Fai clic sul nome del bucket.

  10. Vai alla cartella del prefisso (cyberark-epm/).

  11. Verifica che sia stato creato un nuovo file .ndjson con il timestamp corrente.

Se nei log vengono visualizzati errori:

  • HTTP 401: controlla EPM_USERNAME e EPM_PASSWORD o che l'accesso sia stato ripetuto entro un minuto
  • HTTP 403: l'utente non dispone di un ruolo nel set o è stato raggiunto il limite di utilizzo dell'API del tenant
  • HTTP 404: ID set errato in EPM_SET_IDS
  • HTTP 429: limitazione della frequenza: la funzione riprova con backoff
  • Variabili di ambiente mancanti: verifica che tutte le variabili richieste siano impostate

Recuperare il account di servizio Google SecOps

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

Recupera l'email del account di servizio

  1. Vai a Impostazioni SIEM > Feed.
  2. Fai clic su Aggiungi nuovo feed di notizie.
  3. Fai clic su Configura un singolo feed.
  4. Nel campo Nome feed, inserisci un nome per il feed (ad esempio, CyberArk EPM logs).
  5. Seleziona Google Cloud Storage V2 come Tipo di origine.
  6. Seleziona CyberArk Endpoint Privilege Manager (EPM) come Tipo di log.
  7. Fai clic su Ottieni service account.

  8. Verrà visualizzato un indirizzo email univoco del account di servizio, ad esempio:

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

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

  10. Fai clic su Avanti.

  11. 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://cyberark-epm-secops-logs/cyberark-epm/
      • Sostituisci cyberark-epm-secops-logs con il nome del tuo bucket Cloud Storage.

    • 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 (opzione consigliata per i test).

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

    • Età massima dei 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

  12. Fai clic su Avanti.

  13. Controlla la nuova configurazione del feed di notizie nella schermata Finalizza, quindi fai clic su Invia.

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'email del account di servizio Google SecOps
    • Assegna ruoli: seleziona entrambi i seguenti ruoli:
      • Visualizzatore oggetti Storage: legge gli oggetti log.
      • Storage Legacy Bucket Reader: legge i metadati del bucket. Se hai selezionato l'opzione di eliminazione Elimina file trasferiti e directory vuote, seleziona Storage Legacy Bucket Writer, che concede anche l'autorizzazione di eliminazione.
  6. Fai clic su Salva.

Tabella di mappatura UDM

Campo log Mappatura UDM Logica
accessTargetType_label additional.fields Unita
accessaction_label additional.fields Unita
agentEventCount_label additional.fields Unita
arrivalTime_label additional.fields Unita
deceptionType_label additional.fields Unita
eventCount_label additional.fields Unita
logonAttemptTypeId_label additional.fields Unita
logonStatusId_label additional.fields Unita
modificationTime_label additional.fields Unita
originalFileName_label additional.fields Unita
owner_label additional.fields Unita
packageName_label additional.fields Unita
publisher_field additional.fields Unita
publisher_label additional.fields Unita
setID_label additional.fields Unita
skippedCount_label additional.fields Unita
sourceProcessSigner_field additional.fields Unita
sourceProcessSigner_label additional.fields Unita
sourceType_label additional.fields Unita
threatProtectionActionId_label additional.fields Unita
threatProtectionAction_label additional.fields Unita
user_is_admin_label additional.fields Unita
winEventRecordId_label additional.fields Unita
winEventType_label additional.fields Unita
workingDirectory_label additional.fields Unita
displayName metadata.description Mappato direttamente
eachevent.displayName metadata.description Mappato direttamente
lastEventDate metadata.event_timestamp Analizzato come ISO8601
has_principal metadata.event_type Mappato: true → STATUS_UPDATE
has_principal_user metadata.event_type Mappato: true → USER_UNCATEGORIZED
eachevent.eventType metadata.product_event_type Mappato direttamente
eventType metadata.product_event_type Mappato direttamente
productName metadata.product_version Mappato direttamente
productVersion metadata.product_version Mappato direttamente
url network.http.referral_url Mappato direttamente
agentId principal.asset.asset_id Mappato direttamente
eachevent.agentId principal.asset.asset_id Mappato direttamente
computerName principal.asset.hostname Mappato direttamente
sourceName principal.file.full_path Mappato direttamente
computerName principal.hostname Mappato direttamente
eachevent.computerName principal.hostname Mappato direttamente
operatingSystemType principal.platform Mappato direttamente
operating_system_type principal.platform Mappato: Windows → WINDOWS
company principal.user.company_name Mappato direttamente
eachevent.userName principal.user.userid Mappato direttamente
userName principal.user.userid Mappato direttamente
sec_result security_result Unita
threat_protection_action security_result.action_details Mappato direttamente
fileDescription security_result.description Mappato direttamente
eachevent.policyName security_result.rule_name Mappato direttamente
policyName security_result.rule_name Mappato direttamente
justification security_result.summary Mappato direttamente
fileQualifier target.asset.asset_id Mappato direttamente
software target.asset.software Unita
eachevent.filePath target.file.full_path Mappato direttamente
filePath target.file.full_path Mappato direttamente
eachevent.hash target.file.sha1 Mappato direttamente
hash target.file.sha1 Mappato direttamente
fileSize target.file.size Mappato direttamente
eachevent.processCommandLine target.process.command_line Rinominate/mappate
processCommandLine target.process.command_line Mappato direttamente
fileName target.process.file.names Unita
eachevent.sourceProcessCommandLine target.process.parent_process.command_line Rinominato/mappato
sourceProcessCommandLine target.process.parent_process.command_line Mappato direttamente
each_event.sourceProcessHash target.process.parent_process.file.sha1 Rinominate/mappate
sourceProcessHash target.process.parent_process.file.sha1 Mappato direttamente
sourceProcessPublisher_label target.resource.attribute.labels Unita
exposed_accountName target.user.attribute.labels Unita
exposed_domain target.user.attribute.labels Unita
exposed_source target.user.attribute.labels Unita
exposed_username target.user.attribute.labels Unita
sourceProcessUsername target.user.userid Mappato direttamente
N/D metadata.event_type Costante: STATUS_UPDATE
N/D metadata.product_name Costante: EPM
N/D metadata.vendor_name Costante: CYBERARK
N/D principal.platform Costante: WINDOWS
N/D security_result.alert_state Costante: ALERTING

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.