Recoger registros de autenticación de Duo
En este documento se explica cómo ingerir registros de autenticación de Duo en Google Security Operations. 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, al tiempo que gestiona varios factores y resultados de autenticación para clasificar los eventos de seguridad. El analizador también limpia los datos, convierte los tipos y gestiona los errores para asegurar la calidad y la coherencia de los datos.
Elige uno de los dos métodos de recogida:
- Opción 1: Ingestión directa mediante una API de terceros
- Opción 2: Recoger registros con AWS Lambda y Amazon S3
Antes de empezar
- Instancia de Google SecOps
- Acceso privilegiado al panel de administración de Duo (se requiere el rol de propietario para crear aplicaciones de la API Admin)
- Acceso privilegiado a AWS si se usa la opción 2
Opción 1: Ingerir registros de autenticación de Duo mediante una API de terceros
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, Administrador o Gestor de aplicaciones.
- 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-XXXXXXXX.duosecurity.com)
- Ve a la sección Permisos.
- Deselecciona todas las opciones de permisos, excepto Conceder registro de lectura.
- Haz clic en Guardar cambios.
Configurar un feed en Google SecOps para ingerir registros de autenticación 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 Authentication Logs). - Seleccione API de terceros como Tipo de fuente.
- Seleccione Autenticación de Duo como Tipo de registro.
- Haz clic en Siguiente.
- Especifique los valores de los siguientes parámetros de entrada:
- Nombre de usuario: introduce la clave de integración de Duo.
- Secreto: introduce la clave secreta de Duo.
- Nombre de host de la API: introduce el nombre de host de la API (por ejemplo,
api-XXXXXXXX.duosecurity.com). - Espacio de nombres de recursos: opcional. El espacio de nombres de recursos.
- Etiquetas de ingestión: opcional. Etiqueta que se aplicará 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.
Opción 2: Ingerir registros de autenticación de Duo mediante AWS S3
Recoger las credenciales de la API Admin de Duo
- Inicia sesión en el panel de administración de Duo.
- Ve a Aplicaciones > Proteger una aplicación.
- Busca API Admin en el catálogo de aplicaciones.
- Haz clic en Proteger para añadir la aplicación de la API Admin.
- Copia y guarda los siguientes valores:
- Clave de integración (ikey)
- Clave secreta (skey)
- Nombre de host de la API (por ejemplo,
api-XXXXXXXX.duosecurity.com)
- En Permisos, habilita 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-auth-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 y selecciona la política AmazonS3FullAccess.
- 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, vaya a IAM > Policies > Create policy > JSON tab (IAM > Políticas > Crear política > pestaña JSON).
Introduce 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" } ] }- Sustituye
duo-auth-logssi has introducido otro nombre de segmento.
- Sustituye
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
WriteDuoAuthToS3Roleal rol y haz clic en Crear rol.
Crear la función Lambda
- En la consola de AWS, ve a Lambda > Funciones.
- Haga clic en Crear función > Crear desde cero.
Proporciona los siguientes detalles de configuración:
Ajuste Valor Nombre duo_auth_to_s3Tiempo de ejecución Python 3.13 Arquitectura x86_64 Rol de ejecución WriteDuoAuthToS3RoleUna vez creada la función, abra la pestaña Código, elimine el stub e introduzca 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())Vaya a Configuración > Variables de entorno.
Haz clic en Editar > Añadir nueva variable de entorno.
Introduce las siguientes variables de entorno y sustituye los valores por los tuyos.
Clave Valor de ejemplo S3_BUCKETduo-auth-logsS3_PREFIXduo/auth/STATE_KEYduo/auth/state.jsonDUO_IKEYDIXYZ...DUO_SKEY****************DUO_API_HOSTNAMEapi-XXXXXXXX.duosecurity.comLIMIT500Una vez creada la función, permanece en su página (o abre Lambda > Funciones > tu-función).
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.
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_auth_to_s3. - Nombre:
duo-auth-1h.
- Programación periódica: Precio (
- Haz clic en Crear programación.
Crear un usuario y claves de IAM de solo lectura para Google SecOps
- En la consola de AWS, ve a IAM > Usuarios > Añadir usuarios.
- Haz clic en Add users (Añadir usuarios).
- Proporcione los siguientes detalles de configuración:
- Usuario:
secops-reader - Tipo de acceso: clave de acceso (acceso programático)
- Usuario:
- 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.
En el editor de JSON, introduce la siguiente política:
{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": ["s3:GetObject"], "Resource": "arn:aws:s3:::duo-auth-logs/*" }, { "Effect": "Allow", "Action": ["s3:ListBucket"], "Resource": "arn:aws:s3:::duo-auth-logs" } ] }Asigna el nombre
secops-reader-policy.Ve a Crear política > busca o selecciona > Siguiente > Añadir permisos.
Ve a Credenciales de seguridad > Claves de acceso > Crear clave de acceso.
Descarga el archivo CSV (estos valores se introducen en el feed).
Configurar un feed en Google SecOps para ingerir registros de autenticación 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 Authentication Logs). - Selecciona Amazon S3 V2 como Tipo de fuente.
- Seleccione Autenticación 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-auth-logs/duo/auth/ - 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.
- URI de S3:
- 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 |
|---|---|---|
access_device.browser |
target.resource.attribute.labels.value |
Si access_device.browser está presente, su valor se asigna al modelo de datos unificado. |
access_device.hostname |
principal.hostname |
Si access_device.hostname está presente y no está vacío, su valor se asigna al UDM. Si está vacío y event_type es USER_CREATION, event_type cambia a USER_UNCATEGORIZED. Si access_device.hostname está vacío y existe el campo hostname, se usa el valor de hostname. |
access_device.ip |
principal.ip |
Si access_device.ip existe y es una dirección IPv4 válida, su valor se asigna al UDM. Si no es una dirección IPv4 válida, se añade como valor de cadena a additional.fields con la clave access_device.ip. |
access_device.location.city |
principal.location.city |
Si está presente, el valor se asigna al UDM. |
access_device.location.country |
principal.location.country_or_region |
Si está presente, el valor se asigna al UDM. |
access_device.location.state |
principal.location.state |
Si está presente, el valor se asigna al UDM. |
access_device.os |
principal.platform |
Si está presente, el valor se traduce al valor de UDM correspondiente (MAC, WINDOWS o LINUX). |
access_device.os_version |
principal.platform_version |
Si está presente, el valor se asigna al UDM. |
application.key |
target.resource.id |
Si está presente, el valor se asigna al UDM. |
application.name |
target.application |
Si está presente, el valor se asigna al UDM. |
auth_device.ip |
target.ip |
Si está presente y no es "None", el valor se asigna al UDM. |
auth_device.location.city |
target.location.city |
Si está presente, el valor se asigna al UDM. |
auth_device.location.country |
target.location.country_or_region |
Si está presente, el valor se asigna al UDM. |
auth_device.location.state |
target.location.state |
Si está presente, el valor se asigna al UDM. |
auth_device.name |
target.hostname O target.user.phone_numbers |
Si auth_device.name está presente y es un número de teléfono (después de la normalización), se añade a target.user.phone_numbers. De lo contrario, se asigna a target.hostname. |
client_ip |
target.ip |
Si está presente y no es "None", el valor se asigna al UDM. |
client_section |
target.resource.attribute.labels.value |
Si client_section está presente, su valor se asigna al UDM con la clave client_section. |
dn |
target.user.userid |
Si dn está presente y user.name y username no lo están, se extrae userid del campo dn mediante grok y se asigna al modelo de datos unificado. El valor de event_type es USER_LOGIN. |
event_type |
metadata.product_event_type Y metadata.event_type |
El valor se asigna a metadata.product_event_type. También se usa para determinar el metadata.event_type: "authentication" se convierte en USER_LOGIN, "enrollment" se convierte en USER_CREATION y, si está vacío o no es ninguno de los dos, se convierte en GENERIC_EVENT. |
factor |
extensions.auth.mechanism Y extensions.auth.auth_details |
El valor se traduce al valor de UDM auth.mechanism correspondiente (HARDWARE_KEY, REMOTE_INTERACTIVE, LOCAL, OTP). El valor original también se asigna a extensions.auth.auth_details. |
hostname |
principal.hostname |
Si está presente y access_device.hostname está vacío, el valor se asigna al modelo de datos unificado. |
log_format |
target.resource.attribute.labels.value |
Si log_format está presente, su valor se asigna al UDM con la clave log_format. |
log_level.__class_uuid__ |
target.resource.attribute.labels.value |
Si log_level.__class_uuid__ está presente, su valor se asigna al UDM con la clave __class_uuid__. |
log_level.name |
target.resource.attribute.labels.value Y security_result.severity |
Si log_level.name está presente, su valor se asigna al UDM con la clave name. Si el valor es "info", security_result.severity se define como INFORMATIONAL. |
log_logger.unpersistable |
target.resource.attribute.labels.value |
Si log_logger.unpersistable está presente, su valor se asigna al UDM con la clave unpersistable. |
log_namespace |
target.resource.attribute.labels.value |
Si log_namespace está presente, su valor se asigna al UDM con la clave log_namespace. |
log_source |
target.resource.attribute.labels.value |
Si log_source está presente, su valor se asigna al UDM con la clave log_source. |
msg |
security_result.summary |
Si está presente y reason está vacío, el valor se asigna al modelo de datos unificado. |
reason |
security_result.summary |
Si está presente, el valor se asigna al UDM. |
result |
security_result.action_details Y security_result.action |
Si está presente, el valor se asigna a security_result.action_details. "success" o "SUCCESS" se traduce como security_result.action ALLOW; de lo contrario, se traduce como BLOCK. |
server_section |
target.resource.attribute.labels.value |
Si server_section está presente, su valor se asigna al UDM con la clave server_section. |
server_section_ikey |
target.resource.attribute.labels.value |
Si server_section_ikey está presente, su valor se asigna al UDM con la clave server_section_ikey. |
status |
security_result.action_details Y security_result.action |
Si está presente, el valor se asigna a security_result.action_details. "Permitir" se traduce como security_result.action ALLOW y "Rechazar" como BLOCK. |
timestamp |
metadata.event_timestamp Y event.timestamp |
El valor se convierte en una marca de tiempo y se asigna a metadata.event_timestamp y event.timestamp. |
txid |
metadata.product_log_id Y network.session_id |
El valor se asigna tanto a metadata.product_log_id como a network.session_id. |
user.groups |
target.user.group_identifiers |
Todos los valores del array se añaden a target.user.group_identifiers. |
user.key |
target.user.product_object_id |
Si está presente, el valor se asigna al UDM. |
user.name |
target.user.userid |
Si está presente, el valor se asigna al UDM. |
username |
target.user.userid |
Si está presente y user.name no lo está, el valor se asigna al UDM. El valor de event_type es USER_LOGIN. |
| (Lógica del analizador) | metadata.vendor_name |
Siempre se le asigna el valor "DUO_SECURITY". |
| (Lógica del analizador) | metadata.product_name |
Siempre se le asigna el valor "MULTI-FACTOR_AUTHENTICATION". |
| (Lógica del analizador) | metadata.log_type |
Se toma del campo log_type de nivel superior del registro sin procesar. |
| (Lógica del analizador) | extensions.auth.type |
Siempre se le asigna el valor "SSO". |
¿Necesitas más ayuda? Recibe respuestas de los miembros de la comunidad y de los profesionales de Google SecOps.