O Google usa tecnologia de IA na tradução de conteúdos para seu idioma de preferência. As traduções com IA podem ter erros.

Coletar registros do Recordia

Compatível com:

Google SecOps SIEM

Este documento explica como ingerir registros do Recordia no Google Security Operations usando o Google Cloud Storage V2.

A Recordia (da Comunycarse) é uma plataforma de gravação de chamadas e compliance para serviços financeiros e centrais de atendimento. Ele captura e arquiva interações de voz e vídeo, fornecendo ferramentas para gerenciamento de qualidade, compliance regulatório e análise. O analisador extrai campos dos metadados de chamadas e eventos de auditoria do Recordia e os mapeia para o modelo de dados unificado (UDM), capturando detalhes da sessão, informações do autor/recebedor da chamada, atividade do usuário e gravidade da segurança.

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 Storage do Cloud ativada
Permissões para criar e gerenciar intervalos do GCS
Permissões para gerenciar políticas do IAM em buckets do GCS
Permissões para criar serviços do Cloud Run, tópicos do Pub/Sub e jobs do Cloud Scheduler
Acesso privilegiado à plataforma Recordia com acesso à API ativado. Entre em contato com o suporte da Comunycarse para receber as credenciais da API.

Criar um bucket do Google Cloud Storage

Acesse o Console do Google Cloud.
Selecione ou crie um projeto.
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 globalmente exclusivo, por exemplo, `recordia-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

Clique em Criar.

Coletar credenciais da API Recordia

Receber credenciais de API

Entre em contato com o gerente da sua conta do Recordia ou com o suporte da Comunycarse para pedir acesso à API.
Consiga as seguintes credenciais:
- URL base da API: o URL base da sua instância do Recordia (por exemplo, https://api.recordia.net)
- Token da API: um token de portador para autenticação.
Observação: as credenciais e os endpoints da API podem variar de acordo com a implantação. Consulte a documentação da API Recordia para saber o método de autenticação exato e os endpoints disponíveis.

Verifique as permissões

Para verificar se as credenciais da API têm as permissões necessárias:

Faça login no portal de administração do Recordia.
Acesse a seção Configurações ou API.
Confirme se o acesso à API está ativado na sua conta.
Se não for possível acessar as configurações da API, entre em contato com o suporte da Comunycarse.

Testar o acesso à API

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

# Replace with your actual credentials
API_TOKEN="<your-api-token>"
API_BASE="https://api.recordia.net"

# Test API access - list recordings
curl -v -H "Authorization: Bearer ${API_TOKEN}" \
  "${API_BASE}/v1/recordings?from=2024-01-01T00:00:00Z&to=2024-01-02T00:00:00Z"

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 recordia-collector-sa
- Descrição da conta de serviço: insira Service account for Cloud Run function to collect Recordia logs
Clique em Criar e continuar.
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 de funções do Cloud.
Clique em Continuar.
Clique em Concluído.

Essas funções são necessárias para:

Administrador de objetos do Storage: grava registros no 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, recordia-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, recordia-collector-sa@PROJECT_ID.iam.gserviceaccount.com).
- Atribuir papéis: selecione Administrador de objetos do Storage.
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 recordia-logs-trigger
- Não mude as outras configurações.
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 metadados de gravação e eventos de auditoria da API Recordia 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	`recordia-collector`
Região	Selecione a região correspondente 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 Gatilho (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 do Pub/Sub (recordia-logs-trigger).
4. Clique em Salvar.
Na seção Autenticação:
1. Selecione Exigir autenticação.
2. Confira o Identity and Access Management (IAM).
Observação: o Pub/Sub processa automaticamente a autenticação ao invocar a função.
Role a tela para baixo e abra Contêineres, rede, segurança.
Acesse a guia Segurança:
- Conta de serviço: selecione a conta de serviço (recordia-collector-sa).

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	Descrição
`GCS_BUCKET`	`recordia-logs`	Nome do bucket do GCS
`GCS_PREFIX`	`recordia`	Prefixo para arquivos de registro
`STATE_KEY`	`recordia/state.json`	Caminho do arquivo de estado
`API_BASE`	`https://api.recordia.net`	URL de base da API Recordia
`API_TOKEN`	`your-bearer-token`	Token do portador da API Recordia
`MAX_RECORDS`	`1000`	Máximo de registros por execução
`PAGE_SIZE`	`100`	Registros por página
`LOOKBACK_HOURS`	`24`	Período de lookback inicial

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)
Acesse a guia Configurações:
- Na seção Recursos:
  - Memória: selecione 512 MiB ou mais.
  - CPU: selecione 1
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.
Clique em Criar.
Aguarde a criação do serviço (1 a 2 minutos).
Depois que o serviço for criado, o editor de código inline será aberto automaticamente.

