Coletar registros de operações do Zoom
Este documento explica como ingerir registros de operações do Zoom no Google Security Operations usando o Google Cloud Storage. O analisador transforma os registros brutos em um modelo de dados unificado (UDM). Ele extrai campos da mensagem de registro bruta, realiza limpeza e normalização de dados e mapeia as informações extraídas para os campos correspondentes do UDM, enriquecendo os dados para análise e correlação em um sistema SIEM.
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 Zoom
Coletar pré-requisitos de registros de operações do Zoom
- Faça login no Zoom App Marketplace.
- Acesse Desenvolver > Criar app > OAuth de servidor para servidor.
- Crie o app e adicione o seguinte escopo:
report:read:operation_logs:admin(oureport:read:admin). Em Credenciais do app, copie e salve os seguintes detalhes em um local seguro:
- ID da conta
- Client-ID
- Client Secret
Verifique as permissões
Para verificar se a conta tem as permissões necessárias:
- Faça login na sua conta do Zoom.
- Acesse Administrador > Gerenciamento de contas > Perfil da conta.
- Se você conseguir acessar as configurações da conta e visualizar os registros de operações, terá as permissões necessárias.
- Se você não conseguir acessar essas opções, entre em contato com o administrador do Zoom para conceder as permissões necessárias.
Testar o acesso à API
Teste suas credenciais antes de prosseguir com a integração:
# Replace with your actual credentials ZOOM_ACCOUNT_ID="<your-account-id>" ZOOM_CLIENT_ID="<your-client-id>" ZOOM_CLIENT_SECRET="<your-client-secret>" # Get OAuth token TOKEN=$(curl -s -X POST "https://zoom.us/oauth/token?grant_type=account_credentials&account_id=${ZOOM_ACCOUNT_ID}" \ -u "${ZOOM_CLIENT_ID}:${ZOOM_CLIENT_SECRET}" \ | grep -o '"access_token":"[^"]*"' | cut -d'"' -f4) # Test API access curl -v -H "Authorization: Bearer ${TOKEN}" \ "https://api.zoom.us/v2/report/operationlogs?from=$(date -u -d '1 day ago' +%Y-%m-%d)&to=$(date -u +%Y-%m-%d)&page_size=10"
Criar um bucket do Google Cloud Storage
- Acesse o Console do Google Cloud.
- Selecione seu projeto ou crie um novo.
- No menu de navegação, acesse Cloud Storage > Buckets.
- Clique em Criar bucket.
Informe os seguintes detalhes de configuração:
Configuração Valor Nomeie seu bucket Insira um nome exclusivo globalmente, por exemplo, zoom-operation-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 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
- No Console do GCP, acesse IAM e administrador > Contas de serviço.
- Clique em Criar conta de serviço.
- Informe os seguintes detalhes de configuração:
- Nome da conta de serviço: insira
zoom-operationlogs-sa. - Descrição da conta de serviço: insira
Service account for Cloud Run function to collect Zoom operation logs.
- Nome da conta de serviço: insira
- Clique em Criar e continuar.
- Na seção Conceder acesso a essa conta de serviço ao projeto, adicione os seguintes papéis:
- Clique em Selecionar papel.
- Pesquise e selecione Administrador de objetos do Storage.
- Clique em + Adicionar outro papel.
- Pesquise e selecione Invocador do Cloud Run.
- Clique em + Adicionar outro papel.
- Pesquise e selecione Invocador do Cloud Functions.
- Clique em Continuar.
- 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.
- 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:
- Acesse Cloud Storage > Buckets.
- Clique no nome do bucket (por exemplo,
zoom-operation-logs). - Acesse a guia Permissões.
- Clique em Conceder acesso.
- Informe os seguintes detalhes de configuração:
- Adicionar principais: insira o e-mail da conta de serviço (por exemplo,
zoom-operationlogs-sa@PROJECT_ID.iam.gserviceaccount.com). - Atribuir papéis: selecione Administrador de objetos do Storage.
- Adicionar principais: insira o e-mail da conta de serviço (por exemplo,
- 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.
- No Console do GCP, acesse Pub/Sub > Tópicos.
- Selecione Criar tópico.
- Informe os seguintes detalhes de configuração:
- ID do tópico: insira
zoom-operationlogs-trigger. - Não altere as outras configurações.
- ID do tópico: insira
- 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 Zoom e gravá-los no GCS.
- No console do GCP, acesse o Cloud Run.
- Clique em Criar serviço.
- Selecione Função (use um editor in-line para criar uma função).
Na seção Configurar, forneça os seguintes detalhes de configuração:
Configuração Valor Nome do serviço zoom-operationlogs-to-gcsRegiã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. Na seção Acionador (opcional):
- Clique em + Adicionar gatilho.
- Selecione Cloud Pub/Sub.
- Em Selecionar um tópico do Cloud Pub/Sub, escolha
zoom-operationlogs-trigger. - Clique em Salvar.
Na seção Autenticação:
- Selecione Exigir autenticação.
- Confira o Identity and Access Management (IAM).
Role a tela para baixo e abra Contêineres, rede, segurança.
Acesse a guia Segurança:
- Conta de serviço: selecione
zoom-operationlogs-sa.
- Conta de serviço: selecione
Acesse a guia Contêineres:
- Clique em Variáveis e secrets.
- Clique em + Adicionar variável para cada variável de ambiente:
Nome da variável Valor de exemplo GCS_BUCKETzoom-operation-logsGCS_PREFIXzoom/operationlogs/STATE_KEYzoom/operationlogs/state.jsonZOOM_ACCOUNT_ID<your-zoom-account-id>ZOOM_CLIENT_ID<your-zoom-client-id>ZOOM_CLIENT_SECRET<your-zoom-client-secret>PAGE_SIZE300TIMEOUT30Na seção Variáveis e secrets, role a tela até Solicitações:
- Tempo limite da solicitação: insira
600segundos (10 minutos).
- Tempo limite da solicitação: insira
Acesse a guia Configurações:
- Na seção Recursos:
- Memória: selecione 512 MiB ou mais.
- CPU: selecione 1.
- Na seção Recursos:
Na seção Escalonamento de revisão:
- Número mínimo de instâncias: insira
0. - Número máximo de instâncias: insira
100ou ajuste com base na carga esperada.
- Número mínimo de instâncias: insira
Clique em Criar.
Aguarde a criação do serviço (1 a 2 minutos).
Depois que o serviço é criado, o editor de código inline é aberto automaticamente.
Adicionar código da função
- Insira main em Ponto de entrada da função.
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, date, timedelta, timezone import base64 import uuid import gzip import io # 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() # Environment variables GCS_BUCKET = os.environ.get('GCS_BUCKET') GCS_PREFIX = os.environ.get('GCS_PREFIX', 'zoom/operationlogs/') STATE_KEY = os.environ.get('STATE_KEY', 'zoom/operationlogs/state.json') ZOOM_ACCOUNT_ID = os.environ.get('ZOOM_ACCOUNT_ID') ZOOM_CLIENT_ID = os.environ.get('ZOOM_CLIENT_ID') ZOOM_CLIENT_SECRET = os.environ.get('ZOOM_CLIENT_SECRET') PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '300')) TIMEOUT = int(os.environ.get('TIMEOUT', '30')) TOKEN_URL = "https://zoom.us/oauth/token" REPORT_URL = "https://api.zoom.us/v2/report/operationlogs" @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch Zoom operation logs and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ if not all([GCS_BUCKET, ZOOM_ACCOUNT_ID, ZOOM_CLIENT_ID, ZOOM_CLIENT_SECRET]): print('Error: Missing required environment variables') return try: bucket = storage_client.bucket(GCS_BUCKET) # Get OAuth token token = get_token() # Load state state = load_state(bucket, STATE_KEY) cursor_date = state.get('cursor_date', date.today().isoformat()) print(f'Processing logs for date: {cursor_date}') # Fetch logs from_date = cursor_date to_date = cursor_date total_written = 0 next_token = state.get('next_page_token') while True: page = fetch_page(token, from_date, to_date, next_token) items = page.get('operation_logs', []) or [] if items: write_chunk(bucket, items, datetime.now(timezone.utc)) total_written += len(items) next_token = page.get('next_page_token') if not next_token: break # Advance to next day if we've finished this date today = date.today().isoformat() if cursor_date < today: nxt = (datetime.fromisoformat(cursor_date) + timedelta(days=1)).date().isoformat() state['cursor_date'] = nxt state['next_page_token'] = None else: # stay on today; continue later with next_page_token=None state['next_page_token'] = None save_state(bucket, STATE_KEY, state) print(f'Successfully processed {total_written} logs for {from_date}') except Exception as e: print(f'Error processing logs: {str(e)}') raise def get_token(): """Get OAuth 2.0 access token from Zoom.""" params = f"grant_type=account_credentials&account_id={ZOOM_ACCOUNT_ID}" basic = base64.b64encode(f"{ZOOM_CLIENT_ID}:{ZOOM_CLIENT_SECRET}".encode('utf-8')).decode('utf-8') headers = { 'Authorization': f'Basic {basic}', 'Content-Type': 'application/x-www-form-urlencoded', 'Accept': 'application/json', 'Host': 'zoom.us' } response = http.request( 'POST', TOKEN_URL, body=params, headers=headers, timeout=TIMEOUT ) if response.status != 200: print(f'Token request failed: {response.status}') response_text = response.data.decode('utf-8') print(f'Response body: {response_text}') raise Exception(f'Failed to get OAuth token: {response.status}') body = json.loads(response.data.decode('utf-8')) return body['access_token'] 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)}') # Initial state: start today today = date.today().isoformat() return {'cursor_date': today, 'next_page_token': None} def save_state(bucket, key, state): """Save state to GCS.""" try: state['updated_at'] = datetime.now(timezone.utc).isoformat() 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)}') def write_chunk(bucket, items, ts): """Write log chunk to GCS.""" key = f"{GCS_PREFIX}{ts:%Y/%m/%d}/zoom-operationlogs-{uuid.uuid4()}.json.gz" buf = io.BytesIO() with gzip.GzipFile(fileobj=buf, mode='w') as gz: for rec in items: gz.write((json.dumps(rec) + '\n').encode('utf-8')) buf.seek(0) blob = bucket.blob(key) blob.upload_from_file(buf, content_type='application/gzip') print(f'Wrote {len(items)} logs to {key}') return key def fetch_page(token, from_date, to_date, next_page_token): """Fetch a page of logs from Zoom API.""" params = { 'from': from_date, 'to': to_date, 'page_size': str(PAGE_SIZE) } if next_page_token: params['next_page_token'] = next_page_token # Build query string query_string = '&'.join([f"{k}={v}" for k, v in params.items()]) url = f"{REPORT_URL}?{query_string}" headers = { 'Authorization': f'Bearer {token}', 'Accept': 'application/json' } response = http.request( 'GET', url, headers=headers, timeout=TIMEOUT ) if response.status != 200: print(f'API request failed: {response.status}') response_text = response.data.decode('utf-8') print(f'Response body: {response_text}') raise Exception(f'Failed to fetch logs: {response.status}') return json.loads(response.data.decode('utf-8'))- Segundo arquivo: requirements.txt:
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0Clique em Implantar para salvar e implantar a função.
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.
- No Console do GCP, acesse o Cloud Scheduler.
- Clique em Criar job.
Informe os seguintes detalhes de configuração:
Configuração Valor Nome zoom-operationlogs-schedule-15minRegião Selecione a mesma região da função do Cloud Run Frequência */15 * * * *(a cada 15 minutos)Fuso horário Selecione o fuso horário (UTC recomendado) Tipo de destino Pub/Sub Tópico Selecionar zoom-operationlogs-triggerCorpo da mensagem {}(objeto JSON vazio)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 * * * *Padrão (recomendado) A cada hora 0 * * * *Volume baixo A cada 6 horas 0 */6 * * *Processamento em lote
Testar a integração
- No console do Cloud Scheduler, encontre seu job (por exemplo,
zoom-operationlogs-schedule-15min). - Clique em Executar à força para acionar o job manualmente.
- Aguarde alguns segundos.
- Acesse Cloud Run > Serviços.
- Clique no nome da função (
zoom-operationlogs-to-gcs). - Clique na guia Registros.
Verifique se a função foi executada com sucesso. Procure o seguinte:
Processing logs for date: YYYY-MM-DD Page 1: Retrieved X events Wrote X records to zoom/operationlogs/YYYY/MM/DD/zoom-operationlogs-UUID.json.gz Successfully processed X logs for YYYY-MM-DDAcesse Cloud Storage > Buckets.
Clique no nome do bucket (
zoom-operation-logs).Navegue até a pasta de prefixo (
zoom/operationlogs/).Verifique se um novo arquivo
.json.gzfoi criado com o carimbo de data/hora atual.
Se você encontrar erros nos registros:
- HTTP 401: verifique as credenciais da API Zoom nas variáveis de ambiente
- HTTP 403: verifique se o app Zoom tem o escopo
report:read:operation_logs:admin - 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 GCS. Você precisa conceder a essa conta de serviço acesso ao seu bucket.
Receber o e-mail da conta de serviço
- Acesse Configurações do SIEM > Feeds.
- Clique em Adicionar novo feed.
- Clique em Configurar um único feed.
- No campo Nome do feed, insira um nome para o feed (por exemplo,
Zoom Operation Logs). - Selecione Google Cloud Storage V2 como o Tipo de origem.
- Selecione Registros de operações do Zoom como o Tipo de registro.
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.comCopie 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.
- Acesse Cloud Storage > Buckets.
- Clique no nome do bucket (por exemplo,
zoom-operation-logs). - Acesse a guia Permissões.
- Clique em Conceder acesso.
- 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.
Clique em Salvar.
Configurar um feed no Google SecOps para ingerir registros de operações do Zoom
- Acesse Configurações do SIEM > Feeds.
- Clique em Adicionar novo feed.
- Clique em Configurar um único feed.
- No campo Nome do feed, insira um nome para o feed (por exemplo,
Zoom Operation Logs). - Selecione Google Cloud Storage V2 como o Tipo de origem.
- Selecione Registros de operações do Zoom como o Tipo de registro.
- Clique em Próxima.
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://zoom-operation-logs/zoom/operationlogs/Substitua:
zoom-operation-logs: o nome do bucket do GCS.zoom/operationlogs/: caminho do prefixo em que 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.
Clique em Próxima.
Revise a nova configuração do feed na tela Finalizar e clique em Enviar.
Tabela de mapeamento do UDM
| Campo de registro | Mapeamento do UDM | Lógica |
|---|---|---|
| ação | metadata.product_event_type | O campo de registro bruto "action" é mapeado para esse campo do UDM. |
| category_type | additional.fields.key | O campo de registro bruto "category_type" é mapeado para esse campo do UDM. |
| category_type | additional.fields.value.string_value | O campo de registro bruto "category_type" é mapeado para esse campo do UDM. |
| Departamento | target.user.department | O campo de registro bruto "Department" (extraído do campo "operation_detail") é mapeado para esse campo do UDM. |
| Descrição | target.user.role_description | O campo de registro bruto "Description" (extraído do campo "operation_detail") é mapeado para esse campo do UDM. |
| Nome de exibição | target.user.user_display_name | O campo de registro bruto "Nome de exibição" (extraído do campo "operation_detail") é mapeado para esse campo do UDM. |
| Endereço de e-mail | target.user.email_addresses | O campo de registro bruto "Email Address" (extraído do campo "operation_detail") é mapeado para esse campo do UDM. |
| Nome | target.user.first_name | O campo de registro bruto "Nome" (extraído do campo "operation_detail") é mapeado para esse campo do UDM. |
| Cargo | target.user.title | O campo de registro bruto "Cargo" (extraído do campo "operation_detail") é mapeado para esse campo do UDM. |
| Sobrenome | target.user.last_name | O campo de registro bruto "Sobrenome" (extraído do campo "operation_detail") é mapeado para esse campo do UDM. |
| Local | target.location.name | O campo de registro bruto "Location" (extraído do campo "operation_detail") é mapeado para esse campo do UDM. |
| operation_detail | metadata.description | O campo de registro bruto "operation_detail" é mapeado para esse campo do UDM. |
| operador | principal.user.email_addresses | O campo de registro bruto "operator" é mapeado para esse campo do UDM se corresponder a uma regex de e-mail. |
| operador | principal.user.userid | O campo de registro bruto "operator" é mapeado para esse campo do UDM se não corresponder a uma expressão regular de e-mail. |
| Nome do cômodo | target.user.attribute.labels.value | O campo de registro bruto "Nome da sala" (extraído do campo "operation_detail") é mapeado para esse campo do UDM. |
| Nome do papel | target.user.attribute.roles.name | O campo de registro bruto "Nome da função" (extraído do campo "operation_detail") é mapeado para esse campo do UDM. |
| tempo | metadata.event_timestamp.seconds | O campo de registro bruto "time" é analisado e mapeado para esse campo do UDM. |
| Tipo | target.user.attribute.labels.value | O campo de registro bruto "Type" (extraído do campo "operation_detail") é mapeado para esse campo do UDM. |
| Função do usuário | target.user.attribute.roles.name | O campo de registro bruto "Função do usuário" (extraído do campo "operation_detail") é mapeado para esse campo do UDM. |
| Tipo de usuário | target.user.attribute.labels.value | O campo de registro bruto "Tipo de usuário" (extraído do campo "operation_detail") é mapeado para esse campo do UDM. |
| metadata.log_type | O valor "ZOOM_OPERATION_LOGS" é atribuído a esse campo da UDM. | |
| metadata.vendor_name | O valor "ZOOM" é atribuído a esse campo da UDM. | |
| metadata.product_name | O valor "ZOOM_OPERATION_LOGS" é atribuído a esse campo da UDM. | |
| metadata.event_type | O valor é determinado com base na seguinte lógica: 1. Se o campo "event_type" não estiver vazio, o valor dele será usado. 1. Se os campos "operator", "email" ou "email2" não estiverem vazios, o valor será definido como "USER_UNCATEGORIZED". 1. Caso contrário, o valor será "GENERIC_EVENT". | |
| json_data | about.user.attribute.labels.value | O campo de registro bruto "json_data" (extraído do campo "operation_detail") é analisado como JSON. Os campos "assistant" e "options" de cada elemento da matriz JSON analisada são mapeados para o campo "value" da matriz "labels" na UDM. |
| json_data | about.user.userid | O campo de registro bruto "json_data" (extraído do campo "operation_detail") é analisado como JSON. O campo "userId" de cada elemento da matriz JSON analisada (exceto o primeiro) é mapeado para o campo "userid" do objeto "about.user" na UDM. |
| json_data | target.user.attribute.labels.value | O campo de registro bruto "json_data" (extraído do campo "operation_detail") é analisado como JSON. Os campos "assistant" e "options" do primeiro elemento da matriz JSON analisada são mapeados para o campo "value" da matriz "labels" na UDM. |
| json_data | target.user.userid | O campo de registro bruto "json_data" (extraído do campo "operation_detail") é analisado como JSON. O campo "userId" do primeiro elemento da matriz JSON analisada é mapeado para o campo "userid" do objeto "target.user" na UDM. |
| target.user.email_addresses | O campo de registro bruto "email" (extraído do campo "operation_detail") é mapeado para esse campo do UDM. | |
| email2 | target.user.email_addresses | O campo de registro bruto "email2" (extraído do campo "operation_detail") é mapeado para esse campo do UDM. |
| papel | target.user.attribute.roles.name | O campo de registro bruto "role" (extraído do campo "operation_detail") é mapeado para esse campo do UDM. |
Precisa de mais ajuda? Receba respostas de membros da comunidade e profissionais do Google SecOps.