detection_testing_execution.py

import argparse
import copy
import csv
# from ctypes.wintypes import tagRECT
import json
import os
import queue
import random
import secrets
import shutil
import string
import subprocess
import sys
import threading
import time
from collections import OrderedDict
from datetime import datetime, timedelta
from posixpath import basename
from tempfile import mkdtemp
from timeit import default_timer as timer
from typing import Union
from urllib.parse import urlparse
import signal


import docker
import requests
import requests.packages.urllib3
from docker.client import DockerClient
from requests import get



from modules import (container_manager, new_arguments2,
                     testing_service, validate_args, utils)
from modules.github_service import GithubService
from modules.validate_args import validate, validate_and_write, ES_APP_NAME

SPLUNK_CONTAINER_APPS_DIR = "/opt/splunk/etc/apps"
index_file_local_path = "indexes.conf.tar"
index_file_container_path = os.path.join(SPLUNK_CONTAINER_APPS_DIR, "search")

# Should be the last one we copy.
datamodel_file_local_path = "datamodels.conf.tar"
datamodel_file_container_path = os.path.join(
    SPLUNK_CONTAINER_APPS_DIR, "Splunk_SA_CIM")


authorizations_file_local_path = "authorize.conf.tar"
authorizations_file_container_path = "/opt/splunk/etc/system/local"

CONTAINER_APP_DIRECTORY = "apps"

MAX_RECOMMENDED_CONTAINERS_BEFORE_WARNING = 2





def copy_local_apps_to_directory(apps: dict[str, dict], splunkbase_username:tuple[str,None] = None, splunkbase_password:tuple[str,None] = None, mock:bool = False, target_directory:str = "apps") -> str:
    if mock is True:
        target_directory = os.path.join("prior_config", target_directory)
        
        # Remove the apps directory or the prior config directory.  If it's just an apps directory, then we don't want
        #to remove that.
        shutil.rmtree(target_directory, ignore_errors=True)
    try:
        # Make sure the directory exists.  If it already did, that's okay. Don't delete anything from it
        # We want to re-use previously downloaded apps
        os.makedirs(target_directory, exist_ok = True)
        
    except Exception as e:
        raise(Exception(f"Some error occured when trying to make the {target_directory}: [{str(e)}]"))

    
    for key, item in apps.items():

        # These apps are URLs that will be passed.  The apps will be downloaded and installed by the container
        # # Get the file from an http source
        splunkbase_info = True if ('app_number' in item and item['app_number'] is not None and 
                                  'app_version' in item and item['app_version'] is not None) else False
        splunkbase_creds = True if (splunkbase_username is not None and 
                                   splunkbase_password is not None) else False
        can_download_from_splunkbase = splunkbase_info and splunkbase_creds

        

        #local apps can either have a local_path or an http_path
        if 'local_path' in item:
            source_path = os.path.abspath(os.path.expanduser(item['local_path']))
            base_name = os.path.basename(source_path)
            dest_path = os.path.join(target_directory, base_name)
            try:
                print(f"copying {os.path.relpath(source_path)} to {os.path.relpath(dest_path)}")
                shutil.copy(source_path, dest_path)
                item['local_path'] = dest_path
            except shutil.SameFileError as e:
                # Same file, not a real error.  The copy just doesn't happen
                print("err:%s" % (str(e)))
                pass
            except Exception as e:
                print("Error copying ESCU Package [%s] to [%s]: [%s].\n\tQuitting..." % (
                    source_path, dest_path, str(e)), file=sys.stderr)
                sys.exit(1)

        
        elif can_download_from_splunkbase is True:
            #Don't do anything, this will be downloaded from splunkbase
            pass
        elif splunkbase_info is True and splunkbase_creds is False and mock is True:
            #Don't need to do anything, when this actually runs the apps will be downloaded from Splunkbase
            #There is another opportunity to provide the creds then
            pass
        elif 'http_path' in item and can_download_from_splunkbase is False:
            http_path = item['http_path']
            try:
                url_parse_obj = urlparse(http_path)
                path_after_host = url_parse_obj[2].rstrip('/') #removes / at the end, if applicable
                base_name = path_after_host.rpartition('/')[-1] #just get the file name
                dest_path = os.path.join(target_directory, base_name) #write the whole path
                utils.download_file_from_http(http_path, dest_path, verbose_print=True)
                #we need to update the local path because this is used to copy it into the container later
                item['local_path'] = dest_path
                #Remove the HTTP Path, we will use the local_path instead
            except Exception as e:
                print("Error trying to download %s @ %s: [%s].  This app is required.\n\tQuitting..."%(key, http_path, str(e)),file=sys.stderr)
                sys.exit(1)

        elif splunkbase_info is False:
            print(f"Error - trying to install an app [{key}] that does not have 'local_path', 'http_path', "
                   "or 'app_version' and 'app_number' for installing from Splunkbase.\n\tQuitting...")
            sys.exit(1)
    return target_directory


