Google 會運用 AI 技術將內容翻譯成你偏好的語言，但可能會出錯。

收集 MuleSoft Anypoint 平台記錄

支援：

Google SecOps SIEM

本文說明如何使用 Cloud Storage，將 MuleSoft Anypoint 平台記錄中的稽核追蹤記錄事件，擷取至 Google Security Operations。

事前準備

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

Google SecOps 執行個體
已啟用 Cloud Storage API 的 Google Cloud 專案
建立及管理 Cloud Storage bucket 的權限
管理 Cloud Storage 值區身分與存取權管理 (IAM) 政策的權限
建立 Cloud Run 函式、Pub/Sub 主題和 Cloud Scheduler 工作的權限
建立服務帳戶的權限
MuleSoft Anypoint Platform 的特殊存取權

取得 MuleSoft 組織 ID

登入 Anypoint Platform。
依序前往「存取權管理」>「機構」。
在「商家群組」表格中，按一下貴機構的名稱。
複製組織 ID (例如 0a12b3c4-d5e6-789f-1021-1a2b34cd5e6f)。

或者，前往 MuleSoft Business Groups，然後從網址複製 ID。

建立 MuleSoft 連結應用程式

登入 Anypoint Platform。
依序前往「存取權管理」>「已連結的應用程式」>「建立應用程式」。
請提供下列設定詳細資料：
- 應用程式名稱：輸入不重複的名稱 (例如 Google SecOps export)。
- 選取「App acts on its own behalf (client credentials)」。
依序點選「新增範圍」>「稽核記錄檢視器」>「下一步」。
選取需要記錄的每個營業地點群組。
依序點選「下一步」> 新增範圍。
按一下「儲存」，然後複製「用戶端 ID」和「用戶端密鑰」。

注意： 視組織權限而定，可能需要其他範圍。如果遇到權限錯誤，請確認已授予「已連結的應用程式」存取所有必要營業地點群組的權限。

建立 Cloud Storage bucket

前往 Google Cloud 控制台。
選取專案或建立新專案。
在導覽選單中，依序前往「Cloud Storage」>「Bucket」。
按一下「建立值區」。

提供下列設定詳細資料：

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

點選「建立」。

為 Cloud Run 函式建立服務帳戶

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

建立服務帳戶

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

這些角色適用於：

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

授予 Cloud Storage bucket 的 IAM 權限

在 Cloud Storage bucket 中授予服務帳戶 write 權限，方法如下：

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

建立 Pub/Sub 主題

建立 Pub/Sub 主題，供 Cloud Scheduler 發布訊息，並供 Cloud Run 函式訂閱，步驟如下：

在 GCP 控制台中，前往「Pub/Sub」>「主題」。
按一下「建立主題」。
請提供下列設定詳細資料：
- 主題 ID：輸入 mulesoft-audit-trigger。
- 其他設定保留預設值。
點選「建立」。

建立 Cloud Run 函式來收集記錄

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

前往 GCP 控制台的「Cloud Run」。
點選「編寫函式」。
在「設定」部分，提供下列設定詳細資料：

設定值

服務名稱 mulesoft-audit-collector

區域選取與 Cloud Storage bucket 相符的區域 (例如 us-central1)

執行階段 選取 Python 3.12 以上版本
在「觸發條件」部分，按一下「新增觸發條件」，然後選取「Pub/Sub 觸發條件」。
在「Eventarc trigger」(Eventarc 觸發條件) 窗格中，提供下列設定詳細資料：
- 觸發條件名稱：保留系統產生的名稱，或輸入觸發條件的名稱。
- 觸發條件類型：選取「Google 來源」。
- 事件供應商：選取「Pub/Sub」。
- 事件類型：選取「google.cloud.pubsub.topic.v1.messagePublished」。
- 「選取 Cloud Pub/Sub 主題」：選擇主題 mulesoft-audit-trigger。
- 區域：選取與函式相同的區域。
- 服務帳戶：選取服務帳戶 mulesoft-logs-collector-sa。
按一下「儲存觸發條件」。

注意： 觸發條件選項來自 Eventarc。如果沒有列出「Pub/Sub 觸發條件」，請為專案啟用 Eventarc API，然後重新載入頁面。如果窗格回報服務帳戶缺少「Eventarc Event Receiver」(Eventarc 事件接收者) 或「Cloud Run Invoker」(Cloud Run 叫用者) 角色，請按一下「Grant」(授權)。
在「Authentication」(驗證) 部分：
1. 選取「需要驗證」。
2. 請檢查身分與存取權管理 (IAM)。
注意： Pub/Sub 會在叫用函式時自動處理驗證。
前往並展開「容器、網路、安全性」。
前往「安全性」分頁標籤：
- 服務帳戶：選取 mulesoft-logs-collector-sa 服務帳戶。

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

