Se usó la API de Cloud Translation para traducir esta página.

Recopila registros de la plataforma de ZeroFox

Compatible con:

Google SecOps SIEM

En este documento, se explica cómo transferir registros de la plataforma de ZeroFox a las Operaciones de seguridad de Google con Amazon S3.

Antes de comenzar

Asegúrate de cumplir con los siguientes requisitos previos:

Es una instancia de Google SecOps.
Acceso privilegiado al arrendatario de la plataforma de ZeroFox
Acceso con privilegios a AWS (S3, Identity and Access Management [IAM], Lambda, EventBridge).

Obtén los requisitos previos de ZeroFox

Accede a la plataforma de ZeroFox en https://cloud.zerofox.com.
Ve a Conectores de datos > Feeds de datos de la API.
- URL directa (después del acceso): https://cloud.zerofox.com/data_connectors/api
- Si no ves este elemento de menú, comunícate con tu administrador de ZeroFox para obtener acceso.
Haz clic en Generar token o Crear token de acceso personal.
Proporciona los siguientes detalles de configuración:
- Nombre: Ingresa un nombre descriptivo (por ejemplo, Google SecOps S3 Ingestion).
- Vencimiento: Selecciona un período de rotación según la política de seguridad de tu organización.
- Permissions/Feeds: Selecciona permisos de lectura para Alerts, CTI feeds y otros tipos de datos que desees exportar.
Haz clic en Generar.
Copia y guarda el token de acceso personal generado en una ubicación segura (no podrás volver a verlo).
Guarda ZEROFOX_BASE_URL: https://api.zerofox.com (predeterminado para la mayoría de los arrendatarios)

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

Crea un bucket de Amazon S3 siguiendo esta guía del usuario: Crea un bucket
Guarda el Nombre y la Región del bucket para futuras referencias (por ejemplo, zerofox-platform-logs).
Crea un usuario siguiendo esta guía del usuario: Cómo crear un usuario de IAM.
Selecciona el usuario creado.
Selecciona la pestaña Credenciales de seguridad.
Haz clic en Crear clave de acceso en la sección Claves de acceso.
Selecciona Servicio de terceros como Caso de uso.
Haz clic en Siguiente.
Opcional: Agrega una etiqueta de descripción.
Haz clic en Crear clave de acceso.
Haz clic en Descargar archivo .CSV para guardar la clave de acceso y la clave de acceso secreta para consultarlas en el futuro.
Haz clic en Listo.
Selecciona la pestaña Permisos.
Haz clic en Agregar permisos en la sección Políticas de permisos.
Selecciona Agregar permisos.
Selecciona Adjuntar políticas directamente.
Busca la política AmazonS3FullAccess.
Selecciona la política.
Haz clic en Siguiente.
Haz clic en Agregar permisos.

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

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

JSON de la política (reemplaza zerofox-platform-logs si ingresaste un nombre de bucket diferente):

{
  "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"
    }
  ]
}

Haz clic en Siguiente > Crear política.
Ve a IAM > Roles > Crear rol > Servicio de AWS > Lambda.
Adjunta la política recién creada.
Asigna el nombre ZeroFoxPlatformToS3Role al rol y haz clic en Crear rol.

Crea la función Lambda

En la consola de AWS, ve a Lambda > Functions > Create function.
Haz clic en Author from scratch.
Proporciona los siguientes detalles de configuración:

Configuración Valor

Nombre zerofox_platform_to_s3

Tiempo de ejecución Python 3.13

Arquitectura x86_64

Rol de ejecución ZeroFoxPlatformToS3Role

Configuración	Valor
Nombre	`zerofox_platform_to_s3`
Tiempo de ejecución	Python 3.13
Arquitectura	x86_64
Rol de ejecución	`ZeroFoxPlatformToS3Role`

Después de crear la función, abre la pestaña Code, borra el código auxiliar y pega 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())

Ve a Configuration > Environment variables.
Haz clic en Editar > Agregar nueva variable de entorno.

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`	`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

Después de crear la función, permanece en su página (o abre Lambda > Funciones > tu-función).
Selecciona la pestaña Configuración.
En el panel Configuración general, haz clic en Editar.
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

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

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

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

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"
    }
  ]
}

Nombre = secops-reader-policy.
Haz clic en Crear política > busca o selecciona > Siguiente > Agregar permisos.
Crea una clave de acceso para secops-reader: Credenciales de seguridad > Claves de acceso.
Haz clic en Crear clave de acceso.
Descarga el .CSV. (Pegarás estos valores en el feed).

Configura un feed en Google SecOps para transferir registros de la plataforma de ZeroFox

Ve a Configuración de SIEM > Feeds.
Haz clic en + Agregar feed nuevo.
En el campo Nombre del feed, ingresa un nombre para el feed (por ejemplo, ZeroFox Platform Logs).
Selecciona Amazon S3 V2 como el Tipo de fuente.
Selecciona ZeroFox Platform como el Tipo de registro.
Haz clic en Siguiente.
Especifica valores para los siguientes parámetros de entrada:
- URI de S3: s3://zerofox-platform-logs/zerofox/platform/
- 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.
Haz clic en Siguiente.
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.