本頁面由 Cloud Translation API 翻譯而成。

收集 Harness IO 稽核記錄

支援的國家/地區：

Google SecOps SIEM

本文說明如何使用 Amazon S3，將 Harness IO 稽核記錄擷取至 Google Security Operations。

事前準備

Google SecOps 執行個體
Harness 的特殊存取權 (API 金鑰和帳戶 ID)
AWS 的特殊存取權 (S3、IAM、Lambda、EventBridge)

取得個人帳戶的 Harness API 金鑰和帳戶 ID

登入 Harness 網頁版 UI。
前往「使用者設定檔」>「我的 API 金鑰」。
選取「API 金鑰」。
輸入 API 金鑰的名稱。
按一下 [儲存]。
選取新 API 金鑰下方的「權杖」。
輸入權杖的名稱。
按一下 [產生憑證]。
複製權杖並儲存在安全的位置。
複製並儲存您的「帳戶 ID」 (顯示在 Harness 網址和「帳戶設定」中)。

選用：取得服務帳戶的 Harness API 金鑰和帳戶 ID

登入 Harness 網頁版 UI。
建立服務帳戶
依序前往「帳戶設定」>「存取控制」。
選取「服務帳戶」> 選取要建立 API 金鑰的服務帳戶。
在「API 金鑰」下方，選取「API 金鑰」。
輸入 API 金鑰的名稱。
按一下 [儲存]。
選取新 API 金鑰下方的「權杖」。
輸入權杖名稱。
按一下 [產生憑證]。
複製權杖並儲存在安全的位置。
複製並儲存您的「帳戶 ID」 (顯示在 Harness 網址和「帳戶設定」中)。

為 Google SecOps 設定 AWS S3 值區和 IAM

按照本使用指南建立 Amazon S3 值區：建立值區
儲存 bucket 的「名稱」和「區域」，以供日後參考 (例如 harness-io)。
按照這份使用者指南建立使用者：建立 IAM 使用者。
選取建立的「使用者」。
選取「安全憑證」分頁標籤。
在「Access Keys」部分中，按一下「Create Access Key」。
選取「第三方服務」做為「用途」。
點選「下一步」。
選用：新增說明標記。
按一下「建立存取金鑰」。
按一下「下載 CSV 檔案」，儲存「存取金鑰」和「私密存取金鑰」，以供日後使用。
按一下 [完成]。
選取 [權限] 分頁標籤。
在「Permissions policies」(權限政策) 區段中，按一下「Add permissions」(新增權限)。
選取「新增權限」。
選取「直接附加政策」
搜尋並選取 AmazonS3FullAccess 政策。
點選「下一步」。
按一下「Add permissions」。

設定 S3 上傳的身分與存取權管理政策和角色

在 AWS 管理控制台中，依序前往「IAM」>「Policies」>「Create policy」>「JSON」分頁標籤。

輸入下列政策：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowPutHarnessObjects",
      "Effect": "Allow",
      "Action": "s3:PutObject",
      "Resource": "arn:aws:s3:::harness-io/*"
    },
    {
      "Sid": "AllowGetStateObject",
      "Effect": "Allow",
      "Action": "s3:GetObject",
      "Resource": "arn:aws:s3:::harness-io/harness/audit/state.json"
    }
  ]
}

如果您輸入了其他值區名稱，請將 harness-io 替換為該名稱：

依序點選「Next」>「Create policy」。
依序前往「IAM」>「Roles」>「Create role」>「AWS service」>「Lambda」。
附加新建立的政策。
為角色命名 WriteHarnessToS3Role，然後按一下「建立角色」。

建立 Lambda 函式

在 AWS 控制台中，依序前往「Lambda」>「Functions」>「Create function」。
按一下「從頭開始撰寫」。
請提供下列設定詳細資料：

設定值

名稱 harness_io_to_s3

執行階段 Python 3.13

架構 x86_64

執行角色 WriteHarnessToS3Role

設定	值
名稱	`harness_io_to_s3`
執行階段	Python 3.13
架構	x86_64
執行角色	`WriteHarnessToS3Role`

建立函式後，開啟「程式碼」分頁，刪除存根並輸入下列程式碼 (harness_io_to_s3.py)。

#!/usr/bin/env python3

import os, json, time, urllib.parse
from urllib.request import Request, urlopen
from urllib.error import HTTPError, URLError
import boto3

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"]  # x-api-key token
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", "100")), 100)  # <=100
START_MINUTES_BACK = int(os.environ.get("START_MINUTES_BACK", "60"))

s3 = boto3.client("s3")
HDRS = {"x-api-key": API_KEY, "Content-Type": "application/json", "Accept": "application/json"}

def _read_state():
    try:
        obj = s3.get_object(Bucket=BUCKET, Key=STATE_KEY)
        j = json.loads(obj["Body"].read())
        return j.get("since"), j.get("pageToken")
    except Exception:
        return None, None

def _write_state(since_ms: int, page_token: str | None):
    body = json.dumps({"since": since_ms, "pageToken": page_token}).encode("utf-8")
    s3.put_object(Bucket=BUCKET, Key=STATE_KEY, Body=body, ContentType="application/json")

def _http_post(path: str, body: dict, query: dict, timeout: int = 60, max_retries: int = 5) -> dict:
    qs = urllib.parse.urlencode(query)
    url = f"{API_BASE}{path}?{qs}"
    data = json.dumps(body).encode("utf-8")

    attempt, backoff = 0, 1.0
    while True:
        req = Request(url, data=data, method="POST")
        for k, v in HDRS.items():
            req.add_header(k, v)
        try:
            with urlopen(req, timeout=timeout) as r:
                return json.loads(r.read().decode("utf-8"))
        except HTTPError as e:
            if (e.code == 429 or 500 <= e.code <= 599) and attempt < max_retries:
                time.sleep(backoff)
                attempt += 1
                backoff *= 2
                continue
            raise
        except URLError:
            if attempt < max_retries:
                time.sleep(backoff)
                attempt += 1
                backoff *= 2
                continue
            raise

