收集 UKG 記錄

驗證權限

如要確認 API 憑證是否具備必要存取權，請按照下列步驟操作：

1. 登入 UKG Pro 管理入口網站。
2. 依序點選「選單」>「管理」>「系統設定」>「安全性」>「稽核記錄」。
3. 確認您可以查看稽核記錄。
4. 如果無法存取稽核資料，請與 UKG Pro 管理員聯絡，要求對方授予必要權限。

測試 API 存取權

• 請先測試憑證，再繼續進行整合：

  # Replace with your actual credentials
  ACCESS_TOKEN="your-access-token"
  API_KEY="your-api-key"
  SERVICE_HOST="your-service-host"
  
  # Test API access - fetch audit logs
  curl -v -H "Authorization: Bearer ${ACCESS_TOKEN}" \
    -H "US-Customer-Api-Key: ${API_KEY}" \
    "https://${SERVICE_HOST}/personnel/v1/audit-logs?startDate=2024-01-01&endDate=2024-01-02"
  

為 Cloud Run 函式建立服務帳戶

Cloud Run 函式需要具備 GCS bucket 寫入權限，且可由 Pub/Sub 叫用的服務帳戶。

建立服務帳戶

1. 在 GCP 控制台中，依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。
2. 按一下「Create Service Account」(建立服務帳戶)。
3. 請提供下列設定詳細資料：
   • 服務帳戶名稱：輸入 ukg-collector-sa。
   • 服務帳戶說明：輸入 Service account for Cloud Run function to collect UKG logs。
4. 按一下「建立並繼續」。
5. 在「將專案存取權授予這個服務帳戶」部分，新增下列角色：
   1. 按一下「選擇角色」。
   2. 搜尋並選取「Storage 物件管理員」。
   3. 點選「+ 新增其他角色」。
   4. 搜尋並選取「Cloud Run Invoker」。
   5. 點選「+ 新增其他角色」。
   6. 搜尋並選取「Cloud Functions Invoker」(Cloud Functions 叫用者)。
6. 按一下「繼續」。
7. 按一下 [完成]。

這些角色適用於：

• Storage 物件管理員：將記錄檔寫入 GCS bucket，並管理狀態檔案
• Cloud Run 叫用者：允許 Pub/Sub 叫用函式
• Cloud Functions 叫用者：允許函式叫用

授予 GCS 值區的 IAM 權限

將 GCS bucket 的寫入權限授予服務帳戶：

1. 依序前往「Cloud Storage」>「Buckets」。
2. 按一下 bucket 名稱 (例如 ukg-logs)。
3. 前往「權限」分頁標籤。
4. 按一下「授予存取權」。
5. 請提供下列設定詳細資料：
   • 新增主體：輸入服務帳戶電子郵件地址 (例如 ukg-collector-sa@your-project.iam.gserviceaccount.com)。
   • 指派角色：選取「Storage 物件管理員」。
6. 按一下 [儲存]。

建立 Pub/Sub 主題

建立 Pub/Sub 主題，Cloud Scheduler 會將訊息發布至該主題，而 Cloud Run 函式會訂閱該主題。

1. 在 GCP Console 中，前往「Pub/Sub」>「Topics」(主題)。
2. 按一下「建立主題」。
3. 請提供下列設定詳細資料：
   • 主題 ID：輸入 ukg-trigger。
   • 其他設定保留預設值。
4. 點選「建立」。

建立 Cloud Run 函式來收集記錄

Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發，從 UKG Pro API 擷取記錄並寫入 GCS。

1. 前往 GCP Console 的「Cloud Run」。
2. 按一下「Create service」(建立服務)。
3. 選取「函式」 (使用內嵌編輯器建立函式)。

4. 在「設定」部分，提供下列設定詳細資料：

   設定	值
   服務名稱	ukg-collector
   區域	選取與 GCS bucket 相符的區域 (例如 us-central1)
   執行階段	選取 Python 3.12 以上版本

5. 在「Trigger (optional)」(觸發條件 (選用)) 專區：

   1. 按一下「+ 新增觸發條件」。
   2. 選取「Cloud Pub/Sub」。
   3. 在「選取 Cloud Pub/Sub 主題」中，選擇主題 ukg-trigger。
   4. 按一下 [儲存]。

6. 在「Authentication」(驗證) 部分：

   1. 選取「需要驗證」。
   2. 檢查 Identity and Access Management (IAM)。

7. 向下捲動並展開「容器」、「網路」和「安全性」。

8. 前往「安全性」分頁：

   • 服務帳戶：選取服務帳戶 ukg-collector-sa。

