Citrix Monitor Service のログを収集する

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

Google SecOps SIEM

注: この機能には、Google Security Operations サービス固有の規約の pre-GA サービス規約が適用されます。pre-GA の機能についてはサポートが制限されます。また、pre-GA の機能の変更は pre-GA の他のバージョンと互換性がない可能性があります。 詳細については、Google SecOps 技術サポート サービス ガイドラインと Google SecOps サービス固有の規約をご覧ください。

このドキュメントでは、Google Cloud Storage を使用して Citrix Monitor Service ログを Google Security Operations に取り込む方法について説明します。パーサーは、未加工の JSON 形式のログを Google SecOps UDM に準拠した構造化形式に変換します。未加工のログから関連するフィールドを抽出し、対応する UDM フィールドにマッピングします。また、ユーザー情報、マシンの詳細、ネットワーク アクティビティなどの追加コンテキストでデータを拡充します。

始める前に

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

• Google SecOps インスタンス
• Cloud Storage API が有効になっている GCP プロジェクト
• GCS バケットを作成および管理する権限
• GCS バケットの IAM ポリシーを管理する権限
• Cloud Run 関数、Pub/Sub トピック、Cloud Scheduler ジョブを作成する権限
• サービス アカウントの作成と IAM ロールの管理権限
• Citrix Cloud テナントへの特権アクセス
• Citrix Cloud API 認証情報（クライアント ID、クライアント シークレット、顧客 ID）

Citrix Monitor Service の前提条件を収集する

1. Citrix Cloud Console にログインします。
2. [Identity and Access Management> API Access] に移動します。
3. [クライアントを作成] をクリックします。

4. 次の詳細をコピーして安全な場所に保存します。

   • Client-ID
   • Client Secret
   • お客様 ID（Citrix Cloud コンソールに表示）
   • API ベース URL:
     • 米国/EU/AP-S: https://api.cloud.com
     • 日本: https://api.citrixcloud.jp
   注: DaaS の Citrix Monitor Service データにアクセスする際に選択した地理的リージョンに基づいてエンドポイントを使用します。米国/EU/AP-S リージョンでは https://api.cloud.com/monitorodata、日本リージョンでは https://api.citrixcloud.jp/monitorodata を使用します。

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

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

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

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

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

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

Cloud Run 関数には、GCS バケットに書き込む権限を持つサービス アカウントが必要です。

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

1. GCP Console で、[IAM と管理>サービス アカウント] に移動します。
2. [サービス アカウントを作成] をクリックします。
3. 次の構成の詳細を指定します。
   • サービス アカウント名: 「citrix-monitor-collector-sa」と入力します。
   • サービス アカウントの説明: 「Service account for Cloud Run function to collect Citrix Monitor Service 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. バケット名をクリックします。
3. [権限] タブに移動します。
4. [アクセス権を付与] をクリックします。
5. 次の構成の詳細を指定します。
   • プリンシパルを追加: サービス アカウントのメールアドレス（citrix-monitor-collector-sa@PROJECT_ID.iam.gserviceaccount.com）を入力します。
   • ロールを割り当てる: [Storage オブジェクト管理者] を選択します。
6. [保存] をクリックします。

Pub/Sub トピックの作成

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

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

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

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

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

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

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

5. [トリガー（省略可）] セクションで、次の操作を行います。

   1. [+ トリガーを追加] をクリックします。
   2. [Cloud Pub/Sub] を選択します。
   3. [Cloud Pub/Sub トピックを選択してください] で、トピック（citrix-monitor-trigger）を選択します。
   4. [保存] をクリックします。

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

   1. [認証が必要] を選択します。
   2. Identity and Access Management（IAM）を確認します。
   注: Pub/Sub は、関数を呼び出すときに認証を自動的に処理します。