def _write_page(obj: dict, now: float, page_index: int) -> str:
    ts = time.strftime("%Y/%m/%d/%H%M%S", time.gmtime(now))
    key = f"{PREFIX}/{ts}-page{page_index:05d}.json"
    s3.put_object(
        Bucket=BUCKET,
        Key=key,
        Body=json.dumps(obj, separators=(",", ":")).encode("utf-8"),
        ContentType="application/json",
    )
    return key

def fetch_and_store():
    now_s = time.time()
    since_ms, page_token = _read_state()
    if since_ms is None:
        since_ms = int((now_s - START_MINUTES_BACK * 60) * 1000)
    until_ms = int(now_s * 1000)

    page_index = 0
    total = 0

    while True:
        body = {"startTime": since_ms, "endTime": until_ms}
        query = {"accountIdentifier": ACCOUNT_ID, "pageSize": PAGE_SIZE}
        if page_token:
            query["pageToken"] = page_token
        else:
            query["pageIndex"] = page_index

        data = _http_post("/audit/api/audits/listV2", body, query)
        _write_page(data, now_s, page_index)

        entries = []
        for key in ("data", "content", "response", "resource", "resources", "items"):
            if isinstance(data.get(key), list):
                entries = data[key]
                break
        total += len(entries) if isinstance(entries, list) else 0

        next_token = (
            data.get("pageToken")
            or (isinstance(data.get("meta"), dict) and data["meta"].get("pageToken"))
            or (isinstance(data.get("metadata"), dict) and data["metadata"].get("pageToken"))
        )

        if next_token:
            page_token = next_token
            page_index += 1
            continue

        if len(entries) < PAGE_SIZE:
            break
        page_index += 1

    _write_state(until_ms, None)
    return {"pages": page_index + 1, "objects_estimate": total}

def lambda_handler(event=None, context=None):
    return fetch_and_store()

if __name__ == "__main__":
    print(lambda_handler())

依序前往「Configuration」>「Environment variables」>「Edit」>「Add new environment variable」。

輸入下列環境變數，並將換成您的值：

鍵	範例
`S3_BUCKET`	`harness-io`
`S3_PREFIX`	`harness/audit/`
`STATE_KEY`	`harness/audit/state.json`
`HARNESS_ACCOUNT_ID`	`123456789`
`HARNESS_API_KEY`	`harness_xxx_token`
`HARNESS_API_BASE`	`https://app.harness.io`
`PAGE_SIZE`	`100`
`START_MINUTES_BACK`	`60`

建立函式後，請留在函式頁面 (或依序開啟「Lambda」>「Functions」>「your‑function」)。
選取「設定」分頁標籤。
在「一般設定」面板中，按一下「編輯」。
將「Timeout」(逾時間隔) 變更為「5 minutes (300 seconds)」(5 分鐘 (300 秒))，然後按一下「Save」(儲存)。

注意： 逾時是上限，也就是說，如果函式在計時器達到 5 分鐘時仍在執行，AWS Lambda 就會終止函式。如果執行時間較短，系統會正常完成作業，並只收取實際執行時間的費用。

建立 EventBridge 排程

依序前往「Amazon EventBridge」>「Scheduler」>「Create schedule」。
請提供下列設定詳細資料：
- 週期性時間表：費率 (1 hour)。
- 目標：您的 Lambda 函式。
- 名稱：harness-io-1h。
按一下「建立時間表」。

為 Google SecOps 建立唯讀 IAM 使用者和金鑰

在 AWS 控制台中，依序前往「IAM」>「Users」，然後按一下「Add users」。
提供下列設定詳細資料：
- 使用者：輸入不重複的名稱 (例如 secops-reader)
- 存取權類型：選取「存取金鑰 - 程式輔助存取」
- 按一下「建立使用者」。
附加最低讀取權限政策 (自訂)：依序選取「使用者」secops-reader「權限」「新增權限」「直接附加政策」「建立政策」

在 JSON 編輯器中輸入下列政策：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["s3:GetObject"],
      "Resource": "arn:aws:s3:::<your-bucket>/*"
    },
    {
      "Effect": "Allow",
      "Action": ["s3:ListBucket"],
      "Resource": "arn:aws:s3:::<your-bucket>"
    }
  ]
}

將名稱設為 secops-reader-policy。
依序前往「建立政策」> 搜尋/選取 >「下一步」>「新增權限」。
依序前往「安全憑證」>「存取金鑰」>「建立存取金鑰」。
下載 CSV (這些值會輸入至動態饋給)。

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

依序前往「SIEM 設定」>「動態饋給」。
按一下「新增動態消息」。
在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Harness IO)。
選取「Amazon S3 V2」做為「來源類型」。
選取「Harness IO」做為「記錄類型」。
點選「下一步」。
指定下列輸入參數的值：
- S3 URI：s3://harness-io/harness/audit/
- 來源刪除選項：根據偏好選取刪除選項。
- 檔案存在時間上限：預設為 180 天。
- 存取金鑰 ID：具有 S3 值區存取權的使用者存取金鑰。
- 存取密鑰：具有 S3 bucket 存取權的使用者私密金鑰。
- 資產命名空間：資產命名空間。
- 擷取標籤：要套用至這個動態饋給事件的標籤。
點選「下一步」。
在「Finalize」畫面上檢查新的動態饋給設定，然後按一下「Submit」。

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