9. 前往「容器」分頁：

   1. 按一下「變數與密鑰」。
   2. 針對每個環境變數，按一下「+ 新增變數」：

   變數名稱	範例值	說明
   GCS_BUCKET	ukg-logs	GCS bucket 名稱
   GCS_PREFIX	ukg	記錄檔的前置字串
   STATE_KEY	ukg/state.json	狀態檔案路徑
   SERVICE_HOST	your-service-host	UKG Pro 服務主機網址
   UKG_USERNAME	your-service-username	服務帳戶使用者名稱
   UKG_PASSWORD	your-service-password	服務帳戶密碼
   UKG_API_KEY	your-api-key	客戶 API 金鑰
   MAX_RECORDS	10000	每次執行的記錄數上限
   PAGE_SIZE	100	每頁記錄數
   LOOKBACK_HOURS	24	初始回溯期

10. 在「變數與密鑰」分頁中向下捲動至「要求」：

    • 要求逾時：輸入 600 秒 (10 分鐘)。

11. 前往「容器」中的「設定」分頁：

    • 在「資源」部分：
      • 記憶體：選取 512 MiB 以上。
      • CPU：選取 1。

12. 在「修訂版本資源調度」部分：

    • 執行個體數量下限：輸入 0。
    • 「Maximum number of instances」(執行個體數量上限)：輸入 100 (或根據預期負載調整)。

13. 點選「建立」。

14. 等待服務建立完成 (1 到 2 分鐘)。

15. 服務建立完成後，系統會自動開啟內嵌程式碼編輯器。

新增函式程式碼

1. 在「函式進入點」中輸入 main。

