收集 Aware 稽核記錄

支援的國家/地區：

Google SecOps SIEM

本文說明如何使用 Google Cloud Storage，將 Aware 稽核記錄檔擷取至 Google Security Operations。Aware 是協作智慧平台，可為企業協作工具提供洞察資料和治理功能。

事前準備

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

Google SecOps 執行個體
已啟用 Cloud Storage API 的 GCP 專案
建立及管理 GCS 值區的權限
管理 Google Cloud Storage 值區 IAM 政策的權限
建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
Aware 租戶的特殊存取權

收集 Aware API 的必要條件

登入 Aware 管理控制台。
依序前往「系統設定」>「整合」>「API 權杖」。
按一下「+ API Token」，然後授予「Audit Logs Read-only」權限。
複製下列詳細資料並存放在安全的地方：
- API 權杖
- API 基準網址：端點為 https://api.aware.work/external/system/auditlogs/v1
注意： Aware 管理控制台中顯示的 API 基本網址應與 https://api.aware.work/external/system/auditlogs/v1 相符。如果租戶使用其他基本網址，您需要更新函式程式碼中的 AWARE_ENDPOINT 常數。

建立 Google Cloud Storage 值區

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

請提供下列設定詳細資料：

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

點選「建立」。

為 Cloud Run 函式建立服務帳戶

Cloud Run 函式需要具備 GCS bucket 寫入權限的服務帳戶，並由 Pub/Sub 叫用。

建立服務帳戶

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

這些角色適用於：

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

授予 GCS 值區的 IAM 權限

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

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

建立 Pub/Sub 主題

建立 Pub/Sub 主題，Cloud Scheduler 會將訊息發布至該主題，而 Cloud Run 函式會訂閱該主題。

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

建立 Cloud Run 函式來收集記錄

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

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

設定值

服務名稱 aware-audit-poller

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

執行階段 選取「Python 3.12」以上版本
在「Trigger (optional)」(觸發條件 (選用)) 專區：
1. 按一下「+ 新增觸發條件」。
2. 選取「Cloud Pub/Sub」。
3. 在「選取 Cloud Pub/Sub 主題」中，選擇主題 aware-audit-trigger。
4. 按一下 [儲存]。
在「Authentication」(驗證) 部分：
1. 選取「需要驗證」。
2. 檢查 Identity and Access Management (IAM)。
注意： 叫用函式時，Pub/Sub 會自動處理驗證。
向下捲動並展開「Containers, Networking, Security」。
前往「安全性」分頁：
- 服務帳戶：選取服務帳戶 aware-audit-poller-sa。
前往「容器」分頁：
1. 按一下「變數與密鑰」。
2. 針對每個環境變數，按一下「+ 新增變數」：
  
  變數名稱範例值
  
  GCS_BUCKET aware-audit-logs
  
  GCS_PREFIX aware/audit/
  
  STATE_KEY aware/state.json
  
  AWARE_API_TOKEN your-aware-api-token
  
  MAX_PER_PAGE 500
在「變數與密鑰」分頁中向下捲動至「要求」：
- 要求逾時：輸入 600 秒 (10 分鐘)。
前往「容器」中的「設定」分頁：
- 在「資源」部分：
  - 記憶體：選取 512 MiB 以上。
  - CPU：選取 1。
- 按一下 [完成]。
捲動至「執行環境」：
- 選取「預設」 (建議選項)。
在「修訂版本資源調度」部分：
- 執行個體數量下限：輸入 0。
- 「Maximum number of instances」(執行個體數量上限)：輸入 100 (或根據預期負載調整)。
點選「建立」。
等待服務建立完成 (1 到 2 分鐘)。
服務建立完成後，系統會自動開啟內嵌程式碼編輯器。

設定	值
服務名稱	`aware-audit-poller`
區域	選取與 GCS bucket 相符的區域 (例如 `us-central1`)
執行階段	選取「Python 3.12」以上版本

變數名稱	範例值
`GCS_BUCKET`	`aware-audit-logs`
`GCS_PREFIX`	`aware/audit/`
`STATE_KEY`	`aware/state.json`
`AWARE_API_TOKEN`	`your-aware-api-token`
`MAX_PER_PAGE`	`500`

新增函式程式碼

在「Function entry point」(函式進入點) 中輸入 main

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

第一個檔案：main.py:

import functions_framework
from google.cloud import storage
import json
import os
import urllib3
from datetime import datetime, timedelta, timezone
import gzip
import io
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()

AWARE_ENDPOINT = "https://api.aware.work/external/system/auditlogs/v1"

