Esta página se ha traducido con Cloud Translation API.

Recoger registros de telefonía de Duo

Disponible en:

SecOps de Google SIEM

En este documento se explica cómo ingerir registros de telefonía de Duo en Google Security Operations mediante Amazon S3. El analizador extrae campos de los registros, los transforma y los asigna al modelo de datos unificado (UDM). Gestiona varios formatos de registro de Duo, convierte marcas de tiempo, extrae información de los usuarios, detalles de la red y resultados de seguridad, y, por último, estructura la salida en el formato UDM estandarizado.

Antes de empezar

Asegúrate de que cumples los siguientes requisitos previos:

Instancia de Google SecOps.
Acceso privilegiado al panel de administración de Duo con el rol Propietario.
Acceso privilegiado a AWS (S3, Gestión de Identidades y Accesos [IAM], Lambda y EventBridge).

Recopilar los requisitos previos de Duo (credenciales de API)

Inicia sesión en el panel de administración de Duo como administrador con el rol Propietario.
Ve a Aplicaciones > Catálogo de aplicaciones.
Busca la entrada de API Admin en el catálogo.
Haz clic en + Añadir para crear la aplicación.
Copia y guarda en un lugar seguro los siguientes detalles:
- Clave de integración
- Clave secreta
- Nombre de host de la API (por ejemplo, api-yyyyyyyy.duosecurity.com)
En la sección Permisos, desmarca todas las opciones de permisos excepto Conceder registro de lectura.
Haz clic en Guardar cambios.

Configurar un segmento de AWS S3 y IAM para Google SecOps

Crea un segmento de Amazon S3 siguiendo esta guía de usuario: Crear un segmento.
Guarda el nombre y la región del segmento para consultarlos más adelante (por ejemplo, duo-telephony-logs).
Crea un usuario siguiendo esta guía: Crear un usuario de gestión de identidades y accesos.
Selecciona el Usuario creado.
Selecciona la pestaña Credenciales de seguridad.
En la sección Claves de acceso, haz clic en Crear clave de acceso.
Selecciona Servicio de terceros en Caso práctico.
Haz clic en Siguiente.
Opcional: añade una etiqueta de descripción.
Haz clic en Crear clave de acceso.
Haz clic en Descargar archivo .CSV para guardar la clave de acceso y la clave de acceso secreta para futuras consultas.
Haz clic en Listo.
Selecciona la pestaña Permisos.
En la sección Políticas de permisos, haz clic en Añadir permisos.
Selecciona Añadir permisos.
Seleccione Adjuntar políticas directamente.
Busca la política AmazonS3FullAccess.
Selecciona la política.
Haz clic en Siguiente.
Haz clic en Añadir permisos.

Configurar la política y el rol de gestión de identidades y accesos para las subidas de S3

En la consola de AWS, ve a IAM > Políticas.
Haz clic en Crear política > pestaña JSON.
Copia y pega la siguiente política.

JSON de la política (sustituye duo-telephony-logs si has introducido otro nombre de contenedor):

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

Haz clic en Siguiente > Crear política.
Ve a IAM > Roles > Crear rol > Servicio de AWS > Lambda.
Adjunte la política que acaba de crear.
Dale el nombre duo-telephony-lambda-role al rol y haz clic en Crear rol.

Crear la función Lambda

En la consola de AWS, ve a Lambda > Funciones > Crear función.
Haz clic en Crear desde cero.
Proporciona los siguientes detalles de configuración:

Ajuste Valor

Nombre duo-telephony-logs-collector

Tiempo de ejecución Python 3.13

Arquitectura x86_64

Rol de ejecución duo-telephony-lambda-role

Ajuste	Valor
Nombre	`duo-telephony-logs-collector`
Tiempo de ejecución	Python 3.13
Arquitectura	x86_64
Rol de ejecución	`duo-telephony-lambda-role`

Una vez creada la función, abra la pestaña Código, elimine el stub y pegue el siguiente 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}")

Vaya a Configuración > Variables de entorno.
Haz clic en Editar > Añadir nueva variable de entorno.

Introduce las variables de entorno que se indican en la siguiente tabla y sustituye los valores de ejemplo por los tuyos.

Variables de entorno

Clave	Valor de ejemplo
`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`

Una vez creada la función, permanece en su página (o abre Lambda > Funciones > duo-telephony-logs-collector).
Seleccione la pestaña Configuración.
En el panel Configuración general, haz clic en Editar.
Cambia Tiempo de espera a 5 minutos (300 segundos) y haz clic en Guardar.

Nota: El tiempo de espera es un límite superior estricto, lo que significa que, si la función sigue ejecutándose cuando el temporizador llegue a los 5 minutos, AWS Lambda la finalizará. Si la ejecución es más corta, se completará con normalidad y solo se te cobrará por el tiempo de ejecución real.

