Google は AI 技術を使用して、コンテンツをご希望の言語に翻訳しています。AI 翻訳には誤りが含まれる場合があります。

Sonrai Security のログを収集する

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

Google SecOps SIEM

このドキュメントでは、Google Cloud Storage V2 を使用して Sonrai Security ログを Google Security Operations に取り込む方法について説明します。

Sonrai Security は、マルチクラウド環境全体で ID、データ、アクセスガバナンスを実現するエンタープライズクラウドセキュリティプラットフォームです。GraphQL API を介して、セキュリティ検出結果、チケット、監査データを提供します。

始める前に

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

Google SecOps インスタンス
Cloud Storage API が有効になっている GCP プロジェクト
GCS バケットを作成および管理する権限
GCS バケットの IAM ポリシーを管理する権限
Cloud Run サービス、Pub/Sub トピック、Cloud Scheduler ジョブを作成する権限
管理者権限と API キーアクセスによる Sonrai Security への特権アクセス

Google Cloud Storage バケットを作成する

Google Cloud コンソールに移動します。
プロジェクトを選択するか、新しいプロジェクトを作成します。
ナビゲーションメニューで、[Cloud Storage > バケット] に移動します。
[バケットを作成] をクリックします。

次の構成情報を提供してください。

設定	値
バケットに名前を付ける	グローバルに一意の名前（`sonrai-security-logs` など）を入力します。
ロケーションタイプ	ニーズに応じて選択します（リージョン、デュアルリージョン、マルチリージョン）。
ロケーション	場所を選択します（例: `us-central1`）。
ストレージクラス	Standard（アクセス頻度の高いログにおすすめ）
アクセス制御	均一（推奨）
保護ツール	省略可: オブジェクトのバージョニングまたは保持ポリシーを有効にする

[作成] をクリックします。

Sonrai Security API 認証情報を収集する

API キーを生成

管理者として Sonrai Security プラットフォームにログインします。
[設定] > [API キー] に移動します。
[API キーを作成] をクリックします。
キーの名前を入力します（例: Google Security Operations Integration）。
生成された API キーをコピーして、安全な場所に保存します。

組織の URL を特定する

API エンドポイントは、Sonrai Security 組織に基づいています。書式は次のとおりです。

コンポーネント	値
GraphQL エンドポイント	`https://YOUR_ORG.sonraisecurity.com/graphql`

権限を確認する

API キーに必要なアクセス権があることを確認するには:

Sonrai Security プラットフォームにログインします。
[設定] > [API キー] に移動します。
API キーがリストに表示され、有効になっていることを確認します。
キーにチケット、検出結果、監査データをクエリする権限があることを確認します。
権限が制限されている場合は、Sonrai Security 管理者にお問い合わせください。

テスト API へのアクセス

統合に進む前に、認証情報をテストします。

# Replace with your actual credentials
API_KEY="your-api-key"
ORG_URL="https://YOUR_ORG.sonraisecurity.com"

# Test API access - query tickets
curl -v -X POST \
  -H "Authorization: Bearer ${API_KEY}" \
  -H "Content-Type: application/json" \
  -d '{"query": "{ Tickets(where: {createdDate: {op: GT, value: \"2024-01-01T00:00:00Z\"}}) { count } }"}' \
  "${ORG_URL}/graphql"

Cloud Run functions のサービスアカウントを作成する

Cloud Run functions には、GCS バケットへの書き込み権限があり、Pub/Sub によって呼び出されるサービスアカウントが必要です。

サービスアカウントの作成

GCP コンソールで、[IAM と管理>サービスアカウント] に移動します。
[サービスアカウントを作成] をクリックします。
次の構成情報を指定します。
- サービスアカウント名: 「sonrai-collector-sa」と入力します。
- サービスアカウントの説明: 「Service account for Cloud Run function to collect Sonrai Security logs」と入力します。
[作成して続行] をクリックします。
[このサービスアカウントにプロジェクトへのアクセスを許可する] セクションで、次のロールを追加します。
1. [ロールを選択] をクリックします。
2. [ストレージオブジェクト管理者] を検索して選択します。
3. [+ 別のロールを追加] をクリックします。
4. [Cloud Run 起動元] を検索して選択します。
5. [+ 別のロールを追加] をクリックします。
6. [Cloud Functions 起動元] を検索して選択します。
[続行] をクリックします。
[完了] をクリックします。

これらのロールは、次の操作に必要です。

Storage オブジェクト管理者: ログを GCS バケットに書き込み、状態ファイルを管理する
Cloud Run 起動元: Pub/Sub が関数を呼び出すことを許可します。
Cloud Functions 起動元: 関数の呼び出しを許可する

