Digital Shadows SearchLight-Logs erfassen

Unterstützt in:

In diesem Dokument wird beschrieben, wie Sie Digital Shadows SearchLight-Logs mit Amazon S3 in Google Security Operations aufnehmen. Der Parser extrahiert Sicherheitsereignisdaten aus den JSON-Logs. Es initialisiert Felder des Unified Data Model (UDM), parst die JSON-Nutzlast, ordnet relevante Felder dem UDM-Schema zu, extrahiert Entitäten wie E-Mail und Hostname mithilfe von Grok-Mustern und erstellt die Objekte security_result und metadata innerhalb des UDM-Ereignisses.

Hinweise

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

  • Eine Google SecOps-Instanz.
  • Privilegierter Zugriff auf den Digital Shadows SearchLight-Mandanten.
  • Privilegierter Zugriff auf AWS (S3, Identity and Access Management (IAM), Lambda, EventBridge).

Voraussetzungen für Digital Shadows SearchLight erfassen (IDs, API-Schlüssel, Organisations-IDs, Tokens)

  1. Melden Sie sich im Digital Shadows SearchLight-Portal an.
  2. Gehen Sie zu den Einstellungen> API-Anmeldedaten.
  3. Erstellen Sie einen neuen API-Client oder ein neues Schlüsselpaar.
  4. Kopieren Sie die folgenden Details und speichern Sie sie an einem sicheren Ort:
    • API-Schlüssel
    • API Secret (API-Secret)
    • Konto-ID
    • API-Basis-URL: https://api.searchlight.app/v1 oder https://portal-digitalshadows.com/api/v1

AWS S3-Bucket und IAM für Google SecOps konfigurieren

  1. Erstellen Sie einen Amazon S3-Bucket. Folgen Sie dazu der Anleitung unter Bucket erstellen.
  2. Speichern Sie den Namen und die Region des Buckets zur späteren Verwendung (z. B. digital-shadows-logs).
  3. Erstellen Sie einen Nutzer gemäß dieser Anleitung: IAM-Nutzer erstellen.
  4. Wählen Sie den erstellten Nutzer aus.
  5. Wählen Sie den Tab Sicherheitsanmeldedaten aus.
  6. Klicken Sie im Abschnitt Zugriffsschlüssel auf Zugriffsschlüssel erstellen.
  7. Wählen Sie Drittanbieterdienst als Anwendungsfall aus.
  8. Klicken Sie auf Weiter.
  9. Optional: Fügen Sie ein Beschreibungstag hinzu.
  10. Klicken Sie auf Zugriffsschlüssel erstellen.
  11. Klicken Sie auf CSV-Datei herunterladen, um den Access Key (Zugriffsschlüssel) und den Secret Access Key (geheimer Zugriffsschlüssel) für die zukünftige Verwendung zu speichern.
  12. Klicken Sie auf Fertig.
  13. Wählen Sie den Tab Berechtigungen aus.
  14. Klicken Sie im Bereich Berechtigungsrichtlinien auf Berechtigungen hinzufügen.
  15. Wählen Sie Berechtigungen hinzufügen aus.
  16. Wählen Sie Richtlinien direkt anhängen aus.
  17. Suchen Sie nach der Richtlinie AmazonS3FullAccess.
  18. Wählen Sie die Richtlinie aus.
  19. Klicken Sie auf Weiter.
  20. Klicken Sie auf Berechtigungen hinzufügen.