2. 在內嵌程式碼編輯器中建立兩個檔案：

   • 第一個檔案：main.py:

   import functions_framework
   from google.cloud import storage
   import json
   import os
   import urllib3
   from datetime import datetime, timezone, timedelta
   import time
   import base64
   
   # 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()
   
   # Environment variables
   GCS_BUCKET = os.environ.get('GCS_BUCKET')
   GCS_PREFIX = os.environ.get('GCS_PREFIX', 'ukg')
   STATE_KEY = os.environ.get('STATE_KEY', 'ukg/state.json')
   SERVICE_HOST = os.environ.get('SERVICE_HOST', '').rstrip('/')
   UKG_USERNAME = os.environ.get('UKG_USERNAME', '')
   UKG_PASSWORD = os.environ.get('UKG_PASSWORD', '')
   UKG_API_KEY = os.environ.get('UKG_API_KEY', '')
   MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '10000'))
   PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '100'))
   LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))
   
   def parse_datetime(value: str) -> datetime:
       """Parse ISO datetime string to datetime object."""
       if value.endswith("Z"):
           value = value[:-1] + "+00:00"
       return datetime.fromisoformat(value)
   
   def get_token():
       """Get OAuth 2.0 access token from UKG Pro."""
       token_url = f"https://{SERVICE_HOST}/authentication/token"
   
       auth_string = f"{UKG_USERNAME}:{UKG_PASSWORD}"
       auth_bytes = auth_string.encode('utf-8')
       auth_b64 = base64.b64encode(auth_bytes).decode('utf-8')
   
       headers = {
           'Authorization': f'Basic {auth_b64}',
           'US-Customer-Api-Key': UKG_API_KEY,
           'Content-Type': 'application/x-www-form-urlencoded',
           'Accept': 'application/json'
       }
   
       body = 'grant_type=client_credentials'
   
       backoff = 1.0
       max_retries = 3
   
       for attempt in range(max_retries):
           response = http.request('POST', token_url, body=body.encode('utf-8'), headers=headers)
   
           if response.status == 429:
               retry_after = int(response.headers.get('Retry-After', str(int(backoff))))
               print(f"Rate limited (429) on token request. Retrying after {retry_after}s...")
               time.sleep(retry_after)
               backoff = min(backoff * 2, 30.0)
               continue
   
           if response.status != 200:
               raise RuntimeError(f"Failed to get access token: {response.status} - {response.data.decode('utf-8')}")
   
           data = json.loads(response.data.decode('utf-8'))
           return data['access_token']
   
       raise RuntimeError(f"Failed to get token after {max_retries} retries due to rate limiting")
   
   @functions_framework.cloud_event
   def main(cloud_event):
       """
       Cloud Run function triggered by Pub/Sub to fetch UKG logs and write to GCS.
   
       Args:
           cloud_event: CloudEvent object containing Pub/Sub message
       """
   
       if not all([GCS_BUCKET, SERVICE_HOST, UKG_USERNAME, UKG_PASSWORD, UKG_API_KEY]):
           print('Error: Missing required environment variables')
           return
   
       try:
           bucket = storage_client.bucket(GCS_BUCKET)
   
           # Load state
           state = load_state(bucket, STATE_KEY)
   
           # Determine time window
           now = datetime.now(timezone.utc)
           last_time = None
   
           if isinstance(state, dict) and state.get("last_event_time"):
               try:
                   last_time = parse_datetime(state["last_event_time"])
                   last_time = last_time - timedelta(minutes=2)
               except Exception as e:
                   print(f"Warning: Could not parse last_event_time: {e}")
   
           if last_time is None:
               last_time = now - timedelta(hours=LOOKBACK_HOURS)
   
           print(f"Fetching logs from {last_time.isoformat()} to {now.isoformat()}")
   
           # Get access token
           token = get_token()
   
           # Fetch logs
           records, newest_event_time = fetch_logs(
               token=token,
               start_time=last_time,
               end_time=now,
               page_size=PAGE_SIZE,
               max_records=MAX_RECORDS,
           )
   
           if not records:
               print("No new log records found.")
               save_state(bucket, STATE_KEY, now.isoformat())
               return
   
           # Write to GCS as NDJSON
           timestamp = now.strftime('%Y%m%d_%H%M%S')
           object_key = f"{GCS_PREFIX}/logs_{timestamp}.ndjson"
           blob = bucket.blob(object_key)
   
           ndjson = '\n'.join([json.dumps(record, ensure_ascii=False) for record in records]) + '\n'
           blob.upload_from_string(ndjson, content_type='application/x-ndjson')
   
           print(f"Wrote {len(records)} records to gs://{GCS_BUCKET}/{object_key}")
   
           if newest_event_time:
               save_state(bucket, STATE_KEY, newest_event_time)
           else:
               save_state(bucket, STATE_KEY, now.isoformat())
   
           print(f"Successfully processed {len(records)} records")
   
       except Exception as e:
           print(f'Error processing logs: {str(e)}')
           raise
   
   def load_state(bucket, key):
       """Load state from GCS."""
       try:
           blob = bucket.blob(key)
           if blob.exists():
               state_data = blob.download_as_text()
               return json.loads(state_data)
       except Exception as e:
           print(f"Warning: Could not load state: {e}")
   
       return {}
   
   def save_state(bucket, key, last_event_time_iso: str):
       """Save the last event timestamp to GCS state file."""
       try:
           state = {'last_event_time': last_event_time_iso}
           blob = bucket.blob(key)
           blob.upload_from_string(
               json.dumps(state, indent=2),
               content_type='application/json'
           )
           print(f"Saved state: last_event_time={last_event_time_iso}")
       except Exception as e:
           print(f"Warning: Could not save state: {e}")
   
   def fetch_logs(token: str, start_time: datetime, end_time: datetime, page_size: int, max_records: int):
       """
       Fetch audit logs from UKG Pro API with pagination and rate limiting.
   
       Args:
           token: OAuth 2.0 access token
           start_time: Start time for log query
           end_time: End time for log query
           page_size: Number of records per page
           max_records: Maximum total records to fetch
   
       Returns:
           Tuple of (records list, newest_event_time ISO string)
       """
       endpoint = f"https://{SERVICE_HOST}/personnel/v1/audit-logs"
   
       headers = {
           'Authorization': f'Bearer {token}',
           'US-Customer-Api-Key': UKG_API_KEY,
           'Accept': 'application/json',
           'User-Agent': 'GoogleSecOps-UKGCollector/1.0'
       }
   
       records = []
       newest_time = None
       page_num = 0
       backoff = 1.0
       page = 1
   
       start_date = start_time.strftime('%Y-%m-%d')
       end_date = end_time.strftime('%Y-%m-%d')
   
       while True:
           page_num += 1
   
           if len(records) >= max_records:
               print(f"Reached max_records limit ({max_records})")
               break
   
           url = f"{endpoint}?startDate={start_date}&endDate={end_date}&page={page}&per_page={page_size}"
   
           try:
               response = http.request('GET', url, headers=headers)
   
               if response.status == 429:
                   retry_after = int(response.headers.get('Retry-After', str(int(backoff))))
                   print(f"Rate limited (429). Retrying after {retry_after}s...")
                   time.sleep(retry_after)
                   backoff = min(backoff * 2, 30.0)
                   continue
   
               backoff = 1.0
   
               if response.status != 200:
                   print(f"HTTP Error: {response.status}")
                   response_text = response.data.decode('utf-8')
                   print(f"Response body: {response_text}")
                   return [], None
   
               data = json.loads(response.data.decode('utf-8'))
   
               # UKG may return results in different formats
               if isinstance(data, list):
                   page_results = data
               else:
                   page_results = data.get('data', data.get('results', data.get('auditLogs', [])))
   
               if not page_results:
                   print(f"No more results (empty page)")
                   break
   
               print(f"Page {page_num}: Retrieved {len(page_results)} events")
               records.extend(page_results)
   
               # Track newest event time
               for event in page_results:
                   try:
                       event_time = event.get('auditDate') or event.get('dateTime') or event.get('timestamp')
                       if event_time:
                           if newest_time is None or parse_datetime(event_time) > parse_datetime(newest_time):
                               newest_time = event_time
                   except Exception as e:
                       print(f"Warning: Could not parse event time: {e}")
   
               if len(page_results) < page_size:
                   print(f"Reached last page (size={len(page_results)} < limit={page_size})")
                   break
   
               page += 1
   
           except Exception as e:
               print(f"Error fetching logs: {e}")
               return [], None
   
       print(f"Retrieved {len(records)} total records from {page_num} pages")
       return records, newest_time
   

   • 第二個檔案：requirements.txt:

   functions-framework==3.*
   google-cloud-storage==2.*
   urllib3>=2.0.0
   

