mirror of
https://github.com/splunk/security_content
synced 2026-06-08 17:32:49 +00:00
Branch was auto-updated.
This commit is contained in:
pyth0n1c
@@ -27,6 +27,7 @@ class BAFactoryOutputDto:
|
||||
class BAFactory():
|
||||
input_dto: BAFactoryInputDto
|
||||
output_dto: BAFactoryOutputDto
|
||||
ids: dict[str,list[str]] = {}
|
||||
|
||||
def __init__(self, output_dto: BAFactoryOutputDto) -> None:
|
||||
self.output_dto = output_dto
|
||||
@@ -68,12 +69,14 @@ class BAFactory():
|
||||
type_string = "Detections"
|
||||
self.input_dto.director.constructDetection(self.input_dto.detection_builder, file, [], [], [], self.output_dto.tests, {}, [], [])
|
||||
detection = self.input_dto.detection_builder.getObject()
|
||||
Utils.add_id(self.ids, detection, file)
|
||||
if not detection.deprecated and not detection.experimental:
|
||||
self.output_dto.detections.append(detection)
|
||||
elif type == SecurityContentType.unit_tests:
|
||||
type_string = "Unit Tests"
|
||||
self.input_dto.director.constructTest(self.input_dto.basic_builder, file)
|
||||
test = self.input_dto.basic_builder.getObject()
|
||||
Utils.add_id(self.ids, test, file)
|
||||
self.output_dto.tests.append(test)
|
||||
else:
|
||||
raise(Exception(f"Unsupported content type: [{type}]"))
|
||||
@@ -87,6 +90,12 @@ class BAFactory():
|
||||
print(e)
|
||||
validation_error_found = True
|
||||
|
||||
#Check for any duplicate IDs. The structure is uses
|
||||
# to track them, self.ids, is populated previously in this
|
||||
# function every time content is adde.
|
||||
# This will also print out the duplicates if they exist.
|
||||
validation_error_found |= Utils.check_ids_for_duplicates(self.ids)
|
||||
|
||||
print(f"\r{f'{type_string} Progress'.rjust(23)}: [{progress_percent:3.0f}%]...", end="", flush=True)
|
||||
print("Done!")
|
||||
|
||||
|
||||
@@ -15,6 +15,7 @@ from bin.contentctl_project.contentctl_core.application.builder.playbook_builder
|
||||
from bin.contentctl_project.contentctl_core.application.builder.director import Director
|
||||
from bin.contentctl_project.contentctl_core.application.factory.utils.utils import Utils
|
||||
from bin.contentctl_project.contentctl_core.domain.entities.link_validator import LinkValidator
|
||||
from bin.contentctl_project.contentctl_core.domain.entities.security_content_object import SecurityContentObject
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FactoryInputDto:
|
||||
@@ -42,16 +43,19 @@ class FactoryOutputDto:
|
||||
lookups: list
|
||||
tests: list
|
||||
|
||||
|
||||
|
||||
class Factory():
|
||||
input_dto: FactoryInputDto
|
||||
output_dto: FactoryOutputDto
|
||||
|
||||
ids: dict[str,list[str]] = {}
|
||||
|
||||
def __init__(self, output_dto: FactoryOutputDto) -> None:
|
||||
self.output_dto = output_dto
|
||||
|
||||
|
||||
|
||||
|
||||
def execute(self, input_dto: FactoryInputDto) -> None:
|
||||
self.input_dto = input_dto
|
||||
print("Creating Security Content - ESCU. This may take some time...")
|
||||
@@ -69,7 +73,6 @@ class Factory():
|
||||
|
||||
|
||||
def createSecurityContent(self, type: SecurityContentType) -> list:
|
||||
|
||||
objects = []
|
||||
if type == SecurityContentType.deployments:
|
||||
files = Utils.get_all_yml_files_from_directory(os.path.join(self.input_dto.input_path, str(type.name), 'ESCU'))
|
||||
@@ -78,85 +81,12 @@ class Factory():
|
||||
else:
|
||||
files = Utils.get_all_yml_files_from_directory(os.path.join(self.input_dto.input_path, str(type.name)))
|
||||
|
||||
# Instead of failing on the first error, just keep track of
|
||||
# whether or not an error was found. This way, we can
|
||||
# report all of the errors on a single run so that the
|
||||
# user can see all the errors they need to fix.
|
||||
validation_error_found = False
|
||||
|
||||
#Code below us commented out, but allows threaded construction of detections
|
||||
#No issues have been observed running construction in parallel
|
||||
'''
|
||||
import copy
|
||||
import threading
|
||||
NUM_THREADS = 1
|
||||
|
||||
def make_detections_thread(index:int):
|
||||
my_dto = copy.deepcopy(self.input_dto)
|
||||
for i in range(index, len(files), NUM_THREADS):
|
||||
file = files[i]
|
||||
if 'ssa__' in file:
|
||||
continue
|
||||
try:
|
||||
if type == SecurityContentType.lookups:
|
||||
my_dto.director.constructLookup(my_dto.basic_builder, file)
|
||||
self.output_dto.lookups.append(my_dto.basic_builder.getObject())
|
||||
|
||||
elif type == SecurityContentType.macros:
|
||||
my_dto.director.constructMacro(my_dto.basic_builder, file)
|
||||
self.output_dto.macros.append(my_dto.basic_builder.getObject())
|
||||
|
||||
elif type == SecurityContentType.deployments:
|
||||
my_dto.director.constructDeployment(my_dto.basic_builder, file)
|
||||
self.output_dto.deployments.append(my_dto.basic_builder.getObject())
|
||||
|
||||
elif type == SecurityContentType.playbooks:
|
||||
my_dto.director.constructPlaybook(my_dto.playbook_builder, file)
|
||||
self.output_dto.playbooks.append(my_dto.playbook_builder.getObject())
|
||||
|
||||
elif type == SecurityContentType.baselines:
|
||||
my_dto.director.constructBaseline(my_dto.baseline_builder, file, self.output_dto.deployments)
|
||||
baseline = my_dto.baseline_builder.getObject()
|
||||
self.output_dto.baselines.append(baseline)
|
||||
|
||||
elif type == SecurityContentType.investigations:
|
||||
my_dto.director.constructInvestigation(my_dto.investigation_builder, file)
|
||||
investigation = my_dto.investigation_builder.getObject()
|
||||
self.output_dto.investigations.append(investigation)
|
||||
|
||||
elif type == SecurityContentType.stories:
|
||||
my_dto.director.constructStory(my_dto.story_builder, file,
|
||||
self.output_dto.detections, self.output_dto.baselines, self.output_dto.investigations)
|
||||
story = my_dto.story_builder.getObject()
|
||||
self.output_dto.stories.append(story)
|
||||
|
||||
elif type == SecurityContentType.detections:
|
||||
my_dto.director.constructDetection(my_dto.detection_builder, file,
|
||||
self.output_dto.deployments, self.output_dto.playbooks, self.output_dto.baselines,
|
||||
self.output_dto.tests, my_dto.attack_enrichment, self.output_dto.macros,
|
||||
self.output_dto.lookups, my_dto.force_cached_or_offline)
|
||||
detection = my_dto.detection_builder.getObject()
|
||||
self.output_dto.detections.append(detection)
|
||||
|
||||
elif type == SecurityContentType.unit_tests:
|
||||
my_dto.director.constructTest(my_dto.basic_builder, file)
|
||||
test = my_dto.basic_builder.getObject()
|
||||
self.output_dto.tests.append(test)
|
||||
|
||||
except ValidationError as e:
|
||||
print('\nValidation Error for file ' + file)
|
||||
print(e)
|
||||
validation_error_found = True
|
||||
|
||||
|
||||
|
||||
builder_threads = []
|
||||
|
||||
for t in range(NUM_THREADS):
|
||||
thread = threading.Thread(target=make_detections_thread,args=(t,))
|
||||
thread.start()
|
||||
builder_threads.append(thread)
|
||||
|
||||
for thread in builder_threads:
|
||||
thread.join()
|
||||
'''
|
||||
|
||||
already_ran = False
|
||||
progress_percent = 0
|
||||
type_string = "UNKNOWN TYPE"
|
||||
@@ -173,33 +103,43 @@ class Factory():
|
||||
if type == SecurityContentType.lookups:
|
||||
type_string = "Lookups"
|
||||
self.input_dto.director.constructLookup(self.input_dto.basic_builder, file)
|
||||
self.output_dto.lookups.append(self.input_dto.basic_builder.getObject())
|
||||
lookup = self.input_dto.basic_builder.getObject()
|
||||
Utils.add_id(self.ids, lookup, file)
|
||||
self.output_dto.lookups.append(lookup)
|
||||
|
||||
elif type == SecurityContentType.macros:
|
||||
type_string = "Macros"
|
||||
self.input_dto.director.constructMacro(self.input_dto.basic_builder, file)
|
||||
self.output_dto.macros.append(self.input_dto.basic_builder.getObject())
|
||||
macro = self.input_dto.basic_builder.getObject()
|
||||
Utils.add_id(self.ids, macro, file)
|
||||
self.output_dto.macros.append(macro)
|
||||
|
||||
elif type == SecurityContentType.deployments:
|
||||
type_string = "Deployments"
|
||||
self.input_dto.director.constructDeployment(self.input_dto.basic_builder, file)
|
||||
self.output_dto.deployments.append(self.input_dto.basic_builder.getObject())
|
||||
deployment = self.input_dto.basic_builder.getObject()
|
||||
Utils.add_id(self.ids, deployment, file)
|
||||
self.output_dto.deployments.append(deployment)
|
||||
|
||||
elif type == SecurityContentType.playbooks:
|
||||
type_string = "Playbooks"
|
||||
self.input_dto.director.constructPlaybook(self.input_dto.playbook_builder, file)
|
||||
self.output_dto.playbooks.append(self.input_dto.playbook_builder.getObject())
|
||||
playbook = self.input_dto.playbook_builder.getObject()
|
||||
Utils.add_id(self.ids, playbook, file)
|
||||
self.output_dto.playbooks.append(playbook)
|
||||
|
||||
elif type == SecurityContentType.baselines:
|
||||
type_string = "Baselines"
|
||||
self.input_dto.director.constructBaseline(self.input_dto.baseline_builder, file, self.output_dto.deployments)
|
||||
baseline = self.input_dto.baseline_builder.getObject()
|
||||
Utils.add_id(self.ids, baseline, file)
|
||||
self.output_dto.baselines.append(baseline)
|
||||
|
||||
elif type == SecurityContentType.investigations:
|
||||
type_string = "Investigations"
|
||||
self.input_dto.director.constructInvestigation(self.input_dto.investigation_builder, file)
|
||||
investigation = self.input_dto.investigation_builder.getObject()
|
||||
Utils.add_id(self.ids, investigation, file)
|
||||
self.output_dto.investigations.append(investigation)
|
||||
|
||||
elif type == SecurityContentType.stories:
|
||||
@@ -207,6 +147,7 @@ class Factory():
|
||||
self.input_dto.director.constructStory(self.input_dto.story_builder, file,
|
||||
self.output_dto.detections, self.output_dto.baselines, self.output_dto.investigations)
|
||||
story = self.input_dto.story_builder.getObject()
|
||||
Utils.add_id(self.ids, story, file)
|
||||
self.output_dto.stories.append(story)
|
||||
|
||||
elif type == SecurityContentType.detections:
|
||||
@@ -216,12 +157,14 @@ class Factory():
|
||||
self.output_dto.tests, self.input_dto.attack_enrichment, self.output_dto.macros,
|
||||
self.output_dto.lookups, self.input_dto.force_cached_or_offline)
|
||||
detection = self.input_dto.detection_builder.getObject()
|
||||
Utils.add_id(self.ids, detection, file)
|
||||
self.output_dto.detections.append(detection)
|
||||
|
||||
elif type == SecurityContentType.unit_tests:
|
||||
type_string = "Unit Tests"
|
||||
self.input_dto.director.constructTest(self.input_dto.basic_builder, file)
|
||||
test = self.input_dto.basic_builder.getObject()
|
||||
Utils.add_id(self.ids, test, file)
|
||||
self.output_dto.tests.append(test)
|
||||
|
||||
else:
|
||||
@@ -235,7 +178,14 @@ class Factory():
|
||||
print('\nValidation Error for file ' + file)
|
||||
print(e)
|
||||
validation_error_found = True
|
||||
|
||||
|
||||
|
||||
#Check for any duplicate IDs. The structure is uses
|
||||
# to track them, self.ids, is populated previously in this
|
||||
# function every time content is adde.
|
||||
# This will also print out the duplicates if they exist.
|
||||
validation_error_found |= Utils.check_ids_for_duplicates(self.ids)
|
||||
|
||||
print(f"\r{f'{type_string} Progress'.rjust(23)}: [{progress_percent:3.0f}%]...", end="", flush=True)
|
||||
print("Done!")
|
||||
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
import os
|
||||
|
||||
from bin.contentctl_project.contentctl_core.domain.entities.security_content_object import SecurityContentObject
|
||||
|
||||
class Utils:
|
||||
|
||||
@staticmethod
|
||||
@@ -10,4 +12,24 @@ class Utils:
|
||||
if file.endswith(".yml"):
|
||||
listOfFiles.append(os.path.join(dirpath, file))
|
||||
|
||||
return sorted(listOfFiles)
|
||||
return sorted(listOfFiles)
|
||||
|
||||
@staticmethod
|
||||
def add_id(id_dict:dict[str, list[str]], obj:SecurityContentObject, path:str) -> None:
|
||||
if hasattr(obj, "id"):
|
||||
obj_id = obj.id
|
||||
if obj_id in id_dict:
|
||||
id_dict[obj_id].append(path)
|
||||
else:
|
||||
id_dict[obj_id] = [path]
|
||||
# Otherwise, no ID so nothing to add....
|
||||
|
||||
@staticmethod
|
||||
def check_ids_for_duplicates(id_dict:dict[str, list[str]])->bool:
|
||||
validation_error = False
|
||||
for key, values in id_dict.items():
|
||||
if len(values) > 1:
|
||||
validation_error = True
|
||||
id_conflicts_string = '\n\t* '.join(values)
|
||||
print(f"\nError validating id [{key}] - duplicate ID is used for the following content: \n\t* {id_conflicts_string}")
|
||||
return validation_error
|
||||
@@ -1,7 +1,7 @@
|
||||
name: Spring4Shell Payload URL Request
|
||||
id: 2850c734-2d44-4431-8139-1a56f6f54c01
|
||||
id: 9d44d649-7d67-4559-95c1-8022ff49420b
|
||||
version: 1
|
||||
date: '2022-04-05'
|
||||
date: '2022-07-12'
|
||||
author: Michael Haag, Splunk
|
||||
type: TTP
|
||||
datamodel:
|
||||
|
||||
Reference in New Issue
Block a user