GCS バケットに対する IAM 権限を付与する

GCS バケットに対する書き込み権限をサービスアカウントに付与します。

[Cloud Storage] > [バケット] に移動します。
バケット名（sonrai-security-logs など）をクリックします。
[権限] タブに移動します。
[アクセス権を付与] をクリックします。
次の構成情報を指定します。
- プリンシパルを追加: サービスアカウントのメールアドレス（例: sonrai-collector-sa@your-project.iam.gserviceaccount.com）を入力します。
- ロールを割り当てる: [ストレージオブジェクト管理者] を選択します。
[保存] をクリックします。

Pub/Sub トピックの作成

Cloud Scheduler がパブリッシュし、Cloud Run 関数がサブスクライブする Pub/Sub トピックを作成します。

GCP Console で、[Pub/Sub> トピック] に移動します。
[トピックを作成] をクリックします。
次の構成情報を指定します。
- トピック ID: 「sonrai-trigger」と入力します。
- その他の設定はデフォルトのままにします。
[作成] をクリックします。

ログを収集する Cloud Run 関数を作成する

Cloud Run 関数は、Cloud Scheduler からの Pub/Sub メッセージによってトリガーされ、Sonrai Security GraphQL API からログを取得して GCS に書き込みます。

GCP Console で、[Cloud Run] に移動します。
[サービスを作成] をクリックします。
[関数] を選択します（インラインエディタを使用して関数を作成します）。

[構成] セクションで、次の構成の詳細を指定します。

設定	値
サービス名	`sonrai-collector`
リージョン	GCS バケットと一致するリージョンを選択します（例: `us-central1`）。
ランタイム	[Python 3.12] 以降を選択します。

[トリガー（省略可）] セクションで、次の操作を行います。
1. [+ トリガーを追加] をクリックします。
2. [Cloud Pub/Sub] を選択します。
3. [Cloud Pub/Sub トピックを選択してください] で、トピック sonrai-trigger を選択します。
4. [保存] をクリックします。
[認証] セクションで、次の操作を行います。
1. [認証が必要] を選択します。
2. Identity and Access Management（IAM）を確認します。
注: Pub/Sub は、関数を呼び出すときに認証を自動的に処理します。
下にスクロールして、[コンテナ、ネットワーキング、セキュリティ] を開きます。
[セキュリティ] タブに移動します。
- サービスアカウント: サービスアカウント sonrai-collector-sa を選択します。

[コンテナ] タブに移動します。

[変数とシークレット] をクリックします。
環境変数ごとに [+ 変数を追加] をクリックします。

変数名	値の例	説明
`GCS_BUCKET`	`sonrai-security-logs`	GCS バケット名
`GCS_PREFIX`	`sonrai`	ログファイルの接頭辞
`STATE_KEY`	`sonrai/state.json`	状態ファイルのパス
`API_KEY`	`your-api-key`	Sonrai Security API キー
`ORG_URL`	`https://YOUR_ORG.sonraisecurity.com`	Sonrai 組織の URL
`MAX_RECORDS`	`10000`	実行あたりの最大レコード数
`PAGE_SIZE`	`500`	1 ページあたりのレコード数
`LOOKBACK_HOURS`	`24`	初期ルックバック期間

[変数とシークレット] タブで [リクエスト] まで下にスクロールします。
- リクエストのタイムアウト: 600 秒（10 分）を入力します。
[コンテナ] の [設定] タブに移動します。
- [リソース] セクションで次の操作を行います。
  - メモリ: 512 MiB 以上を選択します。
  - CPU: [1] を選択します。
[リビジョンスケーリング] セクションで、次の操作を行います。
- [インスタンスの最小数] に「0」と入力します。
- インスタンスの最大数: 100 と入力します（または、想定される負荷に基づいて調整します）。
[作成] をクリックします。
サービスが作成されるまで待ちます（1 ～ 2 分）。
サービスを作成すると、インラインコードエディタが自動的に開きます。

関数コードを追加する

[関数のエントリポイント] に「main」と入力します。

インラインコードエディタで、次の 2 つのファイルを作成します。

最初のファイル: main.py:

import functions_framework
from google.cloud import storage
import json
import os
import urllib3
from datetime import datetime, timezone, timedelta
import time

# Initialize HTTP client with timeouts
http = urllib3.PoolManager(
  timeout=urllib3.Timeout(connect=5.0, read=30.0),
  retries=False,
)

