收集 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

  1. 依次前往 SIEM 设置 > Feed
  2. 点击添加新 Feed
  3. 在下一页上,点击配置单个 Feed
  4. Feed 名称字段中,输入 Feed 的名称(例如 OCI Cloud Guard Problems)。
  5. 选择 Webhook 作为来源类型
  6. 选择 Oracle Cloud Infrastructure - Oracle Cloud Guard 作为日志类型
  7. 点击下一步
  8. 为以下输入参数指定值:
    • 拆分分隔符(可选):留空(每次函数调用发送一个事件)。
    • 资产命名空间资产命名空间
    • 注入标签:要应用于此 Feed 中事件的标签
  9. 点击下一步
  10. 最终确定界面中查看新的 Feed 配置,然后点击提交

生成并保存密钥

创建 Feed 后,您必须生成用于身份验证的密钥:

  1. 在 Feed 详情页面上,点击生成密钥
  2. 系统会显示一个对话框,其中包含密钥。
  3. 复制并妥善保存此密钥。

重要提示:密钥只会显示一次,之后无法再检索。如果丢失,您必须生成新的密钥。

获取 Feed 端点网址

  1. 前往相应 Feed 的详细信息标签页。
  2. 端点信息部分,复制 Feed 端点网址
  3. 网址格式为:

    https://malachiteingestion-pa.googleapis.com/v2/unstructuredlogentries:batchCreate
    

    https://<REGION>-malachiteingestion-pa.googleapis.com/v2/unstructuredlogentries:batchCreate
    
  4. 保存此网址以供后续步骤使用。

  5. 点击完成

创建 Google Cloud API 密钥

Chronicle 需要 API 密钥才能进行身份验证。在 Google Cloud Console 中创建受限 API 密钥。

创建 API 密钥

  1. 前往 Google Cloud 控制台的“凭据”页面
  2. 选择您的项目(与您的 Chronicle 实例关联的项目)。
  3. 依次点击创建凭据> API 密钥
  4. 系统会创建一个 API 密钥,并在对话框中显示该密钥。
  5. 点击修改 API 密钥以限制密钥。

限制 API 密钥

  1. API 密钥设置页面中:
    • 名称:输入一个描述性名称(例如 Chronicle Webhook API Key
  2. API 限制下:
    1. 选择限制密钥
    2. 选择 API 下拉菜单中,搜索并选择 Google SecOps API(或 Chronicle API)。
  3. 点击保存
  4. 从页面顶部的 API 密钥字段复制 API 密钥值。
  5. 安全地保存 API 密钥。

启用 Cloud Guard Cloud Event 响应器

Cloud Guard 使用云事件响应程序将问题详细信息发送到 OCI Events 服务。Cloud Event responder 是 Responder recipe 的一部分,必须启用并设置为自动执行。

在 Responder 配方中启用 Cloud 事件规则

  1. 登录 Oracle Cloud Infrastructure 控制台
  2. 打开导航菜单,然后依次选择身份和安全 > Cloud Guard > 配方
  3. 选择 Responder Recipes
  4. 点击附加到目标上的回答者配方。
  5. 响应者规则部分中,找到 Cloud Events 规则。
  6. 如果状态为已停用
    1. 打开操作菜单(三点状图标),然后选择修改
    2. 状态更改为已启用
    3. 点击保存

在目标设备上配置自动执行

  1. Oracle Cloud Infrastructure 控制台中,打开导航菜单,然后依次选择身份与安全 > Cloud Guard > 配置 > 目标
  2. 选择目标隔离区。
  3. 资源下,选择回答者配方
  4. 点击“回答者配方”链接。
  5. 找到 Cloud Event 响应者规则。
  6. 打开操作菜单(三点状图标),然后选择修改
  7. 规则触发器设置为自动执行
  8. 选中确认自动执行复选框。
  9. 点击保存

创建 OCI Functions 应用

创建 OCI Functions 应用,用于托管负责将 Cloud Guard 事件转发到 Google SecOps 的函数。

创建应用

  1. Oracle Cloud Infrastructure 控制台中,打开导航菜单,然后依次选择开发者服务 > 函数 > 应用
  2. 选择要在其中创建应用的区间。
  3. 点击创建应用
  4. 提供以下配置详细信息:
    • 名称:输入 cloudguard-chronicle-forwarder
    • VCN:选择具有互联网访问权限的 VCN(函数需要此权限才能访问 Chronicle webhook 端点)
    • 子网:选择已配置 NAT 网关的公共子网或专用子网
  5. 点击创建

创建和部署函数

使用 OCI Cloud Shell 或具有 Fn Project CLI 的本地环境来创建和部署函数。

  1. 从 Oracle Cloud Infrastructure 控制台打开 OCI Cloud Shell(点击顶部导航栏中的 Cloud Shell 图标)。
  2. 为 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
    
  3. 初始化新的 Python 函数:

    fn init --runtime python cloudguard-to-chronicle
    cd cloudguard-to-chronicle
    
  4. 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
            )
    
  5. 替换 requirements.txt 文件的内容:

    fdk>=0.1.0
    requests>=2.25.0
    
  6. 将函数部署到应用:

    fn -v deploy --app cloudguard-chronicle-forwarder
    
  7. 等待部署完成。