3. 點選「部署」即可儲存並部署函式。

4. 等待部署作業完成 (2 到 3 分鐘)。

建立 Cloud Scheduler 工作

Cloud Scheduler 會定期將訊息發布至 Pub/Sub 主題，觸發 Cloud Run 函式。

1. 前往 GCP 主控台的「Cloud Scheduler」。
2. 點選「建立工作」。

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

   設定	值
   名稱	ukg-collector-hourly
   區域	選取與 Cloud Run 函式相同的區域
   頻率	0 * * * * (每小時整點)
   時區	選取時區 (建議使用世界標準時間)
   目標類型	Pub/Sub
   主題	選取主題 ukg-trigger
   郵件內文	{} (空白 JSON 物件)

4. 點選「建立」。

排程頻率選項

根據記錄檔量和延遲時間要求選擇頻率：

測試整合項目

1. 在 Cloud Scheduler 控制台中，找出您的工作 (ukg-collector-hourly)。
2. 按一下「強制執行」即可手動觸發。
3. 等待幾秒鐘，然後依序前往「Cloud Run」>「Services」(服務) >「ukg-collector」>「Logs」(記錄)。

4. 確認函式是否已順利執行。請注意以下事項：

   Fetching logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00
   Page 1: Retrieved X events
   Wrote X records to gs://ukg-logs/ukg/logs_YYYYMMDD_HHMMSS.ndjson
   Successfully processed X records
   

5. 檢查 GCS 值區 (ukg-logs)，確認記錄是否已寫入。

如果記錄中顯示錯誤：

• HTTP 401：檢查環境變數中的 API 憑證
• HTTP 403：確認服務帳戶在 UKG Pro 中具備必要權限
• HTTP 429：頻率限制 - 函式會自動重試，並延遲一段時間
• 無法取得存取權杖：確認 SERVICE_HOST、UKG_USERNAME、UKG_PASSWORD 和 UKG_API_KEY 是否正確

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

1. 依序前往「SIEM 設定」>「動態饋給」。
2. 按一下「新增動態消息」。
3. 按一下「設定單一動態饋給」。
4. 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 UKG Logs)。
5. 選取「Google Cloud Storage V2」做為「來源類型」。
6. 選取「UKG」做為「記錄類型」。

7. 按一下「取得服務帳戶」，系統會顯示專屬服務帳戶電子郵件地址，例如：

   chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
   

8. 複製這個電子郵件地址，後續步驟會用到。

9. 點選「下一步」。

10. 指定下列輸入參數的值：

    • 儲存空間 bucket URL：輸入 GCS bucket URI，並加上前置路徑：

      gs://ukg-logs/ukg/
      

      • 取代：
        • ukg-logs：您的 GCS bucket 名稱。
        • ukg：儲存記錄的選用前置字元/資料夾路徑 (如為根目錄，請留空)。

    • 來源刪除選項：根據偏好設定選取刪除選項：

      • 永不：移轉後一律不刪除任何檔案 (建議用於測試)。
      • 刪除已轉移的檔案：成功轉移檔案後刪除檔案。

      • 刪除已轉移的檔案和空白目錄：成功轉移後刪除檔案和空白目錄。

    • 檔案存在時間上限：包含在過去天數內修改的檔案 (預設為 180 天)。

    • 資產命名空間：資產命名空間。

    • 擷取標籤：要套用至這個動態饋給事件的標籤。

11. 點選「下一步」。

12. 在「Finalize」(完成) 畫面中檢查新的動態饋給設定，然後按一下「Submit」(提交)。

將 IAM 權限授予 Google SecOps 服務帳戶

Google SecOps 服務帳戶需要 GCS bucket 的「Storage 物件檢視者」角色。

1. 依序前往「Cloud Storage」>「Buckets」。
2. 按一下 bucket 名稱 (ukg-logs)。
3. 前往「權限」分頁標籤。
4. 按一下「授予存取權」。
5. 請提供下列設定詳細資料：
   • 新增主體：貼上 Google SecOps 服務帳戶電子郵件地址。
   • 指派角色：選取「Storage 物件檢視者」。

6. 按一下 [儲存]。

UDM 對應表

變更記錄

查看這個剖析器的變更記錄

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