Code42 Incydr のコア データセットを収集する

以下でサポートされています。

このドキュメントでは、Amazon S3 を使用して Code42 Incydr のコア データセット（ユーザー、セッション、監査、ケース、必要に応じてファイル イベント）を Google Security Operations に取り込む方法について説明します。

始める前に

  • Google SecOps インスタンス
  • Code42 Incydr への特権アクセス
  • AWS（S3、IAM、Lambda、EventBridge）への特権アクセス

移行元の前提条件（ID、API キー、組織 ID、トークン）を収集する

  1. Code42 Incydr ウェブ UI にログインします。
  2. [管理> 統合> API クライアント] に移動します。
  3. 新しいクライアントを作成します。
  4. 次の詳細をコピーして安全な場所に保存します。
    1. クライアント ID。
    2. クライアント シークレット。
    3. ベース URL:（例: https://api.us.code42.com、https://api.us2.code42.com、https://api.ie.code42.com、https://api.gov.code42.com）。

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

  1. バケットの作成のユーザーガイドに沿って、Amazon S3 バケットを作成します。
  2. 後で使用するために、バケットの名前とリージョンを保存します。
  3. IAM ユーザーの作成のユーザーガイドに沿って、ユーザーを作成します。
  4. 作成したユーザーを選択します。
  5. [セキュリティ認証情報] タブを選択します。
  6. [アクセスキー] セクションで [アクセスキーを作成] をクリックします。
  7. [ユースケース] として [サードパーティ サービス] を選択します。
  8. [次へ] をクリックします。
  9. 省略可: 説明タグを追加します。
  10. [アクセスキーを作成] をクリックします。
  11. [CSV ファイルをダウンロード] をクリックして、[アクセスキー] と [シークレット アクセスキー] を保存し、後で使用できるようにします。
  12. [完了] をクリックします。
  13. [権限] タブを選択します。
  14. [権限ポリシー] セクションで、[権限を追加] をクリックします。
  15. [権限を追加] を選択します。
  16. [ポリシーを直接アタッチする] を選択します。
  17. AmazonS3FullAccess ポリシーを検索して選択します。
  18. [次へ] をクリックします。
  19. [権限を追加] をクリックします。

Code42 Incydr をポーリングするように AWS Lambda を設定する（変換なし）

  1. AWS コンソールで、[Lambda] > [Functions] > [Create function] に移動します。
  2. [Author from scratch] をクリックします。
  3. 次の構成の詳細を入力します。
    • 名前: 一意でわかりやすい名前を入力します（例: code42-incydr-pull）。
    • ランタイム: [Python 3.13] を選択します。
    • 権限: s3:PutObject と Cloudwatch を含むロールを選択します。
  4. [関数を作成] をクリックします。
  5. [Configuration] > [General configuration] > [Edit] を選択します。
  6. Timeout=5m と Memory=1024 MB を構成します。
  7. [保存] をクリックします。
  8. [構成] > [環境変数] > [編集] > [追加] を選択します。
    1. INCYDR_BASE_URL = https://api.us.code42.com
    2. INCYDR_CLIENT_ID = <Client ID>
    3. INCYDR_CLIENT_SECRET = <Client Secret>
    4. S3_BUCKET = code42-incydr
    5. S3_PREFIX = code42/
    6. PAGE_SIZE = 500
    7. LOOKBACK_MINUTES = 60
    8. STREAMS = users,sessions,audit,cases
    9. 省略可: FE_ADV_QUERY_JSON = ``
    10. 省略可: FE_PAGE_SIZE = 1000
  9. [保存] をクリックします。

  10. [コード] を選択し、次の Python コードを入力します。

    import base64, json, os, time
from datetime import datetime, timedelta, timezone
from urllib.parse import urlencode
from urllib.request import Request, urlopen
import boto3

BASE = os.environ["INCYDR_BASE_URL"].rstrip("/")
CID = os.environ["INCYDR_CLIENT_ID"]
CSECRET = os.environ["INCYDR_CLIENT_SECRET"]
BUCKET = os.environ["S3_BUCKET"]
PREFIX_BASE = os.environ.get("S3_PREFIX", "code42/")
PAGE_SIZE = int(os.environ.get("PAGE_SIZE", "500"))
LOOKBACK_MINUTES = int(os.environ.get("LOOKBACK_MINUTES", "60"))
STREAMS = [s.strip() for s in os.environ.get("STREAMS", "users").split(",") if s.strip()]
FE_ADV_QUERY_JSON = os.environ.get("FE_ADV_QUERY_JSON", "").strip()
FE_PAGE_SIZE = int(os.environ.get("FE_PAGE_SIZE", "1000"))

