Recopila registros de Digital Shadows SearchLight

Compatible con:

Google SecOps SIEM

Nota: Esta función está sujeta a las Condiciones de las Ofertas de la fase previa a la DG de las Condiciones Específicas del Servicio de Operaciones de Seguridad de Google. Las características de la fase previa a la DG pueden tener asistencia limitada, y es posible que los cambios en las características de la fase previa a la DG no sean compatibles con otras versiones de la fase previa a la DG. Para obtener más información, consulta los Lineamientos de los Servicios de Asistencia Técnica de Google SecOps y las Condiciones Específicas del Servicio de Google SecOps.

En este documento, se explica cómo transferir registros de Digital Shadows SearchLight a Google Security Operations con Amazon S3. El analizador extrae los datos de eventos de seguridad de los registros JSON. Inicializa los campos del Modelo de datos unificado (UDM), analiza la carga útil de JSON, asigna los campos pertinentes al esquema del UDM, extrae entidades como el correo electrónico y el nombre de host con patrones de Grok, y construye los objetos security_result y metadata dentro del evento del UDM.

Antes de comenzar

Asegúrate de cumplir con los siguientes requisitos previos:

• Es una instancia de Google SecOps.
• Acceso con privilegios al usuario de Digital Shadows SearchLight
• Acceso con privilegios a AWS (S3, Identity and Access Management [IAM], Lambda, EventBridge).

Recopila los requisitos previos de Digital Shadows SearchLight (IDs, claves de API, IDs de organización, tokens).

1. Accede al portal de Digital Shadows SearchLight.
2. Ve a Configuración > Credenciales de API.
3. Crea un nuevo cliente de API o un par de claves.
4. Copia y guarda en una ubicación segura los siguientes detalles:
   • Clave de API
   • Secreto de API
   • ID de la cuenta
   • URL base de la API: https://api.searchlight.app/v1 o https://portal-digitalshadows.com/api/v1

Configura el bucket de AWS S3 y el IAM para Google SecOps

1. Crea un bucket de Amazon S3 siguiendo esta guía del usuario: Crea un bucket
2. Guarda el Nombre y la Región del bucket para futuras referencias (por ejemplo, digital-shadows-logs).
3. Crea un usuario siguiendo esta guía del usuario: Cómo crear un usuario de IAM.
4. Selecciona el usuario creado.
5. Selecciona la pestaña Credenciales de seguridad.
6. Haz clic en Crear clave de acceso en la sección Claves de acceso.
7. Selecciona Servicio de terceros como Caso de uso.
8. Haz clic en Siguiente.
9. Opcional: Agrega una etiqueta de descripción.
10. Haz clic en Crear clave de acceso.
11. Haz clic en Descargar archivo .CSV para guardar la clave de acceso y la clave de acceso secreta para consultarlas en el futuro.
12. Haz clic en Listo.
13. Selecciona la pestaña Permisos.
14. Haz clic en Agregar permisos en la sección Políticas de permisos.
15. Selecciona Agregar permisos.
16. Selecciona Adjuntar políticas directamente.
17. Busca la política AmazonS3FullAccess.
18. Selecciona la política.
19. Haz clic en Siguiente.
20. Haz clic en Agregar permisos.

Configura la política y el rol de IAM para las cargas de S3

1. En la consola de AWS, ve a IAM > Políticas.
2. Haz clic en Crear política > pestaña JSON.
3. Copia y pega la siguiente política.

