Coletar registros de telefonia do Duo

Compatível com:

Google SecOps SIEM

Este documento explica como ingerir registros de telefonia do Duo no Google Security Operations usando o Google Cloud Storage. O analisador extrai campos dos registros, transforma e mapeia para o modelo de dados unificado (UDM). Ele processa vários formatos de registro do Duo, convertendo carimbos de data/hora, extraindo informações do usuário, detalhes da rede e resultados de segurança e, por fim, estruturando a saída no formato UDM padronizado.

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 serviços do Cloud Run, tópicos do Pub/Sub e jobs do Cloud Scheduler
Acesso privilegiado ao painel de administração do Duo com a função de proprietário

Coletar os pré-requisitos do Duo (credenciais da API)

Faça login no painel de administração do Duo como administrador com a função de proprietário.
Acesse Aplicativos > Catálogo de aplicativos.
Localize a entrada da API Admin no catálogo.
Clique em + Adicionar para criar o aplicativo.
Copie e salve em um local seguro os seguintes detalhes:
- Chave de integração
- Chave secreta
- Nome do host da API (por exemplo, api-yyyyyyyy.duosecurity.com)
Na seção Permissões, marque apenas a caixa de seleção da permissão Ler telefonia e desmarque todas as outras.
Clique em Salvar alterações.

Verifique as permissões

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

Faça login no painel de administração do Duo.
Acesse Aplicativos > Proteger um aplicativo.
Localize o aplicativo da API Admin.
Clique no nome do aplicativo para ver os detalhes.
Na seção Permissões, verifique se apenas a opção Ler telefonia está marcada.
Se outras permissões estiverem marcadas ou se "Ler telefonia" não estiver marcada, atualize as permissões e clique em Salvar alterações.

Testar o acesso à API

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

# Replace with your actual credentials
DUO_IKEY="your-integration-key"
DUO_SKEY="your-secret-key"
DUO_HOST="api-yyyyyyyy.duosecurity.com"

