Recopila registros de autenticación de Duo

En este documento, se explica cómo transferir registros de autenticación de Duo a las Operaciones de seguridad de Google con Amazon S3. El analizador extrae los registros de los mensajes con formato JSON. Transforma los datos de registro sin procesar en el Modelo de datos unificado (UDM), asignando campos como el usuario, el dispositivo, la aplicación, la ubicación y los detalles de autenticación, y, al mismo tiempo, controla varios factores y resultados de autenticación para categorizar los eventos de seguridad. El analizador también realiza la limpieza de datos, la conversión de tipos y el manejo de errores para garantizar la calidad y la coherencia de los datos.

Antes de comenzar

• Instancia de Google SecOps
• Acceso privilegiado al arrendatario de Duo (aplicación de la API de Admin)
• Acceso con privilegios a AWS (S3, IAM, Lambda, EventBridge)

Configura la aplicación de la API de Duo Admin

1. Accede al Panel de administración de Duo.
2. Ve a Aplicaciones > Proteger una aplicación.
3. Agrega la aplicación de la API de Admin.
4. Copia y guarda los siguientes valores en una ubicación segura:
   • Clave de integración (ikey)
   • Clave secreta (skey)
   • Nombre de host de la API (por ejemplo, api-XXXXXXXX.duosecurity.com)
5. En Permissions, habilita Grant read log (para leer los registros de autenticación).
6. Guarda la aplicación.

Configura el bucket de AWS S3 y el IAM para Google SecOps

1. Crea un bucket de Amazon S3 siguiendo esta guía del usuario: Crea un bucket
2. Guarda el Nombre y la Región del bucket para futuras referencias (por ejemplo, duo-auth-logs).
3. Crea un usuario siguiendo esta guía del usuario: Cómo crear un usuario de IAM.
4. Selecciona el usuario creado.
5. Selecciona la pestaña Credenciales de seguridad.
6. Haz clic en Crear clave de acceso en la sección Claves de acceso.
7. Selecciona Servicio de terceros como el Caso de uso.
8. Haz clic en Siguiente.
9. Opcional: Agrega una etiqueta de descripción.
10. Haz clic en Crear clave de acceso.
11. Haz clic en Descargar archivo CSV para guardar la clave de acceso y la clave de acceso secreta para usarlas más adelante.
12. Haz clic en Listo.
13. Selecciona la pestaña Permisos.
14. Haz clic en Agregar permisos en la sección Políticas de permisos.
15. Selecciona Agregar permisos.
16. Selecciona Adjuntar políticas directamente.
17. Busca y selecciona la política AmazonS3FullAccess.
18. Haz clic en Siguiente.
19. Haz clic en Agregar permisos.

Configura la política y el rol de IAM para las cargas de S3

1. Ve a la consola de AWS > IAM > Políticas > Crear política > pestaña JSON.

2. Ingresa la siguiente política:

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

   • Reemplaza duo-auth-logs si ingresaste un nombre de bucket diferente.

3. Haz clic en Siguiente > Crear política.

4. Ve a IAM > Roles > Crear rol > Servicio de AWS > Lambda.

5. Adjunta la política recién creada.

6. Asigna el nombre WriteDuoAuthToS3Role al rol y haz clic en Crear rol.

Crea la función Lambda

1. En la consola de AWS, ve a Lambda > Functions > Create function.
2. Haz clic en Crear desde cero.

3. Proporciona los siguientes detalles de configuración:

   Configuración	Valor
   Nombre	duo_auth_to_s3
   Tiempo de ejecución	Python 3.13
   Arquitectura	x86_64
   Rol de ejecución	WriteDuoAuthToS3Role

