收集 Aware 稽核記錄

剖析器版本：1.0

支援的服務：

Google SecOps SIEM

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

事前準備

請確認您符合下列必要條件：

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

Collect Aware API 必要條件

1. 登入 Aware 管理控制台。
2. 依序前往「System Settings」>「Integrations」>「API Tokens」。
3. 按一下「+ API Token」，然後授予「稽核記錄 Read-only」權限。

4. 請複製下列詳細資料，並存放在安全的位置：

   • 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 bucket

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

5. 提供下列設定詳細資料：

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

6. 點選「建立」。

建立 Cloud Run 函式的服務帳戶

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

建立服務帳戶

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

這些角色適用於：

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

授予 GCS bucket 的 IAM 權限

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

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

建立 Pub/Sub 主題

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

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

建立 Cloud Run 函式來收集記錄

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

1. 在 GCP 控制台中，前往「Cloud Run」。
2. 按一下「Create service」(建立服務)。
3. 選取「函式」(使用直接修改建立函式)。

4. 在「Configure」(設定) 部分，提供下列設定詳細資料：

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

5. 在「Trigger (optional)」(觸發條件 (選用)) 部分：

   1. 按一下「+ 新增觸發條件」。
   2. 選取「Cloud Pub/Sub」。
   3. 在「Select a Cloud Pub/Sub topic」(選取 Cloud Pub/Sub 主題) 中，選擇主題 aware-audit-trigger。
   4. 按一下「儲存」。

6. 在「Authentication」(驗證) 部分：

   1. 選取「需要驗證」。
   2. Identity and Access Management (IAM)。
   注意：Pub/Sub 會在叫用函式時自動處理驗證。

7. 向下捲動並展開「容器、網路、安全性」。

8. 前往「安全性」分頁：

   • 「Service account」(服務帳戶)：選取服務帳戶 aware-audit-poller-sa。

9. 前往「容器」分頁：

   1. 按一下「Variables & Secrets」(變數與密鑰)。

   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

10. 在「變數與密鑰」分頁中，向下捲動至「要求」：

    • 要求逾時：輸入 600 秒 (10 分鐘)。

11. 前往「容器」的「設定」分頁：

    • 在「Resources」(資源) 部分：
      • 「記憶體」：選取「512 MiB」或以上。
      • CPU：選取「1」。
    • 按一下「完成」。

12. 捲動至「Execution environment」(執行環境)：

    • 選取「預設」(建議)。

13. 在「Revision scaling」(修訂版本資源調度) 部分：

    • 「Minimum number of instances」(執行個體數量下限)：輸入 0。
    • 「執行個體數量上限」：輸入 100 (或依預期負載調整)。

14. 點選「建立」。

15. 等待服務建立完成 (1 至 2 分鐘)。

16. 服務建立完成後，內嵌程式碼編輯器會自動開啟。

新增函式程式碼

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

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

   • 第一個檔案：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
   

3. 按一下「部署」，即可儲存並部署函式。

4. 等待部署作業完成 (2 至 3 分鐘)。

建立 Cloud Scheduler 工作

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

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

3. 提供下列設定詳細資料：

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

4. 點選「建立」。

排程頻率選項

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

  頻率	Cron 運算式	用途
  每 5 分鐘	*/5 * * * *	高流量、低延遲
  每 15 分鐘	*/15 * * * *	中等音量
  每小時	0 * * * *	標準 (建議)
  每 6 小時	0 */6 * * *	低量、批次處理
  每天	0 0 * * *	歷來資料集合

  注意：頻繁輪詢 (例如每 5 分鐘一次) 可能會受到 Aware API 頻率限制。監控非 200 回應，並視需要調整排程。

測試排程器工作

1. 在 Cloud Scheduler 控制台中找到您的工作。
2. 按一下「強制執行」，手動觸發。
3. 稍候幾秒，然後依序前往「Cloud Run」>「服務」>「aware-audit-poller」>「記錄」。
4. 確認函式是否執行成功。
5. 查看 GCS bucket，確認記錄已寫入。

擷取 Google SecOps 服務帳戶

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

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

1. 依序前往「設定」>「資料管理」>「動態饋給」。如果使用的是 SIEM 專用平台，請依序前往「設定」>「動態饋給」。
2. 按一下「新增動態饋給」。
3. 按一下「設定單一動態饋給」。
4. 在「Feed name」(動態饋給名稱) 欄位中，輸入動態饋給名稱 (例如 Aware Audit logs)。
5. 選取「Google Cloud Storage V2」做為「來源類型」。
6. 在「Log type」(記錄類型) 中選取「Aware Audit」(Aware 稽核)。

7. 按一下「Get Service Account」。系統會顯示專屬服務帳戶電子郵件地址，例如：

   [e-mail]
   

8. 複製這個電子郵件地址，下一個步驟將會用到。

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

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

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

1. 依序前往「Cloud Storage」>「Bucket」。
2. 按一下 bucket 名稱。
3. 前往「權限」分頁。
4. 按一下「授予存取權」。
5. 提供下列設定詳細資料：
   • 新增主體：貼上 Google SecOps 服務帳戶電子郵件地址。
   • 指派角色：選取「Storage 物件檢視者」。

6. 按一下「儲存」。

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

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

1. 依序前往「設定」>「資料管理」>「動態饋給」。如果使用的是 SIEM 專用平台，請依序前往「設定」>「動態饋給」。
2. 按一下「新增動態饋給」。
3. 按一下「設定單一動態饋給」。
4. 在「Feed name」(動態饋給名稱) 欄位中，輸入動態饋給名稱 (例如 Aware Audit logs)。
5. 選取「Google Cloud Storage V2」做為「來源類型」。
6. 在「Log type」(記錄類型) 中選取「Aware Audit」(Aware 稽核)。
7. 點選「下一步」。

8. 指定下列輸入參數的值：

   • Storage bucket URL：輸入 GCS bucket URI，並加上前置字元路徑：

     gs://aware-audit-logs/aware/audit/
     

     • 更改項目：

       • aware-audit-logs：GCS bucket 名稱。
       • aware/audit/：：記錄檔的儲存前置字元/資料夾路徑。
     注意：URI 結尾一律要加上斜線 (/)。

   • 來源刪除選項：根據偏好選取刪除選項：

     • Never：轉移後一律不刪除任何檔案 (建議用於測試)。
     • 刪除已轉移的檔案：在成功轉移後刪除檔案。

     • 刪除已轉移的檔案和空白目錄：成功轉移後，刪除檔案和空白目錄。

       注意：如果選取刪除選項，服務帳戶必須具備 Storage 物件管理員角色，而非 Storage 物件檢視者。請據此更新 IAM 權限。

   • 檔案存在時間上限：包含過去幾天內修改過的檔案。預設值為 180 天。

   • Asset namespace：資產命名空間。

   • 擷取標籤：要套用至這個動態饋給事件的標籤。

9. 點選「下一步」。

10. 在「完成」畫面中檢查新的新聞動態消息設定，然後按一下「提交」。

變更記錄

查看這個剖析器的變更記錄

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