IAM-Richtlinie und ‑Rolle für S3-Uploads konfigurieren

  1. Rufen Sie in der AWS-Konsole IAM > Richtlinien auf.
  2. Klicken Sie auf Richtlinie erstellen> Tab „JSON“.
  3. Kopieren Sie die folgende Richtlinie und fügen Sie sie ein.
  4. Policy JSON (ersetzen Sie digital-shadows-logs, wenn Sie einen anderen Bucket-Namen eingegeben haben):

    {
      "Version": "2012-10-17",
      "Statement": [
        {
          "Sid": "AllowPutObjects",
          "Effect": "Allow",
          "Action": "s3:PutObject",
          "Resource": "arn:aws:s3:::digital-shadows-logs/*"
        },
        {
          "Sid": "AllowGetStateObject",
          "Effect": "Allow",
          "Action": "s3:GetObject",
          "Resource": "arn:aws:s3:::digital-shadows-logs/digital-shadows-searchlight/state.json"
        }
      ]
    }
    
  5. Klicken Sie auf Weiter > Richtlinie erstellen.

  6. Rufen Sie IAM > Rollen > Rolle erstellen > AWS-Service > Lambda auf.

  7. Hängen Sie die neu erstellte Richtlinie an.

  8. Geben Sie der Rolle den Namen digital-shadows-lambda-role und klicken Sie auf Rolle erstellen.

