Coletar registros de telefonia do Duo

Compatível com:

Google SecOps SIEM

Observação: esse recurso é coberto pelos Termos de produtos pré-GA dos Termos específicos de serviço do Google Security Operations. Os recursos pré-GA podem ter suporte limitado, e as mudanças realizadas podem não ser compatíveis com outras versões pré-GA. Para mais informações, consulte as diretrizes do serviço de suporte técnico do Google SecOps e os Termos específicos de serviço do Google SecOps.

Este documento explica como ingerir registros de telefonia do Duo no Google Security Operations usando o Amazon S3. 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, converte carimbos de data/hora, extrai informações do usuário, detalhes da rede e resultados de segurança e, por fim, estrutura a saída no formato UDM padronizado.

Antes de começar

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

• Instância do Google SecOps.
• Acesso privilegiado ao Painel de administração do Duo com a função de Proprietário.
• Acesso privilegiado à AWS (S3, Identity and Access Management (IAM), Lambda, EventBridge).

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

1. Faça login no painel de administração do Duo como administrador com a função de Proprietário.
2. Acesse Aplicativos > Catálogo de aplicativos.
3. Localize a entrada da API Admin no catálogo.
4. Clique em + Adicionar para criar o aplicativo.
5. 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)
6. Na seção Permissões, desmarque todas as opções de permissão, exceto Conceder leitura de registros.
7. Clique em Salvar alterações.

Configurar o bucket do AWS S3 e o IAM para o Google SecOps

1. Crie um bucket do Amazon S3 seguindo este guia do usuário: Como criar um bucket
2. Salve o Nome e a Região do bucket para referência futura (por exemplo, duo-telephony-logs).
3. Crie um usuário seguindo este guia: Como criar um usuário do IAM.
4. Selecione o usuário criado.
5. Selecione a guia Credenciais de segurança.
6. Clique em Criar chave de acesso na seção Chaves de acesso.
7. Selecione Serviço de terceiros como Caso de uso.
8. Clique em Próxima.
9. Opcional: adicione uma tag de descrição.
10. Clique em Criar chave de acesso.
11. Clique em Fazer o download do arquivo .CSV para salvar a chave de acesso e a chave de acesso secreta para referência futura.
12. Clique em Concluído.
13. Selecione a guia Permissões.
14. Clique em Adicionar permissões na seção Políticas de permissões.
15. Selecione Adicionar permissões.
16. Selecione Anexar políticas diretamente.
17. Pesquise a política AmazonS3FullAccess.
18. Selecione a política.
19. Clique em Próxima.
20. Clique em Adicionar permissões

Configurar a política e o papel do IAM para uploads do S3

1. No console da AWS, acesse IAM > Políticas.
2. Clique em Criar política > guia JSON.
3. Copie e cole a política a seguir.

4. JSON da política (substitua duo-telephony-logs se você tiver inserido um nome de bucket diferente):

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Sid": "AllowPutObjects",
         "Effect": "Allow",
         "Action": "s3:PutObject",
         "Resource": "arn:aws:s3:::duo-telephony-logs/*"
       },
       {
         "Sid": "AllowGetStateObject",
         "Effect": "Allow",
         "Action": "s3:GetObject",
         "Resource": "arn:aws:s3:::duo-telephony-logs/duo-telephony/state.json"
       }
     ]
   }
   

5. Clique em Próxima > Criar política.

6. Acesse IAM > Funções > Criar função > Serviço da AWS > Lambda.

7. Anexe a política recém-criada.

8. Nomeie a função como duo-telephony-lambda-role e clique em Criar função.

Criar a função Lambda

1. No console da AWS, acesse Lambda > Functions > Create function.
2. Clique em Criar do zero.

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

   Configuração	Valor
   Nome	duo-telephony-logs-collector
   Ambiente de execução	Python 3.13
   Arquitetura	x86_64
   Função de execução	duo-telephony-lambda-role