7. 下にスクロールして、[コンテナ、ネットワーキング、セキュリティ] を開きます。

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

   • サービス アカウント: サービス アカウントを選択します（citrix-monitor-collector-sa）。

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

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

   変数名	値の例
   GCS_BUCKET	citrix-monitor-logs
   GCS_PREFIX	citrix_monitor
   STATE_KEY	citrix_monitor/state.json
   CITRIX_CLIENT_ID	your-client-id
   CITRIX_CLIENT_SECRET	your-client-secret
   CITRIX_CUSTOMER_ID	your-customer-id
   API_BASE	https://api.cloud.com
   ENTITIES	Machines,Sessions,Connections,Applications,Users
   PAGE_SIZE	1000
   LOOKBACK_MINUTES	75
   USE_TIME_FILTER	true

10. [変数とシークレット] タブで [リクエスト] まで下にスクロールします。

    • リクエストのタイムアウト: 600 秒（10 分）を入力します。

11. [コンテナ] の [設定] タブに移動します。

    • [リソース] セクションで次の操作を行います。
      • メモリ: 512 MiB 以上を選択します。
      • CPU: [1] を選択します。
    • [完了] をクリックします。

12. [実行環境] までスクロールします。

    • [デフォルト]（推奨）を選択します。

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, timedelta, timezone
   import uuid
   
   # Citrix Cloud OAuth2 endpoint template
   TOKEN_URL_TMPL = "{api_base}/cctrustoauth2/{customerid}/tokens/clients"
   DEFAULT_API_BASE = "https://api.cloud.com"
   MONITOR_BASE_PATH = "/monitorodata"
   
   # Initialize HTTP client
   http = urllib3.PoolManager()
   
   # Initialize Storage client
   storage_client = storage.Client()
   
   def http_post_form(url, data_dict):
       """POST form data to get authentication token."""
       encoded_data = urllib3.request.urlencode(data_dict)
       response = http.request(
           'POST',
           url,
           body=encoded_data,
           headers={
               'Accept': 'application/json',
               'Content-Type': 'application/x-www-form-urlencoded'
           }
       )
       return json.loads(response.data.decode('utf-8'))
   
   def http_get_json(url, headers):
       """GET JSON data from API endpoint."""
       response = http.request('GET', url, headers=headers)
       return json.loads(response.data.decode('utf-8'))
   
   def get_citrix_token(api_base, customer_id, client_id, client_secret):
       """Get Citrix Cloud authentication token."""
       url = TOKEN_URL_TMPL.format(
           api_base=api_base.rstrip('/'),
           customerid=customer_id
       )
       payload = {
           'grant_type': 'client_credentials',
           'client_id': client_id,
           'client_secret': client_secret
       }
       response = http_post_form(url, payload)
       return response['access_token']
   
   def build_entity_url(api_base, entity, filter_query=None, top=None):
       """Build OData URL with optional filter and pagination."""
       base = api_base.rstrip('/') + MONITOR_BASE_PATH + '/' + entity
       params = []
       if filter_query:
           # Encode filter query with safe characters for OData
           encoded_filter = urllib3.request.urlencode({'$filter': filter_query})[9:]  # Remove '$filter='
           params.append('$filter=' + encoded_filter)
       if top:
           params.append('$top=' + str(top))
       return base + ('?' + '&'.join(params) if params else '')
   
   def fetch_entity_rows(entity, start_iso=None, end_iso=None, page_size=1000, headers=None, api_base=DEFAULT_API_BASE):
       """Fetch entity data with optional time filtering and pagination."""
       first_url = None
       if start_iso and end_iso:
           filter_query = f"(ModifiedDate ge {start_iso} and ModifiedDate lt {end_iso})"
           first_url = build_entity_url(api_base, entity, filter_query, page_size)
       else:
           first_url = build_entity_url(api_base, entity, None, page_size)
   
       url = first_url
       while url:
           try:
               data = http_get_json(url, headers)
               items = data.get('value', [])
               for item in items:
                   yield item
               url = data.get('@odata.nextLink')
           except Exception as e:
               # If ModifiedDate filtering fails, fall back to unfiltered query
               if 'Bad Request' in str(e) and start_iso and end_iso:
                   print(f"ModifiedDate filter not supported for {entity}, falling back to unfiltered query")
                   url = build_entity_url(api_base, entity, None, page_size)
                   continue
               else:
                   raise
   
   def load_state(bucket, key):
       """Read the last processed timestamp from GCS state file."""
       try:
           blob = bucket.blob(key)
           if blob.exists():
               content = blob.download_as_text()
               state = json.loads(content)
               timestamp_str = state.get('last_hour_utc')
               if timestamp_str:
                   return datetime.fromisoformat(timestamp_str.replace('Z', '+00:00')).replace(tzinfo=None)
       except Exception as e:
           print(f"Warning: Could not load state: {str(e)}")
       return None
   
   def save_state(bucket, key, dt_utc):
       """Write the current processed timestamp to GCS state file."""
       state = {'last_hour_utc': dt_utc.isoformat() + 'Z'}
       blob = bucket.blob(key)
       blob.upload_from_string(
           json.dumps(state, separators=(',', ':')),
           content_type='application/json'
       )
   
   def write_ndjson_to_gcs(bucket, key, rows):
       """Write rows as NDJSON to GCS."""
       body_lines = []
       for row in rows:
           json_line = json.dumps(row, separators=(',', ':'), ensure_ascii=False)
           body_lines.append(json_line)
       body = '\n'.join(body_lines) + '\n'
       blob = bucket.blob(key)
       blob.upload_from_string(body, content_type='application/x-ndjson')
   
   @functions_framework.cloud_event
   def main(cloud_event):
       """
       Cloud Run function triggered by Pub/Sub to fetch Citrix Monitor Service logs and write to GCS.
   
       Args:
           cloud_event: CloudEvent object containing Pub/Sub message
       """
   
       # Get environment variables
       bucket_name = os.environ.get('GCS_BUCKET')
       prefix = os.environ.get('GCS_PREFIX', 'citrix_monitor').strip('/')
       state_key = os.environ.get('STATE_KEY') or f"{prefix}/state.json"
       customer_id = os.environ.get('CITRIX_CUSTOMER_ID')
       client_id = os.environ.get('CITRIX_CLIENT_ID')
       client_secret = os.environ.get('CITRIX_CLIENT_SECRET')
       api_base = os.environ.get('API_BASE', DEFAULT_API_BASE)
       entities = [e.strip() for e in os.environ.get('ENTITIES', 'Machines,Sessions,Connections,Applications,Users').split(',') if e.strip()]
       page_size = int(os.environ.get('PAGE_SIZE', '1000'))
       lookback_minutes = int(os.environ.get('LOOKBACK_MINUTES', '75'))
       use_time_filter = os.environ.get('USE_TIME_FILTER', 'true').lower() == 'true'
   
       if not all([bucket_name, customer_id, client_id, client_secret]):
           print('Error: Missing required environment variables')
           return
   
       try:
           # Get GCS bucket
           bucket = storage_client.bucket(bucket_name)
   
           # Time window calculation
           now = datetime.utcnow()
           fallback_hour = (now - timedelta(minutes=lookback_minutes)).replace(minute=0, second=0, microsecond=0)
           last_processed = load_state(bucket, state_key)
           target_hour = (last_processed + timedelta(hours=1)) if last_processed else fallback_hour
           start_iso = target_hour.isoformat() + 'Z'
           end_iso = (target_hour + timedelta(hours=1)).isoformat() + 'Z'
   
           # Authentication
           token = get_citrix_token(api_base, customer_id, client_id, client_secret)
           headers = {
               'Authorization': f'CWSAuth bearer={token}',
               'Citrix-CustomerId': customer_id,
               'Accept': 'application/json',
               'Accept-Encoding': 'gzip, deflate, br',
               'User-Agent': 'citrix-monitor-gcs-collector/1.0'
           }
   
           total_records = 0
   
           # Process each entity type
           for entity in entities:
               rows_batch = []
               try:
                   entity_generator = fetch_entity_rows(
                       entity=entity,
                       start_iso=start_iso if use_time_filter else None,
                       end_iso=end_iso if use_time_filter else None,
                       page_size=page_size,
                       headers=headers,
                       api_base=api_base
                   )
   
                   for row in entity_generator:
                       # Store raw Citrix data directly for proper parser recognition
                       rows_batch.append(row)
   
                       # Write in batches to avoid memory issues
                       if len(rows_batch) >= 1000:
                           gcs_key = f"{prefix}/{entity}/year={target_hour.year:04d}/month={target_hour.month:02d}/day={target_hour.day:02d}/hour={target_hour.hour:02d}/part-{uuid.uuid4().hex}.ndjson"
                           write_ndjson_to_gcs(bucket, gcs_key, rows_batch)
                           total_records += len(rows_batch)
                           rows_batch = []
   
               except Exception as ex:
                   print(f"Error processing entity {entity}: {str(ex)}")
                   continue
   
               # Write remaining records
               if rows_batch:
                   gcs_key = f"{prefix}/{entity}/year={target_hour.year:04d}/month={target_hour.month:02d}/day={target_hour.day:02d}/hour={target_hour.hour:02d}/part-{uuid.uuid4().hex}.ndjson"
                   write_ndjson_to_gcs(bucket, gcs_key, rows_batch)
                   total_records += len(rows_batch)
   
           # Update state file
           save_state(bucket, state_key, target_hour)
   
           print(f"Successfully processed {total_records} records for hour {start_iso}")
           print(f"Entities processed: {', '.join(entities)}")
   
       except Exception as e:
           print(f'Error processing Citrix Monitor logs: {str(e)}')
           raise
   

   • 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. 次の構成情報を提供してください。

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

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

