Collecter les journaux de la plate-forme MuleSoft Anypoint

Compatible avec :

Google SecOps SIEM

Ce document explique comment ingérer des événements de piste d'audit provenant des journaux de la plate-forme MuleSoft Anypoint dans Google Security Operations à l'aide de Cloud Storage.

Remarque : MuleSoft est un type de journal sans analyseur prédéfini. Ce guide ne couvre donc que l'ingestion des journaux. Vous devez configurer un analyseur personnalisé pour convertir les journaux MuleSoft en UDM.

Avant de commencer

Assurez-vous de remplir les conditions suivantes :

• Une instance Google SecOps
• Un projet Google Cloud avec l'API Cloud Storage activée
• Autorisations pour créer et gérer des buckets Cloud Storage
• Autorisations permettant de gérer les stratégies Identity and Access Management (IAM) sur les buckets Cloud Storage
• Autorisations pour créer des fonctions Cloud Run, des sujets Pub/Sub et des jobs Cloud Scheduler
• Autorisations permettant de créer des comptes de service
• Accès privilégié à la plate-forme MuleSoft Anypoint

Obtenir l'ID d'organisation MuleSoft

1. Connectez-vous à la plate-forme Anypoint.
2. Accédez à Access Management> Organizations (Gestion des accès > Organisations).
3. Dans le tableau Groupes d'établissements, cliquez sur le nom de votre organisation.
4. Copiez l'ID d'organisation (par exemple, 0a12b3c4-d5e6-789f-1021-1a2b34cd5e6f).

Vous pouvez également accéder à MuleSoft Business Groups et copier l'ID à partir de l'URL.

Créer l'application connectée MuleSoft

1. Connectez-vous à la plate-forme Anypoint.
2. Accédez à Gestion des accès> Applications connectées> Créer une application.
3. Fournissez les informations de configuration suivantes :
   • Nom de l'application : saisissez un nom unique (par exemple, Google SecOps export).
   • Sélectionnez L'application agit en son propre nom (identifiants client).
4. Cliquez sur Ajouter des autorisations> Lecteur du journal d'audit > Suivant.
5. Sélectionnez tous les groupes d'établissements dont vous avez besoin des journaux.
6. Cliquez sur Suivant > Ajouter des champs d'application.

7. Cliquez sur Save (Enregistrer), puis copiez l'ID client et le code secret du client.

   Remarque : Des niveaux d'accès supplémentaires peuvent être requis en fonction des autorisations de l'organisation. Si vous rencontrez des erreurs d'autorisation, vérifiez que l'application connectée a été autorisée à accéder à tous les groupes d'établissements requis.

Créer un bucket Cloud Storage

1. Accédez à la consoleGoogle 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, mulesoft-audit-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 fréquemment consultés)
   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

La fonction Cloud Run nécessite un compte de service disposant des autorisations nécessaires pour écrire dans le bucket Cloud Storage.

Créer le 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 mulesoft-logs-collector-sa.
   • Description du compte de service : saisissez Service account for Cloud Run function to collect MuleSoft Anypoint logs.
4. Cliquez sur Créer et continuer.
5. Dans la section Autoriser ce compte de service à accéder au projet :
   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.

Ces rôles sont requis pour :

• Administrateur des objets Storage : écrire des journaux dans le bucket Cloud Storage 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 le bucket Cloud Storage

Accordez au compte de service les autorisations write sur le bucket Cloud Storage comme suit :

