Raccogliere gli audit log della console NetApp (in precedenza BlueXP)
Questo documento spiega come importare i log di controllo di NetApp Console (in precedenza BlueXP) in Google Security Operations utilizzando Google Cloud Storage V2.
NetApp Console è un control plane unificato per la gestione dei servizi di archiviazione e dati multi-cloud ibridi in ambienti on-premise e cloud. Il servizio Audit registra le operazioni eseguite dai servizi della Console, inclusi gli indirizzi IP di origine, gli spazi di lavoro, gli agenti della Console utilizzati e altri dati di telemetria utili per l'analisi forense e i requisiti di conformità.
Prima di iniziare
Assicurati di soddisfare i seguenti prerequisiti:
- Un'istanza Google SecOps
- Un progetto GCP con le API Cloud Storage, Cloud Run, Pub/Sub e Cloud Scheduler abilitate
- 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 amministrativo alla console NetApp con autorizzazioni per creare service account
- Il tuo ID account NetApp Console
Configurare l'accesso all'API della console NetApp
Per consentire a Google SecOps di recuperare i log di controllo, devi creare un account di servizio con le autorizzazioni appropriate e generare le credenziali API.
Recuperare l'ID account
- Vai alla console NetApp utilizzando un browser.
- Accedi utilizzando le tue credenziali della console NetApp o del sito di assistenza NetApp.
- Fai clic sul menu a discesa Account nella parte superiore della pagina.
- Fai clic su Gestisci account per l'account selezionato.
Nella sezione Panoramica, copia il valore dell'ID account.
Crea un account di servizio
- Nella console NetApp, vai ad Administration > Identity and access (Amministrazione > Identità e accesso).
- Seleziona Membri.
- Seleziona Aggiungi un membro.
- In Tipo di membro, seleziona Service account.
- Inserisci un nome per il account di servizio (ad esempio,
Google SecOps Integration). Lascia deselezionata l'opzione Utilizza l'autenticazione JWT con chiave privata per utilizzare l'autenticazione con client secret.
Nella sezione Seleziona un'organizzazione, una cartella o un progetto, seleziona la tua organizzazione.
In Categoria, seleziona Organizzazione.
In Ruolo, seleziona Visualizzatore organizzazione.
Fai clic su Aggiungi.
Registra le credenziali API
Dopo aver creato il account di servizio, viene visualizzata una finestra di dialogo con le credenziali:
- ID client: il tuo identificatore client univoco (ad esempio,
TvPPs4SeM5smEElsGmdDUznljhN3YY8s) - Client secret: la tua chiave segreta API
- Scarica o copia sia l'ID client che il client secret in una posizione sicura.
- Fai clic su Chiudi.
Verifica le autorizzazioni
Per verificare che il account di servizio disponga delle autorizzazioni richieste:
- Accedi alla console NetApp.
- Vai ad Amministrazione > Identità e accesso.
- Seleziona Membri.
- Individua il account di servizio che hai creato e verifica che disponga del ruolo Visualizzatore organizzazione.
- Se il ruolo non è assegnato, fai clic sul nome del account di servizio e aggiorna il ruolo a Visualizzatore organizzazione.
Testare l'accesso API
Testa le tue credenziali prima di procedere con l'integrazione:
# Replace with your actual credentials CLIENT_ID="your-client-id" CLIENT_SECRET="your-client-secret" ACCOUNT_ID="your-account-id" # Obtain access token TOKEN=$(curl -s -X POST "https://netapp-cloud-account.auth0.com/oauth/token" \ -H "Content-Type: application/json" \ -d '{ "grant_type": "client_credentials", "client_id": "'"${CLIENT_ID}"'", "client_secret": "'"${CLIENT_SECRET}"'", "audience": "https://api.cloud.netapp.com" }' | python3 -c "import sys,json; print(json.load(sys.stdin)['access_token'])") # Test audit API access curl -s -H "Authorization: Bearer ${TOKEN}" \ "https://cloudmanager.cloud.netapp.com/audit/${ACCOUNT_ID}?offset=0" \ | python3 -m json.tool
Una risposta positiva restituisce un oggetto JSON contenente i record di controllo per l'account specificato.
Crea 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 netapp-bluexp-audit-logs).Tipo di località Scegli in base alle tue esigenze (regione singola, a due regioni, multiregionale) 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) Abilita il controllo delle versioni degli oggetti o la policy di conservazione Fai clic su Crea.
Crea un account di servizio per la funzione Cloud Run
- Nella console Google Cloud, vai a IAM e amministrazione > Account di servizio.
- Fai clic su Crea account di servizio.
- Fornisci i seguenti dettagli di configurazione:
- Nome del service account: inserisci
netapp-bluexp-collector-sa - Descrizione service account: inserisci
Service account for Cloud Run function to collect NetApp Console audit logs
- Nome del service account: inserisci
- Fai clic su Crea e continua.
- Nella sezione Concedi a questo account di servizio l'accesso al progetto, aggiungi i seguenti ruoli:
- Fai clic su Seleziona un ruolo.
- Cerca e seleziona Storage Object Admin.
- Fai clic su + Aggiungi un altro ruolo.
- Cerca e seleziona Cloud Run Invoker.
- Fai clic su + Aggiungi un altro ruolo.
- Cerca e seleziona Cloud Functions Invoker.
- Fai clic su Continua.
- Fai clic su Fine.
Concedi autorizzazioni IAM sul bucket GCS
- Vai a Cloud Storage > Bucket.
- Fai clic sul nome del bucket (
netapp-bluexp-audit-logs). - Vai alla scheda Autorizzazioni.
- Fai clic su Concedi l'accesso.
- Fornisci i seguenti dettagli di configurazione:
- Aggiungi entità: inserisci l'email del account di servizio (
netapp-bluexp-collector-sa@PROJECT_ID.iam.gserviceaccount.com). - Assegna i ruoli: seleziona Storage Object Admin.
- Aggiungi entità: inserisci l'email del account di servizio (
- Fai clic su Salva.
Crea argomento Pub/Sub
- Nella console Google Cloud, vai a Pub/Sub > Argomenti.
- Fai clic su Crea argomento.
- Fornisci i seguenti dettagli di configurazione:
- ID argomento: inserisci
netapp-bluexp-audit-trigger - Lascia invariate le altre impostazioni predefinite.
- ID argomento: inserisci
- 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 audit log dall'API NetApp Console Audit e scriverli in GCS.
- Nella console Google Cloud, vai a Cloud Run.
- Fai clic su Crea servizio.
- Seleziona Funzione (utilizza un editor in linea per creare una funzione).
Nella sezione Configura, fornisci i seguenti dettagli di configurazione:
Impostazione Valore Nome servizio netapp-bluexp-collectorRegione Seleziona la regione corrispondente al tuo bucket GCS (ad esempio, us-central1)Tempo di esecuzione 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
netapp-bluexp-audit-trigger. - Fai clic su Salva.
Nella sezione Autenticazione:
- Seleziona Richiedi autenticazione.
- Seleziona Identity and Access Management (IAM).
Scorri verso il basso ed espandi Container, networking, sicurezza.
Vai alla scheda Sicurezza:
- Service account: seleziona
netapp-bluexp-collector-sa.
- Service account: seleziona
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 Descrizione GCS_BUCKETnetapp-bluexp-audit-logsNome bucket GCS GCS_PREFIXnetapp-bluexp-auditPrefisso per i file di log STATE_KEYnetapp-bluexp-audit/state.jsonPercorso file di stato NETAPP_CLIENT_IDyour-client-idID client del account di servizio della console NetApp NETAPP_CLIENT_SECRETyour-client-secretClient secret account di servizio del servizio NetApp Console NETAPP_ACCOUNT_IDaccount-AbCdEfGhID account NetApp Console MAX_RECORDS5000Numero massimo di record per esecuzione LOOKBACK_HOURS24Periodo di riferimento iniziale Nella sezione Variabili e secret, scorri verso il basso fino a Richieste:
- Timeout richiesta: inserisci
600secondi (10 minuti)
- Timeout richiesta: inserisci
Vai alla scheda Impostazioni:
- Nella sezione Risorse:
- Memoria: seleziona 512 MiB o un valore superiore.
- CPU: seleziona 1
- Nella sezione Risorse:
Nella sezione Scalabilità della revisione:
- Numero minimo di istanze: inserisci
0 - Numero massimo di istanze: inserisci
100
- Numero minimo di istanze: inserisci
Fai clic su Crea.
Attendi la creazione del servizio (1-2 minuti).
Dopo aver creato il servizio, si aprirà automaticamente l'editor di codice incorporato.
Aggiungi codice per la funzione
- Inserisci main nel campo Entry point (Punto di ingresso).
Nell'editor di codice incorporato, crea due 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 http = urllib3.PoolManager( timeout=urllib3.Timeout(connect=5.0, read=30.0), retries=False, ) storage_client = storage.Client() GCS_BUCKET = os.environ.get('GCS_BUCKET') GCS_PREFIX = os.environ.get('GCS_PREFIX', 'netapp-bluexp-audit') STATE_KEY = os.environ.get('STATE_KEY', 'netapp-bluexp-audit/state.json') NETAPP_CLIENT_ID = os.environ.get('NETAPP_CLIENT_ID') NETAPP_CLIENT_SECRET = os.environ.get('NETAPP_CLIENT_SECRET') NETAPP_ACCOUNT_ID = os.environ.get('NETAPP_ACCOUNT_ID') MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '5000')) LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24')) TOKEN_URL = 'https://netapp-cloud-account.auth0.com/oauth/token' API_BASE = 'https://cloudmanager.cloud.netapp.com' AUDIENCE = 'https://api.cloud.netapp.com' def to_unix_millis(dt): if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) dt = dt.astimezone(timezone.utc) return int(dt.timestamp() * 1000) def parse_datetime(value): if value.endswith('Z'): value = value[:-1] + '+00:00' return datetime.fromisoformat(value) def get_access_token(): body = json.dumps({ 'grant_type': 'client_credentials', 'client_id': NETAPP_CLIENT_ID, 'client_secret': NETAPP_CLIENT_SECRET, 'audience': AUDIENCE, }).encode('utf-8') response = http.request( 'POST', TOKEN_URL, body=body, headers={'Content-Type': 'application/json'} ) if response.status != 200: raise Exception( f"Token request failed: {response.status} - " f"{response.data.decode('utf-8')}" ) data = json.loads(response.data.decode('utf-8')) token = data.get('access_token') if not token: raise Exception('No access_token in token response') print('Successfully obtained NetApp Console access token') return token @functions_framework.cloud_event def main(cloud_event): if not all([GCS_BUCKET, NETAPP_CLIENT_ID, NETAPP_CLIENT_SECRET, NETAPP_ACCOUNT_ID]): print('Error: Missing required environment variables') return try: bucket = storage_client.bucket(GCS_BUCKET) state = load_state(bucket) now = datetime.now(timezone.utc) if isinstance(state, dict) and state.get('last_event_time'): try: last_val = state['last_event_time'] if last_val.endswith('Z'): last_val = last_val[:-1] + '+00:00' last_time = datetime.fromisoformat(last_val) last_time = last_time - timedelta(minutes=2) except Exception as e: print(f'Warning: Could not parse last_event_time: {e}') last_time = now - timedelta(hours=LOOKBACK_HOURS) else: last_time = now - timedelta(hours=LOOKBACK_HOURS) print(f'Fetching audit logs from {last_time.isoformat()} to {now.isoformat()}') token = get_access_token() records, newest_time = fetch_audit_logs(token, last_time, now) if not records: print('No new audit records found.') save_state(bucket, now.isoformat()) return timestamp = now.strftime('%Y%m%d_%H%M%S') object_key = f'{GCS_PREFIX}/netapp_bluexp_audit_{timestamp}.ndjson' blob = bucket.blob(object_key) ndjson = '\n'.join( [json.dumps(r, ensure_ascii=False, default=str) for r in records] ) + '\n' blob.upload_from_string(ndjson, content_type='application/x-ndjson') print(f'Wrote {len(records)} records to gs://{GCS_BUCKET}/{object_key}') save_state(bucket, newest_time if newest_time else now.isoformat()) print(f'Successfully processed {len(records)} audit records') except Exception as e: print(f'Error processing audit logs: {str(e)}') raise def fetch_audit_logs(token, start_time, end_time): endpoint = f'{API_BASE}/audit/{NETAPP_ACCOUNT_ID}' headers = { 'Authorization': f'Bearer {token}', 'Accept': 'application/json', 'User-Agent': 'GoogleSecOps-NetAppBlueXPCollector/1.0', } start_millis = to_unix_millis(start_time) end_millis = to_unix_millis(end_time) records = [] newest_time = None offset = 0 page_num = 0 backoff = 1.0 while True: page_num += 1 if len(records) >= MAX_RECORDS: print(f'Reached max_records limit ({MAX_RECORDS})') break url = ( f'{endpoint}' f'?fromLastModified={start_millis}' f'&toLastModified={end_millis}' f'&offset={offset}' ) try: response = http.request('GET', url, headers=headers) if response.status == 429: retry_after = int(response.headers.get('Retry-After', str(int(backoff)))) print(f'Rate limited (429). Retrying after {retry_after}s...') time.sleep(retry_after) backoff = min(backoff * 2, 30.0) continue backoff = 1.0 if response.status != 200: print(f'HTTP Error: {response.status}') response_text = response.data.decode('utf-8') print(f'Response body: {response_text}') return [], None data = json.loads(response.data.decode('utf-8')) page_results = data.get('auditEntries', []) if not page_results: print('No more results (empty page)') break print(f'Page {page_num}: Retrieved {len(page_results)} audit records') records.extend(page_results) for entry in page_results: try: last_modified = entry.get('lastModified') if last_modified: entry_dt = datetime.fromtimestamp( last_modified / 1000, tz=timezone.utc ) entry_time = entry_dt.isoformat() if newest_time is None or parse_datetime(entry_time) > parse_datetime(newest_time): newest_time = entry_time except Exception as e: print(f'Warning: Could not parse entry time: {e}') count = data.get('count', len(page_results)) if count < 100: print(f'Reached last page (count={count})') break offset += count except Exception as e: print(f'Error fetching audit logs: {e}') return [], None print(f'Retrieved {len(records)} total audit records from {page_num} pages') return records, newest_time def load_state(bucket): try: blob = bucket.blob(STATE_KEY) if blob.exists(): return json.loads(blob.download_as_text()) except Exception as e: print(f'Warning: Could not load state: {e}') return {} def save_state(bucket, last_event_time_iso): try: state = { 'last_event_time': last_event_time_iso, 'last_run': datetime.now(timezone.utc).isoformat(), } blob = bucket.blob(STATE_KEY) blob.upload_from_string( json.dumps(state, indent=2), content_type='application/json' ) print(f'Saved state: last_event_time={last_event_time_iso}') except Exception as e: print(f'Warning: Could not save state: {e}')- 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 pubblicherà messaggi nell'argomento Pub/Sub a intervalli regolari, attivando la funzione Cloud Run.
- Nella console Google Cloud, vai a Cloud Scheduler.
- Fai clic su Crea job.
Fornisci i seguenti dettagli di configurazione:
Impostazione Valore Nome netapp-bluexp-collector-hourlyRegione 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 netapp-bluexp-audit-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 |
Testare l'integrazione
- Nella console Cloud Scheduler, trova il tuo job (
netapp-bluexp-collector-hourly). - Fai clic su Forza esecuzione per attivare il job manualmente.
- Attendi alcuni secondi.
- Vai a Cloud Run > Servizi.
- Fai clic su
netapp-bluexp-collector. - Fai clic sulla scheda Log.
Verifica che la funzione sia stata eseguita correttamente. Cerca:
Fetching audit logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00 Successfully obtained NetApp Console access token Page 1: Retrieved X audit records Wrote X records to gs://netapp-bluexp-audit-logs/netapp-bluexp-audit/netapp_bluexp_audit_YYYYMMDD_HHMMSS.ndjson Successfully processed X audit recordsVai a Cloud Storage > Bucket.
Fai clic su
netapp-bluexp-audit-logs.Vai alla cartella
netapp-bluexp-audit/.Verifica che sia stato creato un nuovo file
.ndjsoncon il timestamp corrente.
Se visualizzi errori nei log:
- HTTP 401: verifica che le variabili di ambiente
NETAPP_CLIENT_IDeNETAPP_CLIENT_SECRETsiano corrette - HTTP 403: verifica che il account di servizio abbia il ruolo Organization viewer nella console NetApp
- HTTP 429: limitazione della frequenza: la funzione verrà ritentata automaticamente con backoff esponenziale
- Variabili di ambiente mancanti: verifica che tutte le variabili richieste siano impostate nella configurazione di Cloud Run Functions
Recuperare il account di servizio Google SecOps
- 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,
NetApp Console Audit Logs). - Seleziona Google Cloud Storage V2 come Tipo di origine.
- Seleziona NetApp Console (formerly BlueXP) come Log type.
Fai clic su Ottieni service account. Verrà visualizzato un indirizzo email del account di servizio univoco, ad esempio:
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.comCopia questo indirizzo email per utilizzarlo nel passaggio successivo.
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://netapp-bluexp-audit-logs/netapp-bluexp-audit/
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 i file trasferiti e le 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.
Concedi le autorizzazioni IAM al account di servizio Google SecOps
- Vai a Cloud Storage > Bucket.
- Fai clic su
netapp-bluexp-audit-logs. - Vai alla scheda Autorizzazioni.
- Fai clic su Concedi l'accesso.
- Fornisci i seguenti dettagli di configurazione:
- Aggiungi entità: incolla l'email del account di servizio Google SecOps
- Assegna i ruoli: seleziona Storage Object Viewer
Fai clic su Salva.
Tabella di mappatura UDM
| Campo log | Mappatura UDM | Logica |
|---|---|---|
| endTime | additional.fields | Unite alle etichette di ciascuna se non vuote |
| responseData | additional.fields | |
| weHash | additional.fields | |
| wePublicId | additional.fields | |
| hasFailedRecord | additional.fields | |
| azione | additional.fields | |
| servizio | additional.fields | |
| hasRecords | additional.fields | |
| workspaceId | additional.fields | |
| lastModified | additional.fields | |
| startTime | additional.fields | |
| agentName | metadata.event_type | Impostato su STATUS_UPDATE se agentName non è vuoto, altrimenti GENERIC_EVENT |
| agentId | metadata.product_deployment_id | Valore copiato direttamente se non vuoto |
| requestId | metadata.product_log_id | Valore copiato direttamente se non vuoto |
| rete | rete | Rinomina direttamente se non è vuoto |
| agentName | principal.asset.hostname | Valore copiato direttamente se non è vuoto |
| fileName | principal.file.names | Unito se non vuoto |
| agentName | principal.hostname | Valore copiato direttamente se non è vuoto |
| accountId | principal.resource.id | Valore copiato direttamente se non è vuoto |
| resourceName | principal.resource.name | Valore copiato direttamente se non vuoto |
| accountId | principal.resource.product_object_id | Valore di accountId se non è vuoto, altrimenti di resourceId se non è vuoto |
| resourceId | principal.resource.product_object_id | |
| principalId | principal.user.userid | Valore copiato direttamente se non è vuoto |
| stato | security_result.action | Imposta su ALLOW se status =~ (?i)SUCCESS, BLOCK se in FAILURE o UNSUCCESSFUL_ATTEMPT |
| stato | security_result.action_details | Valore copiato direttamente se non è vuoto |
| destinazione | destinazione | Rinomina direttamente se non è vuoto |
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.