Mengumpulkan log audit dan masalah tingkat grup Snyk

Didukung di:

Panduan ini menjelaskan cara Anda dapat menyerap log audit dan masalah tingkat grup Snyk ke Google Security Operations menggunakan Amazon S3.

Sebelum memulai

Pastikan Anda memenuhi prasyarat berikut:

  • Instance Google SecOps
  • Akses istimewa ke Grup Snyk (token API dengan akses baca; ID Grup)
  • Akses istimewa ke AWS (S3, IAM, Lambda, EventBridge)

Mendapatkan ID Grup Snyk dan token API

  1. Di UI Snyk, buka Account settings > API token dan buat API token.
  2. Salin dan simpan token di lokasi yang aman untuk digunakan nanti sebagai SNYK_TOKEN.
  3. Beralih ke Grup Anda dan buka Setelan grup.
  4. Salin dan simpan ID Grup dari URL (https://app.snyk.io/group/<GROUP_ID>/...) untuk digunakan nanti sebagai GROUP_ID.
  5. Endpoint API dasar: https://api.snyk.io (ganti dengan API_BASE jika diperlukan).
  6. Tetapkan peran Admin Grup kepada pengguna dengan token. (Pengguna harus dapat melihat Log Audit Grup dan Masalah Grup).

Mengonfigurasi bucket AWS S3 dan IAM untuk Google SecOps

  1. Buat bucket Amazon S3 dengan mengikuti panduan pengguna ini: Membuat bucket
  2. Simpan Name dan Region bucket untuk referensi di masa mendatang (misalnya, snyk-group-logs).
  3. Buat pengguna dengan mengikuti panduan pengguna ini: Membuat pengguna IAM.
  4. Pilih Pengguna yang dibuat.
  5. Pilih tab Kredensial keamanan.
  6. Klik Create Access Key di bagian Access Keys.
  7. Pilih Layanan pihak ketiga sebagai Kasus penggunaan.
  8. Klik Berikutnya.
  9. Opsional: tambahkan tag deskripsi.
  10. Klik Create access key.
  11. Klik Download CSV file untuk menyimpan Access Key dan Secret Access Key untuk digunakan nanti.
  12. Klik Selesai.
  13. Pilih tab Izin.
  14. Klik Tambahkan izin di bagian Kebijakan izin.
  15. Pilih Tambahkan izin.
  16. Pilih Lampirkan kebijakan secara langsung
  17. Telusuri dan pilih kebijakan AmazonS3FullAccess.
  18. Klik Berikutnya.
  19. Klik Add permissions.

Mengonfigurasi kebijakan dan peran IAM untuk upload S3

  1. Di konsol AWS, buka IAM > Policies > Create policy > JSON tab.

  2. Masukkan kebijakan berikut (mencakup akses tulis untuk semua objek di bucket dan akses baca ke file status yang digunakan Lambda Anda):

    {
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "PutAllSnykGroupObjects",
      "Effect": "Allow",
      "Action": ["s3:PutObject", "s3:GetObject"],
      "Resource": "arn:aws:s3:::snyk-group-logs/*"
    }
  ]
}
    • Ganti snyk-group-logs jika Anda memasukkan nama bucket yang berbeda.

  3. Klik Berikutnya > Buat kebijakan.

  4. Buka IAM > Roles > Create role > AWS service > Lambda.

  5. Lampirkan kebijakan yang baru dibuat.

  6. Beri nama peran WriteSnykGroupToS3Role, lalu klik Buat peran.

Buat fungsi Lambda

  1. Di Konsol AWS, buka Lambda > Functions > Create function.
  2. Klik Buat dari awal.
  3. Berikan detail konfigurasi berikut:
Setelan Nilai
Nama snyk_group_audit_issues_to_s3
Runtime Python 3.13
Arsitektur x86_64
Peran eksekusi WriteSnykGroupToS3Role

  1. Setelah fungsi dibuat, buka tab Code, hapus stub, dan masukkan kode berikut (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())

  2. Buka Configuration > Environment variables > Edit > Add new environment variable.

  3. Masukkan variabel lingkungan berikut, ganti dengan nilai Anda:

    Kunci Contoh
    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

  4. Setelah fungsi dibuat, tetap buka halamannya (atau buka Lambda > Functions > <your-function>).

  5. Pilih tab Configuration

  6. Di panel General configuration, klik Edit.

  7. Ubah Waktu tunggu menjadi 5 menit (300 detik), lalu klik Simpan.

Membuat jadwal EventBridge

  1. Buka Amazon EventBridge > Scheduler > Create schedule.
  2. Berikan detail konfigurasi berikut:
    • Jadwal berulang: Tarif (1 hour).
    • Target: Fungsi Lambda Anda snyk_group_audit_issues_to_s3.
    • Name: snyk-group-audit-issues-1h.
  3. Klik Buat jadwal.

Opsional: Buat pengguna & kunci IAM hanya baca untuk Google SecOps

  1. Di Konsol AWS, buka IAM > Pengguna > Tambahkan pengguna.
  2. Klik Add users.
  3. Berikan detail konfigurasi berikut:
    • Pengguna: secops-reader.
    • Jenis akses: Kunci akses — Akses terprogram.
  4. Klik Buat pengguna.
  5. Lampirkan kebijakan baca minimal (kustom): Pengguna > secops-reader > Izin > Tambahkan izin > Lampirkan kebijakan secara langsung > Buat kebijakan.

  6. Di editor JSON, masukkan kebijakan berikut:

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

  7. Tetapkan nama ke secops-reader-policy.

  8. Buka Buat kebijakan > cari/pilih > Berikutnya > Tambahkan izin.

  9. Buka Kredensial keamanan > Kunci akses > Buat kunci akses.

  10. Download CSV (nilai ini dimasukkan ke dalam feed).

Mengonfigurasi feed di Google SecOps untuk menyerap log audit dan masalah tingkat Grup Snyk

  1. Buka Setelan SIEM > Feed.
  2. Klik + Tambahkan Feed Baru.
  3. Di kolom Nama feed, masukkan nama untuk feed (misalnya, Snyk Group Audit/Issues).
  4. Pilih Amazon S3 V2 sebagai Jenis sumber.
  5. Pilih Snyk Group level audit/issues logs sebagai Log type.
  6. Klik Berikutnya.
  7. Tentukan nilai untuk parameter input berikut:
    • URI S3: s3://snyk-group-logs/snyk/group/
    • Opsi penghapusan sumber: Pilih opsi penghapusan sesuai preferensi Anda.
    • Usia File Maksimum: Menyertakan file yang diubah dalam jumlah hari terakhir. Defaultnya adalah 180 hari.
    • ID Kunci Akses: Kunci akses pengguna dengan akses ke bucket S3.
    • Kunci Akses Rahasia: Kunci rahasia pengguna dengan akses ke bucket S3.
    • Namespace aset: snyk.group
    • Label penyerapan: Tambahkan jika diinginkan.
  8. Klik Berikutnya.
  9. Tinjau konfigurasi feed baru Anda di layar Selesaikan, lalu klik Kirim.

Perlu bantuan lain? Dapatkan jawaban dari anggota Komunitas dan profesional Google SecOps.