收集 Rippling 活動記錄

支援的國家/地區:

本文說明如何使用 Google Cloud Storage,將 Rippling 活動記錄擷取至 Google Security Operations。Rippling 是員工管理平台,提供人資、IT 和財務解決方案,包括薪資、福利、新進人員訓練、裝置管理和應用程式佈建。透過 Company Activity API,您可以取得 Rippling 平台中管理員和使用者動作的稽核記錄。

事前準備

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

  • Google SecOps 執行個體
  • 已啟用 Cloud Storage API 的 GCP 專案
  • 建立及管理 GCS 值區的權限
  • 管理 Google Cloud Storage 值區 IAM 政策的權限
  • 建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
  • Rippling 的特殊存取權 (可存取公司活動的 API 權杖)

建立 Google Cloud Storage 值區

  1. 前往 Google Cloud 控制台
  2. 選取專案或建立新專案。
  3. 在導覽選單中,依序前往「Cloud Storage」>「Bucket」
  4. 按一下「建立值區」
  5. 請提供下列設定詳細資料:

    設定
    為 bucket 命名 輸入全域不重複的名稱 (例如 rippling-activity-logs)
    位置類型 根據需求選擇 (區域、雙區域、多區域)
    位置 選取地點 (例如 us-central1)
    儲存空間級別 標準 (建議用於經常存取的記錄)
    存取控管 統一 (建議)
    保護工具 選用:啟用物件版本管理或保留政策
  6. 點選「建立」

收集 Rippling API 憑證

  1. 登入 Rippling 管理員
  2. 依序前往「Search」>「API Tokens」
    • 替代路徑:依序點選「設定」>「公司設定」>「API 權杖」
  3. 按一下「Create API Token」(建立 API 憑證)
  4. 請提供下列設定詳細資料:
    • 名稱:輸入有意義的不重複名稱 (例如 Google SecOps GCS Export)。
    • API 版本:選取「Base API (v1)」
    • 範圍/權限:啟用 company:activity:read (公司活動的必要權限)。
  5. 點選「建立」
  6. 複製權杖值並儲存在安全的位置。您會將其做為不記名權杖使用。

為 Cloud Run 函式建立服務帳戶

Cloud Run 函式需要具備 GCS bucket 寫入權限的服務帳戶。

建立服務帳戶

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

這些角色適用於:

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

