收集 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 控制台。
  2. 前往 Identity and Access Management > API Access。
  3. 点击创建客户端。
  4. 复制以下详细信息并将其保存在安全的位置：
    • 客户端 ID (Client ID)
    • 客户端密钥 (Client Secret)
    • 客户 ID（可在 Citrix Cloud 控制台中查看）
    • API 基本网址：
      • 全球：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. 点击 Download CSV file（下载 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 > 函数 > 创建函数。
  2. 点击从头开始创作。

  3. 提供以下配置详细信息：

    设置 值
    名称 CitrixMonitorCollector
    运行时 Python 3.13
    架构 x86_64
    执行角色 CitrixMonitorLambdaRole

  4. 创建函数后，打开 Code 标签页，删除桩代码并输入以下代码 (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. 选择配置标签页。

  10. 在常规配置面板中，点击修改。

  11. 将超时更改为 5 分钟（300 秒），然后点击保存。

创建 EventBridge 计划

  1. 依次前往 Amazon EventBridge > 调度程序 > 创建计划。
  2. 提供以下配置详细信息：
    • 周期性安排：频率 (1 hour)
    • 目标：您的 Lambda 函数 CitrixMonitorCollector
    • 名称：CitrixMonitorCollector-1h
  3. 点击创建时间表。

可选：为 Google SecOps 创建只读 IAM 用户和密钥

  1. 在 AWS 控制台中，依次前往 IAM > 用户 > 添加用户。
  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-monitor-logs/*"
    },
    {
      "Effect": "Allow",
      "Action": ["s3:ListBucket"],
      "Resource": "arn:aws:s3:::citrix-monitor-logs"
    }
  ]
}

  7. 将名称设置为 secops-reader-policy。

  8. 依次前往创建政策 > 搜索/选择 > 下一步 > 添加权限。

  9. 依次前往安全凭据 > 访问密钥 > 创建访问密钥。

  10. 下载 CSV（这些值会输入到 Feed 中）。

在 Google SecOps 中配置 Feed 以注入 Citrix Monitor Service 日志

  1. 依次前往 SIEM 设置> Feed。
  2. 点击 + 添加新 Feed。
  3. 在Feed 名称字段中，输入 Feed 的名称（例如 Citrix Monitor Service logs）。
  4. 选择 Amazon S3 V2 作为来源类型。
  5. 选择 Citrix Monitor 作为日志类型。
  6. 点击下一步。
  7. 为以下输入参数指定值：
    • S3 URI：s3://citrix-monitor-logs/citrix_monitor/
    • 来源删除选项：根据您的偏好设置选择删除选项。
    • 文件存在时间上限：包含在过去指定天数内修改的文件。默认值为 180 天。
    • 访问密钥 ID：有权访问 S3 存储桶的用户访问密钥。
    • 私有访问密钥：具有 S3 存储桶访问权限的用户私有密钥。
    • 资产命名空间：资产命名空间。
    • 注入标签：应用于此 Feed 中事件的标签。
  8. 点击下一步。
  9. 在最终确定界面中查看新的 Feed 配置，然后点击提交。

需要更多帮助？从社区成员和 Google SecOps 专业人士那里获得解答。