配置函数环境变量

  1. Oracle Cloud Infrastructure 控制台中,打开导航菜单,然后依次选择开发者服务 > 函数 > 应用
  2. 点击 cloudguard-chronicle-forwarder 应用。
  3. 资源下,选择函数
  4. 点击 cloudguard-to-chronicle 函数。
  5. 资源下,选择配置
  6. 点击修改,然后添加以下键值对:

    CHRONICLE_ENDPOINT Chronicle 网络钩子端点网址(例如 https://malachiteingestion-pa.googleapis.com/v2/unstructuredlogentries:batchCreate
    CHRONICLE_API_KEY 为 Chronicle webhook 创建的 Google Cloud API 密钥
    CHRONICLE_SECRET Chronicle Webhook 密钥
  7. 点击保存

为 OCI Functions 创建 IAM 政策

OCI Functions 服务和 Events 服务需要 IAM 政策才能调用函数。

为函数创建动态群组

  1. Oracle Cloud Infrastructure 控制台中,打开导航菜单,然后依次选择身份与安全 > 网域
  2. 选择默认网域。
  3. 身份网域资源下,选择动态群组
  4. 点击创建动态群组
  5. 提供以下配置详细信息:

    • 名称:输入 cloudguard-chronicle-functions
    • 说明:输入 Dynamic group for Cloud Guard to Chronicle forwarder function
    • 匹配规则:输入以下规则:
    ALL {resource.type = 'fnfunc', resource.compartment.id = '<compartment_OCID>'}
    
  6. 点击创建

创建 IAM 政策

  1. Oracle Cloud Infrastructure 控制台中,打开导航菜单,然后依次选择身份与安全 > 政策
  2. 确保您位于根区间中。
  3. 点击创建政策
  4. 提供以下配置详细信息:
    • 名称:输入 cloudguard-chronicle-policy
    • 说明:输入 Allows Events Service to invoke Cloud Guard Chronicle forwarder function
    • 区间:确保已选择根区间。
  5. Policy Builder 部分中,切换 Show manual editor
  6. 声明字段中,输入以下政策声明:

    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>
    
  7. 点击创建

为 Cloud Guard 创建 OCI Events 规则

创建一项事件规则,以便在 Cloud Guard 检测到、关闭或修复问题时触发 OCI 函数。

创建“事件”规则

  1. Oracle Cloud Infrastructure 控制台中,打开导航菜单,然后依次选择可观测性和管理 > 事件服务 > 规则

  2. 点击创建规则

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

    • 显示名称:输入 cloudguard-to-chronicle
    • 说明:输入 Forwards Cloud Guard problem events to Google SecOps via OCI Functions
  4. 规则条件部分中:

    1. 条件设置为事件类型
    2. 服务名称设置为 Cloud Guard
    3. 活动类型设置为以下值(可选择多个):
      • 检测到 - 问题
      • 已关闭 - 问题
      • 已解决 - 问题
  5. 操作部分中:

    1. 操作类型设置为函数
    2. 函数区间设置为包含函数应用的区间。
    3. Function Application 设置为 cloudguard-chronicle-forwarder
    4. 函数设置为 cloudguard-to-chronicle
  6. 点击创建规则

可选:按风险等级过滤

如需仅转发严重程度较高的问题,请向“事件”规则添加属性过滤条件:

  1. 规则条件部分,点击 + 其他条件
  2. 条件设置为属性
  3. 属性名称设置为 riskLevel
  4. 属性值设置为所需的风险级别:

    • CRITICAL
    • HIGH

验证集成

手动测试函数

  1. Oracle Cloud Infrastructure 控制台中,打开导航菜单,然后依次选择开发者服务 > 函数 > 应用
  2. 点击 cloudguard-chronicle-forwarder 应用。
  3. 资源下,选择函数
  4. 点击 cloudguard-to-chronicle 函数。
  5. 点击调用以使用示例数据测试函数,或使用 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
    
  6. 依次前往开发者服务 > 函数 > 应用 > cloudguard-chronicle-forwarder,然后选择日志部分,验证函数日志。

  7. 查找确认转发成功的日志条目:

    Received Cloud Guard event: com.oraclecloud.cloudguard.problemdetected
    Successfully forwarded event to Chronicle (HTTP 200)
    

在 Google SecOps 中验证

  1. 登录 Google SecOps
  2. 前往搜索,然后查询日志类型为 OCI_CLOUDGUARD 的事件。
  3. 验证 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 支持多种身份验证方法。选择供应商支持的方法。

如果您的供应商支持自定义 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 从变更日志映射

更新日志

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

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