splunk-security_content/bin/docker_detection_tester/modules/splunk_sdk.py

from os import error
import sys
from time import sleep
import splunklib.client as client
import splunklib.results as results
import requests
import time
import timeit
import datetime
from typing import Union

DEFAULT_EVENT_HOST = "ATTACK_DATA_HOST"
DEFAULT_DATA_INDEX = "main"
FAILURE_SLEEP_INTERVAL_SECONDS = 60

def enable_delete_for_admin(splunk_host:str, splunk_port:int, splunk_password:str)->bool:
    try:
        service = client.connect(
            host=splunk_host,
            port=splunk_port,
            username='admin',
            password=splunk_password
        )
    except Exception as e:
        raise(Exception("Unable to connect to Splunk instance: " + str(e)))
        

    #write the following contents to /opt/splunk/etc/system/local/authorize.conf
    "[role_admin]"\
    "delete_by_keyword = enabled"\
    "grantableRoles = admin"\
    "importRoles = can_delete;user;power_user"\
    "srchIndexesAllowed = *;_*;main"\
    "srchIndexesDefault = main"\
    "srchMaxTime = 8640000"

    #Run the following search, equivalent to running ./splunk reload auth, to get the settings to take effect
    
    update_changed_auth_search = "| rest splunk_server=* /services/authentication/providers/services/_reload"


    try:
        job = service.jobs.create(update_changed_auth_search)
    except Exception as e:
        error_message = "Unable to enable delete: %s"%(str(e))
        return False
    
    input("Waiting for you to check that delete has been enabled with: %s"%(update_changed_auth_search))
    return True
    '''
    # search and replace \\ with \\\
    # search = search.replace('\\','\\\\')
    role = service.roles['admin']
    try:
        role.grant('delete_by_keyword')
    except Exception as e:
        print("Error - failed trying to grant 'can_delete' privs to admin: [%s]"%(str(e)))
        return False
    '''
    return True




def get_number_of_indexed_events(splunk_host, splunk_port, splunk_password, index:str, event_host:str=DEFAULT_EVENT_HOST, sourcetype:Union[str,None]=None )->int:

    try:
        service = client.connect(
            host=splunk_host,
            port=splunk_port,
            username='admin',
            password=splunk_password
        )
    except Exception as e:
        raise(Exception("Unable to connect to Splunk instance: " + str(e)))

    if sourcetype is not None:
        search = f'''search index="{index}" sourcetype="{sourcetype}" host="{event_host}" | stats count'''
    else:
        search = f'''search index="{index}" host="{event_host}" | stats count'''
    kwargs = {"exec_mode":"blocking"}
    try:
        job = service.jobs.create(search, **kwargs)
  
        #This returns the count in string form, not as an int. For example:
        #OrderedDict([('count', '59630')])
        results_stream = job.results(output_mode='json')
        count = None
        for res in results.JSONResultsReader(results_stream):
            if 'count' in res:
                count = int(res['count'],10)
        if count is None:
            raise Exception(f"Expected the get_number_of_indexed_events search to only return 1 count, but got {len(search_results)} instead.")
        
        return count    

    except Exception as e:
        raise Exception("Error trying to get the count while waiting for indexing to complete: %s"%(str(e)))
        
    


def wait_for_indexing_to_complete(splunk_host, splunk_port, splunk_password, sourcetype:str, index:str, check_interval_seconds:int=10)->bool:
    startTime = timeit.default_timer()
    previous_count = -1
    time.sleep(check_interval_seconds)
    while True:
        new_count = get_number_of_indexed_events(splunk_host, splunk_port, splunk_password, index=index, sourcetype=sourcetype)
        #print(f"Previous Count [{previous_count}] New Count [{new_count}]")
        if previous_count == -1:
            previous_count = new_count
        else:
            if new_count == previous_count:
                stopTime = timeit.default_timer()
                return True
            else:
                previous_count = new_count
        
        #If new_count is really low, then the server is taking some extra time to index the data.
        # So sleep for longer to make sure that we give time to complete (or at least process more
        # events so we don't return from this function prematurely) 
        if new_count < 2:
            time.sleep(check_interval_seconds*3)
        else:
            time.sleep(check_interval_seconds)
        

def test_baseline_search(splunk_host, splunk_port, splunk_password, search, pass_condition, baseline_name, baseline_file, earliest_time, latest_time)->dict:
    try:
        service = client.connect(
            host=splunk_host,
            port=splunk_port,
            username='admin',
            password=splunk_password
        )
    except Exception as e:
        raise(Exception("Unable to connect to Splunk instance: " + str(e)))
        


    # search and replace \\ with \\\
    # search = search.replace('\\','\\\\')

    if search.startswith('|'):
        updated_search = search
    else:
        updated_search = 'search ' + search

    kwargs = {"exec_mode": "blocking",
              "dispatch.earliest_time": earliest_time,
              "dispatch.latest_time": latest_time}

    splunk_search = updated_search + ' ' + pass_condition

    try:
        job = service.jobs.create(splunk_search, **kwargs)
    except Exception as e:
        raise(Exception("Unable to execute baseline: " + str(e)))
        

    test_results = dict()
    test_results['diskUsage'] = job['diskUsage']
    test_results['runDuration'] = job['runDuration']
    test_results['baseline_name'] = baseline_name
    test_results['baseline_file'] = baseline_file
    test_results['scanCount'] = job['scanCount']

    if int(job['resultCount']) != 1:
        print("Test failed for baseline: " + baseline_name)
        test_results['error'] = True
        return test_results
    else:
        print("Test successful for baseline: " + baseline_name)
        test_results['error'] = False
        return test_results