Adicionar código da função

Insira main no campo Ponto de entrada.

No editor de código inline, 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, timedelta
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()

# Environment variables
GCS_BUCKET = os.environ.get('GCS_BUCKET')
GCS_PREFIX = os.environ.get('GCS_PREFIX', 'recordia')
STATE_KEY = os.environ.get('STATE_KEY', 'recordia/state.json')
API_BASE = os.environ.get('API_BASE')
API_TOKEN = os.environ.get('API_TOKEN')
MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '1000'))
PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '100'))
LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))

def parse_datetime(value: str) -> datetime:
  """Parse ISO datetime string to datetime object."""
  if value.endswith("Z"):
    value = value[:-1] + "+00:00"
  return datetime.fromisoformat(value)

@functions_framework.cloud_event
def main(cloud_event):
  """
  Cloud Run function triggered by Pub/Sub to fetch Recordia
  recording metadata and audit events and write to GCS.

  Args:
    cloud_event: CloudEvent object containing Pub/Sub message
  """

  if not all([GCS_BUCKET, API_BASE, API_TOKEN]):
    print('Error: Missing required environment variables')
    return

  try:
    # Get GCS bucket
    bucket = storage_client.bucket(GCS_BUCKET)

    # Load state
    state = load_state(bucket, STATE_KEY)

    # Determine time window
    now = datetime.now(timezone.utc)
    last_time = None

    if isinstance(state, dict) and state.get("last_event_time"):
      try:
        last_time = parse_datetime(state["last_event_time"])
        last_time = last_time - timedelta(minutes=2)
      except Exception as e:
        print(f"Warning: Could not parse last_event_time: {e}")

    if last_time is None:
      last_time = now - timedelta(hours=LOOKBACK_HOURS)

    print(f"Fetching logs from {last_time.isoformat()} to {now.isoformat()}")

    # Fetch recordings
    records, newest_event_time = fetch_recordings(
      api_base=API_BASE,
      token=API_TOKEN,
      start_time=last_time,
      end_time=now,
      page_size=PAGE_SIZE,
      max_records=MAX_RECORDS,
    )

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

    # Write to GCS as NDJSON
    timestamp = now.strftime('%Y%m%d_%H%M%S')
    object_key = f"{GCS_PREFIX}/logs_{timestamp}.ndjson"
    blob = bucket.blob(object_key)

    ndjson = '\n'.join([json.dumps(record, ensure_ascii=False) 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}")

    # Update state with newest event time
    if newest_event_time:
      save_state(bucket, STATE_KEY, newest_event_time)
    else:
      save_state(bucket, STATE_KEY, now.isoformat())

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

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

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: {e}")

  return {}

def save_state(bucket, key, last_event_time_iso: str):
  """Save the last event timestamp to GCS state file."""
  try:
    state = {'last_event_time': last_event_time_iso}
    blob = bucket.blob(key)
    blob.upload_from_string(
      json.dumps(state, indent=2),
      content_type='application/json'
    )
    print(f"Saved state: last_event_time={last_event_time_iso}")
  except Exception as e:
    print(f"Warning: Could not save state: {e}")

def fetch_recordings(api_base: str, token: str, start_time: datetime, end_time: datetime, page_size: int, max_records: int):
  """
  Fetch recording metadata from Recordia API with pagination.

  Args:
    api_base: API base URL
    token: Bearer token for authentication
    start_time: Start time for log query
    end_time: End time for log query
    page_size: Number of records per page
    max_records: Maximum total records to fetch

  Returns:
    Tuple of (records list, newest_event_time ISO string)
  """
  base_url = api_base.rstrip('/')
  endpoint = f"{base_url}/v1/recordings"

  headers = {
    'Authorization': f'Bearer {token}',
    'Accept': 'application/json',
    'Content-Type': 'application/json',
    'User-Agent': 'GoogleSecOps-RecordiaCollector/1.0',
  }

  records = []
  newest_time = None
  page_num = 0
  backoff = 1.0
  offset = 0

  from_time = start_time.strftime('%Y-%m-%dT%H:%M:%SZ')
  to_time = end_time.strftime('%Y-%m-%dT%H:%M:%SZ')

  while True:
    page_num += 1

    if len(records) >= max_records:
      print(f"Reached max_records limit ({max_records})")
      break

    current_limit = min(page_size, max_records - len(records))
    url = f"{endpoint}?from={from_time}&to={to_time}&offset={offset}&limit={current_limit}"

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

      # Handle rate limiting with exponential backoff
      if response.status == 429:
        retry_after = int(response.headers.get('Retry-After', str(int(backoff))))
        print(f"Rate limited (429). Retrying after {retry_after}s...")
        time.sleep(retry_after)
        backoff = min(backoff * 2, 30.0)
        continue

      backoff = 1.0

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

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

      if isinstance(data, list):
        page_results = data
      else:
        page_results = data.get('recordings', data.get('results', []))

      if not page_results:
        print(f"No more results (empty page)")
        break

      print(f"Page {page_num}: Retrieved {len(page_results)} recordings")
      records.extend(page_results)

      # Track newest event time
      for event in page_results:
        try:
          event_time = event.get('startTime') or event.get('timestamp') or event.get('date')
          if event_time:
            if newest_time is None or parse_datetime(event_time) > parse_datetime(newest_time):
              newest_time = event_time
        except Exception as e:
          print(f"Warning: Could not parse event time: {e}")

      # Check for more results
      if len(page_results) < current_limit:
        print(f"Reached last page (size={len(page_results)} < limit={current_limit})")
        break

      offset += len(page_results)

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

  print(f"Retrieved {len(records)} total recordings from {page_num} pages")
  return records, newest_time

Segundo arquivo:requirements.txt:

functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0

Clique em Implantar para salvar e implantar a função.
Aguarde a conclusão da implantação (2 a 3 minutos).

Observação: a configuração do acionador do Pub/Sub cria automaticamente as assinaturas e permissões necessárias.

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.

No console do GCP, acesse o Cloud Scheduler.
Clique em Criar job.

Informe os seguintes detalhes de configuração:

Configuração	Valor
Nome	`recordia-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	Selecione o tópico do Pub/Sub (`recordia-logs-trigger`).
Corpo 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 * * *`	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

No console do Cloud Scheduler, encontre seu job.
Clique em Forçar execução para acionar o job manualmente.
Aguarde alguns segundos.
Acesse Cloud Run > Serviços.
Clique no nome da função (recordia-collector).
Clique na guia Registros.

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

Fetching logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00
Page 1: Retrieved X recordings
Wrote X records to gs://recordia-logs/recordia/logs_YYYYMMDD_HHMMSS.ndjson
Successfully processed X records

Acesse Cloud Storage > Buckets.
Clique no nome do bucket (recordia-logs).
Navegue até a pasta de prefixo (recordia/).
Verifique se um novo arquivo .ndjson foi criado com o carimbo de data/hora atual.

Se você encontrar erros nos registros:

HTTP 401: verifique o token da API nas variáveis de ambiente
HTTP 403: verifique se o token da API tem acesso ao endpoint de gravações.
HTTP 429: limitação de taxa. A função vai tentar de novo automaticamente com espera.
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

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, Recordia Logs).
Selecione Google Cloud Storage V2 como o Tipo de origem.
Selecione Recordia como o Tipo de registro.
Clique em Receber conta de serviço.
Um e-mail exclusivo da conta de serviço será exibido, por exemplo:
```
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
```
Copie esse endereço de e-mail para usar na próxima etapa.

