收集 Menlo Security Isolation Platform (MSIP) 日志

支持的平台：

Google SecOps SIEM

本指南介绍了如何使用 Google Cloud Storage 将 Menlo Security Isolation Platform (MSIP) 日志注入到 Google SecOps。

Menlo Security 以隔离为中心的方法将网页浏览和文档检索操作分别在用户设备和远离端点的隔离型一次性虚拟容器 (DVC) 中进行。所有风险代码都在隔离的 DVC 中执行，绝不会到达端点。系统只会将安全的展示数据发送到用户的浏览器。该平台可全面记录 Web 访问、审核事件、电子邮件安全、DLP 违规和 SMTP 活动。

准备工作

确保您满足以下前提条件：

• Google SecOps 实例
• 对 Menlo Security Security Admin Portal的特权访问权限
• 在 Menlo Security 中创建 API 令牌的权限
• 启用了 Cloud Storage、Cloud Run、Pub/Sub 和 Cloud Scheduler API 的 GCP 项目

配置 Menlo Security API 访问权限

如需让 Cloud Run 函数能够检索日志，您需要创建具有 Log Export API 权限的 API 令牌。

创建 API 令牌

1. 登录 https://admin.menlosecurity.com 处的 Menlo Security 管理员门户。
2. 依次前往设置> 身份验证 > API 令牌。
3. 点击 Create Token。
4. 在 Token Name（令牌名称）字段中，输入一个描述性名称（例如 Chronicle GCS Integration）。
5. 在权限部分中，选择 Log Export API。
6. 点击创建或保存。
7. 复制并安全地存储生成的 API 令牌。

重要提示：API 令牌仅显示一次。创建后立即将其存储在安全的位置。

所需 API 权限

• API 令牌需要以下权限：

  权限/范围	访问权限级别	用途
  日志导出 API	读取	从 Menlo Security API 检索日志

  注意： 日志提取只需要 Log Export API 权限。无需其他权限。

创建 Google Cloud Storage 存储桶

1. 前往 Google Cloud 控制台。
2. 选择您的项目或创建新项目。
3. 在导航菜单中，依次前往 Cloud Storage > 存储分区。
4. 点击创建存储分区。

5. 提供以下配置详细信息：

   设置	值
   为存储桶命名	输入一个全局唯一的名称（例如 menlo-security-logs）
   位置类型	根据您的需求进行选择（区域级、双区域、多区域）
   位置	选择营业地点（例如 us-central1）
   存储类别	标准（建议用于经常访问的日志）
   访问权限控制	均匀（推荐）
   保护工具	可选：启用对象版本控制或保留政策

6. 点击创建。

为 Cloud Run 函数创建服务账号

创建服务账号

1. 在 GCP 控制台中，依次前往 IAM 和管理 > 服务账号。
2. 点击创建服务账号。
3. 提供以下配置详细信息：
   • 服务账号名称：输入 menlo-logs-collector-sa
   • 服务账号说明：输入 Service account for Cloud Run function to collect Menlo Security logs
4. 点击创建并继续。
5. 在向此服务账号授予对项目的访问权限部分中，添加以下角色：
   1. 点击选择角色。
   2. 搜索并选择 Storage Object Admin。
   3. 点击 + 添加其他角色。
   4. 搜索并选择 Cloud Run Invoker。
   5. 点击 + 添加其他角色。
   6. 搜索并选择 Cloud Functions Invoker。
6. 点击继续。
7. 点击完成。

授予对 GCS 存储桶的 IAM 权限

1. 前往 Cloud Storage > 存储分区。
2. 点击您的存储桶名称 (menlo-security-logs)。
3. 前往权限标签页。
4. 点击授予访问权限。
5. 提供以下配置详细信息：
   • 添加主账号：输入服务账号电子邮件地址 (menlo-logs-collector-sa@PROJECT_ID.iam.gserviceaccount.com)
   • 分配角色：选择 Storage Object Admin
6. 点击保存。

创建 Pub/Sub 主题

1. 在 GCP 控制台中，前往 Pub/Sub > 主题。
2. 点击创建主题。
3. 提供以下配置详细信息：
   • 主题 ID：输入 menlo-logs-trigger
   • 将其他设置保留为默认值