# Initialize Storage client
storage_client = storage.Client()

# Environment variables
GCS_BUCKET = os.environ.get('GCS_BUCKET')
GCS_PREFIX = os.environ.get('GCS_PREFIX', 'sonrai')
STATE_KEY = os.environ.get('STATE_KEY', 'sonrai/state.json')
API_KEY = os.environ.get('API_KEY', '')
ORG_URL = os.environ.get('ORG_URL', '').rstrip('/')
MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '10000'))
PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '500'))
LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))

def parse_datetime(value: str) -> datetime:
  """Parse ISO datetime string to datetime object."""
  if value.endswith("Z"):
    value = value[:-1] + "+00:00"
  return datetime.fromisoformat(value)

@functions_framework.cloud_event
def main(cloud_event):
  """
  Cloud Run function triggered by Pub/Sub to fetch Sonrai Security logs and write to GCS.

  Args:
    cloud_event: CloudEvent object containing Pub/Sub message
  """

  if not all([GCS_BUCKET, API_KEY, ORG_URL]):
    print('Error: Missing required environment variables')
    return

  try:
    bucket = storage_client.bucket(GCS_BUCKET)

    # Load state
    state = load_state(bucket, STATE_KEY)

    # Determine time window
    now = datetime.now(timezone.utc)
    last_time = None

    if isinstance(state, dict) and state.get("last_event_time"):
      try:
        last_time = parse_datetime(state["last_event_time"])
        last_time = last_time - timedelta(minutes=2)
      except Exception as e:
        print(f"Warning: Could not parse last_event_time: {e}")

    if last_time is None:
      last_time = now - timedelta(hours=LOOKBACK_HOURS)

    print(f"Fetching tickets from {last_time.isoformat()} to {now.isoformat()}")

    # Fetch tickets/findings
    records, newest_event_time = fetch_tickets(
      start_time=last_time,
      end_time=now,
      page_size=PAGE_SIZE,
      max_records=MAX_RECORDS,
    )

    if not records:
      print("No new records found.")
      save_state(bucket, STATE_KEY, now.isoformat())
      return

    # Write to GCS as NDJSON
    timestamp = now.strftime('%Y%m%d_%H%M%S')
    object_key = f"{GCS_PREFIX}/logs_{timestamp}.ndjson"
    blob = bucket.blob(object_key)

    ndjson = '\n'.join([json.dumps(record, ensure_ascii=False) for record in records]) + '\n'
    blob.upload_from_string(ndjson, content_type='application/x-ndjson')

    print(f"Wrote {len(records)} records to gs://{GCS_BUCKET}/{object_key}")

    if newest_event_time:
      save_state(bucket, STATE_KEY, newest_event_time)
    else:
      save_state(bucket, STATE_KEY, now.isoformat())

    print(f"Successfully processed {len(records)} records")

  except Exception as e:
    print(f'Error processing logs: {str(e)}')
    raise

def load_state(bucket, key):
  """Load state from GCS."""
  try:
    blob = bucket.blob(key)
    if blob.exists():
      state_data = blob.download_as_text()
      return json.loads(state_data)
  except Exception as e:
    print(f"Warning: Could not load state: {e}")

  return {}

def save_state(bucket, key, last_event_time_iso: str):
  """Save the last event timestamp to GCS state file."""
  try:
    state = {'last_event_time': last_event_time_iso}
    blob = bucket.blob(key)
    blob.upload_from_string(
      json.dumps(state, indent=2),
      content_type='application/json'
    )
    print(f"Saved state: last_event_time={last_event_time_iso}")
  except Exception as e:
    print(f"Warning: Could not save state: {e}")

def graphql_query(query: str):
  """Execute a GraphQL query against Sonrai Security API with rate limiting."""
  endpoint = f"{ORG_URL}/graphql"

  headers = {
    'Authorization': f'Bearer {API_KEY}',
    'Content-Type': 'application/json',
    'Accept': 'application/json',
    'User-Agent': 'GoogleSecOps-SonraiCollector/1.0'
  }

  body = json.dumps({'query': query})

  backoff = 1.0
  max_retries = 3

  for attempt in range(max_retries):
    response = http.request('POST', endpoint, body=body.encode('utf-8'), headers=headers)

    if response.status == 429:
      retry_after = int(response.headers.get('Retry-After', str(int(backoff))))
      print(f"Rate limited (429). Retrying after {retry_after}s...")
      time.sleep(retry_after)
      backoff = min(backoff * 2, 30.0)
      continue

    if response.status != 200:
      print(f"HTTP Error: {response.status} - {response.data.decode('utf-8')}")
      return None

    data = json.loads(response.data.decode('utf-8'))

    if 'errors' in data:
      print(f"GraphQL errors: {json.dumps(data['errors'])}")
      return None

    return data.get('data')

  print(f"Failed after {max_retries} retries due to rate limiting")
  return None

