Raccogliere i log di BeyondTrust Endpoint Privilege Management (EPM)
Questo documento spiega come importare i log di BeyondTrust Endpoint Privilege Management (EPM) in Google Security Operations utilizzando Google Cloud Storage. Il parser si concentra sulla trasformazione dei dati di log JSON non elaborati da BeyondTrust Endpoint in un formato strutturato conforme a Chronicle UDM. Innanzitutto, inizializza i valori predefiniti per vari campi e poi analizza il payload JSON, mappando successivamente campi specifici dal log non elaborato nei campi UDM corrispondenti all'interno dell'oggetto event.idm.read_only_udm.
Prima di iniziare
Assicurati di disporre dei seguenti prerequisiti:
- Un'istanza Google SecOps
- Un progetto GCP con l'API Cloud Storage abilitata
- Autorizzazioni per creare e gestire bucket GCS
- Autorizzazioni per gestire le policy IAM nei bucket GCS
- Autorizzazioni per creare servizi Cloud Run, argomenti Pub/Sub e job Cloud Scheduler
- Accesso con privilegi al tenant o all'API di BeyondTrust Endpoint Privilege Management
Creazione di un bucket Google Cloud Storage
- Vai alla console Google Cloud.
- Seleziona il tuo progetto o creane uno nuovo.
- Nel menu di navigazione, vai a Cloud Storage > Bucket.
- Fai clic su Crea bucket.
Fornisci i seguenti dettagli di configurazione:
Impostazione Valore Assegna un nome al bucket Inserisci un nome univoco globale (ad esempio beyondtrust-epm-logs).Tipo di località Scegli in base alle tue esigenze (regione singola, doppia regione, più regioni) Località Seleziona la posizione (ad esempio, us-central1).Classe di archiviazione Standard (consigliato per i log a cui si accede di frequente) Controllo dell'accesso Uniforme (consigliato) Strumenti di protezione (Facoltativo) Attiva il controllo delle versioni degli oggetti o la policy di conservazione Fai clic su Crea.
Raccogli le credenziali API di BeyondTrust EPM
- Accedi alla console web BeyondTrust Privilege Management come amministratore.
- Vai a Configurazione > Impostazioni > Impostazioni API.
- Fai clic su Crea un account API.
- Fornisci i seguenti dettagli di configurazione:
- Nome: inserisci
Google SecOps Collector. - Accesso API: abilita Audit (lettura) e altri ambiti in base alle esigenze.
- Nome: inserisci
- Copia e salva l'ID client e il client secret.
- Copia l'URL di base dell'API, in genere
https://<your-tenant>-services.pm.beyondtrustcloud.com(lo utilizzerai come BPT_API_URL).
Crea un service account per la funzione Cloud Run
La funzione Cloud Run richiede un service account con autorizzazioni di scrittura nel bucket GCS e di invocazione da parte di Pub/Sub.
Crea service account
- Nella console Google Cloud, vai a IAM e amministrazione > Service Accounts.
- Fai clic su Crea service account.
- Fornisci i seguenti dettagli di configurazione:
- Nome del service account: inserisci
beyondtrust-epm-collector-sa. - Descrizione service account: inserisci
Service account for Cloud Run function to collect BeyondTrust EPM logs.
- Nome del service account: inserisci
- Fai clic su Crea e continua.
- Nella sezione Concedi a questo service account l'accesso al progetto, aggiungi i seguenti ruoli:
- Fai clic su Seleziona un ruolo.
- Cerca e seleziona Amministratore oggetti di archiviazione.
- Fai clic su + Aggiungi un altro ruolo.
- Cerca e seleziona Cloud Run Invoker.
- Fai clic su + Aggiungi un altro ruolo.
- Cerca e seleziona Invoker di Cloud Functions.
- Fai clic su Continua.
- Fai clic su Fine.
Questi ruoli sono necessari per:
- Amministratore oggetti Storage: scrive i log nel bucket GCS e gestisce i file di stato
- Cloud Run Invoker: consente a Pub/Sub di richiamare la funzione
- Cloud Functions Invoker: consente la chiamata di funzioni
Concedi autorizzazioni IAM sul bucket GCS
Concedi al service account le autorizzazioni di scrittura sul bucket GCS:
- Vai a Cloud Storage > Bucket.
- Fai clic sul nome del bucket.
- Vai alla scheda Autorizzazioni.
- Fai clic su Concedi l'accesso.
- Fornisci i seguenti dettagli di configurazione:
- Aggiungi entità: inserisci l'email del service account (ad es.
beyondtrust-epm-collector-sa@PROJECT_ID.iam.gserviceaccount.com). - Assegna i ruoli: seleziona Storage Object Admin.
- Aggiungi entità: inserisci l'email del service account (ad es.
- Fai clic su Salva.
Crea argomento Pub/Sub
Crea un argomento Pub/Sub a cui Cloud Scheduler pubblicherà e a cui la funzione Cloud Run si iscriverà.
- Nella console GCP, vai a Pub/Sub > Argomenti.
- Fai clic su Crea argomento.
- Fornisci i seguenti dettagli di configurazione:
- ID argomento: inserisci
beyondtrust-epm-trigger. - Lascia le altre impostazioni sui valori predefiniti.
- ID argomento: inserisci
- Fai clic su Crea.
Crea una funzione Cloud Run per raccogliere i log
La funzione Cloud Run viene attivata dai messaggi Pub/Sub di Cloud Scheduler per recuperare i log dall'API BeyondTrust EPM e li scrive in GCS.
- Nella console GCP, vai a Cloud Run.
- Fai clic su Crea servizio.
- Seleziona Funzione (usa un editor in linea per creare una funzione).
Nella sezione Configura, fornisci i seguenti dettagli di configurazione:
Impostazione Valore Nome servizio beyondtrust-epm-collectorRegione Seleziona la regione corrispondente al tuo bucket GCS (ad esempio us-central1)Runtime Seleziona Python 3.12 o versioni successive Nella sezione Trigger (facoltativo):
- Fai clic su + Aggiungi trigger.
- Seleziona Cloud Pub/Sub.
- In Seleziona un argomento Cloud Pub/Sub, scegli l'argomento
beyondtrust-epm-trigger. - Fai clic su Salva.
Nella sezione Autenticazione:
- Seleziona Richiedi autenticazione.
- Controlla Identity and Access Management (IAM).
Scorri verso il basso ed espandi Container, networking, sicurezza.
Vai alla scheda Sicurezza:
- Service account: seleziona il service account
beyondtrust-epm-collector-sa.
- Service account: seleziona il service account
Vai alla scheda Container:
- Fai clic su Variabili e secret.
- Fai clic su + Aggiungi variabile per ogni variabile di ambiente:
Nome variabile Valore di esempio GCS_BUCKETbeyondtrust-epm-logsGCS_PREFIXbeyondtrust-epm/STATE_KEYbeyondtrust-epm/state.jsonBPT_API_URLhttps://yourtenant-services.pm.beyondtrustcloud.comCLIENT_IDyour-client-idCLIENT_SECRETyour-client-secretOAUTH_SCOPEmanagement-apiRECORD_SIZE1000MAX_ITERATIONS10Scorri verso il basso nella scheda Variabili e secret fino a Richieste:
- Timeout richiesta: inserisci
600secondi (10 minuti).
- Timeout richiesta: inserisci
Vai alla scheda Impostazioni in Container:
- Nella sezione Risorse:
- Memoria: seleziona 512 MiB o un valore superiore.
- CPU: seleziona 1.
- Fai clic su Fine.
- Nella sezione Risorse:
Scorri fino a Ambiente di esecuzione:
- Seleziona Predefinito (opzione consigliata).
Nella sezione Scalabilità della revisione:
- Numero minimo di istanze: inserisci
0. - Numero massimo di istanze: inserisci
100(o modifica in base al carico previsto).
- Numero minimo di istanze: inserisci
Fai clic su Crea.
Attendi la creazione del servizio (1-2 minuti).
Dopo aver creato il servizio, si apre automaticamente l'editor di codice incorporato.
Aggiungi codice per la funzione
- Inserisci main in Entry point della funzione
Nell'editor di codice incorporato, crea due file:
- Primo file: main.py:
import functions_framework from google.cloud import storage import json import os import urllib3 from datetime import datetime, timezone, timedelta import time # Initialize HTTP client http = urllib3.PoolManager() # Initialize Storage client storage_client = storage.Client() @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch logs from BeyondTrust EPM API and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ # Get environment variables 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') # BeyondTrust EPM API credentials api_url = os.environ.get('BPT_API_URL') client_id = os.environ.get('CLIENT_ID') client_secret = os.environ.get('CLIENT_SECRET') oauth_scope = os.environ.get('OAUTH_SCOPE', 'management-api') record_size = int(os.environ.get('RECORD_SIZE', '1000')) max_iterations = int(os.environ.get('MAX_ITERATIONS', '10')) if not all([bucket_name, api_url, client_id, client_secret]): print('Error: Missing required environment variables') return try: # Get GCS bucket bucket = storage_client.bucket(bucket_name) # Load state (last processed timestamp) state = load_state(bucket, state_key) last_timestamp = state.get('last_timestamp', (datetime.utcnow() - timedelta(hours=24)).strftime("%Y-%m-%dT%H:%M:%SZ")) print(f'Processing logs since {last_timestamp}') # Get OAuth access token token = get_oauth_token(api_url, client_id, client_secret, oauth_scope) # Fetch audit events events = fetch_audit_events(api_url, token, last_timestamp, record_size, max_iterations) if events: # Store events in GCS current_timestamp = datetime.utcnow() filename = f"{prefix}beyondtrust-epm-events-{current_timestamp.strftime('%Y%m%d_%H%M%S')}.json" store_events_to_gcs(bucket, filename, events) # Update state with latest timestamp latest_timestamp = get_latest_event_timestamp(events) save_state(bucket, state_key, {'last_timestamp': latest_timestamp, 'updated_at': datetime.utcnow().isoformat() + 'Z'}) print(f'Successfully processed {len(events)} events and stored to {filename}') else: print('No new events found') except Exception as e: print(f'Error processing logs: {str(e)}') raise def get_oauth_token(api_url, client_id, client_secret, scope): """ Get OAuth access token using client credentials flow for BeyondTrust EPM. Uses the correct endpoint: /oauth/token """ token_url = f"{api_url}/oauth/token" headers = {'Content-Type': 'application/x-www-form-urlencoded'} body = f"grant_type=client_credentials&client_id={client_id}&client_secret={client_secret}&scope={scope}" response = http.request('POST', token_url, headers=headers, body=body, timeout=urllib3.Timeout(60.0)) if response.status != 200: raise RuntimeError(f"Token request failed: {response.status} {response.data[:256]!r}") token_data = json.loads(response.data.decode('utf-8')) return token_data['access_token'] def fetch_audit_events(api_url, access_token, last_timestamp, record_size, max_iterations): """ Fetch audit events using the BeyondTrust EPM API endpoint: /management-api/v2/AuditEvents with query parameters for filtering and pagination """ headers = { 'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json' } all_events = [] current_start_date = last_timestamp iterations = 0 # Enforce maximum RecordSize limit of 1000 based on BeyondTrust documentation record_size_limited = min(record_size, 1000) while iterations < max_iterations: iterations += 1 if len(all_events) >= 10000: print(f"Reached maximum events limit (10000)") break # Use the BeyondTrust EPM API endpoint for audit events query_url = f"{api_url}/management-api/v2/AuditEvents" params = { 'StartDate': current_start_date, 'RecordSize': record_size_limited } # Construct URL with query parameters query_string = '&'.join([f"{k}={v}" for k, v in params.items()]) full_url = f"{query_url}?{query_string}" try: response = http.request('GET', full_url, headers=headers, timeout=urllib3.Timeout(300.0)) # Handle rate limiting with exponential backoff if response.status == 429: retry_after = int(response.headers.get('Retry-After', '30')) print(f"Rate limited (429). Retrying after {retry_after}s...") time.sleep(retry_after) continue if response.status != 200: raise RuntimeError(f"API request failed: {response.status} {response.data[:256]!r}") response_data = json.loads(response.data.decode('utf-8')) events = response_data.get('events', []) if not events: break print(f"Page {iterations}: Retrieved {len(events)} events") all_events.extend(events) # If we got fewer events than RecordSize, we've reached the end if len(events) < record_size_limited: break # For pagination, update StartDate to the timestamp of the last event last_event = events[-1] last_timestamp = extract_event_timestamp(last_event) if not last_timestamp: print("Warning: Could not find timestamp in last event for pagination") break # Convert to datetime and add 1 second to avoid retrieving the same event again try: dt = parse_timestamp(last_timestamp) dt = dt + timedelta(seconds=1) current_start_date = dt.strftime("%Y-%m-%dT%H:%M:%SZ") except Exception as e: print(f"Error parsing timestamp {last_timestamp}: {e}") break except Exception as e: print(f"Error fetching page {iterations}: {e}") break return all_events def extract_event_timestamp(event): """Extract timestamp from event, checking multiple possible fields""" # Check common timestamp fields timestamp_fields = ['event.dateTime', 'event.timestamp', 'timestamp', 'eventTime', 'dateTime', 'whenOccurred', 'date', 'time', 'event.ingested'] # Try nested event.dateTime first (common in BeyondTrust) if isinstance(event, dict) and isinstance(event.get("event"), dict): ts = event["event"].get("dateTime") if ts: return ts ts = event["event"].get("timestamp") if ts: return ts # Fallback to other timestamp fields for field in timestamp_fields: if field in event and event[field]: return event[field] return None def parse_timestamp(timestamp_str): """Parse timestamp string to datetime object, handling various formats""" if isinstance(timestamp_str, (int, float)): # Unix timestamp (in milliseconds or seconds) if timestamp_str > 1e12: # Milliseconds return datetime.fromtimestamp(timestamp_str / 1000, tz=timezone.utc) else: # Seconds return datetime.fromtimestamp(timestamp_str, tz=timezone.utc) if isinstance(timestamp_str, str): # Try different string formats try: # ISO format with Z if timestamp_str.endswith('Z'): return datetime.fromisoformat(timestamp_str.replace('Z', '+00:00')) # ISO format with timezone elif '+' in timestamp_str or timestamp_str.endswith('00:00'): return datetime.fromisoformat(timestamp_str) # ISO format without timezone (assume UTC) else: dt = datetime.fromisoformat(timestamp_str) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt except ValueError: pass raise ValueError(f"Could not parse timestamp: {timestamp_str}") def load_state(bucket, key): """Load state from GCS.""" try: blob = bucket.blob(key) if blob.exists(): state_data = blob.download_as_text() return json.loads(state_data) except Exception as e: print(f'Warning: Could not load state: {str(e)}') return {} def save_state(bucket, key, state): """Save state to GCS.""" try: blob = bucket.blob(key) blob.upload_from_string( json.dumps(state), content_type='application/json' ) except Exception as e: print(f'Warning: Could not save state: {str(e)}') def store_events_to_gcs(bucket, key, events): """Store events as JSONL (one JSON object per line) in GCS""" # Convert to JSONL format (one JSON object per line) jsonl_content = '\n'.join(json.dumps(event, default=str) for event in events) blob = bucket.blob(key) blob.upload_from_string(jsonl_content, content_type='application/x-ndjson') def get_latest_event_timestamp(events): """Get the latest timestamp from the events for state tracking""" if not events: return datetime.utcnow().isoformat() + 'Z' latest = None for event in events: timestamp = extract_event_timestamp(event) if timestamp: try: event_dt = parse_timestamp(timestamp) event_iso = event_dt.isoformat() + 'Z' if latest is None or event_iso > latest: latest = event_iso except Exception as e: print(f"Error parsing event timestamp {timestamp}: {e}") continue return latest or datetime.utcnow().isoformat() + 'Z'- Secondo file: requirements.txt::
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0Fai clic su Esegui il deployment per salvare la funzione ed eseguirne il deployment.
Attendi il completamento del deployment (2-3 minuti).
Crea job Cloud Scheduler
Cloud Scheduler pubblica messaggi nell'argomento Pub/Sub a intervalli regolari, attivando la funzione Cloud Run.
- Nella console di GCP, vai a Cloud Scheduler.
- Fai clic su Crea job.
Fornisci i seguenti dettagli di configurazione:
Impostazione Valore Nome beyondtrust-epm-collector-hourlyRegione Seleziona la stessa regione della funzione Cloud Run Frequenza 0 * * * *(ogni ora, all'ora)Fuso orario Seleziona il fuso orario (UTC consigliato) Tipo di target Pub/Sub Argomento Seleziona l'argomento beyondtrust-epm-triggerCorpo del messaggio {}(oggetto JSON vuoto)Fai clic su Crea.
Opzioni di frequenza di pianificazione
Scegli la frequenza in base al volume dei log e ai requisiti di latenza:
Frequenza Espressione cron Caso d'uso Ogni 5 minuti */5 * * * *Volume elevato, bassa latenza Ogni 15 minuti */15 * * * *Volume medio Ogni ora 0 * * * *Standard (consigliato) Ogni 6 ore 0 */6 * * *Volume basso, elaborazione batch Ogni giorno 0 0 * * *Raccolta dei dati storici
Testa il job di pianificazione
- Nella console Cloud Scheduler, trova il job.
- Fai clic su Forza esecuzione per attivare manualmente.
- Attendi qualche secondo e vai a Cloud Run > Servizi > beyondtrust-epm-collector > Log.
- Verifica che la funzione sia stata eseguita correttamente.
- Controlla il bucket GCS per verificare che i log siano stati scritti.
Recuperare il service account Google SecOps
Google SecOps utilizza un service account univoco per leggere i dati dal tuo bucket GCS. Devi concedere a questo service account l'accesso al tuo bucket.
Recuperare l'email del service account
- Vai a Impostazioni SIEM > Feed.
- Fai clic su Aggiungi nuovo feed.
- Fai clic su Configura un singolo feed.
- Nel campo Nome feed, inserisci un nome per il feed (ad esempio,
BeyondTrust EPM logs). - Seleziona Google Cloud Storage V2 come Tipo di origine.
- Seleziona BeyondTrust Endpoint Privilege Management come Tipo di log.
Fai clic su Ottieni service account. Verrà visualizzata un'email univoca del service account, ad esempio:
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.comCopia questo indirizzo email per utilizzarlo nel passaggio successivo.
Concedi le autorizzazioni IAM al service account Google SecOps
Il service account Google SecOps deve avere il ruolo Visualizzatore oggetti Storage nel bucket GCS.
- Vai a Cloud Storage > Bucket.
- Fai clic sul nome del bucket.
- Vai alla scheda Autorizzazioni.
- Fai clic su Concedi l'accesso.
- Fornisci i seguenti dettagli di configurazione:
- Aggiungi entità: incolla l'email del service account Google SecOps.
- Assegna i ruoli: seleziona Visualizzatore oggetti Storage.
Fai clic su Salva.
Configura un feed in Google SecOps per importare i log di BeyondTrust EPM
- Vai a Impostazioni SIEM > Feed.
- Fai clic su Aggiungi nuovo feed.
- Fai clic su Configura un singolo feed.
- Nel campo Nome feed, inserisci un nome per il feed (ad esempio,
BeyondTrust EPM logs). - Seleziona Google Cloud Storage V2 come Tipo di origine.
- Seleziona BeyondTrust Endpoint Privilege Management come Tipo di log.
- Fai clic su Avanti.
Specifica i valori per i seguenti parametri di input:
URL del bucket di archiviazione: inserisci l'URI del bucket GCS con il percorso del prefisso:
gs://beyondtrust-epm-logs/beyondtrust-epm/Sostituisci:
beyondtrust-epm-logs: il nome del bucket GCS.beyondtrust-epm/: (Facoltativo) prefisso/percorso della cartella in cui vengono archiviati i log (lascia vuoto per la radice).
Esempi:
- Bucket radice:
gs://beyondtrust-epm-logs/ - Con prefisso:
gs://beyondtrust-epm-logs/beyondtrust-epm/
- Bucket radice:
Opzione di eliminazione dell'origine: seleziona l'opzione di eliminazione in base alle tue preferenze:
- Mai: non elimina mai i file dopo i trasferimenti (opzione consigliata per i test).
- Elimina file trasferiti: elimina i file dopo il trasferimento riuscito.
Elimina file trasferiti e directory vuote: elimina i file e le directory vuote dopo il trasferimento riuscito.
Età massima del file: includi i file modificati nell'ultimo numero di 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.
Fai clic su Avanti.
Controlla la nuova configurazione del feed nella schermata Finalizza e poi fai clic su Invia.
Tabella di mappatura UDM
| Campo log | Mappatura UDM | Funzione logica |
|---|---|---|
| agent.id | principal.asset.attribute.labels.value | Mappato all'etichetta con la chiave agent_id |
| agent.version | principal.asset.attribute.labels.value | Mappato all'etichetta con la chiave agent_version |
| ecs.version | principal.asset.attribute.labels.value | Mappato all'etichetta con la chiave ecs_version |
| event_data.reason | metadata.description | Descrizione dell'evento dal log non elaborato |
| event_datas.ActionId | metadata.product_log_id | Identificatore log specifico del prodotto |
| file.path | principal.file.full_path | Percorso file completo dell'evento |
| headers.content_length | additional.fields.value.string_value | Mappato all'etichetta con key content_length |
| headers.content_type | additional.fields.value.string_value | Mappato all'etichetta con key content_type |
| headers.http_host | additional.fields.value.string_value | Mappato all'etichetta con la chiave http_host |
| headers.http_version | network.application_protocol_version | Versione del protocollo HTTP |
| headers.request_method | network.http.method | Metodo di richiesta HTTP |
| host.hostname | principal.hostname | Nome host principale |
| host.hostname | principal.asset.hostname | Nome host dell'asset principale |
| host.ip | principal.asset.ip | Indirizzo IP dell'asset principale |
| host.ip | principal.ip | Indirizzo IP principale |
| host.mac | principal.mac | Indirizzo MAC principale |
| host.os.platform | principal.platform | Imposta su MAC se è uguale a macOS |
| host.os.version | principal.platform_version | Versione sistema operativo |
| labels.related_item_id | metadata.product_log_id | Identificatore dell'elemento correlato |
| process.command_line | principal.process.command_line | Riga di comando del processo |
| process.name | additional.fields.value.string_value | Mappato all'etichetta con la chiave process_name |
| process.parent.name | additional.fields.value.string_value | Mappato all'etichetta con la chiave process_parent_name |
| process.parent.pid | principal.process.parent_process.pid | PID del processo padre convertito in stringa |
| process.pid | principal.process.pid | Process PID convertito in stringa |
| user.id | principal.user.userid | Identificatore utente |
| user.name | principal.user.user_display_name | Nome visualizzato dell'utente |
| N/D | metadata.event_timestamp | Timestamp dell'evento impostato sul timestamp della voce di log |
| N/D | metadata.event_type | GENERIC_EVENT se non è presente un principal, altrimenti STATUS_UPDATE |
| N/D | network.application_protocol | Imposta su HTTP se il campo http_version contiene HTTP |
Hai bisogno di ulteriore assistenza? Ricevi risposte dai membri della community e dai professionisti di Google SecOps.