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 アクセス トークンを生成する

  1. システム管理者アカウントで Smartsheet アカウントにログインします。
  2. 左側のナビゲーション バーの下部にある [アカウント](プロフィール画像)を選択し、[個人設定] に移動します。
  3. [API Access] タブに移動します。
  4. [新しいアクセス トークンを生成] をクリックします。
  5. トークンの名前を入力します(例: SecOps SIEM Integration)。
  6. [OK] をクリックします。
  7. アクセス トークンをコピーして安全に保存します。

権限を確認する

アカウントに必要な権限があることを確認するには:

  1. Smartsheet アカウントにログインします。
  2. [アカウント](プロフィール画像)> [個人設定] > [API アクセス] に移動します。
  3. [Manage API Access Tokens] ページが表示され、トークンを生成できる場合は、必要な権限が付与されています。
  4. これらのオプションにアクセスできない場合は、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 バケットを作成する

  1. Google Cloud コンソールに移動します。
  2. プロジェクトを選択するか、新しいプロジェクトを作成します。
  3. ナビゲーション メニューで、[Cloud Storage > バケット] に移動します。
  4. [バケットを作成] をクリックします。
  5. 次の構成情報を提供してください。

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

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

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

サービス アカウントを作成する

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

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

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

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

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

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

Pub/Sub トピックの作成

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

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

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

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

  1. GCP Console で、[Cloud Run] に移動します。
  2. [サービスを作成] をクリックします。
  3. [関数] を選択します(インライン エディタを使用して関数を作成します)。
  4. [構成] セクションで、次の構成の詳細を指定します。

    設定 値
    サービス名 smartsheet-logs-to-gcs
    リージョン GCS バケットと一致するリージョンを選択します(例: us-central1)。
    ランタイム [Python 3.12] 以降を選択します。
  5. [トリガー(省略可)] セクションで、次の操作を行います。

    1. [+ トリガーを追加] をクリックします。
    2. [Cloud Pub/Sub] を選択します。
    3. [Cloud Pub/Sub トピックを選択してください] で、smartsheet-logs-trigger を選択します。
    4. [保存] をクリックします。
  6. [認証] セクションで、次の操作を行います。

    1. [認証が必要] を選択します。
    2. Identity and Access Management(IAM)を確認します。
  7. 下にスクロールして、[コンテナ、ネットワーキング、セキュリティ] を開きます。

  8. [セキュリティ] タブに移動します。

    • サービス アカウント: smartsheet-logs-sa を選択します。
  9. [コンテナ] タブに移動します。

    1. [変数とシークレット] をクリックします。
    2. 環境変数ごとに [+ 変数を追加] をクリックします。
    変数名 値の例
    GCS_BUCKET smartsheet-audit-logs
    GCS_PREFIX smartsheet/events/
    STATE_KEY smartsheet/events/state.json
    SMARTSHEET_TOKEN <your-smartsheet-access-token>
    MAX_COUNT 1000
    TIMEOUT 30
  10. [変数とシークレット] セクションで、[リクエスト] まで下にスクロールします。

    • リクエストのタイムアウト: 600 秒(10 分)を入力します。
  11. [設定] タブに移動します。

    • [リソース] セクションで次の操作を行います。
      • メモリ: 512 MiB 以上を選択します。
      • CPU: [1] を選択します。
  12. [リビジョン スケーリング] セクションで、次の操作を行います。

    • [インスタンスの最小数] に「0」と入力します。
    • インスタンスの最大数: 100 と入力します(または、想定される負荷に基づいて調整します)。
  13. [作成] をクリックします。

  14. サービスが作成されるまで待ちます(1 ~ 2 分)。

  15. サービスを作成すると、インライン コードエディタが自動的に開きます。

関数のコードを追加する

  1. [関数のエントリ ポイント] に「main」と入力します。
  2. インライン コードエディタで、次の 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_available
      
    • 2 つ目のファイル: requirements.txt:

      functions-framework==3.*
      google-cloud-storage==2.*
      urllib3>=2.0.0
      
  3. [デプロイ] をクリックして、関数を保存してデプロイします。

  4. デプロイが完了するまで待ちます(2 ~ 3 分)。

