MuleSoft Anypoint-Plattformprotokolle erfassen

Unterstützt in:

In diesem Dokument wird beschrieben, wie Sie Audit-Trail-Ereignisse aus MuleSoft Anypoint-Plattform-Logs mithilfe von Cloud Storage in Google Security Operations aufnehmen.

Hinweis

Prüfen Sie, ob folgende Voraussetzungen erfüllt sind:

  • Eine Google SecOps-Instanz
  • Ein Google Cloud -Projekt mit aktivierter Cloud Storage API
  • Berechtigungen zum Erstellen und Verwalten von Cloud Storage-Buckets
  • Berechtigungen zum Verwalten von IAM-Richtlinien (Identity and Access Management) für Cloud Storage-Buckets
  • Berechtigungen zum Erstellen von Cloud Run-Funktionen, Pub/Sub-Themen und Cloud Scheduler-Jobs
  • Berechtigungen zum Erstellen von Dienstkonten
  • Privilegierter Zugriff auf die MuleSoft Anypoint Platform

MuleSoft-Organisations-ID abrufen

  1. Melden Sie sich in der Anypoint Platform an.
  2. Gehen Sie zu Zugriffsverwaltung > Organisationen.
  3. Klicken Sie in der Tabelle Gruppen für Unternehmensstandorte auf den Namen Ihrer Organisation.
  4. Kopieren Sie die Organisations-ID (z. B. 0a12b3c4-d5e6-789f-1021-1a2b34cd5e6f).

Alternativ können Sie auch zu MuleSoft Business Groups gehen und die ID aus der URL kopieren.

MuleSoft Connected App erstellen

  1. Melden Sie sich in der Anypoint Platform an.
  2. Rufen Sie Zugriffsverwaltung > „Verbundene Apps“ > „App erstellen“ auf.
  3. Geben Sie die folgenden Konfigurationsdetails an:
    • App-Name: Geben Sie einen eindeutigen Namen ein (z. B. Google SecOps export).
    • Wählen Sie App acts on its own behalf (client credentials) (App agiert im eigenen Namen (Clientanmeldedaten)) aus.
  4. Klicken Sie auf Bereiche hinzufügen> Audit-Log-Anzeige > Weiter.
  5. Wählen Sie alle Unternehmensgruppen aus, deren Protokolle Sie benötigen.
  6. Klicken Sie auf Weiter> Bereiche hinzufügen.

  7. Klicken Sie auf Speichern und kopieren Sie die Client-ID und den Clientschlüssel.

Cloud Storage-Bucket erstellen

  1. Rufen Sie die Google Cloud Console auf.
  2. Wählen Sie Ihr Projekt aus oder erstellen Sie ein neues.
  3. Rufen Sie im Navigationsmenü Cloud Storage > Buckets auf.
  4. Klicken Sie auf Bucket erstellen.

  5. Geben Sie die folgenden Konfigurationsdetails an:

    Einstellung Wert
    Bucket benennen Geben Sie einen global eindeutigen Namen ein, z. B. mulesoft-audit-logs.
    Standorttyp Wählen Sie die Option aus, die am besten zu Ihren Anforderungen passt (Region, Dual-Region, Multi-Region).
    Standort Wählen Sie den Speicherort aus, z. B. us-central1.
    Speicherklasse Standard (empfohlen für Logs, auf die häufig zugegriffen wird)
    Zugriffskontrolle Einheitlich (empfohlen)
    Schutztools Optional: Objektversionierung oder Aufbewahrungsrichtlinie aktivieren

  6. Klicken Sie auf Erstellen.

Dienstkonto für die Cloud Run-Funktion erstellen

Für die Cloud Run-Funktion ist ein Dienstkonto mit Berechtigungen zum Schreiben in den Cloud Storage-Bucket erforderlich.

Erstellen Sie das Dienstkonto:

  1. Wechseln Sie in der GCP Console zu IAM und Verwaltung > Dienstkonten.
  2. Klicken Sie auf Dienstkonto erstellen.
  3. Geben Sie die folgenden Konfigurationsdetails an:
    • Name des Dienstkontos: Geben Sie mulesoft-logs-collector-sa ein.
    • Beschreibung des Dienstkontos: Geben Sie Service account for Cloud Run function to collect MuleSoft Anypoint logs ein.
  4. Klicken Sie auf Erstellen und fortfahren.
  5. Gehen Sie im Bereich Diesem Dienstkonto Zugriff auf das Projekt erteilen so vor:
    1. Klicken Sie auf Rolle auswählen.
    2. Suchen Sie nach Storage Object Admin und wählen Sie die Rolle aus.
    3. Klicken Sie auf + Weitere Rolle hinzufügen.
    4. Suchen Sie nach Cloud Run Invoker und wählen Sie die Rolle aus.
    5. Klicken Sie auf + Weitere Rolle hinzufügen.
    6. Suchen Sie nach Cloud Functions Invoker und wählen Sie die Rolle aus.
  6. Klicken Sie auf Weiter.
  7. Klicken Sie auf Fertig.

