收集 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。
将以下值复制并保存在安全位置：
- 客户端 ID：用于 API 身份验证的公开标识符
- 客户端密钥：用于 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。
您可以在 Looker 实例中依次前往管理 > API，找到 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 存储桶，以便 Google SecOps 提取。
创建 Google Cloud Storage 存储桶
- 前往 Google Cloud 控制台。
- 选择您的项目或创建新项目。
- 在导航菜单中，依次前往 Cloud Storage > 存储分区。
- 点击创建存储分区。
提供以下配置详细信息：
设置 值 为存储桶命名 输入一个全局唯一的名称（例如 looker-audit-logs-gcs）位置类型 根据您的需求进行选择（区域级、双区域、多区域） 位置 选择营业地点（例如 us-central1）存储类别 标准（建议用于经常访问的日志） 访问权限控制 均匀（推荐） 保护工具 可选：启用对象版本控制或保留政策 点击创建。
为 Cloud Run 函数创建服务账号
- 在 GCP 控制台中，依次前往 IAM 和管理 > 服务账号。
- 点击创建服务账号。
- 提供以下配置详细信息：
- 服务账号名称：输入
looker-audit-collector-sa - 服务账号说明：输入
Service account for Cloud Run function to collect Looker audit logs
- 服务账号名称：输入
- 点击创建并继续。
- 在向此服务账号授予对项目的访问权限部分中，添加以下角色：
- 点击选择角色。
- 搜索并选择 Storage Object Admin。
- 点击 + 添加其他角色。
- 搜索并选择 Cloud Run Invoker。
- 点击 + 添加其他角色。
- 搜索并选择 Cloud Functions Invoker。
- 点击继续。
- 点击完成。
授予对 GCS 存储桶的 IAM 权限
- 前往 Cloud Storage > 存储分区。
- 点击您的存储桶名称 (
looker-audit-logs-gcs)。 - 前往权限标签页。
- 点击授予访问权限。
- 提供以下配置详细信息：
- 添加主账号：输入服务账号电子邮件地址 (
looker-audit-collector-sa@PROJECT_ID.iam.gserviceaccount.com) - 分配角色：选择 Storage Object Admin
- 添加主账号：输入服务账号电子邮件地址 (
- 点击保存。
创建 Pub/Sub 主题
- 在 GCP 控制台中，前往 Pub/Sub > 主题。
- 点击创建主题。
- 提供以下配置详细信息：
- 主题 ID：输入
looker-audit-trigger - 将其他设置保留为默认值
- 主题 ID：输入
- 点击创建。
创建 Cloud Run 函数以收集日志
- 在 GCP 控制台中，前往 Cloud Run。
- 点击创建服务。
- 选择函数（使用内嵌编辑器创建函数）。
在配置部分中，提供以下配置详细信息：
设置 值 Service 名称 looker-audit-collector区域 选择与您的 GCS 存储桶匹配的区域（例如 us-central1）运行时 选择 Python 3.12 或更高版本 在触发器（可选）部分中：
- 点击 + 添加触发器。
- 选择 Cloud Pub/Sub。
- 在选择 Cloud Pub/Sub 主题中，选择
looker-audit-trigger。 - 点击保存。
在身份验证部分中：
- 选择需要进行身份验证。
- 检查 Identity and Access Management (IAM)。
向下滚动并展开容器、网络、安全性。
前往安全性标签页：
- 服务账号：选择
looker-audit-collector-sa
- 服务账号：选择
前往容器标签页：
- 点击变量和密钥。
- 为每个环境变量点击+ 添加变量：
变量名称 示例值 说明 GCS_BUCKETlooker-audit-logs-gcsGCS 存储桶名称 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每次查询的最大网页数 在变量和 Secret 部分中，向下滚动到请求：
- 请求超时：输入
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 Console 中，前往 Cloud Scheduler。
- 点击创建作业。
提供以下配置详细信息：
设置 值 名称 looker-audit-collector-hourly区域 选择与 Cloud Run 函数相同的区域 频率 0 * * * *（每小时一次，整点时）时区 选择时区（建议选择世界协调时间 [UTC]） 目标类型 Pub/Sub 主题 选择 looker-audit-trigger消息正文 {}（空 JSON 对象）点击创建。
测试集成
- 在 Cloud Scheduler 控制台中，找到您的作业 (
looker-audit-collector-hourly)。 - 点击强制运行以手动触发作业。
- 等待几秒钟。
- 前往 Cloud Run > 服务。
- 点击
looker-audit-collector。 - 点击日志标签页。
验证函数是否已成功执行。查找：
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 > 存储分区。
点击
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 设置 > Feed。
- 点击添加新 Feed。
- 点击配置单个 Feed。
- 在 Feed 名称字段中，输入 Feed 的名称（例如
Looker Audit Logs GCS）。 - 选择 Google Cloud Storage V2 作为来源类型。
- 选择 Looker 审核作为日志类型。
点击获取服务账号。系统会显示一个唯一的服务账号电子邮件地址，例如：
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com复制此电子邮件地址，以便在下一步中使用。
点击下一步。
为以下输入参数指定值：
存储桶网址：输入带有前缀路径的 GCS 存储桶 URI：
gs://looker-audit-logs-gcs/looker-audit/
来源删除选项：根据您的偏好选择删除选项：
- 永不：转移后永不删除任何文件（建议用于测试）。
- 删除已转移的文件：在成功转移后删除文件。
删除已转移的文件和空目录：成功转移后删除文件和空目录。
文件存在时间上限：包含在过去指定天数内修改的文件（默认值为 180 天）
资产命名空间：资产命名空间
注入标签：要应用于此 Feed 中事件的标签
点击下一步。
在最终确定界面中查看新的 Feed 配置，然后点击提交。
向 Google SecOps 服务账号授予 IAM 权限
- 前往 Cloud Storage > 存储分区。
- 点击
looker-audit-logs-gcs。 - 前往权限标签页。
- 点击授予访问权限。
- 提供以下配置详细信息：
- 添加主账号：粘贴 Google SecOps 服务账号电子邮件地址
- 分配角色：选择 Storage Object Viewer
点击保存。
选项 B：使用 Amazon S3 配置注入
此选项使用 AWS Lambda 函数轮询 Looker API 以获取系统活动审核事件，并将这些事件写入 S3 存储桶，以便 Google SecOps 提取。
为 Google SecOps 配置 AWS S3 存储桶和 IAM
- 按照以下用户指南创建 Amazon S3 存储桶：创建存储桶
- 保存存储桶名称和区域以供日后参考（例如
looker-audit-logs）。 - 按照以下用户指南创建用户：创建 IAM 用户。
- 选择创建的用户。
- 选择安全凭据标签页。
- 在访问密钥部分中，点击创建访问密钥。
- 选择第三方服务作为使用情形。
- 点击下一步。
- 可选：添加说明标记。
- 点击创建访问密钥。
- 点击下载 .csv 文件以保存访问密钥和密钥，供日后参考。
- 点击完成。
- 选择权限标签页。
- 在权限政策部分中，点击添加权限。
- 选择添加权限。
- 选择直接附加政策。
- 搜索 AmazonS3FullAccess 政策。
- 选择相应政策。
- 点击下一步。
- 点击添加权限。
为 S3 上传配置 IAM 政策和角色
- 在 AWS 控制台中，依次前往 IAM > 政策 > 创建政策 > JSON 标签页。
- 复制并粘贴以下政策。
政策 JSON（如果您输入了其他存储桶名称，请替换
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 > 函数 > 创建函数。
- 点击从头开始创作。
提供以下配置详细信息：
设置 值 名称 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}")依次前往配置 > 环境变量 > 修改 > 添加新的环境变量。
输入以下环境变量，并将其替换为您的值。
环境变量
| 键 | 示例值 |
|---|---|
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 > 函数 > 您的函数）。
- 选择配置标签页。
- 在常规配置面板中，点击修改。
将超时更改为 5 分钟（300 秒），然后点击保存。
创建 EventBridge 计划
- 依次前往 Amazon EventBridge > 调度器 > 创建调度。
- 提供以下配置详细信息：
- 周期性安排：费率 (
1 hour) - 目标：您的 Lambda 函数
LookerAuditCollector - 名称：
LookerAuditCollector-1h
- 周期性安排：费率 (
- 点击创建时间表。
在 Google SecOps 中配置 Feed 以注入 Looker 审核日志
- 依次前往 SIEM 设置 > Feed。
- 点击添加新 Feed。
- 在下一页上，点击配置单个 Feed。
- 为Feed 名称输入一个唯一名称。
- 选择 Amazon S3 V2 作为来源类型。
- 选择 Looker 审核作为日志类型。
- 点击下一步，然后点击提交。
为以下字段指定值：
- S3 URI：
s3://looker-audit-logs/looker-audit/ - 来源删除选项：根据您的偏好选择删除选项
- 文件存在时间上限：包含在过去指定天数内修改的文件（默认值为 180 天）
- 访问密钥 ID：有权访问 S3 存储桶的用户访问密钥
- 私有访问密钥：具有 S3 存储桶访问权限的用户私有密钥
- 资产命名空间：资产命名空间
- 注入标签：要应用于此 Feed 中事件的标签
- S3 URI：
点击下一步，然后点击提交。
Looker 系统活动数据参考
下表介绍了此集成功能收集的 Looker 系统活动探索中的关键数据：
| 探索 | 收集的数据 | 保留 |
|---|---|---|
| 事件 | 用户身份验证事件、内容创建和修改、权限更改、API 调用、预定交付事件、下载事件 | 90 天（默认值） |
| 历史记录 | 查询执行历史记录、信息中心和 Look 访问权限、查询运行时和状态、查询来源（界面、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 | 直接复制值 |
| 历史记录处理后标题 | 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 | 如果 User-ID 不为空，则返回 User-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 专业人士的解答。