收集 Google Cloud Looker 稽核記錄
支援的國家/地區：
Google SecOps
SIEM
本文說明如何使用 Google Cloud Storage 或 Amazon S3，將 Google Cloud Looker 稽核記錄擷取至 Google Security Operations。
Looker 商業智慧與資料分析平台可協助機構探索、分析及分享即時業務洞察資料。Looker 的系統活動模型會追蹤使用者驗證事件、查詢執行記錄、資訊主頁和 Look 存取權、內容建立和修改、下載事件、API 呼叫、排定傳送時間的事件，以及權限變更。
事前準備
請確認您已完成下列事前準備事項：
- Google SecOps 執行個體
- 具備管理員存取權或
see_system_activity權限的 Looker 執行個體 - Looker API 憑證 (用戶端 ID 和用戶端密鑰)
- 如果是 GCS 路徑：已啟用 Cloud Storage、Cloud Run、Pub/Sub 和 Cloud Scheduler API 的 GCP 專案
- S3 路徑：AWS (S3、IAM) 的特殊存取權
設定 Looker API 憑證
如要讓 Google SecOps 擷取系統活動稽核資料，您必須在 Looker 中建立 API 憑證，並設定具備必要權限的服務帳戶。
建立 Looker 服務帳戶以存取 API
- 以管理員身分登入 Looker 執行個體。
- 依序前往「管理」>「使用者」。
- 按一下「新增使用者」。
- 在「電子郵件」欄位中，輸入服務帳戶電子郵件地址 (例如
chronicle-integration@yourcompany.com)。 在「角色」下方，選取包含
see_system_activity權限的角色。按一下 [儲存]。
產生 API3 憑證
- 依序前往「管理」>「使用者」。
- 找出您建立的服務帳戶使用者，然後按一下「編輯」。
- 向下捲動至「API 金鑰」部分。
- 點選「New API Key」(新增 API 金鑰)。
複製下列值並妥善保存：
- 用戶端 ID：用於 API 驗證的公開 ID
- 用戶端密鑰：用於 API 驗證的私密金鑰
找出 Looker API 基礎網址
Looker API 基礎網址的格式如下：
https://<instance_name>.cloud.looker.com
如果是託管於 Google Cloud、Microsoft Azure 的 Looker 執行個體，以及 2020 年 7 月 7 日當天或之後建立的 AWS 執行個體，API 會使用連接埠 443 (預設 HTTPS)。如果是較舊的 AWS 代管執行個體，API 可能會使用 19999 連接埠。
如要查看 API 主機網址，請在 Looker 執行個體中依序前往「管理」>「API」。
測試 API 存取權
請先測試憑證，再繼續進行整合：
# Replace with your actual credentials LOOKER_BASE_URL="https://your-instance.cloud.looker.com" CLIENT_ID="your-client-id" CLIENT_SECRET="your-client-secret" # Obtain access token TOKEN=$(curl -s -X POST "${LOOKER_BASE_URL}/api/4.0/login" \ -d "client_id=${CLIENT_ID}&client_secret=${CLIENT_SECRET}" \ | python3 -c "import sys,json; print(json.load(sys.stdin)['access_token'])") # Test System Activity access curl -s -H "Authorization: token ${TOKEN}" \ "${LOOKER_BASE_URL}/api/4.0/queries/run/json" \ -X POST \ -H "Content-Type: application/json" \ -d '{"model":"system__activity","view":"event","fields":["event.name","event.created_time"],"limit":"5","sorts":["event.created_time desc"]}' \ | python3 -m json.tool
成功的回應會傳回近期 Looker 事件的 JSON 陣列。
選項 A：使用 Google Cloud Storage 設定擷取作業
這個選項會使用 Cloud Run 函式輪詢 Looker API，取得系統活動稽核事件，並將這些事件寫入 GCS bucket，供 Google SecOps 擷取。
建立 Google Cloud Storage bucket
- 前往 Google Cloud 控制台。
- 選取專案或建立新專案。
- 在導覽選單中，依序前往「Cloud Storage」>「Bucket」。
- 按一下「建立值區」。
請提供下列設定詳細資料：
設定 值 為 bucket 命名 輸入全域不重複的名稱 (例如 looker-audit-logs-gcs)位置類型 根據需求選擇 (區域、雙區域、多區域) 位置 選取位置 (例如 us-central1)儲存空間級別 標準 (建議用於經常存取的記錄) 存取控管 統一 (建議) 保護工具 選用：啟用物件版本管理或資料保留政策 點選「建立」。
為 Cloud Run 函式建立服務帳戶
- 在 GCP 控制台中，依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。
- 按一下「Create Service Account」(建立服務帳戶)。
- 請提供下列設定詳細資料：
- 服務帳戶名稱：輸入
looker-audit-collector-sa - 服務帳戶說明：輸入
Service account for Cloud Run function to collect Looker audit logs
- 服務帳戶名稱：輸入
- 按一下「建立並繼續」。
- 在「將專案存取權授予這個服務帳戶」部分，新增下列角色：
- 按一下「選擇角色」。
- 搜尋並選取「Storage 物件管理員」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Run Invoker」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Functions Invoker」(Cloud Functions 叫用者)。
- 按一下「繼續」。
- 按一下 [完成]。
授予 GCS 值區的 IAM 權限
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下 bucket 名稱 (
looker-audit-logs-gcs)。 - 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料：
- 新增主體：輸入服務帳戶電子郵件地址 (
looker-audit-collector-sa@PROJECT_ID.iam.gserviceaccount.com) - 指派角色：選取「Storage 物件管理員」
- 新增主體：輸入服務帳戶電子郵件地址 (
- 按一下 [儲存]。
建立 Pub/Sub 主題
- 在 GCP Console 中，前往「Pub/Sub」>「Topics」(主題)。
- 按一下「建立主題」。
- 請提供下列設定詳細資料：
- 主題 ID：輸入
looker-audit-trigger - 其他設定保留預設值
- 主題 ID：輸入
- 點選「建立」。
建立 Cloud Run 函式來收集記錄
- 前往 GCP Console 的「Cloud Run」。
- 按一下「Create service」(建立服務)。
- 選取「函式」 (使用內嵌編輯器建立函式)。
在「設定」部分，提供下列設定詳細資料：
設定 值 服務名稱 looker-audit-collector區域 選取與 GCS bucket 相符的區域 (例如 us-central1)執行階段 選取 Python 3.12 以上版本 在「Trigger (optional)」(觸發條件 (選用)) 專區：
- 按一下「+ 新增觸發條件」。
- 選取「Cloud Pub/Sub」。
- 在「Select a Cloud Pub/Sub topic」(選取 Cloud Pub/Sub 主題) 中，選擇
looker-audit-trigger。 - 按一下 [儲存]。
在「Authentication」(驗證) 部分：
- 選取「需要驗證」。
- 檢查 Identity and Access Management (IAM)。
向下捲動並展開「容器」、「網路」和「安全性」。
前往「安全性」分頁：
- 服務帳戶：選取
looker-audit-collector-sa
- 服務帳戶：選取
前往「容器」分頁：
- 按一下「變數與密鑰」。
- 針對每個環境變數，按一下「+ 新增變數」：
變數名稱 範例值 說明 GCS_BUCKETlooker-audit-logs-gcsGCS bucket 名稱 GCS_PREFIXlooker-audit記錄檔的前置字串 STATE_KEYlooker-audit/state.json狀態檔案路徑 LOOKER_BASE_URLhttps://your-instance.cloud.looker.comLooker API 基礎網址 LOOKER_CLIENT_IDyour-client-idLooker API 用戶端 ID LOOKER_CLIENT_SECRETyour-client-secretLooker API 用戶端密鑰 LOOKBACK_HOURS24初始回溯期 PAGE_SIZE5000每個 API 頁面的記錄 MAX_PAGES20每項查詢的頁數上限 在「變數與密鑰」部分，向下捲動至「要求」：
- 要求逾時：輸入
600秒 (10 分鐘)
- 要求逾時：輸入
前往「設定」分頁：
- 在「資源」部分：
- 記憶體：選取 512 MiB 以上
- CPU：選取 1
- 在「資源」部分：
在「修訂版本資源調度」部分：
- 執行個體數量下限：輸入
0 - 執行個體數量上限：輸入
100
- 執行個體數量下限：輸入
點選「建立」。
等待服務建立完成 (1 到 2 分鐘)。
服務建立完成後，系統會自動開啟內嵌程式碼編輯器。
新增函式程式碼
- 在「進入點」欄位中輸入「main」。
- 在內嵌程式碼編輯器中，建立兩個檔案：
main.py:
import functions_framework from google.cloud import storage import json import os import urllib3 import urllib.parse from datetime import datetime, timezone, timedelta http = urllib3.PoolManager( timeout=urllib3.Timeout(connect=10.0, read=60.0), retries=False, ) storage_client = storage.Client() GCS_BUCKET = os.environ.get('GCS_BUCKET') GCS_PREFIX = os.environ.get('GCS_PREFIX', 'looker-audit') STATE_KEY = os.environ.get('STATE_KEY', 'looker-audit/state.json') LOOKER_BASE_URL = os.environ.get('LOOKER_BASE_URL', '').rstrip('/') CLIENT_ID = os.environ.get('LOOKER_CLIENT_ID') CLIENT_SECRET = os.environ.get('LOOKER_CLIENT_SECRET') LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24')) PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '5000')) MAX_PAGES = int(os.environ.get('MAX_PAGES', '20')) @functions_framework.cloud_event def main(cloud_event): if not all([GCS_BUCKET, LOOKER_BASE_URL, CLIENT_ID, CLIENT_SECRET]): print('Error: Missing required environment variables') return try: bucket = storage_client.bucket(GCS_BUCKET) state = load_state(bucket) now = datetime.now(timezone.utc) if isinstance(state, dict) and state.get('last_event_time'): try: last_val = state['last_event_time'] if last_val.endswith('Z'): last_val = last_val[:-1] + '+00:00' last_time = datetime.fromisoformat(last_val) last_time = last_time - timedelta(minutes=2) except Exception as e: print(f"Warning: Could not parse last_event_time: {e}") last_time = now - timedelta(hours=LOOKBACK_HOURS) else: last_time = now - timedelta(hours=LOOKBACK_HOURS) print(f"Fetching events from {last_time.isoformat()} to {now.isoformat()}") token = get_access_token() events = fetch_system_activity( token, 'event', [ 'event.id', 'event.name', 'event.category', 'event.created_time', 'event.is_api_call', 'event.is_admin', 'event.is_looker_employee', 'user.id', 'user.name', 'user.email' ], 'event.created_time', last_time, now ) history = fetch_system_activity( token, 'history', [ 'history.id', 'history.created_time', 'history.completed_time', 'history.status', 'history.source', 'history.issuer_source', 'history.runtime', 'history.message', 'query.id', 'query.model', 'query.view', 'user.id', 'user.name', 'user.email', 'dashboard.id', 'dashboard.title', 'look.id', 'look.title' ], 'history.created_time', last_time, now ) all_records = [] for e in events: e['_looker_record_type'] = 'event' all_records.append(e) for h in history: h['_looker_record_type'] = 'history' all_records.append(h) if not all_records: print("No new records found.") save_state(bucket, now.isoformat()) return timestamp = now.strftime('%Y%m%d_%H%M%S') object_key = f"{GCS_PREFIX}/looker_audit_{timestamp}.ndjson" blob = bucket.blob(object_key) ndjson = '\n'.join( [json.dumps(r, ensure_ascii=False, default=str) for r in all_records] ) + '\n' blob.upload_from_string(ndjson, content_type='application/x-ndjson') print(f"Wrote {len(all_records)} records to gs://{GCS_BUCKET}/{object_key}") newest = find_newest_time(events, history) save_state(bucket, newest if newest else now.isoformat()) print(f"Successfully processed {len(all_records)} records " f"(events: {len(events)}, history: {len(history)})") except Exception as e: print(f'Error processing logs: {str(e)}') raise def get_access_token(): url = f"{LOOKER_BASE_URL}/api/4.0/login" encoded_body = urllib.parse.urlencode({ 'client_id': CLIENT_ID, 'client_secret': CLIENT_SECRET }).encode('utf-8') response = http.request( 'POST', url, body=encoded_body, headers={'Content-Type': 'application/x-www-form-urlencoded'} ) if response.status != 200: raise Exception( f"Looker login failed: {response.status} - " f"{response.data.decode('utf-8')}" ) data = json.loads(response.data.decode('utf-8')) token = data.get('access_token') if not token: raise Exception("No access_token in login response") print("Successfully obtained Looker API access token") return token def fetch_system_activity(token, view, fields, time_field, start_time, end_time): start_str = start_time.strftime('%Y-%m-%d %H:%M:%S') end_str = end_time.strftime('%Y-%m-%d %H:%M:%S') all_records = [] offset = 0 for page in range(MAX_PAGES): query_body = { "model": "system__activity", "view": view, "fields": fields, "filters": { time_field: f"{start_str} to {end_str}" }, "sorts": [f"{time_field} asc"], "limit": str(PAGE_SIZE), "offset": str(offset) } url = f"{LOOKER_BASE_URL}/api/4.0/queries/run/json" response = http.request( 'POST', url, body=json.dumps(query_body).encode('utf-8'), headers={ 'Authorization': f'token {token}', 'Content-Type': 'application/json' } ) if response.status == 429: print(f"Rate limited on {view} query. Stopping pagination.") break if response.status != 200: print(f"{view} query failed: {response.status} - " f"{response.data.decode('utf-8')}") break page_results = json.loads(response.data.decode('utf-8')) if not page_results: break all_records.extend(page_results) print(f"{view} page {page + 1}: {len(page_results)} records " f"(total: {len(all_records)})") if len(page_results) < PAGE_SIZE: break offset += PAGE_SIZE print(f"Total {view} records fetched: {len(all_records)}") return all_records def find_newest_time(events, history): newest = None for e in events: t = e.get('event.created_time') if t and (newest is None or t > newest): newest = t for h in history: t = h.get('history.created_time') if t and (newest is None or t > newest): newest = t return newest def load_state(bucket): try: blob = bucket.blob(STATE_KEY) if blob.exists(): return json.loads(blob.download_as_text()) except Exception as e: print(f"Warning: Could not load state: {e}") return {} def save_state(bucket, last_event_time_iso): try: state = { 'last_event_time': last_event_time_iso, 'last_run': datetime.now(timezone.utc).isoformat() } blob = bucket.blob(STATE_KEY) blob.upload_from_string( json.dumps(state, indent=2), content_type='application/json' ) print(f"Saved state: last_event_time={last_event_time_iso}") except Exception as e: print(f"Warning: Could not save state: {e}")requirements.txt:
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0
- 點選「部署」即可儲存並部署函式。
等待部署作業完成 (2 到 3 分鐘)。
建立 Cloud Scheduler 工作
- 前往 GCP 主控台的「Cloud Scheduler」。
- 點選「建立工作」。
請提供下列設定詳細資料：
設定 值 名稱 looker-audit-collector-hourly區域 選取與 Cloud Run 函式相同的區域 頻率 0 * * * *(每小時整點)時區 選取時區 (建議使用世界標準時間) 目標類型 Pub/Sub 主題 選取「 looker-audit-trigger」郵件內文 {}(空白 JSON 物件)點選「建立」。
測試整合項目
- 在 Cloud Scheduler 控制台中，找出您的工作 (
looker-audit-collector-hourly)。 - 按一下「強制執行」即可手動觸發工作。
- 稍等幾秒鐘。
- 前往「Cloud Run」>「Services」。
- 按一下
looker-audit-collector。 - 按一下 [Logs] (記錄) 分頁標籤。
確認函式是否已順利執行。尋找：
Fetching events from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00 Successfully obtained Looker API access token event page 1: X records (total: X) history page 1: X records (total: X) Wrote X records to gs://looker-audit-logs-gcs/looker-audit/looker_audit_YYYYMMDD_HHMMSS.ndjson Successfully processed X records (events: X, history: X)依序前往「Cloud Storage」>「Buckets」。
按一下
looker-audit-logs-gcs。前往
looker-audit/資料夾。確認是否已建立新的
.ndjson檔案，且檔案名稱包含目前的時間戳記。
如果記錄中顯示錯誤：
- HTTP 401：確認
LOOKER_CLIENT_ID和LOOKER_CLIENT_SECRET環境變數是否正確 - HTTP 403：確認 Looker 使用者是否具備
see_system_activity權限 - HTTP 429：頻率限制，函式會停止分頁，並在下次排定的執行作業中繼續
- 缺少環境變數：確認 Cloud Run 函式設定中已設定所有必要變數
擷取 Google SecOps 服務帳戶
- 依序前往「SIEM 設定」>「動態饋給」。
- 按一下「新增動態消息」。
- 按一下「設定單一動態饋給」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
Looker Audit Logs GCS)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 選取「Looker 稽核」做為「記錄類型」。
按一下「取得服務帳戶」。系統會顯示專屬服務帳戶電子郵件地址，例如：
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com複製這個電子郵件地址，以便在下一步中使用。
點選「下一步」。
指定下列輸入參數的值：
儲存空間 bucket URL：輸入 GCS bucket URI，並加上前置路徑：
gs://looker-audit-logs-gcs/looker-audit/
來源刪除選項：根據偏好設定選取刪除選項：
- 永不：移轉後一律不刪除任何檔案 (建議用於測試)。
- 刪除已轉移的檔案：成功轉移檔案後刪除檔案。
刪除已轉移的檔案和空白目錄：成功轉移後刪除檔案和空白目錄。
檔案存在時間上限：包含在過去天數內修改的檔案 (預設為 180 天)
資產命名空間：資產命名空間
擷取標籤：要套用至這個動態饋給事件的標籤
點選「下一步」。
在「Finalize」(完成) 畫面中檢查新的動態饋給設定，然後按一下「Submit」(提交)。
將 IAM 權限授予 Google SecOps 服務帳戶
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下
looker-audit-logs-gcs。 - 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料：
- 新增主體：貼上 Google SecOps 服務帳戶電子郵件地址
- 指派角色：選取「Storage 物件檢視者」
按一下 [儲存]。
選項 B：使用 Amazon S3 設定擷取作業
這個選項會使用 AWS Lambda 函式輪詢 Looker API，取得系統活動稽核事件，並將這些事件寫入 S3 bucket，供 Google SecOps 擷取。
為 Google SecOps 設定 AWS S3 值區和 IAM
- 按照這份使用者指南建立 Amazon S3 bucket：建立 bucket
- 請儲存 bucket 的「名稱」和「區域」，以供日後參考 (例如
looker-audit-logs)。 - 請按照這份使用者指南建立使用者：建立 IAM 使用者。
- 選取建立的「使用者」。
- 選取「安全憑證」分頁標籤。
- 在「Access Keys」部分中，按一下「Create Access Key」。
- 選取「第三方服務」做為「用途」。
- 點選「下一步」。
- 新增說明標記 (如有需要)。
- 按一下「建立存取金鑰」。
- 按一下「Download .csv file」(下載 .csv 檔案)，儲存「Access Key」(存取金鑰) 和「Secret Access Key」(私密存取金鑰)，以供日後參考。
- 按一下 [完成]。
- 選取「權限」分頁標籤。
- 在「權限政策」部分中，按一下「新增權限」。
- 選取「新增權限」。
- 選取「直接附加政策」。
- 搜尋「AmazonS3FullAccess」AmazonS3FullAccess政策。
- 選取政策。
- 點選「下一步」。
- 按一下「Add permissions」。
設定 S3 上傳的身分與存取權管理政策和角色
- 在 AWS 控制台中，依序前往「IAM」>「Policies」>「Create policy」>「JSON」分頁標籤。
- 複製並貼上下列政策。
政策 JSON (如果您輸入的 bucket 名稱不同，請替換
looker-audit-logs)：{ "Version": "2012-10-17", "Statement": [ { "Sid": "AllowPutObjects", "Effect": "Allow", "Action": "s3:PutObject", "Resource": "arn:aws:s3:::looker-audit-logs/*" }, { "Sid": "AllowGetStateObject", "Effect": "Allow", "Action": "s3:GetObject", "Resource": "arn:aws:s3:::looker-audit-logs/looker-audit/state.json" } ] }依序點選「下一步」>「建立政策」。
依序前往「IAM」>「角色」>「建立角色」>「AWS 服務」>「Lambda」。
附加新建立的政策。
為角色命名
LookerAuditCollectorRole，然後按一下「建立角色」。
建立 Lambda 函式
- 在 AWS 控制台中，依序前往「Lambda」>「Functions」>「Create function」。
- 按一下「從頭開始撰寫」。
請提供下列設定詳細資料：
設定 值 名稱 LookerAuditCollector執行階段 Python 3.13 架構 x86_64 執行角色 LookerAuditCollectorRole建立函式後，開啟「程式碼」分頁，刪除虛設常式並貼上下方程式碼 (LookerAuditCollector.py)。
import urllib3 import json import boto3 import os from datetime import datetime, timezone, timedelta import logging import urllib.parse logger = logging.getLogger() logger.setLevel(logging.INFO) http = urllib3.PoolManager( timeout=urllib3.Timeout(connect=10.0, read=60.0), retries=False, ) s3 = boto3.client('s3') BUCKET = os.environ['S3_BUCKET'] PREFIX = os.environ['S3_PREFIX'] STATE_KEY = os.environ['STATE_KEY'] LOOKER_BASE_URL = os.environ['LOOKER_BASE_URL'].rstrip('/') CLIENT_ID = os.environ['LOOKER_CLIENT_ID'] CLIENT_SECRET = os.environ['LOOKER_CLIENT_SECRET'] LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24')) PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '5000')) MAX_PAGES = int(os.environ.get('MAX_PAGES', '20')) def lambda_handler(event, context): try: state = load_state() now = datetime.now(timezone.utc) if state and state.get('last_event_time'): try: last_time = datetime.fromisoformat( state['last_event_time'].replace('Z', '+00:00') ) last_time = last_time - timedelta(minutes=2) except Exception as e: logger.warning(f"Could not parse last_event_time: {e}") last_time = now - timedelta(hours=LOOKBACK_HOURS) else: last_time = now - timedelta(hours=LOOKBACK_HOURS) logger.info(f"Fetching events from {last_time.isoformat()} to {now.isoformat()}") token = get_access_token() events = fetch_events(token, last_time, now) history = fetch_history(token, last_time, now) all_records = [] for e in events: e['_looker_record_type'] = 'event' all_records.append(e) for h in history: h['_looker_record_type'] = 'history' all_records.append(h) if not all_records: logger.info("No new records found.") save_state(now.isoformat()) return {'statusCode': 200, 'body': json.dumps({'events': 0})} timestamp = now.strftime('%Y%m%d_%H%M%S') object_key = f"{PREFIX}/looker_audit_{timestamp}.ndjson" ndjson = '\n'.join( [json.dumps(r, ensure_ascii=False, default=str) for r in all_records] ) + '\n' s3.put_object( Bucket=BUCKET, Key=object_key, Body=ndjson.encode('utf-8'), ContentType='application/x-ndjson' ) logger.info(f"Wrote {len(all_records)} records to s3://{BUCKET}/{object_key}") newest_time = find_newest_time(events, history) save_state(newest_time if newest_time else now.isoformat()) return { 'statusCode': 200, 'body': json.dumps({ 'events': len(events), 'history': len(history), 'total': len(all_records) }) } except Exception as e: logger.error(f"Lambda execution failed: {str(e)}") raise def get_access_token(): url = f"{LOOKER_BASE_URL}/api/4.0/login" encoded_body = urllib.parse.urlencode({ 'client_id': CLIENT_ID, 'client_secret': CLIENT_SECRET }).encode('utf-8') response = http.request( 'POST', url, body=encoded_body, headers={'Content-Type': 'application/x-www-form-urlencoded'} ) if response.status != 200: raise Exception( f"Login failed with status {response.status}: " f"{response.data.decode('utf-8')}" ) data = json.loads(response.data.decode('utf-8')) token = data.get('access_token') if not token: raise Exception("No access_token in login response") logger.info("Successfully obtained Looker API access token") return token def fetch_events(token, start_time, end_time): start_str = start_time.strftime('%Y-%m-%d %H:%M:%S') end_str = end_time.strftime('%Y-%m-%d %H:%M:%S') all_events = [] offset = 0 for page in range(MAX_PAGES): query_body = { "model": "system__activity", "view": "event", "fields": [ "event.id", "event.name", "event.category", "event.created_time", "event.is_api_call", "event.is_admin", "event.is_looker_employee", "user.id", "user.name", "user.email" ], "filters": { "event.created_time": f"{start_str} to {end_str}" }, "sorts": ["event.created_time asc"], "limit": str(PAGE_SIZE), "offset": str(offset) } url = f"{LOOKER_BASE_URL}/api/4.0/queries/run/json" response = http.request( 'POST', url, body=json.dumps(query_body).encode('utf-8'), headers={ 'Authorization': f'token {token}', 'Content-Type': 'application/json' } ) if response.status == 429: logger.warning("Rate limited on events query. Stopping pagination.") break if response.status != 200: logger.error( f"Events query failed: {response.status} - " f"{response.data.decode('utf-8')}" ) break page_results = json.loads(response.data.decode('utf-8')) if not page_results: logger.info(f"Events: No more results at offset {offset}") break all_events.extend(page_results) logger.info( f"Events page {page + 1}: Retrieved {len(page_results)} records " f"(total: {len(all_events)})" ) if len(page_results) < PAGE_SIZE: break offset += PAGE_SIZE logger.info(f"Total events fetched: {len(all_events)}") return all_events def fetch_history(token, start_time, end_time): start_str = start_time.strftime('%Y-%m-%d %H:%M:%S') end_str = end_time.strftime('%Y-%m-%d %H:%M:%S') all_history = [] offset = 0 for page in range(MAX_PAGES): query_body = { "model": "system__activity", "view": "history", "fields": [ "history.id", "history.created_time", "history.completed_time", "history.status", "history.source", "history.issuer_source", "history.runtime", "history.message", "query.id", "query.model", "query.view", "user.id", "user.name", "user.email", "dashboard.id", "dashboard.title", "look.id", "look.title" ], "filters": { "history.created_time": f"{start_str} to {end_str}" }, "sorts": ["history.created_time asc"], "limit": str(PAGE_SIZE), "offset": str(offset) } url = f"{LOOKER_BASE_URL}/api/4.0/queries/run/json" response = http.request( 'POST', url, body=json.dumps(query_body).encode('utf-8'), headers={ 'Authorization': f'token {token}', 'Content-Type': 'application/json' } ) if response.status == 429: logger.warning("Rate limited on history query. Stopping pagination.") break if response.status != 200: logger.error( f"History query failed: {response.status} - " f"{response.data.decode('utf-8')}" ) break page_results = json.loads(response.data.decode('utf-8')) if not page_results: logger.info(f"History: No more results at offset {offset}") break all_history.extend(page_results) logger.info( f"History page {page + 1}: Retrieved {len(page_results)} records " f"(total: {len(all_history)})" ) if len(page_results) < PAGE_SIZE: break offset += PAGE_SIZE logger.info(f"Total history records fetched: {len(all_history)}") return all_history def find_newest_time(events, history): newest = None for e in events: t = e.get('event.created_time') if t and (newest is None or t > newest): newest = t for h in history: t = h.get('history.created_time') if t and (newest is None or t > newest): newest = t return newest def load_state(): try: obj = s3.get_object(Bucket=BUCKET, Key=STATE_KEY) return json.loads(obj['Body'].read().decode('utf-8')) except s3.exceptions.NoSuchKey: logger.info("No previous state found, starting fresh") return None except Exception as e: logger.warning(f"Could not load state: {e}") return None def save_state(last_event_time): state = { 'last_event_time': last_event_time, 'last_run': datetime.now(timezone.utc).isoformat() } s3.put_object( Bucket=BUCKET, Key=STATE_KEY, Body=json.dumps(state, indent=2).encode('utf-8'), ContentType='application/json' ) logger.info(f"Saved state: last_event_time={last_event_time}")依序前往「Configuration」>「環境變數」>「Edit」>「Add new environment variable」。
輸入下列環境變數，並將 換成您的值。
環境變數
| 鍵 | 範例值 |
|---|---|
S3_BUCKET |
looker-audit-logs |
S3_PREFIX |
looker-audit/ |
STATE_KEY |
looker-audit/state.json |
LOOKER_BASE_URL |
https://your-instance.cloud.looker.com |
LOOKER_CLIENT_ID |
your-looker-client-id |
LOOKER_CLIENT_SECRET |
your-looker-client-secret |
LOOKBACK_HOURS |
24 |
PAGE_SIZE |
5000 |
MAX_PAGES |
20 |
- 建立函式後，請留在函式頁面 (或依序開啟「Lambda」>「Functions」>「your-function」)。
- 選取「設定」分頁標籤。
- 在「一般設定」面板中，按一下「編輯」。
將「Timeout」(逾時間隔) 變更為「5 minutes (300 seconds)」(5 分鐘 (300 秒))，然後按一下「Save」(儲存)。
建立 EventBridge 排程
- 依序前往「Amazon EventBridge」>「Scheduler」>「Create schedule」。
- 請提供下列設定詳細資料：
- 週期性時間表：費率 (
1 hour) - 目標：您的 Lambda 函式
LookerAuditCollector - Name (名稱)：
LookerAuditCollector-1h
- 週期性時間表：費率 (
- 按一下「建立時間表」。
在 Google SecOps 中設定資訊提供，擷取 Looker 稽核記錄
- 依序前往「SIEM 設定」>「動態饋給」。
- 按一下「新增動態消息」。
- 在下一個頁面中，按一下「設定單一動態饋給」。
- 輸入動態饋給名稱的專屬名稱。
- 選取「Amazon S3 V2」做為「來源類型」。
- 選取「Looker 稽核」做為「記錄類型」。
- 依序點按「繼續」和「提交」。
為下列欄位指定值：
- S3 URI：
s3://looker-audit-logs/looker-audit/ - 來源刪除選項：根據偏好設定選取刪除選項
- 檔案存在時間上限：包含在過去天數內修改的檔案 (預設為 180 天)
- 存取金鑰 ID：具有 S3 值區存取權的使用者存取金鑰
- 存取密鑰：具有 S3 bucket 存取權的使用者私密金鑰
- 資產命名空間：資產命名空間
- 擷取標籤：要套用至這個動態饋給事件的標籤
- S3 URI：
依序點按「繼續」和「提交」。
Looker 系統活動記錄資料參考資料
下表說明這項整合服務收集的 Looker 系統活動探索資料，以及可從中取得的重要資料：
| 探索 | 收集到的資料 | 保留 |
|---|---|---|
| 事件 | 使用者驗證事件、內容建立和修改、權限變更、API 呼叫、排定傳送時間的事件、下載事件 | 90 天 (預設) |
| 歷史記錄 | 查詢執行記錄、資訊主頁和 Look 存取權、查詢執行時間和狀態、查詢來源 (UI、API、排程) | 90 天 (預設) |
UDM 對應表
| 記錄欄位 | UDM 對應 | 邏輯 |
|---|---|---|
| 群組編輯連結 | additional.fields.Group_Edit_Link_label.value.string_value | 直接複製值 |
| 群組 ID | additional.fields.Group_ID_label.value.string_value | 直接複製值 |
| 記錄最近一次跑步的長度 (以秒為單位) | additional.fields.History_Most_Recent_Run_Length_in_Seconds_label.value.string_value | 直接複製值 |
| 歷史記錄 Slug | additional.fields.History_Slug_label.value.string_value | 直接複製值 |
| 記錄來源 | additional.fields.History_Source_label.value.string_value | 直接複製值 |
| 記錄狀態 | additional.fields.History_Status_label.value.string_value | 直接複製值 |
| Look Link | additional.fields.Look_Link_label.value.string_value | 直接複製值 |
| Look 名稱 | additional.fields.Look_Title_label.value.string_value | 直接複製值 |
| 使用者編輯連結 | additional.fields.User_Edit_Link_label.value.string_value | 直接複製值 |
| 使用者主資料夾 | additional.fields.User_Home_Folder_label.value.string_value | 直接複製值 |
| dashboard.link | additional.fields.dashboard_link_label.value.string_value | 直接複製值 |
| dashboard.title | additional.fields.dashboard_title_label.value.string_value | 直接複製值 |
| history.source | additional.fields.history_source_label.value.string_value | 直接複製值 |
| history.status | additional.fields.history_status_label.value.string_value | 直接複製值 |
| history.id | additional.fields.id_label.value.string_value | 已轉換為字串 |
| history.connection_name | additional.fields.name_label.value.string_value | 直接複製值 |
| query.model | additional.fields.query_model_label.value.string_value | 直接複製值 |
| query.view | additional.fields.query_view_label.value.string_value | 直接複製值 |
| sql_text.text | additional.fields.sql_text_text_label.value.string_value | 直接複製值 |
| 記錄建立時間 | metadata.event_timestamp | 使用 ISO8601、RFC 3339 或 yyyy-MM-dd HH:mm:ss 格式剖析 |
| has_principal_user | metadata.event_type | 如果 has_principal 和 has_target 皆為 true，則設為「NETWORK_CONNECTION」；如果 has_principal_user 為 true，則設為「USER_UNCATEGORIZED」；如果 has_principal 為 true，則設為「STATUS_UPDATE」；否則設為「GENERIC_EVENT」 |
| has_principal | metadata.event_type | |
| has_target | metadata.event_type | |
| 使用者電子郵件 | principal.email | 如果使用者電子郵件地址不為空值，則為該地址，否則為 user.email |
| user.email | principal.email | |
| 群組名稱 | principal.group.group_display_name | 直接複製值 |
| 使用者 ID | principal.user.product_object_id | 如果使用者 ID 不為空白，則為使用者 ID 的值，否則為 user.id |
| user.id | principal.user.product_object_id | |
| 使用者名稱 | principal.user.userid | 如果使用者名稱不為空白，則為使用者名稱的值，否則為 user.name |
| user.name | principal.user.userid | |
| Kevin Liu | security_result.category_details | 如果 Kevin Liu 不為空白，則與「Kevin_Liu_label」合併；如果記錄 ID 不為空白，則與「History_ID_label」合併；如果記錄建立日期不為空白，則與「History_Created_Date_label」合併 |
| 記錄 ID | security_result.category_details | |
| 記錄建立日期 | security_result.category_details | |
| Look 說明 | security_result.description | 直接複製值 |
| 依使用者名稱排序 | target.hostname | 如果使用者開發分支版本名稱不是空白，系統就會複製值 |
user.name |
principal.user.user_display_name |
從變更記錄對應 |
變更記錄
還有其他問題嗎？向社群成員和 Google SecOps 專業人員尋求答案。