Diese Rollen sind erforderlich für:

  • Storage-Objekt-Administrator: Protokolle in den Cloud Storage-Bucket schreiben und Statusdateien verwalten
  • Cloud Run Invoker: Pub/Sub darf die Funktion aufrufen.
  • Cloud Functions-Invoker: Funktionsaufruf zulassen

IAM-Berechtigungen für den Cloud Storage-Bucket gewähren

Gewähren Sie dem Dienstkonto die Berechtigungen write für den Cloud Storage-Bucket:

  1. Rufen Sie Cloud Storage > Buckets auf.
  2. Klicken Sie auf den Namen Ihres Buckets.
  3. Wechseln Sie zum Tab Berechtigungen.
  4. Klicken Sie auf Zugriff erlauben.
  5. Geben Sie die folgenden Konfigurationsdetails an:
    • Hauptkonten hinzufügen: Geben Sie die E-Mail-Adresse des Dienstkontos ein (z. B. mulesoft-logs-collector-sa@PROJECT_ID.iam.gserviceaccount.com).
    • Rollen zuweisen: Wählen Sie Storage Object Admin aus.
  6. Klicken Sie auf Speichern.

Pub/Sub-Thema erstellen

Erstellen Sie ein Pub/Sub-Thema, in dem Cloud Scheduler veröffentlicht und das von der Cloud Run-Funktion abonniert wird:

  1. Rufen Sie in der GCP Console Pub/Sub > Themen auf.
  2. Klicken Sie auf Thema erstellen.
  3. Geben Sie die folgenden Konfigurationsdetails an:
    • Themen-ID: Geben Sie mulesoft-audit-trigger ein.
    • Übernehmen Sie die anderen Einstellungen.
  4. Klicken Sie auf Erstellen.

Cloud Run-Funktion zum Erfassen von Logs erstellen

Die Cloud Run-Funktion wird durch Pub/Sub-Nachrichten von Cloud Scheduler ausgelöst, um Logs von der MuleSoft Anypoint API abzurufen und in Cloud Storage zu schreiben.

  1. Rufen Sie in der GCP Console Cloud Run auf.
  2. Klicken Sie auf Funktion schreiben.

  3. Geben Sie im Abschnitt Konfigurieren die folgenden Konfigurationsdetails an:

    Einstellung Wert
    Dienstname mulesoft-audit-collector
    Region Wählen Sie die Region aus, die Ihrem Cloud Storage-Bucket entspricht, z. B. us-central1.
    Laufzeit Wählen Sie Python 3.12 oder höher aus.

  4. Klicken Sie im Bereich Trigger auf Trigger hinzufügen und wählen Sie Pub/Sub-Trigger aus.

  5. Geben Sie im Bereich Eventarc-Trigger die folgenden Konfigurationsdetails an:

    • Triggername: Behalten Sie den generierten Namen bei oder geben Sie einen Namen für den Trigger ein.
    • Triggertyp: Wählen Sie Google-Quellen aus.
    • Ereignisanbieter: Wählen Sie Pub/Sub aus.
    • Ereignistyp: Wählen Sie google.cloud.pubsub.topic.v1.messagePublished aus.
    • Cloud Pub/Sub-Thema auswählen: Wählen Sie das Thema mulesoft-audit-trigger aus.
    • Region: Wählen Sie dieselbe Region wie für die Funktion aus.
    • Dienstkonto: Wählen Sie das Dienstkonto mulesoft-logs-collector-sa aus.

  6. Klicken Sie auf Trigger speichern.

  7. Im Abschnitt Authentifizierung:

    1. Wählen Sie Authentifizierung erforderlich aus.
    2. Weitere Informationen finden Sie unter Identity and Access Management (IAM).

  8. Maximieren Sie Container, Netzwerk, Sicherheit.

  9. Rufen Sie den Tab Sicherheit auf:

    • Dienstkonto: Wählen Sie das Dienstkonto mulesoft-logs-collector-sa aus.

  10. Rufen Sie den Tab Container auf:

    1. Klicken Sie auf Variablen und Secrets.
    2. Klicken Sie für jede Umgebungsvariable auf + Variable hinzufügen:
    Variablenname Beispielwert Beschreibung
    GCS_BUCKET mulesoft-audit-logs Name des Cloud Storage-Buckets
    GCS_PREFIX mulesoft-audit Präfix für Logdateien
    STATE_KEY mulesoft-audit-state.json Zustandspfad, außerhalb des Logpräfixes
    MULE_ORG_ID your_org_id MuleSoft-Organisations-ID
    CLIENT_ID your_client_id Client-ID der verbundenen App
    CLIENT_SECRET your_client_secret Clientschlüssel der verbundenen App
    MAX_RECORDS 10000 Maximale Anzahl von Datensätzen pro Ausführung
    PAGE_SIZE 200 Einträge pro Seite (API-Maximum: 200)
    LOOKBACK_HOURS 24 Anfänglicher Rückschauzeitraum bei der ersten Ausführung

  11. Rufen Sie den Tab Variablen und Secrets → Anfragen auf:

    • Zeitüberschreitung bei Anfrage: Geben Sie 600 Sekunden (10 Minuten) ein.

  12. Rufen Sie den Tab Einstellungen unter Container auf:

    • Im Abschnitt Ressourcen:
      • Arbeitsspeicher: Wählen Sie 512 MiB oder höher aus.
      • CPU: Wählen Sie 1 aus.
    • Klicken Sie auf Fertig.

  13. Rufen Sie die Ausführungsumgebung auf:

    • Wählen Sie Standard (empfohlen) aus.

  14. Im Abschnitt Versionsskalierung:

    • Mindestanzahl von Instanzen: Geben Sie 0 ein.
    • Maximale Anzahl von Instanzen: Geben Sie 100 ein (oder passen Sie den Wert an die erwartete Last an).

  15. Klicken Sie auf Erstellen.

  16. Warten Sie, bis der Dienst erstellt wurde (1–2 Minuten).

  17. Nachdem der Dienst erstellt wurde, wird der Inline-Code-Editor automatisch geöffnet.