4. Depois que a função for criada, abra a guia Código, exclua o stub e cole o código a seguir (duo-telephony-logs-collector.py).

   import json
   import boto3
   import os
   import hmac
   import hashlib
   import base64
   import urllib.parse
   import urllib.request
   import email.utils
   from datetime import datetime, timedelta, timezone
   from typing import Dict, Any, List, Optional
   from botocore.exceptions import ClientError
   
   s3 = boto3.client('s3')
   
   def lambda_handler(event, context):
       """
       Lambda function to fetch Duo telephony logs and store them in S3.
       """
       try:
           # Get configuration from environment variables
           bucket_name = os.environ['S3_BUCKET']
           s3_prefix = os.environ['S3_PREFIX'].rstrip('/')
           state_key = os.environ['STATE_KEY']
           integration_key = os.environ['DUO_IKEY']
           secret_key = os.environ['DUO_SKEY']
           api_hostname = os.environ['DUO_API_HOST']
   
           # Load state
           state = load_state(bucket_name, 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:
               if next_offset:
                   # Use offset for pagination
                   params = {
                       'limit': '1000',
                       'next_offset': next_offset
                   }
               else:
                   # Initial request with time range
                   params = {
                       'mintime': str(mintime),
                       'maxtime': str(maxtime),
                       'limit': '1000',
                       '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
                           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
               else:
                   has_more = False
   
           # Save logs to S3 if any were fetched
           if logs:
               timestamp = datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')
               key = f"{s3_prefix}/telephony_{timestamp}.json"
   
               # Format logs as newline-delimited JSON
               log_data = '\n'.join(json.dumps(log) for log in logs)
   
               s3.put_object(
                   Bucket=bucket_name,
                   Key=key,
                   Body=log_data.encode('utf-8'),
                   ContentType='application/x-ndjson'
               )
   
               print(f"Saved {len(logs)} telephony logs to s3://{bucket_name}/{key}")
           else:
               print("No new telephony logs found")
   
           # Save state
           save_state(bucket_name, state_key, state)
   
           return {
               'statusCode': 200,
               'body': json.dumps({
                   'message': f'Successfully processed {len(logs)} telephony logs',
                   'logs_count': len(logs)
               })
           }
   
       except Exception as e:
           print(f"Error: {str(e)}")
           return {
               'statusCode': 500,
               'body': json.dumps({'error': str(e)})
           }
   
   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]:  # Rate limit or server error
                   if attempt < max_retries - 1:
                       wait_time = (2 ** attempt) * 2  # Exponential backoff
                       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
       req = urllib.request.Request(url)
       req.add_header('Authorization', f'Basic {auth}')
       req.add_header('Date', now)
       req.add_header('Host', host)
       req.add_header('User-Agent', 'duo-telephony-s3-ingestor/1.0')
   
       try:
           with urllib.request.urlopen(req, timeout=30) as response:
               data = json.loads(response.read().decode('utf-8'))
               if data.get('stat') == 'OK':
                   return data.get('response', {})
               else:
                   raise Exception(f"API error: {data.get('message', 'Unknown error')}")
       except urllib.error.HTTPError as e:
           error_body = e.read().decode('utf-8')
           raise Exception(f"HTTP error {e.code}: {error_body}")
   
   def load_state(bucket: str, key: str) -> Dict[str, Any]:
       """Load state from S3."""
       try:
           response = s3.get_object(Bucket=bucket, Key=key)
           return json.loads(response['Body'].read().decode('utf-8'))
       except ClientError as e:
           if e.response.get('Error', {}).get('Code') in ('NoSuchKey', '404'):
               return {}
           print(f"Error loading state: {e}")
           return {}
       except Exception as e:
           print(f"Error loading state: {e}")
           return {}
   
   def save_state(bucket: str, key: str, state: Dict[str, Any]):
       """Save state to S3."""
       try:
           s3.put_object(
               Bucket=bucket,
               Key=key,
               Body=json.dumps(state).encode('utf-8'),
               ContentType='application/json'
           )
       except Exception as e:
           print(f"Error saving state: {e}")
   

5. Acesse Configuração > Variáveis de ambiente.

6. Clique em Editar > Adicionar nova variável de ambiente.

7. Insira as variáveis de ambiente fornecidas na tabela a seguir, substituindo os valores de exemplo pelos seus.

   Variáveis de ambiente

   Chave	Valor de exemplo
   S3_BUCKET	duo-telephony-logs
   S3_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

8. Depois que a função for criada, permaneça na página dela ou abra Lambda > Functions > duo-telephony-logs-collector.

9. Selecione a guia Configuração.

10. No painel Configuração geral, clique em Editar.

11. Mude Tempo limite para 5 minutos (300 segundos) e clique em Salvar.

    Observação: o tempo limite é um limite superior fixo. Isso significa que, se a função ainda estiver em execução quando o timer atingir 5 minutos, o AWS Lambda vai encerrá-la. Execuções mais curtas terminam normalmente, e você só recebe cobranças pelo tempo de execução real.

Criar uma programação do EventBridge

1. Acesse Amazon EventBridge > Scheduler > Criar programação.
2. Informe os seguintes detalhes de configuração:
   • Programação recorrente: Taxa (1 hour).
   • Destino: sua função Lambda duo-telephony-logs-collector.
   • Nome: duo-telephony-logs-1h.
3. Clique em Criar programação.

(Opcional) Criar um usuário e chaves do IAM somente leitura para o Google SecOps

1. Acesse Console da AWS > IAM > Usuários.
2. Clique em Add users.
3. Informe os seguintes detalhes de configuração:
   • Usuário: insira secops-reader.
   • Tipo de acesso: selecione Chave de acesso – Acesso programático.
4. Clique em Criar usuário.
5. Anexe a política de leitura mínima (personalizada): Usuários > secops-reader > Permissões > Adicionar permissões > Anexar políticas diretamente > Criar política.

6. JSON:

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Effect": "Allow",
         "Action": ["s3:GetObject"],
         "Resource": "arn:aws:s3:::duo-telephony-logs/*"
       },
       {
         "Effect": "Allow",
         "Action": ["s3:ListBucket"],
         "Resource": "arn:aws:s3:::duo-telephony-logs"
       }
     ]
   }
   