def ensure_security_content(branch: str, commit_hash: Union[str,None], pr_number: Union[int, None], persist_security_content: bool) -> tuple[GithubService, bool]:
    if persist_security_content is True and os.path.exists("security_content"):
        print("****** You chose --persist_security_content and the security_content directory exists. "
              "We will not check out the repo again. Please be aware, this could cause issues if your "
              "repo is out of date or if a previous build failed to download all required tools and "\
              "libraries.  If this occurs, it is suggested to change the "\
              "persist_security_content setting to false. ******")
              
        github_service = GithubService(
            branch, commit_hash, persist_security_content=persist_security_content)

    else:
        if persist_security_content is True and not os.path.exists("security_content"):
            print("Error - you chose --persist_security_content but the security_content directory does not exist!"
                  "  We will check it out for you.")
            persist_security_content = False

        elif os.path.exists("security_content/"):
            print("Deleting the security_content directory")
            try:
                shutil.rmtree("security_content/", ignore_errors=True)
                print("Successfully removed security_content directory")
            except Exception as e:
                print(
                    "Error - could not remove the security_content directory: [%s].\n\tQuitting..." % (str(e)))
                sys.exit(1)

        if pr_number:
            github_service = GithubService(branch, commit_hash, pr_number)
        else:
            github_service = GithubService(branch, commit_hash)

    return github_service, persist_security_content


def generate_escu_app(persist_security_content: bool = False) -> str:
    # Go into the security content directory
    print("****GENERATING ESCU APP****")
    os.chdir("security_content")
    if persist_security_content is False:
        commands = ["python ../../../contentctl.py --path . --skip_enrichment generate --product ESCU --output dist/escu"]
    else:
        commands = ["python ../../../contentctl.py --path . --skip_enrichment generate --product ESCU --output dist/escu"]
    ret = subprocess.run("; ".join(commands),
                         shell=True, capture_output=True)
    if ret.returncode != 0:
        print(f"Error generating new content0.\n\tQuitting and dumping error...\n{str(ret.stderr)}\n{str(ret.stdout)}")
        sys.exit(1)

    ret = subprocess.run("tar -czf DA-ESS-ContentUpdate.spl -C dist/escu .",
                         shell=True, capture_output=True)
    if ret.returncode != 0:
        print("Error generating new content1.\n\tQuitting and dumping error...\n[%s]" % (
            ret.stderr))
        sys.exit(1)

    output_file_name = "DA-ESS-ContentUpdate-latest.tar.gz"
    output_file_path_from_slim_latest = os.path.join(
        "upload", output_file_name)
    output_file_path_from_security_content = os.path.join(
        "slim_packaging", output_file_path_from_slim_latest)
    output_file_path_from_root = os.path.join(
        "security_content", output_file_path_from_security_content)

    if persist_security_content is True:
        try:
            os.remove(output_file_path_from_security_content)
        except FileNotFoundError:
            # No problem if we fail to remove it, that just means it wasn't there and we didn't need to
            pass
        except Exception as e:
            print("Error deleting the (possibly) existing old ESCU File: [%s]" % (
                str(e)), file=sys.stderr)
            sys.exit(1)

        # There remove the latest file if it exists
        commands = ["cd slim_packaging",
                    "cp -R ../dist/escu DA-ESS-ContentUpdate",
                    "mkdir upload",
                    "tar -czf upload/DA-ESS-ContentUpdate*.tar.gz DA-ESS-ContentUpdate",
                    "cp upload/DA-ESS-ContentUpdate*.tar.gz %s" % (output_file_path_from_slim_latest)]

    else:
        os.mkdir("slim_packaging")
        commands = ["rm -rf slim_packaging/slim_latest",
                    "mkdir slim_packaging",
                    "cd slim_packaging",
                    "cp -R ../dist/escu DA-ESS-ContentUpdate",
                    "mkdir upload",
                    "tar -czf upload/DA-ESS-ContentUpdate*.tar.gz DA-ESS-ContentUpdate",
                    "cp upload/DA-ESS-ContentUpdate*.tar.gz %s" % (output_file_path_from_slim_latest)]

    ret = subprocess.run("; ".join(commands),
                         shell=True, capture_output=True)
    if ret.returncode != 0: 
        print("Command List:\n%s" % (commands))
        print("Error generating new ESCU Package.\n\tQuitting and dumping error...\n[%s]" % (
            ret.stderr.decode('utf-8')), file=sys.stderr)
        sys.exit(1)
    os.chdir("../")

    return output_file_path_from_root



