このページは 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 UI で、[Account settings] > [API token] に移動し、API トークンを生成します。
トークンをコピーして安全な場所に保存し、後で SNYK_TOKEN として使用します。
グループに切り替えて、グループ設定を開きます。
URL（https://app.snyk.io/group/<GROUP_ID>/...）からグループ ID をコピーして保存し、後で GROUP_ID として使用します。
ベース API エンドポイント: https://api.snyk.io（必要に応じて API_BASE でオーバーライド）。
トークンを持つユーザーにグループ管理者ロールを割り当てます。（ユーザーは Group Audit Logs と Group Issues を表示できる必要があります）。

Google SecOps 用に AWS S3 バケットと IAM を構成する

バケットの作成のユーザーガイドに沿って、Amazon S3 バケットを作成します。
後で参照できるように、バケットの名前とリージョンを保存します（例: snyk-group-logs）。
IAM ユーザーの作成のユーザーガイドに沿って、ユーザーを作成します。
作成したユーザーを選択します。
[セキュリティ認証情報] タブを選択します。
[アクセスキー] セクションで [アクセスキーを作成] をクリックします。
[ユースケース] で [サードパーティサービス] を選択します。
[次へ] をクリックします。
省略可: 説明タグを追加します。
[アクセスキーを作成] をクリックします。
[CSV ファイルをダウンロード] をクリックして、[アクセスキー] と [シークレットアクセスキー] を保存し、後で使用できるようにします。
[完了] をクリックします。
[権限] タブを選択します。
[権限ポリシー] セクションで、[権限を追加] をクリックします。
[権限を追加] を選択します。
[ポリシーを直接アタッチする] を選択します。
AmazonS3FullAccess ポリシーを検索して選択します。
[次へ] をクリックします。
[権限を追加] をクリックします。

S3 アップロードの IAM ポリシーとロールを構成する

AWS コンソールで、[IAM] > [ポリシー] > [ポリシーの作成] > [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 を置き換えます。
[次へ] > [ポリシーを作成] をクリックします。
[IAM] > [ロール] > [ロールの作成] > [AWS サービス] > [Lambda] に移動します。
新しく作成したポリシーを関連付けます。
ロールに「WriteSnykGroupToS3Role」という名前を付けて、[ロールを作成] をクリックします。

Lambda 関数を作成する

AWS コンソールで、[Lambda] > [Functions] > [Create function] に移動します。
[Author from scratch] をクリックします。
次の構成情報を提供してください。

設定	値
名前	`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())

[構成> 環境変数 > 編集 > 新しい環境変数を追加] に移動します。

次の環境変数を入力し、実際の値に置き換えます。

キー	例
`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] > [関数] > <your-function> を開きます）。
[CONFIGURATION] タブを選択します。
[全般設定] パネルで、[編集] をクリックします。
[Timeout] を [5 minutes (300 seconds)] に変更し、[Save] をクリックします。

注: タイムアウトは上限値です。つまり、タイマーが 5 分に達したときに関数がまだ実行されている場合、AWS Lambda は関数を終了します。実行時間が短くなった場合は、通常どおり終了します。料金は実際の実行時間に対してのみ課金されます。

EventBridge スケジュールを作成する

Amazon EventBridge > Scheduler > スケジュールの作成に移動します。
次の構成の詳細を入力します。
- 定期的なスケジュール: レート（1 hour）。
- ターゲット: Lambda 関数 snyk_group_audit_issues_to_s3。
- 名前: snyk-group-audit-issues-1h
[スケジュールを作成] をクリックします。

省略可: Google SecOps 用の読み取り専用の IAM ユーザーと鍵を作成する

AWS コンソールで、[IAM] > [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 をダウンロードします（これらの値はフィードに入力されます）。

Snyk グループレベルの監査ログと問題ログを取り込むように Google SecOps でフィードを構成する

[SIEM 設定] > [フィード] に移動します。
[+ 新しいフィードを追加] をクリックします。
[フィード名] フィールドに、フィードの名前を入力します（例: Snyk Group Audit/Issues）。
[ソースタイプ] として [Amazon S3 V2] を選択します。
[ログタイプ] として [Snyk Group level audit/issues logs] を選択します。
[次へ] をクリックします。
次の入力パラメータの値を指定します。
- S3 URI: s3://snyk-group-logs/snyk/group/
- Source deletion options: 必要に応じて削除オプションを選択します。
- ファイルの最大経過日数: 指定した日数以内に変更されたファイルを含めます。デフォルトは 180 日です。
- アクセスキー ID: S3 バケットにアクセスできるユーザーアクセスキー。
- シークレットアクセスキー: S3 バケットにアクセスできるユーザーのシークレットキー。
- アセットの Namespace: snyk.group
- 取り込みラベル: 必要に応じて追加します。
[次へ] をクリックします。
[Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。

さらにサポートが必要な場合 コミュニティメンバーや Google SecOps のプロフェッショナルから回答を得ることができます。