このページは Cloud Translation API によって翻訳されました。

Digital Shadows SearchLight ログを収集する

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

Google SecOps SIEM

このドキュメントでは、Amazon S3 を使用して Digital Shadows SearchLight ログを Google Security Operations に取り込む方法について説明します。パーサーは、JSON ログからセキュリティイベントデータを抽出します。統合データモデル（UDM）フィールドを初期化し、JSON ペイロードを解析し、関連するフィールドを UDM スキーマにマッピングし、grok パターンを使用してメールやホスト名などのエンティティを抽出し、UDM イベント内に security_result オブジェクトと metadata オブジェクトを構築します。

始める前に

次の前提条件を満たしていることを確認してください。

Google SecOps インスタンス。
Digital Shadows SearchLight テナントへの特権アクセス。
AWS（S3、Identity and Access Management（IAM）、Lambda、EventBridge）への特権アクセス。

Digital Shadows SearchLight の前提条件（ID、API キー、組織 ID、トークン）を収集する

Digital Shadows SearchLight ポータルにログインします。
[設定] > [API 認証情報] に移動します。
新しい API クライアントまたは鍵ペアを作成します。
次の詳細をコピーして安全な場所に保存します。
- API キー
- API Secret
- アカウント ID
- API ベース URL: https://api.searchlight.app/v1 または https://portal-digitalshadows.com/api/v1

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

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

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

AWS コンソールで、[IAM] > [ポリシー] に移動します。
[ポリシーを作成> [JSON] タブ] をクリックします。
次のポリシーをコピーして貼り付けます。

ポリシー JSON（別のバケット名を入力した場合は digital-shadows-logs を置き換えます）:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowPutObjects",
      "Effect": "Allow",
      "Action": "s3:PutObject",
      "Resource": "arn:aws:s3:::digital-shadows-logs/*"
    },
    {
      "Sid": "AllowGetStateObject",
      "Effect": "Allow",
      "Action": "s3:GetObject",
      "Resource": "arn:aws:s3:::digital-shadows-logs/digital-shadows-searchlight/state.json"
    }
  ]
}

[次へ] > [ポリシーを作成] をクリックします。
[IAM] > [ロール] > [ロールの作成] > [AWS サービス] > [Lambda] に移動します。
新しく作成したポリシーを関連付けます。
ロールに「digital-shadows-lambda-role」という名前を付けて、[ロールを作成] をクリックします。

Lambda 関数を作成する

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

設定値

名前 digital-shadows-collector

ランタイム Python 3.13

アーキテクチャ x86_64

実行ロール digital-shadows-lambda-role

設定	値
名前	`digital-shadows-collector`
ランタイム	Python 3.13
アーキテクチャ	x86_64
実行ロール	`digital-shadows-lambda-role`

関数を作成したら、[コード] タブを開き、スタブを削除して次のコード（digital-shadows-collector.py）を貼り付けます。

import json
import os
import base64
import logging
import time
from datetime import datetime, timedelta, timezone
from urllib.parse import urlencode

import boto3
import urllib3

logger = logging.getLogger()
logger.setLevel(logging.INFO)

HTTP = urllib3.PoolManager(retries=False)

def _basic_auth_header(key: str, secret: str) -> str:
    token = base64.b64encode(f"{key}:{secret}".encode("utf-8")).decode("utf-8")
    return f"Basic {token}"

def _load_state(s3, bucket, key, default_days=30) -> str:
    """Return ISO8601 checkpoint (UTC)."""
    try:
        obj = s3.get_object(Bucket=bucket, Key=key)
        state = json.loads(obj["Body"].read().decode("utf-8"))
        ts = state.get("last_timestamp")
        if ts:
            return ts
    except s3.exceptions.NoSuchKey:
        pass
    except Exception as e:
        logger.warning(f"State read error: {e}")
    return (datetime.now(timezone.utc) - timedelta(days=default_days)).isoformat()

def _save_state(s3, bucket, key, ts: str) -> None:
    s3.put_object(
        Bucket=bucket,
        Key=key,
        Body=json.dumps({"last_timestamp": ts}).encode("utf-8"),
        ContentType="application/json",
    )

def _get_json(url: str, headers: dict, params: dict, backoff_s=2, max_retries=3) -> dict:
    qs = f"?{urlencode(params)}" if params else ""
    for attempt in range(max_retries):
        r = HTTP.request("GET", f"{url}{qs}", headers=headers)
        if r.status == 200:
            return json.loads(r.data.decode("utf-8"))
        if r.status in (429, 500, 502, 503, 504):
            wait = backoff_s * (2 ** attempt)
            logger.warning(f"HTTP {r.status} from DS API, retrying in {wait}s")
            time.sleep(wait)
            continue
        raise RuntimeError(f"DS API error {r.status}: {r.data[:200]}")
    raise RuntimeError("Exceeded retry budget for DS API")

