收集 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 bucket，Google SecOps 會從該 bucket 擷取記錄。
事前準備
請確認您已完成下列事前準備事項：
- Google SecOps 執行個體
- 已啟用 Cloud Storage API 的 GCP 專案
- 建立及管理 GCS bucket 的權限
- 管理 Google Cloud Storage 值區 IAM 政策的權限
- 建立 Cloud Run 函式、Pub/Sub 主題和 Cloud Scheduler 工作的權限
- 建立服務帳戶及管理 IAM 角色的權限
- 具備管理員角色的 Citrix Cloud 租戶特殊存取權
- Citrix Cloud API 憑證 (用戶端 ID、用戶端密鑰、客戶 ID)
收集 Citrix Monitor Service API 憑證
- 登入 Citrix Cloud 控制台。
- 依序前往「Identity and Access Management」(身分與存取權管理)>「API Access」(API 存取權)。
- 按一下「Create Client」(建立用戶端)。
複製下列詳細資料並存放在安全的地方：
- 用戶端 ID
- 用戶端密碼
- 客戶 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"
如果收到 HTTP 200 回應和 JSON 資料，表示您的憑證運作正常。
建立 Google Cloud Storage bucket
- 前往 Google Cloud 控制台。
- 選取專案或建立新專案。
- 在導覽選單中，依序前往「Cloud Storage」>「Bucket」。
- 按一下「建立值區」。
請提供下列設定詳細資料：
設定 值 為 bucket 命名 輸入全域不重複的名稱 (例如 citrix-monitor-logs)位置類型 根據需求選擇 (區域、雙區域、多區域) 位置 選取位置 (例如 us-central1)儲存空間級別 標準 (建議用於經常存取的記錄) 存取控管 統一 (建議) 保護工具 選用：啟用物件版本管理或保留政策 點選「建立」。
為 Cloud Run 函式建立服務帳戶
Cloud Run 函式需要具備 GCS bucket 寫入權限的服務帳戶。
建立服務帳戶
- 在 GCP 控制台中，依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。
- 按一下「Create Service Account」(建立服務帳戶)。
- 提供下列設定詳細資料：
- 服務帳戶名稱：輸入
citrix-monitor-collector-sa。 - 服務帳戶說明：輸入
Service account for Cloud Run function to collect Citrix Monitor Service logs。
- 服務帳戶名稱：輸入
- 按一下「建立並繼續」。
- 在「將專案存取權授予這個服務帳戶」部分：
- 按一下「選擇角色」。
- 搜尋並選取「Storage 物件管理員」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Run Invoker」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Functions Invoker」(Cloud Functions 叫用者)。
- 按一下「繼續」。
- 按一下 [完成]。
這些角色適用於：
- Storage 物件管理員：將記錄檔寫入 GCS bucket，並管理狀態檔案
- Cloud Run 叫用者：允許 Pub/Sub 叫用函式
- Cloud Functions 叫用者：允許函式叫用
授予 GCS 值區的 IAM 權限
將 GCS bucket 的寫入權限授予服務帳戶：
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下 bucket 名稱。
- 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 提供下列設定詳細資料：
- 新增主體：輸入服務帳戶電子郵件地址 (
citrix-monitor-collector-sa@PROJECT_ID.iam.gserviceaccount.com)。 - 指派角色：選取「Storage 物件管理員」。
- 新增主體：輸入服務帳戶電子郵件地址 (
- 按一下 [儲存]。
建立 Pub/Sub 主題
建立 Pub/Sub 主題，Cloud Scheduler 會將訊息發布至該主題，而 Cloud Run 函式會訂閱該主題。
- 在 GCP Console 中，前往「Pub/Sub」>「Topics」(主題)。
- 按一下「建立主題」。
- 提供下列設定詳細資料：
- 主題 ID：輸入
citrix-monitor-trigger。 - 其他設定保留預設值。
- 主題 ID：輸入
- 點選「建立」。
建立 Cloud Run 函式來收集記錄
Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發，從 Citrix Monitor Service OData API 擷取記錄，並寫入 GCS。
- 前往 GCP Console 的「Cloud Run」。
- 按一下「Create service」(建立服務)。
- 選取「函式」 (使用內嵌編輯器建立函式)。
在「設定」部分，提供下列設定詳細資料：
設定 值 服務名稱 citrix-monitor-collector區域 選取與 GCS bucket 相符的區域 (例如 us-central1)執行階段 選取 Python 3.12 以上版本 在「Trigger (optional)」(觸發條件 (選用)) 專區：
- 按一下「+ 新增觸發條件」。
- 選取「Cloud Pub/Sub」。
- 在「選取 Cloud Pub/Sub 主題」中，選擇主題 (
citrix-monitor-trigger)。 - 按一下 [儲存]。
在「Authentication」(驗證) 部分：
- 選取「需要驗證」。
- 檢查 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。 - 「Maximum number of instances」(執行個體數量上限)：輸入
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 主控台的「Cloud Scheduler」。
- 點選「建立工作」。
請提供下列設定詳細資料：
設定 值 名稱 citrix-monitor-collector-hourly區域 選取與 Cloud Run 函式相同的區域 頻率 0 * * * *(每小時整點)時區 選取時區 (建議使用世界標準時間) 目標類型 Pub/Sub 主題 選取主題 ( citrix-monitor-trigger)郵件內文 {}(空白 JSON 物件)點選「建立」。
測試整合項目
- 在 Cloud Scheduler 控制台中找出您的工作。
- 按一下「強制執行」即可手動觸發工作。
- 稍等幾秒鐘。
- 前往「Cloud Run」>「Services」。
- 按一下函式名稱 (
citrix-monitor-collector)。 - 按一下 [Logs] (記錄) 分頁標籤。
確認函式是否執行成功。尋找：
Successfully processed X records for hour YYYY-MM-DDTHH:MM:SSZ Entities processed: Machines, Sessions, Connections, Applications, Users依序前往「Cloud Storage」>「Buckets」。
點按 bucket 名稱。
前往前置字元資料夾 (
citrix_monitor/)。確認是否已建立新的
.ndjson檔案。
如果記錄中顯示錯誤：
- HTTP 401：檢查環境變數中的 API 憑證
- HTTP 403：確認 Citrix Cloud 帳戶具備管理員權限
- HTTP 429：頻率限制 - 函式會自動重試並延遲
- 缺少環境變數：檢查是否已設定所有必要變數
在 Google SecOps 中設定動態饋給，擷取 Citrix Monitor Service 記錄
- 依序前往「SIEM 設定」>「動態饋給」。
- 按一下「新增動態消息」。
- 按一下「設定單一動態饋給」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
Citrix Monitor Service logs)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 選取「Citrix Monitor」做為「記錄類型」。
按一下「取得服務帳戶」。系統會顯示專屬服務帳戶電子郵件地址，例如：
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com複製這個電子郵件地址，後續步驟會用到。
點選「下一步」。
指定下列輸入參數的值：
儲存空間 bucket URL：輸入 GCS bucket URI，並加上前置路徑：
gs://citrix-monitor-logs/citrix_monitor/- 取代：
citrix-monitor-logs：您的 GCS bucket 名稱。citrix_monitor：儲存記錄的選用前置字元/資料夾路徑 (如為根目錄，請留空)。
- 取代：
來源刪除選項：根據偏好設定選取刪除選項：
- 永不：移轉後一律不刪除任何檔案 (建議用於測試)。
- 刪除已轉移的檔案：成功轉移檔案後刪除檔案。
- 刪除已轉移的檔案和空白目錄：成功轉移後刪除檔案和空白目錄。
檔案存在時間上限：包含在過去天數內修改的檔案。預設值為 180 天。
資產命名空間：資產命名空間。
擷取標籤：要套用至這個動態饋給事件的標籤。
點選「下一步」。
在「Finalize」(完成) 畫面中檢查新的動態饋給設定，然後按一下「Submit」(提交)。
將 IAM 權限授予 Google SecOps 服務帳戶
Google SecOps 服務帳戶需要 GCS bucket 的「Storage 物件檢視者」角色。
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下 bucket 名稱。
- 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 提供下列設定詳細資料：
- 新增主體：貼上 Google SecOps 服務帳戶電子郵件地址。
- 指派角色：選取「Storage 物件檢視者」。
按一下 [儲存]。
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 專業人員尋求解答。