Coletar registros do CloudM

Versão do analisador:6.0

Compatível com:

Neste documento, explicamos como ingerir registros do CloudM no Google Security Operations usando o Google Cloud Storage V2.

O CloudM é uma plataforma SaaS para o Google Workspace e o Microsoft 365 que oferece automação de fluxo de trabalho para integração e desintegração de usuários, backup, arquivamento e migração de dados. O CloudM Automate gera um registro de auditoria completo de todas as ações realizadas no seu domínio, incluindo eventos de gerenciamento de usuários, etapas do fluxo de trabalho de desligamento, mudanças de configuração e operações relacionadas à segurança. Os dados do registro de auditoria do último ano são preservados.

Antes de começar

Verifique se você tem os pré-requisitos a seguir:

  • Uma instância do Google SecOps
  • Um projeto do Google Cloud com a API Storage Cloud ativada
  • Permissões para criar e gerenciar buckets do Cloud Storage
  • Permissões para gerenciar políticas do Identity and Access Management (IAM) em buckets do Cloud Storage
  • Permissões para criar serviços do Cloud Run, tópicos do Pub/Sub e jobs do Cloud Scheduler
  • Acesso de administrador à sua instância do CloudM Automate com a permissão Editar configurações globais
  • URL da sua instância do CloudM Automate (por exemplo, yourcompany.cloudm.io)
  • Seu ID de domínio do CloudM

Coletar credenciais do CloudM Automate

Criar um papel personalizado para acesso aos registros da API

  1. Faça login na sua instância do CloudM Automate.
  2. Acesse Configurações > Funções.
  3. Clique em Adicionar função para criar uma função.
  4. No campo Nome da função, insira um nome descritivo (por exemplo, Google SecOps Log Reader).

  5. Na lista de permissões, ative a seguinte permissão:

    • Visualizar registros: permite visualizar todos os registros do aplicativo.

  6. Salve a função.

Criar uma conta de serviço e atribuir o papel

  1. No CloudM Automate, acesse Configurações > Funções.
  2. Crie ou identifique uma conta de serviço para usar no acesso à API.
  3. Atribua o papel Leitor de registros do Google SecOps à conta de serviço.
  4. Verifique se o papel foi atribuído com o escopo global para que a conta de serviço possa acessar os registros em todo o domínio.

Receber o token de acesso da conta de serviço

  1. Gere um token de acesso para a conta de serviço.
  2. O token de acesso é usado como um token do portador no cabeçalho Authorization ao fazer solicitações de API para a API CloudM Logs.

  3. Anote os seguintes valores:

    • URL da instância do Automate: o URL da sua instância do CloudM Automate (por exemplo, yourcompany.cloudm.io)
    • ID do domínio: o identificador do seu domínio do CloudM.
    • Token de acesso da conta de serviço: o token do portador para autenticação da API.

Verifique as permissões

Para verificar se a conta tem as permissões necessárias:

  1. Faça login no CloudM Automate.
  2. Acesse Configurações > Funções.
  3. Verifique se a conta de serviço tem a permissão Ver registros atribuída com escopo global.
  4. Se essa opção não estiver disponível, entre em contato com o administrador para receber as permissões Editar configurações globais e Ver registros.

Testar o acesso à API

  • Teste suas credenciais antes de prosseguir com a integração:

    # Replace with your actual credentials
CLOUDM_INSTANCE="yourcompany.cloudm.io"
DOMAIN_ID="your-domain-id"
ACCESS_TOKEN="your-access-token"

# Test API access
curl -v -H "Authorization: Bearer ${ACCESS_TOKEN}" \
  "https://${CLOUDM_INSTANCE}/_ah/api/events/v1/${DOMAIN_ID}?from=$(date -u +%Y-%m-%d)&to=$(date -u +%Y-%m-%d)"

    Uma resposta bem-sucedida retorna uma matriz JSON de eventos de registro de auditoria.

Permissões de API necessárias

  • A conta de serviço precisa da seguinte permissão:

    Permissão Nível de acesso Finalidade
    Ver registros Global Recuperar todos os eventos de registro de auditoria do CloudM Automate

Criar Google Cloud bucket do Storage

  1. Acesse o Console do Google Cloud.
  2. Selecione ou crie um projeto.
  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 globalmente exclusivo (por exemplo, cloudm-audit-logs)
    Tipo de local Escolha de acordo com 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 Cloud Storage e ser invocada pelo Pub/Sub.

