Workday ユーザー アクティビティ ログを収集する

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

このドキュメントでは、Cloud Storage V2 を使用して Workday ユーザー アクティビティ ログを Google Security Operations に取り込む方法について説明します。

Workday は、ユーザー アクティビティ、認証、セキュリティ監査ログを生成するクラウドベースの人材管理(HCM)および財務プラットフォームです。ユーザー アクティビティ ロギング REST API を使用すると、これらのログにプログラムでアクセスできます。

始める前に

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

  • Google SecOps インスタンス
  • {storage_api} が有効になっている Google Cloud プロジェクト
  • Cloud Storage バケットを作成して管理する権限
  • Cloud Storage バケットで Identity and Access Management(IAM)ポリシーを管理する権限
  • Cloud Run サービス、Pub/Sub トピック、Cloud Scheduler ジョブを作成する権限
  • セキュリティ管理者または同等の権限による Workday への特権アクセス
  • Workday テナントでユーザー アクティビティ ロギングが有効になっている

Cloud Storage バケットを作成する

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

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

Workday API 認証情報を収集する

ユーザー アクティビティ ロギングを有効にする

  1. セキュリティ管理者として Workday テナントにログインします。
  2. Workday の検索バーで、[Edit Tenant Setup - System] タスクを検索して開きます。
  3. [セキュリティ] セクションで、[ユーザー アクティビティのロギングを有効にする] チェックボックスをオンにします。
  4. [OK] をクリックして変更を保存します。

統合システム ユーザー(ISU)を作成する

  1. Workday の検索バーで、「統合システム ユーザーを作成」を検索します。
  2. [ユーザー名](ISU_SIEM_Export など)と強力な [パスワード] を入力します。
  3. [次回ログイン時に新しいパスワードを要求する] チェックボックスをオフにします。
  4. [OK] をクリックします。
  5. [Create Security Group] を検索し、[統合システム セキュリティ グループ(制限なし)] を作成します。
  6. このセキュリティ グループに ISU(ISU_SIEM_Export)を追加します。
  7. [View Domain] を検索し、[User Activity Logging] ドメインを見つけます。このドメインの ISU セキュリティ グループに Get アクセス権を付与します。
  8. [保留中のセキュリティ ポリシーの変更を有効にする] を検索して、変更を有効にします。

OAuth 用に API クライアントを登録する

  1. Workday の検索バーで、「Register API Client for Integrations」を検索します。
  2. [クライアント名](例: SIEM_OAuth_Client)を入力します。
  3. [Non-Expiring Refresh Tokens] オプションを選択します。
  4. スコープ System(または User Activity Logging API に必要なスコープ)を追加します。
  5. [OK] をクリックします。
  6. 次の詳細をコピーして安全な場所に保存します。
    • クライアント ID: API クライアント識別子。
    • Client Secret: API クライアント シークレット。
  7. [Manage Refresh Tokens for Integrations] を検索します。
  8. ISU アカウント(ISU_SIEM_Export)を選択します。
  9. API クライアントの新しい更新トークンを生成します。
  10. [Refresh Token] の値をコピーして保存します。

テナント URL を確認する

API エンドポイントは、Workday テナントに基づいています。書式は次のとおりです。

コンポーネント 値
トークン エンドポイント https://HOST/ccx/oauth2/TENANT/token
Activity Logging API https://HOST/ccx/api/privacy/v1/TENANT/activityLogging

権限を確認する

ISU に必要な権限があることを確認するには:

  1. セキュリティ管理者として Workday にログインします。
  2. [統合システム ユーザーを表示] を検索し、ISU(ISU_SIEM_Export)を選択します。
  3. ユーザーが、ユーザー アクティビティ ロギング ドメインへの Get アクセス権を持つセキュリティ グループのメンバーであることを確認します。
  4. アクセス権がない場合は、セキュリティ グループを更新して、保留中のセキュリティ ポリシーの変更を有効にします。