Funktionscode hinzufügen

  1. Geben Sie main in das Feld Funktionseinstiegspunkt ein.

  2. Erstellen Sie im Inline-Code-Editor zwei Dateien:

    • Erste Datei – 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

    • Zweite Datei – requirements.txt:

      functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0

  3. Klicken Sie auf Bereitstellen, um die Funktion zu speichern und bereitzustellen.

  4. Warten Sie, bis die Bereitstellung abgeschlossen ist (2–3 Minuten).

Was Sie bedenken sollten

  • Ratenbegrenzung:Für den Audit-Log-Abfrageendpunkt gelten Ratenbegrenzungen pro IP-Adresse in den drei Steuerungsebenen. Die US-Steuerungsebene erlaubt 700 Anfragen pro Minute und IP-Adresse, während die EU- und Gov-Steuerungsebenen 40 Anfragen pro Minute und IP-Adresse erlauben. Die Funktion implementiert exponentiellen Backoff, um die Ratenbegrenzung automatisch zu verarbeiten.

  • Ablauf von Token:Zugriffstoken laufen in der Regel etwa 30 bis 60 Minuten nach der Ausstellung ab. Die Funktion fordert bei jeder Ausführung ein neues Token an, was innerhalb dieses Zeitfensters liegt.

  • Deduplizierung und Status: Die Funktion speichert einen Zeitstempel der neuesten Ereigniszeit und die Hashes der Ereignisse innerhalb des Überlappungsfensters in der Statusdatei. Bei jedem Lauf werden die Daten zwei Minuten vor dem Wasserzeichen noch einmal gelesen, um spät aufgezeichnete Einträge zu erfassen. Die bereits geschriebenen Einträge werden verworfen. Daher werden bei einem erneuten Versuch, einem manuellen Lauf oder einer Planerverzögerung keine doppelten Ereignisse erzeugt. Bei einem fehlgeschlagenen API-Aufruf schlägt auch der Lauf fehl und das Wasserzeichen bleibt unverändert. Beim nächsten Lauf wird dasselbe Zeitfenster noch einmal versucht.

  • Aufbewahrung von Audit-Logs:Audit-Logs haben eine standardmäßige Aufbewahrungsdauer von einem Jahr. Wenn Ihre Organisation vor dem 10. Juli 2023 erstellt wurde und Sie die Aufbewahrungsdauer nicht manuell geändert haben, beträgt die Aufbewahrungsdauer sechs Jahre. Laden Sie Protokolle regelmäßig herunter, wenn Sie sie über den konfigurierten Aufbewahrungszeitraum hinaus aufbewahren müssen.

Cloud Scheduler-Job erstellen

