收集 Harness IO 稽核記錄
支援：
Google SecOps
SIEM
本文說明如何使用 Google Cloud Storage，將 Harness IO 稽核記錄擷取至 Google Security Operations。Harness 是持續推送軟體更新和 DevOps 平台，提供軟體推送、功能旗標、雲端成本管理和安全性測試工具。
事前準備
請確認您已完成下列事前準備事項：
- Google SecOps 執行個體
- 已啟用 Cloud Storage API 的 GCP 專案
- 建立及管理 GCS bucket 的權限
- 管理 GCS bucket IAM 政策的權限
- 建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
- 具備 Harness 特殊存取權，可執行下列操作：
- 建立 API 金鑰
- 存取稽核記錄
- 查看帳戶設定
收集 Harness API 憑證
在 Harness 中建立 API 金鑰
- 登入 Harness Platform。
- 按一下您的 使用者個人資料。
- 前往「我的 API 金鑰」。
- 按一下「+ API 金鑰」。
- 提供下列設定詳細資料：
- 名稱：輸入描述性名稱 (例如
Google SecOps Integration)。 - 說明：選填說明。
- 名稱：輸入描述性名稱 (例如
- 按一下 [儲存]。
- 按一下「+ Token」建立新權杖。
- 請提供下列設定詳細資料：
- 「Name」(名稱)：輸入
Chronicle Feed Token。 - 設定到期時間：選取適當的到期時間或「無到期時間」 (用於正式環境)。
- 「Name」(名稱)：輸入
- 按一下 [產生憑證]。
複製並妥善儲存權杖值。這個權杖會做為
x-api-key標頭值。
取得 Harness 帳戶 ID
- 在 Harness 平台中，記下網址中的帳戶 ID。
範例網址：https://app.harness.io/ng/account/YOUR_ACCOUNT_ID/...。YOUR_ACCOUNT_ID 部分是您的帳戶 ID。
或者，前往「帳戶設定」>「總覽」，即可查看「帳戶 ID」。
複製並儲存帳戶 ID，以供 Cloud Run 函式使用。
建立 Google Cloud Storage bucket
- 前往 Google Cloud 控制台。
- 選取專案或建立新專案。
- 在導覽選單中，依序前往「Cloud Storage」>「Bucket」。
- 按一下「建立值區」。
提供下列設定詳細資料：
設定 值 為 bucket 命名 輸入全域不重複的名稱 (例如 harness-io-logs)位置類型 請根據需求選擇 (單一區域、雙區域、多區域) 位置 選取位置 (例如 us-central1)儲存空間級別 標準 (建議用於經常存取的記錄) 存取控管 統一 (建議) 保護工具 選用：啟用物件版本管理或保留政策 點選「建立」。
為 Cloud Run 函式建立服務帳戶
Cloud Run 函式需要具備寫入 GCS bucket 權限，且可由 Pub/Sub 叫用的服務帳戶。
建立服務帳戶
- 在 GCP 控制台中，依序前往「IAM 與管理」>「服務帳戶」。
- 按一下「Create Service Account」(建立服務帳戶)。
- 請提供下列設定詳細資料：
- 「Service account name」(服務帳戶名稱)：輸入
harness-audit-collector-sa。 - 服務帳戶說明：輸入
Service account for Cloud Run function to collect Harness IO audit logs。
- 「Service account name」(服務帳戶名稱)：輸入
- 按一下「建立並繼續」。
- 在「將專案存取權授予這個服務帳戶」部分，新增下列角色：
- 按一下「請選擇角色」。
- 搜尋並選取「Storage 物件管理員」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Run Invoker」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Functions Invoker」(Cloud Functions 叫用者)。
- 按一下「繼續」。
- 按一下「完成」。
這些角色適用於：
- Storage 物件管理員：將記錄檔寫入 GCS bucket 並管理狀態檔案
- Cloud Run 叫用者：允許 Pub/Sub 叫用函式
- Cloud Functions 叫用者：允許叫用函式
授予 GCS bucket 的 IAM 權限
將 GCS bucket 的寫入權限授予服務帳戶：
- 依序前往「Cloud Storage」 > 「Bucket」。
- 按一下 bucket 名稱。
- 前往「權限」分頁。
- 按一下「授予存取權」。
- 提供下列設定詳細資料：
- 新增主體：輸入服務帳戶電子郵件地址 (例如
harness-audit-collector-sa@PROJECT_ID.iam.gserviceaccount.com)。 - 指派角色：選取「Storage 物件管理員」。
- 新增主體：輸入服務帳戶電子郵件地址 (例如
- 按一下 [儲存]。
建立 Pub/Sub 主題
建立 Pub/Sub 主題，Cloud Scheduler 會發布至該主題，Cloud Run 函式則會訂閱該主題。
- 在 GCP 控制台中，前往「Pub/Sub」>「主題」。
- 按一下「建立主題」。
- 提供下列設定詳細資料：
- 主題 ID：輸入
harness-audit-trigger。 - 其他設定保留預設值。
- 主題 ID：輸入
- 點選「建立」。
建立 Cloud Run 函式來收集記錄
Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發，從 Harness API 擷取記錄，並將記錄寫入 GCS。
- 在 GCP 控制台中，前往「Cloud Run」。
- 按一下「Create service」(建立服務)。
- 選取「函式」 (使用內嵌編輯器建立函式)。
在「設定」部分，提供下列設定詳細資料：
設定 值 服務名稱 harness-audit-collector區域 選取與 GCS bucket 相符的區域 (例如 us-central1)執行階段 選取 Python 3.12 以上版本 在「觸發條件 (選用)」部分：
- 按一下「+ 新增觸發條件」。
- 選取「Cloud Pub/Sub」。
- 在「選取 Cloud Pub/Sub 主題」中，選擇 Pub/Sub 主題 (
harness-audit-trigger)。 - 按一下 [儲存]。
在「Authentication」(驗證) 部分：
- 選取「需要驗證」。
- Identity and Access Management (IAM)。
向下捲動並展開「容器、網路、安全性」。
前往「安全性」分頁標籤：
- 服務帳戶：選取服務帳戶 (
harness-audit-collector-sa)。
- 服務帳戶：選取服務帳戶 (
前往「容器」分頁：
- 按一下「變數與密鑰」。
- 針對每個環境變數，按一下「+ 新增變數」：
變數名稱 範例值 說明 HARNESS_ACCOUNT_ID您的 Harness 帳戶 ID Harness 提供的帳戶 ID HARNESS_API_KEY您的 API 金鑰權杖 具備 audit:read 權限的權杖 GCS_BUCKETharness-io-logsGCS bucket 名稱 GCS_PREFIXharness/auditGCS 物件的前置字串 STATE_KEYharness/audit/state.jsonGCS 中的狀態檔案路徑 - 選用環境變數：
變數名稱 預設值 說明 HARNESS_API_BASEhttps://app.harness.ioHarness API 基準網址 (自行代管執行個體的覆寫) PAGE_SIZE50每個網頁的事件數 (最多 100 個) START_MINUTES_BACK60初始回溯期 (以分鐘為單位) FILTER_MODULES無 以半形逗號分隔的模組 (例如 CD,CI,CE)FILTER_ACTIONS無 以半形逗號分隔的動作 (例如 CREATE,UPDATE,DELETE)STATIC_FILTER無 預先定義的篩選器： EXCLUDE_LOGIN_EVENTS或EXCLUDE_SYSTEM_EVENTSMAX_RETRIES3頻率限制的重試次數上限 在「變數與密鑰」分頁中向下捲動至「要求」：
- 要求逾時：輸入
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 time # Initialize HTTP client http = urllib3.PoolManager() # Initialize Storage client storage_client = storage.Client() # Configuration from Environment Variables API_BASE = os.environ.get("HARNESS_API_BASE", "https://app.harness.io").rstrip("/") ACCOUNT_ID = os.environ["HARNESS_ACCOUNT_ID"] API_KEY = os.environ["HARNESS_API_KEY"] BUCKET = os.environ["GCS_BUCKET"] PREFIX = os.environ.get("GCS_PREFIX", "harness/audit").strip("/") STATE_KEY = os.environ.get("STATE_KEY", "harness/audit/state.json") PAGE_SIZE = min(int(os.environ.get("PAGE_SIZE", "50")), 100) START_MINUTES_BACK = int(os.environ.get("START_MINUTES_BACK", "60")) # Optional filters FILTER_MODULES = os.environ.get("FILTER_MODULES", "").split(",") if os.environ.get("FILTER_MODULES") else None FILTER_ACTIONS = os.environ.get("FILTER_ACTIONS", "").split(",") if os.environ.get("FILTER_ACTIONS") else None STATIC_FILTER = os.environ.get("STATIC_FILTER") MAX_RETRIES = int(os.environ.get("MAX_RETRIES", "3")) # HTTP headers for Harness API HDRS = { "x-api-key": API_KEY, "Content-Type": "application/json", "Accept": "application/json", } def read_state(bucket): """Read checkpoint state from GCS.""" try: blob = bucket.blob(STATE_KEY) if blob.exists(): state_data = blob.download_as_text() state = json.loads(state_data) since_ms = state.get("since") page_token = state.get("pageToken") print(f"State loaded: since={since_ms}, pageToken={page_token}") return since_ms, page_token except Exception as e: print(f"Warning: Could not load state: {e}") print("No state file found, starting fresh collection") start_time = datetime.now(timezone.utc) - timedelta(minutes=START_MINUTES_BACK) since_ms = int(start_time.timestamp() * 1000) print(f"Initial since timestamp: {since_ms} ({start_time.isoformat()})") return since_ms, None def write_state(bucket, since_ms, page_token=None): """Write checkpoint state to GCS.""" state = { "since": since_ms, "pageToken": page_token, "lastRun": int(time.time() * 1000), "lastRunISO": datetime.now(timezone.utc).isoformat() } try: blob = bucket.blob(STATE_KEY) blob.upload_from_string( json.dumps(state, indent=2), content_type="application/json" ) print(f"State saved: since={since_ms}, pageToken={page_token}") except Exception as e: print(f"Error writing state: {e}") raise def fetch_harness_audits(since_ms, page_token=None, retry_count=0): """ Fetch audit logs from Harness API with retry logic. API Endpoint: POST /audit/api/audits/listV2 """ try: # Build URL with query parameters url = ( f"{API_BASE}/audit/api/audits/listV2" f"?accountIdentifier={ACCOUNT_ID}" f"&pageSize={PAGE_SIZE}" ) if page_token: url += f"&pageToken={page_token}" print(f"Fetching from: {url[:100]}...") # Build request body with time filter and optional filters body_data = { "startTime": since_ms, "endTime": int(time.time() * 1000), "filterType": "Audit" } if FILTER_MODULES: body_data["modules"] = [m.strip() for m in FILTER_MODULES if m.strip()] print(f"Applying module filter: {body_data['modules']}") if FILTER_ACTIONS: body_data["actions"] = [a.strip() for a in FILTER_ACTIONS if a.strip()] print(f"Applying action filter: {body_data['actions']}") if STATIC_FILTER: body_data["staticFilter"] = STATIC_FILTER print(f"Applying static filter: {STATIC_FILTER}") # Make POST request response = http.request( 'POST', url, body=json.dumps(body_data).encode('utf-8'), headers=HDRS, timeout=30.0 ) resp_data = json.loads(response.data.decode('utf-8')) if "status" not in resp_data: print(f"Response missing 'status' field: {response.data[:200]}") # Check response status if resp_data.get("status") != "SUCCESS": error_msg = resp_data.get("message", "Unknown error") raise Exception(f"API returned status: {resp_data.get('status')} - {error_msg}") # Extract data from response structure data_obj = resp_data.get("data", {}) if not data_obj: print("Response 'data' object is empty or missing") events = data_obj.get("content", []) has_next = data_obj.get("hasNext", False) next_token = data_obj.get("pageToken") print(f"API response: {len(events)} events, hasNext={has_next}, pageToken={next_token}") if not events and data_obj: print(f"Empty events but data present. Data keys: {list(data_obj.keys())}") return { "events": events, "hasNext": has_next, "pageToken": next_token } except Exception as e: if hasattr(e, 'status') and e.status == 429: retry_after = 60 print(f"Rate limit exceeded. Retry after {retry_after} seconds (attempt {retry_count + 1}/{MAX_RETRIES})") if retry_count < MAX_RETRIES: print(f"Waiting {retry_after} seconds before retry...") time.sleep(retry_after) print(f"Retrying request (attempt {retry_count + 2}/{MAX_RETRIES})") return fetch_harness_audits(since_ms, page_token, retry_count + 1) else: raise Exception(f"Max retries ({MAX_RETRIES}) exceeded for rate limiting") print(f"Error in fetch_harness_audits: {e}") raise def upload_to_gcs(bucket, events): """Upload audit events to GCS in JSONL format.""" if not events: print("No events to upload") return None try: # Create JSONL content (one JSON object per line) jsonl_lines = [json.dumps(event) for event in events] jsonl_content = "\n".join(jsonl_lines) # Generate GCS key with timestamp timestamp = datetime.now(timezone.utc) key = ( f"{PREFIX}/" f"{timestamp:%Y/%m/%d}/" f"harness-audit-{timestamp:%Y%m%d-%H%M%S}.jsonl" ) # Upload to GCS blob = bucket.blob(key) blob.upload_from_string( jsonl_content, content_type="application/x-ndjson" ) blob.metadata = { "event-count": str(len(events)), "source": "harness-audit-function", "collection-time": timestamp.isoformat() } blob.patch() print(f"Uploaded {len(events)} events to gs://{BUCKET}/{key}") return key except Exception as e: print(f"Error uploading to GCS: {e}") raise @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch Harness audit logs and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ print("=== Harness Audit Collection Started ===") print(f"Configuration: API_BASE={API_BASE}, ACCOUNT_ID={ACCOUNT_ID[:8]}..., PAGE_SIZE={PAGE_SIZE}") if FILTER_MODULES: print(f"Module filter enabled: {FILTER_MODULES}") if FILTER_ACTIONS: print(f"Action filter enabled: {FILTER_ACTIONS}") if STATIC_FILTER: print(f"Static filter enabled: {STATIC_FILTER}") try: # Get GCS bucket bucket = storage_client.bucket(BUCKET) # Step 1: Read checkpoint state since_ms, page_token = read_state(bucket) if page_token: print("Resuming pagination from saved pageToken") else: since_dt = datetime.fromtimestamp(since_ms / 1000, tz=timezone.utc) print(f"Starting new collection from: {since_dt.isoformat()}") # Step 2: Collect all events with pagination all_events = [] current_page_token = page_token page_count = 0 max_pages = 100 has_next = True while has_next and page_count < max_pages: page_count += 1 print(f"--- Fetching page {page_count} ---") # Fetch one page of results result = fetch_harness_audits(since_ms, current_page_token) # Extract events events = result.get("events", []) all_events.extend(events) print(f"Page {page_count}: {len(events)} events (total: {len(all_events)})") # Check pagination status has_next = result.get("hasNext", False) current_page_token = result.get("pageToken") if not has_next: print("Pagination complete (hasNext=False)") break if not current_page_token: print("hasNext=True but no pageToken, stopping pagination") break # Small delay between pages to avoid rate limiting time.sleep(0.5) if page_count >= max_pages: print(f"Reached max pages limit ({max_pages}), stopping") # Step 3: Upload collected events to GCS if all_events: gcs_key = upload_to_gcs(bucket, all_events) print(f"Successfully uploaded {len(all_events)} total events") else: print("No new events to upload") gcs_key = None # Step 4: Update checkpoint state if not has_next: # Pagination complete - update since to current time for next run new_since = int(time.time() * 1000) write_state(bucket, new_since, None) print(f"Pagination complete, state updated with new since={new_since}") else: # Pagination incomplete - save pageToken for continuation write_state(bucket, since_ms, current_page_token) print("Pagination incomplete, saved pageToken for next run") # Step 5: Log result result = { "status": "Success", "eventsCollected": len(all_events), "pagesProcessed": page_count, "paginationComplete": not has_next, "gcsKey": gcs_key, "filters": { "modules": FILTER_MODULES, "actions": FILTER_ACTIONS, "staticFilter": STATIC_FILTER } } print(f"Collection completed: {json.dumps(result)}") except Exception as e: print(f"Collection failed: {e}") raise finally: print("=== Harness Audit Collection Finished ===")- 第二個檔案：requirements.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」。
- 點選「建立工作」。
提供下列設定詳細資料：
設定 值 名稱 harness-audit-hourly區域 選取與 Cloud Run 函式相同的區域 頻率 0 * * * *(每小時整點)時區 選取時區 (建議選取世界標準時間) 目標類型 Pub/Sub 主題 選取 Pub/Sub 主題 ( harness-audit-trigger)郵件內文 {}(空白 JSON 物件)點選「建立」。
排程頻率選項
根據記錄檔量和延遲時間需求選擇頻率：
頻率 Cron 運算式 用途 每 5 分鐘 */5 * * * *高容量、低延遲 每 15 分鐘 */15 * * * *中等音量 每小時 0 * * * *標準 (建議) 每 6 小時 0 */6 * * *低量、批次處理 每天 0 0 * * *歷來資料集合
測試整合項目
- 在 Cloud Scheduler 控制台中找到工作。
- 按一下「強制執行」，手動觸發工作。
- 稍等幾秒鐘。
- 依序前往「Cloud Run」 > 「服務」。
- 按一下函式名稱 (
harness-audit-collector)。 - 點選「記錄」分頁標籤。
確認函式是否執行成功。請找出以下項目：
=== Harness Audit Collection Started === State loaded: since=... or No state file found, starting fresh collection --- Fetching page 1 --- API response: X events, hasNext=... Uploaded X events to gs://harness-io-logs/harness/audit/... Successfully processed X records === Harness Audit Collection Finished ===依序前往「Cloud Storage」 > 「Bucket」。
按一下 bucket 名稱。
前往前置字元資料夾 (
harness/audit/)。確認系統已建立新的
.jsonl檔案，並加上目前的時間戳記。
如果記錄中顯示錯誤：
- HTTP 401：檢查環境變數中的 API 憑證
- HTTP 403：確認帳戶是否具備必要權限
- HTTP 429：頻率限制 - 函式會自動重試，並採用輪詢機制
缺少環境變數：檢查是否已設定所有必要變數
擷取 Google SecOps 服務帳戶
Google SecOps 會使用專屬服務帳戶，從 GCS bucket 讀取資料。您必須授予這個服務帳戶值區存取權。
取得服務帳戶電子郵件地址
- 依序前往 「SIEM 設定」 > 「動態饋給」。
- 按一下「新增動態饋給」。
- 按一下「設定單一動態饋給」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
Harness Audit Logs)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 選取「Harness IO」做為「記錄類型」。
按一下「取得服務帳戶」。系統會顯示專屬服務帳戶電子郵件地址，例如：
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com複製這個電子郵件地址，以便在下一步中使用。
將 IAM 權限授予 Google SecOps 服務帳戶
Google SecOps 服務帳戶需要 GCS bucket 的「Storage 物件檢視者」角色。
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下 bucket 名稱。
- 前往「權限」分頁。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料：
- 新增主體：貼上 Google SecOps 服務帳戶電子郵件地址。
- 指派角色：選取「Storage 物件檢視者」。
按一下 [儲存]。
在 Google SecOps 中設定動態饋給，擷取 Harness IO 記錄
- 依序前往「SIEM 設定」>「動態饋給」。
- 按一下「新增動態饋給」。
- 按一下「設定單一動態饋給」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
Harness Audit Logs)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 選取「Harness IO」做為「記錄類型」。
- 點選「下一步」。
指定下列輸入參數的值：
儲存空間 bucket 網址：輸入 GCS bucket URI，並加上前置路徑：
gs://harness-io-logs/harness/audit/更改項目：
harness-io-logs：您的 GCS bucket 名稱。harness/audit：：儲存記錄的前置字元/資料夾路徑。
範例：
- 根 bucket：
gs://company-logs/ - 加上前置字元：
gs://company-logs/harness-logs/ - 含子資料夾：
gs://company-logs/harness/audit/
- 根 bucket：
來源刪除選項：根據偏好選取刪除選項：
- 永不：一律不會在轉移後刪除任何檔案 (建議用於測試)。
- 刪除已轉移的檔案：在成功轉移後刪除檔案。
刪除已轉移的檔案和空目錄：成功轉移後，刪除檔案和空目錄。
檔案存在時間上限：納入在過去天數內修改的檔案。預設值為 180 天。
資產命名空間：資產命名空間。輸入
harness.audit。擷取標籤：要套用至這個動態饋給事件的標籤。
點選「下一步」。
在「Finalize」(完成) 畫面中檢查新的動態饋給設定，然後按一下「Submit」(提交)。
變更記錄
還有其他問題嗎？向社群成員和 Google SecOps 專業人員尋求解答。