Coletar registros do CRM do Zendesk

Compatível com:

Este documento explica como ingerir registros do gerenciamento de relacionamento com o cliente (CRM) do Zendesk no Google Security Operations usando o Google Cloud Storage. O CRM do Zendesk oferece recursos de suporte ao cliente e gerenciamento de tíquetes. A plataforma rastreia interações com clientes, tíquetes de suporte e atividades administrativas usando registros de auditoria e dados de tíquetes.

Antes de começar

Verifique se você atende os seguintes pré-requisitos:

  • Uma instância do Google SecOps
  • Um projeto do GCP com a API Cloud Storage ativada
  • Permissões para criar e gerenciar buckets do GCS
  • Permissões para gerenciar políticas do IAM em buckets do GCS
  • Permissões para criar funções do Cloud Run, tópicos do Pub/Sub e jobs do Cloud Scheduler
  • Acesso privilegiado ao Zendesk (é necessário ter a função de administrador para criar um token de API)
  • Plano Enterprise do Zendesk (necessário para acesso à API de registros de auditoria)

Conferir os pré-requisitos do Zendesk

Confirmar plano e função

Você precisa ser um administrador do Zendesk para criar tokens de API ou clientes OAuth. A API Audit Logs está disponível apenas no plano Enterprise e retorna no máximo 100 registros por página. Se a sua conta não for Enterprise, você ainda poderá coletar dados incrementais de tíquetes.

Ativar o acesso ao token da API (uma única vez)

  1. Na Central de administração, acesse Apps e integrações > APIs > API Zendesk.
  2. Na guia Configurações, ative o Acesso por token.

Gerar um token de API (para autenticação básica)

  1. Acesse Apps e integrações > APIs > API do Zendesk.
  2. Clique no botão Adicionar token de API.
  3. Se quiser, adicione uma descrição do token da API.
  4. Clique em Criar.
  5. Copie e salve o token da API agora. Não será possível vê-lo novamente.
  6. Salve o e-mail do administrador que será autenticado com esse token.

(Opcional) Crie um cliente OAuth (para autenticação de portador em vez de token de API)

  1. Acesse Apps e integrações > APIs > API do Zendesk.
  2. Clique na guia Clientes OAuth.
  3. Clique em Adicionar cliente OAuth.
  4. Preencha o Nome do cliente, o Identificador exclusivo (automático) e os URLs de redirecionamento (podem ser um marcador de posição se você só gerar tokens com a API).
  5. Clique em Salvar.
  6. Crie um token de acesso para a integração e conceda os escopos mínimos exigidos por este guia:
    • tickets:read (para ingressos incrementais)
    • auditlogs:read (para registros de auditoria; somente Enterprise)
  7. Copie o token de acesso (cole na variável de ambiente ZENDESK_BEARER_TOKEN) e registre o ID/segredo do cliente com segurança (para fluxos futuros de atualização de token).

Registre seu URL de base do Zendesk

Use https://<your_subdomain>.zendesk.com (cole na variável de ambiente ZENDESK_BASE_URL).