1. Accédez à Cloud Storage > Buckets.
2. Cliquez sur le nom de votre bucket.
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 (par exemple, mulesoft-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

Créez un sujet Pub/Sub auquel Cloud Scheduler publiera des messages et auquel la fonction Cloud Run s'abonnera, comme suit :

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 sujet : saisissez mulesoft-audit-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

La fonction Cloud Run sera déclenchée par les messages Pub/Sub de Cloud Scheduler pour extraire les journaux de l'API MuleSoft Anypoint et les écrire dans Cloud Storage.

1. Dans la console GCP, accédez à Cloud Run.
2. Cliquez sur Écrire une fonction.

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

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

4. Dans la section Déclencheur, cliquez sur Ajouter un déclencheur, puis sélectionnez Déclencheur Pub/Sub.

5. Dans le volet Déclencheur Eventarc, fournissez les informations de configuration suivantes :

   • Nom du déclencheur : conservez le nom généré ou saisissez-en un.
   • Type de déclencheur : sélectionnez Sources Google.
   • Fournisseur d'événements : sélectionnez Pub/Sub.
   • Type d'événement : sélectionnez google.cloud.pubsub.topic.v1.messagePublished.
   • Sélectionner un sujet Cloud Pub/Sub : choisissez le sujet mulesoft-audit-trigger.
   • Région : sélectionnez la même région que la fonction.
   • Compte de service : sélectionnez le compte de service mulesoft-logs-collector-sa.

6. Cliquez sur Enregistrer le déclencheur.

   Remarque : Les options de déclencheur proviennent d'Eventarc. Si l'option Déclencheur Pub/Sub n'est pas listée, activez l'API Eventarc pour le projet et actualisez la page. Si le volet indique que le compte de service ne dispose pas du rôle Destinataire des événements Eventarc ou Demandeur Cloud Run, cliquez sur Accorder.

7. Dans la section Authentification :

   1. Sélectionnez Exiger l'authentification.
   2. Vérifiez Identity and Access Management (IAM).
   Remarque : Pub/Sub gère automatiquement l'authentification lors de l'appel de la fonction.

8. Accédez à Conteneurs, mise en réseau, sécurité et développez cette section.

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

   • Compte de service : sélectionnez le compte de service mulesoft-logs-collector-sa.

10. 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	mulesoft-audit-logs	Nom du bucket Cloud Storage
    GCS_PREFIX	mulesoft-audit	Préfixe des fichiers journaux
    STATE_KEY	mulesoft-audit-state.json	Chemin d'état, en dehors du préfixe du journal
    MULE_ORG_ID	your_org_id	ID d'organisation MuleSoft
    CLIENT_ID	your_client_id	ID client de l'application connectée
    CLIENT_SECRET	your_client_secret	Code secret du client de l'application connectée
    MAX_RECORDS	10000	Nombre maximal d'enregistrements par exécution
    PAGE_SIZE	200	Nombre d'entrées par page (200 maximum pour l'API)
    LOOKBACK_HOURS	24	Période d'analyse initiale lors de la première exécution

    Remarque : STATE_KEY pointe volontairement en dehors du préfixe du journal. Le flux ingère chaque objet sous l'URI de son bucket et, lorsqu'une option de suppression est sélectionnée, supprime ce qu'il a transféré. En conservant le fichier d'état en dehors de ce préfixe, vous empêchez son ingestion en tant que données de journaux et sa suppression, ce qui réinitialiserait la collecte et ingérerait à nouveau les événements en double.

11. Accédez à l'onglet Variables et secrets, puis à Requêtes :

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

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

13. Accédez à Environnement d'exécution :

    • Sélectionnez Par défaut (recommandé).

14. 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).

15. Cliquez sur Créer.

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

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