Lambda-Funktion erstellen

  1. Rufen Sie in der AWS Console Lambda > Funktionen > Funktion erstellen auf.
  2. Klicken Sie auf Von Grund auf erstellen.
  3. Geben Sie die folgenden Konfigurationsdetails an:

    Einstellung Wert
    Name digital-shadows-collector
    Laufzeit Python 3.13
    Architektur x86_64
    Ausführungsrolle digital-shadows-lambda-role
  4. Nachdem die Funktion erstellt wurde, öffnen Sie den Tab Code, löschen Sie den Stub und fügen Sie den folgenden Code (digital-shadows-collector.py) ein.

    import json
    import os
    import base64
    import logging
    import time
    from datetime import datetime, timedelta, timezone
    from urllib.parse import urlencode
    
    import boto3
    import urllib3
    
    logger = logging.getLogger()
    logger.setLevel(logging.INFO)
    
    HTTP = urllib3.PoolManager(retries=False)
    
    def _basic_auth_header(key: str, secret: str) -> str:
        token = base64.b64encode(f"{key}:{secret}".encode("utf-8")).decode("utf-8")
        return f"Basic {token}"
    
    def _load_state(s3, bucket, key, default_days=30) -> str:
        """Return ISO8601 checkpoint (UTC)."""
        try:
            obj = s3.get_object(Bucket=bucket, Key=key)
            state = json.loads(obj["Body"].read().decode("utf-8"))
            ts = state.get("last_timestamp")
            if ts:
                return ts
        except s3.exceptions.NoSuchKey:
            pass
        except Exception as e:
            logger.warning(f"State read error: {e}")
        return (datetime.now(timezone.utc) - timedelta(days=default_days)).isoformat()
    
    def _save_state(s3, bucket, key, ts: str) -> None:
        s3.put_object(
            Bucket=bucket,
            Key=key,
            Body=json.dumps({"last_timestamp": ts}).encode("utf-8"),
            ContentType="application/json",
        )
    
    def _get_json(url: str, headers: dict, params: dict, backoff_s=2, max_retries=3) -> dict:
        qs = f"?{urlencode(params)}" if params else ""
        for attempt in range(max_retries):
            r = HTTP.request("GET", f"{url}{qs}", headers=headers)
            if r.status == 200:
                return json.loads(r.data.decode("utf-8"))
            if r.status in (429, 500, 502, 503, 504):
                wait = backoff_s * (2 ** attempt)
                logger.warning(f"HTTP {r.status} from DS API, retrying in {wait}s")
                time.sleep(wait)
                continue
            raise RuntimeError(f"DS API error {r.status}: {r.data[:200]}")
        raise RuntimeError("Exceeded retry budget for DS API")
    
    def _collect(api_base, headers, path, since_ts, account_id, page_size, max_pages, time_param):
        items = []
        for page in range(max_pages):
            params = {
                "limit": page_size,
                "offset": page * page_size,
                time_param: since_ts,
            }
            if account_id:
                params["account-id"] = account_id
    
            data = _get_json(f"{api_base}/{path}", headers, params)
            batch = data.get("items") or data.get("data") or []
            if not batch:
                break
            items.extend(batch)
            if len(batch) < page_size:
                break
        return items
    
    def lambda_handler(event, context):
        # Required
        s3_bucket  = os.environ["S3_BUCKET"]
        api_key    = os.environ["DS_API_KEY"]
        api_secret = os.environ["DS_API_SECRET"]
    
        # Optional / defaults
        s3_prefix  = os.environ.get("S3_PREFIX", "digital-shadows-searchlight/")
        state_key  = os.environ.get("STATE_KEY", "digital-shadows-searchlight/state.json")
        api_base   = os.environ.get("API_BASE", "https://api.searchlight.app/v1")
        account_id = os.environ.get("DS_ACCOUNT_ID", "")
        page_size  = int(os.environ.get("PAGE_SIZE", "100"))
        max_pages  = int(os.environ.get("MAX_PAGES", "10"))
    
        s3 = boto3.client("s3")
        last_ts = _load_state(s3, s3_bucket, state_key)
        logger.info(f"Checkpoint: {last_ts}")
    
        headers = {
            "Authorization": _basic_auth_header(api_key, api_secret),
            "Accept": "application/json",
            "User-Agent": "Chronicle-DigitalShadows-S3/1.0",
        }
    
        records = []
    
        # Incidents (time filter often 'published-after' or 'updated-since' depending on tenancy)
        incidents = _collect(api_base, headers, "incidents", last_ts, account_id, page_size, max_pages, time_param="published-after")
        for incident in incidents:
            incident['_source_type'] = 'incident'
        records.extend(incidents)
    
        # Intelligence incidents (alerts)
        intel_incidents = _collect(api_base, headers, "intel-incidents", last_ts, account_id, page_size, max_pages, time_param="published-after")
        for intel in intel_incidents:
            intel['_source_type'] = 'intelligence_incident'
        records.extend(intel_incidents)
    
        # Indicators (IOCs)
        indicators = _collect(api_base, headers, "indicators", last_ts, account_id, page_size, max_pages, time_param="lastUpdated-after")
        for indicator in indicators:
            indicator['_source_type'] = 'ioc'
        records.extend(indicators)
    
        if records:
            # Choose newest timestamp seen in this batch
            newest = max(
                (r.get("updated") or r.get("raised") or r.get("lastUpdated") or last_ts)
                for r in records
            )
            key = f"{s3_prefix}digital_shadows_{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}.json"
            body = "\n".join(json.dumps(r, separators=(",", ":")) for r in records).encode("utf-8")
    
            s3.put_object(
                Bucket=s3_bucket,
                Key=key,
                Body=body,
                ContentType="application/x-ndjson",
            )
            _save_state(s3, s3_bucket, state_key, newest)
            msg = f"Wrote {len(records)} records to s3://{s3_bucket}/{key}"
        else:
            msg = "No new records"
    
        logger.info(msg)
        return {"statusCode": 200, "body": msg}
    
  5. Rufen Sie Konfiguration > Umgebungsvariablen auf.

  6. Klicken Sie auf Bearbeiten> Neue Umgebungsvariable hinzufügen.

  7. Geben Sie die in der folgenden Tabelle aufgeführten Umgebungsvariablen ein und ersetzen Sie die Beispielwerte durch Ihre Werte.

    Umgebungsvariablen

    Schlüssel Beispielwert
    S3_BUCKET digital-shadows-logs
    S3_PREFIX digital-shadows-searchlight/
    STATE_KEY digital-shadows-searchlight/state.json
    DS_API_KEY <your-6-character-api-key>
    DS_API_SECRET <your-32-character-api-secret>
    API_BASE https://api.searchlight.app/v1 (oder https://portal-digitalshadows.com/api/v1)
    DS_ACCOUNT_ID <your-account-id> (für die meisten Mandanten erforderlich)
    PAGE_SIZE 100
    MAX_PAGES 10
  8. Bleiben Sie nach dem Erstellen der Funktion auf der zugehörigen Seite oder öffnen Sie Lambda > Funktionen > Ihre Funktion.

  9. Wählen Sie den Tab Konfiguration aus.

  10. Klicken Sie im Bereich Allgemeine Konfiguration auf Bearbeiten.

  11. Ändern Sie Zeitlimit in 5 Minuten (300 Sekunden) und klicken Sie auf Speichern.