Observação: cada instância do Google SecOps tem uma conta de serviço exclusiva. Não use contas de serviço de outras documentações ou exemplos.
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://recordia-logs/recordia/
```
  - Substitua:
    - recordia-logs: o nome do bucket do GCS.
    - recordia: prefixo/caminho da pasta opcional onde os registros são armazenados (deixe em branco para a raiz).
Observação: sempre inclua a barra invertida (/) no final do URI.
- 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.
    
    Observação: se você selecionar uma opção de exclusão, a conta de serviço precisará ter o papel Administrador de objetos do Storage em vez de Leitor de objetos do Storage. Atualize as permissões do IAM de acordo com a necessidade.
- Idade máxima do arquivo: inclua arquivos modificados no último número de 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
Clique em Próxima.
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 Leitor de objetos do Storage no seu bucket do GCS.

Acesse Cloud Storage > Buckets.
Clique no nome do bucket (por exemplo, recordia-logs).
Acesse a guia Permissões.
Clique em Conceder acesso.
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.
Clique em Salvar.

Observação: se você planeja usar a opção de exclusão "Excluir arquivos transferidos" ou "Excluir arquivos transferidos e diretórios vazios", conceda o papel Administrador de objetos do Storage em vez de Leitor de objetos do Storage.

Tabela de mapeamento de UDM

Campo de registro	Mapeamento do UDM	Lógica
record_id, batch_id, ProviderId, ProviderGroupId, TrackingId, RecordingType, Format, Proprietário, NotEncrypted, Hash, CustomData, CustomDataFormat	additional.fields	Mesclado com objetos de rótulo construídos de cada campo, se não estiver vazio, em que label.key é definido como uma string específica e label.value.string_value como o valor do campo.
client_ip,client_ip1	intermediary.ip	Mesclado com client_ip e depois com client_ip1
corp_user	intermediary.user.userid	Valor copiado diretamente
descrição	metadata.description	Valor copiado diretamente
auth_user,src_ip	metadata.event_type	Definido como "USER_UNCATEGORIZED" se auth_user não estiver vazio, caso contrário, "STATUS_UNCATEGORIZED" se src_ip não estiver vazio, caso contrário, "GENERIC_EVENT"
	metadata.product_event_type	Definido como "RECORDING BATCH" se event_info corresponder a "RECORDING BATCH"
CallId	metadata.product_log_id	Valor copiado diretamente
user_agent	network.http.parsed_user_agent	Valor copiado e convertido para o tipo parseduseragent
user_agent	network.http.user_agent	Valor copiado diretamente
session_id	network.session_id	Valor copiado diretamente
src_ip	principal.ip	Valor copiado diretamente
url	principal.url	Valor copiado diretamente
CallingParty	principal.user.phone_numbers	Valor copiado diretamente
auth_user	principal.user.userid	Valor copiado diretamente
gravidade,	security_result.action_details	O valor é copiado diretamente se a gravidade estiver em ["INFO", "DEBUG"]
gravidade,	security_result.severity	Defina como "LOW" se a gravidade estiver em ["INFO", "DEBUG"]
CalledParty	target.user.phone_numbers	Valor copiado diretamente
	metadata.vendor_name	Definido como "Comunycarse"
	metadata.product_name	Definido como "Recordia"

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.