Raccogliere i log di Zendesk CRM

Supportato in:

Questo documento spiega come importare i log di Zendesk Customer Relationship Management (CRM) in Google Security Operations utilizzando Google Cloud Storage. Zendesk CRM offre funzionalità di assistenza clienti e gestione dei ticket. La piattaforma monitora le interazioni con i clienti, i ticket di assistenza e le attività amministrative tramite gli audit log e i dati dei ticket.

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 funzioni Cloud Run, argomenti Pub/Sub e job Cloud Scheduler
  • Accesso con privilegi a Zendesk (ruolo Amministratore richiesto per la creazione del token API)
  • Piano Zendesk Enterprise (richiesto per l'accesso all'API Audit Logs)

Requisiti di Zendesk

Conferma il piano e il ruolo

Per creare token API o client OAuth, devi essere un amministratore Zendesk. L'API Audit Logs è disponibile solo nel piano Enterprise e restituisce un massimo di 100 record per pagina. Se il tuo account non è Enterprise, puoi comunque raccogliere i dati incrementali dei biglietti.

Attivare l'accesso al token API (una tantum)

  1. Nel Centro amministrativo, vai ad App e integrazioni > API > API Zendesk.
  2. Nella scheda Settings (Impostazioni), attiva Token Access (Accesso tramite token).