4. 点击创建。

创建 Cloud Run 函数以收集日志

1. 在 GCP 控制台中，前往 Cloud Run。
2. 点击创建服务。
3. 选择函数（使用内嵌编辑器创建函数）。

4. 在配置部分中，提供以下配置详细信息：

   设置	值
   Service 名称	menlo-logs-collector
   区域	选择与您的 GCS 存储桶匹配的区域（例如 us-central1）
   运行时	选择 Python 3.12 或更高版本

5. 在触发器（可选）部分中：

   1. 点击 + 添加触发器。
   2. 选择 Cloud Pub/Sub。
   3. 在选择 Cloud Pub/Sub 主题中，选择 menlo-logs-trigger。
   4. 点击保存。

6. 在身份验证部分中：

   1. 选择需要进行身份验证。
   2. 检查 Identity and Access Management (IAM)。
   注意： 调用函数时，Pub/Sub 会自动处理身份验证。

7. 向下滚动并展开容器、网络、安全性。

8. 前往安全性标签页：

   • 服务账号：选择 menlo-logs-collector-sa

9. 前往容器标签页：

   1. 点击变量和密钥。
   2. 为每个环境变量点击+ 添加变量：

   变量名称	示例值	说明
   GCS_BUCKET	menlo-security-logs	GCS 存储桶名称
   GCS_PREFIX	menlo-logs	日志文件的前缀
   STATE_KEY	menlo-logs/state.json	状态文件路径
   MENLO_API_TOKEN	your-menlo-api-token	Menlo Security API 令牌
   MENLO_API_HOST	logs.menlosecurity.com	Menlo Security API 主机
   MENLO_LOG_TYPES	web,audit,email,dlp	以逗号分隔的日志类型
   LOOKBACK_HOURS	1	初始回溯期

   注意： MENLO_LOG_TYPES 变量接受要收集的日志类型的逗号分隔列表。支持的值包括：web、audit、email、dlp、attachment、smtp。

10. 在变量和 Secret 部分中，向下滚动到请求：

    • 请求超时：输入 600 秒（10 分钟）

11. 前往设置标签页：

    • 在资源部分中：
      • 内存：选择 512 MiB 或更高值
      • CPU：选择 1

12. 在修订版本伸缩部分中：

    • 实例数下限：输入 0
    • 实例数上限：输入 100

13. 点击创建。

14. 等待服务创建完成（1-2 分钟）。

15. 创建服务后，系统会自动打开内嵌代码编辑器。

添加函数代码

1. 在入口点字段中输入 main。

