Recolha registos de telefonia do Duo

Compatível com:

Google secops SIEM

Nota: esta funcionalidade está abrangida pelos Termos das Ofertas de Pré-DG dos Termos Específicos do serviço Google Security Operations. As funcionalidades de pré-DG podem ter suporte limitado e as alterações às funcionalidades de pré-DG podem não ser compatíveis com outras versões de pré-DG. Para mais informações, consulte as diretrizes dos serviços de apoio técnico do Google SecOps e os Termos Específicos dos Serviços do Google SecOps.

Este documento explica como carregar registos de telefonia do Duo para o Google Security Operations através do Amazon S3. O analisador extrai campos dos registos, transforma-os e mapeia-os para o modelo de dados unificado (UDM). Processa vários formatos de registo do Duo, converte as datas/horas, extrai informações do utilizador, detalhes da rede e resultados de segurança e, por fim, estrutura o resultado no formato UDM padronizado.

Antes de começar

Certifique-se de que cumpre os seguintes pré-requisitos:

• 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 e EventBridge).

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

1. Inicie sessão no painel de administração do Duo como administrador com a função de proprietário.
2. Aceda a Aplicações > Catálogo de aplicações.
3. Localize a entrada da API Admin no catálogo.
4. Clique em + Adicionar para criar a aplicação.
5. Copie e guarde numa localização segura os seguintes detalhes:
   • Chave de integração
   • Chave secreta
   • Nome do anfitrião da API (por exemplo, api-yyyyyyyy.duosecurity.com)
6. Na secção Autorizações, desmarque todas as opções de autorização, exceto Conceder registo de leitura.
7. Clique em Guardar alterações.

Configure o contentor do AWS S3 e o IAM para o Google SecOps

1. Crie um contentor do Amazon S3 seguindo este guia do utilizador: Criar um contentor
2. Guarde o nome e a região do contentor para referência futura (por exemplo, duo-telephony-logs).
3. Crie um utilizador seguindo este guia do utilizador: criar um utilizador do IAM.
4. Selecione o utilizador criado.
5. Selecione o separador Credenciais de segurança.
6. Clique em Criar chave de acesso na secção Chaves de acesso.
7. Selecione Serviço de terceiros como Exemplo de utilização.
8. Clicar em Seguinte.
9. Opcional: adicione uma etiqueta de descrição.
10. Clique em Criar chave de acesso.
11. Clique em Transferir ficheiro .CSV para guardar a chave de acesso e a chave de acesso secreta para referência futura.
12. Clique em Concluído.
13. Selecione o separador Autorizações.
14. Clique em Adicionar autorizações na secção Políticas de autorizações.
15. Selecione Adicionar autorizações.
16. Selecione Anexar políticas diretamente.
17. Pesquise a política AmazonS3FullAccess.
18. Selecione a política.
19. Clicar em Seguinte.
20. Clique em Adicionar autorizações.

Configure a política e a função de IAM para carregamentos do S3

1. Na consola da AWS, aceda a IAM > Políticas.
2. Clique em Criar política > separador JSON.
3. Copie e cole a seguinte política.