前往「容器」分頁：

按一下「Variables & Secrets」(變數與密鑰)。
針對每個環境變數，按一下「+ 新增變數」：

變數名稱	範例值	說明
`GCS_BUCKET`	`mulesoft-audit-logs`	Cloud Storage Bucket 名稱
`GCS_PREFIX`	`mulesoft-audit`	記錄檔的前置字串
`STATE_KEY`	`mulesoft-audit-state.json`	狀態路徑，位於記錄前置字元外
`MULE_ORG_ID`	`your_org_id`	MuleSoft 機構組織 ID
`CLIENT_ID`	`your_client_id`	連結的應用程式用戶端 ID
`CLIENT_SECRET`	`your_client_secret`	連結的應用程式用戶端密鑰
`MAX_RECORDS`	`10000`	每次執行的記錄數上限
`PAGE_SIZE`	`200`	每頁項目數 (API 上限為 200)
`LOOKBACK_HOURS`	`24`	首次執行的初始回溯期

前往「變數與密鑰」分頁，然後點選「要求」：
- 要求逾時：輸入 600 秒 (10 分鐘)。
前往「容器」的「設定」分頁：
- 在「資源」部分：
  - 記憶體：選取 512 MiB 以上。
  - CPU：選取 1。
- 按一下「完成」。
前往「執行環境」：
- 選取「預設」 (建議)。
在「修訂版本資源調度」部分：
- 執行個體數量下限：輸入 0。
- 執行個體數量上限：輸入 100 (或根據預期負載調整)。
點選「建立」。
等待服務建立完成 (1 至 2 分鐘)。
服務建立完成後，系統會自動開啟內嵌程式碼編輯器。

新增函式程式碼

在「函式進入點」中輸入 main

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

第一個檔案 - main.py:

import functions_framework
from google.cloud import storage
from google.cloud.exceptions import NotFound
import hashlib
import json
import os
import urllib3
from urllib.parse import urlencode
from datetime import datetime, timezone, timedelta
import time

# 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()

# MuleSoft API endpoints
TOKEN_URL = "https://anypoint.mulesoft.com/accounts/api/v2/oauth2/token"
AUDIT_URL = "https://anypoint.mulesoft.com/audit/v2/organizations/{org_id}/query?cursorPagination=true"

# Environment variables
GCS_BUCKET = os.environ.get('GCS_BUCKET')
GCS_PREFIX = os.environ.get('GCS_PREFIX', 'mulesoft-audit')
# STATE_KEY must stay OUTSIDE GCS_PREFIX. The feed ingests every object under
# its bucket URI and, with a deletion option selected, deletes what it
# transferred. A state file inside the prefix would be ingested as log data and
# then deleted, resetting collection and re-ingesting duplicates.
STATE_KEY = os.environ.get('STATE_KEY', 'mulesoft-audit-state.json')
MULE_ORG_ID = os.environ.get('MULE_ORG_ID')
CLIENT_ID = os.environ.get('CLIENT_ID')
CLIENT_SECRET = os.environ.get('CLIENT_SECRET')
MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '10000'))
# The Audit Log Query API returns at most 200 entries per JSON query.
PAGE_SIZE = min(int(os.environ.get('PAGE_SIZE', '200')), 200)
LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))
# The query re-reads this many minutes before the watermark so that events the
# platform records late are still collected. Re-read events are removed by hash.
OVERLAP_MINUTES = int(os.environ.get('OVERLAP_MINUTES', '2'))

class FetchError(Exception):
    """Raised when the MuleSoft API call fails.

    The watermark must never advance on a failed fetch, otherwise the events in
    the failed window are skipped permanently.
    """

def to_unix_millis(dt: datetime) -> int:
    """Convert datetime to Unix epoch milliseconds."""
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return int(dt.astimezone(timezone.utc).timestamp() * 1000)

def parse_datetime(value: str) -> datetime:
    """Parse an ISO 8601 datetime string into an aware datetime."""
    if value.endswith("Z"):
        value = value[:-1] + "+00:00"
    dt = datetime.fromisoformat(value)
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return dt

def event_keys(event: dict) -> set:
    """Return the identity this event can be recognised by.

    Audit log entries carry no unique event ID, so a record counts as already
    collected when its full JSON content was seen before. The overlap window
    returns the same JSON again, which this catches without any knowledge of
    the payload schema.
    """
    return {'sha256:' + hashlib.sha256(
        json.dumps(event, sort_keys=True, ensure_ascii=False).encode('utf-8')
    ).hexdigest()}