2. 在内嵌代码编辑器中，创建两个文件：

   • main.py:

   import functions_framework
   from google.cloud import storage
   import os
   import json
   import urllib3
   import time
   import uuid
   from datetime import datetime, timezone, timedelta
   
   http = urllib3.PoolManager(
     timeout=urllib3.Timeout(connect=5.0, read=60.0),
     retries=False,
   )
   
   storage_client = storage.Client()
   
   GCS_BUCKET = os.environ.get('GCS_BUCKET')
   GCS_PREFIX = os.environ.get('GCS_PREFIX', 'menlo-logs')
   STATE_KEY = os.environ.get('STATE_KEY', 'menlo-logs/state.json')
   API_TOKEN = os.environ.get('MENLO_API_TOKEN')
   API_HOST = os.environ.get('MENLO_API_HOST', 'logs.menlosecurity.com')
   LOG_TYPES = os.environ.get('MENLO_LOG_TYPES', 'web').split(',')
   LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '1'))
   
   @functions_framework.cloud_event
   def main(cloud_event):
     if not all([GCS_BUCKET, API_TOKEN, API_HOST]):
       print('Error: Missing required environment variables')
       return
   
     try:
       bucket = storage_client.bucket(GCS_BUCKET)
       state = load_state(bucket)
       now = datetime.now(timezone.utc).replace(microsecond=0)
   
       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 logs from {last_time.isoformat()} to {now.isoformat()}")
   
       total = 0
       newest_time = None
   
       for log_type in LOG_TYPES:
         log_type = log_type.strip()
         print(f"Fetching {log_type} logs...")
         events = fetch_logs(log_type, API_TOKEN, API_HOST, last_time, now)
   
         if events:
           object_key = f"{GCS_PREFIX}/{now:%Y/%m/%d}/menlo-{log_type}-{uuid.uuid4()}.ndjson"
           blob = bucket.blob(object_key)
   
           ndjson = '\n'.join(
             [json.dumps(ev, ensure_ascii=False) for ev in events]
           ) + '\n'
           blob.upload_from_string(ndjson, content_type='application/x-ndjson')
   
           print(f"Uploaded {len(events)} {log_type} events to gs://{GCS_BUCKET}/{object_key}")
           total += len(events)
   
           for ev in events:
             ev_time = ev.get('timestamp') or ev.get('time') or ev.get('created_at')
             if ev_time and (newest_time is None or str(ev_time) > str(newest_time)):
               newest_time = str(ev_time)
         else:
           print(f"No {log_type} events found")
   
       save_state(bucket, newest_time if newest_time else now.isoformat())
       print(f"Successfully processed {total} total events")
   
     except Exception as e:
       print(f'Error processing logs: {str(e)}')
       raise
   
   def fetch_logs(log_type, token, api_host, start, end):
     url = f"https://{api_host}/api/v1/logs/{log_type}"
     headers = {
       'Authorization': f'Token {token}',
       'Accept': 'application/json',
     }
     params = {
       'start': start.strftime('%Y-%m-%dT%H:%M:%SZ'),
       'end': end.strftime('%Y-%m-%dT%H:%M:%SZ'),
     }
     query = '&'.join(f'{k}={v}' for k, v in params.items())
     full_url = f'{url}?{query}'
   
     all_events = []
     backoff = 1.0
   
     while full_url:
       response = http.request('GET', full_url, headers=headers)
   
       if response.status == 429:
         retry_after = int(response.headers.get('Retry-After', str(int(backoff))))
         print(f"Rate limited. Waiting {retry_after}s...")
         time.sleep(retry_after)
         backoff = min(backoff * 2, 60.0)
         continue
   
       if response.status != 200:
         print(f"API error {response.status}: {response.data.decode('utf-8')}")
         break
   
       backoff = 1.0
       data = json.loads(response.data.decode('utf-8'))
       events = data if isinstance(data, list) else data.get('events', data.get('data', []))
       all_events.extend(events)
   
       full_url = data.get('next') if isinstance(data, dict) else None
   
     print(f"Total {log_type} events fetched: {len(all_events)}")
     return all_events
   
   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
   

3. 点击部署以保存并部署该函数。

4. 等待部署完成（2-3 分钟）。

   注意： Pub/Sub 触发器配置会自动创建必要的订阅和权限。

创建 Cloud Scheduler 作业

1. 在 GCP Console 中，前往 Cloud Scheduler。
2. 点击创建作业。

3. 提供以下配置详细信息：

   设置	值
   名称	menlo-logs-collector-hourly
   区域	选择与 Cloud Run 函数相同的区域
   频率	0 * * * *（每小时一次，整点时）
   时区	选择时区（建议选择世界协调时间 [UTC]）
   目标类型	Pub/Sub
   主题	选择menlo-logs-trigger
   消息正文	{}（空 JSON 对象）

4. 点击创建。

测试集成

1. 在 Cloud Scheduler 控制台中，找到您的作业 (menlo-logs-collector-hourly)。
2. 点击强制运行以手动触发作业。
3. 等待几秒钟。
4. 前往 Cloud Run > 服务。
5. 点击 menlo-logs-collector。
6. 点击日志标签页。

7. 验证函数是否已成功执行。查找：

   Fetching logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00
   Fetching web logs...
   Total web events fetched: X
   Uploaded X web events to gs://menlo-security-logs/menlo-logs/YYYY/MM/DD/menlo-web-UUID.ndjson
   Successfully processed X total events
   

8. 前往 Cloud Storage > 存储分区。

9. 点击 menlo-security-logs。

10. 转到 menlo-logs/ 文件夹。

11. 验证是否已创建包含当前日期的新 .ndjson 文件。

如果您在日志中看到错误，请执行以下操作：

