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

Branch was auto-updated.

Browse Source
This commit is contained in:
srv-rr-gh-researchbt
2023-05-02 15:02:48 -07:00
committed by GitHub
parent 33c68b02ea c80c0aecec
commit 44ecbd9829
10 changed files with 664 additions and 393 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/detection-smoketesting.yml
+48
View File
@@ -0,0 +1,48 @@
name: detection-smoketesting
on:
schedule:
- cron: "44 4 * * *"
jobs:
docker-detection-smoketest:
runs-on: ubuntu-latest
steps:
- name: Checkout Repo
uses: actions/checkout@v2
#with:
# ref: develop
- uses: actions/setup-python@v2
with:
python-version: "3.9" #Available versions here - https://github.com/actions/python-versions/releases easy to change/make a matrix/use pypy
architecture: "x64" # optional x64 or x86. Defaults to x64 if not specified
cache: "pip"
- name: Install Python Dependencies
run: |
python -m venv .venv
source .venv/bin/activate
python -m pip install wheel
python -m pip install -r requirements.txt
- name: Run the Smoketesting
run: |
source .venv/bin/activate
cd bin/docker_detection_tester
python detection_testing_execution.py run --branch develop --mode smoketest --config_file test_config_github_actions.json --num_containers 1
# Summarize all results in a different step so they are easy to read/jump to
- name: Run the Smoketesting
run: |
source .venv/bin/activate
cd bin/docker_detection_tester
python summarize_json.py -o test_results/summary_smoketest.json -f test_results/summary.json --smoketest
# Upload Results even on failure. Results are even more important in the event of a failure
- name: Upload Test Results Files
uses: actions/upload-artifact@v2
if: always()
with:
name: smoketest_results
path: |
bin/docker_detection_tester/test_results/summary.json
bin/docker_detection_tester/test_results/summary_smoketest.json
bin/docker_detection_tester/modules/github_service.py
+34 -8
View File
@@ -36,7 +36,6 @@ class GithubService:
PR_number: Union[int, None] = None,
persist_security_content: bool = False,
):
self.security_content_branch = security_content_branch
if persist_security_content:
print("Getting handle on existing security_content repo!")
@@ -118,7 +117,6 @@ class GithubService:
self.commit_hash = commit_hash
def update_and_commit_passed_tests(self, results: list[dict]) -> bool:
changed_file_paths = []
for result in results:
detection_obj_path = os.path.join(
@@ -192,11 +190,9 @@ class GithubService:
types_to_test: list[str],
exclude_ssa: bool = True,
) -> list[str]:
pruned_tests = []
for detection in detection_files:
if os.path.basename(detection).startswith(SSA_PREFIX) and exclude_ssa:
continue
with open(detection, "r") as d:
@@ -214,7 +210,6 @@ class GithubService:
and description["tags"].get("manual_test", False)
)
):
if len(description.get("tests", [])) == 0:
print(
Exception(
@@ -361,7 +356,6 @@ class GithubService:
types: list[str],
detections_list: Union[list[str], None],
) -> list[str]:
if mode == "changes":
tests = self.get_changed_detection_files(folders, types)
elif mode == "selected":
@@ -387,6 +381,8 @@ class GithubService:
elif mode == "all":
tests = self.get_all_tests_and_detections(folders, types)
elif mode == "smoketest":
tests = self.get_everything_including_experimental_and_deprecated(types)
else:
print(
"Error, unsupported mode [%s]. Mode must be one of %s", file=sys.stderr
@@ -400,9 +396,40 @@ class GithubService:
detection_file_list: list[str],
types_to_test: list[str] = ["Anomaly", "Hunting", "TTP"],
) -> list[str]:
return self.prune_detections(detection_file_list, types_to_test)
def get_everything_including_experimental_and_deprecated(
self,
types_to_test: list[str] = ["Anomaly", "Hunting", "TTP"],
exclude_ssa: bool = True,
) -> list[str]:
all_detections = glob.glob(
os.path.join(DETECTION_ROOT_PATH, "**", f"*{DETECTION_FILE_EXTENSION}"),
recursive=True,
)
only_selected_detection_types: list[str] = []
for detection in all_detections:
if os.path.basename(detection).startswith(SSA_PREFIX) and exclude_ssa:
continue
with open(detection, "r") as d:
description: dict = yaml.safe_load(d)
detection_filepath_without_security_content = str(
pathlib.Path(*pathlib.Path(detection).parts[1:])
)
if description.get("type", None) not in types_to_test:
continue
only_selected_detection_types.append(
detection_filepath_without_security_content
)
print(
f"Number of non-ssa tests including experimental and deprecated: {len(only_selected_detection_types)}"
)
return only_selected_detection_types
def get_all_tests_and_detections(
self,
folders: list[str] = ["endpoint", "cloud", "network"],
@@ -428,7 +455,6 @@ class GithubService:
folders=["endpoint", "cloud", "network"],
types_to_test=["Anomaly", "Hunting", "TTP"],
) -> list[str]:
branch1 = self.security_content_branch
branch2 = "develop"
g = git.Git("security_content")
bin/docker_detection_tester/modules/splunk_container.py
+153 -117
View File
@@ -19,11 +19,14 @@ import threading
import wrapt_timeout_decorator
import sys
import traceback
SPLUNKBASE_URL = "https://splunkbase.splunk.com/app/%d/release/%s/download"
SPLUNK_START_ARGS = "--accept-license"
#Give ten minutes to start - this is probably enough time
MAX_CONTAINER_START_TIME_SECONDS = 60*20
# Give ten minutes to start - this is probably enough time
MAX_CONTAINER_START_TIME_SECONDS = 60 * 20
class SplunkContainer:
def __init__(
self,
@@ -40,7 +43,7 @@ class SplunkContainer:
splunkbase_password: Union[str, None] = None,
splunk_ip: str = "127.0.0.1",
interactive_failure: bool = False,
interactive:bool = False
interactive: bool = False,
):
self.interactive_failure = interactive_failure
self.interactive = interactive
@@ -48,7 +51,7 @@ class SplunkContainer:
self.client = docker.client.from_env()
self.full_docker_hub_path = full_docker_hub_path
self.container_password = container_password
self.apps = apps
self.files_to_copy_to_container = files_to_copy_to_container
@@ -63,15 +66,14 @@ class SplunkContainer:
self.management_port = management_port_tuple[1]
self.container = self.make_container()
self.thread = threading.Thread(target=self.run_container, )
self.thread = threading.Thread(
target=self.run_container,
)
self.container_start_time = -1
self.test_start_time = -1
self.num_tests_completed = 0
def prepare_apps_path(
self,
apps: OrderedDict,
@@ -80,42 +82,51 @@ class SplunkContainer:
) -> tuple[str, bool]:
apps_to_install = []
#We don't require credentials unless we install at least one splunkbase app
# We don't require credentials unless we install at least one splunkbase app
require_credentials = False
#If the username and password are supplied, then we will use splunkbase...
#assuming that the app_name and app_number are supplied. Note that if a
#local_path is supplied, then it should override this option!
# If the username and password are supplied, then we will use splunkbase...
# assuming that the app_name and app_number are supplied. Note that if a
# local_path is supplied, then it should override this option!
if splunkbase_username is not None and splunkbase_password is not None:
use_splunkbase = True
else:
use_splunkbase = False
for app_name, app_info in self.apps.items():
if use_splunkbase is True and 'local_path' not in app_info:
target = SPLUNKBASE_URL % (app_info["app_number"], app_info["app_version"])
if use_splunkbase is True and "local_path" not in app_info:
target = SPLUNKBASE_URL % (
app_info["app_number"],
app_info["app_version"],
)
apps_to_install.append(target)
#We will require credentials since we are installing at least one splunkbase app
# We will require credentials since we are installing at least one splunkbase app
require_credentials = True
#Some paths may have a local_path and an HTTP path defined. Default to the local_path first,
#mostly because we may have copied it before into the cache to speed up start time.
elif 'local_path' in app_info:
app_file_name = os.path.basename(app_info['local_path'])
# Some paths may have a local_path and an HTTP path defined. Default to the local_path first,
# mostly because we may have copied it before into the cache to speed up start time.
elif "local_path" in app_info:
app_file_name = os.path.basename(app_info["local_path"])
app_file_container_path = os.path.join("/tmp/apps", app_file_name)
apps_to_install.append(app_file_container_path)
elif 'http_path' in app_info:
apps_to_install.append(app_info['http_path'])
apps_to_install.append(app_file_container_path)
elif "http_path" in app_info:
apps_to_install.append(app_info["http_path"])
else:
if use_splunkbase is True:
print("Error, the app %s: %s could not be installed from Splunkbase because "
"--splunkbase_username and.or --splunkbase_password were not provided."
"\n\tQuitting..."%(app_name,app_info), file=sys.stderr)
print(
"Error, the app %s: %s could not be installed from Splunkbase because "
"--splunkbase_username and.or --splunkbase_password were not provided."
"\n\tQuitting..." % (app_name, app_info),
file=sys.stderr,
)
else:
print("Error, the app %s: %s has no http_path or local_path.\n\tQuitting..."%(app_name,app_info), file=sys.stderr)
print(
"Error, the app %s: %s has no http_path or local_path.\n\tQuitting..."
% (app_name, app_info),
file=sys.stderr,
)
sys.exit(1)
return ",".join(apps_to_install), require_credentials
def make_environment(
@@ -131,13 +142,12 @@ class SplunkContainer:
splunk_apps_url, require_credentials = self.prepare_apps_path(
apps, splunkbase_username, splunkbase_password
)
if require_credentials:
env["SPLUNKBASE_USERNAME"] = splunkbase_username
env["SPLUNKBASE_PASSWORD"] = splunkbase_password
env["SPLUNK_APPS_URL"] = splunk_apps_url
return env
def make_ports(self, *ports: tuple[str, int]) -> dict[str, int]:
@@ -203,26 +213,25 @@ class SplunkContainer:
)
successful_copy = True
except Exception as e:
#print("Failed copy of [%s] file to [%s] on CONTAINER [%s]: [%s]\n...we will try again"%(local_file_path, container_file_path, self.container_name, str(e)))
# print("Failed copy of [%s] file to [%s] on CONTAINER [%s]: [%s]\n...we will try again"%(local_file_path, container_file_path, self.container_name, str(e)))
time.sleep(10)
successful_copy = False
#print("Successfully copied [%s] to [%s] on [%s]"% (local_file_path, container_file_path, self.container_name))
# print("Successfully copied [%s] to [%s] on [%s]"% (local_file_path, container_file_path, self.container_name))
return successful_copy
def stopContainer(self,timeout=10) -> bool:
try:
def stopContainer(self, timeout=10) -> bool:
try:
container = self.client.containers.get(self.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
# 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)
self.synchronization_object.containerFailure()
return True
except Exception as e:
# Container does not exist, or we could not get it. Throw and error
print("Error stopping docker container [%s]"%(self.container_name))
print("Error stopping docker container [%s]" % (self.container_name))
return False
def removeContainer(
self, removeVolumes: bool = True, forceRemove: bool = True
@@ -241,8 +250,7 @@ class SplunkContainer:
# No need to print that the container has been removed, it is expected behavior
return True
except Exception as e:
print("Could not remove Docker Container [%s]" % (
self.container_name))
print("Could not remove Docker Container [%s]" % (self.container_name))
raise (Exception(f"CONTAINER REMOVE ERROR: {str(e)}"))
def get_container_summary(self) -> str:
@@ -253,7 +261,8 @@ class SplunkContainer:
total_time_string = "NOT STARTED"
else:
total_time_rounded = datetime.timedelta(
seconds=round(current_time - self.container_start_time))
seconds=round(current_time - self.container_start_time)
)
total_time_string = str(total_time_rounded)
# Time that the container setup took
@@ -261,7 +270,8 @@ class SplunkContainer:
setup_time_string = "NOT SET UP"
else:
setup_secounds_rounded = datetime.timedelta(
seconds=round(self.test_start_time - self.container_start_time))
seconds=round(self.test_start_time - self.container_start_time)
)
setup_time_string = str(setup_secounds_rounded)
# Time that the tests have been running
@@ -269,23 +279,34 @@ class SplunkContainer:
testing_time_string = "NO TESTS COMPLETED"
else:
testing_seconds_rounded = datetime.timedelta(
seconds=round(current_time - self.test_start_time))
seconds=round(current_time - self.test_start_time)
)
# Get the approximate time per test. This is a clunky way to get rid of decimal
# seconds.... but it works
timedelta_per_test = testing_seconds_rounded/self.num_tests_completed
timedelta_per_test_rounded = timedelta_per_test - \
datetime.timedelta(
microseconds=timedelta_per_test.microseconds)
timedelta_per_test = testing_seconds_rounded / self.num_tests_completed
timedelta_per_test_rounded = timedelta_per_test - datetime.timedelta(
microseconds=timedelta_per_test.microseconds
)
testing_time_string = "%s (%d tests @ %s per test)" % (
testing_seconds_rounded, self.num_tests_completed, timedelta_per_test_rounded)
testing_seconds_rounded,
self.num_tests_completed,
timedelta_per_test_rounded,
)
summary_str = "Summary for %s\n\t"\
"Total Time : [%s]\n\t"\
"Container Start Time: [%s]\n\t"\
"Test Execution Time : [%s]\n" % (
self.container_name, total_time_string, setup_time_string, testing_time_string)
summary_str = (
"Summary for %s\n\t"
"Total Time : [%s]\n\t"
"Container Start Time: [%s]\n\t"
"Test Execution Time : [%s]\n"
% (
self.container_name,
total_time_string,
setup_time_string,
testing_time_string,
)
)
return summary_str
@@ -293,59 +314,56 @@ class SplunkContainer:
self,
seconds_between_attempts: int = 10,
) -> bool:
# The smarter version of this will try to hit one of the pages,
# probably the login page, and when that is available it means that
# splunk is fully started and ready to go. Until then, we just
# use a simple sleep
while True:
try:
service = splunk_sdk.client.connect(host=self.splunk_ip, port=self.management_port, username='admin', password=self.container_password)
service = splunk_sdk.client.connect(
host=self.splunk_ip,
port=self.management_port,
username="admin",
password=self.container_password,
)
if service.restart_required:
#The sleep below will wait
# The sleep below will wait
pass
else:
return True
except Exception as e:
# There is a good chance the server is restarting, so the SDK connection failed.
# Or, we tried to check restart_required while the server was restarting. In the
# calling function, we have a timeout, so it's okay if this function could get
# calling function, we have a timeout, so it's okay if this function could get
# stuck in an infinite loop (the caller will generate a timeout error)
pass
time.sleep(seconds_between_attempts)
#@wrapt_timeout_decorator.timeout(MAX_CONTAINER_START_TIME_SECONDS, timeout_exception=RuntimeError)
# @wrapt_timeout_decorator.timeout(MAX_CONTAINER_START_TIME_SECONDS, timeout_exception=RuntimeError)
def setup_container(self):
self.container.start()
# def shutdown_signal_handler(sig, frame):
# shutdown_client = docker.client.from_env()
# errorCount = 0
# print(f"Shutting down {self.container_name}...", file=sys.stderr)
# try:
# container = shutdown_client.containers.get(self.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
# #so stopping can be useful if we want to debug any container failure
# container.stop(timeout=10)
# print(f"{self.container_name} shut down successfully", file=sys.stderr)
# print(f"{self.container_name} shut down successfully", file=sys.stderr)
# except Exception as e:
# print(f"Error trying to shut down {self.container_name}. It may have already shut down. Stop it youself with 'docker containter stop {self.container_name}", 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!
# import os
# os._exit(1)
# import signal
# signal.signal(signal.SIGINT, shutdown_signal_handler)
@@ -357,29 +375,30 @@ class SplunkContainer:
print("Finished copying files to [%s]" % (self.container_name))
self.wait_for_splunk_ready()
def successfully_finish_tests(self)->None:
def successfully_finish_tests(self) -> None:
try:
if self.num_tests_completed == 0:
print("Container [%s] did not find any tests and will not start.\n"\
"This does not mean there was an error!"%(self.container_name))
print(
"Container [%s] did not find any tests and will not start.\n"
"This does not mean there was an error!" % (self.container_name)
)
else:
print("Container [%s] has finished running [%d] detections, time to stop the container."
% (self.container_name, self.num_tests_completed))
print(
"Container [%s] has finished running [%d] detections, time to stop the container."
% (self.container_name, self.num_tests_completed)
)
# remove the container
self.removeContainer()
except Exception as e:
print(
"Error stopping or removing the container: [%s]" % (str(e)))
print("Error stopping or removing the container: [%s]" % (str(e)))
return None
def run_container(self) -> None:
print("Starting the container [%s]" % (self.container_name))
# Try to get something from the queue. Check this early on
# before launching the container because it can save us a lot of time!
detection_to_test = self.synchronization_object.getTest()
@@ -387,45 +406,48 @@ class SplunkContainer:
return self.successfully_finish_tests()
self.container_start_time = timeit.default_timer()
container_start_time = timeit.default_timer()
try:
self.setup_container()
except Exception as e:
print("There was an exception starting the container [%s]: [%s]. Shutting down container"%(self.container_name,str(e)),file=sys.stdout)
print(
"There was an exception starting the container [%s]: [%s]. Shutting down container"
% (self.container_name, str(e)),
file=sys.stdout,
)
self.stopContainer()
elapsed_rounded = round(timeit.default_timer() - container_start_time)
time_string = (datetime.timedelta(seconds=elapsed_rounded))
print("Container [%s] FAILED in [%s]"%(self.container_name, time_string))
time_string = datetime.timedelta(seconds=elapsed_rounded)
print("Container [%s] FAILED in [%s]" % (self.container_name, time_string))
return None
#GTive some info about how long the container took to start up
# GTive some info about how long the container took to start up
elapsed_rounded = round(timeit.default_timer() - container_start_time)
time_string = (datetime.timedelta(seconds=elapsed_rounded))
print("Container [%s] took [%s] to start"%(self.container_name, time_string))
time_string = datetime.timedelta(seconds=elapsed_rounded)
print("Container [%s] took [%s] to start" % (self.container_name, time_string))
self.synchronization_object.start_barrier.wait()
# Sleep for a small random time so that containers drift apart and don't synchronize their testing
time.sleep(random.randint(1, 30))
self.test_start_time = timeit.default_timer()
while detection_to_test is not None:
if self.synchronization_object.checkContainerFailure():
self.container.stop()
print("Container [%s] successfully stopped early due to failure" % (self.container_name))
print(
"Container [%s] successfully stopped early due to failure"
% (self.container_name)
)
return None
current_test_start_time = timeit.default_timer()
# Sleep for a small random time so that containers drift apart and don't synchronize their testing
#time.sleep(random.randint(1, 30))
# time.sleep(random.randint(1, 30))
# There is a detection to test
print("Container [%s]--->[%s]" %
(self.container_name, detection_to_test))
print("Container [%s]--->[%s]" % (self.container_name, detection_to_test))
try:
result = testing_service.test_detection_wrapper(
self.container_name,
@@ -435,36 +457,50 @@ class SplunkContainer:
detection_to_test,
self.synchronization_object.attack_data_root_folder,
wait_on_failure=self.interactive_failure,
wait_on_completion = self.interactive
wait_on_completion=self.interactive,
smoketest=self.synchronization_object.summarization_reproduce_failure_config[
"mode"
]
== "smoketest",
)
self.synchronization_object.addResult(
result,
duration_string=str(
datetime.timedelta(
seconds=round(
timeit.default_timer() - current_test_start_time
)
)
),
)
self.synchronization_object.addResult(result, duration_string = str(datetime.timedelta(seconds=round(timeit.default_timer() - current_test_start_time))))
# Remove the data from the test that we just ran. We MUST do this when running on CI because otherwise, we will download
# a massive amount of data over the course of a long path and will run out of space on the relatively small CI runner drive
shutil.rmtree(result["attack_data_directory"],ignore_errors=True)
shutil.rmtree(result["attack_data_directory"], ignore_errors=True)
except Exception as e:
import traceback
print(
"Warning - uncaught error in detection test for [%s] - this should not happen: [%s]"
% (detection_to_test, str(e))
)
print(traceback.print_exc())
self.synchronization_object.addError(
{"detection_file": detection_to_test,
"detection_error": str(e)}, duration_string = str(datetime.timedelta(seconds=round(timeit.default_timer() - current_test_start_time)))
{"detection_file": detection_to_test, "detection_error": str(e)},
duration_string=str(
datetime.timedelta(
seconds=round(
timeit.default_timer() - current_test_start_time
)
)
),
)
self.num_tests_completed += 1
# Try to get something from the queue
detection_to_test = self.synchronization_object.getTest()
#We failed to get a test from the queue, so we must be done gracefully! Quit
return self.successfully_finish_tests()
# We failed to get a test from the queue, so we must be done gracefully! Quit
return self.successfully_finish_tests()
bin/docker_detection_tester/modules/testing_service.py
+222 -131
View File
@@ -1,7 +1,6 @@
import re
#import ansible_runner
# import ansible_runner
import yaml
import uuid
import sys
@@ -19,185 +18,277 @@ import datetime
import http.client
def test_detection_wrapper(container_name:str, splunk_ip:str, splunk_password:str, splunk_port:int,
detection_file:str, attack_data_root_folder, wait_on_failure:bool=False, wait_on_completion:bool=False)->dict:
def test_detection_wrapper(
container_name: str,
splunk_ip: str,
splunk_password: str,
splunk_port: int,
detection_file: str,
attack_data_root_folder,
wait_on_failure: bool = False,
wait_on_completion: bool = False,
smoketest: bool = False,
) -> dict:
one_test_start = timeit.default_timer()
uuid_var = str(uuid.uuid4())
result_test, indices_to_delete = test_detection(splunk_ip, splunk_port, container_name, splunk_password, detection_file, uuid_var, attack_data_root_folder)
result_test, indices_to_delete = test_detection(
splunk_ip,
splunk_port,
container_name,
splunk_password,
detection_file,
uuid_var,
attack_data_root_folder,
smoketest,
)
one_test_stop = timeit.default_timer()
if result_test is None:
#We failed so early in the process that we could not produce any meaningful result
raise(Exception("Test execution Error"))
# We failed so early in the process that we could not produce any meaningful result
raise (Exception("Test execution Error"))
#enter = input("Run some tests from [%s] on [%s] - we don't delete until you hit enter :)"%(container_name, test_file))
# enter = input("Run some tests from [%s] on [%s] - we don't delete until you hit enter :)"%(container_name, test_file))
# delete test data
search_string = result_test['detection_result']['search_string']
#get pretty time info
elapsed_search_time_string = str(datetime.timedelta(seconds=round(one_test_stop - one_test_start)))
search_string = result_test["detection_result"]["search_string"]
#search failed if there was an error or the detection failed to produce the expected result
#print("Elapsed search time: %s"%(elapsed_search_time_string))
if (wait_on_failure or wait_on_completion) and (result_test['detection_result']['error'] or not result_test['detection_result']['success']):
wait_on_delete = {'message':"\n\n\n****SEARCH FAILURE : Allowing time to debug search/data****"}
# get pretty time info
elapsed_search_time_string = str(
datetime.timedelta(seconds=round(one_test_stop - one_test_start))
)
# search failed if there was an error or the detection failed to produce the expected result
# print("Elapsed search time: %s"%(elapsed_search_time_string))
if (wait_on_failure or wait_on_completion) and (
result_test["detection_result"]["error"]
or not result_test["detection_result"]["success"]
):
wait_on_delete = {
"message": "\n\n\n****SEARCH FAILURE : Allowing time to debug search/data****"
}
elif wait_on_completion:
wait_on_delete = {'message':"\n\n\n****SEARCH SUCCESS : Allowing time to examine search/data****"}
wait_on_delete = {
"message": "\n\n\n****SEARCH SUCCESS : Allowing time to examine search/data****"
}
else:
wait_on_delete = None
splunk_sdk.delete_attack_data(
splunk_ip,
splunk_password,
splunk_port,
wait_on_delete,
search_string,
detection_file,
indices=indices_to_delete,
)
splunk_sdk.delete_attack_data(splunk_ip, splunk_password, splunk_port, wait_on_delete, search_string, detection_file, indices = indices_to_delete)
return result_test
return result_test
import splunklib.client as client
def get_service(splunk_ip:str, splunk_port:int, splunk_password:str):
def get_service(splunk_ip: str, splunk_port: int, splunk_password: str):
try:
service = client.connect(
host=splunk_ip,
port=splunk_port,
username='admin',
password=splunk_password
host=splunk_ip, port=splunk_port, username="admin", password=splunk_password
)
except Exception as e:
raise(Exception("Unable to connect to Splunk instance: " + str(e)))
raise (Exception("Unable to connect to Splunk instance: " + str(e)))
return service
def test_detection(splunk_ip:str, splunk_port:int, container_name:str, splunk_password:str, detection_file:str, uuid_var, attack_data_root_folder)->Tuple[Union[dict,None], set[str]]:
def test_detection(
splunk_ip: str,
splunk_port: int,
container_name: str,
splunk_password: str,
detection_file: str,
uuid_var,
attack_data_root_folder,
smoketest: bool,
) -> Tuple[Union[dict, None], set[str]]:
detection_file_obj = load_file(os.path.join("security_content/", detection_file))
if not detection_file_obj:
print("Not detection_file_obj!")
raise(Exception("No test file object found for [%s]"%detection_file))
#print(test_file_obj)
raise (Exception("No test file object found for [%s]" % detection_file))
# write entry dynamodb
#aws_service.add_detection_results_in_dynamo_db('eu-central-1', uuid_var , uuid_test, test_file_obj['tests'][0]['name'], test_file_obj['tests'][0]['file'], str(int(time.time())))
#epoch_time = str(int(time.time()))
abs_folder_path = mkdtemp(prefix="DATA_", dir=attack_data_root_folder)
#We want the relative path, so we convert it as required
tests:dict = detection_file_obj.get("tests", {})
if len(tests) > 1:
print(f"****WARNING - THIS DETECTION CONTAINS {len(tests)} TESTS BUT WE WILL ONLY RUN 1")
test = tests[0]
indices_to_delete = set()
abs_folder_path = mkdtemp(prefix="DATA_", dir=attack_data_root_folder)
if smoketest:
result_detection = splunk_sdk.test_detection_search(
splunk_ip,
splunk_port,
splunk_password,
detection_file_obj["search"],
"",
detection_file_obj["name"],
detection_file,
"-24h",
"now",
attempts_remaining=1,
)
for attack_data in test['attack_data']:
url = attack_data['data']
if 'custom_index' in attack_data:
print(f"Found a custom index for {detection_file}: {attack_data['custom_index']}")
data_upload_index = attack_data['custom_index']
else:
data_upload_index = splunk_sdk.DEFAULT_DATA_INDEX
result_test = {}
test = {"name": detection_file_obj["name"] + " Smoketest"}
result_test["baselines_result"] = []
else:
# print(test_file_obj)
indices_to_delete.add(data_upload_index)
_, target_file = mkstemp(prefix="attack_data_", dir=abs_folder_path)
utils.download_file_from_http(url, target_file, overwrite_file=True)
# write entry dynamodb
# aws_service.add_detection_results_in_dynamo_db('eu-central-1', uuid_var , uuid_test, test_file_obj['tests'][0]['name'], test_file_obj['tests'][0]['file'], str(int(time.time())))
# epoch_time = str(int(time.time()))
# Update timestamps before replay
if 'update_timestamp' in attack_data:
if attack_data['update_timestamp'] == True:
data_manipulation = DataManipulation()
data_manipulation.manipulate_timestamp(target_file, attack_data['sourcetype'], attack_data['source'])
#replay_attack_dataset(container_name, splunk_password, folder_name, "test0", attack_data['sourcetype'], attack_data['source'], attack_data['file_name'])
try:
service = get_service(splunk_ip, splunk_port, splunk_password)
test_index = service.indexes[data_upload_index]
with open(target_file, 'rb') as target:
test_index.submit(target.read(), sourcetype=attack_data['sourcetype'], source=attack_data['source'], host=splunk_sdk.DEFAULT_EVENT_HOST)
except http.client.HTTPException as e:
raise(Exception(f"Failed to submit detection file {target_file} to Splunk Server: {str(e)}"))
except Exception as e:
raise(Exception(f"Failed to submit detection file {target_file} to Splunk Server: {str(e)}"))
# We want the relative path, so we convert it as required
tests: dict = detection_file_obj.get("tests", {})
if len(tests) > 1:
print(
f"****WARNING - THIS DETECTION CONTAINS {len(tests)} TESTS BUT WE WILL ONLY RUN 1"
)
test = tests[0]
if not splunk_sdk.wait_for_indexing_to_complete(splunk_ip, splunk_port, splunk_password, attack_data['sourcetype'], data_upload_index):
raise Exception("There was an error waiting for indexing to complete.")
#Allow some time for the data to be ingested and processed
#print("begin sleep 30")
#time.sleep(60)
for attack_data in test["attack_data"]:
url = attack_data["data"]
#print("end sleep 30")
result_test = {}
if "custom_index" in attack_data:
print(
f"Found a custom index for {detection_file}: {attack_data['custom_index']}"
)
data_upload_index = attack_data["custom_index"]
else:
data_upload_index = splunk_sdk.DEFAULT_DATA_INDEX
if 'baselines' in test:
results_baselines = []
for baseline_obj in test['baselines']:
baseline_file_name = baseline_obj['file']
baseline = load_file(os.path.join(os.path.dirname(__file__), '../security_content', baseline_file_name))
result_obj = dict()
result_obj['baseline'] = baseline_obj['name']
result_obj['baseline_file'] = baseline_file_name
print("Making test_baseline_search request to: [%s:%d]"%(splunk_ip, splunk_port))
result = splunk_sdk.test_baseline_search(splunk_ip, splunk_port, splunk_password, baseline['search'], baseline_obj['pass_condition'], baseline['name'], baseline_file_name, baseline_obj['earliest_time'], baseline_obj['latest_time'])
#we don't seem to be doing anything with this loop... are we supposed to have the following line belwo?
results_baselines.append(result)
indices_to_delete.add(data_upload_index)
result_test['baselines_result'] = results_baselines
_, target_file = mkstemp(prefix="attack_data_", dir=abs_folder_path)
result_detection = splunk_sdk.test_detection_search(splunk_ip, splunk_port, splunk_password, detection_file_obj['search'], test.get('pass_condition', '| stats count | where count > 0'), detection_file_obj['name'], detection_file, test.get('earliest_time', '-24h'), test.get('latest_time', 'now'))
if result_detection['error']:
print("There was an error running the search: %s"%(result_detection['search_string']))
utils.download_file_from_http(url, target_file, overwrite_file=True)
# Update timestamps before replay
if "update_timestamp" in attack_data:
if attack_data["update_timestamp"] == True:
data_manipulation = DataManipulation()
data_manipulation.manipulate_timestamp(
target_file, attack_data["sourcetype"], attack_data["source"]
)
# replay_attack_dataset(container_name, splunk_password, folder_name, "test0", attack_data['sourcetype'], attack_data['source'], attack_data['file_name'])
try:
service = get_service(splunk_ip, splunk_port, splunk_password)
test_index = service.indexes[data_upload_index]
with open(target_file, "rb") as target:
test_index.submit(
target.read(),
sourcetype=attack_data["sourcetype"],
source=attack_data["source"],
host=splunk_sdk.DEFAULT_EVENT_HOST,
)
result_detection['detection_name'] = test['name']
result_detection['detection_file'] = detection_file
result_test['detection_result'] = result_detection
result_test['attack_data_directory'] = abs_folder_path
except http.client.HTTPException as e:
raise (
Exception(
f"Failed to submit detection file {target_file} to Splunk Server: {str(e)}"
)
)
except Exception as e:
raise (
Exception(
f"Failed to submit detection file {target_file} to Splunk Server: {str(e)}"
)
)
if not splunk_sdk.wait_for_indexing_to_complete(
splunk_ip,
splunk_port,
splunk_password,
attack_data["sourcetype"],
data_upload_index,
):
raise Exception("There was an error waiting for indexing to complete.")
# Allow some time for the data to be ingested and processed
# print("begin sleep 30")
# time.sleep(60)
# print("end sleep 30")
result_test = {}
if "baselines" in test:
results_baselines = []
for baseline_obj in test["baselines"]:
baseline_file_name = baseline_obj["file"]
baseline = load_file(
os.path.join(
os.path.dirname(__file__),
"../security_content",
baseline_file_name,
)
)
result_obj = dict()
result_obj["baseline"] = baseline_obj["name"]
result_obj["baseline_file"] = baseline_file_name
print(
"Making test_baseline_search request to: [%s:%d]"
% (splunk_ip, splunk_port)
)
result = splunk_sdk.test_baseline_search(
splunk_ip,
splunk_port,
splunk_password,
baseline["search"],
baseline_obj["pass_condition"],
baseline["name"],
baseline_file_name,
baseline_obj["earliest_time"],
baseline_obj["latest_time"],
)
# we don't seem to be doing anything with this loop... are we supposed to have the following line belwo?
results_baselines.append(result)
result_test["baselines_result"] = results_baselines
result_detection = splunk_sdk.test_detection_search(
splunk_ip,
splunk_port,
splunk_password,
detection_file_obj["search"],
test.get("pass_condition", "| stats count | where count > 0"),
detection_file_obj["name"],
detection_file,
test.get("earliest_time", "-24h"),
test.get("latest_time", "now"),
)
if result_detection["error"]:
print(
"There was an error running the search: %s"
% (result_detection["search_string"])
)
result_detection["detection_name"] = test["name"]
result_detection["detection_file"] = detection_file
result_test["detection_result"] = result_detection
result_test["attack_data_directory"] = abs_folder_path
return result_test, indices_to_delete
def load_file(file_path):
try:
with open(file_path, 'r', encoding="utf-8") as stream:
with open(file_path, "r", encoding="utf-8") as stream:
try:
file = list(yaml.safe_load_all(stream))[0]
except yaml.YAMLError as exc:
raise(Exception("ERROR: parsing YAML for {0}:[{1}]".format(file_path, str(exc))))
raise (
Exception(
"ERROR: parsing YAML for {0}:[{1}]".format(file_path, str(exc))
)
)
except Exception as e:
raise(Exception("ERROR: opening {0}:[{1}]".format(file_path, str(e))))
raise (Exception("ERROR: opening {0}:[{1}]".format(file_path, str(e))))
return file
bin/docker_detection_tester/modules/validate_args.py
+7 -8
View File
@@ -142,14 +142,14 @@ setup_schema = {
"URL_TOOLBOX": {
"app_number": 2734,
"app_version": "1.9.2",
"http_path": "https://attack-range-appbinaries.s3.us-west-2.amazonaws.com/Latest/url-toolbox_192.tgz"
},
"SPLUNK_TA_FIX_WINDOWS":{
"http_path": "https://attack-range-appbinaries.s3.us-west-2.amazonaws.com/Latest/url-toolbox_192.tgz",
},
"SPLUNK_TA_FIX_WINDOWS": {
"app_number": 9999,
"app_version": "1.0.0",
"http_path": "https://attack-range-appbinaries.s3.us-west-2.amazonaws.com/Latest/Splunk_TA_fix_windows.tgz"
},
"SPLUNK_TA_MICROSOFT_CLOUD_SERVICES": {
"http_path": "https://attack-range-appbinaries.s3.us-west-2.amazonaws.com/Latest/Splunk_TA_fix_windows.tgz",
},
"SPLUNK_TA_MICROSOFT_CLOUD_SERVICES": {
"app_number": 3110,
"app_version": "4.5.2",
"http_path": "https://attack-range-appbinaries.s3.us-west-2.amazonaws.com/Latest/splunk-add-on-for-microsoft-cloud-services_452.tgz",
@@ -168,7 +168,7 @@ setup_schema = {
},
"mode": {
"type": "string",
"enum": ["changes", "selected", "all"],
"enum": ["changes", "selected", "all", "smoketest"],
"default": "changes",
},
"num_containers": {"type": "integer", "minimum": 1, "default": 1},
@@ -318,7 +318,6 @@ def validate(
# v = jsonschema.Draft201909Validator(argument_schema)
try:
validation_errors, validated_json = jsonschema_errorprinter.check_json(
configuration, setup_schema
)
bin/docker_detection_tester/summarize_json.py
+178 -87
View File
@@ -8,149 +8,240 @@ import os.path
from operator import itemgetter
import copy
def outputResultsJSON(output_filename:str, data:list[dict], baseline:OrderedDict,
failure_manifest_filename = "detection_failure_manifest.json",
output_folder:str="", summarization_reproduce_failure_config:dict={})->tuple[bool,int,int,int,int]:
def outputResultsJSON(
output_filename: str,
data: list[dict],
baseline: OrderedDict,
failure_manifest_filename="detection_failure_manifest.json",
output_folder: str = "",
summarization_reproduce_failure_config: dict = {},
) -> tuple[bool, int, int, int, int]:
success = True
try:
test_count = len(data)
#Passed
pass_count = len([x for x in data if x['success'] == True])
#A failure or an error
fail_count = len([x for x in data if x['success'] == False])
#An error (every error is also a failure)
fail_and_error_count = len([x for x in data if x['error'] == True])
#A failure without an error
fail_without_error_count = len([x for x in data if x['success'] == False and x['error'] == False])
#This number should always be zero...
error_and_success_count = len([x for x in data if x['success'] == True and x['error'] == True])
# Passed
pass_count = len([x for x in data if x["success"] == True])
# A failure or an error
fail_count = len([x for x in data if x["success"] == False])
# An error (every error is also a failure)
fail_and_error_count = len([x for x in data if x["error"] == True])
# A failure without an error
fail_without_error_count = len(
[x for x in data if x["success"] == False and x["error"] == False]
)
# This number should always be zero...
error_and_success_count = len(
[x for x in data if x["success"] == True and x["error"] == True]
)
if error_and_success_count > 0:
print("Error - a test was successful, but also included an error. This should be impossible.",file=sys.stderr)
print(
"Error - a test was successful, but also included an error. This should be impossible.",
file=sys.stderr,
)
success = False
if test_count != (pass_count + fail_count):
print("Error - the total tests [%d] does not equal the pass[%d]/fails[%d]"%(test_count, pass_count,fail_count))
success=False
print(
"Error - the total tests [%d] does not equal the pass[%d]/fails[%d]"
% (test_count, pass_count, fail_count)
)
success = False
if fail_count > 0:
result = "FAIL for %d detections"%(fail_count)
result = "FAIL for %d detections" % (fail_count)
success = False
else:
result = "PASS for all %d detections"%(pass_count)
result = "PASS for all %d detections" % (pass_count)
summary = {
"TOTAL_TESTS": test_count,
"TESTS_PASSED": pass_count,
"TOTAL_FAILURES": fail_count,
"FAIL_ONLY": fail_without_error_count,
"PASS_RATE": calculate_pass_rate(pass_count, test_count),
"FAIL_AND_ERROR": fail_and_error_count,
}
summary={"TOTAL_TESTS": test_count, "TESTS_PASSED": pass_count,
"TOTAL_FAILURES": fail_count, "FAIL_ONLY": fail_without_error_count,
"PASS_RATE": calculate_pass_rate(pass_count, test_count),
"FAIL_AND_ERROR":fail_and_error_count }
data_sorted = sorted(
data, key=lambda k: (-k["error"], k["success"], k["detection_file"])
)
with open(os.path.join(output_folder, output_filename), "w") as jsonFile:
json.dump(
{"summary": summary, "baseline": baseline, "results": data_sorted},
jsonFile,
indent=" ",
)
data_sorted = sorted(data, key = lambda k: (-k['error'], k['success'], k['detection_file']))
with open(os.path.join(output_folder,output_filename), "w") as jsonFile:
json.dump({'summary':summary, 'baseline': baseline, 'results':data_sorted}, jsonFile, indent=" ")
#Generate a failure that the user can download to reproduce and test ONLY the failures locally.
#This makes it easy to test and debug ONLY those that failed. No need to test the ones
#that succeeded!
fail_list = [os.path.join("security_content/detections",x['detection_file'] ) for x in data_sorted if x['success'] == False]
# Generate a failure that the user can download to reproduce and test ONLY the failures locally.
# This makes it easy to test and debug ONLY those that failed. No need to test the ones
# that succeeded!
fail_list = [
os.path.join("security_content/detections", x["detection_file"])
for x in data_sorted
if x["success"] == False
]
if len(fail_list) > 0:
print("FAILURES:")
for failed_test in fail_list:
print(f"\t{failed_test}")
failures_test_override = copy.deepcopy(summarization_reproduce_failure_config)
#Force all tests to be interactive, even if they don't fail (because they failed on this test)
failures_test_override.update({"detections_list": fail_list, "no_interactive_failure":False, "interactive": True,
"num_containers":1, "branch": baseline["branch"], "commit_hash":baseline["commit_hash"],
"mode":"selected", "show_splunk_app_password": True})
with open(os.path.join(output_folder,failure_manifest_filename),"w") as failures:
failures_test_override = copy.deepcopy(
summarization_reproduce_failure_config
)
# Force all tests to be interactive, even if they don't fail (because they failed on this test)
failures_test_override.update(
{
"detections_list": fail_list,
"no_interactive_failure": False,
"interactive": True,
"num_containers": 1,
"branch": baseline["branch"],
"commit_hash": baseline["commit_hash"],
"mode": "selected",
"show_splunk_app_password": True,
}
)
with open(
os.path.join(output_folder, failure_manifest_filename), "w"
) as failures:
validate_args.validate_and_write(failures_test_override, failures)
except Exception as e:
print("There was an error generating [%s]: [%s]"%(output_filename, str(e)),file=sys.stderr)
print(
"There was an error generating [%s]: [%s]" % (output_filename, str(e)),
file=sys.stderr,
)
print(data)
raise(e)
#success = False
#return success, False
raise (e)
# success = False
# return success, False
#note that total failures is fail_count, fail_and_error count is JUST errors (and every error is also a failure)
# note that total failures is fail_count, fail_and_error count is JUST errors (and every error is also a failure)
return success, test_count, pass_count, fail_count, fail_and_error_count
def calculate_pass_rate(pass_count:int, test_count:int)->float:
def calculate_pass_rate(pass_count: int, test_count: int) -> float:
if test_count == 0:
#Assume this means 100% pass rate to avoid divide by zero
# Assume this means 100% pass rate to avoid divide by zero
pass_rate = 1
else:
pass_rate = pass_count / test_count
return pass_rate
def print_summary(test_count: int, pass_count:int, fail_count:int, error_count:int)->None:
print("Summary:"\
f"\n\tTotal Tests: {test_count}"\
f"\n\tTotal Pass : {pass_count}"\
f"\n\tTotal Fail : {fail_count} ({error_count} of these were ERRORS))"\
f"\n\tPass Rate : {calculate_pass_rate(pass_count, test_count):.3f}")
def exit_with_status(test_pass:bool, test_count: int, pass_count:int, fail_count:int, error_count:int)->None:
def print_summary(
test_count: int, pass_count: int, fail_count: int, error_count: int
) -> None:
print(
"Summary:"
f"\n\tTotal Tests: {test_count}"
f"\n\tTotal Pass : {pass_count}"
f"\n\tTotal Fail : {fail_count} ({error_count} of these were ERRORS))"
f"\n\tPass Rate : {calculate_pass_rate(pass_count, test_count):.3f}"
)
def exit_with_status(
test_pass: bool, test_count: int, pass_count: int, fail_count: int, error_count: int
) -> None:
if not test_pass:
print("Result: FAIL")
#print("DURING TESTING, THIS WILL STILL EXIT WITH AN EXIT CODE OF 0 (SUCCESS) TO ALLOW THE WORKFLOW "
# print("DURING TESTING, THIS WILL STILL EXIT WITH AN EXIT CODE OF 0 (SUCCESS) TO ALLOW THE WORKFLOW "
# "TO PASS AND CI/CD TO CONTINUE. THIS WILL BE CHANGED IN A FUTURE VERSION.")
#sys.exit(0)
# sys.exit(0)
sys.exit(1)
else:
print("Result: PASS!")
sys.exit(0)
def finish(test_pass:bool, test_count: int, pass_count:int, fail_count:int, error_count:int)->None:
print_summary(test_count, pass_count, fail_count,error_count)
exit_with_status(test_pass, test_count, pass_count, fail_count,error_count)
def finish(
test_pass: bool, test_count: int, pass_count: int, fail_count: int, error_count: int
) -> None:
print_summary(test_count, pass_count, fail_count, error_count)
exit_with_status(test_pass, test_count, pass_count, fail_count, error_count)
def main():
parser = argparse.ArgumentParser(description="Results Merger")
parser.add_argument('-f', '--files', type=argparse.FileType('r'), required=True, nargs='+', help="The json files you would like to combine into a single file")
parser.add_argument('-o', '--output_filename', type=str, required=True, help="The name of the output file")
parser.add_argument(
"-f",
"--files",
type=argparse.FileType("r"),
required=True,
nargs="+",
help="The json files you would like to combine into a single file",
)
parser.add_argument(
"-o",
"--output_filename",
type=str,
required=True,
help="The name of the output file",
)
parser.add_argument("--smoketest", action=argparse.BooleanOptionalAction)
args = parser.parse_args()
all_data = OrderedDict()
try:
print("We will summarize the files: %s"%(str([f.name for f in args.files])))
print("We will summarize the files: %s" % (str([f.name for f in args.files])))
for f in args.files:
if not f.name.endswith('.json'):
print("Error: passed in file must end in .json - you passed in [%s].\n\tQuitting..."%(f.name))
if not f.name.endswith(".json"):
print(
"Error: passed in file must end in .json - you passed in [%s].\n\tQuitting..."
% (f.name)
)
sys.exit(1)
data = json.loads(f.read())
if 'baseline' in all_data:
#everything has the same baseline, only need to do it once
if "baseline" in all_data:
# everything has the same baseline, only need to do it once
pass
else:
all_data['baseline'] = data['baseline']
if 'results' in all_data:
#this is a list of dictionaries, so add to it
all_data['results'].extend(data['results'])
all_data["baseline"] = data["baseline"]
if "results" in all_data:
# this is a list of dictionaries, so add to it
all_data["results"].extend(data["results"])
else:
all_data['results'] = data['results']
all_data["results"] = data["results"]
test_pass, test_count, pass_count, fail_count, error_count = outputResultsJSON(args.output_filename, all_data['results'], all_data['baseline'])
IGNORE_MESSAGES = [
"Model does not exist", # model not generated by baseline
"Data model 'Identity_Management' was not found", # missing datamodel included with es
"get_asset", # missing macro included with es
"Failed to load model", # when running a model that has not been downloaded separately
"UEBA", # Another missing ES asset
]
if args.smoketest:
new_results = []
for result in all_data["results"]:
if result.get("detection_error", None):
message = result.get("detection_error", None)
ignore = False
for ignore_message in IGNORE_MESSAGES:
if ignore_message in message:
ignore = True
break
if not ignore:
new_results.append(result)
pass
all_data["results"] = new_results
test_pass, test_count, pass_count, fail_count, error_count = outputResultsJSON(
args.output_filename, all_data["results"], all_data["baseline"]
)
finish(test_pass, test_count, pass_count, fail_count, error_count)
except Exception as e:
print("Error generating the summary file: [%s].\n\tQuitting..."%(str(e)))
print("Error generating the summary file: [%s].\n\tQuitting..." % (str(e)))
sys.exit(1)
if __name__=="__main__":
if __name__ == "__main__":
main()
detections/deprecated/prohibited_software_on_endpoint.yml
+20 -17
View File
@@ -1,19 +1,22 @@
name: Prohibited Software On Endpoint
id: a51bfe1a-94f0-48cc-b4e4-b6ae50145893
version: 2
date: '2019-10-11'
date: "2019-10-11"
author: David Dorsey, Splunk
status: deprecated
type: Hunting
description: This search looks for applications on the endpoint that you have marked
description:
This search looks for applications on the endpoint that you have marked
as prohibited.
data_source:
- Sysmon Event ID 1
search: '| tstats `security_content_summariesonly` count min(_time) as firstTime max(_time)
- Sysmon Event ID 1
search:
"| tstats `security_content_summariesonly` count min(_time) as firstTime max(_time)
as lastTime from datamodel=Endpoint.Processes by Processes.dest Processes.user Processes.process_name
| `security_content_ctime(firstTime)`| `security_content_ctime(lastTime)` | `drop_dm_object_name(Processes)`
| `prohibited_softwares` | `prohibited_software_on_endpoint_filter`'
how_to_implement: To successfully implement this search, you must be ingesting data
| `prohibited_processes` | `prohibited_software_on_endpoint_filter`"
how_to_implement:
To successfully implement this search, you must be ingesting data
that records process activity from your hosts to populate the endpoint data model
in the processes node. This is typically populated via endpoint detection-and-response
product, such as Carbon Black or endpoint data sources, such as Sysmon. The data
@@ -27,23 +30,23 @@ known_false_positives: None identified
references: []
tags:
analytic_story:
- Monitor for Unauthorized Software
- 'Emotet Malware DHS Report TA18-201A '
- SamSam Ransomware
- Monitor for Unauthorized Software
- "Emotet Malware DHS Report TA18-201A "
- SamSam Ransomware
asset_type: Endpoint
confidence: 50
impact: 50
message: tbd
observable:
- name: field
type: Unknown
role:
- Unknown
- name: field
type: Unknown
role:
- Unknown
product:
- Splunk Enterprise
- Splunk Enterprise Security
- Splunk Cloud
- Splunk Enterprise
- Splunk Enterprise Security
- Splunk Cloud
required_fields:
- _times
- _times
risk_score: 25
security_domain: endpoint
detections/endpoint/detect_suspicious_processnames_using_pretrained_model_in_dsdl.yml
+2 -2
View File
@@ -1,4 +1,4 @@
name: Detect suspicious processnames using a pretrained model in DSDL
name: Detect suspicious processnames using pretrained model in DSDL
id: a15f8977-ad7d-4669-92ef-b59b97219bf5
version: 1
date: "2023-01-23"
@@ -32,7 +32,7 @@ search: '| tstats `security_content_summariesonly` count min(_time) as firstTime
| rename predicted_label as is_suspicious_score
| rename text as process_name
| where is_suspicious_score > 0.5
| `detect_suspicious_processnames_using_a_pretrained_model_in_dsdl_filter`'
| `detect_suspicious_processnames_using_pretrained_model_in_dsdl_filter`'
how_to_implement: 'Steps to deploy detect suspicious processnames model into Splunk App
DSDL. This detection depends on the Splunk app for Data Science and Deep
lookups/prohibited_softwares.csv
-20
View File
@@ -1,20 +0,0 @@
app,note
remcom.exe,ESCU - This process is an open source replacement to psexec and is not typically seen in an enterprise environment.
pwdump.exe,ESCU - This process is associated with a tool used to dump password hashes on a Windows system.
pwdump2.exe,ESCU - This process is associated with a tool used to dump password hashes on a Windows system.
nc.exe,ESCU - This process is an open source tool used for network communications.
wce.exe,ESCU - This process is associated with a tool used to dump hashes and execute pass-the-hash and pass-the-ticket attacks.
cain.exe,ESCU - This process is associated with a tool used to collect user credentials and execute attacks.
nmap.exe,ESCU - This process is an open source network mapping tool used to identify hosts and listening services on a network.
kidlogger.exe,ESCU - This process is associated with a tool used to collect keyboard input on a host.
isass.exe,ESCU - This process name is used by attackers to hide in plain sight and look like a legitimate Windows system process.
svch0st.exe,ESCU - This process name is used by attackers to hide in plain sight and look like a legitimate Windows system process.
at.exe,ESCU - This process is used to schedule other processes to run. schtasks.exe should be used instead as it provides more flexibility.
getmail.exe,ESCU - This process is seen to be used by attackers to extract email files from host machines.
ntdll.exe,ESCU - This process was identified as malicious by DHS Alert TA18-074A.
netpass.exe,ESCU - This process was identified as malicious by DHS Alert TA18-201A and attackers use this tool to recover all network passwords stored on your system for the current logged-on user.
WebBrowserPassView.exe,ESCU - This process was identified as malicious by DHS Alert TA18-201A and is used by attackers as a password recovery tool that reveals the passwords stored in Web Browsers.
OutlookAddressBookView.exe,ESCU - This process was identified as malicious by DHS Alert TA18-201A and is used by attackers to steal the details of all recipients stored in the address books of Microsoft Outlook.
mailpv.exe,ESCU - This process was identified by DHS Alert TA18-201A and attackers use this tool is a password-recovery tool that reveals the passwords and other account details from various email clients.
NLBrute.exe,ESCU - This process was identified in the SamSam Ransomware Campaign and attackers use this tool to brute force RDP instances with a range of commonly used passwords.
selfdel.exe,ESCU - This executable was delivered in the SamSam Ransomware Campain and the attackers levereged this binary to delete its malicilous activities.
1 app note
2 remcom.exe ESCU - This process is an open source replacement to psexec and is not typically seen in an enterprise environment.
3 pwdump.exe ESCU - This process is associated with a tool used to dump password hashes on a Windows system.
4 pwdump2.exe ESCU - This process is associated with a tool used to dump password hashes on a Windows system.
5 nc.exe ESCU - This process is an open source tool used for network communications.
6 wce.exe ESCU - This process is associated with a tool used to dump hashes and execute pass-the-hash and pass-the-ticket attacks.
7 cain.exe ESCU - This process is associated with a tool used to collect user credentials and execute attacks.
8 nmap.exe ESCU - This process is an open source network mapping tool used to identify hosts and listening services on a network.
9 kidlogger.exe ESCU - This process is associated with a tool used to collect keyboard input on a host.
10 isass.exe ESCU - This process name is used by attackers to hide in plain sight and look like a legitimate Windows system process.
11 svch0st.exe ESCU - This process name is used by attackers to hide in plain sight and look like a legitimate Windows system process.
12 at.exe ESCU - This process is used to schedule other processes to run. schtasks.exe should be used instead as it provides more flexibility.
13 getmail.exe ESCU - This process is seen to be used by attackers to extract email files from host machines.
14 ntdll.exe ESCU - This process was identified as malicious by DHS Alert TA18-074A.
15 netpass.exe ESCU - This process was identified as malicious by DHS Alert TA18-201A and attackers use this tool to recover all network passwords stored on your system for the current logged-on user.
16 WebBrowserPassView.exe ESCU - This process was identified as malicious by DHS Alert TA18-201A and is used by attackers as a password recovery tool that reveals the passwords stored in Web Browsers.
17 OutlookAddressBookView.exe ESCU - This process was identified as malicious by DHS Alert TA18-201A and is used by attackers to steal the details of all recipients stored in the address books of Microsoft Outlook.
18 mailpv.exe ESCU - This process was identified by DHS Alert TA18-201A and attackers use this tool is a password-recovery tool that reveals the passwords and other account details from various email clients.
19 NLBrute.exe ESCU - This process was identified in the SamSam Ransomware Campaign and attackers use this tool to brute force RDP instances with a range of commonly used passwords.
20 selfdel.exe ESCU - This executable was delivered in the SamSam Ransomware Campain and the attackers levereged this binary to delete its malicilous activities.
lookups/prohibited_softwares.yml
-3
View File
@@ -1,3 +0,0 @@
description: A list of processes that have been marked as prohibited
filename: prohibited_softwares.csv
name: prohibited_softwares