Recopila registros del marco de trabajo de análisis de archivos
Versión del analizador: 3.0
En este documento, se explica cómo transferir registros del marco de trabajo de análisis de archivos a Google Security Operations con Google Cloud Storage V2.
El marco de trabajo de análisis de archivos (FSF) es una solución de análisis de archivos recursiva y modular de código abierto desarrollada por Emerson Electric Co. El FSF usa una arquitectura cliente-servidor para analizar archivos y generar resultados de análisis JSON detallados, incluidos metadatos de archivos, coincidencias de firmas de YARA, subobjetos extraídos y metadatos específicos del módulo.
Antes de comenzar
Asegúrate de cumplir con los siguientes requisitos previos:
- Una instancia de Google SecOps
- Un proyecto de GCP con la API de Cloud Storage habilitada
- Permisos para crear y administrar buckets de GCS
- Permisos para administrar políticas de IAM en buckets de GCS
- Una instancia del servidor de FSF implementada con acceso de escritura al directorio de registros
- Acceso raíz o sudo al host del servidor de FSF
Crea un bucket de Google Cloud Storage
- Ve a la consola de Google Cloud.
- Selecciona tu proyecto o crea uno nuevo.
- En el menú de navegación, ve a Cloud Storage > Buckets.
- Haz clic en Crear bucket.
Proporciona los siguientes detalles de configuración:
Parámetro de configuración Valor Asigna un nombre a tu bucket Ingresa un nombre global único (por ejemplo, fsf-logs-secops).Tipo de ubicación Elige según tus necesidades (región, región doble o varias regiones). Ubicación Selecciona la ubicación (por ejemplo, us-central1).Clase de almacenamiento Estándar (recomendado para los registros a los que se accede con frecuencia) Control de acceso Uniforme (recomendado) Herramientas de protección Opcional: Habilita el control de versiones de objetos o la política de retención. Haz clic en Crear.
Configura el directorio de salida de registros de FSF
FSF escribe los resultados del análisis en formato JSON en un directorio de registro configurable. Configura un directorio exclusivo para la transferencia de datos de Google SecOps.
- Conéctate al host del servidor de FSF a través de SSH.
Abre el archivo de configuración del servidor de FSF:
sudo nano /opt/fsf/fsf-server/conf/config.pyUbica el diccionario
SCANNER_CONFIG.Actualiza el parámetro
LOG_PATHa un directorio dedicado:SCANNER_CONFIG = { 'LOG_PATH': '/var/log/fsf', 'YARA_PATH': '/opt/fsf/fsf-server/yara/rules.yara', 'PID_PATH': '/tmp/scanner.pid', 'EXPORT_PATH': '/tmp', 'TIMEOUT': 60, 'MAX_DEPTH': 10 }Guarde y cierre el archivo.
Crea el directorio de registros con los permisos adecuados:
sudo mkdir -p /var/log/fsf sudo chown -R fsf:fsf /var/log/fsf sudo chmod 755 /var/log/fsfReinicia el servidor de FSF para aplicar los cambios:
sudo systemctl restart fsfVerifica que FSF escriba registros en el directorio nuevo:
ls -lh /var/log/fsf/
Instala y configura Fluentd
Fluentd hará un seguimiento de los archivos de registro de FSF y los enviará a Google Cloud Storage.
Instala Fluentd
En el host del servidor de FSF, instala Fluentd (td-agent):
curl -fsSL https://toolbelt.treasuredata.com/sh/install-ubuntu-jammy-td-agent4.sh | shInstala el complemento de salida de GCS:
sudo td-agent-gem install fluent-plugin-gcsVerifica la instalación del complemento:
td-agent-gem list | grep fluent-plugin-gcs
Crea una cuenta de servicio de GCP para Fluentd
- En Google Cloud Console, ve a IAM y administración > Cuentas de servicio.
- Haz clic en Crear cuenta de servicio.
- Proporciona los siguientes detalles de configuración:
- Nombre de la cuenta de servicio: Ingresa
fsf-fluentd-shipper. - Descripción de la cuenta de servicio: Ingresa
Service account for Fluentd to ship FSF logs to GCS.
- Nombre de la cuenta de servicio: Ingresa
- Haz clic en Crear y continuar.
- En la sección Otorga acceso al proyecto a esta cuenta de servicio, haz lo siguiente:
- Haz clic en Seleccionar un rol.
- Busca y selecciona Administrador de objetos de Storage.
- Haz clic en Continuar.
- Haz clic en Listo.
Crear clave de la cuenta de servicio
- En la lista de Cuentas de servicio, haz clic en la cuenta de servicio (
fsf-fluentd-shipper). - Ve a la pestaña Claves.
- Haz clic en Agregar clave > Crear clave nueva.
- Seleccione JSON como tipo de clave.
- Haz clic en Crear.
- El archivo de claves JSON se descargará en tu computadora.
Transfiere el archivo de claves al host del servidor de FSF:
scp /path/to/downloaded-key.json user@fsf-server:/etc/td-agent/gcp-key.jsonEstablece los permisos adecuados en el archivo de claves:
sudo chown td-agent:td-agent /etc/td-agent/gcp-key.json sudo chmod 600 /etc/td-agent/gcp-key.json
Otorga permisos de IAM en el bucket de GCS
- Ve a Cloud Storage > Buckets.
- Haz clic en el nombre del bucket (
fsf-logs-secops). - Ve a la pestaña Permisos.
- Haz clic en Otorgar acceso.
- Proporciona los siguientes detalles de configuración:
- Agregar entidades: Ingresa el correo electrónico de la cuenta de servicio (por ejemplo,
fsf-fluentd-shipper@PROJECT_ID.iam.gserviceaccount.com). - Asignar roles: Selecciona Administrador de objetos de Storage.
- Agregar entidades: Ingresa el correo electrónico de la cuenta de servicio (por ejemplo,
- Haz clic en Guardar.
Configura Fluentd
En el host del servidor de FSF, crea un archivo de configuración de Fluentd:
sudo nano /etc/td-agent/td-agent.confAgrega la siguiente configuración:
# Tail FSF JSON logs <source> @type tail path /var/log/fsf/*.log pos_file /var/log/td-agent/fsf.log.pos tag fsf.scan read_from_head true <parse> @type json time_key timestamp time_format %Y-%m-%dT%H:%M:%S.%L%z </parse> </source> # Ship to Google Cloud Storage <match fsf.scan> @type gcs project YOUR_GCP_PROJECT_ID keyfile /etc/td-agent/gcp-key.json bucket fsf-logs-secops object_key_format %{path}%{time_slice}_%{index}.%{file_extension} path fsf-logs/ <buffer tag,time> @type file path /var/log/td-agent/buffer/gcs timekey 3600 timekey_wait 10m timekey_use_utc true chunk_limit_size 10MB </buffer> <format> @type json </format> store_as json auto_create_bucket false </match>Reemplaza
YOUR_GCP_PROJECT_IDpor el ID de tu proyecto de GCP.Guarde y cierre el archivo.
Crea el directorio del búfer:
sudo mkdir -p /var/log/td-agent/buffer/gcs sudo chown -R td-agent:td-agent /var/log/td-agent/bufferReinicia Fluentd para aplicar la configuración:
sudo systemctl restart td-agentHabilita Fluentd para que se inicie durante el arranque:
sudo systemctl enable td-agentVerifica que Fluentd se esté ejecutando:
sudo systemctl status td-agent
Verifica el envío de registros
Verifica si hay errores en los registros de Fluentd:
sudo tail -f /var/log/td-agent/td-agent.logActiva un análisis de FSF de prueba para generar registros:
echo "test content" > /tmp/test.txt /opt/fsf/fsf-client/fsf_client.py /tmp/test.txt --suppress-reportEspera entre 1 y 2 minutos para que Fluentd procese y envíe los registros.
En la consola de GCP, ve a Cloud Storage > Buckets.
Haz clic en el nombre del bucket (
fsf-logs-secops).Navega al prefijo
fsf-logs/.Verifica que los archivos JSON se creen con marcas de tiempo.
Descarga e inspecciona un archivo para confirmar que contiene resultados del análisis de FSF en formato JSON.
Recupera la cuenta de servicio de Google SecOps
Google SecOps usa una cuenta de servicio única para leer datos de tu bucket de GCS. Debes otorgar acceso a tu bucket a esta cuenta de servicio.
Obtén el correo electrónico de la cuenta de servicio
- Ve a Configuración de SIEM > Feeds.
- Haz clic en Add New Feed.
- Haz clic en Configura un feed único.
- En el campo Nombre del feed, ingresa un nombre para el feed (por ejemplo,
FSF File Scanning Logs). - Selecciona Google Cloud Storage V2 como el Tipo de fuente.
- Selecciona File Scanning Framework como el Tipo de registro.
Haz clic en Obtener cuenta de servicio. Se mostrará un correo electrónico único de la cuenta de servicio, por ejemplo:
secops-12345678@secops-gcp-prod.iam.gserviceaccount.comCopia la dirección de correo electrónico para usarla en el siguiente paso.
Haz clic en Siguiente.
Especifica valores para los siguientes parámetros de entrada:
URL del bucket de almacenamiento: Ingresa el URI del bucket de GCS con la ruta de acceso del prefijo:
gs://fsf-logs-secops/fsf-logs/Opción de eliminación del código fuente: Selecciona la opción de eliminación según tu preferencia:
- Nunca: Nunca borra ningún archivo después de las transferencias (se recomienda para las pruebas).
- Borrar archivos transferidos: Borra los archivos después de que se transfirieron correctamente.
Borrar los archivos transferidos y los directorios vacíos: Borra los archivos y los directorios vacíos después de que se transfirieron correctamente.
Antigüedad máxima del archivo: Incluye los archivos modificados en la cantidad de días especificada (el valor predeterminado es de 180 días).
Espacio de nombres del recurso: Es el espacio de nombres del recurso.
Etiquetas de transferencia: Es la etiqueta que se aplicará a los eventos de este feed.
Haz clic en Siguiente.
Revisa la nueva configuración del feed en la pantalla Finalizar y, luego, haz clic en Enviar.
Otorga permisos de IAM a la cuenta de servicio de Google SecOps
La cuenta de servicio de Google SecOps necesita el rol de visualizador de objetos de Storage en tu bucket de GCS.
- Ve a Cloud Storage > Buckets.
- Haz clic en el nombre del bucket (
fsf-logs-secops). - Ve a la pestaña Permisos.
- Haz clic en Otorgar acceso.
- Proporciona los siguientes detalles de configuración:
- Agregar entidades: Pega el correo electrónico de la cuenta de servicio de Google SecOps.
- Asignar roles: Selecciona Visualizador de objetos de Storage.
- Haz clic en Guardar.
Verifica la transferencia
- Espera entre 10 y 15 minutos para que se complete la transferencia inicial.
- En Google SecOps, ve a Configuración de SIEM > Feeds.
- Ubica el feed (
FSF File Scanning Logs). - Verifica que el Estado se muestre como Activo.
- Haz clic en el nombre del feed para ver las métricas de transferencia.
- Verifica que aumente el recuento de Eventos procesados.
- Ve a Search en Google SecOps.
Ejecuta una búsqueda para verificar que se transfieran los registros de FSF:
metadata.log_type = "FILE_SCANNING_FRAMEWORK"Verifica que los resultados del análisis de FSF aparezcan en los resultados de la búsqueda.
Soluciona problemas
No aparecen registros en GCS
Verifica que FSF escriba registros en
/var/log/fsf/:ls -lh /var/log/fsf/ tail -f /var/log/fsf/*.logVerifica si hay errores en los registros de Fluentd:
sudo tail -f /var/log/td-agent/td-agent.logVerifica que la clave de la cuenta de servicio de GCP sea válida y tenga los permisos correctos.
Verifica que el nombre del bucket en la configuración de Fluentd coincida con el nombre real del bucket.
Errores de permisos de Fluentd
- Verifica que la cuenta de servicio (
fsf-fluentd-shipper) tenga el rol de administrador de objetos de almacenamiento en el bucket. - Verifica que la ruta de acceso a la clave en la configuración de Fluentd sea correcta.
Verifica que el archivo de clave tenga la propiedad y los permisos correctos:
ls -l /etc/td-agent/gcp-key.json
Google SecOps no ingiere registros
- Verifica que la cuenta de servicio de Google SecOps tenga el rol de visualizador de objetos de Storage en el bucket.
- Verifica que el URI del bucket en la configuración del feed sea correcto y que incluya la barra final.
- Verifica que los archivos existan en el bucket de GCS en la ruta de acceso del prefijo especificado.
- Verifica el estado del feed en Configuración de SIEM > Feeds para ver si hay mensajes de error.
Los registros de FSF no tienen el formato esperado
- Verifica que FSF esté configurado para escribir resultados en formato JSON (comportamiento predeterminado).
- Comprueba que la sección
<parse>de Fluentd esté configurada con@type json. Inspecciona un archivo de registro de forma manual para verificar que contenga un JSON válido:
head -n 1 /var/log/fsf/*.log | jq .
Tabla de asignación de UDM
| Campo de registro | Asignación de UDM | Lógica |
|---|---|---|
| CompressType_label, compressed_parents | about.labels | Se fusionó de CompressType_label (clave "Tipo de compresión", valor de Object.EXTRACT_ZIP.Object_0.Compress Type si el mensaje contiene "Tipo de compresión") y compressed_parents (clave "Archivos principales comprimidos", concatenado de Object.EXTRACT_ZIP.Object_0.META_VT_CACHE.vt_data.additional_info.compressed_parents) |
| Object.EXTRACT_EMBEDDED.Object_0.META_BASIC_INFO.MD5, Object.EXTRACT_ZIP.Object_0.META_BASIC_INFO.MD5, Object.EXTRACT_SWF.META_BASIC_INFO.MD5, Object.EXTRACT_GZIP.META_BASIC_INFO.MD5, Object.EXTRACT_CAB.Object_0.META_BASIC_INFO.MD5 | intermediary.file.md5 | Valor de Object.EXTRACT_EMBEDDED.Object_0.META_BASIC_INFO.MD5 si EXTRACT_EMBEDDED está presente; de lo contrario, Object.EXTRACT_ZIP.Object_0.META_BASIC_INFO.MD5 si EXTRACT_ZIP está presente; de lo contrario, Object.EXTRACT_SWF.META_BASIC_INFO.MD5 si EXTRACT_SWF está presente; de lo contrario, Object.EXTRACT_GZIP.META_BASIC_INFO.MD5 si EXTRACT_GZIP está presente; de lo contrario, Object.EXTRACT_CAB.Object_0.META_BASIC_INFO.MD5 |
| Object.EXTRACT_EMBEDDED.Object_0.Description | intermediary.file.mime_type | El valor se copió directamente. |
| Object.EXTRACT_EMBEDDED.Object_0.META_BASIC_INFO.SHA1, Object.EXTRACT_ZIP.Object_0.META_BASIC_INFO.SHA1, Object.EXTRACT_SWF.META_BASIC_INFO.SHA1, Object.EXTRACT_GZIP.META_BASIC_INFO.SHA1, Object.EXTRACT_CAB.Object_0.META_BASIC_INFO.SHA1 | intermediary.file.sha1 | Valor de Object.EXTRACT_EMBEDDED.Object_0.META_BASIC_INFO.SHA1 si EXTRACT_EMBEDDED está presente; de lo contrario, Object.EXTRACT_ZIP.Object_0.META_BASIC_INFO.SHA1 si EXTRACT_ZIP está presente; de lo contrario, Object.EXTRACT_SWF.META_BASIC_INFO.SHA1 si EXTRACT_SWF está presente; de lo contrario, Object.EXTRACT_GZIP.META_BASIC_INFO.SHA1 si EXTRACT_GZIP está presente; de lo contrario, Object.EXTRACT_CAB.Object_0.META_BASIC_INFO.SHA1 |
| Object.EXTRACT_EMBEDDED.Object_0.META_BASIC_INFO.SHA256, Object.EXTRACT_ZIP.Object_0.META_BASIC_INFO.SHA256, Object.EXTRACT_SWF.META_BASIC_INFO.SHA256, Object.EXTRACT_GZIP.META_BASIC_INFO.SHA256, Object.EXTRACT_CAB.Object_0.META_BASIC_INFO.SHA256 | intermediary.file.sha256 | Valor de Object.EXTRACT_EMBEDDED.Object_0.META_BASIC_INFO.SHA256 si EXTRACT_EMBEDDED está presente; de lo contrario, Object.EXTRACT_ZIP.Object_0.META_BASIC_INFO.SHA256 si EXTRACT_ZIP está presente; de lo contrario, Object.EXTRACT_SWF.META_BASIC_INFO.SHA256 si EXTRACT_SWF está presente; de lo contrario, Object.EXTRACT_GZIP.META_BASIC_INFO.SHA256 si EXTRACT_GZIP está presente; de lo contrario, Object.EXTRACT_CAB.Object_0.META_BASIC_INFO.SHA256 |
| Object.EXTRACT_EMBEDDED.Object_0.META_BASIC_INFO.Size, Object.EXTRACT_ZIP.Object_0.META_BASIC_INFO.Size, Object.EXTRACT_SWF.META_BASIC_INFO.Size, Object.EXTRACT_GZIP.META_BASIC_INFO.Size, Object.EXTRACT_CAB.Object_0.META_BASIC_INFO.Size | intermediary.file.size | Valor de Object.EXTRACT_EMBEDDED.Object_0.META_BASIC_INFO.Size si EXTRACT_EMBEDDED está presente; de lo contrario, Object.EXTRACT_ZIP.Object_0.META_BASIC_INFO.Size si EXTRACT_ZIP está presente; de lo contrario, Object.EXTRACT_SWF.META_BASIC_INFO.Size si EXTRACT_SWF está presente; de lo contrario, Object.EXTRACT_GZIP.META_BASIC_INFO.Size si EXTRACT_GZIP está presente; de lo contrario, Object.EXTRACT_CAB.Object_0.META_BASIC_INFO.Size; sin el " .*" final y convertido a uinteger |
| Object.EXTRACT_ZIP.Object_0.META_VT_CACHE.vt_data.scan_id | intermediary.resource.id | El valor se copió directamente. |
| Object.EXTRACT_ZIP.Object_0.META_VT_CACHE.vt_data.permalink | intermediary.url | El valor se copió directamente. |
| Object.META_EMERSON_INFO.results | intermediary.user.email_addresses | Se combinó desde matched_email en el array de resultados |
| Summary.Observations | metadata.description | Se concatenó a partir de un array con el separador ", ", y se quitó la coma inicial. |
| Tiempo de análisis | metadata.event_timestamp | Se convirtió con el filtro de fecha con el formato aaaa-MM-dd HH:mm:ss |
| Fuente | metadata.event_type | Se establece en "SCAN_FILE" si la fuente no está vacía; de lo contrario, se establece en "GENERIC_EVENT". |
| Object.META_VT_CACHE._id | metadata.product_log_id | El valor se copió directamente. |
| result.ad_data.message | network.http.response_code | Se extrae como un número entero con el patrón de grok INT del campo result.ad_data.message. |
| Fuente | principal.hostname | El valor se copió directamente. |
| Object.META_EMERSON_INFO.result_summary, Object.EXTRACT_ZIP.Object_0.META_VT_CACHE.vt_data.verbose_msg | security_result.summary | Se establece en Object.META_EMERSON_INFO.result_summary si está presente; de lo contrario, se establece en Object.EXTRACT_ZIP.Object_0.META_VT_CACHE.vt_data.verbose_msg. |
| Nombre del archivo | target.file.full_path | El valor se copió directamente. |
| Object.META_BASIC_INFO.MD5 | target.file.md5 | El valor se copió directamente. |
| Summary.Yara | target.file.mime_type | Se extrae del primer índice de Summary.Yara, se convierte a mayúsculas y se quita "FT_" si Yara está presente; de lo contrario, se establece en "ZIP" si EXTRACT_ZIP está presente, en "SWF" si EXTRACT_SWF está presente, en "GZIP" si EXTRACT_GZIP está presente y en "CAB" si EXTRACT_CAB está presente. |
| Object.META_BASIC_INFO.SHA1, Object.META_VT_CACHE.SHA1 | target.file.sha1 | Valor de Object.META_BASIC_INFO.SHA1 si no está vacío; de lo contrario, Object.META_VT_CACHE.SHA1 |
| Object.META_BASIC_INFO.SHA256 | target.file.sha256 | El valor se copió directamente. |
| Object.META_BASIC_INFO.Size | target.file.size | Se quitó el " .*" final y se convirtió en uinteger |
| metadata.vendor_name | Se estableció en "EMERSON". | |
| metadata.product_name | Se establece en "FILE SCANNING FRAMEWORK". |
¿Necesitas más ayuda? Obtén respuestas de miembros de la comunidad y profesionales de Google SecOps.