Raccogliere i log della piattaforma ZeroFox

Supportato in:

Questo documento spiega come importare i log della piattaforma ZeroFox in Google Security Operations utilizzando Amazon S3.

Prima di iniziare

Assicurati di soddisfare i seguenti prerequisiti:

  • Un'istanza Google SecOps.
  • Accesso privilegiato al tenant della piattaforma ZeroFox.
  • Accesso privilegiato ad AWS (S3, Identity and Access Management (IAM), Lambda, EventBridge).

Ottenere i prerequisiti di ZeroFox

  1. Accedi alla piattaforma ZeroFox all'indirizzo https://cloud.zerofox.com.
  2. Vai a Data Connectors > API Data Feeds.
    • URL diretto (dopo l'accesso): https://cloud.zerofox.com/data_connectors/api
    • Se non vedi questa voce di menu, contatta l'amministratore di ZeroFox per richiedere l'accesso.
  3. Fai clic su Genera token o Crea token di accesso personale.
  4. Fornisci i seguenti dettagli di configurazione:
    • Nome: inserisci un nome descrittivo (ad esempio, Google SecOps S3 Ingestion).
    • Scadenza: seleziona un periodo di rotazione in base alle norme di sicurezza della tua organizzazione.
    • Autorizzazioni/Feed: seleziona le autorizzazioni di lettura per: Alerts, CTI feeds e altri tipi di dati che vuoi esportare
  5. Fai clic su Genera.
  6. Copia e salva il token di accesso personale generato in una posizione sicura (non potrai visualizzarlo di nuovo).
  7. Salva ZEROFOX_BASE_URL: https://api.zerofox.com (valore predefinito per la maggior parte dei tenant)

Configura il bucket AWS S3 e IAM per Google SecOps

  1. Crea un bucket Amazon S3 seguendo questa guida utente: Creazione di un bucket
  2. Salva il nome e la regione del bucket per riferimento futuro (ad esempio, zerofox-platform-logs).
  3. Crea un utente seguendo questa guida utente: Creazione di un utente IAM.
  4. Seleziona l'utente creato.
  5. Seleziona la scheda Credenziali di sicurezza.
  6. Fai clic su Crea chiave di accesso nella sezione Chiavi di accesso.
  7. Seleziona Servizio di terze parti come Caso d'uso.
  8. Fai clic su Avanti.
  9. (Facoltativo) Aggiungi un tag di descrizione.
  10. Fai clic su Crea chiave di accesso.
  11. Fai clic su Scarica file .CSV per salvare la chiave di accesso e la chiave di accesso segreta per riferimento futuro.
  12. Fai clic su Fine.
  13. Seleziona la scheda Autorizzazioni.
  14. Fai clic su Aggiungi autorizzazioni nella sezione Criteri per le autorizzazioni.
  15. Seleziona Aggiungi autorizzazioni.
  16. Seleziona Allega direttamente i criteri.
  17. Cerca i criteri AmazonS3FullAccess.
  18. Seleziona la policy.
  19. Fai clic su Avanti.
  20. Fai clic su Aggiungi autorizzazioni.

Configura il ruolo e il criterio IAM per i caricamenti S3

  1. Nella console AWS, vai a IAM > Policy.
  2. Fai clic su Crea criterio > scheda JSON.
  3. Copia e incolla i seguenti criteri.
  4. JSON delle policy (sostituisci zerofox-platform-logs se hai inserito un nome del bucket diverso):

    {
      "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. Fai clic su Avanti > Crea criterio.

  6. Vai a IAM > Ruoli > Crea ruolo > Servizio AWS > Lambda.

  7. Allega il criterio appena creato.

  8. Assegna al ruolo il nome ZeroFoxPlatformToS3Role e fai clic su Crea ruolo.

Crea la funzione Lambda

  1. Nella console AWS, vai a Lambda > Funzioni > Crea funzione.
  2. Fai clic su Crea autore da zero.
  3. Fornisci i seguenti dettagli di configurazione:

    Impostazione Valore
    Nome zerofox_platform_to_s3
    Tempo di esecuzione Python 3.13
    Architettura x86_64
    Ruolo di esecuzione ZeroFoxPlatformToS3Role
  4. Dopo aver creato la funzione, apri la scheda Codice, elimina lo stub e incolla il seguente codice (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. Vai a Configurazione > Variabili di ambiente.

  6. Fai clic su Modifica > Aggiungi nuova variabile di ambiente.

  7. Inserisci le variabili di ambiente fornite nella tabella seguente, sostituendo i valori di esempio con i tuoi valori.

    Variabili di ambiente

    Chiave Valore di esempio
    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 (facoltativo) Modello di URL personalizzato con {SINCE}, {PAGE_TOKEN}, {PAGE_SIZE}
    AUTH_HEADER (facoltativo) Authorization: Bearer <token> per l'autenticazione personalizzata
  8. Dopo aver creato la funzione, rimani sulla relativa pagina (o apri Lambda > Funzioni > la tua funzione).

  9. Seleziona la scheda Configurazione.

  10. Nel riquadro Configurazione generale, fai clic su Modifica.

  11. Modifica Timeout impostando 5 minuti (300 secondi) e fai clic su Salva.

Creare una pianificazione EventBridge

  1. Vai a Amazon EventBridge > Scheduler > Crea pianificazione.
  2. Fornisci i seguenti dettagli di configurazione:
    • Programma ricorrente: Tariffa (1 hour).
    • Target: la tua funzione Lambda zerofox_platform_to_s3.
    • Nome: zerofox-platform-1h
  3. Fai clic su Crea pianificazione.

(Facoltativo) Crea chiavi e utenti IAM di sola lettura per Google SecOps

  1. Vai alla console AWS > IAM > Utenti.
  2. Fai clic su Add users (Aggiungi utenti).
  3. Fornisci i seguenti dettagli di configurazione:
    • Utente: inserisci secops-reader.
    • Tipo di accesso: seleziona Chiave di accesso - Accesso programmatico.
  4. Fai clic su Crea utente.
  5. Collega la criterio per la lettura minima (personalizzata): Utenti > secops-reader > Autorizzazioni > Aggiungi autorizzazioni > Collega le norme direttamente > Crea norma.
  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. Name = secops-reader-policy.

  8. Fai clic su Crea criterio > cerca/seleziona > Avanti > Aggiungi autorizzazioni.

  9. Crea la chiave di accesso per secops-reader: Credenziali di sicurezza > Chiavi di accesso.

  10. Fai clic su Crea chiave di accesso.

  11. Scarica il .CSV. Incollerai questi valori nel feed.

Configura un feed in Google SecOps per importare i log della piattaforma ZeroFox

  1. Vai a Impostazioni SIEM > Feed.
  2. Fai clic su + Aggiungi nuovo feed.
  3. Nel campo Nome feed, inserisci un nome per il feed (ad esempio, ZeroFox Platform Logs).
  4. Seleziona Amazon S3 V2 come Tipo di origine.
  5. Seleziona ZeroFox Platform come Tipo di log.
  6. Fai clic su Avanti.
  7. Specifica i valori per i seguenti parametri di input:
    • URI S3: s3://zerofox-platform-logs/zerofox/platform/
    • Opzioni di eliminazione dell'origine: seleziona l'opzione di eliminazione in base alle tue preferenze.
    • Età massima del file: includi i file modificati nell'ultimo numero di giorni. Il valore predefinito è 180 giorni.
    • ID chiave di accesso: chiave di accesso utente con accesso al bucket S3.
    • Chiave di accesso segreta: chiave segreta dell'utente con accesso al bucket S3.
    • Spazio dei nomi dell'asset: lo spazio dei nomi dell'asset.
    • Etichette di importazione: l'etichetta applicata agli eventi di questo feed.
  8. Fai clic su Avanti.
  9. Controlla la nuova configurazione del feed nella schermata Finalizza e poi fai clic su Invia.

Hai bisogno di ulteriore assistenza? Ricevi risposte dai membri della community e dai professionisti di Google SecOps.