def event_time(event: dict) -> str:
    """Return the event timestamp as an ISO 8601 string, or an empty string.

    Audit log entries carry the event time in `timestamp` as Unix milliseconds.
    """
    raw = event.get('timestamp')
    if raw is None or raw == '':
        return ''
    try:
        return datetime.fromtimestamp(int(raw) / 1000, tz=timezone.utc).isoformat()
    except (TypeError, ValueError):
        return ''

@functions_framework.cloud_event
def main(cloud_event):
    """Fetch MuleSoft Anypoint audit logs and write them to Cloud Storage.

    Args:
        cloud_event: CloudEvent object containing the Pub/Sub message.
    """
    if not all([GCS_BUCKET, MULE_ORG_ID, CLIENT_ID, CLIENT_SECRET]):
        # Raise rather than return: a bare return acks the Pub/Sub message and
        # reports the run as successful, silently discarding the schedule tick.
        raise RuntimeError('Missing required environment variables')

    bucket = storage_client.bucket(GCS_BUCKET)
    state = load_state(bucket, STATE_KEY)

    now = datetime.now(timezone.utc)
    watermark = None
    if state.get('last_event_time'):
        watermark = parse_datetime(state['last_event_time'])
    seen_keys = set(state.get('seen_keys', []))

    if watermark is None:
        start_time = now - timedelta(hours=LOOKBACK_HOURS)
    else:
        start_time = watermark - timedelta(minutes=OVERLAP_MINUTES)

    print(f"Fetching audit logs from {start_time.isoformat()} to {now.isoformat()}")

    token = get_token(CLIENT_ID, CLIENT_SECRET)

    # A FetchError here propagates: the run fails, the watermark is untouched,
    # and the next invocation retries the same window.
    records, newest_event_time = fetch_audit(
        org_id=MULE_ORG_ID,
        token=token,
        start_time=start_time,
        end_time=now,
        page_size=PAGE_SIZE,
        max_records=MAX_RECORDS,
    )

    # Remove events already written by an earlier run. Without this, the overlap
    # window re-emits its events on every invocation.
    fresh = [r for r in records if not (event_keys(r) & seen_keys)]
    print(f"Fetched {len(records)} records, {len(fresh)} new after deduplication")

    if not fresh:
        print("No new audit log records found. Watermark left unchanged.")
        return

    if not newest_event_time:
        # A watermark that cannot be derived from the data is not a watermark.
        # Advancing on wall-clock here would skip every late-recorded event.
        raise FetchError('Records were returned but no event timestamp could be parsed')

    timestamp = now.strftime('%Y%m%dT%H%M%SZ')
    object_key = f"{GCS_PREFIX}/audit_{timestamp}.ndjson"
    blob = bucket.blob(object_key)

    ndjson = '\n'.join(json.dumps(record, ensure_ascii=False) for record in fresh) + '\n'
    blob.upload_from_string(ndjson, content_type='application/x-ndjson')

    print(f"Wrote {len(fresh)} records to gs://{GCS_BUCKET}/{object_key}")

    # Advance the watermark only after the data is durably written.
    new_watermark = parse_datetime(newest_event_time)
    if watermark and new_watermark < watermark:
        new_watermark = watermark

    # Retain only the hashes still inside the overlap window, so the state file
    # stays small while covering every event the next query can re-read.
    cutoff = new_watermark - timedelta(minutes=OVERLAP_MINUTES)
    retained = set()
    for record in records:
        stamp = event_time(record)
        if not stamp:
            continue
        if parse_datetime(stamp) >= cutoff:
            retained.update(event_keys(record))

    save_state(bucket, STATE_KEY, {
        'last_event_time': new_watermark.isoformat(),
        'seen_keys': sorted(retained),
    })

    print(f"Successfully processed {len(fresh)} records")

def load_state(bucket, key):
    """Read the collector state from Cloud Storage.

    Only a missing object is treated as a cold start. Any other error is raised:
    swallowing it would silently reset collection to the full lookback window
    and re-ingest that entire period.
    """
    blob = bucket.blob(key)
    try:
        return json.loads(blob.download_as_text())
    except NotFound:
        print('No state file found. Starting from the lookback window.')
        return {}

def save_state(bucket, key, state: dict):
    """Write the collector state to Cloud Storage.

    Failures are raised, not logged. If the state write fails after the data was
    uploaded, the next run repeats the same window and duplicates it.
    """
    blob = bucket.blob(key)
    blob.upload_from_string(
        json.dumps(state, indent=2),
        content_type='application/json',
    )
    print(f"Saved state: last_event_time={state.get('last_event_time')}")