Cloud Scheduler ジョブを作成する

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

  1. GCP Console で、[Cloud Scheduler] に移動します。
  2. [ジョブを作成] をクリックします。
  3. 次の構成情報を提供してください。

    設定 値
    名前 smartsheet-logs-schedule-15min
    リージョン Cloud Run functions と同じリージョンを選択する
    周波数 */15 * * * *(15 分ごと)
    タイムゾーン タイムゾーンを選択(UTC を推奨)
    ターゲット タイプ Pub/Sub
    トピック smartsheet-logs-trigger を選択
    メッセージ本文 {}(空の JSON オブジェクト)
  4. [作成] をクリックします。

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

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

頻度 CRON 式 ユースケース
5 分おき */5 * * * * 大容量、低レイテンシ
15 分ごと */15 * * * * 標準(推奨)
1 時間ごと 0 * * * * 音量が小さい
6 時間ごと 0 */6 * * * バッチ処理

統合をテストする

  1. Cloud Scheduler コンソールで、ジョブ(smartsheet-logs-schedule-15min など)を見つけます。
  2. [強制実行] をクリックして、ジョブを手動でトリガーします。
  3. 数秒待ちます。
  4. Cloud Run > [サービス] に移動します。
  5. 関数名(smartsheet-logs-to-gcs)をクリックします。
  6. [ログ] タブをクリックします。
  7. 関数が正常に実行されたことを確認します。以下のものを探します。

    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
    
  8. [Cloud Storage] > [バケット] に移動します。

  9. バケット名(smartsheet-audit-logs)をクリックします。

  10. 接頭辞フォルダ(smartsheet/events/)に移動します。

  11. 現在のタイムスタンプで新しい .json.gz ファイルが作成されたことを確認します。

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

  • HTTP 401: 環境変数で Smartsheet アクセス トークンを確認する
  • HTTP 403: アカウントにシステム管理者権限があり、イベント レポートが有効になっていることを確認します
  • HTTP 429: レート制限 - 関数はバックオフで自動的に再試行されます
  • 環境変数がありません: 必要な変数がすべて Cloud Run functions の構成で設定されていることを確認します

Google SecOps サービス アカウントを取得する

Google SecOps は、一意のサービス アカウントを使用して GCS バケットからデータを読み取ります。このサービス アカウントにバケットへのアクセス権を付与する必要があります。

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

  1. [SIEM 設定] > [フィード] に移動します。
  2. [Add New Feed] をクリックします。
  3. [単一フィードを設定] をクリックします。
  4. [フィード名] フィールドに、フィードの名前を入力します(例: Smartsheet Audit Logs)。
  5. [ソースタイプ] として [Google Cloud Storage V2] を選択します。
  6. [ログタイプ] として [Smartsheet] を選択します。
  7. [サービス アカウントを取得する] をクリックします。一意のサービス アカウントのメールアドレスが表示されます(例:)。

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
    
  8. このメールアドレスをコピーします。次のステップでこれを使用します。

  9. [次へ] をクリックします。

  10. 次の入力パラメータの値を指定します。

    • ストレージ バケットの URL: 接頭辞パスを含む GCS バケット URI を入力します。
    gs://smartsheet-audit-logs/smartsheet/events/
    
    • 次のように置き換えます。

      • smartsheet-audit-logs: GCS バケット名。
      • smartsheet/events/: ログが保存される接頭辞パス。
    • Source deletion option: 必要に応じて削除オプションを選択します。
      • なし: 転送後にファイルを削除しません(テストにおすすめ)。
      • 転送されたファイルを削除する: 転送が完了した後にファイルを削除します。
      • 転送されたファイルと空のディレクトリを削除する: 転送が完了した後にファイルと空のディレクトリを削除します。
    • ファイルの最大経過日数: 指定した日数以内に変更されたファイルを含めます。デフォルトは 180 日です。
    • アセットの名前空間: アセットの名前空間。
    • Ingestion labels: このフィードのイベントに適用されるラベル。
  11. [次へ] をクリックします。

  12. [Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。

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

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

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

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 のプロフェッショナルから回答を得ることができます。