Ajouter le code de la fonction

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

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

   • Premier fichier : main.py:

     import functions_framework
     from google.cloud import storage
     from google.cloud.exceptions import NotFound
     import hashlib
     import json
     import os
     import urllib3
     from urllib.parse import urlencode
     from datetime import datetime, timezone, timedelta
     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()
     
     # MuleSoft API endpoints
     TOKEN_URL = "https://anypoint.mulesoft.com/accounts/api/v2/oauth2/token"
     AUDIT_URL = "https://anypoint.mulesoft.com/audit/v2/organizations/{org_id}/query?cursorPagination=true"
     
     # Environment variables
     GCS_BUCKET = os.environ.get('GCS_BUCKET')
     GCS_PREFIX = os.environ.get('GCS_PREFIX', 'mulesoft-audit')
     # STATE_KEY must stay OUTSIDE GCS_PREFIX. The feed ingests every object under
     # its bucket URI and, with a deletion option selected, deletes what it
     # transferred. A state file inside the prefix would be ingested as log data and
     # then deleted, resetting collection and re-ingesting duplicates.
     STATE_KEY = os.environ.get('STATE_KEY', 'mulesoft-audit-state.json')
     MULE_ORG_ID = os.environ.get('MULE_ORG_ID')
     CLIENT_ID = os.environ.get('CLIENT_ID')
     CLIENT_SECRET = os.environ.get('CLIENT_SECRET')
     MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '10000'))
     # The Audit Log Query API returns at most 200 entries per JSON query.
     PAGE_SIZE = min(int(os.environ.get('PAGE_SIZE', '200')), 200)
     LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))
     # The query re-reads this many minutes before the watermark so that events the
     # platform records late are still collected. Re-read events are removed by hash.
     OVERLAP_MINUTES = int(os.environ.get('OVERLAP_MINUTES', '2'))
     
     class FetchError(Exception):
         """Raised when the MuleSoft API call fails.
     
         The watermark must never advance on a failed fetch, otherwise the events in
         the failed window are skipped permanently.
         """
     
     def to_unix_millis(dt: datetime) -> int:
         """Convert datetime to Unix epoch milliseconds."""
         if dt.tzinfo is None:
             dt = dt.replace(tzinfo=timezone.utc)
         return int(dt.astimezone(timezone.utc).timestamp() * 1000)
     
     def parse_datetime(value: str) -> datetime:
         """Parse an ISO 8601 datetime string into an aware datetime."""
         if value.endswith("Z"):
             value = value[:-1] + "+00:00"
         dt = datetime.fromisoformat(value)
         if dt.tzinfo is None:
             dt = dt.replace(tzinfo=timezone.utc)
         return dt
     
     def event_keys(event: dict) -> set:
         """Return the identity this event can be recognised by.
     
         Audit log entries carry no unique event ID, so a record counts as already
         collected when its full JSON content was seen before. The overlap window
         returns the same JSON again, which this catches without any knowledge of
         the payload schema.
         """
         return {'sha256:' + hashlib.sha256(
             json.dumps(event, sort_keys=True, ensure_ascii=False).encode('utf-8')
         ).hexdigest()}
     
     def event_time(event: dict) -> str:
         """Return the event timestamp as an ISO 8601 string, or an empty string.
     
         Audit log entries carry the event time in `timestamp` as Unix milliseconds.
         """
         raw = event.get('timestamp')
         if raw is None or raw == '':
             return ''
         try:
             return datetime.fromtimestamp(int(raw) / 1000, tz=timezone.utc).isoformat()
         except (TypeError, ValueError):
             return ''
     
     @functions_framework.cloud_event
     def main(cloud_event):
         """Fetch MuleSoft Anypoint audit logs and write them to Cloud Storage.
     
         Args:
             cloud_event: CloudEvent object containing the Pub/Sub message.
         """
         if not all([GCS_BUCKET, MULE_ORG_ID, CLIENT_ID, CLIENT_SECRET]):
             # Raise rather than return: a bare return acks the Pub/Sub message and
             # reports the run as successful, silently discarding the schedule tick.
             raise RuntimeError('Missing required environment variables')
     
         bucket = storage_client.bucket(GCS_BUCKET)
         state = load_state(bucket, STATE_KEY)
     
         now = datetime.now(timezone.utc)
         watermark = None
         if state.get('last_event_time'):
             watermark = parse_datetime(state['last_event_time'])
         seen_keys = set(state.get('seen_keys', []))
     
         if watermark is None:
             start_time = now - timedelta(hours=LOOKBACK_HOURS)
         else:
             start_time = watermark - timedelta(minutes=OVERLAP_MINUTES)
     
         print(f"Fetching audit logs from {start_time.isoformat()} to {now.isoformat()}")
     
         token = get_token(CLIENT_ID, CLIENT_SECRET)
     
         # A FetchError here propagates: the run fails, the watermark is untouched,
         # and the next invocation retries the same window.
         records, newest_event_time = fetch_audit(
             org_id=MULE_ORG_ID,
             token=token,
             start_time=start_time,
             end_time=now,
             page_size=PAGE_SIZE,
             max_records=MAX_RECORDS,
         )
     
         # Remove events already written by an earlier run. Without this, the overlap
         # window re-emits its events on every invocation.
         fresh = [r for r in records if not (event_keys(r) & seen_keys)]
         print(f"Fetched {len(records)} records, {len(fresh)} new after deduplication")
     
         if not fresh:
             print("No new audit log records found. Watermark left unchanged.")
             return
     
         if not newest_event_time:
             # A watermark that cannot be derived from the data is not a watermark.
             # Advancing on wall-clock here would skip every late-recorded event.
             raise FetchError('Records were returned but no event timestamp could be parsed')
     
         timestamp = now.strftime('%Y%m%dT%H%M%SZ')
         object_key = f"{GCS_PREFIX}/audit_{timestamp}.ndjson"
         blob = bucket.blob(object_key)
     
         ndjson = '\n'.join(json.dumps(record, ensure_ascii=False) for record in fresh) + '\n'
         blob.upload_from_string(ndjson, content_type='application/x-ndjson')
     
         print(f"Wrote {len(fresh)} records to gs://{GCS_BUCKET}/{object_key}")
     
         # Advance the watermark only after the data is durably written.
         new_watermark = parse_datetime(newest_event_time)
         if watermark and new_watermark < watermark:
             new_watermark = watermark
     
         # Retain only the hashes still inside the overlap window, so the state file
         # stays small while covering every event the next query can re-read.
         cutoff = new_watermark - timedelta(minutes=OVERLAP_MINUTES)
         retained = set()
         for record in records:
             stamp = event_time(record)
             if not stamp:
                 continue
             if parse_datetime(stamp) >= cutoff:
                 retained.update(event_keys(record))
     
         save_state(bucket, STATE_KEY, {
             'last_event_time': new_watermark.isoformat(),
             'seen_keys': sorted(retained),
         })
     
         print(f"Successfully processed {len(fresh)} records")
     
     def load_state(bucket, key):
         """Read the collector state from Cloud Storage.
     
         Only a missing object is treated as a cold start. Any other error is raised:
         swallowing it would silently reset collection to the full lookback window
         and re-ingest that entire period.
         """
         blob = bucket.blob(key)
         try:
             return json.loads(blob.download_as_text())
         except NotFound:
             print('No state file found. Starting from the lookback window.')
             return {}
     
     def save_state(bucket, key, state: dict):
         """Write the collector state to Cloud Storage.
     
         Failures are raised, not logged. If the state write fails after the data was
         uploaded, the next run repeats the same window and duplicates it.
         """
         blob = bucket.blob(key)
         blob.upload_from_string(
             json.dumps(state, indent=2),
             content_type='application/json',
         )
         print(f"Saved state: last_event_time={state.get('last_event_time')}")
     
     def retry_after_seconds(response, backoff: float) -> int:
         """Parse Retry-After defensively; it may be an HTTP date, which int() cannot parse."""
         raw = response.headers.get('Retry-After')
         try:
             return int(raw) if raw else int(backoff)
         except (TypeError, ValueError):
             return int(backoff)
     
     def get_token(client_id: str, client_secret: str) -> str:
         """Get an OAuth 2.0 access token from MuleSoft using client credentials."""
         body = urlencode({
             'grant_type': 'client_credentials',
             'client_id': client_id,
             'client_secret': client_secret,
         }).encode('utf-8')
     
         backoff = 1.0
         for attempt in range(5):
             try:
                 response = http.request(
                     'POST',
                     TOKEN_URL,
                     body=body,
                     headers={'Content-Type': 'application/x-www-form-urlencoded'},
                 )
             except Exception as e:
                 raise FetchError(f'Token request failed: {e}') from e
     
             if response.status == 429:
                 wait = retry_after_seconds(response, backoff)
                 print(f"Rate limited (429) on token request. Retrying after {wait}s...")
                 time.sleep(wait)
                 backoff = min(backoff * 2, 30.0)
                 continue
     
             if response.status != 200:
                 raise FetchError(f'Failed to get token: HTTP {response.status} {response.data.decode("utf-8", "replace")}')
     
             try:
                 return json.loads(response.data.decode('utf-8'))['access_token']
             except (json.JSONDecodeError, KeyError) as e:
                 raise FetchError(f'Malformed token response: {e}') from e
     
         raise FetchError('Rate limited repeatedly on token request; giving up without advancing the watermark')
     
     def fetch_audit(org_id: str, token: str, start_time: datetime, end_time: datetime, page_size: int, max_records: int):
         """Fetch audit log entries from the MuleSoft Audit Log Query API.
     
         Uses cursor pagination. The API returns entries in ascending timestamp order,
         which a time watermark requires: with newest-first ordering a truncated run
         would advance the watermark past unread events.
     
         Returns:
             Tuple of (records list, newest_event_time ISO string).
     
         Raises:
             FetchError: on any API or transport failure, so that the caller does not
                 mistake a failed fetch for an empty result and advance the watermark.
         """
         endpoint = AUDIT_URL.format(org_id=org_id)
         headers = {
             'Authorization': f'Bearer {token}',
             'Content-Type': 'application/json',
             'Accept': 'application/json',
         }
     
         records = []
         newest_time = None
         cursor = None
         page_num = 0
         backoff = 1.0
         rate_limit_retries = 0
         MAX_RATE_LIMIT_RETRIES = 5
     
         while True:
             page_num += 1
     
             if len(records) >= max_records:
                 # Stop cleanly. The remaining events stay ahead of the watermark and
                 # are collected by the next run.
                 print(f"Reached max_records limit ({max_records})")
                 break
     
             # Rebuild the full query on every page; the time window must never be dropped.
             body = {
                 'startDate': to_unix_millis(start_time),
                 'endDate': to_unix_millis(end_time),
                 'limit': page_size,
                 'ascending': True,
             }
             if cursor:
                 body['cursor'] = cursor
     
             try:
                 response = http.request(
                     'POST',
                     endpoint,
                     body=json.dumps(body).encode('utf-8'),
                     headers=headers,
                 )
             except Exception as e:
                 raise FetchError(f'Request to {endpoint} failed: {e}') from e
     
             if response.status == 429:
                 rate_limit_retries += 1
                 if rate_limit_retries > MAX_RATE_LIMIT_RETRIES:
                     raise FetchError('Rate limited repeatedly; giving up without advancing the watermark')
                 wait = retry_after_seconds(response, backoff)
                 print(f"Rate limited (429). Retrying after {wait}s...")
                 time.sleep(wait)
                 backoff = min(backoff * 2, 30.0)
                 continue
     
             backoff = 1.0
             rate_limit_retries = 0
     
             if response.status != 200:
                 raise FetchError(f'HTTP {response.status} from {endpoint}: {response.data.decode("utf-8", "replace")}')
     
             try:
                 data = json.loads(response.data.decode('utf-8'))
             except json.JSONDecodeError as e:
                 raise FetchError(f'Malformed JSON response from {endpoint}: {e}') from e
     
             page_results = data.get('data') or []
             if not page_results:
                 print("No more results (empty page)")
                 break
     
             print(f"Page {page_num}: Retrieved {len(page_results)} entries")
             records.extend(page_results)
     
             for event in page_results:
                 stamp = event_time(event)
                 if not stamp:
                     continue
                 if newest_time is None or parse_datetime(stamp) > parse_datetime(newest_time):
                     newest_time = stamp
     
             cursor = data.get('cursor')
             if not cursor:
                 break
     
         print(f"Retrieved {len(records)} total entries from {page_num} pages")
         return records, newest_time
     

     Remarque : Ce script est fourni à titre d'exemple et n'est pas officiellement compatible. N'hésitez pas à nous faire part de vos commentaires pour nous aider à l'améliorer. Pour envoyer des commentaires, cliquez sur Envoyer des commentaires > Commentaires sur le produit de sécurité Google SecOps sur cette page.

   • Deuxième fichier : 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).