Criar conta de serviço

  1. No console doGoogle Cloud , 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 cloudm-audit-collector-sa.
    • Descrição da conta de serviço: insira Service account for Cloud Run function to collect CloudM audit logs
  4. Clique em Criar e continuar.

  5. Na seção Conceder a essa conta de serviço acesso 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 de funções do Cloud.

  6. Clique em Continuar.

  7. Clique em Concluído.

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

    • Administrador de objetos do Storage: grava registros no bucket do Cloud Storage 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 Cloud Storage

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

  1. Acesse Cloud Storage > Buckets.
  2. Clique no nome do bucket (cloudm-audit-logs).
  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 (cloudm-audit-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 doGoogle Cloud , acesse Pub/Sub > Tópicos.
  2. Selecione Criar tópico.

  3. Informe os seguintes detalhes de configuração:

    • ID do tópico: insira cloudm-audit-trigger
    • Não mude as outras configurações.

  4. Clique em Criar.

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

A função do Cloud Run será acionada por mensagens do Pub/Sub do Cloud Scheduler para buscar registros da API CloudM Automate Logs e gravá-los no Cloud Storage.

  1. No console doGoogle Cloud , 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 cloudm-audit-collector
    Região Selecione a região correspondente ao bucket do Cloud Storage (por exemplo, us-central1)
    Ambiente de execução Selecione Python 3.12 ou mais recente.

  5. Na seção Gatilho (opcional):

    1. Clique em + Adicionar gatilho.
    2. Selecione Cloud Pub/Sub.
    3. Em Selecionar um tópico do Cloud Pub/Sub, escolha cloudm-audit-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 para baixo e expanda Contêineres, rede, segurança.

  8. Acesse a guia Segurança:

    • Conta de serviço: selecione cloudm-audit-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 cloudm-audit-logs Nome do bucket do Cloud Storage
      GCS_PREFIX cloudm-audit Prefixo para arquivos de registro
      STATE_KEY cloudm-audit/state.json Caminho do arquivo de estado
      CLOUDM_INSTANCE_URL yourcompany.cloudm.io URL da instância do CloudM Automate
      CLOUDM_DOMAIN_ID your-domain-id Identificador de domínio do CloudM
      CLOUDM_ACCESS_TOKEN your-access-token Token do portador da conta de serviço do CloudM
      LOOKBACK_HOURS 24 Período de lookback inicial

  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

  13. Clique em Criar.

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

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

Adicionar código da função

  1. Insira main no campo Ponto de entrada.

  2. No editor de código inline, crie dois arquivos:

    • main.py:

      import functions_framework
from google.cloud import storage
import json
import os
import urllib3
from datetime import datetime, timezone, timedelta

http = urllib3.PoolManager(
  timeout=urllib3.Timeout(connect=10.0, read=60.0),
  retries=False,
)

storage_client = storage.Client()

GCS_BUCKET = os.environ.get('GCS_BUCKET')
GCS_PREFIX = os.environ.get('GCS_PREFIX', 'cloudm-audit')
STATE_KEY = os.environ.get('STATE_KEY', 'cloudm-audit/state.json')
CLOUDM_INSTANCE_URL = os.environ.get('CLOUDM_INSTANCE_URL', '').rstrip('/')
CLOUDM_DOMAIN_ID = os.environ.get('CLOUDM_DOMAIN_ID')
CLOUDM_ACCESS_TOKEN = os.environ.get('CLOUDM_ACCESS_TOKEN')
LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))

@functions_framework.cloud_event
def main(cloud_event):
  if not all([GCS_BUCKET, CLOUDM_INSTANCE_URL, CLOUDM_DOMAIN_ID, CLOUDM_ACCESS_TOKEN]):
    print('Error: Missing required environment variables')
    return

  try:
    bucket = storage_client.bucket(GCS_BUCKET)
    state = load_state(bucket)
    now = datetime.now(timezone.utc)

    if isinstance(state, dict) and state.get('last_event_date'):
      try:
        last_date = state['last_event_date']
        last_time = datetime.strptime(last_date, '%Y-%m-%d').replace(tzinfo=timezone.utc)
      except Exception as e:
        print(f"Warning: Could not parse last_event_date: {e}")
        last_time = now - timedelta(hours=LOOKBACK_HOURS)
    else:
      last_time = now - timedelta(hours=LOOKBACK_HOURS)

    from_date = last_time.strftime('%Y-%m-%d')
    to_date = now.strftime('%Y-%m-%d')

    print(f"Fetching logs from {from_date} to {to_date}")

    records = fetch_logs(from_date, to_date)

    if not records:
      print("No new log records found.")
      save_state(bucket, to_date)
      return

    timestamp = now.strftime('%Y%m%d_%H%M%S')
    object_key = f"{GCS_PREFIX}/cloudm_audit_{timestamp}.ndjson"
    blob = bucket.blob(object_key)

    ndjson = '\n'.join(
      [json.dumps(record, ensure_ascii=False, default=str) for record in records]
    ) + '\n'
    blob.upload_from_string(ndjson, content_type='application/x-ndjson')

    print(f"Wrote {len(records)} records to gs://{GCS_BUCKET}/{object_key}")

    save_state(bucket, to_date)

    print(f"Successfully processed {len(records)} records")

  except Exception as e:
    print(f'Error processing logs: {str(e)}')
    raise