4. Después de crear la función, abre la pestaña Code, borra el código auxiliar y, luego, ingresa el siguiente código (duo_auth_to_s3.py):

   #!/usr/bin/env python3
   # Lambda: Pull Duo Admin API v2 Authentication Logs to S3 (raw JSON pages)
   # Notes:
   # - Duo v2 requires mintime/maxtime in *milliseconds* (13-digit epoch).
   # - Pagination via metadata.next_offset ("<millis>,<txid>").
   # - We save state (mintime_ms) in ms to resume next run without gaps.
   
   import os, json, time, hmac, hashlib, base64, email.utils, urllib.parse
   from urllib.request import Request, urlopen
   from urllib.error import HTTPError, URLError
   import boto3
   
   DUO_IKEY = os.environ["DUO_IKEY"]
   DUO_SKEY = os.environ["DUO_SKEY"]
   DUO_API_HOSTNAME = os.environ["DUO_API_HOSTNAME"].strip()
   S3_BUCKET = os.environ["S3_BUCKET"]
   S3_PREFIX = os.environ.get("S3_PREFIX", "duo/auth/").strip("/")
   STATE_KEY = os.environ.get("STATE_KEY", "duo/auth/state.json")
   LIMIT = min(int(os.environ.get("LIMIT", "500")), 1000)  # default 100, max 1000
   
   s3 = boto3.client("s3")
   
   def _canon_params(params: dict) -> str:
       parts = []
       for k in sorted(params.keys()):
           v = params[k]
           if v is None:
               continue
           parts.append(f"{urllib.parse.quote(str(k), '~')}={urllib.parse.quote(str(v), '~')}")
       return "&".join(parts)
   
   def _sign(method: str, host: str, path: str, params: dict) -> dict:
       now = email.utils.formatdate()
       canon = "\n".join([now, method.upper(), host.lower(), path, _canon_params(params)])
       sig = hmac.new(DUO_SKEY.encode("utf-8"), canon.encode("utf-8"), hashlib.sha1).hexdigest()
       auth = base64.b64encode(f"{DUO_IKEY}:{sig}".encode()).decode()
       return {"Date": now, "Authorization": f"Basic {auth}"}
   
   def _http(method: str, path: str, params: dict, timeout: int = 60, max_retries: int = 5) -> dict:
       host = DUO_API_HOSTNAME
       assert host.startswith("api-") and host.endswith(".duosecurity.com"), \
           "DUO_API_HOSTNAME must be like api-XXXXXXXX.duosecurity.com"
       qs = _canon_params(params)
       url = f"https://{host}{path}" + (f"?{qs}" if qs else "")
   
       attempt, backoff = 0, 1.0
       while True:
           req = Request(url, method=method.upper())
           req.add_header("Accept", "application/json")
           for k, v in _sign(method, host, path, params).items():
               req.add_header(k, v)
           try:
               with urlopen(req, timeout=timeout) as r:
                   return json.loads(r.read().decode("utf-8"))
           except HTTPError as e:
               if (e.code == 429 or 500 <= e.code <= 599) and attempt < max_retries:
                   time.sleep(backoff); attempt += 1; backoff *= 2; continue
               raise
           except URLError:
               if attempt < max_retries:
                   time.sleep(backoff); attempt += 1; backoff *= 2; continue
               raise
   
   def _read_state_ms() -> int | None:
       try:
           obj = s3.get_object(Bucket=S3_BUCKET, Key=STATE_KEY)
           val = json.loads(obj["Body"].read()).get("mintime")
           if val is None:
               return None
           # Backward safety: if seconds were stored, convert to ms
           return int(val) * 1000 if len(str(int(val))) <= 10 else int(val)
       except Exception:
           return None
   
   def _write_state_ms(mintime_ms: int):
       body = json.dumps({"mintime": int(mintime_ms)}).encode("utf-8")
       s3.put_object(Bucket=S3_BUCKET, Key=STATE_KEY, Body=body, ContentType="application/json")
   
   def _write_page(payload: dict, when_epoch_s: int, page: int) -> str:
       key = f"{S3_PREFIX}/{time.strftime('%Y/%m/%d', time.gmtime(when_epoch_s))}/duo-auth-{page:05d}.json"
       s3.put_object(
           Bucket=S3_BUCKET,
           Key=key,
           Body=json.dumps(payload, separators=(",", ":")).encode("utf-8"),
           ContentType="application/json",
       )
       return key
   
   def fetch_and_store():
       now_s = int(time.time())
       # Duo recommends a ~2-minute delay buffer; use maxtime = now - 120 seconds (in ms)
       maxtime_ms = (now_s - 120) * 1000
       mintime_ms = _read_state_ms() or (maxtime_ms - 3600 * 1000)  # 1 hour on first run
   
       page = 0
       total = 0
       next_offset = None
   
       while True:
           params = {"mintime": mintime_ms, "maxtime": maxtime_ms, "limit": LIMIT}
           if next_offset:
               params["next_offset"] = next_offset
   
           data = _http("GET", "/admin/v2/logs/authentication", params)
           _write_page(data, maxtime_ms // 1000, page)
           page += 1
   
           resp = data.get("response")
           items = resp if isinstance(resp, list) else []
           total += len(items)
   
           meta = data.get("metadata") or {}
           next_offset = meta.get("next_offset")
           if not next_offset:
               break
   
       # Advance window to maxtime_ms for next run
       _write_state_ms(maxtime_ms)
       return {"ok": True, "pages": page, "events": total, "next_mintime_ms": maxtime_ms}
   
   def lambda_handler(event=None, context=None):
       return fetch_and_store()
   
   if __name__ == "__main__":
       print(lambda_handler())
   
   

5. Ve a Configuración > Variables de entorno > Editar > Agregar nueva variable de entorno.

6. Ingresa las siguientes variables de entorno proporcionadas y reemplázalas por tus valores:

   Clave	Ejemplo
   S3_BUCKET	duo-auth-logs
   S3_PREFIX	duo/auth/
   STATE_KEY	duo/auth/state.json
   DUO_IKEY	DIXYZ...
   DUO_SKEY	****************
   DUO_API_HOSTNAME	api-XXXXXXXX.duosecurity.com
   LIMIT	500

7. Después de crear la función, permanece en su página (o abre Lambda > Functions > tu‑función).

8. Selecciona la pestaña Configuración.

9. En el panel Configuración general, haz clic en Editar.

10. Cambia Tiempo de espera a 5 minutos (300 segundos) y haz clic en Guardar.

Crea una programación de EventBridge

1. Ve a Amazon EventBridge > Scheduler > Create schedule.
2. Proporciona los siguientes detalles de configuración:
   • Programación recurrente: Frecuencia (1 hour).
   • Destino: Tu función Lambda.
   • Nombre: duo-auth-1h.
3. Haz clic en Crear programación.

Opcional: Crea un usuario y claves de IAM de solo lectura para Google SecOps

1. En la consola de AWS, ve a IAM > Usuarios y, luego, haz clic en Agregar usuarios.
2. Proporciona los siguientes detalles de configuración:
   • Usuario: Ingresa un nombre único (por ejemplo, secops-reader).
   • Tipo de acceso: Selecciona Clave de acceso: Acceso programático.
   • Haz clic en Crear usuario.
3. Adjunta una política de lectura mínima (personalizada): Usuarios > selecciona secops-reader > Permisos > Agregar permisos > Adjuntar políticas directamente > Crear política

4. En el editor de JSON, ingresa la siguiente política:

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Effect": "Allow",
         "Action": ["s3:GetObject"],
         "Resource": "arn:aws:s3:::<your-bucket>/*"
       },
       {
         "Effect": "Allow",
         "Action": ["s3:ListBucket"],
         "Resource": "arn:aws:s3:::<your-bucket>"
       }
     ]
   }
   