def test_detection_search(splunk_host:str, splunk_port:int, splunk_password:str, search:str, pass_condition:str, 
                          detection_name:str, detection_file:str, earliest_time:str, latest_time:str, attempts_remaining:int=4, 
                          failure_sleep_interval_seconds:int=FAILURE_SLEEP_INTERVAL_SECONDS)->dict:
    #Since this is an attempt, decrement the number of remaining attempts
    attempts_remaining -= 1
    
    if search.startswith('|'):
        updated_search = search
    else:
        updated_search = 'search ' + search 

    kwargs = {"exec_mode": "blocking",
              "dispatch.earliest_time": "-1d",
              "dispatch.latest_time": "now"}

    splunk_search = updated_search + ' ' + pass_condition
    test_results = dict()
    
    #These will always be present. By default, we will say that the
    #test has failed AND there was an error (until they are set otherwise)
    test_results['search_string'] = splunk_search
    test_results['detection_name'] = detection_name
    test_results['detection_file'] = detection_file
    
    test_results['success'] = False
    test_results['error'] = True

    
    try:
        service = client.connect(
            host=splunk_host,
            port=splunk_port,

            username='admin',
            password=splunk_password
        )
    except Exception as e:
        error_message = "Unable to connect to Splunk instance: %s"%(str(e))
        print(error_message,file=sys.stderr)
        test_results['error'] = True
        test_results['detection_error'] = error_message
        return test_results


    # search and replace \\ with \\\
    # search = search.replace('\\','\\\\')

   

    #print("SEARCH: %s"%(splunk_search))


    try:
        job = service.jobs.create(splunk_search, **kwargs)
        results_stream = job.results(output_mode='json')

    except Exception as e:
        
        error_message = "Unable to execute detection: %s"%(str(e))
        print(error_message,file=sys.stderr)
        test_results['error'] = True
        test_results['detection_error'] = error_message
        return test_results

    test_results['diskUsage'] = job['diskUsage']
    test_results['runDuration'] = job['runDuration']
    test_results['scanCount'] = job['scanCount']
    
    #If we get this far, then there was not an error
    #The search may have FAILED, but there was no error in the search
    test_results['error'] = False 


    #Should this be 1 for a pass, or should it be greater than 0?
    if int(job['resultCount']) != 1:
        #print("Test failed for detection: " + detection_name)
        if attempts_remaining > 0:
            print(f"Execution of test failed for [{detection_name}]. Sleeping for [{failure_sleep_interval_seconds} seconds] and trying up to {attempts_remaining} more times...")
            time.sleep(failure_sleep_interval_seconds)
            return test_detection_search(splunk_host, splunk_port, splunk_password, search, pass_condition, detection_name, detection_file, 
                                         earliest_time, latest_time, attempts_remaining=attempts_remaining, 
                                         failure_sleep_interval_seconds=failure_sleep_interval_seconds)
        else:
            test_results['success'] = False
            return test_results
    else:
        #print("Test successful for detection: " + detection_name)
        test_results['success'] = True
        return test_results


def delete_attack_data(splunk_host:str, splunk_password:str, splunk_port:int, wait_on_delete:Union[dict,None], search_string:str, detection_filename:str, indices:list[str]=[DEFAULT_DATA_INDEX], host:str=DEFAULT_EVENT_HOST)->bool:
    
    if wait_on_delete:
        print(wait_on_delete['message'])
        print("FILENAME : [%s]"%(detection_filename))
        print("SEARCH   :\n%s"%(search_string))
        _ = input("****************Press ENTER to Complete Test and DELETE data****************\n\n\n")
    
    try:
        service = client.connect(
            host=splunk_host,
            port=splunk_port,

            username='admin',
            password=splunk_password
        )
    except Exception as e:

        raise(Exception("Unable to connect to Splunk instance: " + str(e)))


    #print(f"Deleting data for {detection_filename}: {indices}")
    for index in indices:
        while (get_number_of_indexed_events(splunk_host, splunk_port, splunk_password, index=index, event_host=host) != 0) :
            splunk_search = f'search index="{index}" host="{host}" | delete'
            kwargs = {
                    "exec_mode": "blocking",
                    "dispatch.earliest_time": "-1d",
                    "dispatch.latest_time": "now"}
            try:
                
                job = service.jobs.create(splunk_search, **kwargs)
                results_stream = job.results(output_mode='json')
                reader = results.JSONResultsReader(results_stream)


            except Exception as e:
                raise(Exception(f"Trouble deleting data using the search {splunk_search}: {str(e)}"))
        
    
    return True