Collecter les journaux de la plate-forme d'isolation Menlo Security (MSIP)

Compatible avec :

Ce guide explique comment ingérer les journaux Menlo Security Isolation Platform (MSIP) dans Google SecOps à l'aide de Google Cloud Storage.

L'approche de Menlo Security, axée sur l'isolation, divise la navigation Web et la récupération de documents entre l'appareil de l'utilisateur et un conteneur virtuel jetable (DVC) isolé, loin du point de terminaison. Tout code risqué est exécuté dans le DVC isolé et n'atteint jamais le point de terminaison. Seules les données d'affichage sécurisées sont envoyées au navigateur de l'utilisateur. La plate-forme fournit une journalisation complète pour l'accès Web, les événements d'audit, la sécurité des e-mails, les cas de non-respect de la protection contre la perte de données et l'activité SMTP.

Avant de commencer

Assurez-vous de remplir les conditions préalables suivantes :

  • Une instance Google SecOps
  • Accès privilégié au portail d'administration Menlo Security
  • Autorisations permettant de créer des jetons d'API dans Menlo Security
  • Un projet GCP avec les API Cloud Storage, Cloud Run, Pub/Sub et Cloud Scheduler activées

Configurer l'accès à l'API Menlo Security

Pour permettre à la fonction Cloud Run de récupérer les journaux, vous devez créer un jeton d'API avec l'autorisation de l'API Log Export.

Créer un jeton d'API

  1. Connectez-vous au portail d'administration Menlo Security à l'adresse https://admin.menlosecurity.com.
  2. Accédez à Settings> Authentication> API Tokens (Paramètres > Authentification > Jetons d'API).
  3. Cliquez sur Créer un jeton.
  4. Dans le champ Nom du jeton, saisissez un nom descriptif (par exemple, Chronicle GCS Integration).
  5. Dans la section Autorisations, sélectionnez API Log Export.
  6. Cliquez sur Créer ou sur Enregistrer.
  7. Copiez et stockez de manière sécurisée le jeton d'API généré.

Important : Le jeton d'API n'est affiché qu'une seule fois. Stockez-le dans un emplacement sécurisé immédiatement après l'avoir créé.

Autorisations d'API requises

  • Le jeton d'API nécessite l'autorisation suivante :

    Autorisation/Champ d'application Niveau d'accès Objectif
    API Log Export Lire Récupérer les journaux de l'API Menlo Security

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, menlo-security-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é)
    Outils de protection Facultatif : Activez la gestion des versions des objets ou la règle de conservation.

  6. Cliquez sur Créer.

Créer un compte de service pour la fonction Cloud Run

Créer un compte de service

  1. Dans la console GCP, accédez à IAM et administration > Comptes de service.
  2. Cliquez sur Créer un compte de service.
  3. Fournissez les informations de configuration suivantes :
    • Nom du compte de service : saisissez menlo-logs-collector-sa.
    • Description du compte de service : saisissez Service account for Cloud Run function to collect Menlo Security logs.
  4. Cliquez sur Créer et continuer.
  5. Dans la section Autoriser ce compte de service à accéder au projet, ajoutez les rôles suivants :
    1. Cliquez sur Sélectionner un rôle.
    2. Recherchez et sélectionnez Administrateur des objets Storage.
    3. Cliquez sur + Ajouter un autre rôle.
    4. Recherchez et sélectionnez Demandeur Cloud Run.
    5. Cliquez sur + Ajouter un autre rôle.
    6. Recherchez et sélectionnez Demandeur Cloud Functions.
  6. Cliquez sur Continuer.
  7. Cliquez sur OK.