7. Name = secops-reader-policy.

8. Clique em Criar política > pesquisar/selecionar > Próxima > Adicionar permissões.

9. Crie uma chave de acesso para secops-reader: Credenciais de segurança > Chaves de acesso.

10. Clique em Criar chave de acesso.

11. Faça o download do .CSV. Cole esses valores no feed.

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

1. Acesse Configurações do SIEM > Feeds.
2. Clique em + Adicionar novo feed.
3. No campo Nome do feed, insira um nome para o feed (por exemplo, Duo Telephony logs).
4. Selecione Amazon S3 V2 como o Tipo de origem.
5. Selecione Registros de telefonia do Duo como o Tipo de registro.
6. Clique em Próxima.
7. Especifique valores para os seguintes parâmetros de entrada:
   • URI do S3: s3://duo-telephony-logs/duo-telephony/
   • Opções de exclusão de fontes: selecione a opção de exclusão de acordo com sua preferência.
   • Idade máxima do arquivo: inclui arquivos modificados no último número de dias. O padrão é de 180 dias.
   • ID da chave de acesso: chave de acesso do usuário com acesso ao bucket do S3.
   • Chave de acesso secreta: chave secreta do usuário com acesso ao bucket do S3.
   • Namespace do recurso: o namespace do recurso.
   • Rótulos de ingestão: o rótulo aplicado aos eventos deste feed.
8. Clique em Próxima.
9. 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
context	metadata.product_event_type	Mapeado diretamente do campo context no registro bruto.
credits	security_result.detection_fields.value	Mapeado diretamente do campo credits no registro bruto, aninhado em um objeto detection_fields com a chave credits correspondente.
eventtype	security_result.detection_fields.value	Mapeado diretamente do campo eventtype no registro bruto, aninhado em um objeto detection_fields com a chave eventtype correspondente.
host	principal.hostname	Mapeado diretamente do campo host no registro bruto, se não for um endereço IP. Definido como um valor estático de "ALLOW" no analisador. Definido como um valor estático de "MECHANISM_UNSPECIFIED" no analisador. Analisado do campo timestamp no registro bruto, que representa segundos desde o período. Definido como "USER_UNCATEGORIZED" se os campos context e host estiverem presentes no registro bruto. Defina como "STATUS_UPDATE" se apenas host estiver presente. Caso contrário, defina como "GENERIC_EVENT". Extraído diretamente do campo log_type do registro bruto. Definido como um valor estático de "Telefonia" no analisador. Definido como um valor estático de "Duo" no analisador.
phone	principal.user.phone_numbers	Mapeado diretamente do campo phone no registro bruto.
phone	principal.user.userid	Mapeado diretamente do campo phone no registro bruto. Definido como um valor estático de "INFORMATIONAL" no analisador. Definido como um valor estático de "Telefonia do Duo" no analisador.
timestamp	metadata.event_timestamp	Analisado do campo timestamp no registro bruto, que representa segundos desde o período.
type	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.