Collecter les journaux Group-IB Threat Intelligence
Ce document explique comment ingérer des journaux Group-IB Threat Intelligence dans Google Security Operations à l'aide de Google Cloud Storage.
Group-IB Threat Intelligence & Attribution (TI&A) est une plate-forme de renseignement sur les cybermenaces qui fournit des données en temps réel sur les acteurs malveillants, les indicateurs de compromission (IOC), les logiciels malveillants, l'infrastructure de commande et de contrôle (C2), les identifiants compromis, les campagnes d'hameçonnage et les failles. Il regroupe les renseignements provenant de sources du Web ouvert, du deep Web et du dark Web, ce qui permet aux équipes de sécurité de détecter les menaces et d'y répondre de manière proactive.
Avant de commencer
Assurez-vous de remplir les conditions suivantes :
- Une instance Google SecOps
- Un compte Group-IB TI&A avec accès à l'API activé
- Accès au portail Group-IB TI&A (
tap.group-ib.com) - Un projet Google Cloud avec les API suivantes activées :
- API Cloud Storage
- API Cloud Functions
- API Cloud Scheduler
- API Cloud Build
Générer une clé API Group-IB
- Connectez-vous au portail Group-IB TI&A sur
https://tap.group-ib.com. - Cliquez sur votre nom en haut à droite, puis sélectionnez Profil.
- Cliquez sur Accéder à mes paramètres.
- Accédez à l'onglet Sécurité et accès.
- Dans la section Jeton personnel, cliquez sur Générer un jeton.
Copiez et enregistrez la clé API de manière sécurisée.
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 Attribuez un nom au bucket Saisissez un nom unique (par exemple, groupib-ti-logs).Type d'emplacement Choisissez en fonction de vos besoins (région, birégion, 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é) Cliquez sur Créer.
Déployer une fonction Cloud pour extraire les données Group-IB
Créez une fonction Cloud qui extrait les données de renseignements sur les menaces de l'API Group-IB TI&A et les écrit dans le bucket GCS sous forme de fichiers NDJSON pour que Google SecOps puisse les ingérer.
Créer la fonction Cloud
- Accédez à la console Google Cloud.
- Accédez à Cloud Functions.
- Cliquez sur Créer une fonction.
Fournissez les informations de configuration suivantes :
Paramètre Valeur Environnement 2e génération Nom de la fonction groupib-to-gcsRégion Sélectionnez la région la plus proche de votre bucket GCS. Type de déclencheur HTTPS Authentification Exiger l'authentification Mémoire allouée 512 Mo (augmentez cette valeur si vous récupérez de grandes collections) Délai avant expiration 540 secondes Cliquez sur Suivant.
Définissez le champ Environnement d'exécution sur
Python 3.11(ou version ultérieure).Définissez le champ Point d'entrée sur
main.Remplacez le contenu de
main.pypar le code suivant :import json import os import requests import functions_framework from datetime import datetime, timedelta, timezone from urllib.parse import urljoin from requests.auth import HTTPBasicAuth from google.cloud import storage GIB_API_URL = os.environ.get('GIB_API_URL', 'https://tap.group-ib.com/api/v2/') GIB_USERNAME = os.environ.get('GIB_USERNAME') GIB_API_KEY = os.environ.get('GIB_API_KEY') GCS_BUCKET = os.environ.get('GCS_BUCKET') GCS_PREFIX = os.environ.get('GCS_PREFIX', 'groupib-ti') COLLECTIONS = os.environ.get('GIB_COLLECTIONS', 'compromised/account,malware/cnc,apt/threat,hi/threat').split(',') DEFAULT_DAYS_BACK = int(os.environ.get('DEFAULT_DAYS_BACK', '3')) # Max items per request: 100 for most collections, 20 for apt/threat and hi/threat BIG_DATA_COLLECTIONS = ['apt/threat', 'hi/threat'] # File to persist seqUpdate values between runs (use GCS for durability) STATE_BLOB = '_state/seq_updates.json' def load_state(): """Load seqUpdate state from GCS.""" client = storage.Client() bucket = client.bucket(GCS_BUCKET) blob = bucket.blob(f"{GCS_PREFIX}/{STATE_BLOB}") if blob.exists(): return json.loads(blob.download_as_text()) return {} def save_state(state): """Save seqUpdate state to GCS.""" client = storage.Client() bucket = client.bucket(GCS_BUCKET) blob = bucket.blob(f"{GCS_PREFIX}/{STATE_BLOB}") blob.upload_from_string(json.dumps(state), content_type='application/json') def gib_request(session, url, params=None): """Send authenticated GET request to Group-IB API.""" resp = session.get(url, params=params) if resp.status_code == 301: raise Exception('IP not whitelisted by Group-IB. Contact Group-IB support.') resp.raise_for_status() return resp.json() def get_seq_update_by_date(session, collection, date_str): """Get seqUpdate value for a collection starting from a given date.""" url = urljoin(GIB_API_URL, 'sequence_list') data = gib_request(session, url, {'date': date_str, 'collection': collection}) return data.get('list', {}).get(collection) def fetch_collection(session, collection, seq_update): """Fetch all new items from a collection starting after the given seqUpdate.""" limit = 20 if collection in BIG_DATA_COLLECTIONS else 100 url = urljoin(GIB_API_URL, f"{collection}/updated") all_items = [] last_seq = seq_update while True: data = gib_request(session, url, {'seqUpdate': str(last_seq), 'limit': limit}) items = data.get('items', []) if not items: break all_items.extend(items) last_seq = items[-1].get('seqUpdate') return all_items, last_seq def write_to_gcs(items, collection_name): """Write items to GCS as NDJSON.""" if not items: return 0 client = storage.Client() bucket = client.bucket(GCS_BUCKET) timestamp = datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S') safe_name = collection_name.replace('/', '_') blob_path = f"{GCS_PREFIX}/{safe_name}_{timestamp}.ndjson" blob = bucket.blob(blob_path) ndjson = '\n'.join(json.dumps(item, ensure_ascii=False) for item in items) + '\n' blob.upload_from_string(ndjson, content_type='application/x-ndjson') return len(items) @functions_framework.http def main(request): """Cloud Function entry point.""" session = requests.Session() session.auth = HTTPBasicAuth(GIB_USERNAME, GIB_API_KEY) session.headers.update({'Accept': '*/*'}) state = load_state() total = 0 for collection in COLLECTIONS: collection = collection.strip() seq_update = state.get(collection) if seq_update is None: default_date = (datetime.now(timezone.utc) - timedelta(days=DEFAULT_DAYS_BACK)).strftime('%Y-%m-%d') seq_update = get_seq_update_by_date(session, collection, default_date) if seq_update is None: continue items, last_seq = fetch_collection(session, collection, seq_update) if items: write_to_gcs(items, collection) total += len(items) state[collection] = last_seq save_state(state) return json.dumps({'status': 'success', 'total_items': total}), 200Remplacez le contenu de
requirements.txtpar les dépendances suivantes :functions-framework==3.* requests>=2.28.0 google-cloud-storage>=2.0.0Cliquez sur Déployer.
Configurer les variables d'environnement
- Après le déploiement, accédez à la page d'informations sur votre fonction.
- Cliquez sur Modifier.
- Développez la section Paramètres d'exécution, de compilation, de connexion et de sécurité.
Sous Variables d'environnement d'exécution, ajoutez les variables suivantes :
Variable Valeur GIB_API_URLhttps://tap.group-ib.com/api/v2/(ouhttps://bt.group-ib.com/api/v2/selon votre région)GIB_USERNAMEAdresse e-mail de votre compte Group-IB GIB_API_KEYVotre clé API Group-IB (jeton personnel) GCS_BUCKETNom de votre bucket GCS (par exemple, groupib-ti-logs)GCS_PREFIXPréfixe des fichiers journaux (par exemple, groupib-ti)GIB_COLLECTIONSListe des collections à récupérer, séparées par une virgule (voir ci-dessous) DEFAULT_DAYS_BACKNombre de jours à analyser lors de la première exécution (par défaut : 3)Cliquez sur Déployer.
Collections Group-IB disponibles
Configurez la variable GIB_COLLECTIONS avec les collections pertinentes pour votre cas d'utilisation :
| Collection | Description | Limite maximale |
|---|---|---|
compromised/account |
Identifiants de compte piratés (identifiant, mot de passe, domaine) | 100 |
compromised/card |
Cartes bancaires piratées | 100 |
compromised/mule |
Comptes de mules financières | 100 |
compromised/imei |
IMEI des appareils mobiles dont la sécurité est compromise | 100 |
compromised/file |
Fichiers compromis avec attribution de logiciels malveillants | 100 |
attacks/ddos |
Données sur les attaques DDoS (adresses IP cibles, domaines) | 100 |
attacks/deface |
Incidents de défiguration de site Web | 100 |
attacks/phishing |
URL et domaines d'hameçonnage | 100 |
attacks/phishing_kit |
Hachages de kits d'hameçonnage et marques cibles | 100 |
bp/phishing |
Protection de la marque : incidents d'hameçonnage | 100 |
bp/phishing_kit |
Protection des marques : kits d'hameçonnage | 100 |
hi/threat |
Rapports sur les menaces cybercriminelles (HI) avec des IOC et un mappage MITRE ATT&CK | 20 |
hi/threat_actor |
Profils d'acteurs de cybermenaces | 100 |
apt/threat |
Rapports sur les menaces APT (États-nations) avec des IOC et un mappage MITRE ATT&CK | 20 |
apt/threat_actor |
Profils d'acteurs malveillants APT | 100 |
malware/cnc |
Indicateurs de serveur de commande et de contrôle (adresses IP, domaines) | 100 |
malware/malware |
Descriptions des logiciels malveillants et niveaux de menace | 100 |
malware/targeted_malware |
Exemples de logiciels malveillants ciblés (hachages, noms de fichiers) | 100 |
osi/git_leak |
Fuites de données de dépôt Git | 100 |
osi/public_leak |
Fuites de données publiques (copier-coller, dumps) | 100 |
osi/vulnerability |
Données sur les failles avec scores CVSS | 100 |
suspicious_ip/tor_node |
Adresses IP des nœuds de sortie Tor | 100 |
suspicious_ip/open_proxy |
Adresses IP de proxy ouvert | 100 |
suspicious_ip/socks_proxy |
Adresses IP du proxy SOCKS | 100 |
Exemple : Pour collecter des données sur les identifiants compromis, l'infrastructure C2 et les APT :
compromised/account,malware/cnc,apt/threat,hi/threat
Planifier la fonction Cloud
Utilisez Cloud Scheduler pour déclencher la fonction à intervalles réguliers.
- Accédez à Cloud Scheduler dans la Google Cloud Console.
- Cliquez sur Créer une tâche.
Fournissez les informations de configuration suivantes :
Paramètre Valeur Nom groupib-to-gcs-scheduleRégion La même région que votre fonction Cloud Fréquence 0 */1 * * *(toutes les heures) ou0 0 * * *(tous les jours)Fuseau horaire Sélectionnez votre fuseau horaire. Sous Configurer l'exécution :
- Type de cible : sélectionnez HTTP.
- URL : saisissez l'URL du déclencheur de la fonction Cloud.
- Méthode HTTP : sélectionnez POST.
- En-tête d'authentification : sélectionnez Ajouter un jeton OIDC.
- Compte de service : sélectionnez un compte de service disposant de l'autorisation
roles/cloudfunctions.invoker.
Cliquez sur Créer.
Tester l'exportation de données
- Dans la console Cloud Scheduler, cliquez sur Forcer l'exécution à côté de votre job pour déclencher la fonction manuellement.
- Consultez les journaux Cloud Functions dans Cloud Logging pour vérifier que les données ont bien été récupérées depuis Group-IB.
- Accédez à Cloud Storage > Buckets dans la Google Cloud Console.
- Cliquez sur le nom de votre bucket (par exemple,
groupib-ti-logs). - Accédez au dossier de préfixe (par exemple,
groupib-ti/). Vérifiez que de nouveaux fichiers
.ndjsonapparaissent dans le bucket.
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 Ajouter un flux.
- Cliquez sur Configurer un flux unique.
- Dans le champ Nom du flux, saisissez un nom pour le flux (par exemple,
Group-IB Threat Intelligence). - Sélectionnez Google Cloud Storage V2 comme Type de source.
- Sélectionnez Group-IB Renseignement sur les menaces comme type de journal.
Cliquez sur Obtenir un compte de service. Une adresse e-mail de compte de service unique 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 (par exemple,
groupib-ti-logs). - 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.
Configurer un flux dans Google SecOps pour ingérer les journaux Group-IB Threat Intelligence
- Accédez à Paramètres SIEM> Flux.
- Cliquez sur Ajouter un flux.
- Cliquez sur Configurer un flux unique.
- Dans le champ Nom du flux, saisissez un nom pour le flux (par exemple,
Group-IB Threat Intelligence). - Sélectionnez Google Cloud Storage V2 comme Type de source.
- Sélectionnez Group-IB Threat Intelligence comme Type de journal.
- 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://groupib-ti-logs/groupib-ti/Remplacez :
groupib-ti-logs: nom de votre bucket GCS.groupib-ti: préfixe/chemin d'accès au dossier dans lequel les journaux sont stockés.
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 une fois le transfert réussi.
- Supprimer les fichiers transférés et les répertoires vides : supprime les fichiers et les répertoires vides une fois le transfert réussi.
Âge maximal du fichier : 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é à appliquer aux événements de ce flux.
Cliquez sur Suivant.
Vérifiez la configuration de votre nouveau flux d'actualités sur l'écran Finaliser, puis cliquez sur Envoyer.
Vous avez encore besoin d'aide ? Obtenez des réponses de membres de la communauté et de professionnels Google SecOps.