Crear una programación de EventBridge

Ve a Amazon EventBridge > Scheduler > Create schedule (Amazon EventBridge > Programador > Crear programación).
Proporcione los siguientes detalles de configuración:
- Programación periódica: Precio (1 hour).
- Destino: tu función Lambda duo-telephony-logs-collector.
- Nombre: duo-telephony-logs-1h.
Haz clic en Crear programación.

(Opcional) Crear un usuario y claves de IAM de solo lectura para Google SecOps

Ve a Consola de AWS > IAM > Usuarios.
Haz clic en Add users (Añadir usuarios).
Proporcione los siguientes detalles de configuración:
- Usuario: introduce secops-reader.
- Tipo de acceso: selecciona Clave de acceso – Acceso programático.
Haz clic en Crear usuario.
Asigna una política de lectura mínima (personalizada): Usuarios > secops-reader > Permisos > Añadir permisos > Asignar políticas directamente > Crear política.

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"
    }
  ]
}

Nombre = secops-reader-policy.
Haz clic en Crear política > busca o selecciona > Siguiente > Añadir permisos.
Crea una clave de acceso para secops-reader: Credenciales de seguridad > Claves de acceso.
Haz clic en Crear clave de acceso.
Descarga la .CSV. (Estos valores se pegarán en el feed).

Configurar un feed en Google SecOps para ingerir registros de telefonía de Duo

Ve a Configuración de SIEM > Feeds.
Haz clic en + Añadir nuevo feed.
En el campo Nombre del feed, introduce un nombre para el feed (por ejemplo, Duo Telephony logs).
Selecciona Amazon S3 V2 como Tipo de fuente.
Selecciona Registros de telefonía de Duo como Tipo de registro.
Haz clic en Siguiente.
Especifique los valores de los siguientes parámetros de entrada:
- URI de S3: s3://duo-telephony-logs/duo-telephony/
- Opciones de eliminación de la fuente: selecciona la opción de eliminación que prefieras.
- Antigüedad máxima del archivo: incluye los archivos modificados en los últimos días. El valor predeterminado es 180 días.
- ID de clave de acceso: clave de acceso de usuario con acceso al bucket de S3.
- Clave de acceso secreta: clave secreta del usuario con acceso al bucket de S3.
- Espacio de nombres de recursos: el espacio de nombres de recursos.
- Etiquetas de ingestión: la etiqueta aplicada a los eventos de este feed.
Haz clic en Siguiente.
Revise la configuración de la nueva fuente en la pantalla Finalizar y, a continuación, haga clic en Enviar.

Tabla de asignación de UDM

Campo de registro	Asignación de UDM	Lógica
`context`	`metadata.product_event_type`	Se asigna directamente desde el campo `context` del registro sin procesar.
`credits`	`security_result.detection_fields.value`	Se asigna directamente desde el campo `credits` del registro sin procesar, anidado en un objeto `detection_fields` con la clave `credits` correspondiente.
`eventtype`	`security_result.detection_fields.value`	Se asigna directamente desde el campo `eventtype` del registro sin procesar, anidado en un objeto `detection_fields` con la clave `eventtype` correspondiente.
`host`	`principal.hostname`	Se asigna directamente desde el campo `host` del registro sin procesar si no es una dirección IP. Se asigna el valor estático "ALLOW" en el analizador. Se asigna el valor estático "MECHANISM_UNSPECIFIED" en el analizador. Se analiza a partir del campo `timestamp` del registro sin procesar, que representa los segundos transcurridos desde el inicio del registro de tiempo. Se asigna el valor "USER_UNCATEGORIZED" si los campos `context` y `host` están presentes en el registro sin procesar. Asigna el valor "STATUS_UPDATE" si solo está presente `host`. De lo contrario, se asigna el valor "GENERIC_EVENT". Se toma directamente del campo `log_type` del registro sin procesar. Se asigna el valor estático "Telephony" en el analizador. Se asigna el valor estático "Duo" en el analizador.
`phone`	`principal.user.phone_numbers`	Se asigna directamente desde el campo `phone` del registro sin procesar.
`phone`	`principal.user.userid`	Se asigna directamente desde el campo `phone` del registro sin procesar. Se define como el valor estático "INFORMATIONAL" en el analizador. Se asigna el valor estático "Duo Telephony" en el analizador.
`timestamp`	`metadata.event_timestamp`	Se analiza a partir del campo `timestamp` del registro sin procesar, que representa los segundos transcurridos desde el inicio del registro de tiempo.
`type`	`security_result.summary`	Se asigna directamente desde el campo `type` del registro sin procesar.

¿Necesitas más ayuda? Recibe respuestas de los miembros de la comunidad y de los profesionales de Google SecOps.