def fetch_logs(from_date, to_date):
  instance = CLOUDM_INSTANCE_URL
  if not instance.startswith('https://'):
    instance = f"https://{instance}"

  endpoint = f"{instance}/_ah/api/events/v1/{CLOUDM_DOMAIN_ID}"

  headers = {
    'Authorization': f'Bearer {CLOUDM_ACCESS_TOKEN}',
    'Accept': 'application/json',
    'User-Agent': 'GoogleSecOps-CloudMCollector/1.0'
  }

  url = f"{endpoint}?from={from_date}&to={to_date}"

  try:
    response = http.request('GET', url, headers=headers)

    if response.status == 429:
      retry_after = int(response.headers.get('Retry-After', '60'))
      print(f"Rate limited (429). Retry after {retry_after}s.")
      return []

    if response.status != 200:
      print(f"HTTP Error: {response.status}")
      response_text = response.data.decode('utf-8')
      print(f"Response body: {response_text}")
      return []

    data = json.loads(response.data.decode('utf-8'))

    if isinstance(data, list):
      records = data
    elif isinstance(data, dict):
      records = data.get('items', data.get('events', [data]))
    else:
      records = []

    print(f"Retrieved {len(records)} events")
    return records

  except Exception as e:
    print(f"Error fetching logs: {e}")
    return []

def load_state(bucket):
  try:
    blob = bucket.blob(STATE_KEY)
    if blob.exists():
      return json.loads(blob.download_as_text())
  except Exception as e:
    print(f"Warning: Could not load state: {e}")
  return {}

def save_state(bucket, last_event_date):
  try:
    state = {
      'last_event_date': last_event_date,
      'last_run': datetime.now(timezone.utc).isoformat()
    }
    blob = bucket.blob(STATE_KEY)
    blob.upload_from_string(
      json.dumps(state, indent=2),
      content_type='application/json'
    )
    print(f"Saved state: last_event_date={last_event_date}")
  except Exception as e:
    print(f"Warning: Could not save state: {e}")

    • 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 vai publicar mensagens no tópico do Pub/Sub em intervalos regulares, acionando a função do Cloud Run.

  1. No Google Cloud console, acesse o Cloud Scheduler.
  2. Clique em Criar job.

  3. Informe os seguintes detalhes de configuração:

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

  4. Clique em Criar.

Testar a integração

  1. No console do Cloud Scheduler, encontre seu job (cloudm-audit-collector-hourly).
  2. Clique em Forçar execução para acionar o job manualmente.
  3. Aguarde alguns segundos.
  4. Acesse Cloud Run > Serviços.
  5. Clique em cloudm-audit-collector.
  6. Clique na guia Registros.

  7. Verifique se a função foi executada com sucesso. Procure por:

    Fetching logs from YYYY-MM-DD to YYYY-MM-DD
Retrieved X events
Wrote X records to gs://cloudm-audit-logs/cloudm-audit/cloudm_audit_YYYYMMDD_HHMMSS.ndjson
Successfully processed X records

  8. Acesse Cloud Storage > Buckets.

  9. Clique em cloudm-audit-logs.

  10. Navegue até a pasta cloudm-audit/.

  11. Verifique se um novo arquivo .ndjson foi criado com o carimbo de data/hora atual.

Se você encontrar erros nos registros:

  • HTTP 401: verifique se a variável de ambiente CLOUDM_ACCESS_TOKEN está correta.
  • HTTP 403: verifique se a conta de serviço tem a permissão Ver registros com escopo global.
  • HTTP 429: limitação de taxa. A função será interrompida e retomada na próxima execução programada.
  • Variáveis de ambiente ausentes: verifique se todas as variáveis necessárias estão definidas na configuração da função do Cloud Run.

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 Cloud Storage. É necessário conceder a essa conta de serviço acesso ao bucket.