テスト API へのアクセス

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

    ```bash
    # Replace with your actual credentials
    CLIENT_ID="your-client-id"
    CLIENT_SECRET="your-client-secret"
    REFRESH_TOKEN="your-refresh-token"
    HOST="your-workday-host"
    TENANT="your-tenant"
    
    # Get OAuth access token
    TOKEN=$(curl -s -X POST "https://${HOST}/ccx/oauth2/${TENANT}/token" \
        -H "Content-Type: application/x-www-form-urlencoded" \
        -d "grant_type=refresh_token&client_id=${CLIENT_ID}&client_secret=${CLIENT_SECRET}&refresh_token=${REFRESH_TOKEN}" \
        | jq -r '.access_token')
    
    # Test API access
    curl -v -H "Authorization: Bearer ${TOKEN}" \
        "https://${HOST}/ccx/api/privacy/v1/${TENANT}/activityLogging?from=2024-01-01T00:00:00Z&to=2024-01-01T01:00:00Z"
    ```
    

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

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

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

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

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

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

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

次のように、Cloud Storage バケットに対する書き込み権限をサービス アカウントに付与します。

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

Pub/Sub トピックの作成

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

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

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

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

  1. GCP Console で、[Cloud Run] に移動します。
  2. [関数を作成] をクリックします。
  3. [構成] セクションで、次の構成の詳細を指定します。

    設定 値
    サービス名 workday-activity-collector
    リージョン Cloud Storage バケットと一致するリージョンを選択します(例: us-central1)。
    ランタイム [Python 3.12] 以降を選択します。
  4. [トリガー] セクションで、[トリガーを追加] をクリックして [Pub/Sub トリガー] を選択します。

  5. [Eventarc トリガー] ペインで、次の構成の詳細を指定します。

    • トリガー名: 生成された名前をそのまま使用するか、トリガーの名前を入力します。
    • トリガーのタイプ: [Google のソース] を選択します。
    • イベント プロバイダ: [Pub/Sub] を選択します。
    • イベントタイプ: [google.cloud.pubsub.topic.v1.messagePublished] を選択します。
    • Cloud Pub/Sub トピックを選択してください: トピック workday-activity-trigger を選択します。
    • リージョン: 関数と同じリージョンを選択します。
    • サービス アカウント: サービス アカウント workday-activity-collector-sa を選択します。
  6. [トリガーを保存] をクリックします。

  7. [認証] セクションで、次の操作を行います。

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

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

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

    1. [変数とシークレット] をクリックします。
    2. 環境変数ごとに [+ 変数を追加] をクリックします。
    変数名 値の例 説明
    GCS_BUCKET workday-activity-logs Cloud Storage バケットの名前
    GCS_PREFIX workday ログファイルの接頭辞
    STATE_KEY workday-state.json ログの接頭辞以外の状態パス
    WORKDAY_HOST your-workday-host Workday のホスト名
    WORKDAY_TENANT your-tenant Workday テナント名
    CLIENT_ID your-client-id OAuth クライアント ID
    CLIENT_SECRET your-client-secret OAuth クライアント シークレット
    REFRESH_TOKEN your-refresh-token OAuth 更新トークン
    MAX_RECORDS 10000 実行あたりの最大レコード数
    PAGE_SIZE 100 1 ページあたりのレコード数
    LOOKBACK_HOURS 24 初期ルックバック期間
  11. [変数とシークレット] タブで、[リクエスト] に移動します。

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

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

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

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

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