Accorder des autorisations IAM sur le bucket GCS

  1. Accédez à Cloud Storage > Buckets.
  2. Cliquez sur le nom de votre bucket (menlo-security-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 : saisissez l'adresse e-mail du compte de service (menlo-logs-collector-sa@PROJECT_ID.iam.gserviceaccount.com).
    • Attribuer des rôles : sélectionnez Administrateur des objets Storage.
  6. Cliquez sur Enregistrer.

Créer un sujet Pub/Sub

  1. Dans la console GCP, accédez à Pub/Sub > Sujets.
  2. Cliquez sur Créer un sujet.
  3. Fournissez les informations de configuration suivantes :
    • ID du thème : saisissez menlo-logs-trigger.
    • Conservez les valeurs par défaut des autres paramètres.
  4. Cliquez sur Créer.

Créer une fonction Cloud Run pour collecter les journaux

  1. Dans la console GCP, accédez à Cloud Run.
  2. Cliquez sur Créer un service.
  3. Sélectionnez Fonction (utilisez un éditeur intégré pour créer une fonction).

  4. Dans la section Configurer, fournissez les informations de configuration suivantes :

    Paramètre Valeur
    Nom du service menlo-logs-collector
    Région Sélectionnez la région correspondant à votre bucket GCS (par exemple, us-central1).
    Durée d'exécution Sélectionnez Python 3.12 ou version ultérieure.

  5. Dans la section Déclencheur (facultatif) :

    1. Cliquez sur + Ajouter un déclencheur.
    2. Sélectionnez Cloud Pub/Sub.
    3. Dans Sélectionner un sujet Cloud Pub/Sub, sélectionnez menlo-logs-trigger.
    4. Cliquez sur Enregistrer.

  6. Dans la section Authentification :

    1. Sélectionnez Exiger l'authentification.
    2. Consultez Identity and Access Management (IAM).

  7. Faites défiler la page vers le bas, puis développez Conteneurs, mise en réseau, sécurité.

  8. Accédez à l'onglet Sécurité :

    • Compte de service : sélectionnez menlo-logs-collector-sa.

  9. Accédez à l'onglet Conteneurs :

    1. Cliquez sur Variables et secrets.
    2. Cliquez sur + Ajouter une variable pour chaque variable d'environnement :
    Nom de la variable Exemple de valeur Description
    GCS_BUCKET menlo-security-logs Nom du bucket GCS
    GCS_PREFIX menlo-logs Préfixe des fichiers journaux
    STATE_KEY menlo-logs/state.json Chemin d'accès au fichier d'état
    MENLO_API_TOKEN your-menlo-api-token Jeton d'API Menlo Security
    MENLO_API_HOST logs.menlosecurity.com Hôte de l'API Menlo Security
    MENLO_LOG_TYPES web,audit,email,dlp Types de journaux séparés par une virgule
    LOOKBACK_HOURS 1 Période d'analyse initiale

  10. Dans la section Variables et secrets, faites défiler la page jusqu'à Requêtes :

    • Délai avant expiration de la requête : saisissez 600 secondes (10 minutes).

  11. Accédez à l'onglet Paramètres :

    • Dans la section Ressources :
      • Mémoire : sélectionnez 512 Mio ou plus.
      • CPU : sélectionnez 1.

  12. Dans la section Scaling de révision :

    • Nombre minimal d'instances : saisissez 0.
    • Nombre maximal d'instances : saisissez 100.

  13. Cliquez sur Créer.

  14. Attendez que le service soit créé (une à deux minutes).

  15. Une fois le service créé, l'éditeur de code intégré s'ouvre automatiquement.

Ajouter un code de fonction

  1. Saisissez main dans le champ Point d'entrée.

  2. Dans l'éditeur de code intégré, créez deux fichiers :

    • main.py:
    import functions_framework
from google.cloud import storage
import os
import json
import urllib3
import time
import uuid
from datetime import datetime, timezone, timedelta

http = urllib3.PoolManager(
  timeout=urllib3.Timeout(connect=5.0, read=60.0),
  retries=False,
)

storage_client = storage.Client()

GCS_BUCKET = os.environ.get('GCS_BUCKET')
GCS_PREFIX = os.environ.get('GCS_PREFIX', 'menlo-logs')
STATE_KEY = os.environ.get('STATE_KEY', 'menlo-logs/state.json')
API_TOKEN = os.environ.get('MENLO_API_TOKEN')
API_HOST = os.environ.get('MENLO_API_HOST', 'logs.menlosecurity.com')
LOG_TYPES = os.environ.get('MENLO_LOG_TYPES', 'web').split(',')
LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '1'))