Encontrar 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, CloudM Audit Logs).
  5. Selecione Google Cloud Storage V2 como o Tipo de origem.
  6. Selecione CloudM como o Tipo de registro.

  7. Clique em Get Service Account.

    Um e-mail exclusivo da conta de serviço será 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.

  9. Clique em Próxima.

  10. Especifique valores para os seguintes parâmetros de entrada:

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

      gs://cloudm-audit-logs/cloudm-audit/

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

      • Excluir arquivos transferidos e diretórios vazios: exclui arquivos e diretórios vazios após a transferência.

    • Idade máxima do arquivo: inclua arquivos modificados nos últimos dias (o padrão é 180 dias).

    • Namespace do recurso: o namespace do recurso.

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

  11. Clique em Próxima.

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

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 bucket do Cloud Storage.

  1. Acesse Cloud Storage > Buckets.
  2. Clique em cloudm-audit-logs.
  3. Acesse a guia Permissões.
  4. Clique em Conceder acesso.

  5. Informe os seguintes detalhes de configuração:

    • Adicionar principais: 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.

Parâmetros da API CloudM Logs

A API CloudM Logs é compatível com os seguintes parâmetros de consulta para filtrar eventos de registro:

Parâmetro Formatar Descrição
byUser Endereço de e-mail Filtre eventos pelo usuário que realizou a ação (análogo a Usuário na interface do CloudM).
de yyyy-MM-dd Data de início do filtro de período
a yyyy-MM-dd Data de término do filtro de período
contextType String Filtrar por tipo de contexto (por exemplo, perfil, grupo, UO)
contextName String Filtre pelo destino de uma ação (por exemplo, um usuário específico que está sendo desativado)
operação String Filtrar por tipo de operação (por exemplo, atribuir alias, suspender usuário)
country Código do país Filtrar por código de país para geolocalização

Tabela de mapeamento de UDM

Campo de registro Mapeamento de UDM Lógica
sobre sobre Valor copiado diretamente
Context_Name about.labels Mesclados como pares de chave-valor de about_Context_Name, about_Context_Type, labels0
Context_Type about.labels
Login_Type about.labels
Emissor additional.fields Mesclado de additional_field0, additional_field1, additional_field2
SAML_code additional.fields
SAML_ACS_Url additional.fields
Operação extensions.auth.type Definido como SSO se a operação corresponder a SSORequest, AUTHTYPE_UNSPECIFIED se Context_Type for LoginUser
Context_Type extensions.auth.type
Carimbo de data/hora metadata.event_timestamp Data e hora e fuso horário extraídos do carimbo de data/hora, fuso horário convertido em compensação, concatenado e analisado como carimbo de data/hora.
Operação metadata.event_type Definido como USER_UNCATEGORIZED se a operação corresponder a Update/Delete/SuspendUser/UnsuspendUser/Create, USER_LOGIN se a operação corresponder a SSORequest/SSORequestFail ou Context_Type for LoginUser, STATUS_UPDATE se IP não estiver vazio, caso contrário, GENERIC_EVENT
Context_Type metadata.event_type
User_Agent network.http.user_agent Valor copiado diretamente
participante participante Renomeado de "principal" se Context_Type != LoginUser, caso contrário, de "target"
destino participante
Organization_Unit principal.administrative_domain Valor copiado diretamente
IP principal.ip Valor copiado diretamente
Cidade principal.location.city Valor copiado diretamente
País principal.location.country_or_region Valor copiado diretamente
Geolocalização principal.location.region_latitude Latitude extraída da geolocalização usando grok
Geolocalização principal.location.region_longitude Longitude extraída da geolocalização usando grok
Região principal.location.state Valor copiado diretamente
Ator principal.user.attribute.roles Definido como "role.name" se o ator não for um e-mail e não estiver vazio. Em seguida, mesclado.
Ator principal.user.email_addresses Valor copiado diretamente se o ator corresponder à regex de e-mail
Mensagem principal.user.userid Nome de usuário extraído da mensagem usando grok
security_result security_result Mesclou o objeto security_result
SAML_code security_result.action Definido como ALLOW se SAML_code corresponder a Success, BLOCK se RequestDenied
Mensagem security_result.description Valor copiado diretamente
Gravidade security_result.severity Defina como maiúsculas se for Error/Critical, INFORMATIONAL se for Info, MEDIUM se for Warning, caso contrário, UNKNOWN_SEVERITY
Operação security_result.summary Valor copiado diretamente
destino destino Renomeado de "target" se Context_Type != LoginUser, caso contrário, de "principal"
participante destino
metadata.product_name metadata.product_name Definido como "CLOUDM"
metadata.vendor_name metadata.vendor_name Defina como "CLOUDM".

Registro de alterações

Ver o registro de alterações deste analisador

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