Recoger registros de la plataforma ZeroFox

Disponible en:

En este documento se explica cómo ingerir registros de la plataforma ZeroFox en Google Security Operations mediante Amazon S3.

Antes de empezar

Asegúrate de que cumples los siguientes requisitos previos:

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

Requisitos previos de ZeroFox

  1. Inicia sesión en la plataforma ZeroFox en https://cloud.zerofox.com.
  2. Vaya a Data Connectors > API Data Feeds.
    • URL directa (después de iniciar sesión): https://cloud.zerofox.com/data_connectors/api
    • Si no ves este elemento de menú, ponte en contacto con tu administrador de ZeroFox para obtener acceso.
  3. Haz clic en Generar token o en Crear token de acceso personal.
  4. Proporcione los siguientes detalles de configuración:
    • Nombre: introduce un nombre descriptivo (por ejemplo, Google SecOps S3 Ingestion).
    • Vencimiento: selecciona un periodo de rotación de acuerdo con la política de seguridad de tu organización.
    • Permisos/Feeds: selecciona permisos de lectura para Alerts, CTI feeds y otros tipos de datos que quieras exportar.
  5. Haz clic en Generar.
  6. Copia y guarda el token de acceso personal generado en un lugar seguro (no podrás volver a verlo).
  7. Guarda ZEROFOX_BASE_URL: https://api.zerofox.com (valor predeterminado para la mayoría de los clientes)

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, zerofox-platform-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 zerofox-platform-logs si has introducido otro nombre de contenedor):

    {
      "Version": "2012-10-17",
      "Statement": [
        {
          "Sid": "AllowPutObjects",
          "Effect": "Allow",
          "Action": "s3:PutObject",
          "Resource": "arn:aws:s3:::zerofox-platform-logs/*"
        },
        {
          "Sid": "AllowGetStateObject",
          "Effect": "Allow",
          "Action": "s3:GetObject",
          "Resource": "arn:aws:s3:::zerofox-platform-logs/zerofox/platform/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 ZeroFoxPlatformToS3Role 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 zerofox_platform_to_s3
    Tiempo de ejecución Python 3.13
    Arquitectura x86_64
    Rol de ejecución ZeroFoxPlatformToS3Role
  4. Una vez creada la función, abra la pestaña Código, elimine el stub y pegue el siguiente código (zerofox_platform_to_s3.py).

    #!/usr/bin/env python3
    # Lambda: Pull ZeroFox Platform data (alerts/incidents/logs) to S3 (no transform)
    
    import os, json, time, urllib.parse
    from urllib.request import Request, urlopen
    from urllib.error import HTTPError, URLError
    import boto3
    
    S3_BUCKET     = os.environ["S3_BUCKET"]
    S3_PREFIX     = os.environ.get("S3_PREFIX", "zerofox/platform/")
    STATE_KEY     = os.environ.get("STATE_KEY", "zerofox/platform/state.json")
    LOOKBACK_SEC  = int(os.environ.get("LOOKBACK_SECONDS", "3600"))
    PAGE_SIZE     = int(os.environ.get("PAGE_SIZE", "200"))
    MAX_PAGES     = int(os.environ.get("MAX_PAGES", "20"))
    HTTP_TIMEOUT  = int(os.environ.get("HTTP_TIMEOUT", "60"))
    HTTP_RETRIES  = int(os.environ.get("HTTP_RETRIES", "3"))
    URL_TEMPLATE  = os.environ.get("URL_TEMPLATE", "")
    AUTH_HEADER   = os.environ.get("AUTH_HEADER", "")  # e.g. "Authorization: Bearer <token>"
    
    ZEROFOX_BASE_URL = os.environ.get("ZEROFOX_BASE_URL", "https://api.zerofox.com")
    ZEROFOX_API_TOKEN = os.environ.get("ZEROFOX_API_TOKEN", "")
    
    s3 = boto3.client("s3")
    
    def _iso(ts: float) -> str:
        return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(ts))
    
    def _load_state() -> dict:
        try:
            obj = s3.get_object(Bucket=S3_BUCKET, Key=STATE_KEY)
            b = obj["Body"].read()
            return json.loads(b) if b else {}
        except Exception:
            return {"last_since": _iso(time.time() - LOOKBACK_SEC)}
    
    def _save_state(st: dict) -> None:
        s3.put_object(
            Bucket=S3_BUCKET, Key=STATE_KEY,
            Body=json.dumps(st, separators=(",", ":")).encode("utf-8"),
            ContentType="application/json",
        )
    
    def _headers() -> dict:
        hdrs = {"Accept": "application/json", "Content-Type": "application/json"}
        if AUTH_HEADER:
            try:
                k, v = AUTH_HEADER.split(":", 1)
                hdrs[k.strip()] = v.strip()
            except ValueError:
                hdrs["Authorization"] = AUTH_HEADER.strip()
        elif ZEROFOX_API_TOKEN:
            hdrs["Authorization"] = f"Bearer {ZEROFOX_API_TOKEN}"
        return hdrs
    
    def _http_get(url: str) -> dict:
        attempt = 0
        while True:
            try:
                req = Request(url, method="GET")
                for k, v in _headers().items():
                    req.add_header(k, v)
                with urlopen(req, timeout=HTTP_TIMEOUT) as r:
                    body = r.read()
                    try:
                        return json.loads(body.decode("utf-8"))
                    except json.JSONDecodeError:
                        return {"raw": body.decode("utf-8", errors="replace")}
            except HTTPError as e:
                if e.code in (429, 500, 502, 503, 504) and attempt < HTTP_RETRIES:
                    retry_after = int(e.headers.get("Retry-After", 1 + attempt))
                    time.sleep(max(1, retry_after))
                    attempt += 1
                    continue
                raise
            except URLError:
                if attempt < HTTP_RETRIES:
                    time.sleep(1 + attempt)
                    attempt += 1
                    continue
                raise
    
    def _put_json(obj: dict, label: str) -> str:
        ts = time.gmtime()
        key = f"{S3_PREFIX}/{time.strftime('%Y/%m/%d/%H%M%S', ts)}-zerofox-{label}.json"
        s3.put_object(
            Bucket=S3_BUCKET, Key=key,
            Body=json.dumps(obj, separators=(",", ":")).encode("utf-8"),
            ContentType="application/json",
        )
        return key
    
    def _extract_next_token(payload: dict):
        next_token = (payload.get("next") or payload.get("next_token") or 
                      payload.get("nextPageToken") or payload.get("next_page_token"))
        if isinstance(next_token, dict):
            return next_token.get("token") or next_token.get("cursor") or next_token.get("value")
        return next_token
    
    def _extract_items(payload: dict) -> list:
        for key in ("results", "data", "alerts", "items", "logs", "events"):
            if isinstance(payload.get(key), list):
                return payload[key]
        return []
    
    def _extract_newest_timestamp(items: list, current: str) -> str:
        newest = current
        for item in items:
            timestamp = (item.get("timestamp") or item.get("created_at") or 
                        item.get("last_modified") or item.get("event_time") or 
                        item.get("log_time") or item.get("updated_at"))
            if isinstance(timestamp, str) and timestamp > newest:
                newest = timestamp
        return newest
    
    def lambda_handler(event=None, context=None):
        st = _load_state()
        since = st.get("last_since") or _iso(time.time() - LOOKBACK_SEC)
    
        # Use URL_TEMPLATE if provided, otherwise construct default alerts endpoint
        if URL_TEMPLATE:
            base_url = URL_TEMPLATE.replace("{SINCE}", urllib.parse.quote(since))
        else:
            base_url = f"{ZEROFOX_BASE_URL}/v1/alerts?since={urllib.parse.quote(since)}"
    
        page_token = ""
        pages = 0
        total_items = 0
        newest_since = since
    
        while pages < MAX_PAGES:
            # Construct URL with pagination
            if URL_TEMPLATE:
                url = (base_url
                      .replace("{PAGE_TOKEN}", urllib.parse.quote(page_token))
                      .replace("{PAGE_SIZE}", str(PAGE_SIZE)))
            else:
                url = f"{base_url}&limit={PAGE_SIZE}"
                if page_token:
                    url += f"&page_token={urllib.parse.quote(page_token)}"
    
            payload = _http_get(url)
            _put_json(payload, f"page-{pages:05d}")
    
            items = _extract_items(payload)
            total_items += len(items)
            newest_since = _extract_newest_timestamp(items, newest_since)
    
            pages += 1
            next_token = _extract_next_token(payload)
            if not next_token:
                break
            page_token = str(next_token)
    
        if newest_since and newest_since != st.get("last_since"):
            st["last_since"] = newest_since
            _save_state(st)
    
        return {"ok": True, "pages": pages, "items": total_items, "since": since, "new_since": newest_since}
    
    if __name__ == "__main__":
        print(lambda_handler())
    
  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 zerofox-platform-logs
    S3_PREFIX zerofox/platform/
    STATE_KEY zerofox/platform/state.json
    ZEROFOX_BASE_URL https://api.zerofox.com
    ZEROFOX_API_TOKEN your-zerofox-personal-access-token
    LOOKBACK_SECONDS 3600
    PAGE_SIZE 200
    MAX_PAGES 20
    HTTP_TIMEOUT 60
    HTTP_RETRIES 3
    URL_TEMPLATE (opcional) Plantilla de URL personalizada con {SINCE}, {PAGE_TOKEN} y {PAGE_SIZE}
    AUTH_HEADER (opcional) Authorization: Bearer <token> para la autenticación personalizada
  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.

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 zerofox_platform_to_s3.
    • Nombre: zerofox-platform-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:::zerofox-platform-logs/*"
        },
        {
          "Effect": "Allow",
          "Action": ["s3:ListBucket"],
          "Resource": "arn:aws:s3:::zerofox-platform-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 la plataforma ZeroFox

  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, ZeroFox Platform Logs).
  4. Selecciona Amazon S3 V2 como Tipo de fuente.
  5. Seleccione ZeroFox Platform como Tipo de registro.
  6. Haz clic en Siguiente.
  7. Especifique los valores de los siguientes parámetros de entrada:
    • URI de S3: s3://zerofox-platform-logs/zerofox/platform/
    • 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.