收集 Citrix Analytics 記錄

支援的國家/地區：

本文說明如何使用 Amazon S3，將 Citrix Analytics 記錄擷取至 Google Security Operations。

事前準備

請確認您已完成下列事前準備事項：

  • Google SecOps 執行個體
  • Citrix Analytics for Performance 租戶的特殊存取權
  • AWS 的具備權限存取權 (S3、IAM、Lambda、EventBridge)

收集 Citrix Analytics 必要條件

  1. 登入 Citrix Cloud 控制台。
  2. 依序前往「Identity and Access Management」(身分與存取權管理)「API Access」(API 存取權)。
  3. 按一下「建立用戶端」。
  4. 複製下列詳細資料並儲存於安全位置：
    • 用戶端 ID
    • 用戶端密碼
    • 客戶 ID (位於 Citrix Cloud 網址或 IAM 頁面)
    • API 基礎網址：https://api.cloud.com/casodata

為 Google SecOps 設定 AWS S3 值區和 IAM

  1. 按照這份使用者指南建立 Amazon S3 值區：建立值區
  2. 儲存 bucket 的「名稱」和「區域」，以供日後參考 (例如 citrix-analytics-logs)。
  3. 按照這份使用者指南建立使用者：建立 IAM 使用者。
  4. 選取建立的「使用者」。
  5. 選取「安全憑證」分頁標籤。
  6. 在「Access Keys」部分中，按一下「Create Access Key」。
  7. 選取「第三方服務」做為「用途」。
  8. 點選「下一步」。
  9. 選用：新增說明標記。
  10. 按一下「建立存取金鑰」。
  11. 按一下「下載 CSV 檔案」，儲存「存取金鑰」和「私密存取金鑰」以供日後使用。
  12. 按一下 [完成]。
  13. 選取 [權限] 分頁標籤。
  14. 在「Permissions policies」(權限政策) 區段中，按一下「Add permissions」(新增權限)。
  15. 選取「新增權限」。
  16. 選取「直接附加政策」
  17. 搜尋並選取 AmazonS3FullAccess 政策。
  18. 點選「下一步」。
  19. 按一下「Add permissions」。

設定 S3 上傳的身分與存取權管理政策和角色

  1. 在 AWS 控制台中，依序前往「IAM」>「Policies」>「Create policy」>「JSON」分頁標籤。

  2. 輸入下列政策：

    {
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowPutObjects",
      "Effect": "Allow",
      "Action": "s3:PutObject",
      "Resource": "arn:aws:s3:::citrix-analytics-logs/*"
    },
    {
      "Sid": "AllowGetStateObject",
      "Effect": "Allow",
      "Action": "s3:GetObject",
      "Resource": "arn:aws:s3:::citrix-analytics-logs/citrix_analytics/state.json"
    }
  ]
}
    • 如果您輸入的值區名稱不同，請替換 citrix-analytics-logs。

  3. 依序點選「Next」>「Create policy」。

  4. 依序前往「IAM」>「Roles」>「Create role」>「AWS service」>「Lambda」。

  5. 附加新建立的政策。

  6. 為角色命名 CitrixAnalyticsLambdaRole，然後按一下「建立角色」。

建立 Lambda 函式

  1. 在 AWS 控制台中，依序前往「Lambda」>「Functions」>「Create function」。
  2. 按一下「從頭開始撰寫」。

  3. 請提供下列設定詳細資料：

    設定 值
    名稱 CitrixAnalyticsCollector
    執行階段 Python 3.13
    架構 x86_64
    執行角色 CitrixAnalyticsLambdaRole

  4. 建立函式後，開啟「程式碼」分頁，刪除存根並輸入下列程式碼 (CitrixAnalyticsCollector.py)：

    import os
import json
import uuid
import datetime
import urllib.parse
import urllib.request
import boto3
import botocore

CITRIX_TOKEN_URL_TMPL = "https://api.cloud.com/cctrustoauth2/{customerid}/tokens/clients"
DEFAULT_API_BASE = "https://api.cloud.com/casodata"

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=30) as response:
        return json.loads(response.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=60) as response:
        return json.loads(response.read().decode("utf-8"))

def get_citrix_token(customer_id, client_id, client_secret):
    """Get Citrix Cloud authentication token."""
    url = CITRIX_TOKEN_URL_TMPL.format(customerid=customer_id)
    payload = {
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret,
    }
    token_response = _http_post_form(url, payload)
    return token_response["access_token"]

def fetch_odata_entity(entity, when_utc, top, headers, api_base):
    """Fetch data from Citrix Analytics OData API with pagination."""
    year = when_utc.year
    month = when_utc.month
    day = when_utc.day
    hour = when_utc.hour

    base_url = f"{api_base.rstrip('/')}/{entity}?year={year:04d}&month={month:02d}&day={day:02d}&hour={hour:02d}"
    skip = 0

    while True:
        url = f"{base_url}&$top={top}&$skip={skip}"
        data = _http_get_json(url, headers)
        items = data.get("value", [])

        if not items:
            break

        for item in items:
            yield item

        if len(items) < top:
            break

        skip += top

def read_state_file(bucket, state_key):
    """Read the last processed timestamp from S3 state file."""
    try:
        obj = s3.get_object(Bucket=bucket, Key=state_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, state_key, dt_utc):
    """Write the current processed timestamp to S3 state file."""
    state_data = {"last_hour_utc": dt_utc.isoformat() + "Z"}
    s3.put_object(
        Bucket=bucket, 
        Key=state_key,
        Body=json.dumps(state_data, separators=(",", ":")),
        ContentType="application/json"
    )

