Harness IO 監査ログを収集する
以下でサポートされています。
Google SecOps
SIEM
このドキュメントでは、Amazon S3 を使用して Harness IO 監査ログを Google Security Operations に取り込む方法について説明します。
始める前に
次の前提条件を満たしていることを確認してください。
- Google SecOps インスタンス
- 次の権限を持つ Harness への特権アクセス:
- API キーを作成する
- アクセス監査ログ
- アカウント設定を表示する
- AWS（S3、IAM、Lambda、EventBridge）への特権アクセス。
Harness API 認証情報を収集する
Harness で API キーを作成する
- Harness Platform にログインします。
- [ユーザー プロフィール] をクリックします。
- [My API Keys] に移動します。
- [+ API キー] をクリックします。
- 次の構成の詳細を入力します。
- 名前: わかりやすい名前を入力します（例:
Google SecOps Integration）。 - 説明: 説明（省略可）。
- 名前: わかりやすい名前を入力します（例:
- [保存] をクリックします。
- [+ トークン] をクリックして、新しいトークンを作成します。
- 次の構成の詳細を入力します。
- 名前: 「
Chronicle Feed Token」と入力します。 - 有効期限を設定: 適切な有効期限または [有効期限なし]（本番環境用）を選択します。
- 名前: 「
- [Generate Token] をクリックします。
トークン値をコピーして安全に保存します。このトークンは
x-api-keyヘッダー値として使用されます。
Harness アカウント ID を取得する
- Harness Platform で、URL のアカウント ID をメモします。
- URL の例:
https://app.harness.io/ng/account/YOUR_ACCOUNT_ID/... YOUR_ACCOUNT_IDの部分はアカウント ID です。
- URL の例:
- または、[アカウント設定> 概要] に移動して、アカウント ID を確認することもできます。
Lambda 関数で使用するために、アカウント ID をコピーして保存します。
Google SecOps 用に AWS S3 バケットと IAM を構成する
- バケットの作成のユーザーガイドに沿って、Amazon S3 バケットを作成します。
- 後で参照できるように、バケットの名前とリージョンを保存します（例:
harness-io-logs）。 - IAM ユーザーの作成のユーザーガイドに沿って、ユーザーを作成します。
- 作成した [User] を選択します。
- [セキュリティ認証情報] タブを選択します。
- [アクセスキー] セクションで [アクセスキーを作成] をクリックします。
- [ユースケース] として [サードパーティ サービス] を選択します。
- [次へ] をクリックします。
- 省略可: 説明タグを追加します。
- [アクセスキーを作成] をクリックします。
- [CSV ファイルをダウンロード] をクリックし、[アクセスキー] と [シークレット アクセスキー] を保存して、今後の参照に備えます。
- [完了] をクリックします。
- [権限] タブを選択します。
- [権限ポリシー] セクションの [権限を追加] をクリックします。
- [権限を追加] を選択します。
- [ポリシーを直接アタッチする] を選択します。
- AmazonS3FullAccess ポリシーを検索します。
- ポリシーを選択します。
- [次へ] をクリックします。
- [権限を追加] をクリックします。
Lambda S3 アップロードの IAM ポリシーとロールを構成する
- AWS コンソールで、[IAM] > [ポリシー] > [ポリシーの作成] > [JSON] タブ に移動します。
次のポリシーをコピーして貼り付けます。
{ "Version": "2012-10-17", "Statement": [ { "Sid": "AllowPutHarnessObjects", "Effect": "Allow", "Action": "s3:PutObject", "Resource": "arn:aws:s3:::harness-io-logs/harness/audit/*" }, { "Sid": "AllowGetStateObject", "Effect": "Allow", "Action": "s3:GetObject", "Resource": "arn:aws:s3:::harness-io-logs/harness/audit/state.json" } ] }- 別のバケット名を入力した場合は、
harness-io-logsを置き換えます。
- 別のバケット名を入力した場合は、
[次へ] をクリックします。
ポリシーに
HarnessToS3Policyという名前を付けて、[ポリシーを作成] をクリックします。[IAM]> [ロール]> [ロールを作成] に移動します。
信頼できるエンティティ タイプとして [AWS service] を選択します。
ユースケースとして [Lambda] を選択します。
[次へ] をクリックします。
次のポリシーを検索して選択します。
HarnessToS3Policy（作成したポリシー）AWSLambdaBasicExecutionRole（CloudWatch Logs の場合）
[次へ] をクリックします。
ロールに「
HarnessAuditLambdaRole」という名前を付けて、[ロールを作成] をクリックします。
Lambda 関数を作成する
- AWS コンソールで、[Lambda] > [Functions] > [Create function] に移動します。
- [Author from scratch] をクリックします。
次の構成情報を提供してください。
設定 値 名前 harness-audit-to-s3ランタイム Python 3.13 アーキテクチャ x86_64 実行ロール HarnessAuditLambdaRole[関数を作成] をクリックします。
関数を作成したら、[コード] タブを開きます。
デフォルトのスタブコードを削除し、次の Lambda 関数コードを入力します。
Lambda 関数コード（
harness_audit_to_s3.py）#!/usr/bin/env python3 """ Harness.io Audit Logs to S3 Lambda Fetches audit logs from Harness API and writes to S3 for Chronicle ingestion. """ import os import json import time import uuid import logging import urllib.parse from datetime import datetime, timedelta, timezone from urllib.request import Request, urlopen from urllib.error import HTTPError, URLError import boto3 # 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["S3_BUCKET"] PREFIX = os.environ.get("S3_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 (NEW) 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") # e.g., "EXCLUDE_LOGIN_EVENTS" MAX_RETRIES = int(os.environ.get("MAX_RETRIES", "3")) # AWS clients s3 = boto3.client("s3") # HTTP headers for Harness API HDRS = { "x-api-key": API_KEY, "Content-Type": "application/json", "Accept": "application/json", } # Logging configuration logger = logging.getLogger() logger.setLevel(logging.INFO) # ============================================ # State Management Functions # ============================================ def _read_state(): """Read checkpoint state from S3.""" try: obj = s3.get_object(Bucket=BUCKET, Key=STATE_KEY) state = json.loads(obj["Body"].read()) since_ms = state.get("since") page_token = state.get("pageToken") logger.info(f"State loaded: since={since_ms}, pageToken={page_token}") return since_ms, page_token except s3.exceptions.NoSuchKey: logger.info("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) logger.info(f"Initial since timestamp: {since_ms} ({start_time.isoformat()})") return since_ms, None except Exception as e: logger.error(f"Error reading state: {e}") raise def _write_state(since_ms: int, page_token: str = None): """Write checkpoint state to S3.""" state = { "since": since_ms, "pageToken": page_token, "lastRun": int(time.time() * 1000), "lastRunISO": datetime.now(timezone.utc).isoformat() } try: s3.put_object( Bucket=BUCKET, Key=STATE_KEY, Body=json.dumps(state, indent=2).encode(), ContentType="application/json" ) logger.info(f"State saved: since={since_ms}, pageToken={page_token}") except Exception as e: logger.error(f"Error writing state: {e}") raise # ============================================ # Harness API Functions # ============================================ def _fetch_harness_audits(since_ms: int, page_token: str = None, retry_count: int = 0): """ Fetch audit logs from Harness API with retry logic. API Endpoint: POST /audit/api/audits/listV2 Documentation: https://apidocs.harness.io/audit/getauditeventlistv2 """ try: # Build URL with query parameters url = ( f"{API_BASE}/audit/api/audits/listV2" f"?accountIdentifier={urllib.parse.quote(ACCOUNT_ID)}" f"&pageSize={PAGE_SIZE}" ) if page_token: url += f"&pageToken={urllib.parse.quote(page_token)}" logger.info(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()] logger.info(f"Applying module filter: {body_data['modules']}") if FILTER_ACTIONS: body_data["actions"] = [a.strip() for a in FILTER_ACTIONS if a.strip()] logger.info(f"Applying action filter: {body_data['actions']}") if STATIC_FILTER: body_data["staticFilter"] = STATIC_FILTER logger.info(f"Applying static filter: {STATIC_FILTER}") logger.debug(f"Request body: {json.dumps(body_data)}") # Make POST request req = Request( url, data=json.dumps(body_data).encode('utf-8'), headers=HDRS, method="POST" ) resp = urlopen(req, timeout=30) resp_text = resp.read().decode('utf-8') resp_data = json.loads(resp_text) if "status" not in resp_data: logger.warning(f"Response missing 'status' field: {resp_text[: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: logger.warning("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") logger.info(f"API response: {len(events)} events, hasNext={has_next}, pageToken={next_token}") if not events and data_obj: logger.info(f"Empty events but data present. Data keys: {list(data_obj.keys())}") return { "events": events, "hasNext": has_next, "pageToken": next_token } except HTTPError as e: error_body = e.read().decode() if hasattr(e, 'read') else '' if e.code == 401: logger.error("Authentication failed: Invalid API key") raise Exception("Invalid Harness API key. Check HARNESS_API_KEY environment variable.") elif e.code == 403: logger.error("Authorization failed: Insufficient permissions") raise Exception("API key lacks required audit:read permissions") elif e.code == 429: retry_after = int(e.headers.get("Retry-After", "60")) logger.warning(f"Rate limit exceeded. Retry after {retry_after} seconds (attempt {retry_count + 1}/{MAX_RETRIES})") if retry_count < MAX_RETRIES: logger.info(f"Waiting {retry_after} seconds before retry...") time.sleep(retry_after) logger.info(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") elif e.code == 400: logger.error(f"Bad request: {error_body}") raise Exception(f"Invalid request parameters: {error_body}") else: logger.error(f"HTTP {e.code}: {e.reason} - {error_body}") raise Exception(f"Harness API error {e.code}: {e.reason}") except URLError as e: logger.error(f"Network error: {e.reason}") raise Exception(f"Failed to connect to Harness API: {e.reason}") except json.JSONDecodeError as e: logger.error(f"Invalid JSON response: {e}") logger.error(f"Response text (first 500 chars): {resp_text[:500] if 'resp_text' in locals() else 'N/A'}") raise Exception("Harness API returned invalid JSON") except Exception as e: logger.error(f"Unexpected error in _fetch_harness_audits: {e}", exc_info=True) raise # ============================================ # S3 Upload Functions # ============================================ def _upload_to_s3(events: list) -> str: """ Upload audit events to S3 in JSONL format. Each line is a complete JSON object (one event per line). """ if not events: logger.info("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 S3 key with timestamp and UUID timestamp = datetime.now(timezone.utc) key = ( f"{PREFIX}/" f"{timestamp:%Y/%m/%d}/" f"harness-audit-{timestamp:%Y%m%d-%H%M%S}-{uuid.uuid4()}.jsonl" ) # Upload to S3 s3.put_object( Bucket=BUCKET, Key=key, Body=jsonl_content.encode('utf-8'), ContentType="application/x-ndjson", Metadata={ "event-count": str(len(events)), "source": "harness-audit-lambda", "collection-time": timestamp.isoformat() } ) logger.info(f"Uploaded {len(events)} events to s3://{BUCKET}/{key}") return key except Exception as e: logger.error(f"Error uploading to S3: {e}", exc_info=True) raise # ============================================ # Main Orchestration Function # ============================================ def fetch_and_store(): """ Main function to fetch audit logs from Harness and store in S3. Handles pagination and state management. """ logger.info("=== Harness Audit Collection Started ===") logger.info(f"Configuration: API_BASE={API_BASE}, ACCOUNT_ID={ACCOUNT_ID[:8]}..., PAGE_SIZE={PAGE_SIZE}") if FILTER_MODULES: logger.info(f"Module filter enabled: {FILTER_MODULES}") if FILTER_ACTIONS: logger.info(f"Action filter enabled: {FILTER_ACTIONS}") if STATIC_FILTER: logger.info(f"Static filter enabled: {STATIC_FILTER}") try: # Step 1: Read checkpoint state since_ms, page_token = _read_state() if page_token: logger.info(f"Resuming pagination from saved pageToken") else: since_dt = datetime.fromtimestamp(since_ms / 1000, tz=timezone.utc) logger.info(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 # Safety limit has_next = True while has_next and page_count < max_pages: page_count += 1 logger.info(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) logger.info(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: logger.info("Pagination complete (hasNext=False)") break if not current_page_token: logger.warning("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: logger.warning(f"Reached max pages limit ({max_pages}), stopping") # Step 3: Upload collected events to S3 if all_events: s3_key = _upload_to_s3(all_events) logger.info(f"Successfully uploaded {len(all_events)} total events") else: logger.info("No new events to upload") s3_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(new_since, None) logger.info(f"Pagination complete, state updated with new since={new_since}") else: # Pagination incomplete - save pageToken for continuation _write_state(since_ms, current_page_token) logger.info(f"Pagination incomplete, saved pageToken for next run") # Step 5: Return result result = { "statusCode": 200, "message": "Success", "eventsCollected": len(all_events), "pagesProcessed": page_count, "paginationComplete": not has_next, "s3Key": s3_key, "filters": { "modules": FILTER_MODULES, "actions": FILTER_ACTIONS, "staticFilter": STATIC_FILTER } } logger.info(f"Collection completed: {json.dumps(result)}") return result except Exception as e: logger.error(f"Collection failed: {e}", exc_info=True) result = { "statusCode": 500, "message": "Error", "error": str(e), "errorType": type(e).__name__ } return result finally: logger.info("=== Harness Audit Collection Finished ===") # ============================================ # Lambda Handler # ============================================ def lambda_handler(event, context): """AWS Lambda handler function.""" return fetch_and_store() # ============================================ # Local Testing # ============================================ if __name__ == "__main__": # For local testing result = lambda_handler(None, None) print(json.dumps(result, indent=2))
[デプロイ] をクリックして、関数コードを保存します。
Lambda 環境変数を構成する
- Lambda 関数のページで、[構成] タブを選択します。
- 左側のサイドバーで [環境変数] をクリックします。
- [編集] をクリックします。
次のそれぞれについて、[環境変数を追加] をクリックします。
必要な環境変数:
キー 値 説明 HARNESS_ACCOUNT_IDHarness アカウント ID Harness のアカウント ID HARNESS_API_KEYAPI キー トークン audit:read 権限を持つトークン S3_BUCKETharness-io-logsS3 バケット名 S3_PREFIXharness/auditS3 オブジェクトの接頭辞 STATE_KEYharness/audit/state.jsonS3 の状態ファイルのパス 省略可能な環境変数:
キー デフォルト値 説明 HARNESS_API_BASEhttps://app.harness.ioHarness API のベース URL 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レート制限の最大再試行回数 [保存] をクリックします。
Lambda のタイムアウトとメモリを構成する
- Lambda 関数のページで、[構成] タブを選択します。
- 左側のサイドバーで [全般設定] をクリックします。
- [編集] をクリックします。
- 次の構成の詳細を入力します。
- メモリ:
256 MB（推奨） - タイムアウト:
5 min 0 sec（300 秒）
- メモリ:
- [保存] をクリックします。
EventBridge スケジュールを作成する
- [Amazon EventBridge] > [Scheduler] > [スケジュールの作成] に移動します。
- 次の構成の詳細を入力します。
- スケジュール名: 「
harness-audit-hourly」と入力します。 - 説明: 説明（省略可）。
- スケジュール名: 「
- [次へ] をクリックします。
- [スケジュール パターン] で、[定期的なスケジュール] を選択します。
- [レートベースのスケジュール] を選択します。
- 次の構成の詳細を入力します。
- レート式: 「
1 hour」と入力します。
- レート式: 「
- [次へ] をクリックします。
- [ターゲット] で、次の構成の詳細を指定します。
- ターゲット API: [AWS Lambda 呼び出し] を選択します。
- Lambda 関数: 関数
harness-audit-to-s3を選択します。
- [次へ] をクリックします。
- スケジュール構成を確認します。
- [スケジュールを作成] をクリックします。
Google SecOps 用の読み取り専用 IAM ユーザーを作成する
この IAM ユーザーにより、Google SecOps は S3 バケットからログを読み取ることができます。
- AWS コンソール > IAM > ユーザー > ユーザーを作成 に移動します。
- 次の構成の詳細を入力します。
- ユーザー名: 「
chronicle-s3-reader」と入力します。
- ユーザー名: 「
- [次へ] をクリックします。
- [ポリシーを直接アタッチする] を選択します。
- [ポリシーを作成] をクリックします。
- [JSON] タブを選択します。
次のポリシーを貼り付けます。
{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "s3:GetObject" ], "Resource": "arn:aws:s3:::harness-io-logs/harness/audit/*" }, { "Effect": "Allow", "Action": [ "s3:ListBucket" ], "Resource": "arn:aws:s3:::harness-io-logs", "Condition": { "StringLike": { "s3:prefix": "harness/audit/*" } } } ] }[次へ] をクリックします。
ポリシーに
ChronicleHarnessS3ReadPolicyという名前を付けます。[ポリシーを作成] をクリックします。
ユーザー作成タブに戻り、ポリシーのリストを更新します。
ChronicleHarnessS3ReadPolicyを検索して選択します。[次へ] をクリックします。
確認して [ユーザーを作成] をクリックします。
閲覧者ユーザーのアクセスキーを作成する
- [IAM ユーザー] ページで、
chronicle-s3-readerユーザーを選択します。 - [セキュリティ認証情報] タブを選択します。
- [アクセスキーを作成] をクリックします。
- ユースケースとして [サードパーティ サービス] を選択します。
- [次へ] をクリックします。
- 省略可: 説明タグを追加します。
- [アクセスキーを作成] をクリックします。
- [CSV ファイルをダウンロード] をクリックして、アクセスキー ID とシークレット アクセスキーを保存します。
- [完了] をクリックします。
Harness IO のログを取り込むように Google SecOps でフィードを構成する
- [SIEM 設定] > [フィード] に移動します。
- [新しく追加] をクリックします。
- 次のページで [単一のフィードを設定] をクリックします。
- [フィード名] フィールドに、フィードの名前を入力します（例:
Harness Audit Logs）。 - [ソースタイプ] として [Amazon S3 V2] を選択します。
- [ログタイプ] として [Harness IO] を選択します。
- [次へ] をクリックします。
次の入力パラメータの値を指定します。
- S3 URI: 接頭辞パスを含む S3 バケット URI を入力します。
s3://harness-io-logs/harness/audit/ Source deletion option: 必要に応じて削除オプションを選択します。
- なし: 転送後にファイルを削除しません（初期設定として推奨）。
- 成功時: 転送が成功すると、すべてのファイルと空のディレクトリが削除されます。
ファイルの最大経過日数: 指定した日数以内に変更されたファイルを含めます。デフォルトは 180 日です。
アクセスキー ID:
chronicle-s3-readerユーザーのアクセスキー ID を入力します。シークレット アクセスキー:
chronicle-s3-readerユーザーのシークレット アクセスキーを入力します。アセットの名前空間: アセットの名前空間。「
harness.audit」と入力します。Ingestion labels: このフィードのイベントに適用される省略可能なラベル。
- S3 URI: 接頭辞パスを含む S3 バケット URI を入力します。
[次へ] をクリックします。
[Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。
さらにサポートが必要な場合 コミュニティ メンバーや Google SecOps のプロフェッショナルから回答を得ることができます。