Digital Shadows SearchLight-Logs erfassen
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)
- Melden Sie sich im Digital Shadows SearchLight-Portal an.
- Gehen Sie zu den Einstellungen> API-Anmeldedaten.
- Erstellen Sie einen neuen API-Client oder ein neues Schlüsselpaar.
- 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
oderhttps://portal-digitalshadows.com/api/v1
AWS S3-Bucket und IAM für Google SecOps konfigurieren
- Erstellen Sie einen Amazon S3-Bucket. Folgen Sie dazu der Anleitung unter Bucket erstellen.
- Speichern Sie den Namen und die Region des Buckets zur späteren Verwendung (z. B.
digital-shadows-logs
). - Erstellen Sie einen Nutzer gemäß dieser Anleitung: IAM-Nutzer erstellen.
- Wählen Sie den erstellten Nutzer aus.
- Wählen Sie den Tab Sicherheitsanmeldedaten aus.
- Klicken Sie im Abschnitt Zugriffsschlüssel auf Zugriffsschlüssel erstellen.
- Wählen Sie Drittanbieterdienst als Anwendungsfall aus.
- Klicken Sie auf Weiter.
- Optional: Fügen Sie ein Beschreibungstag hinzu.
- Klicken Sie auf Zugriffsschlüssel erstellen.
- 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.
- Klicken Sie auf Fertig.
- Wählen Sie den Tab Berechtigungen aus.
- Klicken Sie im Bereich Berechtigungsrichtlinien auf Berechtigungen hinzufügen.
- Wählen Sie Berechtigungen hinzufügen aus.
- Wählen Sie Richtlinien direkt anhängen aus.
- Suchen Sie nach der Richtlinie AmazonS3FullAccess.
- Wählen Sie die Richtlinie aus.
- Klicken Sie auf Weiter.
- Klicken Sie auf Berechtigungen hinzufügen.
IAM-Richtlinie und ‑Rolle für S3-Uploads konfigurieren
- Rufen Sie in der AWS-Konsole IAM > Richtlinien auf.
- Klicken Sie auf Richtlinie erstellen> Tab „JSON“.
- Kopieren Sie die folgende Richtlinie und fügen Sie sie ein.
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" } ] }
Klicken Sie auf Weiter > Richtlinie erstellen.
Rufen Sie IAM > Rollen > Rolle erstellen > AWS-Service > Lambda auf.
Hängen Sie die neu erstellte Richtlinie an.
Geben Sie der Rolle den Namen
digital-shadows-lambda-role
und klicken Sie auf Rolle erstellen.
Lambda-Funktion erstellen
- Rufen Sie in der AWS Console Lambda > Funktionen > Funktion erstellen auf.
- Klicken Sie auf Von Grund auf erstellen.
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
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}
Rufen Sie Konfiguration > Umgebungsvariablen auf.
Klicken Sie auf Bearbeiten> Neue Umgebungsvariable hinzufügen.
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
(oderhttps://portal-digitalshadows.com/api/v1
)DS_ACCOUNT_ID
<your-account-id>
(für die meisten Mandanten erforderlich)PAGE_SIZE
100
MAX_PAGES
10
Bleiben Sie nach dem Erstellen der Funktion auf der zugehörigen Seite oder öffnen Sie Lambda > Funktionen > Ihre Funktion.
Wählen Sie den Tab Konfiguration aus.
Klicken Sie im Bereich Allgemeine Konfiguration auf Bearbeiten.
Ändern Sie Zeitlimit in 5 Minuten (300 Sekunden) und klicken Sie auf Speichern.
EventBridge-Zeitplan erstellen
- Gehen Sie zu Amazon EventBridge > Scheduler > Create schedule (Amazon EventBridge > Scheduler > Zeitplan erstellen).
- Geben Sie die folgenden Konfigurationsdetails an:
- Wiederkehrender Zeitplan: Preis (
1 hour
). - Ziel: Ihre Lambda-Funktion
digital-shadows-collector
. - Name:
digital-shadows-collector-1h
.
- Wiederkehrender Zeitplan: Preis (
- Klicken Sie auf Zeitplan erstellen.
(Optional) IAM-Nutzer mit Lesezugriff und Schlüssel für Google SecOps erstellen
- Rufen Sie die AWS-Konsole > IAM > Nutzer auf.
- Klicken Sie auf Add users (Nutzer hinzufügen).
- Geben Sie die folgenden Konfigurationsdetails an:
- Nutzer: Geben Sie
secops-reader
ein. - Zugriffstyp: Wählen Sie Zugriffsschlüssel – programmatischer Zugriff aus.
- Nutzer: Geben Sie
- Klicken Sie auf Nutzer erstellen.
- Minimale Leseberechtigung (benutzerdefiniert) anhängen: Nutzer > secops-reader > Berechtigungen > Berechtigungen hinzufügen > Richtlinien direkt anhängen > Richtlinie erstellen.
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" } ] }
Name =
secops-reader-policy
.Klicken Sie auf Richtlinie erstellen> suchen/auswählen> Weiter> Berechtigungen hinzufügen.
Erstellen Sie einen Zugriffsschlüssel für
secops-reader
: Sicherheitsanmeldedaten > Zugriffsschlüssel.Klicken Sie auf Zugriffsschlüssel erstellen.
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
- Rufen Sie die SIEM-Einstellungen > Feeds auf.
- Klicken Sie auf + Neuen Feed hinzufügen.
- Geben Sie im Feld Feedname einen Namen für den Feed ein, z. B.
Digital Shadows SearchLight logs
. - Wählen Sie Amazon S3 V2 als Quelltyp aus.
- Wählen Sie Digital Shadows SearchLight als Logtyp aus.
- Klicken Sie auf Weiter.
- 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.
- S3-URI:
- Klicken Sie auf Weiter.
- 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