def write_ndjson_to_s3(bucket, key, records):
    """Write records as NDJSON to S3."""
    body_lines = []
    for record in records:
        json_line = json.dumps(record, 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", "").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", "sessions,machines,users").split(",") if e.strip()]
    top_n = int(os.environ.get("TOP_N", "1000"))
    lookback_minutes = int(os.environ.get("LOOKBACK_MINUTES", "75"))

    # Determine target hour to collect
    now = datetime.datetime.utcnow()
    fallback_target = (now - datetime.timedelta(minutes=lookback_minutes)).replace(minute=0, second=0, microsecond=0)

    last_processed = read_state_file(bucket, state_key)
    if last_processed:
        target_hour = last_processed + datetime.timedelta(hours=1)
    else:
        target_hour = fallback_target

    # Get authentication token
    token = get_citrix_token(customer_id, client_id, client_secret)
    headers = {
        "Authorization": f"CwsAuth bearer={token}",
        "Citrix-CustomerId": customer_id,
        "Accept": "application/json",
        "Content-Type": "application/json",
    }

    total_records = 0

    # Process each entity type
    for entity in entities:
        records = []

        for row in fetch_odata_entity(entity, target_hour, top_n, headers, api_base):
            enriched_record = {
                "citrix_entity": entity,
                "citrix_hour_utc": target_hour.isoformat() + "Z",
                "collection_timestamp": datetime.datetime.utcnow().isoformat() + "Z",
                "raw": row
            }
            records.append(enriched_record)

            # Write in batches to avoid memory issues
            if len(records) >= 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, records)
                total_records += len(records)
                records = []

        # Write remaining records
        if records:
            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, records)
            total_records += len(records)

    # Update state file
    write_state_file(bucket, state_key, target_hour)

    return {
        "statusCode": 200,
        "body": json.dumps({
            "success": True,
            "hour_collected": target_hour.isoformat() + "Z",
            "records_written": total_records,
            "entities_processed": entities
        })
    }

  5. 依序前往「Configuration」>「Environment variables」>「Edit」>「Add new environment variable」。

  6. 輸入下列環境變數，並將 換成您的值：

    鍵 範例值
    S3_BUCKET citrix-analytics-logs
    S3_PREFIX citrix_analytics
    STATE_KEY citrix_analytics/state.json
    CITRIX_CLIENT_ID your-client-id
    CITRIX_CLIENT_SECRET your-client-secret
    API_BASE https://api.cloud.com/casodata
    CITRIX_CUSTOMER_ID your-customer-id
    ENTITIES sessions,machines,users
    TOP_N 1000
    LOOKBACK_MINUTES 75

  7. 建立函式後，請留在該函式頁面 (或依序開啟「Lambda」>「Functions」>「CitrixAnalyticsCollector」)。

  8. 選取「設定」分頁標籤。

  9. 在「一般設定」面板中，按一下「編輯」。

  10. 將「Timeout」(逾時間隔) 變更為「5 minutes (300 seconds)」(5 分鐘 (300 秒))，然後按一下「Save」(儲存)。

建立 EventBridge 排程

  1. 依序前往「Amazon EventBridge」>「Scheduler」>「Create schedule」。
  2. 提供下列設定詳細資料：
    • 週期性時間表：費率 (1 hour)
    • 目標：您的 Lambda 函式 CitrixAnalyticsCollector
    • Name (名稱)：CitrixAnalyticsCollector-1h
  3. 按一下「建立時間表」。

選用：為 Google SecOps 建立唯讀 IAM 使用者和金鑰

  1. 在 AWS 控制台中，依序前往「IAM」>「Users」>「Add users」。
  2. 點選 [Add users] (新增使用者)。
  3. 提供下列設定詳細資料：
    • 使用者：secops-reader
    • 存取類型：存取金鑰 - 程式輔助存取
  4. 按一下「建立使用者」。
  5. 附加最低讀取權限政策 (自訂)：依序選取「使用者」>「secops-reader」>「權限」>「新增權限」>「直接附加政策」>「建立政策」。

  6. 在 JSON 編輯器中輸入下列政策：

    {
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["s3:GetObject"],
      "Resource": "arn:aws:s3:::citrix-analytics-logs/*"
    },
    {
      "Effect": "Allow",
      "Action": ["s3:ListBucket"],
      "Resource": "arn:aws:s3:::citrix-analytics-logs"
    }
  ]
}

  7. 將名稱設為 secops-reader-policy。

  8. 依序前往「建立政策」> 搜尋/選取 >「下一步」>「新增權限」。

  9. 依序前往「安全憑證」>「存取金鑰」>「建立存取金鑰」。

  10. 下載 CSV (這些值會輸入至動態饋給)。

在 Google SecOps 中設定動態饋給，擷取 Citrix Analytics 記錄

  1. 依序前往「SIEM 設定」>「動態饋給」。
  2. 按一下「+ 新增動態消息」。
  3. 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Citrix Analytics Performance logs)。
  4. 選取「Amazon S3 V2」做為「來源類型」。
  5. 選取「Citrix Analytics」做為「記錄類型」。
  6. 點選「下一步」。
  7. 指定下列輸入參數的值：
    • S3 URI：s3://citrix-analytics-logs/citrix_analytics/
    • 來源刪除選項：根據偏好設定選取刪除選項。
    • 檔案存在時間上限：包含在過去天數內修改的檔案。預設為 180 天。
    • 存取金鑰 ID：具有 S3 值區存取權的使用者存取金鑰。
    • 存取密鑰：具有 S3 bucket 存取權的使用者私密金鑰。
    • 資產命名空間：資產命名空間。
    • 擷取標籤：套用至這個動態饋給事件的標籤。
  8. 點選「下一步」。
  9. 在「完成」畫面中檢查新的動態饋給設定，然後按一下「提交」。

還有其他問題嗎？向社群成員和 Google SecOps 專業人員尋求答案。