def finish_mock(settings: dict, detections: list[str], output_file_template: str = "prior_config/config_tests_%d.json")->bool:
    num_containers = settings['num_containers']

    for output_file_index in range(0, num_containers):
        fname = output_file_template % (output_file_index)

        # Get the n'th detection for this file
        detection_tests = detections[output_file_index::num_containers]
        normalized_detection_names = []
        # Normalize the test filename to the name of the detection instead.
        # These are what we should write to the file
        for d in detection_tests:
            filename = os.path.basename(d)
            filename = filename.replace(".test.yml", ".yml")
            leading = os.path.split(d)[0]
            leading = leading.replace("tests/", "detections/")
            new_name = os.path.join(
                "security_content", leading, filename)
            normalized_detection_names.append(new_name)

        # Generate an appropriate config file for this test
        mock_settings = copy.deepcopy(settings)
        # This may be able to support as many as 2 for GitHub Actions...
        # we will have to determine in testing.
        mock_settings['num_containers'] = 1

        # Must be selected since we are passing in a list of detections
        mock_settings['mode'] = 'selected'

        # Pass in the list of detections to run
        mock_settings['detections_list'] = normalized_detection_names

        # We want to persist security content and run with the escu package that we created.
        #Note that if we haven't checked this out yet, we will check it out for you.
        mock_settings['persist_security_content'] = True

        mock_settings['mock'] = False

        # Make sure that it still validates after all of the changes

        try:
            with open(fname, 'w') as outfile:
                validated_settings, b = validate_and_write(configuration=mock_settings, output_file = outfile, strip_credentials=True)
                if validated_settings is None:
                    print(
                        "There was an error validating the updated mock settings.\n\tQuitting...", file=sys.stderr)
                    return False

        except Exception as e:
            print("Error writing config file %s: [%s]\n\tQuitting..." % (
                fname, str(e)), file=sys.stderr)
            return False

    return True