Remarques importantes

• Limitation du débit : le point de terminaison de requête du journal d'audit applique des limites de débit par adresse IP dans les trois plans de contrôle. Le plan de contrôle américain autorise 700 requêtes par minute et par adresse IP, tandis que les plans de contrôle européens et Gov autorisent 40 requêtes par minute et par adresse IP. La fonction implémente un intervalle exponentiel entre les tentatives pour gérer automatiquement la limitation du débit.

• Expiration du jeton : les jetons d'accès expirent généralement 30 à 60 minutes après leur émission. La fonction demande un nouveau jeton pour chaque exécution, ce qui est largement dans les limites de cette fenêtre.

• Déduplication et état : la fonction conserve un filigrane de l'heure de l'événement le plus récent et les hachages des événements dans la fenêtre de chevauchement dans le fichier d'état. Chaque exécution relit les deux minutes précédant le filigrane pour détecter les entrées enregistrées tardivement et supprime celles qu'elle a déjà écrites. Ainsi, une nouvelle tentative, une exécution manuelle ou un retard du planificateur ne produisent pas d'événements en double. Si un appel d'API échoue, l'exécution échoue et le filigrane reste inchangé. L'exécution suivante réessaie donc la même fenêtre.

• Conservation des journaux d'audit : la durée de conservation par défaut des journaux d'audit est d'un an. Si votre organisation a été créée avant le 10 juillet 2023 et que vous n'avez pas modifié manuellement la période de conservation, elle est de six ans. Téléchargez régulièrement les journaux si vous devez les conserver au-delà de la période de conservation configurée.

