Push Cloud Guard Problems to Splunk HEC with OCI SDK

Introduction

Cloud Guard is an OCI Service for Security Posture Management. Cloud Guard’s functionality is to assess, alert and act on OCI resource family by continuously monitoring the current OCI Tenancy where it is enabled. To get a high-level understanding on Cloud Guard refer to this Blog  by Paul Toal. A common use case we see from customers is to export the Security Violations (i.e Problems) identified by Cloud Guard for external consumption, storage and audit for SOC Operations.

Although this use case is very specific, customers could choose one of the following ways for sending events to external sources:

1. Using Responders to generate Events to be consumed by OCI Event Service and trigger OCI function to push to SIEM/Storage of their choice. This operation happens once for every Cloud Guard problem found.
2. Using SDK/APIs to push Cloud Guard Problems to SIEM/Storage of their choice. This is a batch operation that collects Cloud Guard problems for a time frame, to be triggered as a cron process.

In this specific example we will look at use case 2 on extracting cloud guard problems via SDK and sending to an HTTP Event Collector (HEC) in Splunk. Let’s continue with the Steps

Configure Splunk HTTP Collector

Configure a Splunk HTTP Event Collector (HEC) and take a note of Source, HEC Token and URL. These details will be used by Python Module in later steps. Refer to Splunk Documentation for creating and enabling HEC.

Extract Cloud Guard problems with Python SDK

Following code excerpt highlights following activities

1. Initiating Cloud Guard Client
2. Getting Cloud Guard Problems List

You can refer to Python SDK for OCI Cloud Guard  here and here. 

!/usr/bin/env python3
# coding: utf-8
# COPYRIGHT (c) 2020 ORACLE A-TEAM
# THIS SAMPLE CODE IS PROVIDED FOR EDUCATIONAL PURPOSES OR
# TO ASSIST YOUR DEVELOPMENT OR ADMINISTRATION EFFORTS AND
# PROVIDED "AS IS" AND IS NOT SUPPORTED BY ORACLE CORPORATION.
# License: http://www.apache.org/licenses/LICENSE-2.0.html


import oci
…
…

# Initiate Cloud Guard Client
cg = oci.cloud_guard.CloudGuardClient(config={}, signer=signer)
#Set Base to Reporting Region
cg.base_client.set_region('us-ashburn-1')
…
…

try:
    list_problems_response =oci.pagination.list_call_get_all_results(cg.list_problems,
        compartment_id=compartment_ocid,
        compartment_id_in_subtree='TRUE',
        access_level="ACCESSIBLE",
        time_last_detected_greater_than_or_equal_to=start_time,
        time_last_detected_less_than_or_equal_to=end_time).data

…
…

An example Cloud Guard problem output from API in JSON format looks like following:

    {
    "compartment_id": "ocid1.tenancy.oc1..aaaaaaamcompartid2vbxyhczksgjir7xdq",
    "detector_id": "IAAS_ACTIVITY_DETECTOR",
    "detector_rule_id": "VCN_SECURITY_LIST_INGRESS_RULES_CHANGED",
    "id": "ocid1.cloudguardproblem.oc1.iad.amaaaaaac3scrubbedxcjmdz34p3kifm6a",
    "labels": [
        "Network"
    ],
    "lifecycle_detail": "OPEN",
    "lifecycle_state": "ACTIVE",
    "region": "us-ashburn-1",
    "resource_id": "ocid1.saml2idp.oc1..aaaaaaaa64zcsamlidcscufrlob46q/john.doe@mycompany.com",
    "resource_name": "john.doe@mycompany.com",
    "resource_type": "User",
    "risk_level": "MEDIUM",
    "target_id": "ocid1.cloudguardtarget.oc1.iad.amaaatargetxyxqtlhjqs6ujhsbqot7zo2247bhcqoja",
    "time_first_detected": "2020-09-29T20:11:41.978000+00:00",
    "time_last_detected": "2020-09-29T20:11:41.978000+00:00"
}

Push extracted Cloud Guard problems to HEC

def push_problems_to_HEC(cgProblems,host,token):
    url = 'https://'+host+':8088/services/collector/event'
    payload = {"host": "CG_Py_Instance","source": "CG_Problems","event": cgProblems}
    headers = {'Authorization': 'Splunk '+token,}

    try:
        response = requests.request("POST", url, headers=headers, json=payload, verify=False)
        if json.loads(response.text)["text"] == 'Success':
            print("Push to Splunk was Succcessful")
        else:
            print("Push to Splunk Failed")

    except Exception as e:
        print("Exception while Sending data to Splunk\n Message: {}".format(e))
              
    return None    

Extracting from multiple Tenants?

Make sure cross-tenant policies are enabled to 'endorse' , 'allow' and 'admit' for harvesting from multiple tenants from a Compute Instance. The picture then would look like following.

Example Cross Tenant IAM Policies based on above diagram

IAM policies required in Polling Tenant

#poll_dyn_group is a dynamic group/resource principle 
allow dynamic-group poll_dyn_group to read cloud-guard-problems in tenancy
...

define tenancy targetTenant2 as ocid..tenant2
endorse dynamic-group poll_dyn_group to read audit-events in tenancy targetTenant2
endorse dynamic-group poll_dyn_group to read cloud-guard-problems in tenancy targetTenant2
...
define tenancy targetTenant3 as ocid..tenant3
endorse dynamic-group poll_dyn_group  to read audit-events in tenancy targetTenant3
endorse dynamic-group poll_dyn_group to read cloud-guard-problems in tenancy targetTenant3

IAM policies required at every Target Tenant

define tenancy pollTenant as ocid..pollingtenant
define dynamic-group poll_dyn_group as ocid..dynamicgroup..polling..tenant
admit dynamic-group poll_dyn_group of tenancy pollTenant to read cloud-guard-problems in tenancy

Verify Events Sent and Received

At this point python module can be run to verify Problems are sent to Splunk as expected.

From the code sample:

Processing Tenant myTenant STARTED 2020-09-29 21:37:28.915985

Problem Count from Tenant: ocid1.tenancy.oc1..aaaaaamytenantrmohbw2vbxyhczksgjir7xdq : 3690

Processing Tenant myTenant ENDED 2020-09-29 21:38:00.215323 

Push to Splunk was Successful, Code: 0

In Splunk:

Conclusion

Although this blog specifically looks at sending Cloud Guard Problems to Splunk, the same option can be applied for other SIEMs which support similar functionality. If you were to choose the extraction/sending of events via functions, we will cover that as a separate topic in our Cloud Guard series.