def main(args: list[str]):
    #Disable insecure warnings.  We make a number of HTTPS requests to Splunk
    #docker containers that we've set up.  Without this line, we get an 
    #insecure warning every time due to invalid cert.
    requests.packages.urllib3.disable_warnings()

    start_datetime = datetime.now()
    
    action, settings = new_arguments2.parse(args)
    if action == "configure":
        # Done, nothing else to do
        print("Configuration complete!")
        sys.exit(0)
    elif action != "run":
        print("Unsupported action: [%s]" % (action), file=sys.stderr)
        sys.exit(1)

    if settings['mock'] is False:
        # If this is a real run, then make sure Docker is installed and running and usable
        # If this is a mock, then that is not required.  By only checking on a non-mock
        # run, we save ourselves the need to install docker in the CI for the manifest
        # generation step.
        try:
            docker.client.from_env()
        except Exception as e:
            print("Error, failed to get docker client.  Is Docker Installed and Running?\n\t%s" % (str(e)))
            sys.exit(1)
        
        credentials_needed = False
        credential_error = False
        
        
        
        if settings['splunkbase_username'] == None or settings['splunkbase_password'] == None:

            missing_credentials = []
            if settings['splunkbase_username'] == None:
                missing_credentials.append("--splunkbase_username")
            if settings['splunkbase_password'] == None:
                missing_credentials.append("--splunkbase_password")
            
            missing_credentials_string = '\n\t'.join(missing_credentials)
            
            splunkbase_only_apps = []
            for app,content in settings['apps'].items():
                if 'local_path' not in content and 'http_path' not in content:
                    splunkbase_only_apps.append(app)
            if len(splunkbase_only_apps) != 0:
                print(f"Error - you have attempted to install the following apps: {splunkbase_only_apps}, "
                     "but you have not provided a local_path or an http_path in the config file.  Normally, "
                     "we would download these from Splunkbase, but the following credentials are "
                     f"missing:\n\t{missing_credentials_string}\n  Please provide them on the command line "
                     "or in the config file.\n\tQuitting...")
                sys.exit(1)

            print(f"You have listed apps to install but have "\
                  f"not provided\n\t{missing_credentials_string} \nvia the command line or config file. "
                  f"We will download these files from S3 rather than Splunkbase.")
        else:

            print(f"You have listed apps to install and provided Splunkbase credentials. "\
                  f"These apps will be downloaded and installed from Splunkbase!")

        

    
    FULL_DOCKER_HUB_CONTAINER_NAME = "splunk/splunk:%s" % settings['container_tag']

    if settings['num_containers'] > MAX_RECOMMENDED_CONTAINERS_BEFORE_WARNING:
        print("You requested to run with [%d] containers which may use a very large amount of resources "
              "as they all run in parallel.  The maximum suggested number of parallel containers is "
              "[%d].  We will do what you asked, but be warned!" % (settings['num_containers'], MAX_RECOMMENDED_CONTAINERS_BEFORE_WARNING))

    # Check out security content if required
    try:
        #Make sure we fix up the persist_securiy_content argument if it is passed in error (we say it exists but it doesn't)
        github_service, settings['persist_security_content'] = ensure_security_content(
            settings['branch'], settings['commit_hash'], settings['pr_number'], settings['persist_security_content'])
        settings['commit_hash'] = github_service.commit_hash
    except Exception as e:
        print("\nFailure checking out git repository: [%s]"\
              "\n\tCommit Hash: [%s]"\
              "\n\tBranch     : [%s]"\
              "\n\tPR         : [%s]\n\tQuitting..."%
              (str(e),settings['commit_hash'],settings['branch'],settings['pr_number']),file=sys.stderr)
        sys.exit(1)

    #passes = [{'search_string': '| tstats `security_content_summariesonly` count min(_time) as firstTime max(_time) as lastTime from datamodel=Endpoint.Processes where (Processes.process_name ="7z.exe" OR Processes.process_name = "7za.exe" OR Processes.original_file_name = "7z.exe" OR Processes.original_file_name =  "7za.exe") AND (Processes.process="*\\\\C$\\\\*" OR Processes.process="*\\\\Admin$\\\\*" OR Processes.process="*\\\\IPC$\\\\*") by Processes.original_file_name Processes.parent_process_name Processes.parent_process Processes.process_name Processes.process Processes.parent_process_id Processes.process_id  Processes.dest Processes.user | `drop_dm_object_name(Processes)` | `security_content_ctime(firstTime)` | `security_content_ctime(lastTime)` | `7zip_commandline_to_smb_share_path_filter` | stats count | where count > 0', 'detection_name': '7zip CommandLine To SMB Share Path', 'detection_file': 'endpoint/7zip_commandline_to_smb_share_path.yml', 'success': True, 'error': False, 'diskUsage': '286720', 'runDuration': '0.922', 'scanCount': '4897'}]
    #github_service.update_and_commit_passed_tests(passes)
    #sys.exit(0)
    # Make a backup of this config containing the hash and stripped credentials.
    # This makes the test perfectly reproducible.
    reproduce_test_config, _ = validate_args.validate_and_write(settings, output_file=None, strip_credentials=True)
    if reproduce_test_config == None:
        print("Error - there was an error writing out the file to reproduce the test.  This should not happen, as all "\
              "settings should have been validated by this point.\n\tQuitting...",file=sys.stderr)
        sys.exit(1)

    try:
        all_detection_files = github_service.get_detection_files(settings['mode'],
                                                    settings['folders'],
                                                    settings['types'],
                                                    settings['detections_list'])
        
        #We randomly shuffle this because there are likely patterns in searches.  For example,
        #cloud/endpoint/network likely have different impacts on the system.  By shuffling,
        #we spread out this load on a single computer, but also spread it in case
        #we are running on GitHub Actions against multiple machines.  Hopefully, this
        #will reduce that chnaces the some machines run and complete quickly while
        #others take a long time.
        random.shuffle(all_detection_files)        

    except Exception as e:
        print("Error getting test files:\n%s"%(str(e)), file=sys.stderr)
        print("\tQuitting...", file=sys.stderr)
        sys.exit(1)

    print("***This run will test [%d] detections!***"%(len(all_detection_files)))
    

    


    # Check to see if we want to install ESCU and whether it was preeviously generated and we should use that file
    if ES_APP_NAME in settings['apps'] and settings['apps'][ES_APP_NAME]['local_path'] is not None:
        # Using a pregenerated ESCU, no need to build it
        pass

    elif ES_APP_NAME not in settings['apps']:
        print(f"{ES_APP_NAME} was not found in {settings['apps'].keys()}.  We assume this is an error and shut down.\n\t"
              "Quitting...", file=sys.stderr)
        sys.exit(1)
    else:
        # Generate the ESCU package from this branch.
        source_path = generate_escu_app(settings['persist_security_content'])
        settings['apps']['SPLUNK_ES_CONTENT_UPDATE']['local_path'] = source_path
        

    # Copy all the apps, to include ESCU (whether pregenerated or just generated)
    try:
        relative_app_path = copy_local_apps_to_directory(settings['apps'], 
                                     splunkbase_username = settings['splunkbase_username'], 
                                     splunkbase_password = settings['splunkbase_password'], 
                                     mock=settings['mock'], target_directory = CONTAINER_APP_DIRECTORY)
        
        mounts = [{"local_path": os.path.abspath(relative_app_path),
                    "container_path": "/tmp/apps", "type": "bind", "read_only": True}]
    except Exception as e:
        print(f"Error occurred when copying apps to app folder: [{str(e)}]\n\tQuitting...", file=sys.stderr)
        sys.exit(1)


    # If this is a mock run, finish it now
    if settings['mock']:
        #The function below 
        if finish_mock(settings, all_detection_files):
            # mock was successful!
            print("Mock successful!  Manifests generated!")
            sys.exit(0)
        else:
            print("There was an unrecoverage error during the mock.\n\tQuitting...",file=sys.stderr)
            sys.exit(1)



    #Add some files that always need to be copied to to container to set up indexes and datamodels.
    files_to_copy_to_container = OrderedDict()
    files_to_copy_to_container["INDEXES"] = {
        "local_file_path": index_file_local_path, "container_file_path": index_file_container_path}
    files_to_copy_to_container["DATAMODELS"] = {
        "local_file_path": datamodel_file_local_path, "container_file_path": datamodel_file_container_path}
    files_to_copy_to_container["AUTHORIZATIONS"] = {
        "local_file_path": authorizations_file_local_path, "container_file_path": authorizations_file_container_path}
    

    
    def shutdown_signal_handler_setup(sig, frame):
        
        print(f"Signal {sig} received... stopping all [{settings['num_containers']}] containers and shutting down...")
        shutdown_client = docker.client.from_env()
        errorCount = 0
        for container_number in range(settings['num_containers']):
            container_name = settings['local_base_container_name']%container_number
            print(f"Shutting down {container_name}...", file=sys.stderr, end='')
            sys.stdout.flush()
            try:
                container = shutdown_client.containers.get(container_name)
                #Note that stopping does not remove any of the volumes or logs,
                #so stopping can be useful if we want to debug any container failure 
                container.stop(timeout=10)
                print("done", file=sys.stderr)
            except Exception as e:
                print(f"Error trying to shut down {container_name}. It may have already shut down.  Stop it youself with 'docker containter stop {container_name}", sys.stderr)
                errorCount += 1
        if errorCount == 0:
            print("All containers shut down successfully", file=sys.stderr)        
        else:
            print(f"{errorCount} containers may still be running. Find out what is running with:\n\t'docker container ls'\nand shut them down with\n\t'docker container stop CONTAINER_NAME' ", file=sys.stderr)
        
        print("Quitting...",file=sys.stderr)
        #We must use os._exit(1) because sys.exit(1) actually generates an exception which can be caught! And then we don't Quit!
        os._exit(1)
        

            

    #Setup requires a different teardown handler than during execution
    signal.signal(signal.SIGINT, shutdown_signal_handler_setup)
    
    
    try:
        cm = container_manager.ContainerManager(all_detection_files,
                                                FULL_DOCKER_HUB_CONTAINER_NAME,
                                                settings['local_base_container_name'],
                                                settings['num_containers'],
                                                settings['apps'],
                                                settings['branch'],
                                                settings['commit_hash'],
                                                reproduce_test_config,
                                                files_to_copy_to_container=files_to_copy_to_container,
                                                web_port_start=8000,
                                                management_port_start=8089,
                                                mounts=mounts,
                                                show_container_password=settings['show_splunk_app_password'],
                                                container_password=settings['splunk_app_password'],
                                                splunkbase_username=settings['splunkbase_username'],
                                                splunkbase_password=settings['splunkbase_password'],
                                                reuse_image=settings['reuse_image'],
                                                interactive_failure=not settings['no_interactive_failure'],
                                                interactive=settings['interactive'])
    except Exception as e:
        print("Error - unrecoverable error trying to set up the containers: [%s].\n\tQuitting..."%(str(e)),file=sys.stderr)
        sys.exit(1)

    def shutdown_signal_handler_execution(sig, frame):
        #Set that a container has failed which will gracefully stop the other containers.
        #This way we get our full cleanup routine, too!
        print("Got a signal to shut down. Shutting down all containers, please wait...", file=sys.stderr)
        cm.synchronization_object.containerFailure()
    
    #Update the signal handler

    signal.signal(signal.SIGINT, shutdown_signal_handler_execution)
    try:
        result = cm.run_test()
    except Exception as e:
        print("Error - there was an error running the tests: [%s]\n\tQuitting..."%(str(e)),file=sys.stderr)
        sys.exit(1)

    #github_service.update_and_commit_passed_tests(cm.synchronization_object.successes)
    

    #Return code indicates whether testing succeeded and all tests were run.
    #It does NOT indicate that all tests passed!
    if result is True:
        print("Test Execution Successful")
        sys.exit(0)
    else:
        print("Test Execution Failed - review the logs for more details")
        #Because one or more of the threads could be stuck in a certain setup loop, like
        #trying to copy files to a containers (which igonores errors), we must os._exit
        #instead of sys.exit
        os._exit(1)


if __name__ == "__main__":
    main(sys.argv[1:])