Créer une tâche Cloud Scheduler

Cloud Scheduler publiera des messages dans le sujet Pub/Sub à intervalles réguliers, ce qui déclenchera la fonction Cloud Run.

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	mulesoft-audit-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électionnez le thème mulesoft-audit-trigger.
   Corps du message	{} (objet JSON vide)

4. Cliquez sur Créer.

Tester le job du planificateur

1. Dans la console Cloud Scheduler, recherchez votre job.
2. Cliquez sur Exécuter de force pour déclencher manuellement.
3. Patientez quelques secondes, puis accédez à Cloud Run > Services > mulesoft-audit-collector > Journaux.
4. Vérifiez que la fonction s'est exécutée correctement.
5. Vérifiez le bucket Cloud Storage pour confirmer que les journaux ont été écrits.

Configurer un flux dans Google SecOps pour ingérer les journaux MuleSoft

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, MuleSoft Logs).
5. Sélectionnez Google Cloud Storage V2 comme Type de source.
6. Sélectionnez Mulesoft comme type de journal.

7. Cliquez sur Get Service Account (Obtenir un compte de service). Une adresse e-mail de compte de service unique s'affiche, par exemple :

   [e-mail]
   

8. Copiez cette adresse e-mail. Vous en aurez besoin lors de la tâche suivante.

   Remarque : Chaque instance Google SecOps possède un compte de service unique. N'utilisez pas de comptes de service provenant d'autres documentations ou exemples.

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 Cloud Storage :

      gs://mulesoft-audit-logs/
      

      • Remplacez mulesoft-audit-logs par le nom réel du bucket.
      Remarque : Incluez toujours la barre oblique finale (/) à la fin de l'URI.

    • Option de suppression de la source : sélectionnez l'option de suppression de votre choix :

      • Ne jamais supprimer les fichiers : ne supprimez jamais les fichiers de la source (recommandé pour les tests).
      • Supprimer les fichiers transférés et les répertoires vides : supprime les fichiers et les répertoires vides de la source une fois l'extraction réussie.

    • Âge maximal du fichier : incluez les fichiers modifiés au cours d'un certain nombre de 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.

11. Cliquez sur Suivant.

12. Vérifiez la configuration de votre nouveau flux d'actualités 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 de deux rôles sur votre bucket Cloud Storage : Lecteur des objets Storage pour lire les objets journaux et un rôle au niveau du bucket pour lire les métadonnées du bucket.

1. Accédez à Cloud Storage > Buckets.
2. Cliquez sur le nom de votre bucket.
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 les deux options suivantes :
   • Lecteur des objets Storage : lit les objets journaux.
   • Lecteur des anciens buckets Storage : lit les métadonnées du bucket. Si vous avez sélectionné l'option de suppression Supprimer les fichiers transférés et les répertoires vides, sélectionnez plutôt Rédacteur de bucket Storage hérité, qui accorde également l'autorisation de suppression.

6. Cliquez sur Enregistrer.

   Remarque : Accordez les deux rôles. Le rôle Lecteur des objets Storage n'accorde pas l'autorisation au niveau du bucket dont le flux a besoin, et les rôles Ancien lecteur de bucket Storage n'accordent pas l'autorisation de lire le contenu des objets.

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