Coletar registros da Entrust Identity Verification (antiga Onfido)

Compatível com:

Este documento explica como ingerir registros do Entrust Identity Verification (antigo Onfido) no Google Security Operations usando o Google Cloud Storage V2. O analisador transforma os registros brutos de verificação e relatório do Onfido no esquema UDM do Google SecOps.

A Entrust Identity Verification é uma plataforma de verificação de identidade baseada na nuvem que automatiza a verificação de documentos, a análise biométrica e a detecção de fraudes. Ela oferece uma API REST para gerenciar candidatos, verificações e relatórios, permitindo que as organizações integrem fluxos de trabalho de verificação de identidade aos aplicativos.

Antes de começar

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

  • Uma instância do Google SecOps
  • Um projeto do GCP com a API Storage 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
  • Uma conta da Entrust Identity Verification (antiga Onfido) com acesso à API
  • Um token da API Onfido com permissões suficientes para ler verificações e relatórios

Coletar credenciais da API Onfido

Receber token da API

  1. Faça login no painel do Onfido.
  2. Acesse Desenvolvedores > Tokens de API.
  3. Copie um token de API ativo ou clique em Gerar token de API para criar um novo.
  4. Insira um nome para o token (por exemplo, Google Security Operations Integration).
  5. Selecione Ao vivo como o tipo de token.

  6. Copie e salve o token da API com segurança.

Verifique as permissões

Para verificar se o token da API tem as permissões necessárias:

  1. Faça login no painel do Onfido.
  2. Acesse Desenvolvedores > Tokens de API.
  3. Confirme se o token está listado com o status Ativo e não foi revogado.
  4. Verifique se o token tem acesso de leitura a verificações e relatórios testando o acesso à API.

Testar o acesso à API

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

    # Replace with your actual API token
ONFIDO_API_TOKEN="your-api-token"

# Test API access - list checks
curl -v -H "Authorization: Token token=${ONFIDO_API_TOKEN}" \
  "[https://api.onfido.com/v3.6/checks](https://api.onfido.com/v3.6/checks)"

Uma resposta bem-sucedida retorna HTTP 200 com um objeto JSON que contém uma matriz checks.

Criar um bucket do Google Cloud Storage

  1. Acesse o Console do Google Cloud.
  2. Selecione um 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, onfido-verification-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 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 onfido-logs-collector-sa.
    • Descrição da conta de serviço: insira Service account for Cloud Run function to collect Onfido verification 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 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:

  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, onfido-logs-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 onfido-logs-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 será acionada por mensagens do Pub/Sub do Cloud Scheduler para buscar verificações e relatórios da API Onfido e gravar 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 onfido-logs-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 onfido-logs-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 expanda Contêineres, rede, segurança.

  8. Acesse a guia Segurança:

    • Conta de serviço: selecione onfido-logs-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 onfido-verification-logs Nome do bucket do GCS
    GCS_PREFIX onfido-logs Prefixo para arquivos de registro
    STATE_KEY onfido-logs/state.json Caminho do arquivo de estado
    ONFIDO_API_TOKEN your-api-token-here Token da API Onfido
    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

  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 for criado, o editor de código inline será 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, 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', 'onfido-logs')
STATE_KEY = os.environ.get('STATE_KEY', 'onfido-logs/state.json')
ONFIDO_API_TOKEN = os.environ.get('ONFIDO_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'))