@functions_framework.cloud_event
def main(cloud_event):
  if not all([GCS_BUCKET, API_TOKEN, API_HOST]):
    print('Error: Missing required environment variables')
    return

  try:
    bucket = storage_client.bucket(GCS_BUCKET)
    state = load_state(bucket)
    now = datetime.now(timezone.utc).replace(microsecond=0)

    if isinstance(state, dict) and state.get('last_event_time'):
      try:
        last_val = state['last_event_time']
        if last_val.endswith('Z'):
          last_val = last_val[:-1] + '+00:00'
        last_time = datetime.fromisoformat(last_val)
        last_time = last_time - timedelta(minutes=2)
      except Exception as e:
        print(f"Warning: Could not parse last_event_time: {e}")
        last_time = now - timedelta(hours=LOOKBACK_HOURS)
    else:
      last_time = now - timedelta(hours=LOOKBACK_HOURS)

    print(f"Fetching logs from {last_time.isoformat()} to {now.isoformat()}")

    total = 0
    newest_time = None

    for log_type in LOG_TYPES:
      log_type = log_type.strip()
      print(f"Fetching {log_type} logs...")
      events = fetch_logs(log_type, API_TOKEN, API_HOST, last_time, now)

      if events:
        object_key = f"{GCS_PREFIX}/{now:%Y/%m/%d}/menlo-{log_type}-{uuid.uuid4()}.ndjson"
        blob = bucket.blob(object_key)

        ndjson = '\n'.join(
          [json.dumps(ev, ensure_ascii=False) for ev in events]
        ) + '\n'
        blob.upload_from_string(ndjson, content_type='application/x-ndjson')

        print(f"Uploaded {len(events)} {log_type} events to gs://{GCS_BUCKET}/{object_key}")
        total += len(events)

        for ev in events:
          ev_time = ev.get('timestamp') or ev.get('time') or ev.get('created_at')
          if ev_time and (newest_time is None or str(ev_time) > str(newest_time)):
            newest_time = str(ev_time)
      else:
        print(f"No {log_type} events found")

    save_state(bucket, newest_time if newest_time else now.isoformat())
    print(f"Successfully processed {total} total events")

  except Exception as e:
    print(f'Error processing logs: {str(e)}')
    raise

def fetch_logs(log_type, token, api_host, start, end):
  url = f"https://{api_host}/api/v1/logs/{log_type}"
  headers = {
    'Authorization': f'Token {token}',
    'Accept': 'application/json',
  }
  params = {
    'start': start.strftime('%Y-%m-%dT%H:%M:%SZ'),
    'end': end.strftime('%Y-%m-%dT%H:%M:%SZ'),
  }
  query = '&'.join(f'{k}={v}' for k, v in params.items())
  full_url = f'{url}?{query}'

  all_events = []
  backoff = 1.0

  while full_url:
    response = http.request('GET', full_url, headers=headers)

    if response.status == 429:
      retry_after = int(response.headers.get('Retry-After', str(int(backoff))))
      print(f"Rate limited. Waiting {retry_after}s...")
      time.sleep(retry_after)
      backoff = min(backoff * 2, 60.0)
      continue

    if response.status != 200:
      print(f"API error {response.status}: {response.data.decode('utf-8')}")
      break

    backoff = 1.0
    data = json.loads(response.data.decode('utf-8'))
    events = data if isinstance(data, list) else data.get('events', data.get('data', []))
    all_events.extend(events)

    full_url = data.get('next') if isinstance(data, dict) else None

  print(f"Total {log_type} events fetched: {len(all_events)}")
  return all_events

def load_state(bucket):
  try:
    blob = bucket.blob(STATE_KEY)
    if blob.exists():
      return json.loads(blob.download_as_text())
  except Exception as e:
    print(f"Warning: Could not load state: {e}")
  return {}

def save_state(bucket, last_event_time_iso):
  try:
    state = {
      'last_event_time': last_event_time_iso,
      'last_run': datetime.now(timezone.utc).isoformat()
    }
    blob = bucket.blob(STATE_KEY)
    blob.upload_from_string(
      json.dumps(state, indent=2),
      content_type='application/json'
    )
    print(f"Saved state: last_event_time={last_event_time_iso}")
  except Exception as e:
    print(f"Warning: Could not save state: {e}")
    • requirements.txt:
    functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0

  3. Cliquez sur Déployer pour enregistrer et déployer la fonction.

  4. Attendez la fin du déploiement (deux à trois minutes).

