Citrix Monitor Service のログを収集する

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

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

始める前に

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

  • Google SecOps インスタンス
  • Citrix Cloud テナントへの特権アクセス
  • AWS(S3、IAM、Lambda、EventBridge)への特権アクセス

Citrix Monitor Service の前提条件(ID、API キー、組織 ID、トークン)を収集する

  1. Citrix Cloud Console にログインします。
  2. [Identity and Access Management] > [API Access] に移動します。
  3. [Create Client] をクリックします。
  4. 次の詳細をコピーして安全な場所に保存します。
    • クライアント ID
    • Client Secret
    • お客様 ID(Citrix Cloud コンソールに表示)
    • API ベース URL:
      • グローバル: https://api.cloud.com
      • 日本: https://api.citrixcloud.jp

Google SecOps 用に AWS S3 バケットと IAM を構成する

  1. バケットの作成のユーザーガイドに沿って、Amazon S3 バケットを作成します。
  2. 後で参照できるように、バケットの名前リージョンを保存します(例: citrix-monitor-logs)。
  3. IAM ユーザーの作成のユーザーガイドに沿って、ユーザーを作成します。
  4. 作成したユーザーを選択します。
  5. [セキュリティ認証情報] タブを選択します。
  6. [アクセスキー] セクションで [アクセスキーを作成] をクリックします。
  7. [ユースケース] で [サードパーティ サービス] を選択します。
  8. [次へ] をクリックします。
  9. 省略可: 説明タグを追加します。
  10. [アクセスキーを作成] をクリックします。
  11. [CSV ファイルをダウンロード] をクリックして、[アクセスキー] と [シークレット アクセスキー] を保存し、後で使用できるようにします。
  12. [完了] をクリックします。
  13. [権限] タブを選択します。
  14. [権限ポリシー] セクションで、[権限を追加] をクリックします。
  15. [権限を追加] を選択します。
  16. [ポリシーを直接アタッチする] を選択します。
  17. AmazonS3FullAccess ポリシーを検索して選択します。
  18. [次へ] をクリックします。
  19. [権限を追加] をクリックします。

S3 アップロードの IAM ポリシーとロールを構成する

  1. AWS コンソールで、[IAM] > [ポリシー] > [ポリシーの作成] > [JSON] タブに移動します。
  2. 次のポリシーを入力します。

    {
      "Version": "2012-10-17",
      "Statement": [
        {
          "Sid": "AllowPutObjects",
          "Effect": "Allow",
          "Action": "s3:PutObject",
          "Resource": "arn:aws:s3:::citrix-monitor-logs/*"
        },
        {
          "Sid": "AllowGetStateObject",
          "Effect": "Allow",
          "Action": "s3:GetObject",
          "Resource": "arn:aws:s3:::citrix-monitor-logs/citrix_monitor/state.json"
        }
      ]
    }
    
    • 別のバケット名を入力した場合は、citrix-monitor-logs を置き換えます。
  3. [次へ] > [ポリシーを作成] をクリックします。

  4. [IAM] > [ロール] > [ロールの作成] > [AWS サービス] > [Lambda] に移動します。

  5. 新しく作成したポリシーと AWSLambdaBasicExecutionRole 管理ポリシーを関連付けます。

  6. ロールに「CitrixMonitorLambdaRole」という名前を付けて、[ロールを作成] をクリックします。