s3 = boto3.client("s3")

def now_utc():
    return datetime.now(timezone.utc)

def iso_minus(minutes: int):
    return (now_utc() - timedelta(minutes=minutes)).strftime("%Y-%m-%dT%H:%M:%SZ")

def put_bytes(key: str, body: bytes):
    s3.put_object(Bucket=BUCKET, Key=key, Body=body)

def put_json(prefix: str, page_label: str, data):
    ts = now_utc().strftime("%Y/%m/%d/%H%M%S")
    key = f"{PREFIX_BASE}{prefix}{ts}-{page_label}.json"
    put_bytes(key, json.dumps(data).encode("utf-8"))
    return key

def auth_header():
    auth = base64.b64encode(f"{CID}:{CSECRET}".encode()).decode()
    req = Request(f"{BASE}/v1/oauth", data=b"", method="POST")
    req.add_header("Authorization", f"Basic {auth}")
    req.add_header("Accept", "application/json")
    with urlopen(req, timeout=30) as r:
        data = json.loads(r.read().decode())
    return {"Authorization": f"Bearer {data['access_token']}", "Accept": "application/json"}

def http_get(path: str, params: dict | None = None, headers: dict | None = None):
    url = f"{BASE}{path}"
    if params:
        url += ("?" + urlencode(params))
    req = Request(url, method="GET")
    for k, v in (headers or {}).items():
        req.add_header(k, v)
    with urlopen(req, timeout=60) as r:
        return r.read()

def http_post_json(path: str, body: dict, headers: dict | None = None):
    url = f"{BASE}{path}"
    req = Request(url, data=json.dumps(body).encode("utf-8"), method="POST")
    req.add_header("Content-Type", "application/json")
    for k, v in (headers or {}).items():
        req.add_header(k, v)
    with urlopen(req, timeout=120) as r:
        return r.read()

# USERS (/v1/users)

def pull_users(hdrs):
    next_token = None
    pages = 0
    while True:
        params = {"active": "true", "blocked": "false", "pageSize": PAGE_SIZE}
        if next_token:
            params["pgToken"] = next_token
        raw = http_get("/v1/users", params, hdrs)
        data = json.loads(raw.decode())
        put_json("users/", f"users-page-{pages}", data)
        pages += 1
        next_token = data.get("nextPgToken") or data.get("next_pg_token")
        if not next_token:
            break
    return pages

# SESSIONS (/v1/sessions) — alerts live inside sessions

def pull_sessions(hdrs):
    start_iso = iso_minus(LOOKBACK_MINUTES)
    next_token = None
    pages = 0
    while True:
        params = {
            "hasAlerts": "true",
            "startTime": start_iso,
            "pgSize": PAGE_SIZE,
        }
        if next_token:
            params["pgToken"] = next_token
        raw = http_get("/v1/sessions", params, hdrs)
        data = json.loads(raw.decode())
        put_json("sessions/", f"sessions-page-{pages}", data)
        pages += 1
        next_token = data.get("nextPgToken") or data.get("next_page_token")
        if not next_token:
            break
    return pages

# AUDIT LOG (/v1/audit) — CSV export or paged JSON; write as received

def pull_audit(hdrs):
    start_iso = iso_minus(LOOKBACK_MINUTES)
    next_token = None
    pages = 0
    while True:
        params = {"startTime": start_iso, "pgSize": PAGE_SIZE}
        if next_token:
            params["pgToken"] = next_token
        raw = http_get("/v1/audit", params, hdrs)
        try:
            data = json.loads(raw.decode())
            put_json("audit/", f"audit-page-{pages}", data)
            next_token = data.get("nextPgToken") or data.get("next_page_token")
            pages += 1
            if not next_token:
                break
        except Exception:
            ts = now_utc().strftime("%Y/%m/%d/%H%M%S")
            key = f"{PREFIX_BASE}audit/{ts}-audit-export.bin"
            put_bytes(key, raw)
            pages += 1
            break
    return pages

# CASES (/v1/cases)

def pull_cases(hdrs):
    next_token = None
    pages = 0
    while True:
        params = {"pgSize": PAGE_SIZE}
        if next_token:
            params["pgToken"] = next_token
        raw = http_get("/v1/cases", params, hdrs)
        data = json.loads(raw.decode())
        put_json("cases/", f"cases-page-{pages}", data)
        pages += 1
        next_token = data.get("nextPgToken") or data.get("next_page_token")
        if not next_token:
            break
    return pages