def fetch_tickets(start_time: datetime, end_time: datetime, page_size: int, max_records: int):
  """
  Fetch tickets from Sonrai Security GraphQL API with pagination and rate limiting.
  """
  records = []
  newest_time = None
  page_num = 0
  offset = 0

  start_iso = start_time.strftime('%Y-%m-%dT%H:%M:%SZ')

  while True:
    page_num += 1

    if len(records) >= max_records:
      print(f"Reached max_records limit ({max_records})")
      break

    current_limit = min(page_size, max_records - len(records))

    query = f"""{{
              Tickets(
                where: {{createdDate: {{op: GT, value: \"{start_iso}\"}}}}
        limit: {current_limit}
        offset: {offset}
      ) {{
                count
                items {{
                  srn
                  title
                  severity
                  status
                  createdDate
                  updatedDate
                  resourceSrn
                  eventName
                  actionType
                  sourceIp
                  actor {{
                    profile {{
                      srn
                      name
                    }}
          }}
          message
        }}
      }}
    }}"""

    data = graphql_query(query)
    if not data:
      break

    tickets_data = data.get('Tickets', {})
    page_results = tickets_data.get('items', [])

    if not page_results:
      print(f"No more results (empty page)")
      break

    print(f"Page {page_num}: Retrieved {len(page_results)} tickets")
    records.extend(page_results)

    # Track newest event time
    for ticket in page_results:
      try:
        event_time = ticket.get('createdDate')
        if event_time:
          if newest_time is None or parse_datetime(event_time) > parse_datetime(newest_time):
            newest_time = event_time
      except Exception as e:
        print(f"Warning: Could not parse event time: {e}")

    if len(page_results) < current_limit:
      print(f"Reached last page (size={len(page_results)} < limit={current_limit})")
      break

    offset += len(page_results)

  print(f"Retrieved {len(records)} total records from {page_num} pages")
  return records, newest_time

2 つ目のファイル: requirements.txt:

functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0

[デプロイ] をクリックして、関数を保存してデプロイします。
デプロイが完了するまで待ちます（2 ～ 3 分）。

注: Pub/Sub トリガー構成では、必要なサブスクリプションと権限が自動的に作成されます。

Cloud Scheduler ジョブの作成

Cloud Scheduler は、定期的に Pub/Sub トピックにメッセージをパブリッシュし、Cloud Run functions の関数をトリガーします。

GCP Console で、[Cloud Scheduler] に移動します。
[ジョブを作成] をクリックします。

次の構成情報を提供してください。

設定	値
名前	`sonrai-collector-hourly`
リージョン	Cloud Run functions と同じリージョンを選択する
周波数	`0 * * * *`（1 時間ごと、毎正時）
タイムゾーン	タイムゾーンを選択（UTC を推奨）
ターゲットタイプ	Pub/Sub
トピック	トピック「`sonrai-trigger`」を選択します
メッセージ本文	`{}`（空の JSON オブジェクト）

[作成] をクリックします。

スケジュールの頻度オプション

ログの量とレイテンシの要件に基づいて頻度を選択します。

頻度	CRON 式	ユースケース
5 分おき	`/5 * * *`	大容量、低レイテンシ
15 分ごと	`/15 * * *`	検索量が普通
1 時間ごと	`0 * * * *`	標準（推奨）
6 時間ごと	`0 /6 * *`	少量、バッチ処理
毎日	`0 0 * * *`	過去のデータの収集

統合をテストする