Lambda 関数を作成する

  1. AWS コンソールで、[Lambda] > [Functions] > [Create function] に移動します。
  2. [Author from scratch] をクリックします。
  3. 次の構成情報を提供してください。

    設定
    名前 CitrixMonitorCollector
    ランタイム Python 3.13
    アーキテクチャ x86_64
    実行ロール CitrixMonitorLambdaRole
  4. 関数を作成したら、[コード] タブを開き、スタブを削除して次のコード(CitrixMonitorCollector.py)を入力します。

    import os
    import json
    import uuid
    import datetime
    import urllib.parse
    import urllib.request
    import urllib.error
    import boto3
    import botocore
    
    # 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"
    
    s3 = boto3.client("s3")
    
    def http_post_form(url, data_dict):
        """POST form data to get authentication token."""
        data = urllib.parse.urlencode(data_dict).encode("utf-8")
        req = urllib.request.Request(url, data=data, headers={
            "Accept": "application/json",
            "Content-Type": "application/x-www-form-urlencoded",
        })
        with urllib.request.urlopen(req, timeout=45) as resp:
            return json.loads(resp.read().decode("utf-8"))
    
    def http_get_json(url, headers):
        """GET JSON data from API endpoint."""
        req = urllib.request.Request(url, headers=headers)
        with urllib.request.urlopen(req, timeout=90) as resp:
            return json.loads(resp.read().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:
            params.append("$filter=" + urllib.parse.quote(filter_query, safe="()= ':-TZ0123456789"))
        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."""
        # Try ModifiedDate filter if timestamps are provided
        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 urllib.error.HTTPError as e:
                # If ModifiedDate filtering fails, fall back to unfiltered query
                if e.code == 400 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 read_state_file(bucket, key):
        """Read the last processed timestamp from S3 state file."""
        try:
            obj = s3.get_object(Bucket=bucket, Key=key)
            content = obj["Body"].read().decode("utf-8")
            state = json.loads(content)
            timestamp_str = state.get("last_hour_utc")
            if timestamp_str:
                return datetime.datetime.fromisoformat(timestamp_str.replace("Z", "+00:00")).replace(tzinfo=None)
        except botocore.exceptions.ClientError as e:
            if e.response["Error"]["Code"] == "NoSuchKey":
                return None
            raise
        return None
    
    def write_state_file(bucket, key, dt_utc):
        """Write the current processed timestamp to S3 state file."""
        state = {"last_hour_utc": dt_utc.isoformat() + "Z"}
        s3.put_object(
            Bucket=bucket, 
            Key=key, 
            Body=json.dumps(state, separators=(",", ":")), 
            ContentType="application/json"
        )
    
    def write_ndjson_to_s3(bucket, key, rows):
        """Write rows as NDJSON to S3."""
        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").encode("utf-8")
        s3.put_object(
            Bucket=bucket, 
            Key=key, 
            Body=body, 
            ContentType="application/x-ndjson"
        )
    
    def lambda_handler(event, context):
        """Main Lambda handler function."""
    
        # Environment variables
        bucket = os.environ["S3_BUCKET"]
        prefix = os.environ.get("S3_PREFIX", "citrix_monitor").strip("/")
        state_key = os.environ.get("STATE_KEY") or f"{prefix}/state.json"
        customer_id = os.environ["CITRIX_CUSTOMER_ID"]
        client_id = os.environ["CITRIX_CLIENT_ID"]
        client_secret = os.environ["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"
    
        # Time window calculation
        now = datetime.datetime.utcnow()
        fallback_hour = (now - datetime.timedelta(minutes=lookback_minutes)).replace(minute=0, second=0, microsecond=0)
    
        last_processed = read_state_file(bucket, state_key)
        target_hour = (last_processed + datetime.timedelta(hours=1)) if last_processed else fallback_hour
        start_iso = target_hour.isoformat() + "Z"
        end_iso = (target_hour + datetime.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-s3-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:
                        s3_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_s3(bucket, s3_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:
                s3_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_s3(bucket, s3_key, rows_batch)
                total_records += len(rows_batch)
    
        # Update state file
        write_state_file(bucket, state_key, target_hour)
    
        return {
            "statusCode": 200,
            "body": json.dumps({
                "success": True, 
                "hour_collected": start_iso, 
                "records_written": total_records, 
                "entities_processed": entities
            })
        }
    
  5. [構成] > [環境変数] に移動します。

  6. [編集>新しい環境変数を追加] をクリックします。

  7. 次の環境変数を入力し、実際の値に置き換えます。

    キー 値の例
    S3_BUCKET citrix-monitor-logs
    S3_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
  8. 関数を作成したら、そのページにとどまるか、[Lambda] > [関数] > [CitrixMonitorCollector] を開きます。

  9. [CONFIGURATION] タブを選択します。

  10. [全般設定] パネルで、[編集] をクリックします。

  11. [Timeout] を [5 minutes (300 seconds)] に変更し、[Save] をクリックします。

EventBridge スケジュールを作成する

  1. [Amazon EventBridge] > [Scheduler] > [スケジュールの作成] に移動します。
  2. 次の構成の詳細を入力します。
    • 定期的なスケジュール: レート1 hour
    • ターゲット: Lambda 関数 CitrixMonitorCollector
    • 名前: CitrixMonitorCollector-1h
  3. [スケジュールを作成] をクリックします。

省略可: Google SecOps 用の読み取り専用の IAM ユーザーと鍵を作成する

  1. AWS コンソールで、[IAM] > [Users] > [Add users] に移動します。
  2. [ユーザーを追加] をクリックします。
  3. 次の構成の詳細を入力します。
    • ユーザー: secops-reader
    • アクセスタイプ: アクセスキー - プログラムによるアクセス
  4. [ユーザーを作成] をクリックします。
  5. 最小限の読み取りポリシー(カスタム)を関連付ける: [ユーザー] > [secops-reader] > [権限] > [権限を追加] > [ポリシーを直接関連付ける] > [ポリシーを作成]
  6. JSON エディタで、次のポリシーを入力します。

    {
      "Version": "2012-10-17",
      "Statement": [
        {
          "Effect": "Allow",
          "Action": ["s3:GetObject"],
          "Resource": "arn:aws:s3:::citrix-monitor-logs/*"
        },
        {
          "Effect": "Allow",
          "Action": ["s3:ListBucket"],
          "Resource": "arn:aws:s3:::citrix-monitor-logs"
        }
      ]
    }
    
  7. 名前を secops-reader-policy に設定します。

  8. [ポリシーの作成> 検索/選択> 次へ> 権限を追加] に移動します。

  9. [セキュリティ認証情報] > [アクセスキー] > [アクセスキーを作成] に移動します。

  10. CSV をダウンロードします(これらの値はフィードに入力されます)。

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

  1. [SIEM 設定] > [フィード] に移動します。
  2. [+ 新しいフィードを追加] をクリックします。
  3. [フィード名] フィールドに、フィードの名前を入力します(例: Citrix Monitor Service logs)。
  4. [ソースタイプ] として [Amazon S3 V2] を選択します。
  5. [ログタイプ] として [Citrix Monitor] を選択します。
  6. [次へ] をクリックします。
  7. 次の入力パラメータの値を指定します。
    • S3 URI: s3://citrix-monitor-logs/citrix_monitor/
    • Source deletion options: 必要に応じて削除オプションを選択します。
    • ファイルの最大経過日数: 指定した日数以内に変更されたファイルを含めます。デフォルトは 180 日です。
    • アクセスキー ID: S3 バケットにアクセスできるユーザー アクセスキー。
    • シークレット アクセスキー: S3 バケットにアクセスできるユーザーのシークレット キー。
    • アセットの名前空間: アセットの名前空間
    • Ingestion labels: このフィードのイベントに適用されるラベル。
  8. [次へ] をクリックします。
  9. [Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。

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