EventBridge-Zeitplan erstellen

  1. Gehen Sie zu Amazon EventBridge > Scheduler > Create schedule (Amazon EventBridge > Scheduler > Zeitplan erstellen).
  2. Geben Sie die folgenden Konfigurationsdetails an:
    • Wiederkehrender Zeitplan: Preis (1 hour).
    • Ziel: Ihre Lambda-Funktion digital-shadows-collector.
    • Name: digital-shadows-collector-1h.
  3. Klicken Sie auf Zeitplan erstellen.

(Optional) IAM-Nutzer mit Lesezugriff und Schlüssel für Google SecOps erstellen

  1. Rufen Sie die AWS-Konsole > IAM > Nutzer auf.
  2. Klicken Sie auf Add users (Nutzer hinzufügen).
  3. Geben Sie die folgenden Konfigurationsdetails an:
    • Nutzer: Geben Sie secops-reader ein.
    • Zugriffstyp: Wählen Sie Zugriffsschlüssel – programmatischer Zugriff aus.
  4. Klicken Sie auf Nutzer erstellen.
  5. Minimale Leseberechtigung (benutzerdefiniert) anhängen: Nutzer > secops-reader > Berechtigungen > Berechtigungen hinzufügen > Richtlinien direkt anhängen > Richtlinie erstellen.
  6. JSON:

    {
      "Version": "2012-10-17",
      "Statement": [
        {
          "Effect": "Allow",
          "Action": ["s3:GetObject"],
          "Resource": "arn:aws:s3:::digital-shadows-logs/*"
        },
        {
          "Effect": "Allow",
          "Action": ["s3:ListBucket"],
          "Resource": "arn:aws:s3:::digital-shadows-logs"
        }
      ]
    }
    
  7. Name = secops-reader-policy.

  8. Klicken Sie auf Richtlinie erstellen> suchen/auswählen> Weiter> Berechtigungen hinzufügen.

  9. Erstellen Sie einen Zugriffsschlüssel für secops-reader: Sicherheitsanmeldedaten > Zugriffsschlüssel.

  10. Klicken Sie auf Zugriffsschlüssel erstellen.

  11. Laden Sie die Datei .CSV herunter. Sie fügen diese Werte in den Feed ein.

Feed in Google SecOps konfigurieren, um Digital Shadows SearchLight-Logs aufzunehmen

  1. Rufen Sie die SIEM-Einstellungen > Feeds auf.
  2. Klicken Sie auf + Neuen Feed hinzufügen.
  3. Geben Sie im Feld Feedname einen Namen für den Feed ein, z. B. Digital Shadows SearchLight logs.
  4. Wählen Sie Amazon S3 V2 als Quelltyp aus.
  5. Wählen Sie Digital Shadows SearchLight als Logtyp aus.
  6. Klicken Sie auf Weiter.
  7. Geben Sie Werte für die folgenden Eingabeparameter an:
    • S3-URI: s3://digital-shadows-logs/digital-shadows-searchlight/
    • Optionen zum Löschen der Quelle: Wählen Sie die gewünschte Option zum Löschen aus.
    • Maximales Dateialter: Dateien einschließen, die in den letzten Tagen geändert wurden. Der Standardwert ist 180 Tage.
    • Zugriffsschlüssel-ID: Nutzerzugriffsschlüssel mit Zugriff auf den S3-Bucket.
    • Geheimer Zugriffsschlüssel: Der geheime Schlüssel des Nutzers mit Zugriff auf den S3-Bucket.
    • Asset-Namespace: Der Asset-Namespace.
    • Aufnahmelabels: Das Label, das auf die Ereignisse aus diesem Feed angewendet wird.
  8. Klicken Sie auf Weiter.
  9. Prüfen Sie die neue Feedkonfiguration auf dem Bildschirm Abschließen und klicken Sie dann auf Senden.

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