O que salvar para mais tarde

  • URL de base (por exemplo, https://acme.zendesk.com)
  • Endereço de e-mail do usuário administrador (para autenticação de token da API)
  • Token da API (se estiver usando AUTH_MODE=token) ou token de acesso OAuth (se estiver usando AUTH_MODE=bearer)
  • (Opcional): ID/chave secreta do cliente OAuth para gerenciamento do ciclo de vida

Criar um bucket do Google Cloud Storage

  1. Acesse o Console do Google Cloud.
  2. Selecione seu projeto ou crie um novo.
  3. No menu de navegação, acesse Cloud Storage > Buckets.
  4. Clique em Criar bucket.
  5. Informe os seguintes detalhes de configuração:

    Configuração Valor
    Nomeie seu bucket Insira um nome exclusivo globalmente, por exemplo, zendesk-crm-logs.
    Tipo de local Escolha com base nas suas necessidades (região, birregional, multirregional)
    Local Selecione o local (por exemplo, us-central1).
    Classe de armazenamento Padrão (recomendado para registros acessados com frequência)
    Controle de acesso Uniforme (recomendado)
    Ferramentas de proteção Opcional: ativar o controle de versões de objetos ou a política de retenção
  6. Clique em Criar.

Criar uma conta de serviço para a função do Cloud Run

A função do Cloud Run precisa de uma conta de serviço com permissões para gravar no bucket do GCS e ser invocada pelo Pub/Sub.

Criar conta de serviço

  1. No Console do GCP, acesse IAM e administrador > Contas de serviço.
  2. Clique em Criar conta de serviço.
  3. Informe os seguintes detalhes de configuração:
    • Nome da conta de serviço: insira zendesk-crm-collector-sa.
    • Descrição da conta de serviço: insira Service account for Cloud Run function to collect Zendesk CRM logs.
  4. Clique em Criar e continuar.
  5. Na seção Conceder acesso a essa conta de serviço ao projeto, adicione os seguintes papéis:
    1. Clique em Selecionar papel.
    2. Pesquise e selecione Administrador de objetos do Storage.
    3. Clique em + Adicionar outro papel.
    4. Pesquise e selecione Invocador do Cloud Run.
    5. Clique em + Adicionar outro papel.
    6. Pesquise e selecione Invocador do Cloud Functions.
  6. Clique em Continuar.
  7. Clique em Concluído.

Esses papéis são necessários para:

  • Administrador de objetos do Storage: grava registros em um bucket do GCS e gerencia arquivos de estado.
  • Invocador do Cloud Run: permite que o Pub/Sub invoque a função
  • Invocador do Cloud Functions: permite a invocação de funções

Conceder permissões do IAM no bucket do GCS

Conceda permissões de gravação à conta de serviço no bucket do GCS:

  1. Acesse Cloud Storage > Buckets.
  2. Clique no nome do bucket.
  3. Acesse a guia Permissões.
  4. Clique em Conceder acesso.
  5. Informe os seguintes detalhes de configuração:
    • Adicionar principais: insira o e-mail da conta de serviço (por exemplo, zendesk-crm-collector-sa@PROJECT_ID.iam.gserviceaccount.com).
    • Atribuir papéis: selecione Administrador de objetos do Storage.
  6. Clique em Salvar.

Criar tópico Pub/Sub

Crie um tópico do Pub/Sub em que o Cloud Scheduler vai publicar e a função do Cloud Run vai se inscrever.

  1. No Console do GCP, acesse Pub/Sub > Tópicos.
  2. Selecione Criar tópico.
  3. Informe os seguintes detalhes de configuração:
    • ID do tópico: insira zendesk-crm-trigger.
    • Não altere as outras configurações.
  4. Clique em Criar.

Criar uma função do Cloud Run para coletar registros

A função do Cloud Run é acionada por mensagens do Pub/Sub do Cloud Scheduler para buscar registros da API Zendesk e gravá-los no GCS.

  1. No console do GCP, acesse o Cloud Run.
  2. Clique em Criar serviço.
  3. Selecione Função (use um editor in-line para criar uma função).
  4. Na seção Configurar, forneça os seguintes detalhes de configuração:

    Configuração Valor
    Nome do serviço zendesk-crm-collector
    Região Selecione a região que corresponde ao seu bucket do GCS (por exemplo, us-central1).
    Ambiente de execução Selecione Python 3.12 ou uma versão mais recente.
  5. Na seção Acionador (opcional):

    1. Clique em + Adicionar gatilho.
    2. Selecione Cloud Pub/Sub.
    3. Em Selecionar um tópico do Cloud Pub/Sub, escolha o tópico zendesk-crm-trigger.
    4. Clique em Salvar.
  6. Na seção Autenticação:

    1. Selecione Exigir autenticação.
    2. Confira o Identity and Access Management (IAM).
  7. Role a tela para baixo e abra Contêineres, rede, segurança.

  8. Acesse a guia Segurança:

    • Conta de serviço: selecione a conta de serviço zendesk-crm-collector-sa.
  9. Acesse a guia Contêineres:

    1. Clique em Variáveis e secrets.
    2. Clique em + Adicionar variável para cada variável de ambiente:
    Nome da variável Valor de exemplo Descrição
    GCS_BUCKET zendesk-crm-logs Nome do bucket do GCS
    GCS_PREFIX zendesk/crm/ Prefixo para arquivos de registro
    STATE_KEY zendesk/crm/state.json Caminho do arquivo de estado
    ZENDESK_BASE_URL https://your_subdomain.zendesk.com URL base do Zendesk
    AUTH_MODE token Modo de autenticação (token ou bearer)
    ZENDESK_EMAIL analyst@example.com E-mail do administrador para autenticação de token da API
    ZENDESK_API_TOKEN <api_token> Token de API para autenticação
    ZENDESK_BEARER_TOKEN <leave empty unless using OAuth bearer> Token de portador do OAuth (opcional)
    RESOURCES audit_logs,incremental_tickets Recursos a serem coletados
    MAX_PAGES 20 Número máximo de páginas por execução
    LOOKBACK_SECONDS 3600 Período de lookback inicial
    HTTP_TIMEOUT 60 Tempo limite da solicitação HTTP
    HTTP_RETRIES 3 Tentativas de repetição HTTP
  10. Na seção Variáveis e secrets, role a tela para baixo até Solicitações:

    • Tempo limite da solicitação: insira 600 segundos (10 minutos).
  11. Acesse a guia Configurações:

    • Na seção Recursos:
      • Memória: selecione 512 MiB ou mais.
      • CPU: selecione 1.
  12. Na seção Escalonamento de revisão:

    • Número mínimo de instâncias: insira 0.
    • Número máximo de instâncias: insira 100 ou ajuste com base na carga esperada.
  13. Clique em Criar.

  14. Aguarde a criação do serviço (1 a 2 minutos).

  15. Depois que o serviço é criado, o editor de código inline é aberto automaticamente.

Adicionar código da função

  1. Insira main em Ponto de entrada da função.
  2. No editor de código em linha, crie dois arquivos:

    • Primeiro arquivo: 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)}')
    
    • Segundo arquivo: requirements.txt:
    functions-framework==3.*
    google-cloud-storage==2.*
    urllib3>=2.0.0
    
  3. Clique em Implantar para salvar e implantar a função.

  4. Aguarde a conclusão da implantação (2 a 3 minutos).

