Collecter les journaux Group-IB Threat Intelligence

Compatible avec :

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

  1. Connectez-vous au portail Group-IB TI&A sur https://tap.group-ib.com.
  2. Cliquez sur votre nom en haut à droite, puis sélectionnez Profil.
  3. Cliquez sur Accéder à mes paramètres.
  4. Accédez à l'onglet Sécurité et accès.
  5. Dans la section Jeton personnel, cliquez sur Générer un jeton.
  6. Copiez et enregistrez la clé API de manière sécurisée.

Créer un bucket Google Cloud Storage

  1. Accédez à la console Google Cloud.
  2. Sélectionnez votre projet ou créez-en un.
  3. Dans le menu de navigation, accédez à Cloud Storage > Buckets.
  4. Cliquez sur Créer un bucket.
  5. 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é)
  6. 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

  1. Accédez à la console Google Cloud.
  2. Accédez à Cloud Functions.
  3. Cliquez sur Créer une fonction.
  4. Fournissez les informations de configuration suivantes :

    Paramètre Valeur
    Environnement 2e génération
    Nom de la fonction groupib-to-gcs
    Ré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
  5. Cliquez sur Suivant.

  6. Définissez le champ Environnement d'exécution sur Python 3.11 (ou version ultérieure).

  7. Définissez le champ Point d'entrée sur main.

  8. Remplacez le contenu de main.py par 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}), 200
    
  9. Remplacez le contenu de requirements.txt par les dépendances suivantes :

    functions-framework==3.*
    requests>=2.28.0
    google-cloud-storage>=2.0.0
    
  10. Cliquez sur Déployer.

Configurer les variables d'environnement

  1. Après le déploiement, accédez à la page d'informations sur votre fonction.
  2. Cliquez sur Modifier.
  3. Développez la section Paramètres d'exécution, de compilation, de connexion et de sécurité.
  4. Sous Variables d'environnement d'exécution, ajoutez les variables suivantes :

    Variable Valeur
    GIB_API_URL https://tap.group-ib.com/api/v2/ (ou https://bt.group-ib.com/api/v2/ selon votre région)
    GIB_USERNAME Adresse e-mail de votre compte Group-IB
    GIB_API_KEY Votre clé API Group-IB (jeton personnel)
    GCS_BUCKET Nom de votre bucket GCS (par exemple, groupib-ti-logs)
    GCS_PREFIX Préfixe des fichiers journaux (par exemple, groupib-ti)
    GIB_COLLECTIONS Liste des collections à récupérer, séparées par une virgule (voir ci-dessous)
    DEFAULT_DAYS_BACK Nombre de jours à analyser lors de la première exécution (par défaut : 3)
  5. 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.

  1. Accédez à Cloud Scheduler dans la Google Cloud Console.
  2. Cliquez sur Créer une tâche.
  3. Fournissez les informations de configuration suivantes :

    Paramètre Valeur
    Nom groupib-to-gcs-schedule
    Région La même région que votre fonction Cloud
    Fréquence 0 */1 * * * (toutes les heures) ou 0 0 * * * (tous les jours)
    Fuseau horaire Sélectionnez votre fuseau horaire.
  4. 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.
  5. Cliquez sur Créer.

Tester l'exportation de données

  1. Dans la console Cloud Scheduler, cliquez sur Forcer l'exécution à côté de votre job pour déclencher la fonction manuellement.
  2. 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.
  3. Accédez à Cloud Storage > Buckets dans la Google Cloud Console.
  4. Cliquez sur le nom de votre bucket (par exemple, groupib-ti-logs).
  5. Accédez au dossier de préfixe (par exemple, groupib-ti/).
  6. Vérifiez que de nouveaux fichiers .ndjson apparaissent 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

  1. Accédez à Paramètres SIEM> Flux.
  2. Cliquez sur Ajouter un flux.
  3. Cliquez sur Configurer un flux unique.
  4. Dans le champ Nom du flux, saisissez un nom pour le flux (par exemple, Group-IB Threat Intelligence).
  5. Sélectionnez Google Cloud Storage V2 comme Type de source.
  6. Sélectionnez Group-IB Renseignement sur les menaces comme type de journal.
  7. 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.com
    
  8. Copiez 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.

  1. Accédez à Cloud Storage > Buckets.
  2. Cliquez sur le nom de votre bucket (par exemple, groupib-ti-logs).
  3. Accédez à l'onglet Autorisations.
  4. Cliquez sur Accorder l'accès.
  5. 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.
  6. Cliquez sur Enregistrer.

Configurer un flux dans Google SecOps pour ingérer les journaux Group-IB Threat Intelligence

  1. Accédez à Paramètres SIEM> Flux.
  2. Cliquez sur Ajouter un flux.
  3. Cliquez sur Configurer un flux unique.
  4. Dans le champ Nom du flux, saisissez un nom pour le flux (par exemple, Group-IB Threat Intelligence).
  5. Sélectionnez Google Cloud Storage V2 comme Type de source.
  6. Sélectionnez Group-IB Threat Intelligence comme Type de journal.
  7. Cliquez sur Suivant.
  8. 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.

  9. Cliquez sur Suivant.

  10. 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.