def _collect(api_base, headers, path, since_ts, account_id, page_size, max_pages, time_param):
    items = []
    for page in range(max_pages):
        params = {
            "limit": page_size,
            "offset": page * page_size,
            time_param: since_ts,
        }
        if account_id:
            params["account-id"] = account_id

        data = _get_json(f"{api_base}/{path}", headers, params)
        batch = data.get("items") or data.get("data") or []
        if not batch:
            break
        items.extend(batch)
        if len(batch) < page_size:
            break
    return items

def lambda_handler(event, context):
    # Required
    s3_bucket  = os.environ["S3_BUCKET"]
    api_key    = os.environ["DS_API_KEY"]
    api_secret = os.environ["DS_API_SECRET"]

    # Optional / defaults
    s3_prefix  = os.environ.get("S3_PREFIX", "digital-shadows-searchlight/")
    state_key  = os.environ.get("STATE_KEY", "digital-shadows-searchlight/state.json")
    api_base   = os.environ.get("API_BASE", "https://api.searchlight.app/v1")
    account_id = os.environ.get("DS_ACCOUNT_ID", "")
    page_size  = int(os.environ.get("PAGE_SIZE", "100"))
    max_pages  = int(os.environ.get("MAX_PAGES", "10"))

    s3 = boto3.client("s3")
    last_ts = _load_state(s3, s3_bucket, state_key)
    logger.info(f"Checkpoint: {last_ts}")

    headers = {
        "Authorization": _basic_auth_header(api_key, api_secret),
        "Accept": "application/json",
        "User-Agent": "Chronicle-DigitalShadows-S3/1.0",
    }

    records = []

    # Incidents (time filter often 'published-after' or 'updated-since' depending on tenancy)
    incidents = _collect(api_base, headers, "incidents", last_ts, account_id, page_size, max_pages, time_param="published-after")
    for incident in incidents:
        incident['_source_type'] = 'incident'
    records.extend(incidents)

    # Intelligence incidents (alerts)
    intel_incidents = _collect(api_base, headers, "intel-incidents", last_ts, account_id, page_size, max_pages, time_param="published-after")
    for intel in intel_incidents:
        intel['_source_type'] = 'intelligence_incident'
    records.extend(intel_incidents)

    # Indicators (IOCs)
    indicators = _collect(api_base, headers, "indicators", last_ts, account_id, page_size, max_pages, time_param="lastUpdated-after")
    for indicator in indicators:
        indicator['_source_type'] = 'ioc'
    records.extend(indicators)

    if records:
        # Choose newest timestamp seen in this batch
        newest = max(
            (r.get("updated") or r.get("raised") or r.get("lastUpdated") or last_ts)
            for r in records
        )
        key = f"{s3_prefix}digital_shadows_{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}.json"
        body = "\n".join(json.dumps(r, separators=(",", ":")) for r in records).encode("utf-8")

        s3.put_object(
            Bucket=s3_bucket,
            Key=key,
            Body=body,
            ContentType="application/x-ndjson",
        )
        _save_state(s3, s3_bucket, state_key, newest)
        msg = f"Wrote {len(records)} records to s3://{s3_bucket}/{key}"
    else:
        msg = "No new records"

    logger.info(msg)
    return {"statusCode": 200, "body": msg}

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

次の表に示す環境変数を入力し、サンプル値を自分の値に置き換えます。

環境変数

キー	値の例
`S3_BUCKET`	`digital-shadows-logs`
`S3_PREFIX`	`digital-shadows-searchlight/`
`STATE_KEY`	`digital-shadows-searchlight/state.json`
`DS_API_KEY`	`<your-6-character-api-key>`
`DS_API_SECRET`	`<your-32-character-api-secret>`
`API_BASE`	`https://api.searchlight.app/v1`（または `https://portal-digitalshadows.com/api/v1`）
`DS_ACCOUNT_ID`	`<your-account-id>`（ほとんどのテナントで必要）
`PAGE_SIZE`	`100`
`MAX_PAGES`	`10`

関数が作成されたら、そのページにとどまるか、[Lambda] > [関数] > [your-function] を開きます。
[CONFIGURATION] タブを選択します。
[全般設定] パネルで、[編集] をクリックします。
[Timeout] を [5 minutes (300 seconds)] に変更し、[Save] をクリックします。

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

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

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

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

AWS コンソール > IAM > ユーザーに移動します。
[ユーザーを追加] をクリックします。
次の構成の詳細を入力します。
- ユーザー: 「secops-reader」と入力します。
- アクセスの種類: [アクセスキー - プログラムによるアクセス] を選択します。
[ユーザーを作成] をクリックします。
最小限の読み取りポリシー（カスタム）を関連付ける: [ユーザー] > [secops-reader] > [権限] > [権限を追加] > [ポリシーを直接関連付ける] > [ポリシーを作成]。

JSON:

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

名前 = secops-reader-policy。
[ポリシーを作成> 検索/選択> 次へ> 権限を追加] をクリックします。
secops-reader のアクセスキーを作成します（[セキュリティ認証情報>アクセスキー]）。
[アクセスキーを作成] をクリックします。
.CSV をダウンロードします。（これらの値はフィードに貼り付けます）。

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

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

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