收集 Citrix Monitor Service 日志
解析器版本：2.0
支持的平台：
Google SecOps
SIEM
本文档介绍了如何使用 Google Cloud Storage 将 Citrix Monitor Service 日志注入到 Google Security Operations。Citrix Monitor Service 是 Citrix DaaS（以前称为 Citrix Virtual Apps and Desktops Service）的一部分，提供 OData v4 API，用于监控 Citrix 环境中有关机器、会话、连接、应用和用户的数据。Cloud Run 函数会按计划轮询 Citrix Monitor Service OData API，并将收集的日志写入 GCS 存储桶，然后 Google SecOps 会从该存储桶中提取日志。
准备工作
请确保满足以下前提条件：
- 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 API 凭据
- 登录 Citrix Cloud Console。
- 前往 Identity and Access Management > API Access。
- 点击创建客户端。
复制以下详细信息并将其保存在安全的位置：
- 客户端 ID (Client ID)
- 客户端密钥 (Client Secret)
- 客户 ID（显示在 Citrix Cloud 控制台的顶部）
- API 基准网址（取决于您所在的区域）：
- 美国/欧盟/亚太地区（南）：
https://api.cloud.com - 日本：
https://api.citrixcloud.jp - 美国政府：
https://api.cloud.us
- 美国/欧盟/亚太地区（南）：
测试 API 访问权限
在继续进行集成之前，请先测试您的凭据：
# Replace with your actual credentials CITRIX_CUSTOMER_ID="your-customer-id" CITRIX_CLIENT_ID="your-client-id" CITRIX_CLIENT_SECRET="your-client-secret" API_BASE="https://api.cloud.com" # Get bearer token TOKEN=$(curl -s -X POST "${API_BASE}/cctrustoauth2/${CITRIX_CUSTOMER_ID}/tokens/clients" \ -H "Content-Type: application/x-www-form-urlencoded" \ -d "grant_type=client_credentials&client_id=${CITRIX_CLIENT_ID}&client_secret=${CITRIX_CLIENT_SECRET}" \ | python3 -c "import sys,json; print(json.load(sys.stdin)['access_token'])") # Test Monitor OData API access curl -v "${API_BASE}/monitorodata/Machines?\$top=1" \ -H "Authorization: CWSAuth bearer=${TOKEN}" \ -H "Citrix-CustomerId: ${CITRIX_CUSTOMER_ID}" \ -H "Accept: application/json"
如果您收到包含 JSON 数据的 HTTP 200 响应，则表示您的凭据正常运行。
创建 Google Cloud Storage 存储桶
- 前往 Google Cloud 控制台。
- 选择您的项目或创建新项目。
- 在导航菜单中，依次前往 Cloud Storage > 存储分区。
- 点击创建存储分区。
提供以下配置详细信息：
设置 值 为存储桶命名 输入一个全局唯一的名称（例如 citrix-monitor-logs）位置类型 根据您的需求进行选择（区域级、双区域、多区域） 位置 选择营业地点（例如 us-central1）存储类别 标准（建议用于经常访问的日志） 访问权限控制 均匀（推荐） 保护工具 可选：启用对象版本控制或保留政策 点击创建。
为 Cloud Run 函数创建服务账号
Cloud Run 函数需要一个有权写入 GCS 存储桶的服务账号。
创建服务账号
- 在 GCP 控制台中，依次前往 IAM 和管理 > 服务账号。
- 点击创建服务账号。
- 提供以下配置详细信息：
- 服务账号名称：输入
citrix-monitor-collector-sa。 - 服务账号说明：输入
Service account for Cloud Run function to collect Citrix Monitor Service logs。
- 服务账号名称：输入
- 点击创建并继续。
- 在向此服务账号授予对项目的访问权限部分：
- 点击选择角色。
- 搜索并选择 Storage Object Admin。
- 点击 + 添加其他角色。
- 搜索并选择 Cloud Run Invoker。
- 点击 + 添加其他角色。
- 搜索并选择 Cloud Functions Invoker。
- 点击继续。
- 点击完成。
必须拥有这些角色，才能：
- Storage Object Admin：将日志写入 GCS 存储桶并管理状态文件
- Cloud Run Invoker：允许 Pub/Sub 调用函数
- Cloud Functions Invoker：允许调用函数
授予对 GCS 存储桶的 IAM 权限
向服务账号授予对 GCS 存储桶的写入权限：
- 前往 Cloud Storage > 存储分区。
- 点击您的存储桶名称。
- 前往权限标签页。
- 点击授予访问权限。
- 提供以下配置详细信息：
- 添加主账号：输入服务账号电子邮件地址 (
citrix-monitor-collector-sa@PROJECT_ID.iam.gserviceaccount.com)。 - 分配角色：选择 Storage Object Admin。
- 添加主账号：输入服务账号电子邮件地址 (
- 点击保存。
创建 Pub/Sub 主题
创建一个 Pub/Sub 主题，Cloud Scheduler 将向该主题发布消息，而 Cloud Run 函数将订阅该主题。
- 在 GCP 控制台中，前往 Pub/Sub > 主题。
- 点击创建主题。
- 提供以下配置详细信息：
- 主题 ID：输入
citrix-monitor-trigger。 - 将其他设置保留为默认值。
- 主题 ID：输入
- 点击创建。
创建 Cloud Run 函数以收集日志
Cloud Run 函数将由来自 Cloud Scheduler 的 Pub/Sub 消息触发，以从 Citrix Monitor Service OData API 中提取日志并将其写入 GCS。
- 在 GCP 控制台中，前往 Cloud Run。
- 点击创建服务。
- 选择函数（使用内嵌编辑器创建函数）。
在配置部分中，提供以下配置详细信息：
设置 值 Service 名称 citrix-monitor-collector区域 选择与您的 GCS 存储桶匹配的区域（例如 us-central1）运行时 选择 Python 3.12 或更高版本 在触发器（可选）部分中：
- 点击 + 添加触发器。
- 选择 Cloud Pub/Sub。
- 在选择 Cloud Pub/Sub 主题中，选择相应主题 (
citrix-monitor-trigger)。 - 点击保存。
在身份验证部分中：
- 选择需要进行身份验证。
- 检查 Identity and Access Management (IAM)。
向下滚动并展开容器、网络、安全性。
前往安全性标签页：
- 服务账号：选择服务账号 (
citrix-monitor-collector-sa)。
- 服务账号：选择服务账号 (
前往容器标签页：
- 点击变量和密钥。
- 为每个环境变量点击+ 添加变量：
变量名称 示例值 GCS_BUCKETcitrix-monitor-logsGCS_PREFIXcitrix_monitorSTATE_KEYcitrix_monitor/state.jsonCITRIX_CLIENT_IDyour-client-idCITRIX_CLIENT_SECRETyour-client-secretCITRIX_CUSTOMER_IDyour-customer-idAPI_BASEhttps://api.cloud.comENTITIESMachines,Sessions,Connections,Applications,UsersPAGE_SIZE1000LOOKBACK_MINUTES75USE_TIME_FILTERtrue在变量和密钥标签页中，向下滚动到请求：
- 请求超时：输入
600秒（10 分钟）。
- 请求超时：输入
前往容器中的设置标签页：
- 在资源部分中：
- 内存：选择 512 MiB 或更高值。
- CPU：选择 1。
- 点击完成。
- 在资源部分中：
向下滚动到执行环境：
- 选择默认（推荐）。
在修订版本伸缩部分中：
- 实例数下限：输入
0。 - 实例数上限：输入
100（或根据预期负载进行调整）。
- 实例数下限：输入
点击创建。
等待服务创建完成（1-2 分钟）。
创建服务后，系统会自动打开内嵌代码编辑器。
添加函数代码
- 在函数入口点字段中输入 main。
在内嵌代码编辑器中，创建两个文件：
main.py:
import functions_framework from google.cloud import storage import json import os import urllib3 from datetime import datetime, timedelta, timezone import uuid import time # 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 with timeouts http = urllib3.PoolManager( timeout=urllib3.Timeout(connect=5.0, read=30.0), retries=False, ) # 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) if response.status == 429: retry_after = int(response.headers.get('Retry-After', '10')) print(f"Rate limited (429). Retrying after {retry_after}s...") time.sleep(retry_after) response = http.request('GET', url, headers=headers) if response.status != 200: raise Exception(f"HTTP {response.status}: {response.data.decode('utf-8')[:500]}") 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: encoded_filter = urllib3.request.urlencode({'$filter': filter_query})[9:] 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 OData 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 '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) start_iso = None end_iso = None 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: 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)}') raiserequirements.txt:
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0
点击部署以保存并部署该函数。
等待部署完成（2-3 分钟）。
。
创建 Cloud Scheduler 作业
Cloud Scheduler 会定期向 Pub/Sub 主题发布消息，从而触发 Cloud Run 函数。
- 在 GCP Console 中，前往 Cloud Scheduler。
- 点击创建作业。
提供以下配置详细信息：
设置 值 名称 citrix-monitor-collector-hourly区域 选择与 Cloud Run 函数相同的区域 频率 0 * * * *（每小时一次，整点时）时区 选择时区（建议选择世界协调时间 [UTC]） 目标类型 Pub/Sub 主题 选择主题 ( citrix-monitor-trigger)消息正文 {}（空 JSON 对象）点击创建。
测试集成
- 在 Cloud Scheduler 控制台中，找到您的作业。
- 点击强制运行以手动触发作业。
- 等待几秒钟。
- 前往 Cloud Run > 服务。
- 点击函数名称 (
citrix-monitor-collector)。 - 点击日志标签页。
验证函数是否已成功执行。查找：
Successfully processed X records for hour YYYY-MM-DDTHH:MM:SSZ Entities processed: Machines, Sessions, Connections, Applications, Users前往 Cloud Storage > 存储分区。
点击您的存储桶名称。
前往前缀文件夹 (
citrix_monitor/)。验证是否已创建新的
.ndjson文件。
如果您在日志中看到错误，请执行以下操作：
- HTTP 401：检查环境变量中的 API 凭据
- HTTP 403：验证 Citrix Cloud 账号是否具有管理员权限
- HTTP 429：速率限制 - 函数将自动重试并进行退避
- 缺少环境变量：检查是否已设置所有必需的变量
在 Google SecOps 中配置 Feed 以注入 Citrix Monitor Service 日志
- 依次前往 SIEM 设置 > Feed。
- 点击添加新 Feed。
- 点击配置单个 Feed。
- 在 Feed 名称字段中，输入 Feed 的名称（例如
Citrix Monitor Service logs）。 - 选择 Google Cloud Storage V2 作为来源类型。
- 选择 Citrix Monitor 作为日志类型。
点击获取服务账号。系统会显示一个唯一的服务账号电子邮件地址，例如：
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com复制此电子邮件地址。您将在下一步骤中用到它。
点击下一步。
为以下输入参数指定值：
存储桶网址：输入带有前缀路径的 GCS 存储桶 URI：
gs://citrix-monitor-logs/citrix_monitor/- 替换：
citrix-monitor-logs：您的 GCS 存储桶名称。citrix_monitor：存储日志的可选前缀/文件夹路径（留空表示根目录）。
- 替换：
来源删除选项：根据您的偏好选择删除选项：
- 永不：转移后永不删除任何文件（建议用于测试）。
- 删除已转移的文件：在成功转移后删除文件。
- 删除已转移的文件和空目录：成功转移后删除文件和空目录。
文件存在时间上限：包含在过去指定天数内修改过的文件。默认值为 180 天。
资产命名空间：资产命名空间。
注入标签：要应用于此 Feed 中事件的标签。
点击下一步。
在最终确定界面中查看新的 Feed 配置，然后点击提交。
向 Google SecOps 服务账号授予 IAM 权限
Google SecOps 服务账号需要您的 GCS 存储桶的 Storage Object Viewer 角色。
- 前往 Cloud Storage > 存储分区。
- 点击您的存储桶名称。
- 前往权限标签页。
- 点击授予访问权限。
- 提供以下配置详细信息：
- 添加主账号：粘贴 Google SecOps 服务账号电子邮件地址。
- 分配角色：选择 Storage Object Viewer。
点击保存。
UDM 映射表
| 日志字段 | UDM 映射 | 逻辑 |
|---|---|---|
additional_user_id_label |
additional.fields |
已合并 |
CreatedDate |
metadata.event_timestamp |
解析为 ISO8601 |
event_type |
metadata.event_type |
直接映射 |
label_Connection_State |
metadata.ingestion_labels |
已合并 |
label_Log_On_Duration |
metadata.ingestion_labels |
已合并 |
label_Session_Key |
metadata.ingestion_labels |
已合并 |
Machine.AgentVersion |
metadata.product_version |
直接映射 |
Machine.DnsName |
network.dns_domain |
直接映射 |
applicationInstance.Application.BrowserName |
network.http.user_agent |
直接映射 |
CurrentConnectionId |
network.session_id |
直接映射 |
Machine.HostedMachineName |
principal.administrative_domain |
直接映射 |
Machine.Id |
principal.asset.asset_id |
直接映射 |
MachineId |
principal.asset.asset_id |
直接映射 |
Machine.HostingServerName |
principal.hostname |
直接映射 |
Machine.IPAddress |
principal.ip |
已合并 |
connection.ClientAddress |
principal.ip |
已合并 |
connection.ConnectedViaIPAddress |
principal.ip |
已合并 |
user |
principal.user.email_addresses |
已映射：^.+@.+$ → user |
Machine.AssociatedUserFullNames |
principal.user.user_display_name |
直接映射 |
Machine.Sid |
principal.user.windows_sid |
直接映射 |
User.Domain |
target.administrative_domain |
直接映射 |
applicationInstance.Application.Name |
target.application |
直接映射 |
connection.LaunchedViaIPAddress |
target.ip |
已合并 |
applicationInstance.Application.Path |
target.process.file.full_path |
直接映射 |
Machine.Hash |
target.process.file.md5 |
直接映射 |
applicationInstance.Application.AdminFolder |
target.process.parent_process.file.full_path |
直接映射 |
User.Upn |
target.user.email_addresses |
已合并 |
User.FullName |
target.user.user_display_name |
直接映射 |
User.UserName |
target.user.userid |
直接映射 |
User.Sid |
target.user.windows_sid |
直接映射 |
| 不适用 | metadata.event_type |
常量：GENERIC_EVENT |
| 不适用 | metadata.product_name |
常量：CITRIX_MONITOR |
| 不适用 | metadata.vendor_name |
常量：CITRIX_MONITOR |
| 不适用 | target.platform |
常量：WINDOWS |
更新日志
需要更多帮助？获得社区成员和 Google SecOps 专业人士的解答。