# Test API access (requires signing - use Duo's API test tool or Python script)
# Visit https://duo.com/docs/adminapi#testing to test your credentials

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, `duo-telephony-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 duo-telephony-collector-sa.
- Descrição da conta de serviço: insira Service account for Cloud Run function to collect Duo Telephony 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 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.
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, duo-telephony-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, duo-telephony-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 duo-telephony-trigger.
- Não altere as outras configurações.
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 Duo Telephony 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	`duo-telephony-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.

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 do Pub/Sub (duo-telephony-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 (duo-telephony-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
`GCS_BUCKET`	`duo-telephony-logs`
`GCS_PREFIX`	`duo-telephony`
`STATE_KEY`	`duo-telephony/state.json`
`DUO_IKEY`	`<your-integration-key>`
`DUO_SKEY`	`<your-secret-key>`
`DUO_API_HOST`	`api-yyyyyyyy.duosecurity.com`
`MAX_ITERATIONS`	`10`

Role a tela para baixo na guia Variáveis e secrets até Solicitações:
- Tempo limite da solicitação: insira 600 segundos (10 minutos).
Acesse a guia Configurações em Contêineres:
- Na seção Recursos:
  - Memória: selecione 512 MiB ou mais.
  - CPU: selecione 1.
- Clique em Concluído.
Role até Ambiente de execução:
- Selecione Padrão (recomendado).
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 é criado, o editor de código inline é aberto automaticamente.

Adicionar código da função

Insira main no campo Ponto de entrada.

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 hmac
import hashlib
import base64
import urllib.parse
import urllib3
import email.utils
from datetime import datetime, timedelta, timezone
from typing import Dict, Any, List, Optional

# 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 Duo telephony logs 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', 'duo-telephony').rstrip('/')
    state_key = os.environ.get('STATE_KEY', 'duo-telephony/state.json')
    integration_key = os.environ.get('DUO_IKEY')
    secret_key = os.environ.get('DUO_SKEY')
    api_hostname = os.environ.get('DUO_API_HOST')

    if not all([bucket_name, integration_key, secret_key, api_hostname]):
        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)

        # Calculate time range
        now = datetime.now(timezone.utc)

        if state.get('last_offset'):
            # Continue from last offset
            next_offset = state['last_offset']
            logs = []
            has_more = True
        else:
            # Start from last timestamp or 24 hours ago
            mintime = state.get('last_timestamp_ms', int((now - timedelta(hours=24)).timestamp() * 1000))
            # Apply 2-minute delay as recommended by Duo
            maxtime = int((now - timedelta(minutes=2)).timestamp() * 1000)
            next_offset = None
            logs = []
            has_more = True

        # Fetch logs with pagination
        total_fetched = 0
        max_iterations = int(os.environ.get('MAX_ITERATIONS', '10'))

        while has_more and total_fetched < max_iterations:
            page_num = total_fetched + 1

            if next_offset:
                # Use offset for pagination
                params = {
                    'limit': '100',
                    'next_offset': next_offset
                }
            else:
                # Initial request with time range
                params = {
                    'mintime': str(mintime),
                    'maxtime': str(maxtime),
                    'limit': '100',
                    'sort': 'ts:asc'
                }

            # Make API request with retry logic
            response = duo_api_call_with_retry(
                'GET',
                api_hostname,
                '/admin/v2/logs/telephony',
                params,
                integration_key,
                secret_key
            )

            if 'items' in response:
                logs.extend(response['items'])
                total_fetched += 1

            # Check for more data
            if 'metadata' in response and 'next_offset' in response['metadata']:
                next_offset = response['metadata']['next_offset']
                state['last_offset'] = next_offset
            else:
                has_more = False
                state['last_offset'] = None

                # Update timestamp for next run
                if logs:
                    # Get the latest timestamp from logs (ISO 8601 format)
                    latest_ts = max([log.get('ts', '') for log in logs])
                    if latest_ts:
                        # Convert ISO timestamp to milliseconds
                        dt = datetime.fromisoformat(latest_ts.replace('Z', '+00:00'))
                        state['last_timestamp_ms'] = int(dt.timestamp() * 1000) + 1

        # Save logs to GCS if any were fetched
        if logs:
            timestamp = datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')
            blob_name = f"{prefix}/telephony_{timestamp}.json"

            # Format logs as newline-delimited JSON
            log_data = '\n'.join(json.dumps(log) for log in logs)

            blob = bucket.blob(blob_name)
            blob.upload_from_string(
                log_data,
                content_type='application/x-ndjson'
            )

            print(f"Saved {len(logs)} telephony logs to gs://{bucket_name}/{blob_name}")
        else:
            print("No new telephony logs found")

        # Save state
        save_state(bucket, state_key, state)

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

def duo_api_call_with_retry(
    method: str,
    host: str,
    path: str,
    params: Dict[str, str],
    ikey: str,
    skey: str,
    max_retries: int = 3
) -> Dict[str, Any]:
    """Make an authenticated API call to Duo Admin API with retry logic."""
    for attempt in range(max_retries):
        try:
            return duo_api_call(method, host, path, params, ikey, skey)
        except Exception as e:
            if '429' in str(e) or '5' in str(e)[:1]:
                if attempt < max_retries - 1:
                    wait_time = (2 ** attempt) * 2
                    print(f"Retrying after {wait_time} seconds...")
                    import time
                    time.sleep(wait_time)
                    continue
            raise

def duo_api_call(
    method: str,
    host: str,
    path: str,
    params: Dict[str, str],
    ikey: str,
    skey: str
) -> Dict[str, Any]:
    """Make an authenticated API call to Duo Admin API."""
    # Create canonical string for signing using RFC 2822 date format
    now = email.utils.formatdate()
    canon = [now, method.upper(), host.lower(), path]

    # Add parameters
    args = []
    for key in sorted(params.keys()):
        val = params[key]
        args.append(f"{urllib.parse.quote(key, '~')}={urllib.parse.quote(val, '~')}")
    canon.append('&'.join(args))

    canon_str = '\n'.join(canon)

    # Sign the request
    sig = hmac.new(
        skey.encode('utf-8'),
        canon_str.encode('utf-8'),
        hashlib.sha1
    ).hexdigest()

    # Create authorization header
    auth = base64.b64encode(f"{ikey}:{sig}".encode('utf-8')).decode('utf-8')

    # Build URL
    url = f"https://{host}{path}"
    if params:
        url += '?' + '&'.join(args)

    # Make request
    headers = {
        'Authorization': f'Basic {auth}',
        'Date': now,
        'Host': host,
        'User-Agent': 'duo-telephony-gcs-ingestor/1.0'
    }

    try:
        response = http.request('GET', url, headers=headers)
        data = json.loads(response.data.decode('utf-8'))

        if data.get('stat') == 'OK':
            return data.get('response', {})
        else:
            raise Exception(f"API error: {data.get('message', 'Unknown error')}")
    except urllib3.exceptions.HTTPError as e:
        raise Exception(f"HTTP error: {str(e)}")

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 {}

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

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. Os limites de taxa da API Duo são de 5 solicitações por segundo e 1.000 solicitações por minuto. A função usa um limite de 100 registros por página (em vez do máximo de 1.000) para melhorar o desempenho da API e a conformidade com as recomendações do Duo.

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	`duo-telephony-logs-1h`
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 do Pub/Sub (`duo-telephony-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 o job do programador

No console do Cloud Scheduler, encontre seu job (duo-telephony-logs-1h).
Clique em Forçar execução para acionar manualmente.
Aguarde alguns segundos e acesse Cloud Run > Serviços.
Clique no nome da função (duo-telephony-logs-collector).
Clique na guia Registros.
Verifique se a função foi executada com sucesso.
Acesse Cloud Storage > Buckets.
Clique no nome do bucket (duo-telephony-logs).
Navegue até a pasta de prefixo (duo-telephony/).
Verifique se um novo arquivo .json foi criado com os registros.

Se você encontrar erros nos registros:

HTTP 401: verifique as credenciais da API (DUO_IKEY, DUO_SKEY, DUO_API_HOST) nas variáveis de ambiente.
HTTP 403: verifique se o aplicativo da API Admin tem a permissão Ler telefonia ativada.
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 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, Duo Telephony logs).
Selecione Google Cloud Storage V2 como o Tipo de origem.
Selecione Registros de telefonia do Duo 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.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.

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, duo-telephony-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.

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.

Configurar um feed no Google SecOps para ingerir registros da telefonia do Duo

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, Duo Telephony logs).
Selecione Google Cloud Storage V2 como o Tipo de origem.
Selecione Registros de telefonia do Duo 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://duo-telephony-logs/duo-telephony/
```
  - Substitua:
    - duo-telephony-logs: o nome do bucket do GCS.
    - duo-telephony: prefixo/caminho da pasta opcional onde os registros são armazenados (deixe em branco para a raiz).
  - Exemplos:
    - Bucket raiz: gs://duo-telephony-logs/
    - Com prefixo: gs://duo-telephony-logs/duo-telephony/
  Observação: sempre inclua a barra (/) 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 bem-sucedida.
  - Excluir arquivos transferidos e diretórios vazios: exclui arquivos e diretórios vazios após a transferência bem-sucedida.
  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: 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
contexto	metadata.product_event_type	Mapeado diretamente do campo "context" no registro bruto.
créditos	security_result.detection_fields.value	Mapeado diretamente do campo "credits" no registro bruto, aninhado em um objeto "detection_fields" com os créditos de chave correspondentes.
eventtype	security_result.detection_fields.value	Mapeado diretamente do campo "eventtype" no registro bruto, aninhado em um objeto "detection_fields" com o "eventtype" correspondente.
host	principal.hostname	Mapeado diretamente do campo "host" no registro bruto, se não for um endereço IP.
	security_result.action	Definido como um valor estático de "ALLOW" no analisador.
	security_result.detection_fields.value	Definido como um valor estático de "MECHANISM_UNSPECIFIED" no analisador.
	metadata.event_timestamp	Analisado do campo "ts" no registro bruto, que é uma string de carimbo de data/hora ISO 8601.
	metadata.event_type	Definido como "USER_UNCATEGORIZED" se os campos de contexto e host estiverem presentes no registro bruto. Defina como "STATUS_UPDATE" se apenas o organizador estiver presente. Caso contrário, defina como "GENERIC_EVENT".
	metadata.log_type	Extraído diretamente do campo "log_type" do registro bruto.
	metadata.product_name	Definido como um valor estático de "Telefonia" no analisador.
	metadata.vendor_name	Definido como um valor estático de "Duo" no analisador.
telefone	principal.user.phone_numbers	Mapeado diretamente do campo "phone" no registro bruto.
telefone	principal.user.userid	Mapeado diretamente do campo "phone" no registro bruto.
	security_result.severity	Definido como um valor estático de "INFORMATIONAL" no analisador.
	security_result.summary	Definido como um valor estático de "Telefonia do Duo" no analisador.
ts	metadata.event_timestamp	Analisado do campo "ts" no registro bruto, que é uma string de carimbo de data/hora ISO 8601.
tipo	security_result.summary	Mapeado diretamente do campo "type" no registro bruto.

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