Raccogliere i set di dati principali di Code42 Incydr
Questo documento spiega come importare i set di dati principali di Code42 Incydr (Utenti, Audit, Casi e, facoltativamente, Eventi file) in Google Security Operations utilizzando Google Cloud Storage.
Code42 Incydr è una soluzione di gestione dei rischi interni che rileva, analizza e risponde all'esfiltrazione di dati su più dispositivi monitorando l'attività dei file in tempo reale su endpoint, servizi cloud ed email.
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 privilegiato all'API o alla Console di amministrazione di Code42 Incydr con il ruolo Amministratore rischi interni
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 code42-incydr-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 Code42 Incydr
Crea client API
- Accedi alla console web di Code42 Incydr.
- Vai ad Amministrazione > Integrazioni > Client API.
- Fai clic su Crea nuovo client API.
- Nella finestra di dialogo Crea nuovo client API, inserisci un nome per il client (ad esempio,
Google Security Operations Integration). - Copia e salva i seguenti dettagli in una posizione sicura:
- ID client: l'identificatore del client API.
- Secret: la chiave segreta del client API.
- Fai clic su Crea.
Determinare l'URL di base dell'API
L'URL di base dell'API dipende dall'URL della console Code42 Incydr. Verifica l'URL del gateway API nel portale per sviluppatori di Incydr o nella documentazione dell'ambiente del tuo tenant.
Impostazioni predefinite comuni:
URL della console URL di base dell'API https://console.us.code42.comhttps://api.us.code42.comhttps://console.us2.code42.comhttps://api.us2.code42.comhttps://console.ie.code42.comhttps://api.ie.code42.comhttps://console.gov.code42.comhttps://api.gov.code42.com
Verifica le autorizzazioni del client API
Il client API deve disporre delle autorizzazioni appropriate per accedere agli endpoint richiesti:
- Nella console Code42 Incydr, vai ad Administration > Integrations > API Clients.
- Fai clic sul nome del client API che hai creato.
Verifica che il client API abbia accesso ai seguenti ambiti:
- Utenti: accesso in lettura ai dati utente
- Audit log: accesso in lettura agli audit log
- Richieste: accesso in lettura ai dati delle richieste
- Eventi file (facoltativo): accesso in lettura ai dati sugli eventi dei file
Testare l'accesso API
Verifica le tue credenziali prima di procedere con l'integrazione:
# Replace with your actual credentials CLIENT_ID="your-client-id" CLIENT_SECRET="your-client-secret" API_BASE="https://api.us.code42.com" # Get OAuth token TOKEN=$(curl -s -X POST "${API_BASE}/v1/oauth/token" \ -u "${CLIENT_ID}:${CLIENT_SECRET}" \ -H "Content-Type: application/x-www-form-urlencoded" \ -d "grant_type=client_credentials" | jq -r '.access_token') # Test API access curl -v -H "Authorization: Bearer ${TOKEN}" "${API_BASE}/v1/users?pageSize=1"
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
code42-incydr-collector-sa. - Descrizione service account: inserisci
Service account for Cloud Run function to collect Code42 Incydr 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 (ad esempio
code42-incydr-logs). - 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.
code42-incydr-collector-sa@your-project.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
code42-incydr-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 Code42 Incydr 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 code42-incydr-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
code42-incydr-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
code42-incydr-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 Descrizione INCYDR_BASE_URLhttps://api.us.code42.comURL di base dell'API del tuo tenant INCYDR_CLIENT_IDyour-client-idID client API INCYDR_CLIENT_SECRETyour-client-secretAPI client secret GCS_BUCKETcode42-incydr-logsNome bucket GCS GCS_PREFIXcode42/Prefisso per i file di log PAGE_SIZE500Record per pagina LOOKBACK_MINUTES60Periodo di ricerca iniziale STREAMSusers,audit,casesStream di dati separati da virgole FE_QUERY_JSON`` (Facoltativo) JSON della query degli eventi del file FE_PAGE_SIZE1000(Facoltativo) Dimensioni della pagina degli eventi dei file Scorri 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 1024 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, timedelta, timezone import time # Initialize HTTP client http = urllib3.PoolManager( timeout=urllib3.Timeout(connect=5.0, read=30.0), retries=False, ) # Initialize Storage client storage_client = storage.Client() # Environment variables BASE = os.environ.get("INCYDR_BASE_URL", "").rstrip("/") CID = os.environ.get("INCYDR_CLIENT_ID", "") CSECRET = os.environ.get("INCYDR_CLIENT_SECRET", "") BUCKET = os.environ.get("GCS_BUCKET", "") PREFIX_BASE = os.environ.get("GCS_PREFIX", "code42/") PAGE_SIZE = int(os.environ.get("PAGE_SIZE", "500")) LOOKBACK_MINUTES = int(os.environ.get("LOOKBACK_MINUTES", "60")) STREAMS = [s.strip() for s in os.environ.get("STREAMS", "users").split(",") if s.strip()] FE_QUERY_JSON = os.environ.get("FE_QUERY_JSON", "").strip() FE_PAGE_SIZE = int(os.environ.get("FE_PAGE_SIZE", "1000")) def now_utc(): return datetime.now(timezone.utc) def iso_minus(minutes: int): return (now_utc() - timedelta(minutes=minutes)).strftime("%Y-%m-%dT%H:%M:%SZ") def put_json(bucket, prefix: str, page_label: str, data): ts = now_utc().strftime("%Y/%m/%d/%H%M%S") key = f"{PREFIX_BASE}{prefix}{ts}-{page_label}.json" blob = bucket.blob(key) blob.upload_from_string(json.dumps(data), content_type='application/json') return key def get_token(): """Get OAuth 2.0 access token using client credentials flow.""" token_url = f"{BASE}/v1/oauth/token" # Encode credentials import base64 credentials = f"{CID}:{CSECRET}" encoded_credentials = base64.b64encode(credentials.encode('utf-8')).decode('utf-8') headers = { 'Authorization': f'Basic {encoded_credentials}', 'Content-Type': 'application/x-www-form-urlencoded', 'Accept': 'application/json' } body = 'grant_type=client_credentials' backoff = 1.0 max_retries = 3 for attempt in range(max_retries): response = http.request('POST', token_url, body=body, headers=headers) if response.status == 429: retry_after = int(response.headers.get('Retry-After', str(int(backoff)))) print(f"Rate limited (429) on token request. Retrying after {retry_after}s...") time.sleep(retry_after) backoff = min(backoff * 2, 30.0) continue if response.status != 200: raise RuntimeError(f"Failed to get access token: {response.status} - {response.data.decode('utf-8')}") data = json.loads(response.data.decode('utf-8')) return data['access_token'] raise RuntimeError(f"Failed to get token after {max_retries} retries due to rate limiting") def auth_header(): token = get_token() return { "Authorization": f"Bearer {token}", "Accept": "application/json" } def http_get(path: str, params: dict = None, headers: dict = None): url = f"{BASE}{path}" if params: from urllib.parse import urlencode url += "?" + urlencode(params) backoff = 1.0 max_retries = 3 for attempt in range(max_retries): 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 return response.data raise RuntimeError(f"Failed after {max_retries} retries due to rate limiting") def http_post_json(path: str, body: dict, headers: dict = None): url = f"{BASE}{path}" backoff = 1.0 max_retries = 3 for attempt in range(max_retries): response = http.request( 'POST', url, body=json.dumps(body), headers={**headers, 'Content-Type': 'application/json'} ) 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 return response.data raise RuntimeError(f"Failed after {max_retries} retries due to rate limiting") # USERS (/v1/users) def pull_users(bucket, hdrs): next_token = None pages = 0 while True: params = {"active": "true", "blocked": "false", "pageSize": PAGE_SIZE} if next_token: params["pageToken"] = next_token raw = http_get("/v1/users", params, hdrs) data = json.loads(raw.decode('utf-8')) put_json(bucket, "users/", f"users-page-{pages}", data) pages += 1 next_token = data.get("nextPageToken") or data.get("next_page_token") if not next_token: break return pages # AUDIT LOG (/v1/audit/log) def pull_audit(bucket, hdrs): start_iso = iso_minus(LOOKBACK_MINUTES) next_token = None pages = 0 while True: params = {"startTime": start_iso, "pageSize": PAGE_SIZE} if next_token: params["pageToken"] = next_token raw = http_get("/v1/audit/log", params, hdrs) try: data = json.loads(raw.decode('utf-8')) put_json(bucket, "audit/", f"audit-page-{pages}", data) next_token = data.get("nextPageToken") or data.get("next_page_token") pages += 1 if not next_token: break except Exception as e: print(f"Error parsing audit log response: {e}") # Save raw response ts = now_utc().strftime("%Y/%m/%d/%H%M%S") key = f"{PREFIX_BASE}audit/{ts}-audit-export.bin" blob = bucket.blob(key) blob.upload_from_string(raw, content_type='application/octet-stream') pages += 1 break return pages # CASES (/v1/cases) def pull_cases(bucket, hdrs): next_token = None pages = 0 while True: params = {"pageSize": PAGE_SIZE} if next_token: params["pageToken"] = next_token raw = http_get("/v1/cases", params, hdrs) data = json.loads(raw.decode('utf-8')) put_json(bucket, "cases/", f"cases-page-{pages}", data) pages += 1 next_token = data.get("nextPageToken") or data.get("next_page_token") if not next_token: break return pages # FILE EVENTS (/v2/file-events/search) def pull_file_events(bucket, hdrs): if not FE_QUERY_JSON: return 0 try: base_query = json.loads(FE_QUERY_JSON) except Exception as e: print(f"Error: FE_QUERY_JSON is not valid JSON: {e}") return 0 pages = 0 next_token = None while True: body = dict(base_query) body["pageSize"] = FE_PAGE_SIZE if next_token: body["pageToken"] = next_token raw = http_post_json("/v2/file-events/search", body, hdrs) data = json.loads(raw.decode('utf-8')) put_json(bucket, "file_events/", f"fileevents-page-{pages}", data) pages += 1 next_token = ( data.get("nextPageToken") or data.get("next_page_token") or (data.get("file_events") or {}).get("nextPageToken") ) if not next_token: break return pages @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch logs from Code42 Incydr API and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ if not all([BASE, CID, CSECRET, BUCKET]): print('Error: Missing required environment variables') return try: bucket = storage_client.bucket(BUCKET) hdrs = auth_header() report = {} if "users" in STREAMS: print("Fetching users...") report["users_pages"] = pull_users(bucket, hdrs) if "audit" in STREAMS: print("Fetching audit logs...") report["audit_pages"] = pull_audit(bucket, hdrs) if "cases" in STREAMS: print("Fetching cases...") report["cases_pages"] = pull_cases(bucket, hdrs) if "file_events" in STREAMS: print("Fetching file events...") report["file_events_pages"] = pull_file_events(bucket, hdrs) print(f'Successfully processed logs: {json.dumps(report)}') except Exception as e: print(f'Error processing logs: {str(e)}') raise- 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 code42-incydr-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 code42-incydr-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 tuo job (
code42-incydr-hourly). - Fai clic su Forza esecuzione per attivare manualmente.
- Attendi qualche secondo e vai a Cloud Run > Servizi > code42-incydr-collector > Log.
Verifica che la funzione sia stata eseguita correttamente. Ecco alcuni esempi da tenere in considerazione:
Fetching users... Fetching audit logs... Fetching cases... Successfully processed logs: {"users_pages": X, "audit_pages": Y, "cases_pages": Z}Controlla il bucket GCS (
code42-incydr-logs) per verificare che i log siano stati scritti.
Se visualizzi errori nei log:
- HTTP 401: controlla le credenziali API nelle variabili di ambiente
- HTTP 403: verifica che il client API disponga delle autorizzazioni richieste nella console Code42 Incydr
- HTTP 429: limitazione della frequenza: la funzione riproverà automaticamente con backoff
- Impossibile ottenere il token di accesso: verifica che
INCYDR_BASE_URL,INCYDR_CLIENT_IDeINCYDR_CLIENT_SECRETsiano corretti
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,
Code42 Incydr Datasets). - Seleziona Google Cloud Storage V2 come Tipo di origine.
- Seleziona Code42 Incydr 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 (
code42-incydr-logs). - 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 Code42 Incydr
- 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,
Code42 Incydr Datasets). - Seleziona Google Cloud Storage V2 come Tipo di origine.
- Seleziona Code42 Incydr 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://code42-incydr-logs/code42/Sostituisci:
code42-incydr-logs: il nome del bucket GCS.code42/: (Facoltativo) prefisso/percorso della cartella in cui vengono archiviati i log (lascia vuoto per la radice).
Esempi:
- Bucket radice:
gs://code42-incydr-logs/ - Con prefisso:
gs://code42-incydr-logs/code42/
- 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.
Hai bisogno di ulteriore assistenza? Ricevi risposte dai membri della community e dai professionisti di Google SecOps.