收集 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 事件规则和通知主题的权限
  • 访问 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 后,您必须生成用于身份验证的密钥:

  1. 在 Feed 详情页面上,点击生成密钥。
  2. 对话框会显示密钥。
  3. 复制并妥善保存密钥。

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

获取 Feed 端点网址

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

    https://{region}-chronicle.googleapis.com/v1/{parent}/logs:import
    

    其中,{region} 是 Google SecOps 实例的区域(例如 us、europe 或 asia-southeast1),{parent} 是 projects/<PROJECT_ID>/locations/<LOCATION>/instances/<INSTANCE_ID>/logTypes/<LOG_TYPE> 格式的资源路径。

  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 Guard 使用云事件响应程序将问题详细信息发送到 OCI Events 服务。Cloud Event 响应器是响应器配方的一部分,必须启用并设置为自动执行。

在响应器配方中启用 Cloud 事件规则

  1. 登录 Oracle Cloud Infrastructure 控制台。
  2. 打开导航菜单,然后依次选择身份和安全 > Cloud Guard > 配方。
  3. 选择响应者配方。
  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://{region}-chronicle.googleapis.com/v1/{parent}/logs:import)
    CHRONICLE_API_KEY 为 Chronicle webhook 创建的 Google Cloud API 密钥
    CHRONICLE_SECRET Chronicle webhook 密钥
  7. 点击保存。

为 OCI Functions 创建 IAM 政策

OCI Functions 服务和 Events Service 需要 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
    • Compartment:确保选择根区间。
  5. 在政策构建器部分,切换显示手动编辑器。
  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 事件规则

创建事件规则,以便在 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. 将函数应用设置为 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. 点击 Invoke(调用),使用示例数据测试函数,或使用 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 Service 发出以下事件类型:

事件类型 触发器
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 密钥和 Secret 在网址中不可见
  • 更安全(标头不会记录在 Web 服务器访问日志中)
  • 供应商支持的首选方法

方法 2:查询参数

如果您的供应商不支持自定义标头,请将凭据附加到网址。

  • 网址格式:

    <ENDPOINT_URL>?key=<API_KEY>&secret=<SECRET_KEY>
    
  • 示例:

    https://{region}-chronicle.googleapis.com/v1/{parent}/logs:import?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 observer.hostname 来自观测者的值(如果不是 IP)
observer observer.ip Value from observer if 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_host(如果是 IP)
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(如果不为空),否则为主账号(如果是 IP),否则为主账号主机(如果是 IP)
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,则为 target 中的值;否则,如果不是 IP,则为 target_host
target_host target.hostname
目标 target.ip 如果为 IP,则为 target 中的值;如果为 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 专业人士的解答。