@functions_framework.cloud_event
def main(cloud_event):
    """
    Cloud Run function triggered by Pub/Sub to fetch Aware audit logs 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', 'aware/audit/')
    state_key = os.environ.get('STATE_KEY', 'aware/state.json')
    api_token = os.environ.get('AWARE_API_TOKEN')
    max_per_page = int(os.environ.get('MAX_PER_PAGE', '500'))

    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
        state = load_state(bucket, state_key)

        tz_utc = timezone.utc
        now = datetime.now(tz=tz_utc)

        start_date = (
            datetime.fromisoformat(state['last_date']).date()
            if 'last_date' in state
            else (now - timedelta(days=1)).date()
        )
        end_date = now.date()

        total = 0
        day = start_date
        backoff = 1.0

        while day <= end_date:
            day_str = day.strftime('%Y-%m-%d')
            params = {
                'filter': f'startDate:{day_str},endDate:{day_str}',
                'limit': str(max_per_page)
            }
            offset = 1

            out = io.BytesIO()
            with gzip.GzipFile(filename='aware_audit.jsonl', mode='wb', fileobj=out) as gz:
                wrote_any = False

                while True:
                    # Build query URL
                    query_params = {**params, 'offset': str(offset)}
                    query_string = '&'.join([f'{k}={v}' for k, v in query_params.items()])
                    url = f'{AWARE_ENDPOINT}?{query_string}'

                    # Make API request
                    headers = {'X-Aware-Api-Key': api_token}

                    try:
                        response = http.request('GET', url, headers=headers, timeout=30.0)

                        # Handle rate limiting with exponential backoff
                        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)
                            continue

                        backoff = 1.0

                        if response.status != 200:
                            print(f'Error: API returned status {response.status}')
                            break

                        payload = json.loads(response.data.decode('utf-8'))
                        items = (payload.get('value') or {}).get('auditLogData') or []

                        if not items:
                            break

                        for item in items:
                            gz.write((json.dumps(item, separators=(',', ':')) + '\n').encode('utf-8'))
                            total += 1
                            wrote_any = True

                        offset += 1

                    except Exception as e:
                        print(f'Error fetching page: {str(e)}')
                        break

            if wrote_any:
                blob_name = f"{prefix}{day.strftime('%Y/%m/%d')}/aware_audit_{now.strftime('%Y%m%d_%H%M%S')}.jsonl.gz"
                blob = bucket.blob(blob_name)
                blob.upload_from_string(
                    out.getvalue(),
                    content_type='application/json',
                    content_encoding='gzip'
                )
                print(f'Wrote {total} logs to {blob_name}')

            save_state(bucket, state_key, {'last_date': day.isoformat()})
            day += timedelta(days=1)

        print(f'Successfully processed {total} logs')

    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),
            content_type='application/json'
        )
    except Exception as e:
        print(f'Warning: Could not save state: {str(e)}')

第二個檔案：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」。
點選「建立工作」。

請提供下列設定詳細資料：

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

點選「建立」。

排程頻率選項

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

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

測試排程器工作

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

擷取 Google SecOps 服務帳戶

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

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

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

注意： 每個 Google SecOps 執行個體都有專屬的服務帳戶。請勿使用其他說明文件或範例中的服務帳戶。

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

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

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

注意： 如果打算使用「刪除已轉移的檔案」或「刪除已轉移的檔案並清空目錄」刪除選項，請授予「Storage 物件管理員」角色，而非「Storage 物件檢視者」角色。

在 Google SecOps 中設定動態饋給，擷取 Aware 稽核記錄

依序前往「SIEM 設定」>「動態饋給」。
按一下「新增動態消息」。
按一下「設定單一動態饋給」。
在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Aware Audit logs)。
選取「Google Cloud Storage V2」做為「來源類型」。
選取「Aware Audit」做為「記錄類型」。
點選 [下一步]。
指定下列輸入參數的值：
- 儲存空間 bucket URL：輸入 GCS bucket URI，並加上前置路徑：
```
gs://aware-audit-logs/aware/audit/
```
  - 取代：
    - aware-audit-logs：您的 GCS bucket 名稱。
    - aware/audit/：儲存記錄的前置字元/資料夾路徑。
  注意： URI 結尾一律要加上尾端斜線 (/)。
- 來源刪除選項：根據偏好設定選取刪除選項：
  - 永不：移轉後一律不刪除任何檔案 (建議用於測試)。
  - 刪除已轉移的檔案：成功轉移檔案後刪除檔案。
  - 刪除已轉移的檔案和空白目錄：成功轉移後刪除檔案和空白目錄。
    
    注意： 如果選取刪除選項，服務帳戶必須具備「Storage 物件管理員」角色，而非「Storage 物件檢視者」角色。請據此更新 IAM 權限。
- 檔案存在時間上限：包含在過去天數內修改的檔案。預設值為 180 天。
- 資產命名空間：資產命名空間。
- 擷取標籤：要套用至這個動態饋給事件的標籤。
點選 [下一步]。
在「Finalize」(完成) 畫面中檢查新的動態饋給設定，然後按一下「Submit」(提交)。

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