本頁面由 Cloud Translation API 翻譯而成。

收集 Snyk 群組層級的稽核記錄和問題記錄

支援的國家/地區：

Google SecOps SIEM

本指南說明如何使用 Amazon S3，將 Snyk 群組層級的稽核和問題記錄擷取至 Google Security Operations。

事前準備

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

Google SecOps 執行個體
Snyk 群組的特殊存取權 (具有讀取權限的 API 權杖；群組 ID)
AWS (S3、IAM、Lambda、EventBridge) 的具備權限存取權

取得 Snyk 群組 ID 和 API 權杖

在 Snyk 使用者介面中，依序前往「Account settings」>「API token」，然後產生 API 權杖。
複製並將權杖儲存在安全位置，稍後會做為 SNYK_TOKEN 使用。
切換至群組，然後開啟群組設定。
從網址 (https://app.snyk.io/group/<GROUP_ID>/...) 複製並儲存「群組 ID」，以供日後做為 GROUP_ID 使用。
基本 API 端點：https://api.snyk.io (如需覆寫，請使用 API_BASE)。
將「群組管理員」角色指派給擁有權杖的使用者。(使用者必須能夠查看群組稽核記錄和群組問題)。

為 Google SecOps 設定 AWS S3 值區和 IAM

按照這份使用者指南建立 Amazon S3 值區：建立值區
儲存 bucket 的「名稱」和「區域」，以供日後參考 (例如 snyk-group-logs)。
按照這份使用者指南建立使用者：建立 IAM 使用者。
選取建立的「使用者」。
選取「安全憑證」分頁標籤。
在「Access Keys」部分中，按一下「Create Access Key」。
選取「第三方服務」做為「用途」。
點選「下一步」。
選用：新增說明標記。
按一下「建立存取金鑰」。
按一下「下載 CSV 檔案」，儲存「存取金鑰」和「私密存取金鑰」以供日後使用。
按一下 [完成]。
選取 [權限] 分頁標籤。
在「Permissions policies」(權限政策) 區段中，按一下「Add permissions」(新增權限)。
選取「新增權限」。
選取「直接附加政策」
搜尋並選取 AmazonS3FullAccess 政策。
點選「下一步」。
按一下「Add permissions」。

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

在 AWS 控制台中，依序前往「IAM」>「Policies」>「Create policy」>「JSON」分頁標籤。

輸入下列政策 (包括值區下所有物件的寫入權限和 Lambda 使用的狀態檔案讀取權限)：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "PutAllSnykGroupObjects",
      "Effect": "Allow",
      "Action": ["s3:PutObject", "s3:GetObject"],
      "Resource": "arn:aws:s3:::snyk-group-logs/*"
    }
  ]
}

如果您輸入的值區名稱不同，請替換 snyk-group-logs。

依序點選「Next」>「Create policy」。
依序前往「IAM」>「Roles」>「Create role」>「AWS service」>「Lambda」。
附加新建立的政策。
為角色命名 WriteSnykGroupToS3Role，然後按一下「建立角色」。

建立 Lambda 函式

在 AWS 控制台中，依序前往「Lambda」>「Functions」>「Create function」。
按一下「從頭開始撰寫」。
請提供下列設定詳細資料：

設定	值
名稱	`snyk_group_audit_issues_to_s3`
執行階段	Python 3.13
架構	x86_64
執行角色	`WriteSnykGroupToS3Role`

建立函式後，開啟「程式碼」分頁，刪除存根並輸入下列程式碼 (snyk_group_audit_issues_to_s3.py)：

#!/usr/bin/env python3
# Lambda: Pull Snyk Group-level Audit Logs + Issues to S3 (no transform)

import os
import json
import time
import urllib.parse
from urllib.request import Request, urlopen
from urllib.parse import urlparse, parse_qs
from urllib.error import HTTPError
import boto3

API_BASE = os.environ.get("API_BASE", "https://api.snyk.io").rstrip("/")
SNYK_TOKEN = os.environ["SNYK_TOKEN"].strip()
GROUP_ID = os.environ["GROUP_ID"].strip()

BUCKET = os.environ["S3_BUCKET"].strip()
PREFIX = os.environ.get("S3_PREFIX", "snyk/group/").strip()
STATE_KEY = os.environ.get("STATE_KEY", "snyk/group/state.json").strip()