関数のコードを追加する

  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, timedelta
      import time
      import hashlib
      
      # 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', 'workday')
      STATE_KEY = os.environ.get('STATE_KEY', 'workday-state.json')
      WORKDAY_HOST = os.environ.get('WORKDAY_HOST', '')
      WORKDAY_TENANT = os.environ.get('WORKDAY_TENANT', '')
      CLIENT_ID = os.environ.get('CLIENT_ID', '')
      CLIENT_SECRET = os.environ.get('CLIENT_SECRET', '')
      REFRESH_TOKEN = os.environ.get('REFRESH_TOKEN', '')
      MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '10000'))
      PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '100'))
      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)
      
      # Activity Logging entries carry no per-event identifier (taskId and
      # sessionId are reused), so the content hash is the identity.
      ID_FIELDS = ()
      
      def event_keys(event):
          """Return every identity this event can be recognised by."""
          keys = {'sha256:' + hashlib.sha256(
              json.dumps(event, sort_keys=True, ensure_ascii=False).encode('utf-8')
          ).hexdigest()}
          for field in ID_FIELDS:
              value = event.get(field)
              if value:
                  keys.add(f'{field}:{value}')
          return keys
      
      def get_token():
          """Get OAuth 2.0 access token using refresh token flow."""
          token_url = f"https://{WORKDAY_HOST}/ccx/oauth2/{WORKDAY_TENANT}/token"
      
          body = f"grant_type=refresh_token&client_id={CLIENT_ID}&client_secret={CLIENT_SECRET}&refresh_token={REFRESH_TOKEN}"
      
          headers = {
              'Content-Type': 'application/x-www-form-urlencoded',
              'Accept': 'application/json'
          }
      
          backoff = 1.0
          max_retries = 3
      
          for attempt in range(max_retries):
              response = http.request('POST', token_url, 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) on token request. Retrying after {retry_after}s...")
                  time.sleep(retry_after)
                  backoff = min(backoff * 2, 30.0)
                  continue
      
              if response.status != 200:
                  raise RuntimeError(f"Failed to get access token: {response.status} - {response.data.decode('utf-8')}")
      
              data = json.loads(response.data.decode('utf-8'))
              return data['access_token']
      
          raise RuntimeError(f"Failed to get token after {max_retries} retries due to rate limiting")
      
      @functions_framework.cloud_event
      def main(cloud_event):
          """
          Cloud Run function triggered by Pub/Sub to fetch Workday User Activity logs and write to GCS.
      
          Args:
              cloud_event: CloudEvent object containing Pub/Sub message
          """
      
          if not all([GCS_BUCKET, WORKDAY_HOST, WORKDAY_TENANT, CLIENT_ID, CLIENT_SECRET, REFRESH_TOKEN]):
              print('Error: Missing required environment variables')
              return
      
          try:
              bucket = storage_client.bucket(GCS_BUCKET)
      
              # Load state
              state = load_state(bucket, STATE_KEY)
              seen_keys = load_seen(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 logs from {last_time.isoformat()} to {now.isoformat()}")
      
              # Get access token
              token = get_token()
      
              # Fetch logs
              records, newest_event_time = fetch_logs(
                  token=token,
                  start_time=last_time,
                  end_time=now,
                  page_size=PAGE_SIZE,
                  max_records=MAX_RECORDS,
              )
      
              # Drop the events the overlap window re-read from the previous run
              fresh = [r for r in records if not (event_keys(r) & seen_keys)]
              all_keys = set()
              for record in records:
                  all_keys |= event_keys(record)
      
              if not fresh:
                  print("No new log records found.")
                  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 fresh]) + '\n'
              blob.upload_from_string(ndjson, content_type='application/x-ndjson')
      
              print(f"Wrote {len(fresh)} records to gs://{GCS_BUCKET}/{object_key}")
      
              # Remember every fetched record so the next overlap window skips it
              save_seen(bucket, STATE_KEY, all_keys)
      
              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(fresh)} 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}")
              raise
      
          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}")
              raise
      
      def load_seen(bucket, key):
          """Load the event keys written by the previous run (sidecar object)."""
          try:
              blob = bucket.blob(key + '.seen')
              if blob.exists():
                  return set(json.loads(blob.download_as_text()))
          except Exception as e:
              print(f"Warning: Could not load seen keys: {e}")
              raise
      
          return set()
      
      def save_seen(bucket, key, keys):
          """Save the event keys of this run to the sidecar object."""
          try:
              blob = bucket.blob(key + '.seen')
              blob.upload_from_string(
                  json.dumps(sorted(keys)),
                  content_type='application/json'
              )
              print(f"Saved {len(keys)} event keys for deduplication")
          except Exception as e:
              print(f"Warning: Could not save seen keys: {e}")
              raise
      
      def fetch_logs(token: str, start_time: datetime, end_time: datetime, page_size: int, max_records: int):
          """
          Fetch user activity logs from Workday Activity Logging API with pagination and rate limiting.
      
          Args:
              token: OAuth 2.0 access token
              start_time: Start time for log query
              end_time: End time for log query
              page_size: Number of records per page
              max_records: Maximum total records to fetch
      
          Returns:
              Tuple of (records list, newest_event_time ISO string)
          """
          base_url = f"https://{WORKDAY_HOST}/ccx/api/privacy/v1/{WORKDAY_TENANT}/activityLogging"
      
          headers = {
              'Authorization': f'Bearer {token}',
              'Accept': 'application/json',
              'User-Agent': 'GoogleSecOps-WorkdayCollector/1.0'
          }
      
          records = []
          newest_time = None
          page_num = 0
          backoff = 1.0
          offset = 0
      
          from_iso = start_time.strftime('%Y-%m-%dT%H:%M:%SZ')
          to_iso = end_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))
              url = f"{base_url}?from={from_iso}&to={to_iso}&limit={current_limit}&offset={offset}"
      
              try:
                  response = http.request('GET', url, 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
      
                  backoff = 1.0
      
                  if response.status != 200:
                      print(f"HTTP Error: {response.status}")
                      response_text = response.data.decode('utf-8')
                      print(f"Response body: {response_text}")
                      return [], None
      
                  data = json.loads(response.data.decode('utf-8'))
      
                  page_results = data.get('data', [])
      
                  if not page_results:
                      print(f"No more results (empty page)")
                      break
      
                  print(f"Page {page_num}: Retrieved {len(page_results)} events")
                  records.extend(page_results)
      
                  # Track newest event time
                  for event in page_results:
                      try:
                          event_time = event.get('requestTime') or event.get('systemAccount', {}).get('signOnDateTime')
                          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}")
      
                  # Check pagination
                  total = data.get('total', 0)
                  if offset + len(page_results) >= total:
                      print(f"Reached end of results (offset={offset + len(page_results)}, total={total})")
                      break
      
                  offset += len(page_results)
      
              except Exception as e:
                  print(f"Error fetching logs: {e}")
                  return [], None
      
          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
      
  3. [デプロイ] をクリックして、関数を保存してデプロイします。

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

