收集 Oracle Cloud Infrastructure - Oracle Cloud Guard 日志
本文档介绍了如何配置 Oracle Cloud Infrastructure Cloud Guard,以使用 Webhook 将日志推送到 Google Security Operations。
Oracle Cloud Guard 是一项云原生安全服务,可监控、识别并帮助维护 Oracle Cloud Infrastructure (OCI) 的强大安全态势。Cloud Guard 会检查 OCI 资源是否存在与配置相关的安全漏洞,并检测异常活动。当 Cloud Guard 发现与检测器规则的偏差时,它会创建一个问题事件,从而在整个 OCI 租户中提供全面的安全监控。此集成使用 OCI Events 服务和 OCI Functions 将 Cloud Guard 问题事件近乎实时地转发到 Google SecOps。
准备工作
确保您满足以下前提条件:
- Google SecOps 实例
- 已启用 Cloud Guard 的有效 Oracle Cloud Infrastructure 租户
- 对 Oracle Cloud Infrastructure 控制台的管理员访问权限
- 创建和管理 OCI Functions 应用的权限
- 创建 OCI Events 规则和通知主题的权限
- 访问 OCI Cloud Shell 或安装了 Fn Project CLI 的本地环境
- 已安装 Docker(用于在本地构建 OCI 函数)
- 对 Google Cloud Console 的访问权限(用于创建 API 密钥)
在 Google SecOps 中创建 Webhook Feed
创建 Feed
- 依次前往 SIEM 设置 > Feed。
- 点击添加新 Feed。
- 在下一页上,点击配置单个 Feed。
- 在 Feed 名称字段中,输入 Feed 的名称(例如
OCI Cloud Guard Problems)。 - 选择 Webhook 作为来源类型。
- 选择 Oracle Cloud Infrastructure - Oracle Cloud Guard 作为日志类型。
- 点击下一步。
- 为以下输入参数指定值:
- 拆分分隔符(可选):留空(每次函数调用发送一个事件)。
- 资产命名空间:资产命名空间
- 注入标签:要应用于此 Feed 中事件的标签
- 点击下一步。
- 在最终确定界面中查看新的 Feed 配置,然后点击提交。
生成并保存密钥
创建 Feed 后,您必须生成用于身份验证的密钥:
- 在 Feed 详情页面上,点击生成密钥。
- 系统会显示一个对话框,其中包含密钥。
- 复制并妥善保存此密钥。
重要提示:密钥只会显示一次,之后无法再检索。如果丢失,您必须生成新的密钥。
获取 Feed 端点网址
- 前往相应 Feed 的详细信息标签页。
- 在端点信息部分,复制 Feed 端点网址。
网址格式为:
https://malachiteingestion-pa.googleapis.com/v2/unstructuredlogentries:batchCreate或
https://<REGION>-malachiteingestion-pa.googleapis.com/v2/unstructuredlogentries:batchCreate保存此网址以供后续步骤使用。
点击完成。
创建 Google Cloud API 密钥
Chronicle 需要 API 密钥才能进行身份验证。在 Google Cloud Console 中创建受限 API 密钥。
创建 API 密钥
- 前往 Google Cloud 控制台的“凭据”页面。
- 选择您的项目(与您的 Chronicle 实例关联的项目)。
- 依次点击创建凭据> API 密钥。
- 系统会创建一个 API 密钥,并在对话框中显示该密钥。
- 点击修改 API 密钥以限制密钥。
限制 API 密钥
- 在 API 密钥设置页面中:
- 名称:输入一个描述性名称(例如
Chronicle Webhook API Key)
- 名称:输入一个描述性名称(例如
- 在 API 限制下:
- 选择限制密钥。
- 在选择 API 下拉菜单中,搜索并选择 Google SecOps API(或 Chronicle API)。
- 点击保存。
- 从页面顶部的 API 密钥字段复制 API 密钥值。
安全地保存 API 密钥。
启用 Cloud Guard Cloud Event 响应器
Cloud Guard 使用云事件响应程序将问题详细信息发送到 OCI Events 服务。Cloud Event responder 是 Responder recipe 的一部分,必须启用并设置为自动执行。
在 Responder 配方中启用 Cloud 事件规则
- 登录 Oracle Cloud Infrastructure 控制台。
- 打开导航菜单,然后依次选择身份和安全 > Cloud Guard > 配方。
- 选择 Responder Recipes。
- 点击附加到目标上的回答者配方。
- 在响应者规则部分中,找到 Cloud Events 规则。
- 如果状态为已停用:
- 打开操作菜单(三点状图标),然后选择修改。
- 将状态更改为已启用。
- 点击保存。
在目标设备上配置自动执行
- 在 Oracle Cloud Infrastructure 控制台中,打开导航菜单,然后依次选择身份与安全 > Cloud Guard > 配置 > 目标。
- 选择目标隔离区。
- 在资源下,选择回答者配方。
- 点击“回答者配方”链接。
- 找到 Cloud Event 响应者规则。
- 打开操作菜单(三点状图标),然后选择修改。
- 将规则触发器设置为自动执行。
- 选中确认自动执行复选框。
点击保存。
创建 OCI Functions 应用
创建 OCI Functions 应用,用于托管负责将 Cloud Guard 事件转发到 Google SecOps 的函数。
创建应用
- 在 Oracle Cloud Infrastructure 控制台中,打开导航菜单,然后依次选择开发者服务 > 函数 > 应用。
- 选择要在其中创建应用的区间。
- 点击创建应用。
- 提供以下配置详细信息:
- 名称:输入
cloudguard-chronicle-forwarder - VCN:选择具有互联网访问权限的 VCN(函数需要此权限才能访问 Chronicle webhook 端点)
- 子网:选择已配置 NAT 网关的公共子网或专用子网
- 名称:输入
- 点击创建。
创建和部署函数
使用 OCI Cloud Shell 或具有 Fn Project CLI 的本地环境来创建和部署函数。
- 从 Oracle Cloud Infrastructure 控制台打开 OCI Cloud Shell(点击顶部导航栏中的 Cloud Shell 图标)。
为 OCI Functions 应用设置 Fn 上下文:
fn list context fn use context <your-region-context> fn update context oracle.compartment-id <compartment_OCID> fn update context registry <region-key>.ocir.io/<tenancy-namespace>/cloudguard-chronicle初始化新的 Python 函数:
fn init --runtime python cloudguard-to-chronicle cd cloudguard-to-chronicle将
func.py文件的内容替换为以下代码:import io import json import logging import requests from fdk import response def handler(ctx, data: io.BytesIO = None): """ OCI Function that receives Cloud Guard events from OCI Events Service and forwards them to Google SecOps (Chronicle) webhook endpoint. """ logger = logging.getLogger() try: cfg = ctx.Config() chronicle_endpoint = cfg.get("CHRONICLE_ENDPOINT") chronicle_api_key = cfg.get("CHRONICLE_API_KEY") chronicle_secret = cfg.get("CHRONICLE_SECRET") if not all([chronicle_endpoint, chronicle_api_key, chronicle_secret]): logger.error("Missing required configuration: CHRONICLE_ENDPOINT, CHRONICLE_API_KEY, or CHRONICLE_SECRET") return response.Response( ctx, response_data=json.dumps({"error": "Missing configuration"}), headers={"Content-Type": "application/json"}, status_code=500 ) event_data = json.loads(data.getvalue()) logger.info(f"Received Cloud Guard event: {event_data.get('eventType', 'unknown')}") webhook_url = f"{chronicle_endpoint}?key={chronicle_api_key}&secret={chronicle_secret}" headers = { "Content-Type": "application/json" } resp = requests.post( webhook_url, json=event_data, headers=headers, timeout=30 ) if resp.status_code >= 200 and resp.status_code < 300: logger.info(f"Successfully forwarded event to Chronicle (HTTP {resp.status_code})") else: logger.error(f"Failed to forward event to Chronicle (HTTP {resp.status_code}): {resp.text}") return response.Response( ctx, response_data=json.dumps({"status": resp.status_code}), headers={"Content-Type": "application/json"}, status_code=200 ) except Exception as e: logger.error(f"Error processing Cloud Guard event: {str(e)}") return response.Response( ctx, response_data=json.dumps({"error": str(e)}), headers={"Content-Type": "application/json"}, status_code=500 )替换
requirements.txt文件的内容:fdk>=0.1.0 requests>=2.25.0将函数部署到应用:
fn -v deploy --app cloudguard-chronicle-forwarder等待部署完成。
配置函数环境变量
- 在 Oracle Cloud Infrastructure 控制台中,打开导航菜单,然后依次选择开发者服务 > 函数 > 应用。
- 点击 cloudguard-chronicle-forwarder 应用。
- 在资源下,选择函数。
- 点击 cloudguard-to-chronicle 函数。
- 在资源下,选择配置。
点击修改,然后添加以下键值对:
键 值 CHRONICLE_ENDPOINTChronicle 网络钩子端点网址(例如 https://malachiteingestion-pa.googleapis.com/v2/unstructuredlogentries:batchCreate)CHRONICLE_API_KEY为 Chronicle webhook 创建的 Google Cloud API 密钥 CHRONICLE_SECRETChronicle Webhook 密钥 点击保存。
为 OCI Functions 创建 IAM 政策
OCI Functions 服务和 Events 服务需要 IAM 政策才能调用函数。
为函数创建动态群组
- 在 Oracle Cloud Infrastructure 控制台中,打开导航菜单,然后依次选择身份与安全 > 网域。
- 选择默认网域。
- 在身份网域资源下,选择动态群组。
- 点击创建动态群组。
提供以下配置详细信息:
- 名称:输入
cloudguard-chronicle-functions - 说明:输入
Dynamic group for Cloud Guard to Chronicle forwarder function - 匹配规则:输入以下规则:
ALL {resource.type = 'fnfunc', resource.compartment.id = '<compartment_OCID>'}- 名称:输入
点击创建。
创建 IAM 政策
- 在 Oracle Cloud Infrastructure 控制台中,打开导航菜单,然后依次选择身份与安全 > 政策。
- 确保您位于根区间中。
- 点击创建政策。
- 提供以下配置详细信息:
- 名称:输入
cloudguard-chronicle-policy - 说明:输入
Allows Events Service to invoke Cloud Guard Chronicle forwarder function - 区间:确保已选择根区间。
- 名称:输入
- 在 Policy Builder 部分中,切换 Show manual editor。
在声明字段中,输入以下政策声明:
Allow dynamic-group cloudguard-chronicle-functions to use fn-function in compartment <compartment_name> Allow dynamic-group cloudguard-chronicle-functions to use fn-invocation in compartment <compartment_name> Allow service cloudguardevents to use fn-function in compartment <compartment_name> Allow service cloudguardevents to use fn-invocation in compartment <compartment_name>点击创建。
为 Cloud Guard 创建 OCI Events 规则
创建一项事件规则,以便在 Cloud Guard 检测到、关闭或修复问题时触发 OCI 函数。
创建“事件”规则
在 Oracle Cloud Infrastructure 控制台中,打开导航菜单,然后依次选择可观测性和管理 > 事件服务 > 规则。
点击创建规则。
提供以下配置详细信息:
- 显示名称:输入
cloudguard-to-chronicle - 说明:输入
Forwards Cloud Guard problem events to Google SecOps via OCI Functions
- 显示名称:输入
在规则条件部分中:
- 将条件设置为事件类型。
- 将服务名称设置为 Cloud Guard。
- 将活动类型设置为以下值(可选择多个):
- 检测到 - 问题
- 已关闭 - 问题
- 已解决 - 问题
在操作部分中:
- 将操作类型设置为函数。
- 将函数区间设置为包含函数应用的区间。
- 将 Function Application 设置为 cloudguard-chronicle-forwarder。
- 将函数设置为 cloudguard-to-chronicle。
点击创建规则。
可选:按风险等级过滤
如需仅转发严重程度较高的问题,请向“事件”规则添加属性过滤条件:
- 在规则条件部分,点击 + 其他条件。
- 将条件设置为属性。
- 将属性名称设置为
riskLevel。 将属性值设置为所需的风险级别:
CRITICALHIGH
验证集成
手动测试函数
- 在 Oracle Cloud Infrastructure 控制台中,打开导航菜单,然后依次选择开发者服务 > 函数 > 应用。
- 点击 cloudguard-chronicle-forwarder 应用。
- 在资源下,选择函数。
- 点击 cloudguard-to-chronicle 函数。
点击调用以使用示例数据测试函数,或使用 OCI Cloud Shell:
echo '{"eventType":"com.oraclecloud.cloudguard.problemdetected","cloudEventsVersion":"0.1","eventID":"test-event-001","data":{"compartmentId":"ocid1.compartment.oc1..example","resourceName":"test-problem","additionalDetails":{"riskLevel":"HIGH","problemDescription":"Test problem for integration verification","status":"OPEN","region":"us-ashburn-1"}}}' | fn invoke cloudguard-chronicle-forwarder cloudguard-to-chronicle依次前往开发者服务 > 函数 > 应用 > cloudguard-chronicle-forwarder,然后选择日志部分,验证函数日志。
查找确认转发成功的日志条目:
Received Cloud Guard event: com.oraclecloud.cloudguard.problemdetected Successfully forwarded event to Chronicle (HTTP 200)
在 Google SecOps 中验证
- 登录 Google SecOps。
- 前往搜索,然后查询日志类型为
OCI_CLOUDGUARD的事件。 - 验证 Cloud Guard 问题事件是否显示在搜索结果中。
如果活动未显示,请执行以下操作:
- 验证函数配置中的 Chronicle Webhook 端点网址、API 密钥和密钥
- 检查事件规则是否位于正确的 OCI 区域(Cloud Guard 报告区域)
- 确认 Cloud Event 响应程序已启用并设置为自动执行
- 查看函数日志中的错误消息
Cloud Guard 事件类型参考
Cloud Guard 通过 OCI Events 服务发出以下事件类型:
| 事件类型 | 触发器 |
|---|---|
com.oraclecloud.cloudguard.problemdetected |
检测到新问题 |
com.oraclecloud.cloudguard.problemdismissed |
问题被关闭 |
com.oraclecloud.cloudguard.problemremediated |
问题得到解决 |
com.oraclecloud.cloudguard.sightingdetected |
检测到新的目击事件 |
身份验证方法参考
Chronicle webhook Feed 支持多种身份验证方法。选择供应商支持的方法。
方法 1:自定义标头(推荐)
如果您的供应商支持自定义 HTTP 标头,请使用此方法以提高安全性。
请求格式:
POST <ENDPOINT_URL> HTTP/1.1 Content-Type: application/json x-goog-chronicle-auth: <API_KEY> x-chronicle-auth: <SECRET_KEY> { "event": "data", "timestamp": "2025-01-15T10:30:00Z" }
优点:
- API 密钥和密文在网址中不可见
- 更安全(标头不会记录在 Web 服务器访问日志中)
- 供应商支持时的首选方法
方法 2:查询参数
如果您的供应商不支持自定义标头,请将凭据附加到网址。
网址格式:
<ENDPOINT_URL>?key=<API_KEY>&secret=<SECRET_KEY>示例:
https://malachiteingestion-pa.googleapis.com/v2/unstructuredlogentries:batchCreate?key=AIzaSyD...&secret=abcd1234...请求格式:
POST <ENDPOINT_URL>?key=<API_KEY>&secret=<SECRET_KEY> HTTP/1.1 Content-Type: application/json { "event": "data", "timestamp": "2025-01-15T10:30:00Z" }
缺点:
- 网址中显示的凭据
- 可能会记录在 Web 服务器访问日志中
- 安全性不如标头
方法 3:混合(网址 + 标头)
某些配置在网址中使用 API 密钥,在标头中使用密钥。
请求格式:
POST <ENDPOINT_URL>?key=<API_KEY> HTTP/1.1 Content-Type: application/json x-chronicle-auth: <SECRET_KEY> { "event": "data", "timestamp": "2025-01-15T10:30:00Z" }
身份验证标头名称
Chronicle 接受以下身份验证标头名称:
对于 API 密钥:
x-goog-chronicle-auth(推荐)X-Goog-Chronicle-Auth(不区分大小写)
对于密钥:
x-chronicle-auth(推荐)X-Chronicle-Auth(不区分大小写)
Webhook 限制和最佳实践
请求限制
| 限制 | 值 |
|---|---|
| 最大请求大小 | 4 MB |
| 最大 QPS(每秒查询次数) | 15000 |
| 请求超时 | 30 秒 |
| 重试行为 | 自动(使用指数退避算法) |
UDM 映射表
| 日志字段 | UDM 映射 | 逻辑 |
|---|---|---|
| contentType | additional.fields | 从各种其他字段合并而来 |
| cloudEventsVersion | additional.fields | |
| data.additionalDetails.reason | additional.fields | |
| data.additionalDetails.tenantId | additional.fields | |
| data.additionalDetails.problemType | additional.fields | |
| data.resourceName | additional.fields | |
| data.resourceId | additional.fields | |
| 状态 | additional.fields | |
| problem_recommendation | additional.fields | |
| compartment_id | additional.fields | |
| compartment_name | additional.fields | |
| product_event | extensions.auth.type | 如果 product_event 与登录匹配,则设置为“AUTHTYPE_UNSPECIFIED” |
| data.additionalDetails.problemDescription | metadata.description | 如果 data.additionalDetails.problemDescription 不为空,则为该值;否则为 data.additionalDetails.description |
| data.additionalDetails.description | metadata.description | |
| has_user | metadata.event_type | 如果具有 has_user 和 has_target_resource,则设置为 USER_RESOURCE_ACCESS;如果具有 has_user,则设置为 USER_UNCATEGORIZED;如果 product_event 与登录匹配,则设置为 USER_LOGIN;如果具有 has_principal,则设置为 STATUS_UPDATE;否则设置为 GENERIC_EVENT |
| has_target_resource | metadata.event_type | |
| product_event | metadata.event_type | |
| has_principal | metadata.event_type | |
| eventType | metadata.product_event_type | 如果 eventType 不为空,则为 eventType 中的值;否则为 product_event |
| product_event | metadata.product_event_type | |
| eventID | metadata.product_log_id | 直接复制值 |
| 版本 | metadata.product_version | 直接复制值 |
| 观察者 | observer.domain.name | 使用 grok 从观测器中提取的网域 |
| 观察者 | observer.hostname | 如果不是 IP,则来自观测者的值 |
| 观察者 | observer.ip | 如果 IP |
| 主账号 | principal.administrative_domain | 使用 grok 从正文或 principal_host 中提取的网域 |
| principal_host | principal.administrative_domain | |
| data.additionalDetails.problemAdditionalDetails.public_ips | principal.asset.ip | 如果 data.additionalDetails.problemAdditionalDetails.public_ips 不为空,则取该值;否则,如果 vnicDetails.0.vnicPublicIp 不为空,则取该值;否则,如果 principal 为 IP,则取 principal;否则,如果 principal_host 为 IP,则取 principal_host |
| vnicDetails.0.vnicPublicIp | principal.asset.ip | |
| 主账号 | principal.asset.ip | |
| principal_host | principal.asset.ip | |
| 主账号 | principal.hostname | 如果不是 IP,则为来自主账号的值;如果不是 IP,则为 principal_host |
| principal_host | principal.hostname | |
| data.additionalDetails.problemAdditionalDetails.public_ips | principal.ip | 如果 data.additionalDetails.problemAdditionalDetails.public_ips 不为空,则取该值;否则,如果 vnicDetails.0.vnicPublicIp 不为空,则取该值;否则,如果 principal 为 IP,则取 principal;否则,如果 principal_host 为 IP,则取 principal_host |
| vnicDetails.0.vnicPublicIp | principal.ip | |
| 主账号 | principal.ip | |
| principal_host | principal.ip | |
| data.additionalDetails.region | principal.location.name | 直接复制值 |
| principal_port | principal.port | 从 principal_port 转换为整数 |
| data.additionalDetails.resourceName | principal.resource.name | 如果 data.additionalDetails.resourceName 不为空,则为该值;否则为 data.additionalDetails.principalResourceName |
| data.additionalDetails.principalResourceName | principal.resource.name | |
| data.additionalDetails.resourceId | principal.resource.product_object_id | 如果 data.additionalDetails.resourceId 不为空,则返回该值;否则,返回 data.additionalDetails.principalResourceId |
| data.additionalDetails.principalResourceId | principal.resource.product_object_id | |
| data.additionalDetails.resourceType | principal.resource.resource_subtype | 如果 data.additionalDetails.resourceType 不为空,则为该值;否则为 data.additionalDetails.principalResourceType |
| data.additionalDetails.principalResourceType | principal.resource.resource_subtype | |
| data.additionalDetails.resourceName | principal.user.userid | 如果 data.additionalDetails.resourceType == "User",则为 data.additionalDetails.resourceName 中的值;否则,如果 data.additionalDetails.principalResourceType == "User",则为 data.additionalDetails.principalResourceName 中的值 |
| data.additionalDetails.principalResourceName | principal.user.userid | |
| 操作 | security_result.action | 如果操作与屏蔽匹配,则设置为 BLOCK;如果成功,则设置为 ALLOW;如果失败,则设置为 FAIL;否则设置为 UNKNOWN |
| 操作 | security_result.action_details | 直接复制值 |
| data.additionalDetails.tacticName | security_result.attack_details.tactics | 直接复制到 tactic_data.name,然后合并 |
| data.additionalDetails.techniqueName | security_result.attack_details.techniques | 值直接复制到 technique_data.name,然后合并 |
| data.additionalDetails.confidence | security_result.confidence_score | 直接复制值 |
| sightingType | security_result.detection_fields | 从 sightingType、sightingScore、riskScore 标签合并而来 |
| sightingScore | security_result.detection_fields | |
| riskScore | security_result.detection_fields | |
| data.additionalDetails.detectorRuleId | security_result.rule_id | 直接复制值 |
| data.additionalDetails.detectorRuleType | security_result.rule_name | 直接复制值 |
| 和程度上减少 | security_result.severity | 如果严重程度为 INFO,则设置为 INFORMATIONAL;如果严重程度为 Low,则设置为 LOW;如果严重程度为 WARN,则设置为 MEDIUM;如果严重程度为 High,则设置为 HIGH;如果严重程度为 Critical,则设置为 CRITICAL |
| 和程度上减少 | security_result.severity_details | 直接复制值 |
| 目标 | target.administrative_domain | 使用 grok 从目标或 target_host 中提取的网域 |
| target_host | target.administrative_domain | |
| 目标 | target.hostname | 如果不是 IP,则为目标中的值;如果不是 IP,则为 target_host |
| target_host | target.hostname | |
| 目标 | target.ip | 如果为 IP,则为目标中的值;如果为 IP,则为 target_host |
| target_host | target.ip | |
| target_port | target.port | 从 target_port 转换为整数 |
| targetId | target.resource.attribute.labels | 从各种标签字段合并而来 |
| 标签 | target.resource.attribute.labels | |
| vnicAttachmentId | target.resource.attribute.labels | |
| vnicAttachmentDisplayName | target.resource.attribute.labels | |
| subnets | target.resource.attribute.labels | |
| instance_ocid | target.resource.attribute.labels | |
| route_table_ocids | target.resource.attribute.labels | |
| security_list_ocids | target.resource.attribute.labels | |
| data.additionalDetails.impactedResourceName | target.resource.name | 如果 data.additionalDetails.impactedResourceName 不为空,则为该值;否则为 vnicDetails.0.vnicDisplayName |
| vnicDetails.0.vnicDisplayName | target.resource.name | |
| data.additionalDetails.impactedResourceId | target.resource.product_object_id | 如果 data.additionalDetails.impactedResourceId 不为空,则为该值;否则为 vnicDetails.0.vnicId |
| vnicDetails.0.vnicId | target.resource.product_object_id | |
| data.additionalDetails.impactedResourceType | target.resource.resource_subtype | 直接复制值 |
| target_url | target.url | 直接复制值 |
| target_user | target.user.userid | 直接复制值 |
| 产品 | metadata.product_name | 直接复制值 |
| vendor | metadata.vendor_name | 直接复制值 |
data.additionalDetails.tacticName |
security_result.attack_details.tactics.name |
从变更日志映射 |
data.additionalDetails.techniqueName |
security_result.attack_details.techniques.name |
从变更日志映射 |