Criar o job do Cloud Scheduler

O Cloud Scheduler publica mensagens no tópico do Pub/Sub em intervalos regulares, acionando a função do Cloud Run.

  1. No Console do GCP, acesse o Cloud Scheduler.
  2. Clique em Criar job.
  3. Informe os seguintes detalhes de configuração:

    Configuração Valor
    Nome zendesk-crm-collector-hourly
    Região Selecione a mesma região da função do Cloud Run
    Frequência 0 * * * * (a cada hora, na hora)
    Fuso horário Selecione o fuso horário (UTC recomendado)
    Tipo de destino Pub/Sub
    Tópico Selecione o tópico zendesk-crm-trigger.
    Corpo da mensagem {} (objeto JSON vazio)
  4. Clique em Criar.

Opções de frequência de programação

  • Escolha a frequência com base no volume de registros e nos requisitos de latência:

    Frequência Expressão Cron Caso de uso
    A cada 5 minutos */5 * * * * Alto volume e baixa latência
    A cada 15 minutos */15 * * * * Volume médio
    A cada hora 0 * * * * Padrão (recomendado)
    A cada 6 horas 0 */6 * * * Baixo volume, processamento em lote
    Diário 0 0 * * * Coleta de dados históricos

Testar a integração

  1. No console do Cloud Scheduler, encontre seu job.
  2. Clique em Executar à força para acionar o job manualmente.
  3. Aguarde alguns segundos.
  4. Acesse Cloud Run > Serviços.
  5. Clique no nome da função zendesk-crm-collector.
  6. Clique na guia Registros.
  7. Verifique se a função foi executada com sucesso. Procure o seguinte:

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

  9. Clique no nome do bucket.

  10. Navegue até a pasta de prefixo zendesk/crm/.

  11. Verifique se os novos arquivos .json foram criados com o carimbo de data/hora atual.