Cloud Scheduler ジョブを作成する

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

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

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

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

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

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

統合をテストする

  1. Cloud Scheduler コンソールで、ジョブ(workday-activity-collector-hourly)を見つけます。
  2. [強制実行] をクリックして、手動でトリガーします。
  3. 数秒待ってから、[Cloud Run> サービス> workday-activity-collector > ログ] に移動します。
  4. 関数が正常に実行されたことを確認します。以下のものを探します。

    Fetching logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00
    Page 1: Retrieved X events
    Wrote X records to gs://workday-activity-logs/workday/logs_YYYYMMDD_HHMMSS.ndjson
    Successfully processed X records
    
  5. Cloud Storage バケット(workday-activity-logs)をチェックして、ログが書き込まれていることを確認します。

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

  • HTTP 401: 環境変数で OAuth 認証情報を確認する
  • HTTP 403: ISU に User Activity Logging ドメインに必要な権限があることを確認する
  • HTTP 429: レート制限 - 関数はバックオフで自動的に再試行されます
  • アクセス トークンの取得に失敗しました: WORKDAY_HOST、WORKDAY_TENANT、CLIENT_ID、CLIENT_SECRET、REFRESH_TOKEN が正しいことを確認します

Workday ユーザー アクティビティ ログを取り込むように Google SecOps でフィードを構成する

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

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

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

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

    • ストレージ バケット URL: 接頭辞パスを含む Cloud Storage バケット URI を入力します。

      gs://workday-activity-logs/workday/
      
      • 次のように置き換えます。
        • workday-activity-logs: Cloud Storage バケット名。
        • workday: ログが保存されるオプションの接頭辞またはパス(ルートの場合は空のままにします)。
    • Source deletion option: 必要に応じて削除オプションを選択します。

      • ファイルを削除しない: ソースからファイルを削除しません(テストに推奨)。
      • 転送されたファイルと空のディレクトリを削除する: フェッチが正常に完了した後、ソースからファイルと空のディレクトリを削除します。
    • ファイルの最大経過日数: 過去の日数以内に変更されたファイルを含めます(デフォルトは 180 日)。

    • アセットの名前空間: アセットの名前空間。

    • Ingestion labels: このフィードのイベントに適用されるラベル。

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

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

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