Créer une tâche Cloud Scheduler

  1. Dans la console GCP, accédez à Cloud Scheduler.
  2. Cliquez sur Créer une tâche.

  3. Fournissez les informations de configuration suivantes :

    Paramètre Valeur
    Nom menlo-logs-collector-hourly
    Région Sélectionnez la même région que la fonction Cloud Run.
    Fréquence 0 * * * * (toutes les heures, à l'heure pile)
    Fuseau horaire Sélectionnez un fuseau horaire (UTC recommandé).
    Type de cible Pub/Sub
    Sujet Sélectionner menlo-logs-trigger
    Corps du message {} (objet JSON vide)

  4. Cliquez sur Créer.

Tester l'intégration

  1. Dans la console Cloud Scheduler, recherchez votre job (menlo-logs-collector-hourly).
  2. Cliquez sur Exécuter de force pour déclencher le job manuellement.
  3. Patientez quelques secondes.
  4. Accédez à Cloud Run > Services.
  5. Cliquez sur menlo-logs-collector.
  6. Cliquez sur l'onglet Journaux.

  7. Vérifiez que la fonction s'est exécutée correctement. Par exemple :

    Fetching logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00
Fetching web logs...
Total web events fetched: X
Uploaded X web events to gs://menlo-security-logs/menlo-logs/YYYY/MM/DD/menlo-web-UUID.ndjson
Successfully processed X total events

  8. Accédez à Cloud Storage > Buckets.

  9. Cliquez sur menlo-security-logs.

  10. Accédez au dossier menlo-logs/.

  11. Vérifiez que de nouveaux fichiers .ndjson ont été créés avec la date actuelle.

Si des erreurs s'affichent dans les journaux :

  • HTTP 401 : vérifiez que la variable d'environnement MENLO_API_TOKEN est correcte.
  • HTTP 403 : vérifiez que le jeton d'API dispose de l'autorisation API Log Export.
  • HTTP 429 : limitation du débit (la fonction effectuera automatiquement une nouvelle tentative avec un intervalle entre les tentatives)
  • Variables d'environnement manquantes : vérifiez que toutes les variables requises sont définies dans la configuration de la fonction Cloud Run.

Récupérer le compte de service Google SecOps

  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, Menlo Security MSIP Logs).
  5. Sélectionnez Google Cloud Storage V2 comme Type de source.
  6. Sélectionnez Menlo Security 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.

  9. Cliquez sur Suivant.

  10. 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://menlo-security-logs/menlo-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 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 (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

  11. Cliquez sur Suivant.

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

  1. Accédez à Cloud Storage > Buckets.
  2. Cliquez sur menlo-security-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.

Table de mappage UDM