4. JSON da política (substitua duo-telephony-logs se tiver introduzido um nome de contentor 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 Seguinte > Criar política.

6. Aceda a IAM > Funções > Criar função > Serviço AWS > Lambda.

7. Anexe a política criada recentemente.

8. Dê o nome duo-telephony-lambda-role à função e clique em Criar função.

Crie a função Lambda

1. Na consola da AWS, aceda a Lambda > Functions > Create function.
2. Clique em Criar do zero.

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

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

4. Depois de criar a função, abra o separador Código, elimine o stub e cole o seguinte código (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. Aceda a Configuração > Variáveis de ambiente.

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

7. Introduza as variáveis de ambiente fornecidas na tabela seguinte, substituindo os valores de exemplo pelos seus valores.

   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 de criar a função, permaneça na respetiva página (ou abra Lambda > Functions > duo-telephony-logs-collector).

9. Selecione o separador Configuração.

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

11. Altere Tempo limite para 5 minutos (300 segundos) e clique em Guardar.

    Nota: o limite de tempo é um limite superior rígido, o que significa que, se a função ainda estiver em execução quando o temporizador atingir os 5 minutos, o AWS Lambda termina-a. Qualquer execução mais curta termina normalmente. A faturação é feita apenas com base no tempo de execução real.

Crie um horário do EventBridge

1. Aceda a Amazon EventBridge > Scheduler > Create schedule.
2. Indique os seguintes detalhes de configuração:
   • Agenda recorrente: Taxa (1 hour).
   • Alvo: a sua função Lambda duo-telephony-logs-collector.
   • Nome: duo-telephony-logs-1h.
3. Clique em Criar programação.

(Opcional) Crie um utilizador e chaves da IAM só de leitura para o Google SecOps

1. Aceda a AWS Console > IAM > Users.
2. Clique em Adicionar utilizadores.
3. Indique os seguintes detalhes de configuração:
   • Utilizador: introduza secops-reader.
   • Tipo de acesso: selecione Chave de acesso – Acesso programático.
4. Clique em Criar utilizador.
5. Anexe a política de leitura mínima (personalizada): Users > secops-reader > Permissions > Add permissions > Attach policies directly > Create policy.

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. Nome = secops-reader-policy.

8. Clique em Criar política > procure/selecione > Seguinte > Adicionar autorizaçõ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. Transfira o .CSV. (Vai colar estes valores no feed).

Configure um feed no Google SecOps para carregar registos de telefonia do Duo

1. Aceda a Definições do SIEM > Feeds.
2. Clique em + Adicionar novo feed.
3. No campo Nome do feed, introduza um nome para o feed (por exemplo, Duo Telephony logs).
4. Selecione Amazon S3 V2 como o Tipo de origem.
5. Selecione Registos de telefonia do Duo como Tipo de registo.
6. Clicar em Seguinte.
7. Especifique valores para os seguintes parâmetros de entrada:
   • URI do S3: s3://duo-telephony-logs/duo-telephony/
   • Opções de eliminação de origens: selecione a opção de eliminação de acordo com a sua preferência.
   • Idade máxima do ficheiro: inclua ficheiros modificados no último número de dias. A predefinição é 180 dias.
   • ID da chave de acesso: chave de acesso do utilizador com acesso ao contentor do S3.
   • Chave de acesso secreta: chave secreta do utilizador com acesso ao contentor do S3.
   • Espaço de nomes do recurso: o espaço de nomes do recurso.
   • Etiquetas de carregamento: a etiqueta aplicada aos eventos deste feed.
8. Clicar em Seguinte.
9. Reveja a nova configuração do feed no ecrã Finalizar e, de seguida, clique em Enviar.

Tabela de mapeamento do UDM

Campo de registo	Mapeamento do UDM	Lógica
context	metadata.product_event_type	Mapeado diretamente a partir do campo context no registo não processado.
credits	security_result.detection_fields.value	Mapeado diretamente a partir do campo credits no registo não processado, aninhado num objeto detection_fields com a chave correspondente credits.
eventtype	security_result.detection_fields.value	Mapeado diretamente a partir do campo eventtype no registo não processado, aninhado num objeto detection_fields com a chave correspondente eventtype.
host	principal.hostname	Mapeado diretamente a partir do campo host no registo não processado, 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 a partir do campo timestamp no registo não processado, que representa os segundos desde a época. Definido como "USER_UNCATEGORIZED" se os campos context e host estiverem presentes no registo não processado. Definido como "STATUS_UPDATE" se apenas host estiver presente. Caso contrário, defina como "GENERIC_EVENT". Extraído diretamente do campo log_type do registo não processado. 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 a partir do campo phone no registo não processado.
phone	principal.user.userid	Mapeado diretamente a partir do campo phone no registo não processado. Definido como um valor estático de "INFORMATIONAL" no analisador. Definido como um valor estático de "Duo Telephony" no analisador.
timestamp	metadata.event_timestamp	Analisado a partir do campo timestamp no registo não processado, que representa os segundos desde a época.
type	security_result.summary	Mapeado diretamente a partir do campo type no registo não processado.

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