4. JSON de la política (reemplaza digital-shadows-logs si ingresaste un nombre de bucket diferente):

   {
     "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. Haz clic en Siguiente > Crear política.

6. Ve a IAM > Roles > Crear rol > Servicio de AWS > Lambda.

7. Adjunta la política recién creada.

8. Asigna el nombre digital-shadows-lambda-role al rol y haz clic en Crear rol.

Crea la función Lambda

1. En la consola de AWS, ve a Lambda > Functions > Create function.
2. Haz clic en Author from scratch.

3. Proporciona los siguientes detalles de configuración:

   Configuración	Valor
   Nombre	digital-shadows-collector
   Tiempo de ejecución	Python 3.13
   Arquitectura	x86_64
   Rol de ejecución	digital-shadows-lambda-role

4. Después de crear la función, abre la pestaña Code, borra el código auxiliar y pega el siguiente código (digital-shadows-collector.py).

   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. Ve a Configuration > Environment variables.

6. Haz clic en Editar > Agregar nueva variable de entorno.

7. Ingresa las variables de entorno que se proporcionan en la siguiente tabla y reemplaza los valores de ejemplo por tus valores.

   Variables de entorno

   Clave	Valor de ejemplo
   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 (o https://portal-digitalshadows.com/api/v1)
   DS_ACCOUNT_ID	<your-account-id> (obligatorio para la mayoría de los inquilinos)
   PAGE_SIZE	100
   MAX_PAGES	10

8. Después de crear la función, permanece en su página (o abre Lambda > Funciones > tu-función).

9. Selecciona la pestaña Configuración.

10. En el panel Configuración general, haz clic en Editar.

11. Cambia Tiempo de espera a 5 minutos (300 segundos) y haz clic en Guardar.

    Nota: El tiempo de espera es un límite superior fijo, lo que significa que, si la función sigue en ejecución cuando el temporizador llega a los 5 minutos, AWS Lambda la finalizará. Cualquier ejecución más corta simplemente finaliza de forma normal, y solo se te factura el tiempo de ejecución real.

Crea una programación de EventBridge

1. Ve a Amazon EventBridge > Scheduler > Create schedule.
2. Proporciona los siguientes detalles de configuración:
   • Programación recurrente: Frecuencia (1 hour)
   • Destino: Tu función Lambda digital-shadows-collector.
   • Nombre: digital-shadows-collector-1h.
3. Haz clic en Crear programación.

(Opcional) Crea un usuario y claves de IAM de solo lectura para Google SecOps

1. Ve a Consola de AWS > IAM > Usuarios.
2. Haz clic en Agregar usuarios.
3. Proporciona los siguientes detalles de configuración:
   • Usuario: Ingresa secops-reader.
   • Tipo de acceso: Selecciona Clave de acceso: Acceso programático.
4. Haz clic en Crear usuario.
5. Adjunta una política de lectura mínima (personalizada): Usuarios > secops-reader > Permisos > Agregar permisos > Adjuntar políticas directamente > Crear política.

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. Nombre = secops-reader-policy.

8. Haz clic en Crear política > busca o selecciona > Siguiente > Agregar permisos.

9. Crea una clave de acceso para secops-reader: Credenciales de seguridad > Claves de acceso.

10. Haz clic en Crear clave de acceso.

11. Descarga el .CSV. (Pegarás estos valores en el feed).

Configura un feed en Google SecOps para transferir registros de Digital Shadows SearchLight

1. Ve a Configuración de SIEM > Feeds.
2. Haz clic en + Agregar feed nuevo.
3. En el campo Nombre del feed, ingresa un nombre para el feed (por ejemplo, Digital Shadows SearchLight logs).
4. Selecciona Amazon S3 V2 como el Tipo de fuente.
5. Selecciona Digital Shadows SearchLight como el Tipo de registro.
6. Haz clic en Siguiente.
7. Especifica valores para los siguientes parámetros de entrada:
   • URI de S3: s3://digital-shadows-logs/digital-shadows-searchlight/
   • Opciones de borrado de la fuente: Selecciona la opción de borrado según tu preferencia.
   • Antigüedad máxima del archivo: Incluye los archivos modificados en la cantidad de días especificada. El valor predeterminado es de 180 días.
   • ID de clave de acceso: Clave de acceso del usuario con acceso al bucket de S3.
   • Clave de acceso secreta: Clave secreta del usuario con acceso al bucket de S3.
   • Espacio de nombres del recurso: Es el espacio de nombres del recurso.
   • Etiquetas de transmisión: Es la etiqueta que se aplica a los eventos de este feed.
8. Haz clic en Siguiente.
9. Revisa la nueva configuración del feed en la pantalla Finalizar y, luego, haz clic en Enviar.

¿Necesitas más ayuda? Obtén respuestas de miembros de la comunidad y profesionales de Google SecOps.