Skip to content

Commit

Permalink
Merge pull request #219 from splunk/remove_ssa_specific_code
Browse files Browse the repository at this point in the history 
Remove SSA specific code
  • Loading branch information
@pyth0n1c
pyth0n1c authored Aug 21, 2024
2 parents 44ef8d9 + 83e1c08 commit 5d84999
Show file tree
Hide file tree
Showing 12 changed files with 66 additions and 1,157 deletions.
14 changes: 0 additions & 14 deletions contentctl/actions/build.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,6 @@
from contentctl.input.director import Director, DirectorOutputDto
from contentctl.output.conf_output import ConfOutput
from contentctl.output.conf_writer import ConfWriter
from contentctl.output.ba_yml_output import BAYmlOutput
from contentctl.output.api_json_output import ApiJsonOutput
from contentctl.output.data_source_writer import DataSourceWriter
from contentctl.objects.lookup import Lookup
Expand Down Expand Up @@ -86,17 +85,4 @@ def execute(self, input_dto: BuildInputDto) -> DirectorOutputDto:

print(f"Build of '{input_dto.config.app.title}' API successful to {input_dto.config.getAPIPath()}")

if input_dto.config.build_ssa:

srs_path = input_dto.config.getSSAPath() / 'srs'
complex_path = input_dto.config.getSSAPath() / 'complex'
shutil.rmtree(srs_path, ignore_errors=True)
shutil.rmtree(complex_path, ignore_errors=True)
srs_path.mkdir(parents=True)
complex_path.mkdir(parents=True)
ba_yml_output = BAYmlOutput()
ba_yml_output.writeObjects(input_dto.director_output_dto.ssa_detections, str(input_dto.config.getSSAPath()))

print(f"Build of 'SSA' successful to {input_dto.config.getSSAPath()}")

return input_dto.director_output_dto
25 changes: 0 additions & 25 deletions contentctl/actions/convert.py
View file

This file was deleted.

1 change: 0 additions & 1 deletion contentctl/actions/validate.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,6 @@ def execute(self, input_dto: validate) -> DirectorOutputDto:
[],
[],
[],
[],
)

director = Director(director_output_dto)
Expand Down
144 changes: 0 additions & 144 deletions contentctl/input/backend_splunk_ba.py
View file

This file was deleted.

53 changes: 9 additions & 44 deletions contentctl/input/director.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -28,13 +28,11 @@
from contentctl.enrichments.cve_enrichment import CveEnrichment

from contentctl.objects.config import validate
from contentctl.input.ssa_detection_builder import SSADetectionBuilder
from contentctl.objects.enums import SecurityContentType

from contentctl.objects.enums import DetectionStatus
from contentctl.helper.utils import Utils

from contentctl.input.ssa_detection_builder import SSADetectionBuilder
from contentctl.objects.enums import SecurityContentType

from contentctl.objects.enums import DetectionStatus
Expand All @@ -56,7 +54,6 @@ class DirectorOutputDto:
macros: list[Macro]
lookups: list[Lookup]
deployments: list[Deployment]
ssa_detections: list[SSADetection]
data_sources: list[DataSource]
name_to_content_map: dict[str, SecurityContentObject] = field(default_factory=dict)
uuid_to_content_map: dict[UUID, SecurityContentObject] = field(default_factory=dict)
Expand Down Expand Up @@ -98,8 +95,6 @@ def addContentToDictMappings(self, content: SecurityContentObject):
self.stories.append(content)
elif isinstance(content, Detection):
self.detections.append(content)
elif isinstance(content, SSADetection):
self.ssa_detections.append(content)
elif isinstance(content, DataSource):
self.data_sources.append(content)
else:
Expand All @@ -112,11 +107,9 @@ def addContentToDictMappings(self, content: SecurityContentObject):
class Director():
input_dto: validate
output_dto: DirectorOutputDto
ssa_detection_builder: SSADetectionBuilder

def __init__(self, output_dto: DirectorOutputDto) -> None:
self.output_dto = output_dto
self.ssa_detection_builder = SSADetectionBuilder()

def execute(self, input_dto: validate) -> None:
self.input_dto = input_dto
Expand All @@ -129,7 +122,6 @@ def execute(self, input_dto: validate) -> None:
self.createSecurityContent(SecurityContentType.data_sources)
self.createSecurityContent(SecurityContentType.playbooks)
self.createSecurityContent(SecurityContentType.detections)
self.createSecurityContent(SecurityContentType.ssa_detections)