# Page sizes & limits
AUDIT_SIZE = int(os.environ.get("AUDIT_PAGE_SIZE", "100"))       # audit uses 'size' (max 100)
ISSUES_LIMIT = int(os.environ.get("ISSUES_PAGE_LIMIT", "200"))   # issues uses 'limit'
MAX_PAGES = int(os.environ.get("MAX_PAGES", "20"))

# API versions (Snyk REST requires a 'version' param)
AUDIT_API_VERSION = os.environ.get("SNYK_AUDIT_API_VERSION", "2021-06-04").strip()
ISSUES_API_VERSION = os.environ.get("SNYK_ISSUES_API_VERSION", "2024-10-15").strip()

# First-run lookback for audit to avoid huge backfills
LOOKBACK_SECONDS = int(os.environ.get("LOOKBACK_SECONDS", "3600"))

HDRS = {
    "Authorization": f"token {SNYK_TOKEN}",
    "Accept": "application/vnd.api+json",
}

s3 = boto3.client("s3")

def _get_state() -> dict:
    try:
        obj = s3.get_object(Bucket=BUCKET, Key=STATE_KEY)
        return json.loads(obj["Body"].read() or b"{}")
    except Exception:
        return {}

def _put_state(state: dict):
    s3.put_object(
        Bucket=BUCKET,
        Key=STATE_KEY,
        Body=json.dumps(state, separators=(",", ":")).encode("utf-8"),
        ContentType="application/json",
    )

def _iso(ts: float) -> str:
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(ts))

def _http_get(url: str) -> dict:
    req = Request(url, method="GET", headers=HDRS)
    try:
        with urlopen(req, timeout=60) as r:
            return json.loads(r.read().decode("utf-8"))
    except HTTPError as e:
        if e.code in (429, 500, 502, 503, 504):
            delay = int(e.headers.get("Retry-After", "1"))
            time.sleep(max(1, delay))
            with urlopen(req, timeout=60) as r2:
                return json.loads(r2.read().decode("utf-8"))
        raise

def _write_page(kind: str, payload: dict) -> str:
    ts = time.gmtime()
    key = f"{PREFIX.rstrip('/')}/{time.strftime('%Y/%m/%d/%H%M%S', ts)}-snyk-{kind}.json"
    s3.put_object(
        Bucket=BUCKET,
        Key=key,
        Body=json.dumps(payload, separators=(",", ":")).encode("utf-8"),
        ContentType="application/json",
    )
    return key

def _next_href(links: dict | None) -> str | None:
    if not links:
        return None
    nxt = links.get("next")
    if not nxt:
        return None
    if isinstance(nxt, str):
        return nxt
    if isinstance(nxt, dict):
        return nxt.get("href")
    return None

# -------- Audit Logs --------

def pull_audit_logs(state: dict) -> dict:
    cursor = state.get("audit_cursor")
    pages = 0
    total = 0

    base = f"{API_BASE}/rest/groups/{GROUP_ID}/audit_logs/search"
    params: dict[str, object] = {"version": AUDIT_API_VERSION, "size": AUDIT_SIZE}

    if cursor:
        params["cursor"] = cursor
    else:
        now = time.time()
        params["from"] = _iso(now - LOOKBACK_SECONDS)
        params["to"] = _iso(now)

    while pages < MAX_PAGES:
        url = f"{base}?{urllib.parse.urlencode(params, doseq=True)}"
        payload = _http_get(url)
        _write_page("audit", payload)

        data_items = (payload.get("data") or {}).get("items") or []
        if isinstance(data_items, list):
            total += len(data_items)

        nxt = _next_href(payload.get("links"))
        if not nxt:
            break
        q = parse_qs(urlparse(nxt).query)
        cur = (q.get("cursor") or [None])[0]
        if not cur:
            break

        params = {"version": AUDIT_API_VERSION, "size": AUDIT_SIZE, "cursor": cur}
        state["audit_cursor"] = cur
        pages += 1

    return {"pages": pages + 1 if total else pages, "items": total, "cursor": state.get("audit_cursor")}

# -------- Issues --------

def pull_issues(state: dict) -> dict:
    cursor = state.get("issues_cursor")  # stores 'starting_after'
    pages = 0
    total = 0

    base = f"{API_BASE}/rest/groups/{GROUP_ID}/issues"
    params: dict[str, object] = {"version": ISSUES_API_VERSION, "limit": ISSUES_LIMIT}
    if cursor:
        params["starting_after"] = cursor

    while pages < MAX_PAGES:
        url = f"{base}?{urllib.parse.urlencode(params, doseq=True)}"
        payload = _http_get(url)
        _write_page("issues", payload)

        data_items = payload.get("data") or []
        if isinstance(data_items, list):
            total += len(data_items)

        nxt = _next_href(payload.get("links"))
        if not nxt:
            break
        q = parse_qs(urlparse(nxt).query)
        cur = (q.get("starting_after") or [None])[0]
        if not cur:
            break

        params = {"version": ISSUES_API_VERSION, "limit": ISSUES_LIMIT, "starting_after": cur}
        state["issues_cursor"] = cur
        pages += 1

    return {"pages": pages + 1 if total else pages, "items": total, "cursor": state.get("issues_cursor")}

