Raccogliere i set di dati principali di Code42 Incydr

Supportato in:

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 tutti i dispositivi monitorando l'attività dei file in tempo reale su endpoint, servizi cloud ed email.

Prima di iniziare

Assicurati di soddisfare i 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

  1. Vai alla console Google 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 code42-incydr-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) Attiva il controllo delle versioni degli oggetti o la policy di conservazione

  6. Fai clic su Crea.

Raccogli le credenziali API di Code42 Incydr

Crea client API

  1. Accedi alla console web di Code42 Incydr.
  2. Vai ad Amministrazione > Integrazioni > Client API.
  3. Fai clic su Crea nuovo client API.
  4. Nella finestra di dialogo Crea nuovo client API, inserisci un nome per il client (ad esempio, Google Security Operations Integration).
  5. Copia e salva i seguenti dettagli in una posizione sicura:
    • ID client: l'identificatore del client API.
    • Secret: la chiave client secret API.
  6. 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.

  • Valori predefiniti comuni:

    URL console URL di base dell'API
    https://console.us.code42.com https://api.us.code42.com
    https://console.us2.code42.com https://api.us2.code42.com
    https://console.ie.code42.com https://api.ie.code42.com
    https://console.gov.code42.com https://api.gov.code42.com

Verifica le autorizzazioni del client API

Il client API deve disporre delle autorizzazioni appropriate per accedere agli endpoint richiesti:

  1. Nella console Code42 Incydr, vai ad Administration > Integrations > API Clients.
  2. Fai clic sul nome del client API che hai creato.

  3. 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
    • Casi: accesso in lettura ai dati dei casi
    • 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 account di servizio per la funzione Cloud Run

La funzione Cloud Run richiede un account di servizio con autorizzazioni per scrivere nel bucket GCS e per essere richiamato da Pub/Sub.

Crea service account

  1. Nella console Google Cloud, vai a IAM e amministrazione > Service Accounts.
  2. Fai clic su Crea account di servizio.
  3. 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.
  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 Amministratore oggetti di archiviazione.
    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 Invoker di Cloud Functions.
  6. Fai clic su Continua.
  7. 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 account di servizio le autorizzazioni di scrittura sul bucket GCS:

  1. Vai a Cloud Storage > Bucket.
  2. Fai clic sul nome del bucket (ad esempio code42-incydr-logs).
  3. Vai alla scheda Autorizzazioni.
  4. Fai clic su Concedi l'accesso.
  5. Fornisci i seguenti dettagli di configurazione:
    • Aggiungi entità: inserisci l'email del account di servizio (ad esempio, code42-incydr-collector-sa@your-project.iam.gserviceaccount.com).
    • Assegna i ruoli: seleziona Storage Object Admin.
  6. 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à.

  1. Nella console GCP, vai a Pub/Sub > Argomenti.
  2. Fai clic su Crea argomento.
  3. Fornisci i seguenti dettagli di configurazione:
    • ID argomento: inserisci code42-incydr-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 viene attivata dai messaggi Pub/Sub di Cloud Scheduler per recuperare i log dall'API Code42 Incydr e li scrive in GCS.

  1. Nella console GCP, vai a Cloud Run.
  2. Fai clic su Crea servizio.
  3. Seleziona Funzione (usa un editor in linea per creare una funzione).

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

    Impostazione Valore
    Nome servizio code42-incydr-collector
    Regione Seleziona la regione corrispondente al tuo bucket GCS (ad esempio, us-central1)
    Tempo di esecuzione Seleziona Python 3.12 o versioni successive

  5. Nella sezione Trigger (facoltativo):

    1. Fai clic su + Aggiungi trigger.
    2. Seleziona Cloud Pub/Sub.
    3. In Seleziona un argomento Cloud Pub/Sub, scegli l'argomento code42-incydr-trigger.
    4. Fai clic su Salva.

  6. Nella sezione Autenticazione:

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

  7. Scorri verso il basso ed espandi Container, networking, sicurezza.

  8. Vai alla scheda Sicurezza:

    • Service account: seleziona il account di servizio code42-incydr-collector-sa.

  9. 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
    INCYDR_BASE_URL https://api.us.code42.com URL di base dell'API del tuo tenant
    INCYDR_CLIENT_ID your-client-id ID client API
    INCYDR_CLIENT_SECRET your-client-secret API client secret
    GCS_BUCKET code42-incydr-logs Nome del bucket GCS
    GCS_PREFIX code42/ Prefisso per i file di log
    PAGE_SIZE 500 Record per pagina
    LOOKBACK_MINUTES 60 Periodo di riferimento iniziale
    STREAMS users,audit,cases Stream di dati separati da virgole
    FE_QUERY_JSON `` (Facoltativo) JSON della query degli eventi del file
    FE_PAGE_SIZE 1000 (Facoltativo) Dimensioni della pagina degli eventi dei file

  10. Scorri verso il basso nella scheda Variabili e secret fino a Richieste:

    • Timeout richiesta: inserisci 600 secondi (10 minuti).

  11. Vai alla scheda Impostazioni in Container:

    • Nella sezione Risorse:
      • Memoria: seleziona 1024 MiB o un valore superiore.
      • CPU: seleziona 1.
    • Fai clic su Fine.

  12. Scorri fino a Ambiente di esecuzione:

    • Seleziona Predefinito (opzione consigliata).

  13. Nella sezione Scalabilità della revisione:

    • Numero minimo di istanze: inserisci 0.
    • Numero massimo di istanze: inserisci 100 (o modifica in base al carico previsto).

  14. Fai clic su Crea.

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

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

