admin/splunk-security_content
1
0
Fork 0
mirror of https://github.com/splunk/security_content synced 2026-06-08 17:32:49 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
b6fedaa03032f58737a469f6ffc344bcc7b35607
splunk-security_content/bin/docker_detection_tester/detection_testing_execution.py
T
pyth0n1c 6556205305 initial support for pulling tests 
from the detection file instead of
separately. This feature is implemented
as a stopgap as we move to new
tooling.
2023-01-04 11:28:50 -08:00

565 lines
27 KiB
Python
Raw Blame History

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:])
Reference in New Issue View Git Blame Copy Permalink