收集 ZeroFox 平台記錄

支援的國家/地區：

本文說明如何使用 Amazon S3，將 ZeroFox Platform 記錄檔擷取至 Google Security Operations。

事前準備

請確認您已完成下列事前準備事項：

  • Google SecOps 執行個體。
  • ZeroFox Platform 租戶的特殊存取權。
  • AWS 的特殊權限 (S3、Identity and Access Management (IAM)、Lambda、EventBridge)。

取得 ZeroFox 的必要條件

  1. 前往 https://cloud.zerofox.com 登入 ZeroFox Platform。
  2. 依序前往「Data Connectors」>「API Data Feeds」。
    • 直接網址 (登入後)：https://cloud.zerofox.com/data_connectors/api
    • 如果沒有看到這個選單項目，請與 ZeroFox 管理員聯絡，以取得存取權。
  3. 按一下「產生權杖」或「建立個人存取權杖」。
  4. 提供下列設定詳細資料：
    • 名稱：輸入描述性名稱 (例如 Google SecOps S3 Ingestion)。
    • 到期：根據貴機構的安全政策選取輪替週期。
    • 權限/動態消息：選取下列項目的讀取權限：Alerts、CTI feeds，以及要匯出的其他資料類型
  5. 點按「生成」。
  6. 複製並儲存產生的個人存取權杖 (您無法再次查看)。
  7. 儲存 ZEROFOX_BASE_URL：https://api.zerofox.com (大多數租戶的預設值)

為 Google SecOps 設定 AWS S3 值區和 IAM

  1. 按照這份使用者指南建立 Amazon S3 bucket：建立 bucket
  2. 儲存 bucket 的「名稱」和「區域」，以供日後參考 (例如 zerofox-platform-logs)。
  3. 請按照這份使用者指南建立使用者：建立 IAM 使用者。
  4. 選取建立的「使用者」。
  5. 選取「安全憑證」分頁標籤。
  6. 在「Access Keys」部分中，按一下「Create Access Key」。
  7. 選取「第三方服務」做為「用途」。
  8. 點選「下一步」。
  9. 選用：新增說明標記。
  10. 按一下「建立存取金鑰」。
  11. 按一下「下載 .CSV 檔案」，儲存「存取金鑰」和「私密存取金鑰」以供日後參考。
  12. 按一下 [完成]。
  13. 選取「權限」分頁標籤。
  14. 在「權限政策」部分中，按一下「新增權限」。
  15. 選取「新增權限」。
  16. 選取「直接附加政策」。
  17. 搜尋「AmazonS3FullAccess」AmazonS3FullAccess政策。
  18. 選取政策。
  19. 點選「下一步」。
  20. 按一下「Add permissions」。

設定 S3 上傳的身分與存取權管理政策和角色

  1. 在 AWS 控制台中，依序前往「IAM」>「Policies」，
  2. 按一下「建立政策」>「JSON」分頁。
  3. 複製並貼上下列政策。

  4. 政策 JSON (如果您輸入的 bucket 名稱不同，請替換 zerofox-platform-logs)：

    {
  "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. 依序點選「下一步」>「建立政策」。

  6. 依序前往「IAM」>「Roles」>「Create role」>「AWS service」>「Lambda」。

  7. 附加新建立的政策。

  8. 為角色命名 ZeroFoxPlatformToS3Role，然後按一下「建立角色」。

建立 Lambda 函式

  1. 在 AWS 控制台中，依序前往「Lambda」>「Functions」>「Create function」。
  2. 按一下「從頭開始撰寫」。

  3. 請提供下列設定詳細資料：

    設定 值
    名稱 zerofox_platform_to_s3
    執行階段 Python 3.13
    架構 x86_64
    執行角色 ZeroFoxPlatformToS3Role

  4. 建立函式後，開啟「程式碼」分頁，刪除存根並貼上以下程式碼 (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. 依序前往「設定」>「環境變數」。

  6. 依序點選「編輯」> 新增環境變數。

  7. 輸入下表提供的環境變數，並將範例值換成您的值。

    環境變數

    鍵 範例值
    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 (選用) 含有 {SINCE}、{PAGE_TOKEN}、{PAGE_SIZE} 的自訂網址範本
    AUTH_HEADER (選用) Authorization: Bearer <token> 用於自訂驗證

  8. 建立函式後，請留在函式頁面 (或依序開啟「Lambda」>「Functions」>「your-function」)。

  9. 選取「設定」分頁標籤。

  10. 在「一般設定」面板中，按一下「編輯」。

  11. 將「Timeout」(逾時間隔) 變更為「5 minutes (300 seconds)」(5 分鐘 (300 秒))，然後按一下「Save」(儲存)。

建立 EventBridge 排程

  1. 依序前往「Amazon EventBridge」>「Scheduler」>「Create schedule」。
  2. 提供下列設定詳細資料：
    • 週期性時間表：費率 (1 hour)。
    • 目標：您的 Lambda 函式 zerofox_platform_to_s3。
    • 名稱：zerofox-platform-1h。
  3. 按一下「建立時間表」。

(選用) 為 Google SecOps 建立唯讀 IAM 使用者和金鑰

  1. 前往 AWS 控制台 > IAM > 使用者。
  2. 點選 [Add users] (新增使用者)。
  3. 提供下列設定詳細資料：
    • 使用者：輸入 secops-reader。
    • 存取類型：選取「存取金鑰 - 程式輔助存取」。
  4. 按一下「建立使用者」。
  5. 附加最低讀取權限政策 (自訂)：依序選取「Users」(使用者) >「secops-reader」>「Permissions」(權限) >「Add permissions」(新增權限) >「Attach policies directly」(直接附加政策) >「Create policy」(建立政策)。

  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. 依序點選「建立政策」> 搜尋/選取 >「下一步」>「新增權限」。

  9. 為 secops-reader 建立存取金鑰：依序點選「安全憑證」>「存取金鑰」。

  10. 按一下「建立存取金鑰」。

  11. 下載 .CSV。(您會將這些值貼到動態饋給中)。

在 Google SecOps 中設定動態消息，擷取 ZeroFox Platform 記錄

  1. 依序前往「SIEM 設定」>「動態饋給」。
  2. 按一下「+ 新增動態消息」。
  3. 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 ZeroFox Platform Logs)。
  4. 選取「Amazon S3 V2」做為「來源類型」。
  5. 選取「ZeroFox Platform」做為「記錄類型」。
  6. 點選「下一步」。
  7. 指定下列輸入參數的值：
    • S3 URI：s3://zerofox-platform-logs/zerofox/platform/
    • 來源刪除選項：根據偏好設定選取刪除選項。
    • 檔案存在時間上限：包含在過去天數內修改的檔案。預設值為 180 天。
    • 存取金鑰 ID：具有 S3 值區存取權的使用者存取金鑰。
    • 存取密鑰：具有 S3 bucket 存取權的使用者私密金鑰。
    • 資產命名空間：資產命名空間。
    • 擷取標籤：套用至這個動態饋給事件的標籤。
  8. 點選「下一步」。
  9. 在「完成」畫面中檢查新的動態饋給設定，然後按一下「提交」。

還有其他問題嗎？向社群成員和 Google SecOps 專業人員尋求答案。