Genera un token API (per l'autenticazione di base)

  1. Vai ad App e integrazioni > API > API Zendesk.
  2. Fai clic sul pulsante Aggiungi token API.
  3. (Facoltativo) Aggiungi una descrizione del token API.
  4. Fai clic su Crea.
  5. Copia e salva subito il token API (non potrai più visualizzarlo).
  6. Salva l'email dell'amministratore che verrà autenticata con questo token.

(Facoltativo) Crea un client OAuth (per l'autenticazione Bearer anziché il token API)

  1. Vai ad App e integrazioni > API > API Zendesk.
  2. Fai clic sulla scheda Client OAuth.
  3. Fai clic su Aggiungi client OAuth.
  4. Compila i campi Nome client, Identificatore univoco (automatico) e URL di reindirizzamento (può essere un segnaposto se generi token solo con l'API).
  5. Fai clic su Salva.
  6. Crea un token di accesso per l'integrazione e concedi gli ambiti minimi richiesti da questa guida:
    • tickets:read (per i biglietti incrementali)
    • auditlogs:read (per gli audit log; solo Enterprise)
  7. Copia il token di accesso (incollalo nella variabile di ambiente ZENDESK_BEARER_TOKEN) e registra in modo sicuro l'ID client/il client secret (per i futuri flussi di aggiornamento del token).

Registrare l'URL di base di Zendesk

Utilizza https://<your_subdomain>.zendesk.com (incolla nella variabile di ambiente ZENDESK_BASE_URL).

Cosa salvare per un utilizzo futuro

  • URL di base (ad esempio, https://acme.zendesk.com)
  • Indirizzo email dell'utente amministratore (per l'autenticazione con token API)
  • Token API (se utilizzi AUTH_MODE=token) o token di accesso OAuth (se utilizzi AUTH_MODE=bearer)
  • (Facoltativo) ID client/secret OAuth per la gestione del ciclo di vita

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 zendesk-crm-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
  6. Fai clic su Crea.

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

  1. Nella console Google Cloud, vai a IAM e amministrazione > Service Accounts.
  2. Fai clic su Crea service account.
  3. Fornisci i seguenti dettagli di configurazione:
    • Nome del service account: inserisci zendesk-crm-collector-sa.
    • Descrizione service account: inserisci Service account for Cloud Run function to collect Zendesk CRM logs.
  4. Fai clic su Crea e continua.
  5. Nella sezione Concedi a questo service account 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 service account le autorizzazioni di scrittura sul bucket GCS:

  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à: inserisci l'email del service account (ad es. zendesk-crm-collector-sa@PROJECT_ID.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 zendesk-crm-trigger.
    • Lascia le altre impostazioni sui valori predefiniti.
  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 Zendesk e scriverli 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 zendesk-crm-collector
    Regione Seleziona la regione corrispondente al tuo bucket GCS (ad esempio us-central1)
    Runtime 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 zendesk-crm-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 service account zendesk-crm-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
    GCS_BUCKET zendesk-crm-logs Nome bucket GCS
    GCS_PREFIX zendesk/crm/ Prefisso per i file di log
    STATE_KEY zendesk/crm/state.json Percorso file di stato
    ZENDESK_BASE_URL https://your_subdomain.zendesk.com URL di base Zendesk
    AUTH_MODE token Modalità di autenticazione (token o bearer)
    ZENDESK_EMAIL analyst@example.com Email dell'amministratore per l'autenticazione del token API
    ZENDESK_API_TOKEN <api_token> Token API per l'autenticazione
    ZENDESK_BEARER_TOKEN <leave empty unless using OAuth bearer> Token di connessione OAuth (facoltativo)
    RESOURCES audit_logs,incremental_tickets Risorse da raccogliere
    MAX_PAGES 20 Numero massimo di pagine per esecuzione
    LOOKBACK_SECONDS 3600 Periodo di ricerca iniziale
    HTTP_TIMEOUT 60 Timeout richiesta HTTP
    HTTP_RETRIES 3 Tentativi di ripetizione HTTP
  10. Nella sezione Variabili e secret, scorri verso il basso fino a Richieste:

    • Timeout richiesta: inserisci 600 secondi (10 minuti).
  11. Vai alla scheda Impostazioni:

    • Nella sezione Risorse:
      • Memoria: seleziona 512 MiB o un valore superiore.
      • CPU: seleziona 1.
  12. Nella sezione Scalabilità della revisione:

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

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

  15. 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, timezone
    import base64
    import time
    
    # Initialize HTTP client with timeouts
    http = urllib3.PoolManager(
        timeout=urllib3.Timeout(connect=5.0, read=30.0),
        retries=False,
    )
    
    # 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 Zendesk 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', 'zendesk/crm/')
        state_key = os.environ.get('STATE_KEY', 'zendesk/crm/state.json')
    
        base_url = os.environ.get('ZENDESK_BASE_URL', '').rstrip('/')
        auth_mode = os.environ.get('AUTH_MODE', 'token').lower()
        email = os.environ.get('ZENDESK_EMAIL', '')
        api_token = os.environ.get('ZENDESK_API_TOKEN', '')
        bearer = os.environ.get('ZENDESK_BEARER_TOKEN', '')
    
        resources = [r.strip() for r in os.environ.get('RESOURCES', 'audit_logs,incremental_tickets').split(',') if r.strip()]
        max_pages = int(os.environ.get('MAX_PAGES', '20'))
        lookback = int(os.environ.get('LOOKBACK_SECONDS', '3600'))
        http_timeout = int(os.environ.get('HTTP_TIMEOUT', '60'))
        http_retries = int(os.environ.get('HTTP_RETRIES', '3'))
    
        if not all([bucket_name, base_url]):
            print('Error: Missing required environment variables')
            return
    
        try:
            # Get GCS bucket
            bucket = storage_client.bucket(bucket_name)
    
            # Load state
            state = load_state(bucket, state_key)
    
            print(f'Processing resources: {resources}')
    
            summary = []
    
            if 'audit_logs' in resources:
                res = fetch_audit_logs(
                    bucket, prefix, state.get('audit_logs', {}),
                    base_url, auth_mode, email, api_token, bearer,
                    max_pages, http_timeout, http_retries
                )
                state['audit_logs'] = {'next_url': res.get('next_url')}
                summary.append(res)
    
            if 'incremental_tickets' in resources:
                res = fetch_incremental_tickets(
                    bucket, prefix, state.get('incremental_tickets', {}),
                    base_url, auth_mode, email, api_token, bearer,
                    max_pages, lookback, http_timeout, http_retries
                )
                state['incremental_tickets'] = {'cursor': res.get('cursor')}
                summary.append(res)
    
            # Save state
            save_state(bucket, state_key, state)
    
            print(f'Successfully processed logs: {summary}')
    
        except Exception as e:
            print(f'Error processing logs: {str(e)}')
            raise
    
    def get_headers(auth_mode, email, api_token, bearer):
        """Get authentication headers."""
        if auth_mode == 'bearer' and bearer:
            return {
                'Authorization': f'Bearer {bearer}',
                'Accept': 'application/json'
            }
        if auth_mode == 'token' and email and api_token:
            auth_string = f'{email}/token:{api_token}'
            auth_bytes = auth_string.encode('utf-8')
            token = base64.b64encode(auth_bytes).decode('utf-8')
            return {
                'Authorization': f'Basic {token}',
                'Accept': 'application/json'
            }
        raise RuntimeError('Invalid auth settings: provide token (EMAIL + API_TOKEN) or BEARER')
    
    def http_get_json(url, headers, timeout, retries):
        """Make HTTP GET request with retries and exponential backoff."""
        attempt = 0
        backoff = 1.0
        while True:
            try:
                response = http.request('GET', url, headers=headers, timeout=timeout)
                if response.status == 200:
                    return json.loads(response.data.decode('utf-8'))
                elif response.status in (429, 500, 502, 503, 504) and attempt < retries:
                    retry_after = int(response.headers.get('Retry-After', int(backoff)))
                    print(f'HTTP {response.status}: Retrying after {retry_after}s (attempt {attempt + 1}/{retries})')
                    time.sleep(max(1, retry_after))
                    backoff = min(backoff * 2, 30.0)
                    attempt += 1
                    continue
                else:
                    raise Exception(f'HTTP {response.status}: {response.data.decode("utf-8")}')
            except Exception as e:
                if attempt < retries:
                    print(f'Request error: {e}. Retrying after {int(backoff)}s (attempt {attempt + 1}/{retries})')
                    time.sleep(backoff)
                    backoff = min(backoff * 2, 30.0)
                    attempt += 1
                    continue
                raise
    
    def put_page(bucket, prefix, payload, resource):
        """Write page to GCS."""
        ts = datetime.now(timezone.utc)
        key = f'{prefix}{ts.strftime("%Y/%m/%d/%H%M%S")}-zendesk-{resource}.json'
        blob = bucket.blob(key)
        blob.upload_from_string(
            json.dumps(payload),
            content_type='application/json'
        )
        return key
    
    def fetch_audit_logs(bucket, prefix, state, base_url, auth_mode, email, api_token, bearer, max_pages, timeout, retries):
        """Fetch audit logs with pagination."""
        headers = get_headers(auth_mode, email, api_token, bearer)
        next_url = state.get('next_url') or f'{base_url}/api/v2/audit_logs.json'
    
        pages = 0
        written = 0
        last_next = None
    
        while pages < max_pages and next_url:
            data = http_get_json(next_url, headers, timeout, retries)
            put_page(bucket, prefix, data, 'audit_logs')
            written += len(data.get('audit_logs', []))
    
            # Use next_page for pagination
            last_next = data.get('next_page')
            next_url = last_next
            pages += 1
    
            print(f'Audit logs page {pages}: Retrieved {len(data.get("audit_logs", []))} records')
    
        return {
            'resource': 'audit_logs',
            'pages': pages,
            'written': written,
            'next_url': last_next
        }
    
    def fetch_incremental_tickets(bucket, prefix, state, base_url, auth_mode, email, api_token, bearer, max_pages, lookback, timeout, retries):
        """Fetch incremental tickets with cursor-based pagination."""
        headers = get_headers(auth_mode, email, api_token, bearer)
        cursor = state.get('cursor')
    
        if not cursor:
            start = int(time.time()) - lookback
            next_url = f'{base_url}/api/v2/incremental/tickets/cursor.json?start_time={start}'
        else:
            next_url = f'{base_url}/api/v2/incremental/tickets/cursor.json?cursor={cursor}'
    
        pages = 0
        written = 0
        last_cursor = None
    
        while pages < max_pages and next_url:
            data = http_get_json(next_url, headers, timeout, retries)
            put_page(bucket, prefix, data, 'incremental_tickets')
            written += len(data.get('tickets', []))
    
            # Extract cursor from after_cursor field
            last_cursor = data.get('after_cursor')
            if last_cursor:
                next_url = f'{base_url}/api/v2/incremental/tickets/cursor.json?cursor={last_cursor}'
            else:
                next_url = None
    
            pages += 1
    
            print(f'Incremental tickets page {pages}: Retrieved {len(data.get("tickets", []))} records')
    
        return {
            'resource': 'incremental_tickets',
            'pages': pages,
            'written': written,
            'cursor': last_cursor
        }
    
    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 {'audit_logs': {}, 'incremental_tickets': {}}
    
    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)}')
    
    • 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 zendesk-crm-collector-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 zendesk-crm-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

Testare l'integrazione

  1. Nella console Cloud Scheduler, trova il job.
  2. Fai clic su Forza esecuzione per attivare il job manualmente.
  3. Aspetta alcuni secondi.
  4. Vai a Cloud Run > Servizi.
  5. Fai clic sul nome della funzione zendesk-crm-collector.
  6. Fai clic sulla scheda Log.
  7. Verifica che la funzione sia stata eseguita correttamente. Cerca quanto segue:

    Processing resources: ['audit_logs', 'incremental_tickets']
    Audit logs page 1: Retrieved X records
    Incremental tickets page 1: Retrieved X records
    Successfully processed logs: [...]
    
  8. Vai a Cloud Storage > Bucket.

  9. Fai clic sul nome del bucket.

  10. Vai alla cartella del prefisso zendesk/crm/.

  11. Verifica che siano stati creati nuovi file .json con il timestamp corrente.

Se visualizzi errori nei log:

  • HTTP 401: controlla le credenziali API nelle variabili di ambiente
  • HTTP 403: verifica che l'account disponga delle autorizzazioni richieste (ruolo Amministratore, piano Enterprise per i log di controllo)
  • HTTP 429: limitazione della frequenza: la funzione riproverà automaticamente con backoff esponenziale
  • Variabili di ambiente mancanti: controlla che tutte le variabili richieste siano impostate

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

  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, Zendesk CRM logs).
  5. Seleziona Google Cloud Storage V2 come Tipo di origine.
  6. Seleziona Zendesk CRM come Tipo di log.
  7. Fai clic su Ottieni service account. Viene visualizzata un'email del service account univoca, 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 service account Google SecOps

Il service account 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.
  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 service account Google SecOps.
    • Assegna i ruoli: seleziona Visualizzatore oggetti Storage.
  6. Fai clic su Salva.

Configura un feed in Google SecOps per importare i log di Zendesk CRM

  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, Zendesk CRM logs).
  5. Seleziona Google Cloud Storage V2 come Tipo di origine.
  6. Seleziona Zendesk CRM come Tipo di log.
  7. Fai clic su Avanti.
  8. 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://zendesk-crm-logs/zendesk/crm/
      
      • Sostituisci:

        • zendesk-crm-logs: il nome del bucket GCS.
        • zendesk/crm/: il percorso del prefisso/della cartella in cui sono archiviati i log.
    • 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.

  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.