API_BASE = '[https://api.onfido.com/v3.6](https://api.onfido.com/v3.6)'

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 Onfido verification
    checks and reports and write to GCS.

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

    if not all([GCS_BUCKET, ONFIDO_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"])
                # Overlap by 2 minutes to catch any delayed events
                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 checks
        checks, newest_check_time = fetch_checks(
            api_token=ONFIDO_API_TOKEN,
            start_time=last_time,
            end_time=now,
            page_size=PAGE_SIZE,
            max_records=MAX_RECORDS,
        )

        # Fetch reports for each check
        all_records = []
        for check in checks:
            check_record = check.copy()
            check_id = check.get('id')
            if check_id:
                reports = fetch_reports(api_token=ONFIDO_API_TOKEN, check_id=check_id)
                check_record['reports'] = reports
            all_records.append(check_record)

        if not all_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 all_records]) + '\n'
        blob.upload_from_string(ndjson, content_type='application/x-ndjson')

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

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

        print(f"Successfully processed {len(all_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_checks(api_token: str, start_time: datetime, end_time: datetime, page_size: int, max_records: int):
    """
    Fetch verification checks from the Onfido API with pagination and rate limiting.

    Args:
        api_token: Onfido API token
        start_time: Start time for check query
        end_time: End time for check query
        page_size: Number of records per page
        max_records: Maximum total records to fetch

    Returns:
        Tuple of (checks list, newest_event_time ISO string)
    """
    headers = {
        'Authorization': f'Token token={api_token}',
        'Accept': 'application/json',
        'User-Agent': 'GoogleSecOps-OnfidoCollector/1.0'
    }

    records = []
    newest_time = None
    page_num = 0
    backoff = 1.0
    current_page = 1

    while True:
        page_num += 1

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

        url = f"{API_BASE}/checks?page={current_page}&per_page={page_size}"

        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'))

            page_results = data.get('checks', [])

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

            # Filter checks within the time window
            filtered = []
            for check in page_results:
                created_at = check.get('created_at')
                if created_at:
                    try:
                        check_time = parse_datetime(created_at)
                        if start_time <= check_time <= end_time:
                            filtered.append(check)
                        if newest_time is None or check_time > parse_datetime(newest_time):
                            newest_time = created_at
                    except Exception as e:
                        print(f"Warning: Could not parse check time: {e}")
                        filtered.append(check)

            print(f"Page {page_num}: Retrieved {len(page_results)} checks, {len(filtered)} in time window")
            records.extend(filtered)

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

            current_page += 1

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

    print(f"Retrieved {len(records)} total checks from {page_num} pages")
    return records[:max_records], newest_time

def fetch_reports(api_token: str, check_id: str):
    """
    Fetch reports for a specific check from the Onfido API.

    Args:
        api_token: Onfido API token
        check_id: Check ID to fetch reports for

    Returns:
        List of report objects
    """
    headers = {
        'Authorization': f'Token token={api_token}',
        'Accept': 'application/json',
        'User-Agent': 'GoogleSecOps-OnfidoCollector/1.0'
    }

    url = f"{API_BASE}/reports?check_id={check_id}"

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

        if response.status == 429:
            time.sleep(2)
            response = http.request('GET', url, headers=headers)

        if response.status != 200:
            print(f"Error fetching reports for check {check_id}: HTTP {response.status}")
            return []

        data = json.loads(response.data.decode('utf-8'))
        reports = data.get('reports', [])
        return reports

    except Exception as e:
        print(f"Error fetching reports for check {check_id}: {e}")
        return []
    • 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 vai publicar 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 onfido-logs-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 Selecionar onfido-logs-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
Diariamente 0 0 * * * Coleta de dados históricos

Testar a integração

  1. No console do Cloud Scheduler, encontre seu job.
  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 onfido-logs-collector.
  6. Clique na guia Registros.

  7. 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 checks, X in time window
Wrote X records to gs://bucket-name/onfido-logs/logs_YYYYMMDD_HHMMSS.ndjson
Successfully processed X records

  8. Acesse Cloud Storage > Buckets.

  9. Clique no nome do bucket.

  10. Navegue até a pasta onfido-logs/.

  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 o token da API Onfido nas variáveis de ambiente
  • HTTP 403: verifique se o token da API tem acesso ativo e não foi revogado.
  • 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.

Configurar um feed no Google SecOps para ingerir registros do Onfido

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

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

  8. Copie o endereço de e-mail. Você vai usá-la 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 GCS com o caminho do prefixo:

      gs://onfido-verification-logs/onfido-logs/
      • Substitua:
        • onfido-verification-logs: o nome do bucket do GCS.
        • onfido-logs: prefixo/caminho da pasta opcional onde os registros são armazenados (deixe em branco para a raiz).

    • 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 nos últimos 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.

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

Tabela de mapeamento do UDM

Campo de registro Mapeamento do UDM Lógica
read_only_udm.metadata.vendor_name Defina como "ONFIDO".
read_only_udm.metadata.product_name Defina como "ONFIDO".
read_only_udm.metadata.log_type Defina como "ONFIDO".

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.