# FILE EVENTS (/v2/file-events/search) — enabled only if you provide FE_ADV_QUERY_JSON

def pull_file_events(hdrs):
    if not FE_ADV_QUERY_JSON:
        return 0
    try:
        base_query = json.loads(FE_ADV_QUERY_JSON)
    except Exception:
        raise RuntimeError("FE_ADV_QUERY_JSON is not valid JSON")

    pages = 0
    next_token = None
    while True:
        body = dict(base_query)
        body["pgSize"] = FE_PAGE_SIZE
        if next_token:
            body["pgToken"] = next_token
        raw = http_post_json("/v2/file-events/search", body, hdrs)
        data = json.loads(raw.decode())
        put_json("file_events/", f"fileevents-page-{pages}", data)
        pages += 1
        next_token = (
            data.get("nextPgToken")
            or data.get("next_page_token")
            or (data.get("file_events") or {}).get("nextPgToken")
        )
        if not next_token:
            break
    return pages

def handler(event, context):
    hdrs = auth_header()
    report = {}
    if "users" in STREAMS:
        report["users_pages"] = pull_users(hdrs)
    if "sessions" in STREAMS:
        report["sessions_pages"] = pull_sessions(hdrs)
    if "audit" in STREAMS:
        report["audit_pages"] = pull_audit(hdrs)
    if "cases" in STREAMS:
        report["cases_pages"] = pull_cases(hdrs)
    if "file_events" in STREAMS:
        report["file_events_pages"] = pull_file_events(hdrs)
    return report

def lambda_handler(event, context):
    return handler(event, context)

  11. [デプロイ] をクリックします。

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

  1. AWS コンソールで、[Amazon EventBridge] > [ルール] に移動します。
  2. [ルールを作成] をクリックします。
  3. 次の構成の詳細を入力します。
    • スケジュール パターン: [1 時間の固定レート] を選択します。
    • 名前: 一意でわかりやすい名前を入力します（例: code42-incydr-hourly）。
    • ターゲット: [Lambda 関数] を選択し、code42-incydr-pull を選択します。
  4. [ルールを作成] をクリックします。

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

  1. AWS コンソールで、[IAM] > [Users] に移動し、[Add users] をクリックします。
  2. 次の構成の詳細を入力します。
    • ユーザー: 一意の名前を入力します（例: secops-reader）。
    • アクセスタイプ: [Access key - Programmatic access] を選択します。
    • [ユーザーを作成] をクリックします。
  3. 最小限の読み取りポリシー（カスタム）を適用する: [ユーザー] > secops-reader を選択 > [権限] > [権限を追加] > [ポリシーを直接適用] > [ポリシーを作成]

  4. JSON エディタで次のポリシーを入力します。

    {
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "s3:GetObject"
      ],
      "Resource": "arn:aws:s3:::<your-bucket>/*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "s3:ListBucket"
      ],
      "Resource": "arn:aws:s3:::<your-bucket>"
    }
  ]
}

  5. 名前を secops-reader-policy に設定します。

  6. [ポリシーの作成> 検索/選択> 次へ> 権限を追加] に移動します。

  7. [セキュリティ認証情報] > [アクセスキー] > [アクセスキーを作成] に移動します。

  8. CSV をダウンロードします（これらの値はフィードに入力されます）。

Code42 Incydr のログを取り込むように Google SecOps でフィードを構成する

  1. [SIEM 設定] > [フィード] に移動します。
  2. [Add New Feed] をクリックします。
  3. [フィード名] フィールドに、フィードの名前を入力します（例: Code42 Incydr Datasets）。
  4. [ソースタイプ] として [Amazon S3 V2] を選択します。
  5. [ログタイプ] として [Code42 Incydr] を選択します。
  6. [次へ] をクリックします。
  7. 次の入力パラメータの値を指定します。
    • S3 URI: s3://code42-incydr/code42/
    • Source deletion options: 必要に応じて削除オプションを選択します。
    • 最大ファイル経過時間: デフォルトは 180 日です。
    • アクセスキー ID: S3 バケットにアクセスできるユーザー アクセスキー。
    • シークレット アクセスキー: S3 バケットにアクセスできるユーザーのシークレット キー。
    • アセットの名前空間: アセットの名前空間。
    • Ingestion labels: このフィードのイベントに適用されるラベル。
  8. [次へ] をクリックします。
  9. [Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。

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