def retry_after_seconds(response, backoff: float) -> int:
    """Parse Retry-After defensively; it may be an HTTP date, which int() cannot parse."""
    raw = response.headers.get('Retry-After')
    try:
        return int(raw) if raw else int(backoff)
    except (TypeError, ValueError):
        return int(backoff)

def get_token(client_id: str, client_secret: str) -> str:
    """Get an OAuth 2.0 access token from MuleSoft using client credentials."""
    body = urlencode({
        'grant_type': 'client_credentials',
        'client_id': client_id,
        'client_secret': client_secret,
    }).encode('utf-8')

    backoff = 1.0
    for attempt in range(5):
        try:
            response = http.request(
                'POST',
                TOKEN_URL,
                body=body,
                headers={'Content-Type': 'application/x-www-form-urlencoded'},
            )
        except Exception as e:
            raise FetchError(f'Token request failed: {e}') from e

        if response.status == 429:
            wait = retry_after_seconds(response, backoff)
            print(f"Rate limited (429) on token request. Retrying after {wait}s...")
            time.sleep(wait)
            backoff = min(backoff * 2, 30.0)
            continue

        if response.status != 200:
            raise FetchError(f'Failed to get token: HTTP {response.status} {response.data.decode("utf-8", "replace")}')

        try:
            return json.loads(response.data.decode('utf-8'))['access_token']
        except (json.JSONDecodeError, KeyError) as e:
            raise FetchError(f'Malformed token response: {e}') from e

    raise FetchError('Rate limited repeatedly on token request; giving up without advancing the watermark')

def fetch_audit(org_id: str, token: str, start_time: datetime, end_time: datetime, page_size: int, max_records: int):
    """Fetch audit log entries from the MuleSoft Audit Log Query API.

    Uses cursor pagination. The API returns entries in ascending timestamp order,
    which a time watermark requires: with newest-first ordering a truncated run
    would advance the watermark past unread events.

    Returns:
        Tuple of (records list, newest_event_time ISO string).

    Raises:
        FetchError: on any API or transport failure, so that the caller does not
            mistake a failed fetch for an empty result and advance the watermark.
    """
    endpoint = AUDIT_URL.format(org_id=org_id)
    headers = {
        'Authorization': f'Bearer {token}',
        'Content-Type': 'application/json',
        'Accept': 'application/json',
    }

    records = []
    newest_time = None
    cursor = None
    page_num = 0
    backoff = 1.0
    rate_limit_retries = 0
    MAX_RATE_LIMIT_RETRIES = 5

    while True:
        page_num += 1

        if len(records) >= max_records:
            # Stop cleanly. The remaining events stay ahead of the watermark and
            # are collected by the next run.
            print(f"Reached max_records limit ({max_records})")
            break

        # Rebuild the full query on every page; the time window must never be dropped.
        body = {
            'startDate': to_unix_millis(start_time),
            'endDate': to_unix_millis(end_time),
            'limit': page_size,
            'ascending': True,
        }
        if cursor:
            body['cursor'] = cursor

        try:
            response = http.request(
                'POST',
                endpoint,
                body=json.dumps(body).encode('utf-8'),
                headers=headers,
            )
        except Exception as e:
            raise FetchError(f'Request to {endpoint} failed: {e}') from e

        if response.status == 429:
            rate_limit_retries += 1
            if rate_limit_retries > MAX_RATE_LIMIT_RETRIES:
                raise FetchError('Rate limited repeatedly; giving up without advancing the watermark')
            wait = retry_after_seconds(response, backoff)
            print(f"Rate limited (429). Retrying after {wait}s...")
            time.sleep(wait)
            backoff = min(backoff * 2, 30.0)
            continue

        backoff = 1.0
        rate_limit_retries = 0

        if response.status != 200:
            raise FetchError(f'HTTP {response.status} from {endpoint}: {response.data.decode("utf-8", "replace")}')

        try:
            data = json.loads(response.data.decode('utf-8'))
        except json.JSONDecodeError as e:
            raise FetchError(f'Malformed JSON response from {endpoint}: {e}') from e

        page_results = data.get('data') or []
        if not page_results:
            print("No more results (empty page)")
            break

        print(f"Page {page_num}: Retrieved {len(page_results)} entries")
        records.extend(page_results)

        for event in page_results:
            stamp = event_time(event)
            if not stamp:
                continue
            if newest_time is None or parse_datetime(stamp) > parse_datetime(newest_time):
                newest_time = stamp

        cursor = data.get('cursor')
        if not cursor:
            break

    print(f"Retrieved {len(records)} total entries from {page_num} pages")
    return records, newest_time