Cloud Scheduler コンソールで、ジョブ（sonrai-collector-hourly）を見つけます。
[強制実行] をクリックして、手動でトリガーします。
数秒待ってから、[Cloud Run > [サービス] > [sonrai-collector] > [ログ] に移動します。

関数が正常に実行されたことを確認します。以下のものを探します。

Fetching tickets from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00
Page 1: Retrieved X tickets
Wrote X records to gs://sonrai-security-logs/sonrai/logs_YYYYMMDD_HHMMSS.ndjson
Successfully processed X records

GCS バケット（sonrai-security-logs）を調べて、ログが書き込まれたことを確認します。

ログにエラーが表示された場合:

HTTP 401: 環境変数で API キーを確認する
HTTP 403: Sonrai Security プラットフォームで API キーに必要な権限があることを確認する
HTTP 429: レート制限 - 関数はバックオフで自動的に再試行されます
GraphQL エラー: クエリ構文と使用可能なフィールドを確認する

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

[SIEM 設定] > [フィード] に移動します。
[Add New Feed] をクリックします。
[単一フィードを設定] をクリックします。
[フィード名] フィールドに、フィードの名前を入力します（例: Sonrai Security Logs）。
[ソースタイプ] として [Google Cloud Storage V2] を選択します。
[Log type] として [Sonrai Enterprise Cloud Security Solution] を選択します。
[サービスアカウントを取得する] をクリックします。一意のサービスアカウントのメールアドレスが表示されます（例:）。
```
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
```
このメールアドレスをコピーします。次のステップでこれを使用します。

注: 各 Google SecOps インスタンスには一意のサービスアカウントがあります。他のドキュメントや例のサービスアカウントは使用しないでください。
[次へ] をクリックします。
次の入力パラメータの値を指定します。
- ストレージバケットの URL: 接頭辞パスを含む GCS バケット URI を入力します。
```
gs://sonrai-security-logs/sonrai/
```
  - 次のように置き換えます。
    - sonrai-security-logs: GCS バケット名。
    - sonrai: ログが保存されるオプションの接頭辞/フォルダパス（ルートの場合は空のままにします）。
注: URI の末尾には常にスラッシュ（/）を含めてください。
- Source deletion option: 必要に応じて削除オプションを選択します。
  - なし: 転送後にファイルを削除しません（テストにおすすめ）。
  - 転送されたファイルを削除する: 転送が完了した後にファイルを削除します。
  - 転送されたファイルと空のディレクトリを削除する: 転送が完了した後にファイルと空のディレクトリを削除します。
    
    注: 削除オプションを選択する場合は、サービスアカウントに Storage オブジェクト閲覧者ではなく、Storage オブジェクト管理者のロールが必要です。必要に応じて IAM 権限を更新します。
- ファイルの最大経過日数: 過去の日数以内に変更されたファイルを含めます（デフォルトは 180 日）。
- アセットの名前空間: アセットの名前空間。
- Ingestion labels: このフィードのイベントに適用されるラベル。
[次へ] をクリックします。
[Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。

Google SecOps サービスアカウントに IAM 権限を付与する

Google SecOps サービスアカウントには、GCS バケットに対する Storage オブジェクト閲覧者のロールが必要です。

[Cloud Storage] > [バケット] に移動します。
バケット名（sonrai-security-logs）をクリックします。
[権限] タブに移動します。
[アクセス権を付与] をクリックします。
次の構成情報を指定します。
- プリンシパルを追加: Google SecOps サービスアカウントのメールアドレスを貼り付けます。
- ロールを割り当てる: [Storage オブジェクト閲覧者] を選択します。
[保存] をクリックします。

注: 削除オプションの [転送されたファイルを削除] または [転送されたファイルと空のディレクトリを削除] を使用する場合は、Storage オブジェクト閲覧者ではなく、Storage オブジェクト管理者ロールを付与します。

UDM マッピングテーブル

ログフィールド	UDM マッピング	ロジック
actor.profile	principal.resource.attribute.labels	リソースに関連付けられたラベル
actor.profile.srn	target.user.product_object_id	ユーザーのプロダクト固有のオブジェクト ID
actor.profile.srn	target.user.userid	ユーザー ID
actorSrn、variables、gql、type	additional.fields	UDM スキーマの他の場所でキャプチャされていない追加フィールド
actionType	security_result.action_details	実施した対応の詳細
createdDate	metadata.event_timestamp	イベントのタイムスタンプ
event_type	metadata.event_type	イベントのタイプ（例: USER_LOGIN、NETWORK_CONNECTION）
eventName	metadata.product_event_type	プロダクト固有のイベントタイプ
メッセージ	security_result.summary	セキュリティ結果の概要
metadata	metadata	イベントに関するメタデータ
プリンシパル	プリンシパル	プリンシパル（アクター）に関する情報
resourceSrn	principal.resource.name	リソースの名前
security_result	security_result	セキュリティ結果情報
sourceIp	principal.asset.ip	アセットの IP アドレス
sourceIp	principal.ip	プリンシパルに関連付けられた IP アドレス
ターゲット	ターゲット	ターゲットに関する情報
	extensions.auth.type	イベントで使用される認証タイプ
	metadata.product_name	製品名
	metadata.vendor_name	ベンダー/会社名

変更ログ

このパーサーの変更ログを表示する

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