ServiceNow-Audit-Logs

Unterstützt in:

In diesem Dokument wird beschrieben, wie Sie ServiceNow-Audit-Logs mit verschiedenen Methoden in Google Security Operations aufnehmen.

Option A: GCS mit Cloud Run-Funktion

Bei dieser Methode wird eine Cloud Run-Funktion verwendet, um die ServiceNow REST API regelmäßig nach Audit-Logs abzufragen und sie in einem GCS-Bucket zu speichern. Google Security Operations erfasst dann die Logs aus dem GCS-Bucket.

Hinweis

Prüfen Sie, ob die folgenden Voraussetzungen erfüllt sind:

  • Eine Google SecOps-Instanz
  • Privilegierter Zugriff auf den ServiceNow-Mandanten oder die API mit den entsprechenden Rollen (in der Regel admin oder ein Nutzer mit Lesezugriff auf die Tabelle „sys_audit“)
  • Ein GCP-Projekt mit aktivierter Cloud Storage API
  • Berechtigungen zum Erstellen und Verwalten von GCS-Buckets
  • Berechtigungen zum Verwalten von IAM-Richtlinien für GCS-Buckets
  • Berechtigungen zum Erstellen von Cloud Run-Diensten, Pub/Sub-Themen und Cloud Scheduler-Jobs

Voraussetzungen für ServiceNow erfassen (IDs, API-Schlüssel, Organisations-IDs, Tokens)

  1. Melden Sie sich in der ServiceNow Admin Console an.
  2. Klicken Sie auf Systemsicherheit > Nutzer und Gruppen > Nutzer.
  3. Erstellen Sie einen neuen Nutzer oder wählen Sie einen vorhandenen Nutzer mit den entsprechenden Berechtigungen für den Zugriff auf Audit-Logs aus.
  4. Kopieren Sie die folgenden Details und speichern Sie sie an einem sicheren Ort:

    • Nutzername
    • Passwort
    • Instanz-URL (z. B. https://instance.service-now.com)

ACL für Nutzer ohne Administratorberechtigungen konfigurieren

Wenn Sie ein Nutzerkonto verwenden möchten, das kein Administratorkonto ist, müssen Sie eine benutzerdefinierte Zugriffssteuerungsliste (Access Control List, ACL) erstellen, um Lesezugriff auf die Tabelle „sys_audit“ zu gewähren:

  1. Melden Sie sich als Administrator in der ServiceNow Admin Console an.
  2. Rufen Sie Systemsicherheit > Zugriffssteuerung (ACL) auf.
  3. Klicken Sie auf Neu.
  4. Geben Sie die folgenden Konfigurationsdetails an:
    • Typ: Wählen Sie Aufzeichnung aus.
    • Vorgang: Wählen Sie read aus.
    • Name: Geben Sie sys_audit ein.
    • Beschreibung: Geben Sie Allow read access to sys_audit table for Chronicle integration ein.
  5. Fügen Sie im Feld Requires role (Rolle erforderlich) die Rolle hinzu, die Ihrem Integrationsnutzer zugewiesen ist, z. B. chronicle_reader.
  6. Klicken Sie auf Senden.
  7. Prüfen Sie, ob die ACL aktiv ist und der Nutzer die Tabelle „sys_audit“ abfragen kann.

Google 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. servicenow-audit-logs.
    Standorttyp Wählen Sie je nach Bedarf aus (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)
    Zugriffssteuerung Einheitlich (empfohlen)
    Schutzmaßnahmen Optional: Objektversionsverwaltung oder Aufbewahrungsrichtlinie aktivieren
  6. Klicken Sie auf Erstellen.

Dienstkonto für Cloud Run-Funktion erstellen

Die Cloud Run-Funktion benötigt ein Dienstkonto mit Berechtigungen zum Schreiben in den GCS-Bucket und zum Aufrufen durch Pub/Sub.

Dienstkonto erstellen

  1. Wechseln Sie in der GCP Console zu IAM & Verwaltung > Dienstkonten.
  2. Klicken Sie auf Dienstkonto erstellen.
  3. Geben Sie die folgenden Konfigurationsdetails an:
    • Name des Dienstkontos: Geben Sie servicenow-audit-collector-sa ein.
    • Beschreibung des Dienstkontos: Geben Sie Service account for Cloud Run function to collect ServiceNow audit logs ein.
  4. Klicken Sie auf Erstellen und fortfahren.
  5. Fügen Sie im Abschnitt Diesem Dienstkonto Zugriff auf das Projekt erteilen die folgenden Rollen hinzu:
    1. Klicken Sie auf Rolle auswählen.
    2. Suchen Sie nach Storage-Objekt-Administrator 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 GCS-Bucket schreiben und Statusdateien verwalten
  • Cloud Run-Aufrufer: Pub/Sub darf die Funktion aufrufen.
  • Cloud Functions-Invoker: Funktionsaufruf zulassen

IAM-Berechtigungen für GCS-Bucket erteilen

Gewähren Sie dem Dienstkonto Schreibberechtigungen für den GCS-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. servicenow-audit-collector-sa@PROJECT_ID.iam.gserviceaccount.com).
    • Rollen zuweisen: Wählen Sie Storage-Objekt-Administrator 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 servicenow-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 ServiceNow API abzurufen und in GCS zu schreiben.

  1. Rufen Sie in der GCP Console Cloud Run auf.
  2. Klicken Sie auf Dienst erstellen.
  3. Wählen Sie Funktion aus, um eine Funktion mit einem Inline-Editor zu erstellen.
  4. Geben Sie im Abschnitt Konfigurieren die folgenden Konfigurationsdetails an:

    Einstellung Wert
    Dienstname servicenow-audit-collector
    Region Wählen Sie die Region aus, die Ihrem GCS-Bucket entspricht (z. B. us-central1).
    Laufzeit Wählen Sie Python 3.12 oder höher aus.
  5. Im Abschnitt Trigger (optional):

    1. Klicken Sie auf + Trigger hinzufügen.
    2. Wählen Sie Cloud Pub/Sub aus.
    3. Wählen Sie unter Cloud Pub/Sub-Thema auswählen das Pub/Sub-Thema (servicenow-audit-trigger) aus.
    4. Klicken Sie auf Speichern.
  6. Im Abschnitt Authentifizierung:

    1. Wählen Sie Authentifizierung erforderlich aus.
    2. Identitäts- und Zugriffsverwaltung
  7. Scrollen Sie nach unten und maximieren Sie Container, Netzwerk, Sicherheit.

  8. Rufen Sie den Tab Sicherheit auf:

    • Dienstkonto: Wählen Sie das Dienstkonto aus (servicenow-audit-collector-sa).
  9. 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 servicenow-audit-logs Name des GCS-Buckets
    GCS_PREFIX audit-logs Präfix für Protokolldateien
    STATE_KEY audit-logs/state.json Statusdateipfad
    API_BASE_URL https://instance.service-now.com ServiceNow-Instanz-URL
    API_USERNAME your-username ServiceNow-Nutzername
    API_PASSWORD your-password ServiceNow-Passwort
    PAGE_SIZE 1000 Datensätze pro Seite
    MAX_PAGES 1000 Maximale Anzahl abzurufender Seiten
  10. Scrollen Sie im Bereich Variablen und Secrets nach unten zu Anfragen:

    • Zeitlimit für Anfragen: Geben Sie 600 Sekunden (10 Minuten) ein.
  11. Rufen Sie den Tab Einstellungen auf:

    • Im Abschnitt Ressourcen:
      • Arbeitsspeicher: Wählen Sie 512 MiB oder höher aus.
      • CPU: Wählen Sie 1 aus.
  12. 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).
  13. Klicken Sie auf Erstellen.

  14. Warten Sie ein bis zwei Minuten, bis der Dienst erstellt wurde.

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

