收集 FortiCNAPP（以前称为 Lacework）日志
支持的平台：
Google SecOps
SIEM
本文档介绍了如何使用 Google Cloud Storage V2 将 FortiCNAPP（以前称为 Lacework）日志注入到 Google Security Operations。
FortiCNAPP 是一个云原生应用保护平台 (CNAPP)，可在多云环境中提供云安全状况管理、工作负载保护和威胁检测。它会生成可通过 Lacework REST API 收集的提醒、合规性调查结果和审核日志。
准备工作
请确保您满足以下前提条件：
- Google SecOps 实例
- 已启用 Cloud Storage API 的 GCP 项目
- 创建和管理 GCS 存储分区的权限
- 管理 GCS 存储分区的 IAM 政策的权限
- 创建 Cloud Run 服务、Pub/Sub 主题和 Cloud Scheduler 作业的权限
- 对 FortiCNAPP（以前称为 Lacework）控制台的特权访问权限（具有管理员权限）
- 已启用 API 密钥访问权限的 Lacework 账号
创建 Google Cloud Storage 存储桶
- 前往 Google Cloud 控制台。
- 选择您的项目或创建新项目。
- 在导航菜单中，依次前往 Cloud Storage > 存储分区。
点击创建存储分区。
提供以下配置详细信息：
设置 值 为存储桶命名 输入一个全局唯一的名称（例如 lacework-logs）位置类型 根据您的需求进行选择（区域级、双区域、多区域） 位置 选择营业地点（例如 us-central1）存储类别 标准（建议用于经常访问的日志） 访问权限控制 均匀（推荐） 保护工具 可选：启用对象版本控制或保留政策 点击创建。
收集 FortiCNAPP（以前称为 Lacework）API 凭据
生成 API 密钥
- 登录您的 Lacework 控制台。
- 依次前往设置 > 配置 > API 密钥。
- 点击 + 添加新对比项。
- 输入 API 密钥的名称（例如
Google SecOps Integration）。 - （可选）输入说明。
点击保存。
复制以下详细信息并将其保存在安全的位置：
- 密钥 ID：生成的 API 密钥 ID
- 密钥：生成的 API 密钥（仅显示一次）
记下浏览器地址栏中的 Lacework 账号网址。
- 格式：
https://<ACCOUNT>.lacework.net - 示例：如果您的 Lacework 控制台网址为
https://acme.lacework.net，则您的账号名称为acme
- 格式：
验证权限
如需验证账号是否具有所需权限，请执行以下操作：
- 登录 Lacework 控制台。
- 依次前往设置 > 配置 > API 密钥。
- 如果您可以查看“API 密钥”页面并创建密钥，则表示您拥有所需的权限。
- 如果您看不到此选项，请与您的管理员联系，让其授予您管理员级访问权限。
测试 API 访问权限
在继续进行集成之前，请先测试您的凭据：
# Replace with your actual credentials LW_ACCOUNT="your-account-name" LW_KEY_ID="your-api-key-id" LW_SECRET="your-api-secret" # Get a temporary access token TOKEN=$(curl -s -X POST "https://${LW_ACCOUNT}.lacework.net/api/v2/access/tokens" \ -H "X-LW-UAKS: ${LW_SECRET}" \ -H "Content-Type: application/json" \ -d "{\"keyId\": \"${LW_KEY_ID}\", \"expiryTime\": 3600}" | python3 -c "import sys,json; print(json.load(sys.stdin).get('token',''))") # Test API access - list alerts curl -v -H "Authorization: Bearer ${TOKEN}" \ "https://${LW_ACCOUNT}.lacework.net/api/v2/Alerts?startTime=$(date -u -v-1d +%Y-%m-%dT%H:%M:%SZ)&endTime=$(date -u +%Y-%m-%dT%H:%M:%SZ)"
为 Cloud Run 函数创建服务账号
Cloud Run 函数需要一个服务账号，该账号具有写入 GCS 存储桶的权限，并且可以由 Pub/Sub 调用。
创建服务账号
- 在 GCP 控制台中，依次前往 IAM 和管理 > 服务账号。
点击创建服务账号。
提供以下配置详细信息：
- 服务账号名称：输入
lacework-logs-collector-sa - 服务账号说明：输入
Service account for Cloud Run function to collect FortiCNAPP (formerly Lacework) logs
- 服务账号名称：输入
点击创建并继续。
在向此服务账号授予对项目的访问权限部分中，添加以下角色：
- 点击选择角色。
- 搜索并选择 Storage Object Admin。
- 点击 + 添加其他角色。
- 搜索并选择 Cloud Run Invoker。
- 点击 + 添加其他角色。
- 搜索并选择 Cloud Functions Invoker。
点击继续。
点击完成。
必须拥有这些角色，才能：
- Storage Object Admin：将日志写入 GCS 存储桶并管理状态文件
- Cloud Run Invoker：允许 Pub/Sub 调用函数
- Cloud Functions Invoker：允许调用函数
授予对 GCS 存储桶的 IAM 权限
向服务账号授予对 GCS 存储桶的写入权限：
- 前往 Cloud Storage > 存储分区。
- 点击您的存储桶名称（例如
lacework-logs）。 - 前往权限标签页。
点击授予访问权限。
提供以下配置详细信息：
- 添加主账号：输入服务账号电子邮件地址（例如
lacework-logs-collector-sa@PROJECT_ID.iam.gserviceaccount.com） - 分配角色：选择 Storage Object Admin
- 添加主账号：输入服务账号电子邮件地址（例如
点击保存。
创建 Pub/Sub 主题
创建一个 Pub/Sub 主题，Cloud Scheduler 将向该主题发布消息，而 Cloud Run 函数将订阅该主题。
- 在 GCP 控制台中，前往 Pub/Sub > 主题。
点击创建主题。
提供以下配置详细信息：
- 主题 ID：输入
lacework-logs-trigger - 将其他设置保留为默认值
- 主题 ID：输入
点击创建。
创建 Cloud Run 函数以收集日志
Cloud Run 函数将由来自 Cloud Scheduler 的 Pub/Sub 消息触发，以从 FortiCNAPP（以前称为 Lacework）API 中提取日志并将其写入 GCS。
- 在 GCP 控制台中，前往 Cloud Run。
- 点击创建服务。
选择函数（使用内嵌编辑器创建函数）。
在配置部分中，提供以下配置详细信息：
设置 值 Service 名称 lacework-logs-collector区域 选择与您的 GCS 存储桶匹配的区域（例如 us-central1）运行时 选择 Python 3.12 或更高版本 在触发器（可选）部分中：
- 点击 + 添加触发器。
- 选择 Cloud Pub/Sub。
- 在选择 Cloud Pub/Sub 主题部分，选择主题
lacework-logs-trigger。 - 点击保存。
在身份验证部分中：
- 选择需要进行身份验证。
- 检查 Identity and Access Management (IAM)。
向下滚动并展开容器、网络、安全性。
前往安全性标签页：
- 服务账号：选择服务账号
lacework-logs-collector-sa。
- 服务账号：选择服务账号
前往容器标签页：
- 点击变量和密钥。
- 为每个环境变量点击+ 添加变量：
变量名称 示例值 说明 GCS_BUCKETlacework-logsGCS 存储桶名称 GCS_PREFIXlacework日志文件的前缀 STATE_KEYlacework/state.json状态文件路径 LW_ACCOUNTacmeLacework 账号名称 LW_KEY_IDyour-api-key-idLacework API 密钥 ID LW_SECRETyour-api-secretLacework API 密钥 MAX_RECORDS5000每次运行的记录数上限 PAGE_SIZE500每页记录数 LOOKBACK_HOURS24初始回溯期 在变量和 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 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() # Environment variables GCS_BUCKET = os.environ.get('GCS_BUCKET') GCS_PREFIX = os.environ.get('GCS_PREFIX', 'lacework') STATE_KEY = os.environ.get('STATE_KEY', 'lacework/state.json') LW_ACCOUNT = os.environ.get('LW_ACCOUNT') LW_KEY_ID = os.environ.get('LW_KEY_ID') LW_SECRET = os.environ.get('LW_SECRET') MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '5000')) PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '500')) LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24')) # Lacework API base URL API_BASE_TEMPLATE = 'https://{account}.lacework.net/api/v2' # Log endpoints to fetch ENDPOINTS = [ {'name': 'alerts', 'path': '/Alerts', 'time_field': 'startTime', 'results_key': 'data'}, {'name': 'audit_logs', 'path': '/AuditLogs', 'time_field': 'createdTime', 'results_key': 'data'}, ] def get_access_token(api_base: str, key_id: str, secret: str) -> str: """Get a temporary access token from Lacework API.""" token_url = f"{api_base}/access/tokens" body = json.dumps({ 'keyId': key_id, 'expiryTime': 3600 }).encode('utf-8') headers = { 'X-LW-UAKS': secret, 'Content-Type': 'application/json', } response = http.request('POST', token_url, body=body, headers=headers) if response.status != 201: raise Exception(f"Failed to get access token: HTTP {response.status} - {response.data.decode('utf-8')}") token_data = json.loads(response.data.decode('utf-8')) return token_data['token'] @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch FortiCNAPP (formerly Lacework) logs and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ if not all([GCS_BUCKET, LW_ACCOUNT, LW_KEY_ID, LW_SECRET]): print('Error: Missing required environment variables') return try: bucket = storage_client.bucket(GCS_BUCKET) api_base = API_BASE_TEMPLATE.format(account=LW_ACCOUNT) # Get access token token = get_access_token(api_base, LW_KEY_ID, LW_SECRET) print("Successfully obtained access token") # Load state state = load_state(bucket, STATE_KEY) # Determine time window now = datetime.now(timezone.utc) all_records = [] for endpoint in ENDPOINTS: ep_name = endpoint['name'] last_time_str = None if isinstance(state, dict) and state.get(f"last_{ep_name}_time"): try: last_time = parse_datetime(state[f"last_{ep_name}_time"]) # Overlap by 2 minutes to catch any delayed events last_time = last_time - timedelta(minutes=2) last_time_str = last_time.strftime('%Y-%m-%dT%H:%M:%SZ') except Exception as e: print(f"Warning: Could not parse last_{ep_name}_time: {e}") if last_time_str is None: last_time = now - timedelta(hours=LOOKBACK_HOURS) last_time_str = last_time.strftime('%Y-%m-%dT%H:%M:%SZ') end_time_str = now.strftime('%Y-%m-%dT%H:%M:%SZ') print(f"Fetching {ep_name} from {last_time_str} to {end_time_str}") records, newest_event_time = fetch_logs( api_base=api_base, token=token, endpoint=endpoint, start_time=last_time_str, end_time=end_time_str, page_size=PAGE_SIZE, max_records=MAX_RECORDS, ) # Tag records with endpoint type for record in records: record['_lw_log_type'] = ep_name all_records.extend(records) # Update state for this endpoint if newest_event_time: state[f"last_{ep_name}_time"] = newest_event_time else: state[f"last_{ep_name}_time"] = end_time_str print(f"Fetched {len(records)} {ep_name} records") if not all_records: print("No new log records found.") save_state(bucket, STATE_KEY, state) return # Write to GCS as NDJSON timestamp = now.strftime('%Y%m%d_%H%M%S') object_key = f"{GCS_PREFIX}/logs_{timestamp}.ndjson" blob = bucket.blob(object_key) ndjson = '\n'.join([json.dumps(record, ensure_ascii=False) for record 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}") # Save state save_state(bucket, STATE_KEY, state) print(f"Successfully processed {len(all_records)} records") except Exception as e: print(f'Error processing logs: {str(e)}') raise def parse_datetime(value: str) -> datetime: """Parse ISO datetime string to datetime object.""" if value.endswith("Z"): value = value[:-1] + "+00:00" return datetime.fromisoformat(value) 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: {e}") return {} def save_state(bucket, key, state: dict): """Save the state to GCS state file.""" try: blob = bucket.blob(key) blob.upload_from_string( json.dumps(state, indent=2), content_type='application/json' ) print(f"Saved state: {json.dumps(state)}") except Exception as e: print(f"Warning: Could not save state: {e}") def fetch_logs(api_base: str, token: str, endpoint: dict, start_time: str, end_time: str, page_size: int, max_records: int): """ Fetch logs from Lacework API with pagination and rate limiting. Args: api_base: API base URL token: Bearer access token endpoint: Endpoint configuration dict start_time: Start time in ISO format end_time: End time in ISO format page_size: Number of records per page max_records: Maximum total records to fetch Returns: Tuple of (records list, newest_event_time ISO string) """ headers = { 'Authorization': f'Bearer {token}', 'Accept': 'application/json', 'Content-Type': 'application/json', 'User-Agent': 'GoogleSecOps-LaceworkCollector/1.0' } ep_path = endpoint['path'] time_field = endpoint['time_field'] results_key = endpoint['results_key'] records = [] newest_time = None page_num = 0 backoff = 1.0 next_page = None while True: page_num += 1 if len(records) >= max_records: print(f"Reached max_records limit ({max_records}) for {endpoint['name']}") break # Build request URL if next_page: url = next_page else: url = f"{api_base}{ep_path}?startTime={start_time}&endTime={end_time}" try: response = http.request('GET', url, headers=headers) # 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"HTTP Error: {response.status}") response_text = response.data.decode('utf-8') print(f"Response body: {response_text}") return [], None data = json.loads(response.data.decode('utf-8')) page_results = data.get(results_key, []) if not page_results: print(f"No more results (empty page) for {endpoint['name']}") break print(f"Page {page_num}: Retrieved {len(page_results)} {endpoint['name']} events") records.extend(page_results) # Track newest event time for event in page_results: try: event_time = event.get(time_field) if event_time: if newest_time is None or parse_datetime(event_time) > parse_datetime(newest_time): newest_time = event_time except Exception as e: print(f"Warning: Could not parse event time: {e}") # Check for next page via paging object paging = data.get('paging', {}) next_page_url = paging.get('urls', {}).get('nextPage') if not next_page_url: print(f"No more pages for {endpoint['name']}") break next_page = next_page_url except Exception as e: print(f"Error fetching {endpoint['name']} logs: {e}") return [], None print(f"Retrieved {len(records)} total {endpoint['name']} records from {page_num} pages") return records, newest_timerequirements.txt:
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0
点击部署以保存并部署该函数。
等待部署完成（2-3 分钟）。
创建 Cloud Scheduler 作业
Cloud Scheduler 会定期向 Pub/Sub 主题发布消息，从而触发 Cloud Run 函数。
- 在 GCP Console 中，前往 Cloud Scheduler。
点击创建作业。
提供以下配置详细信息：
设置 值 名称 lacework-logs-collector-hourly区域 选择与 Cloud Run 函数相同的区域 频率 0 * * * *（每小时一次，整点时）时区 选择时区（建议选择世界协调时间 [UTC]） 目标类型 Pub/Sub 主题 选择主题 lacework-logs-trigger消息正文 {}（空 JSON 对象）点击创建。
时间表频率选项
根据日志量和延迟时间要求选择频次：
| 频率 | Cron 表达式 | 使用场景 |
|---|---|---|
| 每隔 5 分钟 | */5 * * * * |
大批量、低延迟 |
| 每隔 15 分钟 | */15 * * * * |
搜索量中等 |
| 每小时 | 0 * * * * |
标准（推荐） |
| 每 6 小时 | 0 */6 * * * |
低成交量、批处理 |
| 每天 | 0 0 * * * |
历史数据收集 |
测试集成
- 在 Cloud Scheduler 控制台中，找到您的作业。
- 点击强制运行以手动触发作业。
- 等待几秒钟。
- 前往 Cloud Run > 服务。
- 点击
lacework-logs-collector。 - 点击日志标签页。
验证函数是否已成功执行。查找：
Successfully obtained access token Fetching alerts from YYYY-MM-DDTHH:MM:SSZ to YYYY-MM-DDTHH:MM:SSZ Page 1: Retrieved X alerts events Fetched X alerts records Fetching audit_logs from YYYY-MM-DDTHH:MM:SSZ to YYYY-MM-DDTHH:MM:SSZ Page 1: Retrieved X audit_logs events Fetched X audit_logs records Wrote X records to gs://lacework-logs/lacework/logs_YYYYMMDD_HHMMSS.ndjson Successfully processed X records前往 Cloud Storage > 存储分区。
点击您的存储桶名称 (
lacework-logs)。转到
lacework/文件夹。验证是否已创建具有当前时间戳的新
.ndjson文件。
如果您在日志中看到错误，请执行以下操作：
- HTTP 401：检查环境变量中的 API 凭据，或者令牌可能已过期
- HTTP 403：在 Lacework 控制台中验证 API 密钥是否具有必需的权限
- HTTP 429：速率限制 - 函数将自动重试并进行退避
- 缺少环境变量：检查是否已设置所有必需的变量
在 Google SecOps 中配置 Feed 以注入 FortiCNAPP（前称 Lacework）日志
- 依次前往 SIEM 设置 > Feed。
- 点击添加新 Feed。
- 点击配置单个 Feed。
- 在 Feed 名称字段中，输入 Feed 的名称（例如
Lacework Logs）。 - 选择 Google Cloud Storage V2 作为来源类型。
- 选择 Lacework Cloud Security 作为日志类型。
点击获取服务账号。系统会显示一个唯一的服务账号电子邮件地址，例如：
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com复制此电子邮件地址。
点击下一步。
为以下输入参数指定值：
存储桶网址：输入带有前缀路径的 GCS 存储桶 URI：
gs://lacework-logs/lacework/- 替换：
lacework-logs：您的 GCS 存储桶名称。lacework：存储日志的可选前缀/文件夹路径（留空表示根目录）。
- 替换：
来源删除选项：根据您的偏好选择删除选项：
- 永不：转移后永不删除任何文件（建议用于测试）。
- 删除已转移的文件：在成功转移后删除文件。
删除已转移的文件和空目录：成功转移后删除文件和空目录。
文件存在时间上限：包含在过去指定天数内修改的文件（默认值为 180 天）
资产命名空间：资产命名空间
注入标签：要应用于此 Feed 中事件的标签
点击下一步。
在最终确定界面中查看新的 Feed 配置，然后点击提交。
向 Google SecOps 服务账号授予 IAM 权限
Google SecOps 服务账号需要您的 GCS 存储桶的 Storage Object Viewer 角色。
- 前往 Cloud Storage > 存储分区。
- 点击您的存储桶名称。
- 前往权限标签页。
点击授予访问权限。
提供以下配置详细信息：
- 添加主账号：粘贴 Google SecOps 服务账号电子邮件地址
- 分配角色：选择 Storage Object Viewer
点击保存。
支持的 Lacework Cloud Security 示例日志
代理或机器信息（主机清单）
{ "AGENT_VERSION": "6.7.6-4ce73a7b", "CREATED_TIME": "Thu, 03 Nov 2022 02:09:36 -0700", "HOSTNAME": "host-agent-1", "IP_ADDR": "10.0.0.1", "LAST_UPDATE": "Wed, 18 Oct 2023 17:59:09 -0700", "MID": 6516601498285932156, "MODE": "ebpf", "OS": "Linux", "STATUS": "ACTIVE", "TAGS": { "Account": "999999999999", "AmiId": "ami-00000000000000000", "ExternalIp": "203.0.113.10", "Hostname": "internal-host-1.zone.compute.internal", "InstanceId": "i-00000000000000000", "InternalIp": "172.16.1.10", "LwTokenShort": "DUMMYTOKENABCD123456", "Name": "proxy-DMZ-app-1", "ResourceType": "proxy-machines", "SubnetId": "subnet-00000000000000000", "VmInstanceType": "t3.small", "VmProvider": "AWS", "VpcId": "vpc-00000000000000000", "Zone": "us-west-2a", "arch": "amd64", "falconx.io/application": "proxy-machines", "falconx.io/environment": "prod", "falconx.io/project": "edge", "falconx.io/team": "edge", "os": "linux" } }文件元数据或完整性
{ "CREATED_TIME": "Wed, 18 Oct 2023 17:02:01 -0700", "FILEDATA_HASH": "DUMMYHASH582C741AD91CA817B4718DEAA4E8A83C0B9D92E2", "FILE_PATH": "/usr/local/bin/secure_config", "MID": 7371220731851617371, "MTIME": "Fri, 25 Aug 2023 13:03:09 -0700", "SIZE": 8078 }主机漏洞评估
{ "CVE_PROPS": { "description": "DOCUMENTATION: The MITRE CVE dictionary describes this issue as: " "This CVE ID has been rejected or withdrawn by its CVE Numbering " "Authority for the following reason: This CVE ID has been rejected " "or withdrawn by its CVE Numbering Authority.", "link": "https://vendor.example.com/security/cve/CVE-2021-47472", "metadata": null }, "CVE_RISK_INFO": { "HOST_COUNT": 1249, "IMAGE_COUNT": 0, "PKG_COUNT": 0, "SEVERITY_LEVEL": 2, "score": 0.5154245281584533 }, "CVE_RISK_SCORE": 3.77, "END_TIME": "2024-09-04 07:00:00.000", "EVAL_CTX": { "collector_type": "Agent", "exception_props": [], "hostname": "vuln-host-1.example.net" }, "EVAL_GUID": "3dc61df780e3b722aa59b0ffcac85683", "FEATURE_KEY": { "name": "kernel-headers", "namespace": "centos:7", "package_active": 1, "package_path": "", "version_installed": "0:3.10.0-1160.119.1.el7.tuxcare.els2" }, "MACHINE_TAGS": { "Account": "999999999999", "AmiId": "ami-00000000000000000", "ExternalIp": "203.0.113.10", "Hostname": "ip-172-16-1-10.example-prod.aws.featurespace.net", "InternalIp": "10.0.0.1", "LwTokenShort": "DUMMYTOKENABCD123456", "VmProvider": "AWS", "VpcId": "vpc-00000000000000000", "os": "linux" }, "MID": 5746003737030963813, "PACKAGE_STATUS": "ACTIVE", "REGION": "eu-west-2", "RISK_SCORE": 10, "SEVERITY": "Low", "START_TIME": "2024-09-04 06:00:00.000", "STATUS": "Exception", "VULN_ID": "CVE-2021-47472" }云配置合规性（审核）
{ "ACCOUNT": { "AccountId": "999999999999", "Account_Alias": "" }, "EVAL_TYPE": "LW_SA", "ID": "lacework-global-87", "REASON": "Default security group does not restrict traffic", "RECOMMENDATION": "Ensure the default security group of every Virtual Private Cloud (VPC) restricts all traffic", "REGION": "eu-north-1", "REPORT_TIME": "2024-11-10 18:00:00.000", "RESOURCE_ID": "arn:aws:ec2:eu-west-1:999999999999:security-group/sg-00000000000000000", "SECTION": "", "SEVERITY": "High", "STATUS": "NonCompliant" }DNS 查询或解析
{ "CREATED_TIME": "2024-11-06 05:14:44.329", "DNS_SERVER_IP": "10.0.0.53", "FQDN": "data-service-prod-1234567890.s3.eu-west-2.amazonaws.com", "HOST_IP_ADDR": "172.16.1.20", "MID": 8843985456817096491, "TTL": 5 }映像漏洞评估
{ "CVE_PROPS": null, "EVAL_CTX": { "collector_type": "Agentless", "image_info": { "digest": "sha256:52d5cb782dad7a8a03c8bd1b285bbd32bdbfa8fcc435614bb1e6ceefcf26ae1d", "id": "sha256:31427c44cac7ab632d541181073bbd46a964e4ed38d087d8a47f60bb66eef4df", "registry": "999999999999.dkr.ecr.eu-west-1.amazonaws.com", "repo": "amazon/aws-network-policy-agent" } }, "EVAL_GUID": "3a17a74f0a65eed2bddd2d37bb02e6af", "FEATURE_KEY": { "name": "perl-threads", "namespace": "amzn:2", "version": "1.87-4.amzn2.0.2" }, "FIX_INFO": { "fix_available": 0, "fixed_version": "" }, "IMAGE_ID": "sha256:31427c44cac7ab632d541181073bbd46a964e4ed38d087d8a47f60bb66eef4df", "IMAGE_RISK_INFO": { "factors": [ "cve", "reachability" ], "factors_breakdown": { "cve_counts": { "Critical": 0, "High": 21, "Medium": 73 }, "internet_reachability": "Unknown" } }, "IMAGE_RISK_SCORE": 6.4, "PACKAGE_STATUS": "NO_AGENT_AVAILABLE", "RISK_SCORE": 6.4, "START_TIME": "2024-11-05 19:05:03.553", "STATUS": "GOOD" }网络流量或连接摘要
{ "DST_ENTITY_ID": { "hostname": "service-A.region.amazonaws.com", "ip_internal": 0, "port": 443, "protocol": "TCP" }, "DST_ENTITY_TYPE": "DnsSep", "DST_IN_BYTES": 0, "DST_OUT_BYTES": 0, "ENDPOINT_DETAILS": [ { "dst_ip_addr": "203.0.113.10", "dst_port": 443, "protocol": "TCP", "src_ip_addr": "192.168.1.10" }, { "dst_ip_addr": "198.51.100.5", "dst_port": 443, "protocol": "TCP", "src_ip_addr": "192.168.1.10" } ], "END_TIME": "2024-11-05 21:00:00.000", "NUM_CONNS": 4, "SRC_ENTITY_ID": { "mid": 2080882850610892909, "pid_hash": 744766973756676842 }, "SRC_ENTITY_TYPE": "Process", "SRC_IN_BYTES": 25028, "SRC_OUT_BYTES": 11962, "START_TIME": "2024-11-05 20:00:00.000" }包裹信息或更新
{ "ARCH": "x86_64", "CREATED_TIME": "2024-11-08 01:28:30.566", "MID": 4172267319977985370, "PACKAGE_NAME": "grub2", "VERSION": "2:2.02-0.87.0.2.el7.el7.centos.14.tuxcare.els2" }容器进程活动
{ "CONTAINER_ID": "4853339865add970f72213ec5d76ff51d1308c61a7680cc23c8de20c38c0a8e1", "END_TIME": "2024-11-08 02:00:00.000", "FILE_PATH": "/app/grpc-health-probe", "MID": 3708952045169222383, "PID": 177267, "POD_NAME": "kubernetes-pod-abc", "PPID": 177257, "PROCESS_START_TIME": "2024-11-08 01:43:29.960", "START_TIME": "2024-11-08 01:00:00.000", "UID": 0, "USERNAME": "serviceuser" }一般提醒或事件 (CloudTrail)
{ "EVENT_ID": "413328", "EVENT_NAME": "Unauthorized API Call", "EVENT_TYPE": "CloudTrailDefaultAlert", "SUMMARY": " For account: 999999999999 (and 22 more) : event Unauthorized API Call from a username other " "than whitelisted ones. Replaces lacework-global-29 occurred 3772 times by user " "UDM-PRINCIPAL-ID:UDM-SERVICE-ROLE (and 167 more) ", "START_TIME": "07 Feb 2025 12:00 GMT", "EVENT_CATEGORY": "Aws", "LINK": "https://security.example.net/ui/alert/12345/details", "ACCOUNT": "UDM_ACCOUNT", "SOURCE": "CloudTrail", "subject": { "srcEvent": { "event": { "errorCode": "AccessDenied", "errorMessage": "User: arn:aws:sts::999999999999:assumed-role/UDM-SERVICE-ROLE-IngestionApiRole/UDM-SERVICE-PRINCIPAL " "is not authorized to perform: kinesis:ListShards on resource: " "arn:aws:kinesis:us-east-1:999999999999:stream/ingestion-qa-rel-fraud-review-Stream " "because no identity-based policy allows the kinesis:ListShards action", "eventName": "ListShards", "eventSource": "kinesis.amazonaws.com", "eventTime": "2025-02-07T12:00:24Z", "recipientAccountId": "999999999999", "sourceIPAddress": "firehose.amazonaws.com", "userIdentity": { "accessKeyId": "ACCESSKEYIDDUMMY", "accountId": "999999999999", "arn": "arn:aws:sts::999999999999:assumed-role/UDM-SERVICE-ROLE-IngestionApiRole/UDM-SERVICE-PRINCIPAL", "sessionContext": { "sessionIssuer": { "accountId": "999999999999", "arn": "arn:aws:iam::999999999999:role/UDM-SERVICE-ROLE-IngestionApiRole", "principalId": "PRINCIPALIDDUMMY", "userName": "UDM-SERVICE-ROLE-IngestionApiRole" } } }, "vpcEndpointId": "vpce-00000000000000000" }, "principalId": "PRINCIPALIDDUMMY:UDM-SERVICE-PRINCIPAL", "recipientAccountId": "999999999999", "sourceIPAddress": "firehose.amazonaws.com", "userIdentityName": "UDM-SERVICE-ROLE-IngestionApiRole" } } }
UDM 映射表
| 日志字段 | UDM 映射 | 逻辑 |
|---|---|---|
| alertId | metadata.product_log_id | 直接复制值 |
| alertName | security_result.rule_name | 直接复制值 |
| 和程度上减少 | security_result.severity | 映射到 UDM 严重程度 |
| 状态 | security_result.summary | 直接复制值 |
| alertType | security_result.category_details | 直接复制值 |
| startTime | metadata.event_timestamp | 解析为 ISO 8601 时间戳 |
| endTime | additional.fields | 存储为 end_time 标签 |
| alertInfo.description | security_result.description | 直接复制值 |
| alertInfo.subject | metadata.description | 直接复制值 |
| entityMap.Machine.hostname | principal.hostname | 直接复制值 |
| entityMap.Machine.externalIp | principal.ip | 直接复制值 |
| entityMap.User.username | principal.user.userid | 直接复制值 |
| entityMap.Region.region | principal.location.name | 直接复制值 |
| entityMap.CT_User.accountId | principal.user.product_object_id | 直接复制值 |
event_title |
event.idm.read_only_udm.security_result.summary |
从变更日志映射 |
event_description |
event.idm.read_only_udm.security_result.description |
从变更日志映射 |
summary_details |
event.idm.read_only_udm.security_result.confidence_details |
从变更日志映射 |
target_user_id |
event.idm.read_only_udm.target.user.userid |
从变更日志映射 |
resource_product_id |
event.idm.read_only_udm.target.resource.product_object_id |
从变更日志映射 |
user_role |
event.idm.read_only_udm.target.user.role_name |
从变更日志映射 |
account_name |
event.idm.read_only_udm.target.resource.name |
从变更日志映射 |
application |
event.idm.read_only_udm.target.application |
从变更日志映射 |
event_id |
event.idm.read_only_udm.metadata.product_log_id |
从变更日志映射 |
event_type |
event.idm.read_only_udm.metadata.product_event_type |
从变更日志映射 |
event_severity |
event.idm.read_only_udm.security_result.severity_details |
从变更日志映射 |
lacework_account |
event.idm.read_only_udm.principal.user.userid |
从变更日志映射 |
event_link |
event.idm.read_only_udm.metadata.url_back_to_product |
从变更日志映射 |
starttimevalue |
event.idm.read_only_udm.additional.fields |
从变更日志映射 |
endtimevalue |
event.idm.read_only_udm.additional.fields |
从变更日志映射 |
intgGuid |
event.idm.read_only_udm.target.resource.attribute.labels |
从变更日志映射 |
rec_id |
event.idm.read_only_udm.target.resource.attribute.labels |
从变更日志映射 |
event_source |
event.idm.read_only_udm.metadata.product_name |
从变更日志映射 |
event_timestamp |
event.idm.read_only_udm.metadata.event_timestamp |
从变更日志映射 |
ACCOUNT", "EVENT_CATEGORY", "subject.srcEvent.recipientAccountAlias", "DERIVED_FIELDS.SOURCE", "subject.srcEvent.event.userIdentity.accessKeyId", "subject.srcEvent.event.userIdentity.arn", "subject.srcEvent.event.errorCode", "subject.srcEvent.event.errorMessage", "subject.srcEvent.event.eventID", "subject.srcEvent.event.eventSource", "subject.srcEvent.event.userIdentity.sessionContext.attributes.mfaAuthenticated", "subject.srcEvent.username", "subject.startTime", "subject.srcEvent.eventName", "DERIVED_FIELDS.CATEGORY", "DERIVED_FIELDS.SUBCATEGORY", "subject.dstEvent.gbm_version", "subject.dstEvent.is_visible", "subject.dstEvent.severity", "subject.dstEvent.recipientAccountAlias", "subject.srcEvent.api", "subject.srcEvent.calltype", "subject.srcEvent.gbm_version", "subject.srcEvent.is_visible", and "subject.srcEvent.severity |
additional.fields |
从变更日志映射 |
SUMMARY |
metadata.description |
从变更日志映射 |
EVENT_TYPE |
metadata.product_event_type |
从变更日志映射 |
EVENT_ID |
metadata.product_log_id |
从变更日志映射 |
LINK |
metadata.url_back_to_product |
从变更日志映射 |
subject.srcEvent.event.userAgent", "subject.srcEvent.source |
network.http.user_agent |
从变更日志映射 |
subject.srcEvent.recipientAccountId |
principal.user.groupid |
从变更日志映射 |
subject.srcEvent.principalId |
principal.user.userid |
从变更日志映射 |
subject.srcEvent.event.awsRegion |
security_result.about.asset.attribute.cloud.availability_zone |
从变更日志映射 |
subject.srcEvent.event.eventCategory |
security_result.about.asset.category |
从变更日志映射 |
EVENT_NAME |
security_result.category |
从变更日志映射 |
EVENT_NAME |
security_result.summary |
从变更日志映射 |
subject.srcType |
src.resource.resource_subtype |
从变更日志映射 |
subject.srcEvent.event.userIdentity.sessionContext.attributes.creationDate |
metadata.event_timestamp |
从变更日志映射 |
subject.srcEvent.accountcaller |
principal.resource.product_object_id |
从变更日志映射 |
subject.dstEvent.region |
target.asset.location.name |
从变更日志映射 |
subject.dstEvent.accountcaller |
target.resource.product_object_id |
从变更日志映射 |
subject.dstType |
target.resource.resource_subtype |
从变更日志映射 |
subject.dstEvent.service |
target.url |
从变更日志映射 |
subject.dstEvent.username |
target.user.userid |
从变更日志映射 |
更新日志
需要更多帮助？获得社区成员和 Google SecOps 专业人士的解答。