Aggiungi codice per la funzione

  1. Inserisci main in Entry point della funzione

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

Crea job Cloud Scheduler

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

  1. Nella console di GCP, vai a Cloud Scheduler.
  2. Fai clic su Crea job.

  3. Fornisci i seguenti dettagli di configurazione:

    Impostazione Valore
    Nome code42-incydr-hourly
    Regione 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-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 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

  1. Nella console Cloud Scheduler, trova il tuo job (code42-incydr-hourly).
  2. Fai clic su Forza esecuzione per attivare manualmente.
  3. Attendi qualche secondo e vai a Cloud Run > Servizi > code42-incydr-collector > Log.

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

    Fetching users...
Fetching audit logs...
Fetching cases...
Successfully processed logs: {"users_pages": X, "audit_pages": Y, "cases_pages": Z}

  5. 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 di frequenza: la funzione riproverà automaticamente con backoff
  • Impossibile ottenere il token di accesso: verifica che INCYDR_BASE_URL, INCYDR_CLIENT_ID e INCYDR_CLIENT_SECRET siano corretti

Recuperare il account di servizio Google SecOps

Google SecOps utilizza un account di servizio univoco per leggere i dati dal tuo bucket GCS. 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.
  3. Fai clic su Configura un singolo feed.
  4. Nel campo Nome feed, inserisci un nome per il feed (ad esempio, Code42 Incydr Datasets).
  5. Seleziona Google Cloud Storage V2 come Tipo di origine.
  6. Seleziona Code42 Incydr come Tipo di log.

  7. Fai clic su Ottieni service account. Verrà visualizzata un'email univoca del account di servizio, ad esempio:

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

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

Concedi le autorizzazioni IAM al account di servizio Google SecOps

Il account di servizio Google SecOps deve avere il ruolo Visualizzatore oggetti Storage nel bucket GCS.

  1. Vai a Cloud Storage > Bucket.
  2. Fai clic sul nome del bucket (code42-incydr-logs).
  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 Visualizzatore oggetti Storage.

  6. Fai clic su Salva.

Configura un feed in Google SecOps per importare i log di Code42 Incydr

  1. Vai a Impostazioni SIEM > Feed.
  2. Fai clic su Aggiungi nuovo feed.
  3. Fai clic su Configura un singolo feed.
  4. Nel campo Nome feed, inserisci un nome per il feed (ad esempio, Code42 Incydr Datasets).
  5. Seleziona Google Cloud Storage V2 come Tipo di origine.
  6. Seleziona Code42 Incydr come Tipo di log.
  7. Fai clic su Avanti.

  8. Specifica i valori per i seguenti parametri di input:

    • URL 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/

    • Opzione di eliminazione della fonte: 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.

  9. Fai clic su Avanti.

  10. 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.