Funktionscode hinzufügen

  1. Geben Sie main in das Feld Einstiegspunkt ein.
  2. Erstellen Sie im Inline-Codeeditor zwei Dateien:

    • Erste Datei: main.py::

          import functions_framework
          from google.cloud import storage
          import json
          import os
          import urllib3
          from datetime import datetime, timezone, timedelta
          import time
          import base64
      
          # 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()
      
          # Environment variables
          GCS_BUCKET = os.environ.get('GCS_BUCKET')
          GCS_PREFIX = os.environ.get('GCS_PREFIX', 'audit-logs')
          STATE_KEY = os.environ.get('STATE_KEY', 'audit-logs/state.json')
          API_BASE = os.environ.get('API_BASE_URL')
          USERNAME = os.environ.get('API_USERNAME')
          PASSWORD = os.environ.get('API_PASSWORD')
          PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '1000'))
          MAX_PAGES = int(os.environ.get('MAX_PAGES', '1000'))
      
          def parse_datetime(value: str) -> datetime:
              """Parse ServiceNow datetime string to datetime object."""
              # ServiceNow format: YYYY-MM-DD HH:MM:SS
              try:
                  return datetime.strptime(value, '%Y-%m-%d %H:%M:%S').replace(tzinfo=timezone.utc)
              except ValueError:
                  # Try ISO format as fallback
                  if value.endswith("Z"):
                      value = value[:-1] + "+00:00"
                  return datetime.fromisoformat(value)
      
          @functions_framework.cloud_event
          def main(cloud_event):
              """
              Cloud Run function triggered by Pub/Sub to fetch ServiceNow audit logs and write to GCS.
      
              Args:
                  cloud_event: CloudEvent object containing Pub/Sub message
              """
      
              if not all([GCS_BUCKET, API_BASE, USERNAME, PASSWORD]):
                  print('Error: Missing required environment variables')
                  return
      
              try:
                  # Get GCS bucket
                  bucket = storage_client.bucket(GCS_BUCKET)
      
                  # Load state
                  state = load_state(bucket, STATE_KEY)
      
                  # Determine time window
                  now = datetime.now(timezone.utc)
                  last_time = None
      
                  if isinstance(state, dict) and state.get("last_event_time"):
                      try:
                          last_time = parse_datetime(state["last_event_time"])
                          # Overlap by 2 minutes to catch any delayed events
                          last_time = last_time - timedelta(minutes=2)
                      except Exception as e:
                          print(f"Warning: Could not parse last_event_time: {e}")
      
                  if last_time is None:
                      last_time = now - timedelta(hours=24)
      
                  print(f"Fetching logs from {last_time.strftime('%Y-%m-%d %H:%M:%S')} to {now.strftime('%Y-%m-%d %H:%M:%S')}")
      
                  # Fetch logs
                  records, newest_event_time = fetch_logs(
                      api_base=API_BASE,
                      username=USERNAME,
                      password=PASSWORD,
                      start_time=last_time,
                      end_time=now,
                      page_size=PAGE_SIZE,
                      max_pages=MAX_PAGES,
                  )
      
                  if not records:
                      print("No new log records found.")
                      save_state(bucket, STATE_KEY, now.strftime('%Y-%m-%d %H:%M:%S'))
                      return
      
                  # Write to GCS as NDJSON
                  timestamp = now.strftime('%Y%m%d_%H%M%S')
                  object_key = f"{GCS_PREFIX}/logs_{timestamp}.ndjson"
                  blob = bucket.blob(object_key)
      
                  ndjson = '\n'.join([json.dumps(record, ensure_ascii=False) for record in records]) + '\n'
                  blob.upload_from_string(ndjson, content_type='application/x-ndjson')
      
                  print(f"Wrote {len(records)} records to gs://{GCS_BUCKET}/{object_key}")
      
                  # Update state with newest event time
                  if newest_event_time:
                      save_state(bucket, STATE_KEY, newest_event_time)
                  else:
                      save_state(bucket, STATE_KEY, now.strftime('%Y-%m-%d %H:%M:%S'))
      
                  print(f"Successfully processed {len(records)} records")
      
              except Exception as e:
                  print(f'Error processing logs: {str(e)}')
                  raise
      
          def load_state(bucket, key):
              """Load state from GCS."""
              try:
                  blob = bucket.blob(key)
                  if blob.exists():
                      state_data = blob.download_as_text()
                      return json.loads(state_data)
              except Exception as e:
                  print(f"Warning: Could not load state: {e}")
      
              return {}
      
          def save_state(bucket, key, last_event_time: str):
              """Save the last event timestamp to GCS state file."""
              try:
                  state = {'last_event_time': last_event_time}
                  blob = bucket.blob(key)
                  blob.upload_from_string(
                      json.dumps(state, indent=2),
                      content_type='application/json'
                  )
                  print(f"Saved state: last_event_time={last_event_time}")
              except Exception as e:
                  print(f"Warning: Could not save state: {e}")
      
          def fetch_logs(api_base: str, username: str, password: str, start_time: datetime, end_time: datetime, page_size: int, max_pages: int):
              """
              Fetch logs from ServiceNow sys_audit table with pagination and rate limiting.
      
              Args:
                  api_base: ServiceNow instance URL
                  username: ServiceNow username
                  password: ServiceNow password
                  start_time: Start time for log query
                  end_time: End time for log query
                  page_size: Number of records per page
                  max_pages: Maximum total pages to fetch
      
              Returns:
                  Tuple of (records list, newest_event_time string)
              """
              # Clean up base URL
              base_url = api_base.rstrip('/')
      
              endpoint = f"{base_url}/api/now/table/sys_audit"
      
              # Encode credentials using UTF-8
              auth_string = f"{username}:{password}"
              auth_bytes = auth_string.encode('utf-8')
              auth_b64 = base64.b64encode(auth_bytes).decode('utf-8')
      
              headers = {
                  'Authorization': f'Basic {auth_b64}',
                  'Accept': 'application/json',
                  'Content-Type': 'application/json',
                  'User-Agent': 'GoogleSecOps-ServiceNowCollector/1.0'
              }
      
              records = []
              newest_time = None
              page_num = 0
              backoff = 1.0
              offset = 0
      
              # Format timestamps for ServiceNow (YYYY-MM-DD HH:MM:SS)
              start_time_str = start_time.strftime('%Y-%m-%d %H:%M:%S')
      
              while True:
                  page_num += 1
      
                  if len(records) >= page_size * max_pages:
                      print(f"Reached max_pages limit ({max_pages})")
                      break
      
                  # Build query parameters
                  # Use >= operator for sys_created_on field (on or after)
                  params = []
                  params.append(f"sysparm_query=sys_created_on>={start_time_str}")
                  params.append(f"sysparm_display_value=true")
                  params.append(f"sysparm_limit={page_size}")
                  params.append(f"sysparm_offset={offset}")
      
                  url = f"{endpoint}?{'&'.join(params)}"
      
                  try:
                      response = http.request('GET', url, headers=headers)
      
                      # Handle rate limiting with exponential backoff
                      if response.status == 429:
                          retry_after = int(response.headers.get('Retry-After', str(int(backoff))))
                          print(f"Rate limited (429). Retrying after {retry_after}s...")
                          time.sleep(retry_after)
                          backoff = min(backoff * 2, 30.0)
                          continue
      
                      backoff = 1.0
      
                      if response.status != 200:
                          print(f"HTTP Error: {response.status}")
                          response_text = response.data.decode('utf-8')
                          print(f"Response body: {response_text}")
                          return [], None
      
                      data = json.loads(response.data.decode('utf-8'))
                      page_results = data.get('result', [])
      
                      if not page_results:
                          print(f"No more results (empty page)")
                          break
      
                      print(f"Page {page_num}: Retrieved {len(page_results)} events")
                      records.extend(page_results)
      
                      # Track newest event time
                      for event in page_results:
                          try:
                              event_time = event.get('sys_created_on')
                              if event_time:
                                  if newest_time is None or parse_datetime(event_time) > parse_datetime(newest_time):
                                      newest_time = event_time
                          except Exception as e:
                              print(f"Warning: Could not parse event time: {e}")
      
                      # Check for more results
                      if len(page_results) < page_size:
                          print(f"Reached last page (size={len(page_results)} < limit={page_size})")
                          break
      
                      # Move to next page
                      offset += page_size
      
                      # Small delay to avoid rate limiting
                      time.sleep(0.1)
      
                  except Exception as e:
                      print(f"Error fetching logs: {e}")
                      return [], None
      
              print(f"Retrieved {len(records)} total records 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).

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 servicenow-audit-collector-hourly
    Region Dieselbe Region wie für die Cloud Run-Funktion auswählen
    Frequenz 0 * * * * (jede Stunde, zur vollen Stunde)
    Zeitzone Zeitzone auswählen (UTC empfohlen)
    Zieltyp Pub/Sub
    Thema Wählen Sie das Pub/Sub-Thema aus (servicenow-audit-trigger).
    Nachrichtentext {} (leeres JSON-Objekt)
  4. Klicken Sie auf Erstellen.

Optionen für die Häufigkeit des Zeitplans

  • Wählen Sie die Häufigkeit basierend auf dem Logvolumen und den Latenzanforderungen aus:

    Häufigkeit Cron-Ausdruck Anwendungsfall
    Alle 5 Minuten */5 * * * * Hohes Volumen, niedrige Latenz
    Alle 15 Minuten */15 * * * * Mittleres Suchvolumen
    Stündlich 0 * * * * Standard (empfohlen)
    Alle 6 Stunden 0 */6 * * * Geringes Volumen, Batchverarbeitung
    Täglich 0 0 * * * Erhebung von Verlaufsdaten

Integration testen

  1. Suchen Sie in der Cloud Scheduler-Konsole nach Ihrem Job.
  2. Klicken Sie auf Force run (Ausführung erzwingen), um den Job manuell auszulösen.
  3. Warten Sie einige Sekunden.
  4. Rufen Sie Cloud Run > Dienste auf.
  5. Klicken Sie auf den Namen Ihrer Funktion (servicenow-audit-collector).
  6. Klicken Sie auf den Tab Logs.
  7. Prüfen Sie, ob die Funktion erfolgreich ausgeführt wurde. Achten Sie auf Folgendes:

    Fetching logs from YYYY-MM-DD HH:MM:SS to YYYY-MM-DD HH:MM:SS
    Page 1: Retrieved X events
    Wrote X records to gs://bucket-name/audit-logs/logs_YYYYMMDD_HHMMSS.ndjson
    Successfully processed X records
    
  8. Rufen Sie Cloud Storage > Buckets auf.

  9. Klicken Sie auf den Namen Ihres Buckets.

  10. Rufen Sie den Präfixordner (audit-logs/) auf.

  11. Prüfen Sie, ob eine neue .ndjson-Datei mit dem aktuellen Zeitstempel erstellt wurde.

Wenn in den Logs Fehler angezeigt werden, gehen Sie so vor:

  • HTTP 401: API-Anmeldedaten in Umgebungsvariablen prüfen
  • HTTP 403: Prüfen Sie, ob das Konto die erforderlichen Berechtigungen hat (Administratorrolle oder benutzerdefinierte ACL für sys_audit).
  • HTTP 429: Ratenbegrenzung – die Funktion wird automatisch mit Backoff wiederholt.
  • Fehlende Umgebungsvariablen: Prüfen Sie, ob alle erforderlichen Variablen festgelegt sind.

Google SecOps-Dienstkonto abrufen

Google SecOps verwendet ein eindeutiges Dienstkonto, um Daten aus Ihrem GCS-Bucket zu lesen. Sie müssen diesem Dienstkonto Zugriff auf Ihren Bucket gewähren.

E-Mail-Adresse des Dienstkontos abrufen

  1. Rufen Sie die SIEM-Einstellungen > Feeds auf.
  2. Klicken Sie auf Neuen 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. ServiceNow Audit logs.
  5. Wählen Sie Google Cloud Storage V2 als Quelltyp aus.
  6. Wählen Sie ServiceNow Audit als Logtyp 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 für den nächsten Schritt.

Dem Google SecOps-Dienstkonto IAM-Berechtigungen gewähren

Das Google SecOps-Dienstkonto benötigt die Rolle Storage-Objekt-Betrachter für Ihren GCS-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: Fügen Sie die E‑Mail-Adresse des Google SecOps-Dienstkontos ein.
    • Rollen zuweisen: Wählen Sie Storage-Objekt-Betrachter aus.
  6. Klicken Sie auf Speichern.

Feed in Google SecOps konfigurieren, um ServiceNow-Audit-Logs aufzunehmen

  1. Rufen Sie die SIEM-Einstellungen > Feeds auf.
  2. Klicken Sie auf Neuen 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. ServiceNow Audit logs.
  5. Wählen Sie Google Cloud Storage V2 als Quelltyp aus.
  6. Wählen Sie ServiceNow Audit als Logtyp aus.
  7. Klicken Sie auf Weiter.
  8. Geben Sie Werte für die folgenden Eingabeparameter an:

    • Storage-Bucket-URL: Geben Sie den GCS-Bucket-URI mit dem Präfixpfad ein:

      gs://servicenow-audit-logs/audit-logs/
      
      • Ersetzen Sie:

        • servicenow-audit-logs: Der Name Ihres GCS-Buckets.
        • audit-logs: Präfix/Ordnerpfad, in dem Logs gespeichert werden.
    • Option zum Löschen der Quelle: Wählen Sie die gewünschte Löschoption aus:

      • Nie: Es werden nach Übertragungen nie Dateien gelöscht (empfohlen für Tests).
      • Übertragene Dateien löschen: Dateien werden nach der erfolgreichen Übertragung gelöscht.
      • Übertragene Dateien und leere Verzeichnisse löschen: Löscht Dateien und leere Verzeichnisse nach der erfolgreichen Übertragung.

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

  9. Klicken Sie auf Weiter.

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

Option B: Bindplane-Agent mit Syslog

Bei dieser Methode wird ein BindPlane-Agent verwendet, um ServiceNow-Audit-Logs zu erfassen und an Google Security Operations weiterzuleiten. Da ServiceNow Syslog für Audit-Logs nicht nativ unterstützt, verwenden wir ein Skript, um die ServiceNow REST API abzufragen und die Logs über Syslog an den Bindplane-Agent weiterzuleiten.

Hinweis

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

  • Eine Google SecOps-Instanz
  • Windows Server 2016 oder höher oder Linux-Host mit systemd
  • Netzwerkverbindung zwischen Bindplane-Agent und ServiceNow
  • Wenn Sie den Agent hinter einem Proxy ausführen, müssen die Firewallports gemäß den Anforderungen des Bindplane-Agents geöffnet sein.
  • Privilegierter Zugriff auf die ServiceNow-Verwaltungskonsole oder -Appliance mit den entsprechenden Rollen (in der Regel admin oder Nutzer mit Lesezugriff auf die Tabelle „sys_audit“)

Authentifizierungsdatei für die Aufnahme in Google SecOps abrufen

  1. Melden Sie sich in der Google SecOps-Konsole an.
  2. Rufen Sie die SIEM-Einstellungen > Collection Agent auf.
  3. Klicken Sie auf Herunterladen, um die Datei zur Authentifizierung der Aufnahme herunterzuladen.
  4. Speichern Sie die Datei sicher auf dem System, auf dem Bindplane installiert wird.

Google SecOps-Kundennummer abrufen

  1. Melden Sie sich in der Google SecOps-Konsole an.
  2. Rufen Sie die SIEM-Einstellungen > Profile auf.
  3. Kopieren und speichern Sie die Kunden-ID aus dem Bereich Organisationsdetails.

BindPlane-Agent installieren

Installieren Sie den Bindplane-Agent auf Ihrem Windows- oder Linux-Betriebssystem gemäß der folgenden Anleitung.

Fenstermontage

  1. Öffnen Sie die Eingabeaufforderung oder PowerShell als Administrator.
  2. Führen Sie dazu diesen Befehl aus:

    msiexec /i "https://github.com/observIQ/bindplane-agent/releases/latest/download/observiq-otel-collector.msi" /quiet
    
  3. Warten Sie, bis die Installation abgeschlossen ist.

  4. Überprüfen Sie die Installation mit folgendem Befehl:

    sc query observiq-otel-collector
    

Der Dienst sollte als RUNNING (Wird ausgeführt) angezeigt werden.

Linux-Installation

  1. Öffnen Sie ein Terminal mit Root- oder Sudo-Berechtigungen.
  2. Führen Sie dazu diesen Befehl aus:

    sudo sh -c "$(curl -fsSlL https://github.com/observiq/bindplane-agent/releases/latest/download/install_unix.sh)" install_unix.sh
    
  3. Warten Sie, bis die Installation abgeschlossen ist.

  4. Überprüfen Sie die Installation mit folgendem Befehl:

    sudo systemctl status observiq-otel-collector
    

Der Dienst sollte als aktiv (wird ausgeführt) angezeigt werden.

Zusätzliche Installationsressourcen

Weitere Installationsoptionen und Informationen zur Fehlerbehebung finden Sie in der Installationsanleitung für den Bindplane-Agent.

BindPlane-Agent zum Erfassen von Syslog-Daten und Senden an Google SecOps konfigurieren

Konfigurationsdatei suchen

Linux:bash sudo nano /etc/bindplane-agent/config.yaml

Windows:cmd notepad "C:\Program Files\observIQ OpenTelemetry Collector\config.yaml"

Konfigurationsdatei bearbeiten

Ersetzen Sie den gesamten Inhalt von config.yaml durch die folgende Konfiguration:

```yaml
receivers:
  udplog:
    listen_address: "0.0.0.0:514"

exporters:
  chronicle/servicenow_audit:
    compression: gzip
    creds_file_path: '/path/to/ingestion-authentication-file.json'
    customer_id: '<YOUR_CUSTOMER_ID>'
    endpoint: <CUSTOMER_REGION_ENDPOINT>
    log_type: 'SERVICENOW_AUDIT'
    raw_log_field: body
    ingestion_labels:
      service: servicenow

service:
  pipelines:
    logs/servicenow_to_chronicle:
      receivers:
        - udplog
      exporters:
        - chronicle/servicenow_audit
```

Konfigurationsparameter

Ersetzen Sie die folgenden Platzhalter:

  • listen_address: IP-Adresse und Port, auf dem gelauscht werden soll. Verwenden Sie 0.0.0.0:514, um alle Schnittstellen auf Port 514 zu überwachen.
  • creds_file_path: Vollständiger Pfad zur Datei für die Authentifizierung bei der Aufnahme:
    • Linux: /etc/bindplane-agent/ingestion-auth.json
    • Windows: C:\Program Files\observIQ OpenTelemetry Collector\ingestion-auth.json
  • <YOUR_CUSTOMER_ID>: Kunden-ID aus dem vorherigen Schritt.
  • <CUSTOMER_REGION_ENDPOINT>: Regionale Endpunkt-URL:
    • USA: malachiteingestion-pa.googleapis.com
    • Europa: europe-malachiteingestion-pa.googleapis.com
    • Asien: asia-southeast1-malachiteingestion-pa.googleapis.com
    • Eine vollständige Liste finden Sie unter Regionale Endpunkte.

Konfigurationsdatei speichern

Speichern Sie die Datei nach dem Bearbeiten: * Linux: Drücken Sie Ctrl+O, dann Enter und dann Ctrl+X. * Windows: Klicken Sie auf Datei > Speichern.

Bindplane-Agent neu starten, um die Änderungen zu übernehmen

  • Führen Sie den folgenden Befehl aus, um den Bindplane-Agent unter Linux neu zu starten:

    sudo systemctl restart observiq-otel-collector
    
    1. Prüfen Sie, ob der Dienst ausgeführt wird:

      sudo systemctl status observiq-otel-collector
      
    2. Logs auf Fehler prüfen:

      sudo journalctl -u observiq-otel-collector -f
      
  • Wählen Sie eine der folgenden Optionen aus, um den Bindplane-Agent unter Windows neu zu starten:

    • So verwenden Sie die Eingabeaufforderung oder PowerShell als Administrator:

      net stop observiq-otel-collector && net start observiq-otel-collector
      
    • Services Console verwenden:

    1. Drücken Sie Win+R, geben Sie services.msc ein und drücken Sie die Eingabetaste.
    2. Suchen Sie nach observIQ OpenTelemetry Collector.
    3. Klicken Sie mit der rechten Maustaste und wählen Sie Neu starten aus.

    4. Prüfen Sie, ob der Dienst ausgeführt wird:

      sc query observiq-otel-collector
      
    5. Logs auf Fehler prüfen:

      type "C:\Program Files\observIQ OpenTelemetry Collector\log\collector.log"
      

Skript zum Weiterleiten von ServiceNow-Audit-Logs an Syslog erstellen

Da ServiceNow Syslog für Audit-Logs nicht nativ unterstützt, erstellen wir ein Skript, das die ServiceNow REST API abfragt und die Logs an Syslog weiterleitet. Dieses Skript kann so geplant werden, dass es regelmäßig ausgeführt wird.

Beispiel für Python-Script (Linux)

  • Erstellen Sie eine Datei mit dem Namen servicenow_audit_to_syslog.py und dem folgendem Inhalt:

    import urllib3
    import json
    import datetime
    import base64
    import socket
    import time
    import os
    
    # ServiceNow API details
    BASE_URL = 'https://instance.service-now.com'  # Replace with your ServiceNow instance URL
    USERNAME = 'admin'  # Replace with your ServiceNow username
    PASSWORD = 'password'  # Replace with your ServiceNow password
    
    # Syslog details
    SYSLOG_SERVER = '127.0.0.1'  # Replace with your Bindplane agent IP
    SYSLOG_PORT = 514  # Replace with your Bindplane agent port
    
    # State file to keep track of last run
    STATE_FILE = '/tmp/servicenow_audit_last_run.txt'
    
    # Pagination settings
    PAGE_SIZE = 1000
    MAX_PAGES = 1000
    
    def get_last_run_timestamp():
        try:
            with open(STATE_FILE, 'r') as f:
                return f.read().strip()
        except:
            return '1970-01-01 00:00:00'
    
    def update_state_file(timestamp):
        with open(STATE_FILE, 'w') as f:
            f.write(timestamp)
    
    def send_to_syslog(message):
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        sock.sendto(message.encode(), (SYSLOG_SERVER, SYSLOG_PORT))
        sock.close()
    
    def get_audit_logs(last_run_timestamp):
        """
        Query ServiceNow sys_audit table with proper pagination.
        Uses sys_created_on field for timestamp filtering.
        """
        # Encode credentials using UTF-8
        auth_string = f"{USERNAME}:{PASSWORD}"
        auth_bytes = auth_string.encode('utf-8')
        auth_encoded = base64.b64encode(auth_bytes).decode('utf-8')
    
        # Setup HTTP client
        http = urllib3.PoolManager()
        headers = {
            'Authorization': f'Basic {auth_encoded}',
            'Accept': 'application/json'
        }
    
        results = []
        offset = 0
    
        # Format timestamp for ServiceNow (YYYY-MM-DD HH:MM:SS format)
        # Convert ISO format to ServiceNow format if needed
        if 'T' in last_run_timestamp:
            last_run_timestamp = last_run_timestamp.replace('T', ' ').split('.')[0]
    
        for page in range(MAX_PAGES):
            # Build query with pagination
            # Use >= operator for sys_created_on field (on or after)
            query_params = (
                f"sysparm_query=sys_created_on>={last_run_timestamp}"
                f"&sysparm_display_value=true"
                f"&sysparm_limit={PAGE_SIZE}"
                f"&sysparm_offset={offset}"
            )
    
            url = f"{BASE_URL}/api/now/table/sys_audit?{query_params}"
    
            try:
                response = http.request('GET', url, headers=headers)
    
                if response.status == 200:
                    data = json.loads(response.data.decode('utf-8'))
                    chunk = data.get('result', [])
                    results.extend(chunk)
    
                    # Stop if we got fewer records than PAGE_SIZE (last page)
                    if len(chunk) < PAGE_SIZE:
                        break
    
                    # Move to next page
                    offset += PAGE_SIZE
                else:
                    print(f"Error querying ServiceNow API: {response.status} - {response.data.decode('utf-8')}")
                    break
    
            except Exception as e:
                print(f"Exception querying ServiceNow API: {str(e)}")
                break
    
        return results
    
    def main():
        # Get last run timestamp
        last_run_timestamp = get_last_run_timestamp()
    
        # Current timestamp for this run
        current_timestamp = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    
        # Query ServiceNow API for audit logs
        audit_logs = get_audit_logs(last_run_timestamp)
    
        if audit_logs:
            # Send each log to syslog
            for log in audit_logs:
                # Format the log as JSON
                log_json = json.dumps(log)
    
                # Send to syslog
                send_to_syslog(log_json)
    
                # Sleep briefly to avoid flooding
                time.sleep(0.01)
    
            # Update state file
            update_state_file(current_timestamp)
    
            print(f"Successfully forwarded {len(audit_logs)} audit logs to syslog")
        else:
            print("No new audit logs to forward")
    
    if __name__ == "__main__":
        main()
    

Geplante Ausführung einrichten (Linux)

  • Machen Sie das Skript ausführbar:

    chmod +x servicenow_audit_to_syslog.py
    
  • Erstellen Sie einen Cronjob, um das Skript stündlich auszuführen:

    crontab -e
    
  • Fügen Sie folgende Zeile hinzu:

    0 * * * * /usr/bin/python3 /path/to/servicenow_audit_to_syslog.py >> /tmp/servicenow_audit_to_syslog.log 2>&1
    

PowerShell-Skriptbeispiel (Windows)

  • Erstellen Sie eine Datei mit dem Namen ServiceNow-Audit-To-Syslog.ps1 und dem folgendem Inhalt:

    # ServiceNow API details
    $BaseUrl = 'https://instance.service-now.com'  # Replace with your ServiceNow instance URL
    $Username = 'admin'  # Replace with your ServiceNow username
    $Password = 'password'  # Replace with your ServiceNow password
    
    # Syslog details
    $SyslogServer = '127.0.0.1'  # Replace with your Bindplane agent IP
    $SyslogPort = 514  # Replace with your Bindplane agent port
    
    # State file to keep track of last run
    $StateFile = "$env:TEMP\ServiceNowAuditLastRun.txt"
    
    # Pagination settings
    $PageSize = 1000
    $MaxPages = 1000
    
    function Get-LastRunTimestamp {
        try {
            if (Test-Path $StateFile) {
                return Get-Content $StateFile
            } else {
                return '1970-01-01 00:00:00'
            }
        } catch {
            return '1970-01-01 00:00:00'
        }
    }
    
    function Update-StateFile {
        param([string]$Timestamp)
        Set-Content -Path $StateFile -Value $Timestamp
    }
    
    function Send-ToSyslog {
        param([string]$Message)
        $UdpClient = New-Object System.Net.Sockets.UdpClient
        $UdpClient.Connect($SyslogServer, $SyslogPort)
        $Encoding = [System.Text.Encoding]::ASCII
        $Bytes = $Encoding.GetBytes($Message)
        $UdpClient.Send($Bytes, $Bytes.Length)
        $UdpClient.Close()
    }
    
    function Get-AuditLogs {
        param([string]$LastRunTimestamp)
    
        # Create auth header using UTF-8 encoding
        $Auth = [System.Convert]::ToBase64String([System.Text.Encoding]::UTF8.GetBytes("${Username}:${Password}"))
        $Headers = @{
            Authorization = "Basic ${Auth}"
            Accept = 'application/json'
        }
    
        $Results = @()
        $Offset = 0
    
        # Format timestamp for ServiceNow (YYYY-MM-DD HH:MM:SS format)
        # Convert ISO format to ServiceNow format if needed
        if ($LastRunTimestamp -match 'T') {
            $LastRunTimestamp = $LastRunTimestamp -replace 'T', ' '
            $LastRunTimestamp = $LastRunTimestamp -replace '\.\d+', ''
        }
    
        for ($page = 0; $page -lt $MaxPages; $page++) {
            # Build query with pagination
            # Use >= operator for sys_created_on field (on or after)
            $QueryParams = "sysparm_query=sys_created_on>=${LastRunTimestamp}&sysparm_display_value=true&sysparm_limit=${PageSize}&sysparm_offset=${Offset}"
            $Url = "${BaseUrl}/api/now/table/sys_audit?${QueryParams}"
    
            try {
                $Response = Invoke-RestMethod -Uri $Url -Headers $Headers -Method Get
                $Chunk = $Response.result
                $Results += $Chunk
    
                # Stop if we got fewer records than PageSize (last page)
                if ($Chunk.Count -lt $PageSize) {
                    break
                }
    
                # Move to next page
                $Offset += $PageSize
            } catch {
                Write-Error "Error querying ServiceNow API: $_"
                break
            }
        }
    
        return $Results
    }
    
    # Main execution
    $LastRunTimestamp = Get-LastRunTimestamp
    $CurrentTimestamp = (Get-Date).ToString('yyyy-MM-dd HH:mm:ss')
    
    $AuditLogs = Get-AuditLogs -LastRunTimestamp $LastRunTimestamp
    
    if ($AuditLogs -and $AuditLogs.Count -gt 0) {
        # Send each log to syslog
        foreach ($Log in $AuditLogs) {
            # Format the log as JSON
            $LogJson = $Log | ConvertTo-Json -Compress
    
            # Send to syslog
            Send-ToSyslog -Message $LogJson
    
            # Sleep briefly to avoid flooding
            Start-Sleep -Milliseconds 10
        }
    
        # Update state file
        Update-StateFile -Timestamp $CurrentTimestamp
    
        Write-Output "Successfully forwarded $($AuditLogs.Count) audit logs to syslog"
    } else {
        Write-Output "No new audit logs to forward"
    }
    

Geplante Ausführung einrichten (Windows)

  1. Öffnen Sie den Taskplaner.
  2. Klicken Sie auf Aufgabe erstellen.
  3. Geben Sie die folgende Konfiguration an:
    • Name: ServiceNowAuditToSyslog
    • Sicherheitsoptionen: werden unabhängig davon ausgeführt, ob der Nutzer angemeldet ist oder nicht
  4. Rufen Sie den Tab Trigger auf.
  5. Klicken Sie auf Neu und legen Sie fest, dass der Job stündlich ausgeführt werden soll.
  6. Rufen Sie den Tab Aktionen auf.
  7. Klicken Sie auf Neu und legen Sie Folgendes fest:
    • Aktion: Programm starten
    • Programm/Script: powershell.exe
    • Argumente: -ExecutionPolicy Bypass -File "C:\path\to\ServiceNow-Audit-To-Syslog.ps1"
  8. Klicken Sie auf OK, um die Aufgabe zu speichern.

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