Google SecOps サービス アカウントには、Cloud Storage バケットに対する 2 つのロールが必要です。ログ オブジェクトを読み取るための Storage オブジェクト閲覧者と、バケット メタデータを読み取るためのバケットレベルのロールです。

  1. [Cloud Storage] > [バケット] に移動します。
  2. バケット名(workday-activity-logs)をクリックします。
  3. [権限] タブに移動します。
  4. [アクセス権を付与] をクリックします。
  5. 次の構成情報を指定します。
    • プリンシパルを追加: Google SecOps サービス アカウントのメールアドレスを貼り付けます。
    • ロールを割り当てる: 次の両方を選択します。
      • Storage オブジェクト閲覧者: ログオブジェクトを読み取ります。
      • Storage レガシー バケット読み取り: バケット メタデータを読み取ります。[転送されたファイルと空のディレクトリを削除する] 削除オプションを選択した場合は、代わりに [Storage Legacy Bucket Writer] を選択します。これにより、削除権限も付与されます。
  6. [保存] をクリックします。

UDM マッピング テーブル

ログフィールド UDM マッピング ロジック
Account_Locked__Disabled_or_Expired_label additional.fields 統合済み
Authentication_Type_label additional.fields 統合済み
Created_Moment_label additional.fields 統合済み
Current_OMS_Version_label additional.fields 統合済み
Device_is_Trusted_label additional.fields 統合済み
Elapsed_Time__Minutes_label additional.fields 統合済み
Failed_Sign_On_label additional.fields 統合済み
Hire_Date_for_Signon_Worker_label additional.fields 統合済み
Job_Family_for_Signon_Worker_label additional.fields 統合済み
Operating_System_label additional.fields 統合済み
Pay_Rate_type_label additional.fields 統合済み
Request_Originator_label additional.fields 統合済み
SAML_Identity_Provider_label additional.fields 統合済み
Sign-on_Time_label additional.fields 統合済み
Signoff_Time_label additional.fields 統合済み
Signon_Day_of_the_Week_label additional.fields 統合済み
Signon_Hour_of_Day_label additional.fields 統合済み
Tenant_Access_Read_Only_label additional.fields 統合済み
UI_Client_Type_label additional.fields 統合済み
Worker_Active_for_Signon_Worker_label additional.fields 統合済み
Worker_Type_for_Signon_Worker_label additional.fields 統合済み
awaiting_details_label additional.fields 統合済み
data_label additional.fields 統合済み
evp_nme_label additional.fields 統合済み
evpminus_label additional.fields 統合済み
signon_ip_address_label additional.fields 統合済み
@requestTime metadata.event_timestamp ISO8601 として解析済み
Sign-on_Time metadata.event_timestamp yyyy-MM-ddTHH:mm:ssZZ として解析済み
event_type metadata.event_type 直接マッピングされます。
status_code metadata.product_event_type 直接マッピングされます。
tenant metadata.product_event_type 直接マッピングされます。
taskId metadata.product_log_id 直接マッピングされます。
User_Agent network.http.parsed_user_agent 名前変更済み/マッピング済み
Browser_Type network.http.user_agent 直接マッピングされます。
User_Agent network.http.user_agent 直接マッピングされます。
userAgent network.http.user_agent 直接マッピングされます。
Session_ID network.session_id 直接マッピングされます。
sessionId network.session_id 直接マッピングされます。
Client_TLS_Version network.tls.version 名前変更済み/マッピング済み
Worker principal.asset.hostname 直接マッピングされます。
ip principal.asset.ip 統合済み
ipaddress principal.asset.ip 統合済み
Worker principal.hostname 直接マッピングされます。
manager_name principal.hostname 直接マッピングされます。
ip principal.ip 統合済み
ipaddress principal.ip 統合済み
Signon_Worker_label principal.user.attribute.labels 統合済み
Supervisory_Organization_of_Signon_Worker_label principal.user.attribute.labels 統合済み
System_Account_label principal.user.attribute.labels 統合済み
Workday_Account_Sign_On_label principal.user.attribute.labels 統合済み
user_credentials_label principal.user.attribute.labels 統合済み
User_Name principal.user.email_addresses 統合済み
Role principal.user.user_display_name 直接マッピングされます。
manager_id principal.user.userid 直接マッピングされます。
Active_Session_label security_result.detection_fields 統合済み
Authentication_Failure_Message_label security_result.detection_fields 統合済み
Forgotten_Password_Reset_Request_label security_result.detection_fields 統合済み
Invalid_Credentials_label security_result.detection_fields 統合済み
Is_Device_Managed_label security_result.detection_fields 統合済み
MFA_Exempt_label security_result.detection_fields 統合済み
Password_Changed_label security_result.detection_fields 統合済み
Position_label security_result.detection_fields 統合済み
taskDisplayName security_result.summary 直接マッピングされます。
hostname target.hostname 直接マッピングされます。
CITY_NME target.location.city 直接マッピングされます。
Location target.location.city 直接マッピングされます。
COUNTRY_NME target.location.country_or_region 直接マッピングされます。
Location_Country_for_Signon_Worker target.location.country_or_region 直接マッピングされます。
Location_State target.location.state 直接マッピングされます。
STATE_NME target.location.state 直接マッピングされます。
Exempt_labels target.resource.attribute.labels 統合済み
browser_type_label target.resource.attribute.labels 統合済み
class_of_instance_label target.resource.attribute.labels 統合済み
device target.resource.resource_subtype 直接マッピングされます。
device1 target.resource.resource_subtype 直接マッピングされます。
device target.resource.type マッピング: (?i)(Desktop/Laptop/Mobile/Phone) → DEVICE
device1 target.resource.type マッピング: (?i)(Desktop/Laptop/Mobile/Phone) → DEVICE
BUSINESSGROUP_NME target.user.department 統合済み
email_name target.user.email_addresses 統合済み
EMPLOYEE_ID target.user.employee_id 直接マッピングされます。
Employee_ID target.user.employee_id 直接マッピングされます。
FIRST_NME target.user.first_name 直接マッピングされます。
EMPLOYEE_TYPE target.user.group_identifiers 統合済み
LAST_NME target.user.last_name 直接マッピングされます。
JOBTITLE_NME target.user.title 直接マッピングされます。
title target.user.title 直接マッピングされます。
target.id target.user.userid 直接マッピングされます。
なし metadata.product_name 定数: WORKDAY_USER_ACTIVITY
なし metadata.vendor_name 定数: WORKDAY_USER_ACTIVITY
なし target.resource.type 定数: DEVICE

変更ログ

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

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