Se você encontrar erros nos registros:

  • HTTP 401: verifique as credenciais da API nas variáveis de ambiente
  • HTTP 403: verifique se a conta tem as permissões necessárias (função de administrador, plano Enterprise para registros de auditoria).
  • HTTP 429: limitação de taxa. A função vai tentar de novo automaticamente com espera exponencial.
  • Variáveis de ambiente ausentes: verifique se todas as variáveis necessárias estão definidas.

Recuperar a conta de serviço do Google SecOps

O Google SecOps usa uma conta de serviço exclusiva para ler dados do seu bucket do GCS. Você precisa conceder a essa conta de serviço acesso ao seu bucket.

Receber o e-mail da conta de serviço

  1. Acesse Configurações do SIEM > Feeds.
  2. Clique em Adicionar novo feed.
  3. Clique em Configurar um único feed.
  4. No campo Nome do feed, insira um nome para o feed (por exemplo, Zendesk CRM logs).
  5. Selecione Google Cloud Storage V2 como o Tipo de origem.
  6. Selecione Zendesk CRM como o Tipo de registro.
  7. Clique em Receber conta de serviço. Um e-mail exclusivo da conta de serviço é exibido, por exemplo:

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
    
  8. Copie esse endereço de e-mail para usar na próxima etapa.

Conceder permissões do IAM à conta de serviço do Google SecOps

A conta de serviço do Google SecOps precisa do papel de Leitor de objetos do Storage no seu bucket do GCS.

  1. Acesse Cloud Storage > Buckets.
  2. Clique no nome do bucket.
  3. Acesse a guia Permissões.
  4. Clique em Conceder acesso.
  5. Informe os seguintes detalhes de configuração:
    • Adicionar participantes: cole o e-mail da conta de serviço do Google SecOps.
    • Atribuir papéis: selecione Leitor de objetos do Storage.
  6. Clique em Salvar.

Configurar um feed no Google SecOps para ingerir registros do CRM do Zendesk

  1. Acesse Configurações do SIEM > Feeds.
  2. Clique em Adicionar novo feed.
  3. Clique em Configurar um único feed.
  4. No campo Nome do feed, insira um nome para o feed (por exemplo, Zendesk CRM logs).
  5. Selecione Google Cloud Storage V2 como o Tipo de origem.
  6. Selecione Zendesk CRM como o Tipo de registro.
  7. Clique em Próxima.
  8. Especifique valores para os seguintes parâmetros de entrada:

    • URL do bucket de armazenamento: insira o URI do bucket do GCS com o caminho do prefixo:

      gs://zendesk-crm-logs/zendesk/crm/
      
      • Substitua:

        • zendesk-crm-logs: o nome do bucket do GCS.
        • zendesk/crm/: prefixo/caminho da pasta onde os registros são armazenados.
    • Opção de exclusão da fonte: selecione a opção de exclusão de acordo com sua preferência:

      • Nunca: nunca exclui arquivos após as transferências (recomendado para testes).
      • Excluir arquivos transferidos: exclui os arquivos após a transferência bem-sucedida.
      • Excluir arquivos transferidos e diretórios vazios: exclui arquivos e diretórios vazios após a transferência bem-sucedida.

    • Idade máxima do arquivo: inclui arquivos modificados no último número de dias. O padrão é de 180 dias.

    • Namespace do recurso: o namespace do recurso.

    • Rótulos de ingestão: o rótulo a ser aplicado aos eventos deste feed.

  9. Clique em Próxima.

  10. Revise a nova configuração do feed na tela Finalizar e clique em Enviar.

Precisa de mais ajuda? Receba respostas de membros da comunidade e profissionais do Google SecOps.