Collecter les journaux du framework d'analyse de fichiers
Ce document explique comment ingérer les journaux du framework d'analyse de fichiers dans Google Security Operations à l'aide de Google Cloud Storage V2.
File Scanning Framework (FSF) est une solution d'analyse de fichiers récursive et modulaire open source développée par Emerson Electric Co. FSF utilise une architecture client-serveur pour analyser les fichiers et générer des résultats d'analyse JSON détaillés, y compris les métadonnées des fichiers, les correspondances de signature YARA, les sous-objets extraits et les métadonnées spécifiques aux modules.
Avant de commencer
Assurez-vous de remplir les conditions préalables suivantes :
- Une instance Google SecOps
- Un projet GCP avec l'API Cloud Storage activée
- Autorisations pour créer et gérer des buckets GCS
- Autorisations permettant de gérer les stratégies IAM sur les buckets GCS
- Instance de serveur FSF déployée avec accès en écriture au répertoire de journaux
- Un accès root ou sudo à l'hôte du serveur FSF
Créer un bucket Google Cloud Storage
- Accédez à Google Cloud Console.
- Sélectionnez votre projet ou créez-en un.
- Dans le menu de navigation, accédez à Cloud Storage> Buckets.
- Cliquez sur Créer un bucket.
Fournissez les informations de configuration suivantes :
Paramètre Valeur Nommer votre bucket Saisissez un nom unique (par exemple, fsf-logs-secops).Type d'emplacement Choisissez en fonction de vos besoins (région, birégion ou multirégion). Emplacement Sélectionnez l'emplacement (par exemple, us-central1).Classe de stockage Standard (recommandé pour les journaux auxquels vous accédez fréquemment) Access control (Contrôle des accès) Uniforme (recommandé). Outils de protection Facultatif : Activez la gestion des versions des objets ou la règle de conservation. Cliquez sur Créer.
Configurer le répertoire de sortie des journaux FSF
FSF écrit les résultats de l'analyse JSON dans un répertoire de journaux configurable. Configurez un répertoire dédié à l'ingestion Google SecOps.
- Connectez-vous à l'hôte du serveur FSF via SSH.
Ouvrez le fichier de configuration du serveur FSF :
sudo nano /opt/fsf/fsf-server/conf/config.pyRecherchez le dictionnaire
SCANNER_CONFIG.Mettez à jour le paramètre
LOG_PATHvers un répertoire dédié :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 }Enregistrez et fermez le fichier.
Créez le répertoire de journaux avec les autorisations appropriées :
sudo mkdir -p /var/log/fsf sudo chown -R fsf:fsf /var/log/fsf sudo chmod 755 /var/log/fsfRedémarrez le serveur FSF pour appliquer les modifications :
sudo systemctl restart fsfVérifiez que FSF écrit les journaux dans le nouveau répertoire :
ls -lh /var/log/fsf/
Installer et configurer Fluentd
Fluentd suivra les fichiers journaux FSF et les enverra à Google Cloud Storage.
Installer Fluentd
Sur l'hôte du serveur FSF, installez Fluentd (td-agent) :
curl -fsSL https://toolbelt.treasuredata.com/sh/install-ubuntu-jammy-td-agent4.sh | shInstallez le plug-in de sortie GCS :
sudo td-agent-gem install fluent-plugin-gcsVérifiez l'installation du plug-in :
td-agent-gem list | grep fluent-plugin-gcs
Créer un compte de service GCP pour Fluentd
- Dans la console GCP, accédez à IAM et administration > Comptes de service.
- Cliquez sur Créer un compte de service.
- Fournissez les informations de configuration suivantes :
- Nom du compte de service : saisissez
fsf-fluentd-shipper. - Description du compte de service : saisissez
Service account for Fluentd to ship FSF logs to GCS.
- Nom du compte de service : saisissez
- Cliquez sur Créer et continuer.
- Dans la section Autoriser ce compte de service à accéder au projet :
- Cliquez sur Sélectionner un rôle.
- Recherchez et sélectionnez Administrateur des objets de l'espace de stockage.
- Cliquez sur Continuer.
- Cliquez sur OK.
Créer une clé de compte de service
- Dans la liste Comptes de service, cliquez sur le compte de service (
fsf-fluentd-shipper). - Accédez à l'onglet Clés.
- Cliquez sur Ajouter une clé > Créer une clé.
- Sélectionnez JSON comme type de clé.
- Cliquez sur Créer.
- Le fichier de clé JSON est téléchargé sur votre ordinateur.
Transférez le fichier de clé vers l'hôte du serveur FSF :
scp /path/to/downloaded-key.json user@fsf-server:/etc/td-agent/gcp-key.jsonDéfinissez les autorisations appropriées sur le fichier de clé :
sudo chown td-agent:td-agent /etc/td-agent/gcp-key.json sudo chmod 600 /etc/td-agent/gcp-key.json
Accorder des autorisations IAM sur un bucket GCS
- Accédez à Cloud Storage > Buckets.
- Cliquez sur le nom du bucket (
fsf-logs-secops). - Accédez à l'onglet Autorisations.
- Cliquez sur Accorder l'accès.
- Fournissez les informations de configuration suivantes :
- Ajouter des comptes principaux : saisissez l'adresse e-mail du compte de service (par exemple,
fsf-fluentd-shipper@PROJECT_ID.iam.gserviceaccount.com). - Attribuer des rôles : sélectionnez Administrateur des objets Storage.
- Ajouter des comptes principaux : saisissez l'adresse e-mail du compte de service (par exemple,
- Cliquez sur Enregistrer.
Configurer Fluentd
Sur l'hôte du serveur FSF, créez un fichier de configuration Fluentd :
sudo nano /etc/td-agent/td-agent.confAjoutez la configuration suivante :
# 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>Remplacez
YOUR_GCP_PROJECT_IDpar l'ID de votre projet GCP.Enregistrez et fermez le fichier.
Créez le répertoire du tampon :
sudo mkdir -p /var/log/td-agent/buffer/gcs sudo chown -R td-agent:td-agent /var/log/td-agent/bufferRedémarrez Fluentd pour appliquer la configuration :
sudo systemctl restart td-agentActivez Fluentd pour qu'il démarre au démarrage :
sudo systemctl enable td-agentVérifiez que Fluentd est en cours d'exécution :
sudo systemctl status td-agent
Vérifier l'envoi des journaux
Recherchez les erreurs dans les journaux Fluentd :
sudo tail -f /var/log/td-agent/td-agent.logDéclenchez une analyse FSF de test pour générer des journaux :
echo "test content" > /tmp/test.txt /opt/fsf/fsf-client/fsf_client.py /tmp/test.txt --suppress-reportPatientez une à deux minutes pour que Fluentd traite et envoie les journaux.
Dans la console GCP, accédez à Cloud Storage > Buckets.
Cliquez sur le nom du bucket (
fsf-logs-secops).Accédez au préfixe
fsf-logs/.Vérifiez que des fichiers JSON sont créés avec des codes temporels.
Téléchargez et inspectez un fichier pour confirmer qu'il contient les résultats de l'analyse FSF au format JSON.
Récupérer le compte de service Google SecOps
Google SecOps utilise un compte de service unique pour lire les données de votre bucket GCS. Vous devez accorder à ce compte de service l'accès à votre bucket.
Obtenir l'adresse e-mail du compte de service
- Accédez à Paramètres SIEM> Flux.
- Cliquez sur Add New Feed (Ajouter un flux).
- Cliquez sur Configurer un flux unique.
- Dans le champ Nom du flux, saisissez un nom pour le flux (par exemple,
FSF File Scanning Logs). - Sélectionnez Google Cloud Storage V2 comme Type de source.
- Sélectionnez Framework d'analyse des fichiers comme Type de journal.
Cliquez sur Obtenir un compte de service. Une adresse e-mail unique pour le compte de service s'affiche, par exemple :
secops-12345678@secops-gcp-prod.iam.gserviceaccount.comCopiez l'adresse e-mail pour l'utiliser à l'étape suivante.
Cliquez sur Suivant.
Spécifiez les valeurs des paramètres d'entrée suivants :
URL du bucket de stockage : saisissez l'URI du bucket GCS avec le chemin d'accès au préfixe :
gs://fsf-logs-secops/fsf-logs/Option de suppression de la source : sélectionnez l'option de suppression de votre choix :
- Jamais : ne supprime jamais aucun fichier après les transferts (recommandé pour les tests).
- Supprimer les fichiers transférés : supprime les fichiers après un transfert réussi.
Supprimer les fichiers transférés et les répertoires vides : supprime les fichiers et les répertoires vides après un transfert réussi.
Âge maximal des fichiers : incluez les fichiers modifiés au cours des derniers jours (180 jours par défaut).
Espace de noms de l'élément : espace de noms de l'élément.
Libellés d'ingestion : libellé à appliquer aux événements de ce flux.
Cliquez sur Suivant.
Vérifiez la configuration de votre nouveau flux sur l'écran Finaliser, puis cliquez sur Envoyer.
Accorder des autorisations IAM au compte de service Google SecOps
Le compte de service Google SecOps a besoin du rôle Lecteur des objets Storage sur votre bucket GCS.
- Accédez à Cloud Storage > Buckets.
- Cliquez sur le nom du bucket (
fsf-logs-secops). - Accédez à l'onglet Autorisations.
- Cliquez sur Accorder l'accès.
- Fournissez les informations de configuration suivantes :
- Ajouter des comptes principaux : collez l'adresse e-mail du compte de service Google SecOps.
- Attribuer des rôles : sélectionnez Lecteur d'objets Storage.
- Cliquez sur Enregistrer.
Vérifier l'ingestion
- Attendez 10 à 15 minutes que l'ingestion initiale soit terminée.
- Dans Google SecOps, accédez à Paramètres SIEM > Flux.
- Localisez le flux (
FSF File Scanning Logs). - Vérifiez que l'état Actif s'affiche.
- Cliquez sur le nom du flux pour afficher les métriques d'ingestion.
- Vérifiez que le nombre d'événements ingérés augmente.
- Accédez à Rechercher dans Google SecOps.
Exécutez une requête de recherche pour vérifier que les journaux FSF sont ingérés :
metadata.log_type = "FILE_SCANNING_FRAMEWORK"Vérifiez que les résultats de l'analyse FSF s'affichent dans les résultats de recherche.
Dépannage
Aucun journal n'apparaît dans GCS
Vérifiez que FSF écrit des journaux dans
/var/log/fsf/:ls -lh /var/log/fsf/ tail -f /var/log/fsf/*.logRecherchez les erreurs dans les journaux Fluentd :
sudo tail -f /var/log/td-agent/td-agent.logVérifiez que la clé du compte de service GCP est valide et dispose des autorisations appropriées.
Vérifiez que le nom du bucket dans la configuration Fluentd correspond au nom réel du bucket.
Erreurs d'autorisation Fluentd
- Vérifiez que le compte de service (
fsf-fluentd-shipper) dispose du rôle Administrateur des objets de l'espace de stockage sur le bucket. - Vérifiez que le chemin d'accès à la clé dans la configuration Fluentd est correct.
Vérifiez que le fichier de clé dispose des autorisations et du propriétaire appropriés :
ls -l /etc/td-agent/gcp-key.json
Google SecOps n'ingère pas les journaux
- Vérifiez que le compte de service Google SecOps dispose du rôle Lecteur des objets Storage sur le bucket.
- Vérifiez que l'URI du bucket dans la configuration du flux est correct et inclut la barre oblique finale.
- Vérifiez que les fichiers existent dans le bucket GCS au niveau du chemin d'accès au préfixe spécifié.
- Recherchez les messages d'erreur dans l'état du flux, sous Paramètres du SIEM > Flux.
Journaux FSF non au format attendu
- Vérifiez que FSF est configuré pour écrire la sortie JSON (comportement par défaut).
- Vérifiez que la section
<parse>de Fluentd est configurée avec@type json. Inspectez manuellement un fichier journal pour vérifier qu'il contient un fichier JSON valide :
head -n 1 /var/log/fsf/*.log | jq .
Table de mappage UDM
| Champ du journal | Mappage UDM | Logique |
|---|---|---|
| CompressType_label, compressed_parents | about.labels | Fusionné à partir de CompressType_label (clé "Compress Type", valeur issue de Object.EXTRACT_ZIP.Object_0.Compress Type si le message contient "Compress Type") et compressed_parents (clé "Compressed Parent Files", concaténée à partir 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 | Valeur de Object.EXTRACT_EMBEDDED.Object_0.META_BASIC_INFO.MD5 si EXTRACT_EMBEDDED est présent, sinon Object.EXTRACT_ZIP.Object_0.META_BASIC_INFO.MD5 si EXTRACT_ZIP est présent, sinon Object.EXTRACT_SWF.META_BASIC_INFO.MD5 si EXTRACT_SWF est présent, sinon Object.EXTRACT_GZIP.META_BASIC_INFO.MD5 si EXTRACT_GZIP est présent, sinon Object.EXTRACT_CAB.Object_0.META_BASIC_INFO.MD5 |
| Object.EXTRACT_EMBEDDED.Object_0.Description | intermediary.file.mime_type | Valeur copiée directement |
| 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 | Valeur de Object.EXTRACT_EMBEDDED.Object_0.META_BASIC_INFO.SHA1 si EXTRACT_EMBEDDED est présent, sinon Object.EXTRACT_ZIP.Object_0.META_BASIC_INFO.SHA1 si EXTRACT_ZIP est présent, sinon Object.EXTRACT_SWF.META_BASIC_INFO.SHA1 si EXTRACT_SWF est présent, sinon Object.EXTRACT_GZIP.META_BASIC_INFO.SHA1 si EXTRACT_GZIP est présent, sinon 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 | Valeur de Object.EXTRACT_EMBEDDED.Object_0.META_BASIC_INFO.SHA256 si EXTRACT_EMBEDDED est présent, sinon Object.EXTRACT_ZIP.Object_0.META_BASIC_INFO.SHA256 si EXTRACT_ZIP est présent, sinon Object.EXTRACT_SWF.META_BASIC_INFO.SHA256 si EXTRACT_SWF est présent, sinon Object.EXTRACT_GZIP.META_BASIC_INFO.SHA256 si EXTRACT_GZIP est présent, sinon 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 | Valeur de Object.EXTRACT_EMBEDDED.Object_0.META_BASIC_INFO.Size si EXTRACT_EMBEDDED est présent, sinon Object.EXTRACT_ZIP.Object_0.META_BASIC_INFO.Size si EXTRACT_ZIP est présent, sinon Object.EXTRACT_SWF.META_BASIC_INFO.Size si EXTRACT_SWF est présent, sinon Object.EXTRACT_GZIP.META_BASIC_INFO.Size si EXTRACT_GZIP est présent, sinon Object.EXTRACT_CAB.Object_0.META_BASIC_INFO.Size ; sans le " .*" de fin et converti en uinteger |
| Object.EXTRACT_ZIP.Object_0.META_VT_CACHE.vt_data.scan_id | intermediary.resource.id | Valeur copiée directement |
| Object.EXTRACT_ZIP.Object_0.META_VT_CACHE.vt_data.permalink | intermediary.url | Valeur copiée directement |
| Object.META_EMERSON_INFO.results | intermediary.user.email_addresses | Fusionné à partir de matched_email dans le tableau des résultats |
| Summary.Observations | metadata.description | Concaténation à partir du tableau avec le séparateur ", ", virgule de début supprimée |
| Heure d'analyse | metadata.event_timestamp | Converti à l'aide du filtre de date au format aaaa-MM-jj HH:mm:ss |
| Source | metadata.event_type | Définissez la valeur sur "SCAN_FILE" si la source n'est pas vide, sinon sur "GENERIC_EVENT". |
| Object.META_VT_CACHE._id | metadata.product_log_id | Valeur copiée directement |
| result.ad_data.message | network.http.response_code | Extrait sous forme d'entier à l'aide du modèle grok INT à partir de result.ad_data.message |
| Source | principal.hostname | Valeur copiée directement |
| Object.META_EMERSON_INFO.result_summary, Object.EXTRACT_ZIP.Object_0.META_VT_CACHE.vt_data.verbose_msg | security_result.summary | Définissez la valeur sur Object.META_EMERSON_INFO.result_summary si elle est présente, sinon sur Object.EXTRACT_ZIP.Object_0.META_VT_CACHE.vt_data.verbose_msg. |
| Nom de fichier | target.file.full_path | Valeur copiée directement |
| Object.META_BASIC_INFO.MD5 | target.file.md5 | Valeur copiée directement |
| Summary.Yara | target.file.mime_type | Extrait du premier index de Summary.Yara, mis en majuscules et "FT_" supprimé si Yara est présent, sinon défini sur "ZIP" si EXTRACT_ZIP est présent, "SWF" si EXTRACT_SWF est présent, "GZIP" si EXTRACT_GZIP est présent, "CAB" si EXTRACT_CAB est présent |
| Object.META_BASIC_INFO.SHA1, Object.META_VT_CACHE.SHA1 | target.file.sha1 | Valeur de Object.META_BASIC_INFO.SHA1 si elle n'est pas vide, sinon Object.META_VT_CACHE.SHA1 |
| Object.META_BASIC_INFO.SHA256 | target.file.sha256 | Valeur copiée directement |
| Object.META_BASIC_INFO.Size | target.file.size | Suppression de la chaîne " .*" à la fin et conversion en entier non signé |
| metadata.vendor_name | Défini sur "EMERSON" | |
| metadata.product_name | Définissez-le sur "FILE SCANNING FRAMEWORK" (CADRE D'ANALYSE DES FICHIERS). |
Vous avez encore besoin d'aide ? Obtenez des réponses de membres de la communauté et de professionnels Google SecOps.