• HTTP 401：验证 MENLO_API_TOKEN 环境变量是否正确
• HTTP 403：验证 API 令牌是否具有日志导出 API 权限
• HTTP 429：速率限制 - 函数将自动重试并进行退避
• 缺少环境变量：验证是否已在 Cloud Run 函数配置中设置所有必需的变量

检索 Google SecOps 服务账号

1. 依次前往 SIEM 设置 > Feed。
2. 点击添加新 Feed。
3. 点击配置单个 Feed。
4. 在 Feed 名称字段中，输入 Feed 的名称（例如 Menlo Security MSIP Logs）。
5. 选择 Google Cloud Storage V2 作为来源类型。
6. 选择 Menlo Security 作为日志类型。

7. 点击获取服务账号。系统会显示一个唯一的服务账号电子邮件地址，例如：

   [e-mail]
   

8. 复制此电子邮件地址，以便在下一步中使用。

   注意： 每个 Google SecOps 实例都有一个唯一的服务账号。请勿使用其他文档或示例中的服务账号。

9. 点击下一步。

10. 为以下输入参数指定值：

    • 存储桶网址：输入带有前缀路径的 GCS 存储桶 URI：

      gs://menlo-security-logs/menlo-logs/
      

    注意： 请务必在 URI 末尾添加尾随斜杠 (/)。

    • 来源删除选项：根据您的偏好选择删除选项：

      • 永不：转移后永不删除任何文件（建议用于测试）。
      • 删除已转移的文件：在成功转移后删除文件。

      • 删除已转移的文件和空目录：成功转移后删除文件和空目录。

        注意 ：如果您选择删除选项，服务账号必须具有 Storage Object Admin 角色，而不是 Storage Object Viewer 角色。相应地更新 IAM 权限。

      • 文件存在时间上限：包含在过去指定天数内修改的文件（默认值为 180 天）

      • 资产命名空间：资产命名空间

      • 注入标签：要应用于此 Feed 中事件的标签

11. 点击下一步。

12. 在最终确定界面中查看新的 Feed 配置，然后点击提交。

向 Google SecOps 服务账号授予 IAM 权限

1. 前往 Cloud Storage > 存储分区。
2. 点击 menlo-security-logs。
3. 前往权限标签页。
4. 点击授予访问权限。
5. 提供以下配置详细信息：
   • 添加主账号：粘贴 Google SecOps 服务账号电子邮件地址
   • 分配角色：选择 Storage Object Viewer
6. 点击保存。

注意： 如果您打算使用“删除已转移的文件”或“删除已转移的文件和空目录”删除选项，请授予存储对象管理员角色，而不是存储对象查看者角色。

UDM 映射表