Champ de journal Mappage UDM Logique
customer_name intermediary.hostname Définir si customer_name n'est pas une adresse IP
customer_name intermediary.ip Extrait de customer_name si l'adresse IP est spécifiée, ou des adresses IP xff_ip
xff_ip intermediary.ip
event_data.event_time metadata.event_timestamp Analysé à l'aide du filtre grok et de date
request_type metadata.event_type Logique conditionnelle basée sur ces champs
has_principal metadata.event_type
has_top_url metadata.event_type
has_dest metadata.event_type
log_type metadata.log_type Valeur copiée directement
event_data.name metadata.product_event_type Valeur copiée directement
reqId metadata.product_log_id Valeur copiée directement
event_data.version metadata.product_version Valeur copiée directement
event_data.protocol network.application_protocol Mappé à l'aide d'une liste de protocoles prédéfinie
request_type network.http.method Valeur copiée directement
URL de provenance network.http.referral_url Valeur copiée directement
response_code network.http.response_code Converti en entier
event_data.user-agent network.http.user_agent Valeur copiée directement
full_session_id network.session_id Valeur copiée directement
browser_and_version principal.application Valeur copiée directement
event_data.x-client-ip principal.asset.ip Valeur de event_data.x-client-ip si elle n'est pas vide, sinon src_ip
src_ip principal.asset.ip
groupes principal.group.product_object_id Valeur copiée directement
event_data.x-client-ip principal.ip Valeur de event_data.x-client-ip si elle n'est pas vide, sinon src_ip
src_ip principal.ip
région principal.location.country_or_region Valeur de la région si elle n'est pas vide, sinon egress_country
egress_country principal.location.country_or_region
region_name principal.location.name Valeur copiée directement
egress_ip principal.nat_ip Valeur copiée directement
event_data.userid principal.user.email_addresses Fusionné si l'adresse e-mail correspond à l'expression régulière
event_data.userid principal.user.userid Valeur copiée directement
event_data.domain security_result.about.administrative_domain Valeur copiée directement
event_data.url security_result.about.url Valeur copiée directement
event_data.pe_action security_result.action Mappé sur ALLOW, BLOCK ou ALLOW_WITH_MODIFICATION
event_data.categories security_result.category_details Fusionné à partir de event_data.categories (extrait ou direct) et casb_cat_name
casb_cat_name security_result.category_details
event_data.risk_score security_result.confidence_details Valeur copiée directement
content-type security_result.detection_fields Fusionné à partir de différents champs de libellé
produit security_result.detection_fields
menaces security_result.detection_fields
virusDetails security_result.detection_fields
sandboxResult security_result.detection_fields
fullScanResult security_result.detection_fields
tab_id security_result.detection_fields
pe_reason security_result.detection_fields
ua_type security_result.detection_fields
email_isolation_state security_result.detection_fields
document_url security_result.detection_fields
archive_path security_result.detection_fields
casb_app_name security_result.detection_fields
casb_fun_name security_result.detection_fields
casb_org_name security_result.detection_fields
casb_profile_id security_result.detection_fields
casb_profile_name security_result.detection_fields
casb_profile_type security_result.detection_fields
casb_risk_score security_result.detection_fields
connId security_result.detection_fields
origin_country security_result.detection_fields
egress_country security_result.detection_fields
is_casb_ddl security_result.detection_fields
is_iframe security_result.detection_fields
parent_file_id security_result.detection_fields
parent_filename security_result.detection_fields
parent_sha256 security_result.detection_fields
parent_tid security_result.detection_fields
root_file_id security_result.detection_fields
root_filename security_result.detection_fields
root_sha256 security_result.detection_fields
casb_risk_score security_result.risk_score Converti en float
casb_profile_id security_result.rule_id Valeur copiée directement
pe_rulename security_result.rule_name Valeur copiée directement
casb_profile_name security_result.rule_set Valeur copiée directement
casb_profile_type security_result.rule_type Valeur copiée directement
event_data.severity security_result.severity_details Valeur copiée directement
magicName security_result.summary Concaténé sous la forme "%{magicName} est %{threats}."
menaces security_result.summary
threat_types security_result.threat_name Valeur copiée directement
origin_ip src.ip Valeur copiée directement
origin_country src.location.country_or_region Valeur copiée directement
event_data.dst target.asset.ip Extrait de event_data.dst ou dst à l'aide du modèle IP grok
dst target.asset.ip
filename target.file.full_path Valeur copiée directement
mimeType target.file.mime_type Valeur copiée directement
sha256 target.file.sha256 Valeur copiée directement
file_size target.file.size Converti en float, puis en uinteger
event_data.dst target.ip Extrait de event_data.dst ou dst à l'aide du modèle IP grok
dst target.ip
top_url target.url Extraite à l'aide du format grok d'URI
metadata.product_name metadata.product_name Défini sur "MENLO SECURITY"
metadata.vendor_name metadata.vendor_name Défini sur "MENLO SECURITY"
region event.idm.read_only_udm.principal.location.country_or_region Mappé à partir du journal des modifications
region_name event.idm.read_only_udm.principal.location.name Mappé à partir du journal des modifications
src_ip event.idm.read_only_udm.principal.ip et event.idm.read_only_udm.principal.asset.ip Mappé à partir du journal des modifications
groups event.idm.read_only_udm.principal.group.product_object_id Mappé à partir du journal des modifications
content-type event.idm.read_only_udm.security_result.detection_fields Mappé à partir du journal des modifications
product event.idm.read_only_udm.security_result.detection_fields Mappé à partir du journal des modifications
mimeType event.idm.read_only_udm.target.file.mime_type Mappé à partir du journal des modifications

Journal des modifications

Afficher le journal des modifications pour ce parseur

Vous avez encore besoin d'aide ? Obtenez des réponses de membres de la communauté et de professionnels Google SecOps.