Recopila registros de Akamai Cloud Monitor
En este documento, se explica cómo transferir registros de Akamai Cloud Monitor (balanceador de cargas, Traffic Shaper y ADC) a Google Security Operations con AWS S3. Akamai envía eventos JSON a tu extremo HTTPS; un receptor de API Gateway + Lambda escribe los eventos en S3 (JSONL, gz). El analizador transforma los registros JSON en UDM. Extrae campos de la carga útil JSON, realiza conversiones de tipos de datos, cambia el nombre de los campos para que coincidan con el esquema del UDM y controla la lógica específica para los campos personalizados y la construcción de URLs. También incorpora el manejo de errores y la lógica condicional según la presencia de campos.
Antes de comenzar
Asegúrate de cumplir con los siguientes requisitos previos:
- Instancia de Google SecOps
- Acceso con privilegios a Akamai Control Center y Property Manager
- Acceso privilegiado a AWS*(S3, IAM, Lambda, API Gateway)
Configura el bucket de AWS S3 y el IAM para Google SecOps
- Crea un bucket de Amazon S3 siguiendo esta guía del usuario: Crea un bucket
- Guarda el Nombre y la Región del bucket para futuras referencias (por ejemplo,
akamai-cloud-monitor
). - Crea un usuario siguiendo esta guía del usuario: Cómo crear un usuario de IAM.
- Selecciona el usuario creado.
- Selecciona la pestaña Credenciales de seguridad.
- Haz clic en Crear clave de acceso en la sección Claves de acceso.
- Selecciona Servicio de terceros como el Caso de uso.
- Haz clic en Siguiente.
- Opcional: Agrega 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 usarlas más adelante.
- Haz clic en Listo.
- Selecciona la pestaña Permisos.
- Haz clic en Agregar permisos en la sección Políticas de permisos.
- Selecciona Agregar permisos.
- Selecciona Adjuntar políticas directamente.
- Busca y selecciona la política AmazonS3FullAccess.
- Haz clic en Siguiente.
- Haz clic en Agregar permisos.
Configura la política y el rol de IAM para las cargas de S3 (Lambda)
- En la consola de AWS, ve a IAM > Políticas > Crear política > JSON y pega la siguiente política.
Política de JSON (reemplaza
akamai-cloud-monitor
por el nombre de tu bucket de S3):{ "Version": "2012-10-17", "Statement": [ { "Sid": "AllowPutAkamaiObjects", "Effect": "Allow", "Action": ["s3:PutObject"], "Resource": "arn:aws:s3:::akamai-cloud-monitor/*" } ] }
Haz clic en Siguiente > Crear política.
Ve a IAM > Roles > Crear rol > Servicio de AWS > Lambda.
Adjunta la política en formato JSON.
Asigna el nombre
WriteAkamaiCMToS3Role
al rol y haz clic en Crear rol.
Crea la función Lambda
Configuración | Valor |
---|---|
Nombre | akamai_cloud_monitor_to_s3 |
Tiempo de ejecución | Python 3.13 |
Arquitectura | x86_64 |
Rol de ejecución | WriteAkamaiCMToS3Role |
Después de crear la función, abre la pestaña Code, borra el código auxiliar y, luego, ingresa el siguiente código (
akamai_cloud_monitor_to_s3.py
):#!/usr/bin/env python3 # Lambda: Receive Akamai Cloud Monitor POST, write JSONL (gz) to S3 import os, json, gzip, io, uuid, base64, datetime as dt import boto3 S3_BUCKET = os.environ["S3_BUCKET_NAME"] S3_PREFIX = os.environ.get("S3_PREFIX", "akamai/cloud-monitor/json/").strip("/") + "/" INGEST_TOKEN = os.environ.get("INGEST_TOKEN") # optional shared secret in URL query (?token=...) s3 = boto3.client("s3") def _write_jsonl_gz(objs: list) -> str: key = f"{dt.datetime.utcnow():%Y/%m/%d}/akamai-cloud-monitor-{uuid.uuid4()}.json.gz" buf = io.BytesIO() with gzip.GzipFile(fileobj=buf, mode="w") as gz: for o in objs: gz.write((json.dumps(o, separators=(",", ":")) + "n").encode()) buf.seek(0) s3.upload_fileobj( buf, S3_BUCKET, f"{S3_PREFIX}{key}", ExtraArgs={ "ContentType": "application/json", "ContentEncoding": "gzip", }, ) return f"s3://{S3_BUCKET}/{S3_PREFIX}{key}" def _parse_records_from_event(event) -> list: # HTTP API (Lambda proxy) event: body is a JSON string body = event.get("body") or "" if event.get("isBase64Encoded"): body = base64.b64decode(body).decode("utf-8", "replace") try: data = json.loads(body) except Exception: # accept line-delimited JSON as pass-through try: return [json.loads(line) for line in body.splitlines() if line.strip()] except Exception: return [] if isinstance(data, list): return data if isinstance(data, dict): return [data] return [] def lambda_handler(event, context=None): # Optional shared-secret verification via query parameter (?token=...) if INGEST_TOKEN: qs = event.get("queryStringParameters") or {} token = qs.get("token") if token != INGEST_TOKEN: return {"statusCode": 403, "body": "forbidden"} records = _parse_records_from_event(event) if not records: return {"statusCode": 204, "body": "no content"} key = _write_jsonl_gz(records) return { "statusCode": 200, "headers": {"Content-Type": "application/json"}, "body": json.dumps({"ok": True, "s3_key": key, "count": len(records)}), }
Ve a Configuración > Variables de entorno > Editar.
Haz clic en Agregar nueva variable de entorno y establece los siguientes valores:
Variables de entorno
Clave Ejemplo S3_BUCKET_NAME
akamai-cloud-monitor
S3_PREFIX
akamai/cloud-monitor/json/
INGEST_TOKEN
random-shared-secret
Ve a Configuración > Configuración general.
Haz clic en Editar y establece el Tiempo de espera en 5 minutos (300 segundos).
Haz clic en Guardar.
Crea Amazon API Gateway (extremo HTTPS para Akamai)
- En la consola de AWS, ve a API Gateway > Create API.
- Selecciona HTTP API > Build.
- Proporciona los siguientes detalles de configuración:
- Integraciones: Elige Lambda y selecciona
akamai_cloud_monitor_to_s3
. - Rutas: Agrega ANY
/{proxy+}
o crea una ruta específica (por ejemplo, POST/akamai/cloud-monitor
). - Etapas: Crea o usa $default.
- Integraciones: Elige Lambda y selecciona
- Implementa la API y copia la URL de invocación (por ejemplo,
https://abc123.execute-api.<region>.amazonaws.com
).
Configura Akamai Cloud Monitor para enviar registros
- En Akamai Control Center, abre tu propiedad en Property Manager.
- Haz clic en Agregar regla > elige Administración de Cloud.
- Agrega Cloud Monitor Instrumentation y selecciona los conjuntos de datos requeridos.
- Agrega Cloud Monitor Data Delivery.
- Nombre de host de entrega: Ingresa la URL de invocación de API Gateway (por ejemplo,
abc123.execute-api.<region>.amazonaws.com
). - Ruta de URL de entrega: Tu ruta más un token de consulta opcional, por ejemplo:
/akamai/cloud-monitor?token=<INGEST_TOKEN>
.
- Nombre de host de entrega: Ingresa la URL de invocación de API Gateway (por ejemplo,
- Guarda y activa la versión de la propiedad.
Configura un feed en Google SecOps para transferir datos de Akamai Cloud Monitor (JSON de S3)
- Ve a Configuración de SIEM > Feeds.
- Haz clic en Agregar feed nuevo.
- En el campo Nombre del feed, ingresa un nombre para el feed (por ejemplo,
Akamai Cloud Monitor — S3
). - Selecciona Amazon S3 V2 como el Tipo de fuente.
- Selecciona Akamai Cloud Monitor como el Tipo de registro.
- Haz clic en Siguiente.
- Especifica valores para los siguientes parámetros de entrada:
- URI de S3:
s3://akamai-cloud-monitor/akamai/cloud-monitor/json/
- Opciones de eliminación de la fuente: Indica si se deben borrar los archivos o los directorios después de la transferencia.
- Antigüedad máxima del archivo: Incluye los archivos modificados en la cantidad de días más reciente. El valor predeterminado es de 180 días.
- ID de clave de acceso: Es una clave de acceso alfanumérica de 20 caracteres (p.ej., AKIAIOSFODNN7EXAMPLE).
- Clave de acceso secreta: Es una clave de acceso secreta alfanumérica de 40 caracteres (p.ej., wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY).
- Espacio de nombres del activo:
akamai.cloud_monitor
- Etiquetas de transferencia: Se agregan etiquetas a todos los eventos de este feed (por ejemplo,
source=akamai_cloud_monitor
,format=json
).
- URI de S3:
- Haz clic en Siguiente.
- Revisa la nueva configuración del feed en la pantalla Finalizar y, luego, haz clic en Enviar.
Tabla de asignación de UDM
Campo de registro | Asignación de UDM | Lógica |
---|---|---|
accLang |
network.http.user_agent |
Se asigna directamente si no es "-" o una cadena vacía. |
city |
principal.location.city |
Se asigna directamente si no es "-" o una cadena vacía. |
cliIP |
principal.ip |
Se asigna directamente si no es una cadena vacía. |
country |
principal.location.country_or_region |
Se asigna directamente si no es "-" o una cadena vacía. |
cp |
additional.fields |
Se asigna como un par clave-valor con la clave "cp". |
customField |
about.ip , about.labels , src.ip |
Se analizan como pares clave-valor. Se realiza un tratamiento especial para "eIp" y "pIp" para asignarlos a src.ip y about.ip , respectivamente. Otras claves se asignan como etiquetas dentro de about . |
errorCode |
security_result.summary , security_result.severity |
Si está presente, establece security_result.severity en "ERROR" y asigna el valor a security_result.summary . |
geo.city |
principal.location.city |
Se asigna directamente si city es "-" o una cadena vacía. |
geo.country |
principal.location.country_or_region |
Se asigna directamente si country es "-" o una cadena vacía. |
geo.lat |
principal.location.region_latitude |
Se asigna directamente y se convierte en un valor de coma flotante. |
geo.long |
principal.location.region_longitude |
Se asigna directamente y se convierte en un valor de coma flotante. |
geo.region |
principal.location.state |
Se asigna directamente. |
id |
metadata.product_log_id |
Se asigna directamente si no es una cadena vacía. |
message.cliIP |
principal.ip |
Se asigna directamente si cliIP es una cadena vacía. |
message.fwdHost |
principal.hostname |
Se asigna directamente. |
message.reqHost |
target.hostname , target.url |
Se usa para construir target.url y extraer target.hostname . |
message.reqLen |
network.sent_bytes |
Se asigna directamente y se convierte en un número entero sin signo si totalBytes está vacío o es "-". |
message.reqMethod |
network.http.method |
Se asigna directamente si reqMethod es una cadena vacía. |
message.reqPath |
target.url |
Se agregó a target.url . |
message.reqPort |
target.port |
Se asigna directamente y se convierte en un número entero si reqPort es una cadena vacía. |
message.respLen |
network.received_bytes |
Se asigna directamente y se convierte en un número entero sin signo. |
message.sslVer |
network.tls.version |
Se asigna directamente. |
message.status |
network.http.response_code |
Se asigna directamente y se convierte en un número entero si statusCode está vacío o es "-". |
message.UA |
network.http.user_agent |
Se asigna directamente si UA es "-" o una cadena vacía. |
network.asnum |
additional.fields |
Se asigna como un par clave-valor con la clave "asnum". |
network.edgeIP |
intermediary.ip |
Se asigna directamente. |
network.network |
additional.fields |
Se asigna como un par clave-valor con la clave "network". |
network.networkType |
additional.fields |
Se asigna como un par clave-valor con la clave "networkType". |
proto |
network.application_protocol |
Se usa para determinar network.application_protocol . |
queryStr |
target.url |
Se agrega a target.url si no es "-" o una cadena vacía. |
referer |
network.http.referral_url , about.hostname |
Se asigna directamente si no es "-". El nombre de host extraído se asigna a about.hostname . |
reqHost |
target.hostname , target.url |
Se usa para construir target.url y extraer target.hostname . |
reqId |
metadata.product_log_id , network.session_id |
Se asigna directamente si id es una cadena vacía. También se asigna a network.session_id . |
reqMethod |
network.http.method |
Se asigna directamente si no es una cadena vacía. |
reqPath |
target.url |
Se agrega a target.url si no es "-". |
reqPort |
target.port |
Se asigna directamente y se convierte en un número entero. |
reqTimeSec |
metadata.event_timestamp , timestamp |
Se usa para establecer la marca de tiempo del evento. |
start |
metadata.event_timestamp , timestamp |
Se usa para establecer la marca de tiempo del evento si reqTimeSec es una cadena vacía. |
statusCode |
network.http.response_code |
Se asigna directamente y se convierte en un número entero si no es "-" o una cadena vacía. |
tlsVersion |
network.tls.version |
Se asigna directamente. |
totalBytes |
network.sent_bytes |
Se asigna directamente y se convierte en un número entero sin signo si no está vacío o es "-". |
type |
metadata.product_event_type |
Se asigna directamente. |
UA |
network.http.user_agent |
Se asigna directamente si no es "-" o una cadena vacía. |
version |
metadata.product_version |
Se asigna directamente. |
xForwardedFor |
principal.ip |
Se asigna directamente si no es "-" o una cadena vacía. |
(Lógica del analizador) | metadata.vendor_name |
Se establece en "Akamai". |
(Lógica del analizador) | metadata.product_name |
Se establece en "DataStream". |
(Lógica del analizador) | metadata.event_type |
Se establece en "NETWORK_HTTP". |
(Lógica del analizador) | metadata.product_version |
Se establece en "2" si version es una cadena vacía. |
(Lógica del analizador) | metadata.log_type |
Se debe establecer en "AKAMAI_CLOUD_MONITOR". |
(Lógica del analizador) | network.application_protocol |
Se determina a partir de proto o message.proto . Se establece en "HTTPS" si alguno contiene "HTTPS" (sin distinción entre mayúsculas y minúsculas) o en "HTTP" en cualquier otro caso. |
(Lógica del analizador) | security_result.severity |
Se establece en "INFORMATIONAL" si errorCode es "-" o una cadena vacía. |
(Lógica del analizador) | target.url |
Se construye a partir de protocol , reqHost (o message.reqHost ), reqPath (o message.reqPath ) y queryStr . |
¿Necesitas más ayuda? Obtén respuestas de miembros de la comunidad y profesionales de Google SecOps.