日志字段	UDM 映射	逻辑
customer_name	intermediary.hostname	设置 customer_name 是否为 IP
customer_name	intermediary.ip	如果为 IP，则从 customer_name 中提取；否则从 xff_ip IP 中提取
xff_ip	intermediary.ip
event_data.event_time	metadata.event_timestamp	使用 grok 和日期过滤条件进行解析
request_type	metadata.event_type	基于这些字段的条件逻辑
has_principal	metadata.event_type
has_top_url	metadata.event_type
has_dest	metadata.event_type
log_type	metadata.log_type	直接复制值
event_data.name	metadata.product_event_type	直接复制值
reqId	metadata.product_log_id	直接复制值
event_data.version	metadata.product_version	直接复制值
event_data.protocol	network.application_protocol	使用预定义的协议列表进行映射
request_type	network.http.method	直接复制值
引荐来源网址	network.http.referral_url	直接复制值
response_code	network.http.response_code	转换为整数
event_data.user-agent	network.http.user_agent	直接复制值
full_session_id	network.session_id	直接复制值
browser_and_version	principal.application	直接复制值
event_data.x-client-ip	principal.asset.ip	如果 event_data.x-client-ip 不为空，则为该值；否则为 src_ip
src_ip	principal.asset.ip
groups	principal.group.product_object_id	直接复制值
event_data.x-client-ip	principal.ip	如果 event_data.x-client-ip 不为空，则为该值；否则为 src_ip
src_ip	principal.ip
区域	principal.location.country_or_region	如果区域值不为空，则为区域值；否则为 egress_country
egress_country	principal.location.country_or_region
region_name	principal.location.name	直接复制值
egress_ip	principal.nat_ip	直接复制值
event_data.userid	principal.user.email_addresses	如果匹配电子邮件正则表达式，则合并
event_data.userid	principal.user.userid	直接复制值
event_data.domain	security_result.about.administrative_domain	直接复制值
event_data.url	security_result.about.url	直接复制值
event_data.pe_action	security_result.action	映射到 ALLOW、BLOCK 或 ALLOW_WITH_MODIFICATION
event_data.categories	security_result.category_details	从 event_data.categories（提取或直接）和 casb_cat_name 合并
casb_cat_name	security_result.category_details
event_data.risk_score	security_result.confidence_details	直接复制值
content-type	security_result.detection_fields	从各种标签字段合并而来
产品	security_result.detection_fields
威胁	security_result.detection_fields
virusDetails	security_result.detection_fields
sandboxResult	security_result.detection_fields
fullScanResult	security_result.detection_fields
tab_id	security_result.detection_fields
pe_reason	security_result.detection_fields
ua_type	security_result.detection_fields
email_isolation_state	security_result.detection_fields
document_url	security_result.detection_fields
archive_path	security_result.detection_fields
casb_app_name	security_result.detection_fields
casb_fun_name	security_result.detection_fields
casb_org_name	security_result.detection_fields
casb_profile_id	security_result.detection_fields
casb_profile_name	security_result.detection_fields
casb_profile_type	security_result.detection_fields
casb_risk_score	security_result.detection_fields
connId	security_result.detection_fields
origin_country	security_result.detection_fields
egress_country	security_result.detection_fields
is_casb_ddl	security_result.detection_fields
is_iframe	security_result.detection_fields
parent_file_id	security_result.detection_fields
parent_filename	security_result.detection_fields
parent_sha256	security_result.detection_fields
parent_tid	security_result.detection_fields
root_file_id	security_result.detection_fields
root_filename	security_result.detection_fields
root_sha256	security_result.detection_fields
casb_risk_score	security_result.risk_score	转换为浮点数
casb_profile_id	security_result.rule_id	直接复制值
pe_rulename	security_result.rule_name	直接复制值
casb_profile_name	security_result.rule_set	直接复制值
casb_profile_type	security_result.rule_type	直接复制值
event_data.severity	security_result.severity_details	直接复制值
magicName	security_result.summary	串联为“%{magicName} is %{threats}”。
威胁	security_result.summary
threat_types	security_result.threat_name	直接复制值
origin_ip	src.ip	直接复制值
origin_country	src.location.country_or_region	直接复制值
event_data.dst	target.asset.ip	使用 grok IP 模式从 event_data.dst 或 dst 中提取
dst	target.asset.ip
filename	target.file.full_path	直接复制值
mimeType	target.file.mime_type	直接复制值
sha256	target.file.sha256	直接复制值
file_size	target.file.size	转换为浮点数，然后转换为无符号整数
event_data.dst	target.ip	使用 grok IP 模式从 event_data.dst 或 dst 中提取
dst	target.ip
top_url	target.url	使用 URI grok 模式提取
metadata.product_name	metadata.product_name	设置为“MENLO SECURITY”
metadata.vendor_name	metadata.vendor_name	设置为“MENLO SECURITY”
region	event.idm.read_only_udm.principal.location.country_or_region	从变更日志映射
region_name	event.idm.read_only_udm.principal.location.name	从变更日志映射
src_ip	event.idm.read_only_udm.principal.ip和event.idm.read_only_udm.principal.asset.ip之间	从变更日志映射
groups	event.idm.read_only_udm.principal.group.product_object_id	从变更日志映射
content-type	event.idm.read_only_udm.security_result.detection_fields	从变更日志映射
product	event.idm.read_only_udm.security_result.detection_fields	从变更日志映射
mimeType	event.idm.read_only_udm.target.file.mime_type	从变更日志映射

更新日志

查看相应解析器的更改日志

需要更多帮助？获得社区成员和 Google SecOps 专业人士的解答。