スケジューラ ジョブをテストする

1. Cloud Scheduler コンソールで、ジョブを見つけます。
2. [強制実行] をクリックして手動でトリガーします。
3. 数秒待ってから、[Cloud Run> サービス> citrix-monitor-collector > ログ] に移動します。
4. 関数が正常に実行されたことを確認します。
5. GCS バケットをチェックして、ログが書き込まれたことを確認します。

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

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

サービス アカウントのメールアドレスを取得する

1. [SIEM 設定] > [フィード] に移動します。
2. [Add New Feed] をクリックします。
3. [単一フィードを設定] をクリックします。
4. [フィード名] フィールドに、フィードの名前を入力します（例: Citrix Monitor Service logs）。
5. [ソースタイプ] として [Google Cloud Storage V2] を選択します。
6. [ログタイプ] として [Citrix Monitor] を選択します。

7. [サービス アカウントを取得する] をクリックします。一意のサービス アカウント メールアドレスが表示されます（例:）。

   [e-mail]
   

8. このメールアドレスをコピーして、次のステップで使用します。

   注: 各 Google SecOps インスタンスには一意のサービス アカウントがあります。他のドキュメントや例のサービス アカウントは使用しないでください。

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

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

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

6. [保存] をクリックします。

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

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

1. [SIEM 設定] > [フィード] に移動します。
2. [Add New Feed] をクリックします。
3. [単一フィードを設定] をクリックします。
4. [フィード名] フィールドに、フィードの名前を入力します（例: Citrix Monitor Service logs）。
5. [ソースタイプ] として [Google Cloud Storage V2] を選択します。
6. [ログタイプ] として [Citrix Monitor] を選択します。
7. [次へ] をクリックします。

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

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

     gs://citrix-monitor-logs/citrix_monitor/
     

     • 次のように置き換えます。

       • citrix-monitor-logs: GCS バケット名。
       • citrix_monitor: ログが保存されるオプションの接頭辞/フォルダパス（ルートの場合は空のままにします）。
     注: URI の末尾には常にスラッシュ（/）を付けます。

   • Source deletion option: 必要に応じて削除オプションを選択します。

     • なし: 転送後にファイルを削除しません（テストにおすすめ）。
     • 転送されたファイルを削除する: 転送が完了した後にファイルを削除します。

     • 転送されたファイルと空のディレクトリを削除する: 転送が完了した後にファイルと空のディレクトリを削除します。

       注: 削除オプションを選択する場合は、サービス アカウントに Storage オブジェクト閲覧者ではなく、Storage オブジェクト管理者のロールが必要です。必要に応じて IAM 権限を更新します。

   • ファイルの最大経過日数: 指定した日数以内に変更されたファイルを含めます。デフォルトは 180 日です。

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

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

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

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

ご不明な点がございましたら、コミュニティ メンバーや Google SecOps のプロフェッショナルから回答を得ることができます。