Recopila registros del framework de análisis de archivos
Versión del analizador: 3.0
En este documento, se explica cómo transferir registros de File Scanning Framework a Google Security Operations con Google Cloud Storage V2.
File Scanning Framework (FSF) es una solución de análisis de archivos recursiva, modular y de código abierto desarrollada por Emerson Electric Co. FSF usa una arquitectura cliente-servidor para analizar archivos y generar resultados de análisis JSON detallados, incluidos los metadatos de los archivos, las coincidencias de firmas de YARA, los subobjetos extraídos y los metadatos específicos del módulo.
Antes de comenzar
Asegúrate de cumplir con los siguientes requisitos:
- 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 de 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 único a nivel global (por ejemplo, fsf-logs-secops).Tipo de ubicación Elige una en función de tus necesidades (Región, Birregional o Multirregional). 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 de JSON en un directorio de registro configurable. Configura un directorio dedicado para la transferencia 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.pyBusca 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 esté escribiendo registros en el directorio nuevo:
ls -lh /var/log/fsf/
Instala y configura Fluentd
Fluentd supervisará 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 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 real 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 en el arranque:
sudo systemctl enable td-agentVerifica que Fluentd se esté ejecutando:
sudo systemctl status td-agent
Verifica el envío de registros
Revisa los registros de Fluentd en busca de errores:
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 se estén creando archivos JSON con marcas de tiempo.
Descarga y, luego, inspecciona un archivo para confirmar que contenga 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 > Administración de datos > Feeds. Si estás en una plataforma solo de SIEM, ve a Configuración > Feeds.
- Haz clic en Agregar feed nuevo.
- 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 última cantidad de días (el valor predeterminado es 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 configuración de tu feed de noticias nuevo 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.
- Asigna 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 > Administración de datos > Feeds. Si estás en una plataforma solo de SIEM, ve a Configuración > Feeds.
- Busca 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 el recuento de Eventos transferidos aumente.
- Ve a Búsqueda en Google SecOps.
Ejecuta una consulta de búsqueda para verificar que se estén transfiriendo 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 esté escribiendo registros en
/var/log/fsf/:ls -lh /var/log/fsf/ tail -f /var/log/fsf/*.logRevisa los registros de Fluentd en busca de errores:
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 la clave en la configuración de Fluentd sea correcta.
Verifica que el archivo de claves tenga la propiedad y los permisos correctos:
ls -l /etc/td-agent/gcp-key.json
Google SecOps no transfiere 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 especificada.
- Verifica el estado del feed en Configuración > Administración de datos > 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 JSON (comportamiento predeterminado).
- Verifica 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 combinó CompressType_label (clave "Compress Type", valor de Object.EXTRACT_ZIP.Object_0.Compress Type si el mensaje contiene "Compress Type") y compressed_parents (clave "Compressed Parent Files", concatenada 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 copia 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 en un número entero sin signo |
| Object.EXTRACT_ZIP.Object_0.META_VT_CACHE.vt_data.scan_id | intermediary.resource.id | El valor se copia directamente. |
| Object.EXTRACT_ZIP.Object_0.META_VT_CACHE.vt_data.permalink | intermediary.url | El valor se copia 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ó desde el 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 el origen no está vacío; de lo contrario, se establece en "GENERIC_EVENT". |
| Object.META_VT_CACHE._id | metadata.product_log_id | El valor se copia directamente. |
| result.ad_data.message | network.http.response_code | Se extrae como un número entero con el patrón de grok INT de result.ad_data.message |
| Fuente | principal.hostname | El valor se copia 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, en Object.EXTRACT_ZIP.Object_0.META_VT_CACHE.vt_data.verbose_msg |
| Nombre del archivo | target.file.full_path | El valor se copia directamente. |
| Object.META_BASIC_INFO.MD5 | target.file.md5 | El valor se copia 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, "SWF" si EXTRACT_SWF está presente, "GZIP" si EXTRACT_GZIP está presente, "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 copia directamente. |
| Object.META_BASIC_INFO.Size | target.file.size | Se quita el " .*" final y se convierte en un número entero sin signo |
| metadata.vendor_name | Establece el valor en "EMERSON". | |
| metadata.product_name | Establece en "FILE SCANNING FRAMEWORK". |
¿Necesitas más ayuda? Obtén respuestas de miembros de la comunidad y profesionales de Google SecOps.