from contentctl.objects.abstract_security_content_objects.detection_abstract import MISSING_SOURCES
Expand All @@ -142,12 +134,7 @@ def execute(self, input_dto: validate) -> None:
print("No missing data_sources!")

def createSecurityContent(self, contentType: SecurityContentType) -> None:
if contentType == SecurityContentType.ssa_detections:
files = Utils.get_all_yml_files_from_directory(
os.path.join(self.input_dto.path, "ssa_detections")
)
security_content_files = [f for f in files if f.name.startswith("ssa___")]
elif contentType in [
if contentType in [
SecurityContentType.deployments,
SecurityContentType.lookups,
SecurityContentType.macros,
Expand Down Expand Up @@ -179,43 +166,37 @@ def createSecurityContent(self, contentType: SecurityContentType) -> None:
modelDict = YmlReader.load_file(file)

if contentType == SecurityContentType.lookups:
lookup = Lookup.model_validate(modelDict,context={"output_dto":self.output_dto, "config":self.input_dto})
lookup = Lookup.model_validate(modelDict, context={"output_dto":self.output_dto, "config":self.input_dto})
self.output_dto.addContentToDictMappings(lookup)

elif contentType == SecurityContentType.macros:
macro = Macro.model_validate(modelDict,context={"output_dto":self.output_dto})
macro = Macro.model_validate(modelDict, context={"output_dto":self.output_dto})
self.output_dto.addContentToDictMappings(macro)

elif contentType == SecurityContentType.deployments:
deployment = Deployment.model_validate(modelDict,context={"output_dto":self.output_dto})
deployment = Deployment.model_validate(modelDict, context={"output_dto":self.output_dto})
self.output_dto.addContentToDictMappings(deployment)

elif contentType == SecurityContentType.playbooks:
playbook = Playbook.model_validate(modelDict,context={"output_dto":self.output_dto})
playbook = Playbook.model_validate(modelDict, context={"output_dto":self.output_dto})
self.output_dto.addContentToDictMappings(playbook)

elif contentType == SecurityContentType.baselines:
baseline = Baseline.model_validate(modelDict,context={"output_dto":self.output_dto})
baseline = Baseline.model_validate(modelDict, context={"output_dto":self.output_dto})
self.output_dto.addContentToDictMappings(baseline)

elif contentType == SecurityContentType.investigations:
investigation = Investigation.model_validate(modelDict,context={"output_dto":self.output_dto})
investigation = Investigation.model_validate(modelDict, context={"output_dto":self.output_dto})
self.output_dto.addContentToDictMappings(investigation)

elif contentType == SecurityContentType.stories:
story = Story.model_validate(modelDict,context={"output_dto":self.output_dto})
story = Story.model_validate(modelDict, context={"output_dto":self.output_dto})
self.output_dto.addContentToDictMappings(story)

elif contentType == SecurityContentType.detections:
detection = Detection.model_validate(modelDict,context={"output_dto":self.output_dto, "app":self.input_dto.app})
detection = Detection.model_validate(modelDict, context={"output_dto":self.output_dto, "app":self.input_dto.app})
self.output_dto.addContentToDictMappings(detection)

elif contentType == SecurityContentType.ssa_detections:
self.constructSSADetection(self.ssa_detection_builder, self.output_dto,str(file))
ssa_detection = self.ssa_detection_builder.getObject()
if ssa_detection.status in [DetectionStatus.production.value, DetectionStatus.validation.value]:
self.output_dto.addContentToDictMappings(ssa_detection)

elif contentType == SecurityContentType.data_sources:
data_source = DataSource.model_validate(
modelDict, context={"output_dto": self.output_dto}
Expand Down Expand Up @@ -262,19 +243,3 @@ def createSecurityContent(self, contentType: SecurityContentType) -> None:
f"The following {len(validation_errors)} error(s) were found during validation:\n\n{errors_string}\n\nVALIDATION FAILED"
)

def constructSSADetection(
self,
builder: SSADetectionBuilder,
directorOutput: DirectorOutputDto,
file_path: str,
) -> None:
builder.reset()
builder.setObject(file_path)
builder.addMitreAttackEnrichmentNew(directorOutput.attack_enrichment)
builder.addKillChainPhase()
builder.addCIS()
builder.addNist()
builder.addAnnotations()
builder.addMappings()
builder.addUnitTest()
builder.addRBA()
Loading

0 comments on commit 5d84999

Please sign in to comment.