Cloud Scheduler veröffentlicht in regelmäßigen Abständen Nachrichten im Pub/Sub-Thema und löst so die Cloud Run-Funktion aus.

  1. Rufen Sie in der GCP Console Cloud Scheduler auf.
  2. Klicken Sie auf Job erstellen.

  3. Geben Sie die folgenden Konfigurationsdetails an:

    Einstellung Wert
    Name mulesoft-audit-collector-hourly
    Region Dieselbe Region wie die Cloud Run-Funktion auswählen
    Frequenz 0 * * * * (stündlich zur vollen Stunde)
    Zeitzone Zeitzone auswählen (UTC empfohlen)
    Zieltyp Pub/Sub
    Thema Wählen Sie das Thema mulesoft-audit-trigger aus.
    Inhalt der Nachricht {} (leeres JSON-Objekt)

  4. Klicken Sie auf Erstellen.

Scheduler-Job testen

  1. Suchen Sie in der Cloud Scheduler-Konsole nach Ihrem Job.
  2. Klicken Sie auf Force run (Ausführung erzwingen), um den Trigger manuell auszulösen.
  3. Warten Sie einige Sekunden und rufen Sie Cloud Run > Dienste > mulesoft-audit-collector > Logs auf.
  4. Prüfen Sie, ob die Funktion erfolgreich ausgeführt wurde.
  5. Prüfen Sie im Cloud Storage-Bucket, ob Logs geschrieben wurden.

Feed in Google SecOps konfigurieren, um die MuleSoft-Logs aufzunehmen

  1. Rufen Sie die SIEM-Einstellungen > Feeds auf.
  2. Klicken Sie auf Neuen Nachrichten-Feed hinzufügen.
  3. Klicken Sie auf Einzelnen Feed konfigurieren.
  4. Geben Sie im Feld Feedname einen Namen für den Feed ein, z. B. MuleSoft Logs.
  5. Wählen Sie Google Cloud Storage V2 als Quelltyp aus.
  6. Wählen Sie Mulesoft als Protokolltyp aus.

  7. Klicken Sie auf Dienstkonto abrufen. Es wird eine eindeutige E-Mail-Adresse für das Dienstkonto angezeigt, z. B.:

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com

  8. Kopieren Sie diese E-Mail-Adresse. Sie verwenden es in der nächsten Aufgabe.

  9. Klicken Sie auf Weiter.

  10. Geben Sie Werte für die folgenden Eingabeparameter an:

    • Storage-Bucket-URL: Geben Sie den Cloud Storage-Bucket-URI ein:

      gs://mulesoft-audit-logs/
      • Ersetzen Sie mulesoft-audit-logs durch den tatsächlichen Namen des Buckets.

    • Option zum Löschen der Quelle: Wählen Sie die gewünschte Option zum Löschen aus:

      • Dateien nie löschen: Dateien werden nie aus der Quelle gelöscht (empfohlen für Tests).
      • Übertragene Dateien und leere Verzeichnisse löschen: Dateien und leere Verzeichnisse werden nach einem erfolgreichen Abruf aus der Quelle gelöscht.

    • Maximales Dateialter: Dateien einschließen, die in den letzten Tagen geändert wurden. Der Standardwert ist 180 Tage.

    • Asset-Namespace: Der Asset-Namespace.

    • Aufnahmelabels: Das Label, das auf die Ereignisse aus diesem Feed angewendet werden soll.

  11. Klicken Sie auf Weiter.

  12. Prüfen Sie die neue Feedkonfiguration auf dem Bildschirm Abschließen und klicken Sie dann auf Senden.

Dem Google SecOps-Dienstkonto IAM-Berechtigungen gewähren

Das Google SecOps-Dienstkonto benötigt zwei Rollen für Ihren Cloud Storage-Bucket: Storage-Objekt-Betrachter zum Lesen der Logobjekte und eine Rolle auf Bucket-Ebene zum Lesen der Bucket-Metadaten.

  1. Rufen Sie Cloud Storage > Buckets auf.
  2. Klicken Sie auf den Namen Ihres Buckets.
  3. Wechseln Sie zum Tab Berechtigungen.
  4. Klicken Sie auf Zugriff erlauben.
  5. Geben Sie die folgenden Konfigurationsdetails an:
    • Hauptkonten hinzufügen: Fügen Sie die E‑Mail-Adresse des Google SecOps-Dienstkontos ein.
    • Rollen zuweisen: Wählen Sie die beiden folgenden Rollen aus:
    • Storage-Objekt-Betrachter: Liest die Logobjekte.
    • Leser alter Storage-Buckets: Berechtigt zum Lesen der Bucket-Metadaten. Wenn Sie die Löschoption Übertragene Dateien und leere Verzeichnisse löschen ausgewählt haben, wählen Sie stattdessen Storage Legacy Bucket Writer aus. Dadurch wird auch die Löschberechtigung gewährt.

  6. Klicken Sie auf Speichern.

Benötigen Sie weitere Hilfe? Antworten von Community-Mitgliedern und Google SecOps-Experten erhalten