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

Mostly done with overhaul to arguments. Initial testing show that it looks good. Now to integrate the newly parsed arguments in with the rewritten run behavior.

Browse Source
This commit is contained in:
pyth0n1c
2021-11-18 16:56:46 -08:00
parent aca21ef285
commit 1f5386b776
3 changed files with 70 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
automated_detection_testing/ci/detection_testing_batch/detection_testing_execution.py
+20 -14
View File
@@ -27,6 +27,9 @@ import csv
from requests import get
import json
import requests.packages.urllib3
import modules.new_arguments2
SPLUNK_CONTAINER_APPS_DIR = "/opt/splunk/etc/apps"
index_file_local_path = "indexes.conf.tar"
@@ -38,25 +41,28 @@ datamodel_file_container_path = os.path.join(SPLUNK_CONTAINER_APPS_DIR, "Splunk_
MAX_RECOMMENDED_CONTAINERS_BEFORE_WARNING=2
DEFAULT_CONTAINER_TAG="latest"
LOCAL_BASE_CONTAINER_NAME = "splunk_test_%d"
BASE_CONTAINER_WEB_PORT=8000
BASE_CONTAINER_MANAGEMENT_PORT=8089
DETECTION_TYPES = ['endpoint', 'cloud', 'network']
DETECTION_MODES = ['new', 'all', 'selected']
def main(args):
start_time = timer()
requests.packages.urllib3.disable_warnings()
start_datetime = datetime.now()
action, settings = modules.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)
print("time to run the test!")
sys.exit(0)
parser = argparse.ArgumentParser(description="CI Detection Testing")
parser.add_argument("-b", "--branch", type=str, required=True, help="security content branch")
@@ -92,9 +98,9 @@ def main(args):
parser.add_argument("-split","--split_detections_then_stop", required=False, default=False, action='store_true', help="Should existing images be re-used, or should they be redownloaded?")
start_datetime = datetime.now()
import requests.packages.urllib3
requests.packages.urllib3.disable_warnings()
args = parser.parse_args()
branch = args.branch
uuid_test = args.uuid
automated_detection_testing/ci/detection_testing_batch/modules/new_arguments2.py
+16 -18
View File
@@ -1,12 +1,12 @@
import argparse
import json
from typing import OrderedDict
import validate_args
from typing import OrderedDict, Union
import modules.validate_args as validate_args
import sys
DEFAULT_CONFIG_FILE = "test_config.json"
def configure_action(args) -> bool:
def configure_action(args) -> tuple[str, dict]:
settings = OrderedDict()
if args.input_config_file is None:
settings, schema = validate_args.validate({})
@@ -58,16 +58,18 @@ def configure_action(args) -> bool:
validated_new_settings, schema = validate_args.validate(new_config)
if validated_new_settings == None:
print("Error in the new settings!")
return False
sys.exit(1)
else:
print("New settings worked great. Writing results to: %s" %
print("New settings successful. Writing results to: %s" %
(args.output_config_file.name))
args.output_config_file.write(json.dumps(
validated_new_settings, sort_keys=True, indent=4))
return True
return ("configure", validated_new_settings)
def update_config_with_cli_arguments(args_dict:dict)->dict:
def update_config_with_cli_arguments(args_dict:dict)->tuple[str, dict]:
#First load the config file
settings,_ = validate_args.validate_file(args_dict['config_file'])
@@ -86,19 +88,19 @@ def update_config_with_cli_arguments(args_dict:dict)->dict:
print("Failure while processing updated settings from command line.\n\tQuitting...", file=sys.stderr)
sys.exit(1)
return settings
return ("run", settings)
def run_action(args) -> bool:
def run_action(args) -> tuple[str,dict]:
config = update_config_with_cli_arguments(args.__dict__)
return True
return config
def main(args):
def parse(args)->tuple[str,dict]:
'''
try:
with open(DEFAULT_CONFIG_FILE, 'r') as settings_file:
@@ -158,12 +160,8 @@ def main(args):
# Run the appropriate parser
try:
if args.func(args):
print("Success!")
sys.exit(0)
else:
print("Fail")
sys.exit(1)
action, settings = args.func(args)
return action, settings
except Exception as e:
print("Unknown Error - [%s]" % (str(e)))
sys.exit(1)
@@ -251,4 +249,4 @@ def main(args):
if __name__ == "__main__":
main(sys.argv[1:])
parse(sys.argv[1:])
automated_detection_testing/ci/detection_testing_batch/modules/validate_args.py
+34 -25
View File
@@ -4,7 +4,7 @@ import io
import json
import jsonschema
import jsonschema.exceptions
import jsonschema_errorprinter
import modules.jsonschema_errorprinter as jsonschema_errorprinter
import sys
from typing import Union
@@ -28,17 +28,17 @@ setup_schema = {
"type": "boolean",
"default": False
},
"detections_list": {
"type":["array"],
"type": ["array"],
"items": {
"type":"string"
"type": "string"
},
"default":[],
"default": [],
},
"detections_file":{
"type": ["string","null"],
"detections_file": {
"type": ["string", "null"],
"default": None
},
@@ -57,11 +57,20 @@ setup_schema = {
"type": "string"
},
"local_path": {
"type": "string"
"type": ["string", "null"],
"default": None
},
},
"default": []
}
}
},
"default": [
{
"app_name": "SPLUNK_ES_CONTENT_UPDATE",
"app_number": 3449,
"app_version": "GENERATED",
'local_path': None
}
]
},
"mode": {
@@ -214,53 +223,53 @@ setup_schema = {
}
def validate_file(file:io.TextIOWrapper)->tuple[Union[dict, None], dict]:
def validate_file(file: io.TextIOWrapper) -> tuple[Union[dict, None], dict]:
try:
settings = json.loads(file.read())
return validate(settings)
except Exception as e:
raise(e)
def check_dependencies(settings: dict)->bool:
#Check complex mode dependencies
def check_dependencies(settings: dict) -> bool:
# Check complex mode dependencies
error_free = True
if settings['mode'] == 'selected':
#Make sure that exactly one of the following fields is populated
# Make sure that exactly one of the following fields is populated
if settings['detections_file'] == None and settings['detections_list'] == []:
print("Error - mode was 'selected' but no detections_list or detections_file were supplied.",file=sys.stderr)
print("Error - mode was 'selected' but no detections_list or detections_file were supplied.", file=sys.stderr)
error_free = False
elif settings['detections_file'] != None and settings['detections_list'] != []:
print("Error - mode was 'selected' but detections_list and detections_file were supplied.",file=sys.stderr)
print("Error - mode was 'selected' but detections_list and detections_file were supplied.", file=sys.stderr)
error_free = False
if settings['mode'] != 'selected'and settings['detections_file'] != None:
print("Error - mode was not 'selected' but detections_file was supplied.",file=sys.stderr)
if settings['mode'] != 'selected' and settings['detections_file'] != None:
print("Error - mode was not 'selected' but detections_file was supplied.", file=sys.stderr)
error_free = False
elif settings['mode'] != 'selected' and settings['detections_list'] != []:
print("Error - mode was not 'selected' but detections_list was supplied.",file=sys.stderr)
print("Error - mode was not 'selected' but detections_list was supplied.", file=sys.stderr)
error_free = False
#Returns true if there are not errors
# Returns true if there are not errors
return error_free
def validate(configuration: dict) -> tuple[Union[dict, None], dict]:
#v = jsonschema.Draft201909Validator(argument_schema)
try:
validation_errors, validated_json = jsonschema_errorprinter.check_json(
configuration, setup_schema)
no_complex_errors = check_dependencies(validated_json)
if len(validation_errors) == 0 and no_complex_errors:
print("Input configuration successfully validated!")
return validated_json, setup_schema
elif no_complex_errors == False:
print("Failed due to error(s) listed above.", file=sys.stderr)
return None,setup_schema
return None, setup_schema
else:
print("[%d] failures detected during validation of the configuration!" % (
len(validation_errors)))
len(validation_errors)),file=sys.stderr)
for error in validation_errors:
print(error, end="\n\n", file=sys.stderr)
return None, setup_schema