Recopilar registros de Digital Shadows SearchLight

Disponible en:

SecOps de Google SIEM

Nota: Esta función está cubierta por los Términos de las Ofertas Previas a la Disponibilidad General de los Términos Específicos del Servicio de Google Security Operations. Las funciones previas a la disponibilidad general pueden tener una asistencia limitada y los cambios que se produzcan en ellas podrían no ser compatibles con otras versiones previas a la disponibilidad general. Para obtener más información, consulta las directrices del servicio de asistencia técnica de Google SecOps y los Términos Específicos de los Servicios de Google SecOps.

En este documento se explica cómo ingerir registros de Digital Shadows SearchLight en Google Security Operations mediante Amazon S3. El analizador extrae datos de eventos de seguridad de los registros JSON. Inicializa los campos del modelo de datos unificado (UDM), analiza la carga útil JSON, asigna los campos relevantes al esquema del UDM, extrae entidades como el correo electrónico y el nombre de host mediante patrones grok y crea los objetos security_result y metadata en el evento del UDM.

Antes de empezar

Asegúrate de que cumples los siguientes requisitos previos:

• Una instancia de Google SecOps.
• Acceso privilegiado al arrendatario de Digital Shadows SearchLight.
• Acceso privilegiado a AWS (S3, Gestión de Identidades y Accesos [IAM], Lambda y EventBridge).

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

1. Inicia sesión en el 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 un lugar seguro los siguientes detalles:
   • Clave de API
   • Secreto de API
   • ID de cuenta
   • URL base de la API: https://api.searchlight.app/v1 o https://portal-digitalshadows.com/api/v1

Configurar un segmento de AWS S3 y IAM para Google SecOps

1. Crea un segmento de Amazon S3 siguiendo esta guía de usuario: Crear un segmento.
2. Guarda el nombre y la región del segmento para consultarlos más adelante (por ejemplo, digital-shadows-logs).
3. Crea un usuario siguiendo esta guía: Crear un usuario de gestión de identidades y accesos.
4. Selecciona el Usuario creado.
5. Selecciona la pestaña Credenciales de seguridad.
6. En la sección Claves de acceso, haz clic en Crear clave de acceso.
7. Selecciona Servicio de terceros en Caso práctico.
8. Haz clic en Siguiente.
9. Opcional: añade 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 futuras consultas.
12. Haz clic en Listo.
13. Selecciona la pestaña Permisos.
14. En la sección Políticas de permisos, haz clic en Añadir permisos.
15. Selecciona Añadir permisos.
16. Seleccione Adjuntar políticas directamente.
17. Busca la política AmazonS3FullAccess.
18. Selecciona la política.
19. Haz clic en Siguiente.
20. Haz clic en Añadir permisos.

Configurar la política y el rol de gestión de identidades y accesos para las subidas 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 (sustituye digital-shadows-logs si has introducido otro nombre de contenedor):

   {
     "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. Adjunte la política que acaba de crear.

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

Crear la función Lambda

1. En la consola de AWS, ve a Lambda > Funciones > Crear función.
2. Haz clic en Crear desde cero.

3. Proporciona los siguientes detalles de configuración:

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

4. Una vez creada la función, abra la pestaña Código, elimine el stub y pegue 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. Vaya a Configuración > Variables de entorno.

6. Haz clic en Editar > Añadir nueva variable de entorno.

7. Introduce las variables de entorno que se indican en la siguiente tabla y sustituye los valores de ejemplo por los tuyos.

   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 arrendatarios)
   PAGE_SIZE	100
   MAX_PAGES	10

8. Una vez creada la función, permanece en su página (o abre Lambda > Funciones > tu-función).

9. Seleccione 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 estricto, lo que significa que, si la función sigue ejecutándose cuando el temporizador llegue a los 5 minutos, AWS Lambda la finalizará. Si la ejecución es más corta, se completará con normalidad y solo se te cobrará por el tiempo de ejecución real.

Crear una programación de EventBridge

1. Ve a Amazon EventBridge > Scheduler > Create schedule (Amazon EventBridge > Programador > Crear programación).
2. Proporcione los siguientes detalles de configuración:
   • Programación periódica: Precio (1 hour).
   • Destino: tu función Lambda digital-shadows-collector.
   • Nombre: digital-shadows-collector-1h.
3. Haz clic en Crear programación.

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

1. Ve a Consola de AWS > IAM > Usuarios.
2. Haz clic en Add users (Añadir usuarios).
3. Proporcione los siguientes detalles de configuración:
   • Usuario: introduce secops-reader.
   • Tipo de acceso: selecciona Clave de acceso – Acceso programático.
4. Haz clic en Crear usuario.
5. Asigna una política de lectura mínima (personalizada): Usuarios > secops-reader > Permisos > Añadir permisos > Asignar 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 > Añadir 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 la .CSV. (Estos valores se pegarán en el feed).

Configurar un feed en Google SecOps para ingerir registros de Digital Shadows SearchLight

1. Ve a Configuración de SIEM > Feeds.
2. Haz clic en + Añadir nuevo feed.
3. En el campo Nombre del feed, introduce un nombre para el feed (por ejemplo, Digital Shadows SearchLight logs).
4. Selecciona Amazon S3 V2 como Tipo de fuente.
5. Selecciona Digital Shadows SearchLight como Tipo de registro.
6. Haz clic en Siguiente.
7. Especifique los valores de los siguientes parámetros de entrada:
   • URI de S3: s3://digital-shadows-logs/digital-shadows-searchlight/
   • Opciones de eliminación de la fuente: selecciona la opción de eliminación que prefieras.
   • Antigüedad máxima del archivo: incluye los archivos modificados en los últimos días. El valor predeterminado es 180 días.
   • ID de clave de acceso: clave de acceso de usuario con acceso al bucket de S3.
   • Clave de acceso secreta: clave secreta del usuario con acceso al bucket de S3.
   • Espacio de nombres de recursos: el espacio de nombres de recursos.
   • Etiquetas de ingestión: la etiqueta aplicada a los eventos de este feed.
8. Haz clic en Siguiente.
9. Revise la configuración de la nueva fuente en la pantalla Finalizar y, a continuación, haga clic en Enviar.

¿Necesitas más ayuda? Recibe respuestas de los miembros de la comunidad y de los profesionales de Google SecOps.