5. Configura el nombre como secops-reader-policy.

6. Ve a Crear política > busca o selecciona > Siguiente > Agregar permisos.

7. Ve a Credenciales de seguridad > Claves de acceso > Crear clave de acceso.

8. Descarga el archivo CSV (estos valores se ingresan en el feed).

Configura un feed en Google SecOps para transferir registros de autenticación de Duo

1. Ve a SIEM Settings > Feeds.
2. Haz clic en + Agregar feed nuevo.
3. En el campo Nombre del feed, ingresa un nombre para el feed (por ejemplo, Duo Authentication Logs).
4. Selecciona Amazon S3 V2 como el Tipo de fuente.
5. Selecciona Duo Auth como el Tipo de registro.
6. Haz clic en Siguiente.
7. Especifica valores para los siguientes parámetros de entrada:
   • URI de S3: s3://duo-auth-logs/duo/auth/
   • Opciones de borrado de la fuente: Selecciona la opción de borrado según tu preferencia.
   • Antigüedad máxima del archivo: 180 días de forma predeterminada
   • ID de clave de acceso: Clave de acceso del usuario con acceso al bucket de S3.
   • Clave de acceso secreta: Clave secreta del usuario con acceso al bucket de S3.
   • Espacio de nombres del recurso: Es el espacio de nombres del recurso.
   • Etiquetas de transferencia: Es la etiqueta que se aplica a los eventos de este feed.
8. Haz clic en Siguiente.
9. Revisa la nueva configuración del feed en la pantalla Finalizar y, luego, haz clic en Enviar.

Tabla de asignación de UDM

¿Necesitas más ayuda? Obtén respuestas de miembros de la comunidad y profesionales de Google SecOps.