授予 GCS 值區的 IAM 權限

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

  1. 依序前往「Cloud Storage」>「Buckets」
  2. 按一下 bucket 名稱。
  3. 前往「權限」分頁標籤。
  4. 按一下「授予存取權」
  5. 請提供下列設定詳細資料:
    • 新增主體:輸入服務帳戶電子郵件地址 (例如 rippling-logs-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:輸入 rippling-activity-trigger
    • 其他設定保留預設值。
  4. 點選「建立」

建立 Cloud Run 函式來收集記錄

Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發,從 Rippling Company Activity API 擷取記錄,並寫入 GCS。

  1. 前往 GCP Console 的「Cloud Run」
  2. 按一下「Create service」(建立服務)
  3. 選取「函式」 (使用內嵌編輯器建立函式)。
  4. 在「設定」部分,提供下列設定詳細資料:

    設定
    服務名稱 rippling-activity-collector
    區域 選取與 GCS bucket 相符的區域 (例如 us-central1)
    執行階段 選取「Python 3.12」以上版本
  5. 在「Trigger (optional)」(觸發條件 (選用)) 專區:

    1. 按一下「+ 新增觸發條件」
    2. 選取「Cloud Pub/Sub」
    3. 在「選取 Cloud Pub/Sub 主題」中,選擇主題 rippling-activity-trigger
    4. 按一下 [儲存]
  6. 在「Authentication」(驗證) 部分:

    1. 選取「需要驗證」
    2. 檢查 Identity and Access Management (IAM)
  7. 向下捲動並展開「Containers, Networking, Security」

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

    • 服務帳戶:選取服務帳戶 rippling-logs-collector-sa
  9. 前往「容器」分頁:

    1. 按一下「變數與密鑰」
    2. 針對每個環境變數,按一下「+ 新增變數」
    變數名稱 範例值
    GCS_BUCKET rippling-activity-logs
    GCS_PREFIX rippling/activity/
    STATE_KEY rippling/activity/state.json
    RIPPLING_API_TOKEN your-api-token
    RIPPLING_ACTIVITY_URL https://api.rippling.com/platform/api/company_activity
    LIMIT 1000
    MAX_PAGES 10
    LOOKBACK_MINUTES 60
    END_LAG_SECONDS 120
  10. 在「變數與密鑰」分頁中向下捲動至「要求」

    • 要求逾時:輸入 600 秒 (10 分鐘)。
  11. 前往「容器」中的「設定」分頁:

    • 在「資源」部分:
      • 記憶體:選取 512 MiB 以上。
      • CPU:選取 1
    • 按一下 [完成]。
  12. 捲動至「執行環境」

    • 選取「預設」 (建議選項)。
  13. 在「修訂版本資源調度」部分:

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

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

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

新增函式程式碼

  1. 在「Function entry point」(函式進入點) 中輸入 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
    
    # Initialize HTTP client
    http = urllib3.PoolManager(timeout=urllib3.Timeout(connect=5.0, read=60.0), retries=False)
    
    # Initialize Storage client
    storage_client = storage.Client()
    
    @functions_framework.cloud_event
    def main(cloud_event):
        """
        Cloud Run function triggered by Pub/Sub to fetch logs from Rippling Company Activity API 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', 'rippling/activity/')
        state_key = os.environ.get('STATE_KEY', 'rippling/activity/state.json')
    
        # Rippling API configuration
        api_token = os.environ.get('RIPPLING_API_TOKEN')
        activity_url = os.environ.get('RIPPLING_ACTIVITY_URL', 'https://api.rippling.com/platform/api/company_activity')
        limit = int(os.environ.get('LIMIT', '1000'))
        max_pages = int(os.environ.get('MAX_PAGES', '10'))
        lookback_minutes = int(os.environ.get('LOOKBACK_MINUTES', '60'))
        end_lag_seconds = int(os.environ.get('END_LAG_SECONDS', '120'))
    
        if not all([bucket_name, api_token]):
            print('Error: Missing required environment variables')
            return
    
        try:
            # Get GCS bucket
            bucket = storage_client.bucket(bucket_name)
    
            # Load state (last processed timestamp and cursor)
            state = load_state(bucket, state_key)
            since_iso = state.get('since')
            next_cursor = state.get('next')
    
            # Calculate time window
            run_end = datetime.now(timezone.utc) - timedelta(seconds=end_lag_seconds)
            end_iso = run_end.replace(microsecond=0).isoformat().replace('+00:00', 'Z')
    
            if since_iso is None:
                since_iso = iso_from_epoch(time.time() - lookback_minutes * 60)
            else:
                try:
                    since_iso = (parse_iso(since_iso) + timedelta(seconds=1)).replace(microsecond=0).isoformat().replace('+00:00', 'Z')
                except Exception:
                    since_iso = iso_from_epoch(time.time() - lookback_minutes * 60)
    
            print(f'Processing logs from {since_iso} to {end_iso}')
    
            run_ts_iso = end_iso
            pages = 0
            total = 0
            newest_ts = None
            pending_next = None
    
            # Fetch logs with pagination
            while pages < max_pages:
                params = {'limit': str(limit)}
    
                if next_cursor:
                    params['next'] = next_cursor
                else:
                    params['startDate'] = since_iso
                    params['endDate'] = end_iso
    
                # Build URL with query parameters
                url = build_url(activity_url, params)
    
                # Fetch data from Rippling API
                headers = {
                    'Authorization': f'Bearer {api_token}',
                    'Accept': 'application/json'
                }
    
                # Implement exponential backoff for rate limiting
                backoff = 1.0
                max_retries = 3
                retry_count = 0
    
                while retry_count < max_retries:
                    response = http.request('GET', url, headers=headers, timeout=60.0)
    
                    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)
                        retry_count += 1
                        continue
    
                    break
    
                if response.status != 200:
                    print(f'Error: API returned status {response.status}')
                    break
    
                data = json.loads(response.data.decode('utf-8'))
    
                # Write page to GCS
                write_to_gcs(bucket, prefix, data, run_ts_iso, pages)
    
                # Extract events
                events = data.get('events') or []
                total += len(events) if isinstance(events, list) else 0
    
                # Track newest timestamp
                if isinstance(events, list):
                    for ev in events:
                        t = ev.get('timestamp') or ev.get('time') or ev.get('event_time')
                        if isinstance(t, str):
                            try:
                                dt_ts = parse_iso(t)
                                if newest_ts is None or dt_ts > newest_ts:
                                    newest_ts = dt_ts
                            except Exception:
                                pass
    
                # Check for next page
                nxt = data.get('next')
                pages += 1
    
                if nxt:
                    next_cursor = nxt
                    pending_next = nxt
                    continue
                else:
                    pending_next = None
                    break
    
            # Update state
            new_since_iso = (newest_ts or run_end).replace(microsecond=0).isoformat().replace('+00:00', 'Z')
            save_state(bucket, state_key, {'since': new_since_iso, 'next': pending_next})
    
            print(f'Successfully processed {total} events across {pages} pages')
            print(f'Updated state: since={new_since_iso}, next={pending_next}')
    
        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: {str(e)}')
        return {}
    
    def save_state(bucket, key, state):
        """Save state to GCS."""
        try:
            blob = bucket.blob(key)
            blob.upload_from_string(
                json.dumps(state, separators=(',', ':')),
                content_type='application/json'
            )
        except Exception as e:
            print(f'Warning: Could not save state: {str(e)}')
    
    def write_to_gcs(bucket, prefix, payload, run_ts_iso, page_index):
        """Write payload to GCS."""
        try:
            day_path = parse_iso(run_ts_iso).strftime('%Y/%m/%d')
            key = f"{prefix.strip('/')}/{day_path}/{run_ts_iso.replace(':', '').replace('-', '')}-page{page_index:05d}-company_activity.json"
    
            blob = bucket.blob(key)
            blob.upload_from_string(
                json.dumps(payload, separators=(',', ':')),
                content_type='application/json'
            )
            print(f'Wrote page {page_index} to {key}')
        except Exception as e:
            print(f'Error writing to GCS: {str(e)}')
            raise
    
    def parse_iso(ts):
        """Parse ISO 8601 timestamp."""
        if ts.endswith('Z'):
            ts = ts[:-1] + '+00:00'
        return datetime.fromisoformat(ts)
    
    def iso_from_epoch(sec):
        """Convert epoch seconds to ISO 8601 timestamp."""
        return datetime.fromtimestamp(sec, tz=timezone.utc).replace(microsecond=0).isoformat().replace('+00:00', 'Z')
    
    def build_url(base, params):
        """Build URL with query parameters."""
        if not params:
            return base
        query_string = '&'.join([f'{k}={v}' for k, v in params.items()])
        return f'{base}?{query_string}'
    
    • 第二個檔案: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. 請提供下列設定詳細資料:

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

排程頻率選項

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

    頻率 Cron 運算式 用途
    每 5 分鐘 */5 * * * * 高容量、低延遲
    每 15 分鐘檢查一次 */15 * * * * 普通量
    每小時 0 * * * * 標準 (建議採用)
    每 6 小時 0 */6 * * * 少量、批次處理
    每日 0 0 * * * 歷來資料集合

測試排程器工作

  1. Cloud Scheduler 控制台中找出您的工作。
  2. 按一下「強制執行」即可手動觸發。
  3. 等待幾秒鐘,然後依序前往「Cloud Run」>「Services」>「rippling-activity-collector」>「Logs」
  4. 確認函式是否已順利執行。
  5. 檢查 GCS 值區,確認是否已寫入記錄。

擷取 Google SecOps 服務帳戶

Google SecOps 會使用專屬服務帳戶,從 GCS bucket 讀取資料。您必須授予這個服務帳戶值區存取權。

取得服務帳戶電子郵件地址

  1. 依序前往「SIEM 設定」>「動態饋給」
  2. 按一下「新增動態消息」
  3. 按一下「設定單一動態饋給」
  4. 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Rippling Activity Logs)。
  5. 選取「Google Cloud Storage V2」做為「來源類型」
  6. 選取「Rippling 活動記錄」做為「記錄類型」
  7. 按一下「取得服務帳戶」。系統會顯示不重複的服務帳戶電子郵件地址,例如:

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
    
  8. 複製這個電子郵件地址,以便在下一步中使用。

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

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

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

在 Google SecOps 中設定動態消息,擷取 Rippling 活動記錄

  1. 依序前往「SIEM 設定」>「動態饋給」
  2. 按一下「新增動態消息」
  3. 按一下「設定單一動態饋給」
  4. 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Rippling Activity Logs)。
  5. 選取「Google Cloud Storage V2」做為「來源類型」
  6. 選取「Rippling 活動記錄」做為「記錄類型」
  7. 點選 [下一步]。
  8. 指定下列輸入參數的值:

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

      gs://rippling-activity-logs/rippling/activity/
      
      • 取代:

        • rippling-activity-logs:您的 GCS bucket 名稱。
        • rippling/activity/:儲存記錄的前置字元/資料夾路徑 (必須與 GCS_PREFIX 環境變數相符)。
    • 來源刪除選項:根據偏好設定選取刪除選項:

      • 永不:移轉後一律不刪除任何檔案 (建議用於測試)。
      • 刪除已轉移的檔案:成功轉移檔案後刪除檔案。
      • 刪除已轉移的檔案和空白目錄:成功轉移後刪除檔案和空白目錄。

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

    • 資產命名空間資產命名空間 (例如 rippling.activity)。

    • 擷取標籤:可選用標籤,用於這個動態饋給的事件。

  9. 點選 [下一步]。

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

需要其他協助嗎?向社群成員和 Google SecOps 專業人員尋求答案。