Collecter les journaux Proofpoint Emerging Threats Pro IOC
Ce document explique comment ingérer les journaux Proofpoint Emerging Threats Pro IOC dans Google Security Operations à l'aide de Google Cloud Storage. Emerging Threats Intelligence publie des listes de réputation horaires pour les adresses IP et les domaines au format CSV, avec des données de renseignements sur les menaces, y compris des catégories, des scores et des informations temporelles. Le code du parseur traite les données de renseignements sur les menaces ET_PRO au format CSV. Il extrait les adresses IP, les domaines, les catégories, les scores et d'autres informations pertinentes, et les mappe à la fois à un format d'IOC standardisé et au schéma UDM de Chronicle pour une analyse et une utilisation plus approfondies dans Google SecOps.
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
- Autorisations permettant de créer des services Cloud Run, des sujets Pub/Sub et des tâches Cloud Scheduler
- Abonnement Proofpoint ET Intelligence avec accès aux listes de réputation
- Clé API ET Intelligence obtenue sur https://etadmin.proofpoint.com/api-access
Collecter les conditions préalables d'Emerging Threats Pro
- Connectez-vous au portail d'administration ET Intelligence à l'adresse https://etadmin.proofpoint.com.
- Accédez à Accès à l'API.
- Copiez et enregistrez votre clé API.
- Contactez votre représentant Proofpoint pour obtenir les éléments suivants :
- URL de la liste détaillée de la réputation des adresses IP
- URL de la liste détaillée de la réputation des domaines
ET Intelligence fournit des fichiers CSV distincts pour les listes de réputation des adresses IP et des domaines, qui sont mis à jour toutes les heures. Utilisez le format "Détaillé", qui inclut les colonnes suivantes :
- Liste des domaines : nom de domaine, catégorie, score, première et dernière détection, ports
- Liste d'adresses IP : adresse IP, catégorie, score, première observation, dernière observation, ports
Les URL au format détaillé suivent généralement ce modèle :
- Liste d'adresses IP :
https://rules.emergingthreatspro.com/[your-code]/reputation/detailed-iprepdata.txt - Liste des domaines :
https://rules.emergingthreatspro.com/[your-code]/reputation/detailed-domainrepdata.txt
L'en-tête d'autorisation doit contenir la valeur brute de la clé API sans préfixe "Bearer", conformément aux attentes de l'API Emerging Threats.
Créer un bucket Google Cloud Storage
- Accédez à la console Google Cloud.
- 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, et-pro-ioc-bucket).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 : Activer la gestion des versions des objets ou la règle de conservation Cliquez sur Créer.
Créer un compte de service pour la fonction Cloud Run
La fonction Cloud Run a besoin d'un compte de service disposant des autorisations nécessaires pour écrire dans le bucket GCS et être appelée par Pub/Sub.
Créer un compte de service
- 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
et-pro-ioc-fetcher-sa. - Description du compte de service : saisissez
Service account for Cloud Run function to collect Proofpoint ET Pro IOC logs.
- Nom du compte de service : saisissez
- Cliquez sur Créer et continuer.
- Dans la section Autoriser ce compte de service à accéder au projet, ajoutez les rôles suivants :
- Cliquez sur Sélectionner un rôle.
- Recherchez et sélectionnez Administrateur des objets de l'espace de stockage.
- Cliquez sur + Ajouter un autre rôle.
- Recherchez et sélectionnez Demandeur Cloud Run.
- Cliquez sur + Ajouter un autre rôle.
- Recherchez et sélectionnez Demandeur Cloud Functions.
- Cliquez sur Continuer.
- Cliquez sur OK.
Ces rôles sont requis pour :
- Administrateur des objets Storage : écrire des journaux dans le bucket GCS et gérer les fichiers d'état
- Demandeur Cloud Run : autorise Pub/Sub à appeler la fonction
- Demandeur Cloud Functions : autorise l'appel de fonctions
Accorder des autorisations IAM sur un bucket GCS
Accordez au compte de service des autorisations d'écriture sur le bucket GCS :
- Accédez à Cloud Storage > Buckets.
- Cliquez sur le nom de votre bucket.
- 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,
et-pro-ioc-fetcher-sa@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.
Créer un sujet Pub/Sub
Créez un sujet Pub/Sub auquel Cloud Scheduler publiera des messages et auquel la fonction Cloud Run s'abonnera.
- Dans la console GCP, accédez à Pub/Sub > Sujets.
- Cliquez sur Create topic (Créer un sujet).
- Fournissez les informations de configuration suivantes :
- ID du sujet : saisissez
et-pro-ioc-trigger. - Conservez les valeurs par défaut des autres paramètres.
- ID du sujet : saisissez
- Cliquez sur Créer.
Créer une fonction Cloud Run pour collecter les journaux
La fonction Cloud Run est déclenchée par les messages Pub/Sub de Cloud Scheduler pour extraire les journaux de l'API Proofpoint ET Intelligence et les écrire dans GCS.
- Dans la console GCP, accédez à Cloud Run.
- Cliquez sur Créer un service.
- Sélectionnez Fonction (utilisez un éditeur intégré pour créer une fonction).
Dans la section Configurer, fournissez les informations de configuration suivantes :
Paramètre Valeur Nom du service et-pro-ioc-fetcherRégion Sélectionnez la région correspondant à votre bucket GCS (par exemple, us-central1).Runtime (durée d'exécution) Sélectionnez Python 3.12 ou version ultérieure. Dans la section Déclencheur (facultatif) :
- Cliquez sur + Ajouter un déclencheur.
- Sélectionnez Cloud Pub/Sub.
- Dans Sélectionner un sujet Cloud Pub/Sub, choisissez
et-pro-ioc-trigger. - Cliquez sur Enregistrer.
Dans la section Authentification :
- Sélectionnez Exiger l'authentification.
- Consultez Identity and Access Management (IAM).
Faites défiler la page vers le bas, puis développez Conteneurs, mise en réseau, sécurité.
Accédez à l'onglet Sécurité :
- Compte de service : sélectionnez
et-pro-ioc-fetcher-sa.
- Compte de service : sélectionnez
Accédez à l'onglet Conteneurs :
- Cliquez sur Variables et secrets.
- Cliquez sur + Ajouter une variable pour chaque variable d'environnement :
Nom de la variable Exemple de valeur Description GCS_BUCKETet-pro-ioc-bucketNom du bucket GCS GCS_PREFIXet-pro-iocPréfixe des fichiers journaux STATE_KEYet-pro-ioc/state.jsonChemin d'accès au fichier d'état ET_API_KEYyour-et-api-keyClé API ET Intelligence ET_IP_LIST_URLhttps://rules.emergingthreatspro.com/[your-code]/reputation/detailed-iprepdata.txtURL de la liste détaillée de la réputation des adresses IP ET_DOMAIN_LIST_URLhttps://rules.emergingthreatspro.com/[your-code]/reputation/detailed-domainrepdata.txtURL de la liste détaillée de la réputation des domaines TIMEOUT120Délai avant expiration de la requête HTTP (en secondes) Dans l'onglet Variables et secrets, faites défiler la page jusqu'à Requêtes :
- Délai avant expiration de la requête : saisissez
600secondes (10 minutes).
- Délai avant expiration de la requête : saisissez
Accédez à l'onglet Paramètres dans Conteneurs :
- Dans la section Ressources :
- Mémoire : sélectionnez 512 Mio ou plus.
- CPU : sélectionnez 1.
- Cliquez sur OK.
- Dans la section Ressources :
Faites défiler la page jusqu'à Environnement d'exécution :
- Sélectionnez Par défaut (recommandé).
Dans la section Scaling de révision :
- Nombre minimal d'instances : saisissez
0. - Nombre maximal d'instances : saisissez
100(ou ajustez en fonction de la charge attendue).
- Nombre minimal d'instances : saisissez
Cliquez sur Créer.
Attendez que le service soit créé (1 à 2 minutes).
Une fois le service créé, l'éditeur de code intégré s'ouvre automatiquement.
Ajouter un code de fonction
- Saisissez main dans Point d'entrée de la fonction.
Dans l'éditeur de code intégré, créez deux fichiers :
- Premier fichier : main.py:
import functions_framework from google.cloud import storage import json import os import urllib3 from datetime import datetime, timezone import time # Initialize HTTP client with timeouts http = urllib3.PoolManager( timeout=urllib3.Timeout(connect=5.0, read=30.0), retries=False, ) # Initialize Storage client storage_client = storage.Client() @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch ET Pro IOC reputation lists and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ # Get environment variables bucket_name = os.environ.get('GCS_BUCKET') prefix = os.environ.get('GCS_PREFIX', 'et-pro-ioc').strip('/') state_key = os.environ.get('STATE_KEY', f'{prefix}/state.json') et_api_key = os.environ.get('ET_API_KEY') et_ip_list_url = os.environ.get('ET_IP_LIST_URL') et_domain_list_url = os.environ.get('ET_DOMAIN_LIST_URL') timeout = int(os.environ.get('TIMEOUT', '120')) if not all([bucket_name, et_api_key, et_ip_list_url, et_domain_list_url]): print('Error: Missing required environment variables') return try: # Get GCS bucket bucket = storage_client.bucket(bucket_name) # Generate timestamp for file naming now = datetime.now(timezone.utc) timestamp = now.strftime('%Y/%m/%d/%H%M%S') results = [] errors = [] # Fetch IP reputation list try: print('Fetching IP reputation list...') ip_data = fetch_with_retry(et_ip_list_url, et_api_key, timeout) ip_key = f'{prefix}/ip/{timestamp}.csv' save_to_gcs(bucket, ip_key, ip_data) results.append({'type': 'ip', 'key': ip_key, 'size': len(ip_data)}) print(f'Successfully fetched IP list: {len(ip_data)} bytes') except Exception as e: error_msg = f'Failed to fetch IP list: {str(e)}' print(error_msg) errors.append(error_msg) # Fetch Domain reputation list try: print('Fetching Domain reputation list...') domain_data = fetch_with_retry(et_domain_list_url, et_api_key, timeout) domain_key = f'{prefix}/domain/{timestamp}.csv' save_to_gcs(bucket, domain_key, domain_data) results.append({'type': 'domain', 'key': domain_key, 'size': len(domain_data)}) print(f'Successfully fetched Domain list: {len(domain_data)} bytes') except Exception as e: error_msg = f'Failed to fetch Domain list: {str(e)}' print(error_msg) errors.append(error_msg) # Save state state = { 'last_fetch': now.isoformat(), 'results': results, 'errors': errors } save_state(bucket, state_key, state) if errors: print(f'Completed with {len(errors)} error(s)') else: print('Successfully completed all fetches') except Exception as e: print(f'Error processing logs: {str(e)}') raise def fetch_with_retry(url, api_key, timeout, max_retries=3): """Fetch URL with retry logic for rate limits.""" if not url.lower().startswith('https://'): raise ValueError('Only HTTPS URLs are allowed') headers = {'Authorization': api_key} for attempt in range(max_retries): try: response = http.request('GET', url, headers=headers, timeout=timeout) if response.status == 200: return response.data elif response.status == 429: # Rate limited, wait and retry wait_time = min(30 * (2 ** attempt), 300) print(f'Rate limited, waiting {wait_time}s...') time.sleep(wait_time) else: raise Exception(f'HTTP {response.status}: {response.reason}') except Exception as e: if attempt == max_retries - 1: raise time.sleep(5 * (attempt + 1)) raise Exception(f'Failed to fetch {url} after {max_retries} attempts') def save_to_gcs(bucket, key, content): """Save content to GCS with appropriate content type.""" blob = bucket.blob(key) blob.upload_from_string(content, content_type='text/csv') print(f'Saved {len(content)} bytes to gs://{bucket.name}/{key}') def save_state(bucket, key, state): """Save state to GCS.""" try: blob = bucket.blob(key) blob.upload_from_string( json.dumps(state, indent=2), content_type='application/json' ) except Exception as e: print(f'Warning: Could not save state: {str(e)}')- Deuxième fichier : requirements.txt:
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0Cliquez sur Déployer pour enregistrer et déployer la fonction.
Attendez la fin du déploiement (deux à trois minutes).
Créer une tâche Cloud Scheduler
Cloud Scheduler publie des messages sur le sujet Pub/Sub à intervalles réguliers, ce qui déclenche la fonction Cloud Run.
- Dans la console GCP, accédez à Cloud Scheduler.
- Cliquez sur Créer une tâche.
Fournissez les informations de configuration suivantes :
Paramètre Valeur Nom et-pro-ioc-fetcher-hourlyRégion Sélectionnez la même région que la fonction Cloud Run. Fréquence 0 * * * *(toutes les heures)Fuseau horaire Sélectionnez un fuseau horaire (UTC recommandé). Type de cible Pub/Sub Topic Sélectionner et-pro-ioc-triggerCorps du message {}(objet JSON vide)Cliquez sur Créer.
Options de fréquence de planification
Choisissez la fréquence en fonction du volume de journaux et des exigences de latence :
Fréquence Expression Cron Cas d'utilisation Toutes les heures 0 * * * *Standard (recommandé pour les IOC ET Pro) Toutes les 2 heures 0 */2 * * *Voix plus grave Toutes les 6 heures 0 */6 * * *Mises à jour minimales
Tester l'intégration
- Dans la console Cloud Scheduler, recherchez votre job.
- Cliquez sur Exécuter de force pour déclencher le job manuellement.
- Patientez pendant quelques secondes.
- Accédez à Cloud Run > Services.
- Cliquez sur le nom de la fonction (
et-pro-ioc-fetcher). - Cliquez sur l'onglet Journaux.
Vérifiez que la fonction s'est exécutée correctement. Recherchez les éléments suivants :
Fetching IP reputation list... Successfully fetched IP list: X bytes Fetching Domain reputation list... Successfully fetched Domain list: X bytes Successfully completed all fetchesAccédez à Cloud Storage > Buckets.
Cliquez sur le nom de votre bucket.
Accédez aux dossiers de préfixes (
et-pro-ioc/ip/etet-pro-ioc/domain/).Vérifiez que de nouveaux fichiers
.csvont été créés avec le code temporel actuel.
Si vous constatez des erreurs dans les journaux :
- HTTP 401 : vérifiez ET_API_KEY dans les variables d'environnement
- HTTP 403 : vérifiez que la clé API dispose des autorisations requises.
- HTTP 429 : limitation du débit. La fonction effectuera automatiquement une nouvelle tentative avec un intervalle de temps.
- Variables d'environnement manquantes : vérifiez que toutes les variables requises sont définies.
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,
ET Pro IOC - IP Reputation). - Sélectionnez Google Cloud Storage V2 comme Type de source.
- Sélectionnez Emerging Threats Pro 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 :
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.comCopiez cette adresse e-mail pour l'utiliser à l'étape suivante.
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 de votre bucket.
- 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.
- Attribuez des rôles : sélectionnez Lecteur des objets de l'espace de stockage.
Cliquez sur Enregistrer.
Configurer des flux dans Google SecOps pour ingérer les journaux d'indicateurs de compromission Proofpoint Emerging Threats Pro
Vous devez créer deux flux distincts : un pour la réputation des adresses IP et un pour la réputation des domaines.
Créer un flux de réputation IP
- 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
ET Pro IOC - IP Reputation. - Sélectionnez Google Cloud Storage V2 comme Type de source.
- Sélectionnez Emerging Threats Pro comme type de journal.
- Cliquez sur Suivant.
Spécifiez les valeurs des paramètres d'entrée suivants :
URL du bucket Storage : saisissez l'URI du bucket GCS avec le préfixe du chemin d'accès :
gs://et-pro-ioc-bucket/et-pro-ioc/ip/- Remplacez
et-pro-ioc-bucketpar le nom réel de votre bucket GCS.
- Remplacez
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. La valeur par défaut est de 180 jours.
Espace de noms de l'élément : espace de noms de l'élément.
Libellés d'ingestion : libellé appliqué 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.
Créer un flux de réputation de domaine
Répétez le processus de création du flux :
- 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
ET Pro IOC - Domain Reputation. - Sélectionnez Google Cloud Storage V2 comme Type de source.
- Sélectionnez Emerging Threats Pro comme type de journal.
- Cliquez sur Suivant.
Spécifiez les valeurs des paramètres d'entrée suivants :
- URL du bucket Storage : saisissez l'URI du bucket GCS avec le préfixe du chemin d'accès :
gs://et-pro-ioc-bucket/et-pro-ioc/domain/Remplacez
et-pro-ioc-bucketpar le nom réel de votre bucket GCS.- Option de suppression de la source : sélectionnez l'option de votre choix.
- Âge maximal des fichiers : incluez les fichiers modifiés au cours des derniers jours. La valeur par défaut est de 180 jours.
- Espace de noms de l'élément : espace de noms de l'élément.
- Libellés d'ingestion : libellé appliqué 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.
Table de mappage UDM
| Champ du journal | Mappage UDM | Logique |
|---|---|---|
| catégorie | Ce champ est utilisé dans la logique du parseur, mais n'est pas directement mappé à l'UDM. | Elle détermine la valeur de event.ioc.categorization à l'aide d'une table de recherche. |
| collection_time.nanos | event.idm.entity.metadata.collected_timestamp.nanos | Directement mappé à partir du journal brut. |
| collection_time.seconds | event.idm.entity.metadata.collected_timestamp.seconds | Directement mappé à partir du journal brut. |
| données | Ce champ est analysé en plusieurs champs UDM en fonction de son contenu. | |
| first_seen | event.idm.entity.metadata.interval.start_time | Analysée comme une date et mappée à l'UDM. |
| first_seen | event.ioc.active_timerange.start | Analysée comme une date et mappée à l'UDM. |
| ip_or_domain | event.idm.entity.entity.hostname | Mappé à l'UDM si le modèle grok extrait un hôte du champ. |
| ip_or_domain | event.idm.entity.entity.ip | Mappé à l'UDM si le modèle grok n'extrait pas d'hôte du champ. |
| ip_or_domain | event.ioc.domain_and_ports.domain | Mappé à l'UDM si le modèle grok extrait un hôte du champ. |
| ip_or_domain | event.ioc.ip_and_ports.ip_address | Mappé à l'UDM si le modèle grok n'extrait pas d'hôte du champ. |
| last_seen | event.idm.entity.metadata.interval.end_time | Analysée comme une date et mappée à l'UDM. |
| last_seen | event.ioc.active_timerange.end | Analysée comme une date et mappée à l'UDM. |
| ports | event.idm.entity.entity.labels.value | Analysés, joints avec un délimiteur de virgule et mappés à l'UDM s'il existe plusieurs ports. |
| ports | event.idm.entity.entity.port | Analysé et mappé à l'UDM s'il n'y a qu'un seul port. |
| ports | event.ioc.domain_and_ports.ports | Analysé et mappé sur l'UDM si le modèle Grok extrait un hôte du champ. |
| ports | event.ioc.ip_and_ports.ports | Analysé et mappé sur l'UDM si le modèle Grok n'extrait pas d'hôte du champ. |
| score | event.ioc.confidence_score | Directement mappé à partir du journal brut. |
| event.idm.entity.entity.labels.key | Définissez la valeur sur "ports" s'il existe plusieurs ports. | |
| event.idm.entity.metadata.entity_type | Définissez la valeur sur "DOMAIN_NAME" si le modèle Grok extrait un hôte du champ ip_or_domain. Sinon, définissez-la sur "IP_ADDRESS". | |
| event.idm.entity.metadata.threat.category | Défini sur "SOFTWARE_MALICIOUS". | |
| event.idm.entity.metadata.threat.category_details | Dérivé du champ "Catégorie" à l'aide d'une table de référence. | |
| event.idm.entity.metadata.threat.threat_name | Définissez-le sur "ET Intelligence Rep List". | |
| event.idm.entity.metadata.vendor_name | Définissez-le sur "ET_PRO_IOC". | |
| event.ioc.feed_name | Définissez-le sur "ET Intelligence Rep List". | |
| event.ioc.raw_severity | Défini sur "Malveillant". | |
| timestamp.nanos | Copié depuis collection_time.nanos. | |
| timestamp.seconds | Copié depuis collection_time.seconds. |
Vous avez encore besoin d'aide ? Obtenez des réponses de membres de la communauté et de professionnels Google SecOps.