def lambda_handler(event=None, context=None):
    state = _get_state()
    audit_res = pull_audit_logs(state)
    issues_res = pull_issues(state)
    _put_state(state)
    return {"ok": True, "audit": audit_res, "issues": issues_res}

if __name__ == "__main__":
    print(lambda_handler())

依序前往「Configuration」>「Environment variables」>「Edit」>「Add new environment variable」。

輸入下列環境變數，並將換成您的值：

鍵	範例
`S3_BUCKET`	`snyk-group-logs`
`S3_PREFIX`	`snyk/group/`
`STATE_KEY`	`snyk/group/state.json`
`SNYK_TOKEN`	`xxxxxxxx-xxxx-xxxx-xxxx-xxxx`
`GROUP_ID`	`<group_uuid>`
`API_BASE`	`https://api.snyk.io`
`SNYK_AUDIT_API_VERSION`	`2021-06-04`
`SNYK_ISSUES_API_VERSION`	`2024-10-15`
`AUDIT_PAGE_SIZE`	`100`
`ISSUES_PAGE_LIMIT`	`200`
`MAX_PAGES`	`20`
`LOOKBACK_SECONDS`	`3600`

建立函式後，請留在該函式的頁面 (或依序開啟「Lambda」>「Functions」> <your-function>)。
選取「設定」分頁標籤。
在「一般設定」面板中，按一下「編輯」。
將「Timeout」(逾時間隔) 變更為「5 minutes (300 seconds)」(5 分鐘 (300 秒))，然後按一下「Save」(儲存)。

注意： 逾時是上限，也就是說，如果函式在計時器達到 5 分鐘時仍在執行，AWS Lambda 就會終止函式。如果執行時間較短，系統會正常完成作業，並只收取實際執行時間的費用。

建立 EventBridge 排程

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

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

在 AWS 控制台中，依序前往「IAM」>「Users」>「Add users」。
點選 [Add users] (新增使用者)。
提供下列設定詳細資料：
- 使用者：secops-reader。
- 存取類型：存取金鑰 - 程式輔助存取。
按一下「建立使用者」。
附加最低讀取權限政策 (自訂)：依序選取「使用者」>「secops-reader」>「權限」>「新增權限」>「直接附加政策」>「建立政策」。

在 JSON 編輯器中輸入下列政策：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["s3:GetObject"],
      "Resource": "arn:aws:s3:::snyk-group-logs/*"
    },
    {
      "Effect": "Allow",
      "Action": ["s3:ListBucket"],
      "Resource": "arn:aws:s3:::snyk-group-logs"
    }
  ]
}

將名稱設為 secops-reader-policy。
依序前往「建立政策」> 搜尋/選取 >「下一步」>「新增權限」。
依序前往「安全憑證」>「存取金鑰」>「建立存取金鑰」。
下載 CSV (這些值會輸入至動態饋給)。

在 Google SecOps 中設定動態饋給，擷取 Snyk 群組層級的稽核和問題記錄

依序前往「SIEM 設定」>「動態饋給」。
按一下「+ 新增動態消息」。
在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Snyk Group Audit/Issues)。
選取「Amazon S3 V2」做為「來源類型」。
選取「Snyk Group level audit/issues logs」(Snyk 群組層級稽核/問題記錄) 做為「Log type」(記錄類型)。
點選「下一步」。
指定下列輸入參數的值：
- S3 URI：s3://snyk-group-logs/snyk/group/
- 來源刪除選項：根據偏好設定選取刪除選項。
- 檔案存在時間上限：包含在過去天數內修改的檔案。預設值為 180 天。
- 存取金鑰 ID：具有 S3 值區存取權的使用者存取金鑰。
- 存取密鑰：具有 S3 bucket 存取權的使用者私密金鑰。
- 資產命名空間：snyk.group
- 擷取標籤：視需要新增。
點選「下一步」。
在「完成」畫面中檢查新的動態饋給設定，然後按一下「提交」。

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