第二個檔案 - requirements.txt:

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

點選「部署」即可儲存並部署函式。
等待部署作業完成 (2 至 3 分鐘)。

重要事項

頻率限制：稽核記錄查詢端點會對三個控制層中的每個 IP 套用頻率限制。美國控制層允許每個 IP 每分鐘 700 個要求，而歐盟和政府控制層則允許每個 IP 每分鐘 40 個要求。函式會實作指數輪詢，自動處理頻率限制。
權杖效期：存取權杖通常會在核發後約 30 到 60 分鐘失效。函式會在每次執行時要求新權杖，這完全符合該時間範圍。
重複資料刪除和狀態：函式會在狀態檔案中保留最新事件時間的水印，以及重疊視窗內事件的雜湊。每次執行作業時，系統都會重新讀取浮水印前兩分鐘的資料，以擷取延遲記錄的項目，並捨棄已寫入的項目，因此重試、手動執行或排程器延遲不會產生重複事件。如果 API 呼叫失敗，執行作業就會失敗，且不會變更浮水印，因此下次執行作業時會重試相同時間範圍。
稽核記錄保留：稽核記錄的預設保留期限為一年。如果貴機構是在 2023 年 7 月 10 日前建立，且您未手動變更保留期限，則保留期限為六年。如要保留記錄超過設定的保留期限，請定期下載記錄。

建立 Cloud Scheduler 工作

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

在 GCP 控制台中，前往「Cloud Scheduler」。
點選「建立工作」。

提供下列設定詳細資料：

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

點選「建立」。

測試排程器工作

在 Cloud Scheduler 控制台中找到工作。
按一下「強制執行」即可手動觸發。
等待幾秒鐘，然後依序前往「Cloud Run」>「Services」>「mulesoft-audit-collector」>「Logs」。
確認函式是否執行成功。
檢查 Cloud Storage bucket，確認記錄已寫入。

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

依序前往「SIEM 設定」>「動態饋給」。
按一下「新增動態饋給」。
按一下「設定單一動態饋給」。
在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 MuleSoft Logs)。
選取「Google Cloud Storage V2」做為「來源類型」。
選取「Mulesoft」做為「記錄類型」。
按一下「取得服務帳戶」。系統會顯示專屬服務帳戶電子郵件地址，例如：
```
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
```
複製這個電子郵件地址。後續步驟會用到。

注意： 每個 Google SecOps 執行個體都有專屬的服務帳戶。請勿使用其他說明文件或範例中的服務帳戶。
點選「下一步」。
指定下列輸入參數的值：
- 儲存空間 bucket 網址：輸入 Cloud Storage bucket URI：
```
gs://mulesoft-audit-logs/
```
  - 將 mulesoft-audit-logs 替換為 bucket 的實際名稱。
  注意： URI 結尾一律要加上斜線 (/)。
- 來源刪除選項：根據偏好設定選取刪除選項：
  - 一律不刪除檔案：一律不刪除來源中的檔案 (建議用於測試)。
  - 刪除已轉移的檔案和空白目錄：成功擷取後，從來源刪除檔案和空白目錄。
- 檔案存在時間上限：納入在過去天數內修改的檔案。預設值為 180 天。
- 資產命名空間：資產命名空間。
- 擷取標籤：要套用至這個動態饋給事件的標籤。
點選「下一步」。
在「Finalize」(完成) 畫面中檢查新的動態饋給設定，然後按一下「Submit」(提交)。

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

Google SecOps 服務帳戶需要 Cloud Storage bucket 的兩個角色：用於讀取記錄物件的「Storage 物件檢視者」，以及用於讀取 bucket 中繼資料的 bucket 層級角色。

依序前往「Cloud Storage」>「Buckets」。
按一下 bucket 名稱。
前往「權限」分頁標籤。
按一下「授予存取權」。
請提供下列設定詳細資料：
- 新增主體：貼上 Google SecOps 服務帳戶電子郵件地址。
- 指派角色：選取下列兩項：
- Storage 物件檢視者：讀取記錄物件。
- Storage 舊版 bucket 讀取者：讀取 bucket 中繼資料。如果您選取「刪除已轉移的檔案和空白目錄」刪除選項，請改為選取「Storage Legacy Bucket Writer」，這也會授予刪除權限。
按一下 [儲存]。

注意： 請同時授予這兩個角色。Storage 物件檢視者不會授予動態消息所需的值區層級權限，而 Storage 舊版值區角色不會授予讀取物件內容的權限。

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