Smartsheet のログを収集する
このドキュメントでは、Google Cloud Storage を使用して Smartsheet ログを Google Security Operations に取り込む方法について説明します。
Smartsheet は、エンタープライズ チーム向けにスプレッドシートのようなプロジェクト管理、タスク トラッキング、ワークフローの自動化を提供する共同作業管理プラットフォームです。Event Reporting API は、Smartsheet 組織全体にわたるユーザー アクション、データアクセス、共有の変更、管理オペレーションなど、100 種類以上のイベントタイプを対象とする監査ログを提供します。
始める前に
次の前提条件を満たしていることを確認してください。
- Google SecOps インスタンス
- Cloud Storage API が有効になっている GCP プロジェクト
- GCS バケットを作成および管理する権限
- GCS バケットの IAM ポリシーを管理する権限
- Cloud Run functions、Pub/Sub トピック、Cloud Scheduler ジョブを作成する権限
- イベント レポート アドオンが有効になっている Smartsheet Enterprise プラン
- API アクセス権を持つ Smartsheet システム管理者アカウント
Smartsheet API 認証情報を収集する
API アクセス トークンを生成する
- システム管理者アカウントで Smartsheet アカウントにログインします。
- 左側のナビゲーション バーの下部にある [アカウント](プロフィール画像)を選択し、[個人設定] に移動します。
- [API Access] タブに移動します。
- [新しいアクセス トークンを生成] をクリックします。
- トークンの名前を入力します(例:
SecOps SIEM Integration)。 - [OK] をクリックします。
アクセス トークンをコピーして安全に保存します。
権限を確認する
アカウントに必要な権限があることを確認するには:
- Smartsheet アカウントにログインします。
- [アカウント](プロフィール画像)> [個人設定] > [API アクセス] に移動します。
- [Manage API Access Tokens] ページが表示され、トークンを生成できる場合は、必要な権限が付与されています。
- これらのオプションにアクセスできない場合は、Smartsheet システム管理者に連絡して API アクセス権を付与してもらってください。
テスト API へのアクセス
統合に進む前に、認証情報をテストします。
# Replace with your actual access token SMARTSHEET_TOKEN="<your-access-token>" # Test Event Reporting API access curl -v -H "Authorization: Bearer ${SMARTSHEET_TOKEN}" \ "https://api.smartsheet.com/2.0/events?since=$(date -u -d '1 hour ago' +%Y-%m-%dT%H:%M:%SZ)&maxCount=10"
次のエラーが表示された場合:
- HTTP 401: アクセス トークンが正しいことを確認します。
- HTTP 403: アカウントにシステム管理者権限があり、プランでイベント レポート アドオンが有効になっていることを確認します。
Google Cloud Storage バケットを作成する
- Google Cloud コンソールに移動します。
- プロジェクトを選択するか、新しいプロジェクトを作成します。
- ナビゲーション メニューで、[Cloud Storage > バケット] に移動します。
- [バケットを作成] をクリックします。
次の構成情報を提供してください。
設定 値 バケットに名前を付ける グローバルに一意の名前( smartsheet-audit-logsなど)を入力します。ロケーション タイプ ニーズに応じて選択します(リージョン、デュアルリージョン、マルチリージョン)。 ロケーション 場所を選択します(例: us-central1)。ストレージ クラス Standard(アクセス頻度の高いログにおすすめ) アクセス制御 均一(推奨) 保護ツール 省略可: オブジェクトのバージョニングまたは保持ポリシーを有効にする [作成] をクリックします。
Cloud Run functions のサービス アカウントを作成する
Cloud Run functions には、GCS バケットへの書き込み権限があり、Pub/Sub によって呼び出されるサービス アカウントが必要です。
サービス アカウントを作成する
- GCP コンソールで、[IAM と管理>サービス アカウント] に移動します。
- [サービス アカウントを作成] をクリックします。
- 次の構成情報を指定します。
- サービス アカウント名: 「
smartsheet-logs-sa」と入力します。 - サービス アカウントの説明: 「
Service account for Cloud Run function to collect Smartsheet audit logs」と入力します。
- サービス アカウント名: 「
- [作成して続行] をクリックします。
- [このサービス アカウントにプロジェクトへのアクセスを許可する] セクションで、次のロールを追加します。
- [ロールを選択] をクリックします。
- [ストレージ オブジェクト管理者] を検索して選択します。
- [+ 別のロールを追加] をクリックします。
- [Cloud Run 起動元] を検索して選択します。
- [+ 別のロールを追加] をクリックします。
- [Cloud Functions 起動元] を検索して選択します。
- [続行] をクリックします。
- [完了] をクリックします。
これらのロールは、次の操作に必要です。
- Storage オブジェクト管理者: ログを GCS バケットに書き込み、状態ファイルを管理する
- Cloud Run 起動元: Pub/Sub が関数を呼び出すことを許可します。
- Cloud Functions 起動元: 関数の呼び出しを許可する
GCS バケットに対する IAM 権限を付与する
GCS バケットに対する書き込み権限をサービス アカウントに付与します。
- [Cloud Storage] > [バケット] に移動します。
- バケット名(
smartsheet-audit-logsなど)をクリックします。 - [権限] タブに移動します。
- [アクセス権を付与] をクリックします。
- 次の構成情報を指定します。
- プリンシパルを追加: サービス アカウントのメールアドレス(例:
smartsheet-logs-sa@PROJECT_ID.iam.gserviceaccount.com)を入力します。 - ロールを割り当てる: [ストレージ オブジェクト管理者] を選択します。
- プリンシパルを追加: サービス アカウントのメールアドレス(例:
- [保存] をクリックします。
Pub/Sub トピックの作成
Cloud Scheduler がパブリッシュし、Cloud Run 関数がサブスクライブする Pub/Sub トピックを作成します。
- GCP Console で、[Pub/Sub> トピック] に移動します。
- [トピックを作成] をクリックします。
- 次の構成情報を指定します。
- トピック ID: 「
smartsheet-logs-trigger」と入力します。 - その他の設定はデフォルトのままにします。
- トピック ID: 「
- [作成] をクリックします。
ログを収集する Cloud Run 関数を作成する
Cloud Run 関数は、Cloud Scheduler からの Pub/Sub メッセージによってトリガーされ、Smartsheet Event Reporting API からログを取得して GCS に書き込みます。
- GCP Console で、[Cloud Run] に移動します。
- [サービスを作成] をクリックします。
- [関数] を選択します(インライン エディタを使用して関数を作成します)。
[構成] セクションで、次の構成の詳細を指定します。
設定 値 サービス名 smartsheet-logs-to-gcsリージョン GCS バケットと一致するリージョンを選択します(例: us-central1)。ランタイム [Python 3.12] 以降を選択します。 [トリガー(省略可)] セクションで、次の操作を行います。
- [+ トリガーを追加] をクリックします。
- [Cloud Pub/Sub] を選択します。
- [Cloud Pub/Sub トピックを選択してください] で、
smartsheet-logs-triggerを選択します。 - [保存] をクリックします。
[認証] セクションで、次の操作を行います。
- [認証が必要] を選択します。
- Identity and Access Management(IAM)を確認します。
下にスクロールして、[コンテナ、ネットワーキング、セキュリティ] を開きます。
[セキュリティ] タブに移動します。
- サービス アカウント:
smartsheet-logs-saを選択します。
- サービス アカウント:
[コンテナ] タブに移動します。
- [変数とシークレット] をクリックします。
- 環境変数ごとに [+ 変数を追加] をクリックします。
変数名 値の例 GCS_BUCKETsmartsheet-audit-logsGCS_PREFIXsmartsheet/events/STATE_KEYsmartsheet/events/state.jsonSMARTSHEET_TOKEN<your-smartsheet-access-token>MAX_COUNT1000TIMEOUT30[変数とシークレット] セクションで、[リクエスト] まで下にスクロールします。
- リクエストのタイムアウト:
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 import uuid import gzip import io # 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', 'smartsheet/events/') STATE_KEY = os.environ.get('STATE_KEY', 'smartsheet/events/state.json') SMARTSHEET_TOKEN = os.environ.get('SMARTSHEET_TOKEN') MAX_COUNT = int(os.environ.get('MAX_COUNT', '1000')) TIMEOUT = int(os.environ.get('TIMEOUT', '30')) EVENTS_URL = "https://api.smartsheet.com/2.0/events" @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch Smartsheet Event Reporting audit logs and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ if not all([GCS_BUCKET, SMARTSHEET_TOKEN]): print('Error: Missing required environment variables') return try: bucket = storage_client.bucket(GCS_BUCKET) # Load state state = load_state(bucket, STATE_KEY) stream_position = state.get('stream_position') print(f'Fetching events from stream position: {stream_position or "latest"}') # Fetch events total_written = 0 has_more = True while has_more: events, new_position, more = fetch_events(stream_position) if events: write_chunk(bucket, events, datetime.now(timezone.utc)) total_written += len(events) if new_position: stream_position = new_position has_more = more and total_written < 50000 # Save state state['stream_position'] = stream_position save_state(bucket, STATE_KEY, state) print(f'Successfully processed {total_written} events') 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: {str(e)}') return {} def save_state(bucket, key, state): """Save state to GCS.""" try: state['updated_at'] = datetime.now(timezone.utc).isoformat() blob = bucket.blob(key) blob.upload_from_string( json.dumps(state), content_type='application/json' ) except Exception as e: print(f'Warning: Could not save state: {str(e)}') def write_chunk(bucket, items, ts): """Write log chunk to GCS as compressed NDJSON.""" key = f"{GCS_PREFIX}{ts:%Y/%m/%d}/smartsheet-events-{uuid.uuid4()}.json.gz" buf = io.BytesIO() with gzip.GzipFile(fileobj=buf, mode='w') as gz: for rec in items: gz.write((json.dumps(rec) + '\n').encode('utf-8')) buf.seek(0) blob = bucket.blob(key) blob.upload_from_file(buf, content_type='application/gzip') print(f'Wrote {len(items)} events to {key}') return key def fetch_events(stream_position): """ Fetch events from Smartsheet Event Reporting API. The API uses a streaming model with streamPosition for pagination. On first call (no streamPosition), it returns the current position without events. Subsequent calls return events since the position. Returns: Tuple of (events list, new stream position, has more data) """ headers = { 'Authorization': f'Bearer {SMARTSHEET_TOKEN}', 'Accept': 'application/json' } params = [f'maxCount={MAX_COUNT}'] if stream_position: params.append(f'streamPosition={stream_position}') url = f"{EVENTS_URL}?{'&'.join(params)}" response = http.request( 'GET', url, headers=headers, timeout=TIMEOUT ) if response.status == 429: retry_after = int(response.headers.get('Retry-After', '60')) print(f'Rate limited (429). Retry-After: {retry_after}s') import time time.sleep(min(retry_after, 120)) return fetch_events(stream_position) if response.status != 200: print(f'API request failed: {response.status}') response_text = response.data.decode('utf-8') print(f'Response body: {response_text}') raise Exception(f'Failed to fetch events: {response.status}') data = json.loads(response.data.decode('utf-8')) events = data.get('data', []) or [] new_position = data.get('nextStreamPosition') more_available = data.get('moreEventsAvailable', False) if events: print(f'Retrieved {len(events)} events') return events, new_position, more_available2 つ目のファイル: requirements.txt:
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0
[デプロイ] をクリックして、関数を保存してデプロイします。
デプロイが完了するまで待ちます(2 ~ 3 分)。
Cloud Scheduler ジョブを作成する
Cloud Scheduler は、定期的に Pub/Sub トピックにメッセージをパブリッシュし、Cloud Run functions の関数をトリガーします。
- GCP Console で、[Cloud Scheduler] に移動します。
- [ジョブを作成] をクリックします。
次の構成情報を提供してください。
設定 値 名前 smartsheet-logs-schedule-15minリージョン Cloud Run functions と同じリージョンを選択する 周波数 */15 * * * *(15 分ごと)タイムゾーン タイムゾーンを選択(UTC を推奨) ターゲット タイプ Pub/Sub トピック smartsheet-logs-triggerを選択メッセージ本文 {}(空の JSON オブジェクト)[作成] をクリックします。
スケジュールの頻度オプション
ログの量とレイテンシの要件に基づいて頻度を選択します。
| 頻度 | CRON 式 | ユースケース |
|---|---|---|
| 5 分おき | */5 * * * * |
大容量、低レイテンシ |
| 15 分ごと | */15 * * * * |
標準(推奨) |
| 1 時間ごと | 0 * * * * |
音量が小さい |
| 6 時間ごと | 0 */6 * * * |
バッチ処理 |
統合をテストする
- Cloud Scheduler コンソールで、ジョブ(
smartsheet-logs-schedule-15minなど)を見つけます。 - [強制実行] をクリックして、ジョブを手動でトリガーします。
- 数秒待ちます。
- Cloud Run > [サービス] に移動します。
- 関数名(
smartsheet-logs-to-gcs)をクリックします。 - [ログ] タブをクリックします。
関数が正常に実行されたことを確認します。以下のものを探します。
Fetching events from stream position: <position> Retrieved X events Wrote X events to smartsheet/events/YYYY/MM/DD/smartsheet-events-UUID.json.gz Successfully processed X events[Cloud Storage] > [バケット] に移動します。
バケット名(
smartsheet-audit-logs)をクリックします。接頭辞フォルダ(
smartsheet/events/)に移動します。現在のタイムスタンプで新しい
.json.gzファイルが作成されたことを確認します。
ログにエラーが表示された場合:
- HTTP 401: 環境変数で Smartsheet アクセス トークンを確認する
- HTTP 403: アカウントにシステム管理者権限があり、イベント レポートが有効になっていることを確認します
- HTTP 429: レート制限 - 関数はバックオフで自動的に再試行されます
- 環境変数がありません: 必要な変数がすべて Cloud Run functions の構成で設定されていることを確認します
Google SecOps サービス アカウントを取得する
Google SecOps は、一意のサービス アカウントを使用して GCS バケットからデータを読み取ります。このサービス アカウントにバケットへのアクセス権を付与する必要があります。
Smartsheet のログを取り込むように Google SecOps でフィードを構成する
- [SIEM 設定] > [フィード] に移動します。
- [Add New Feed] をクリックします。
- [単一フィードを設定] をクリックします。
- [フィード名] フィールドに、フィードの名前を入力します(例:
Smartsheet Audit Logs)。 - [ソースタイプ] として [Google Cloud Storage V2] を選択します。
- [ログタイプ] として [Smartsheet] を選択します。
[サービス アカウントを取得する] をクリックします。一意のサービス アカウントのメールアドレスが表示されます(例:)。
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.comこのメールアドレスをコピーします。次のステップでこれを使用します。
[次へ] をクリックします。
次の入力パラメータの値を指定します。
- ストレージ バケットの URL: 接頭辞パスを含む GCS バケット URI を入力します。
gs://smartsheet-audit-logs/smartsheet/events/次のように置き換えます。
smartsheet-audit-logs: GCS バケット名。smartsheet/events/: ログが保存される接頭辞パス。
- Source deletion option: 必要に応じて削除オプションを選択します。
- なし: 転送後にファイルを削除しません(テストにおすすめ)。
- 転送されたファイルを削除する: 転送が完了した後にファイルを削除します。
- 転送されたファイルと空のディレクトリを削除する: 転送が完了した後にファイルと空のディレクトリを削除します。
- ファイルの最大経過日数: 指定した日数以内に変更されたファイルを含めます。デフォルトは 180 日です。
- アセットの名前空間: アセットの名前空間。
- Ingestion labels: このフィードのイベントに適用されるラベル。
[次へ] をクリックします。
[Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。
Google SecOps サービス アカウントに IAM 権限を付与する
Google SecOps サービス アカウントには、GCS バケットに対する Storage オブジェクト閲覧者のロールが必要です。
- [Cloud Storage] > [バケット] に移動します。
- バケット名(
smartsheet-audit-logsなど)をクリックします。 - [権限] タブに移動します。
- [アクセス権を付与] をクリックします。
- 次の構成情報を指定します。
- プリンシパルを追加: Google SecOps サービス アカウントのメールアドレスを貼り付けます。
- ロールを割り当てる: [Storage オブジェクト閲覧者] を選択します。
[保存] をクリックします。
UDM マッピング テーブル
| ログフィールド | UDM マッピング | ロジック |
|---|---|---|
accessLevel_label |
additional.fields |
統合済み |
appName_label |
additional.fields |
統合済み |
attachmentName_label |
additional.fields |
統合済み |
cellLinkSourceSheetId_label |
additional.fields |
統合済み |
dashboardName_label |
additional.fields |
統合済み |
folderName_label |
additional.fields |
統合済み |
formatType_label |
additional.fields |
統合済み |
includeAttachments_label |
additional.fields |
統合済み |
includeDiscussions_label |
additional.fields |
統合済み |
mergeType_label |
additional.fields |
統合済み |
rowCount_label |
additional.fields |
統合済み |
rowsMoved_label |
additional.fields |
統合済み |
sheetId_label |
additional.fields |
統合済み |
sheetName_label |
additional.fields |
統合済み |
sheetRowId_label |
additional.fields |
統合済み |
sourceFolderId_label |
additional.fields |
統合済み |
sourceObjectId_label |
additional.fields |
統合済み |
sourceSheetId_label |
additional.fields |
統合済み |
sourceType_label |
additional.fields |
統合済み |
tokenDisplayValue_label |
additional.fields |
統合済み |
tokenUserId_label |
additional.fields |
統合済み |
userId_label |
additional.fields |
統合済み |
workspaceId_label |
additional.fields |
統合済み |
additionalDetails.accessScopes |
metadata.description |
直接マッピングされます。 |
additionalDetails_tokenExpirationTimestamp |
metadata.event_timestamp |
yyyy-MM-ddTHH:mm:ssZ として解析済み |
eventTimestamp |
metadata.event_timestamp |
yyyy-MM-ddTHH:mm:ssZ として解析済み |
has_principal |
metadata.event_type |
マッピング: true → USER_UNCATEGORIZED |
accessTokenName |
metadata.product_log_id |
直接マッピングされます。 |
eventId |
metadata.product_log_id |
直接マッピングされます。 |
additionalDetails.appClientId |
principal.user.userid |
直接マッピングされます。 |
requestUserId |
principal.user.userid |
直接マッピングされます。 |
object_id |
security_result.about.labels |
統合済み |
source_label |
security_result.about.labels |
統合済み |
object_type |
security_result.about.resource.attribute.labels |
統合済み |
action |
security_result.action_details |
直接マッピングされます。 |
userId |
target.user.userid |
直接マッピングされます。 |
| なし | metadata.event_type |
定数: USER_UNCATEGORIZED |
変更ログ
さらにサポートが必要な場合 コミュニティ メンバーや Google SecOps のプロフェッショナルから回答を得ることができます。