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:
pyth0n1c
2022-03-28 11:48:02 -07:00
committed by GitHub
parent ea6261926a 61fef2deb7
commit e4f71708f8
3 changed files with 61 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bin/contentctl_project/contentctl_core/application/factory/new_content_factory.py
+18 -5
View File
@@ -1,17 +1,18 @@
import os
import uuid
import questionary
from dataclasses import dataclass
from datetime import datetime
from bin.contentctl_project.contentctl_core.domain.entities.enums.enums import SecurityContentType
from bin.contentctl_project.contentctl_core.domain.entities.enums.enums import SecurityContentProduct
from bin.contentctl_project.contentctl_core.application.factory.utils.new_content_questions import NewContentQuestions
@dataclass(frozen=True)
class NewContentFactoryInputDto:
type: SecurityContentType
type: SecurityContentProduct
@dataclass(frozen=True)
@@ -37,9 +38,15 @@ class NewContentFactory():
self.output_dto.obj['author'] = answers['detection_author']
self.output_dto.obj['type'] = answers['detection_type']
self.output_dto.obj['datamodel'] = answers['datamodels']
self.output_dto.obj['description'] = 'UPDATE_DESCRIPTION'
file_name = self.output_dto.obj['name'].replace(' ', '_').replace('-','_').replace('.','_').replace('/','_').lower()
self.output_dto.obj['search'] = answers['detection_search'] + ' | `' + file_name + '_filter`'
if answers['detection_product'] == 'SSA':
answers['datamodels'] = [d.replace(' (SSA)', '') for d in answers['datamodels']]
self.output_dto.obj['datamodel'] = answers['datamodels']
if answers['detection_product'] == 'ESCU':
self.output_dto.obj['datamodel'] = answers['datamodels']
self.output_dto.obj['description'] = 'UPDATE_DESCRIPTION'
if answers['detection_product'] == 'ESCU':
file_name = self.output_dto.obj['name'].replace(' ', '_').replace('-','_').replace('.','_').replace('/','_').lower()
self.output_dto.obj['search'] = answers['detection_search'] + ' | `' + file_name + '_filter`'
self.output_dto.obj['how_to_implement'] = 'UPDATE_HOW_TO_IMPLEMENT'
self.output_dto.obj['known_false_positives'] = 'UPDATE_KNOWN_FALSE_POSITIVES'
self.output_dto.obj['references'] = ['REFERENCE']
@@ -56,11 +63,17 @@ class NewContentFactory():
self.output_dto.obj['tags']['mitre_attack_id'] = [x.strip() for x in answers['mitre_attack_ids'].split(',')]
self.output_dto.obj['tags']['nist'] = ['DE.CM']
self.output_dto.obj['tags']['observable'] = [{'name': 'UPDATE', 'type': 'UPDATE', 'role': ['UPDATE']}]
self.output_dto.obj['tags']['product'] = ['Splunk Enterprise','Splunk Enterprise Security','Splunk Cloud']
if answers['detection_product'] == 'SSA':
self.output_dto.obj['tags']['risk_severity'] = 'UPDATE: <low>, <medium>, <high>'
if answers['detection_product'] == 'ESCU':
self.output_dto.obj['tags']['product'] = ['Splunk Enterprise','Splunk Enterprise Security','Splunk Cloud']
if answers['detection_product'] == 'SSA':
self.output_dto.obj['tags']['product'] = ['Splunk Behavioral Analytics']
self.output_dto.obj['tags']['required_fields'] = ['UPDATE']
self.output_dto.obj['tags']['risk_score'] = 'UPDATE (impact * confidence)/100'
self.output_dto.obj['tags']['security_domain'] = answers['security_domain']
self.output_dto.obj['source'] = answers['detection_kind']
elif input_dto.type == SecurityContentType.stories:
questions = NewContentQuestions.get_questions_story()
bin/contentctl_project/contentctl_core/application/factory/utils/new_content_questions.py
+18
View File
@@ -5,6 +5,16 @@ class NewContentQuestions():
@classmethod
def get_questions_detection(self) -> list:
questions = [
{
'type': 'select',
'message': 'what product is this for',
'name': 'detection_product',
'choices': [
'ESCU',
'SSA'
],
'default': 'ESCU'
},
{
'type': 'select',
'message': 'what kind of detection is this',
@@ -50,6 +60,12 @@ class NewContentQuestions():
'name': 'datamodels',
'choices': [
'Endpoint',
'Endpoint_Processes (SSA)',
'Endpoint_Registry (SSA)',
'Endpoint_Filesystem (SSA)',
'Endpoint_ResourceAccess (SSA)',
'Endpoint_AccountManagement (SSA)',
'Intrusion_Detection (SSA)',
'Authentication',
'Change',
'Email',
@@ -61,6 +77,7 @@ class NewContentQuestions():
'Web',
'Risk'
],
'default': 'Endpoint'
},
{
'type': 'text',
@@ -88,6 +105,7 @@ class NewContentQuestions():
'Actions on Objectives',
'Denial of Service'
],
'default': 'Exploitation'
},
{
'type': 'select',
bin/contentctl_project/contentctl_infrastructure/adapter/obj_to_yml_adapter.py
+25 -8
View File
@@ -18,8 +18,8 @@ class ObjToYmlAdapter(Adapter):
def writeObjects(self, objects: list, output_path: str, type: SecurityContentType = None) -> None:
for obj in objects:
file_name = "ssa___" + self.convertNameToFileName(obj.name)
for obj in objects:
file_name = "ssa___" + self.convertNameToFileName(obj.name, obj.tags)
if self.isComplexBARule(obj.search):
file_path = os.path.join(output_path, 'complex', file_name)
else:
@@ -85,13 +85,13 @@ class ObjToYmlAdapter(Adapter):
def writeObjectNewContent(self, object: dict, type: SecurityContentType) -> None:
if type == SecurityContentType.detections:
file_path = os.path.join(os.path.dirname(__file__), '../../../../detections', object['source'], self.convertNameToFileName(object['name']))
file_path = os.path.join(os.path.dirname(__file__), '../../../../detections', object['source'], self.convertNameToFileName(object['name'],object['tags']['product']))
test_obj = {}
test_obj['name'] = object['name'] + ' Unit Test'
test_obj['tests'] = [
{
'name': object['name'],
'file': object['source'] + '/' + self.convertNameToFileName(object['name']),
'file': object['source'] + '/' + self.convertNameToFileName(object['name'],object['tags']['product']),
'pass_condition': '| stats count | where count > 0',
'earliest_time': '-24h',
'latest_time': 'now',
@@ -105,23 +105,40 @@ class ObjToYmlAdapter(Adapter):
]
}
]
file_path_test = os.path.join(os.path.dirname(__file__), '../../../../tests', object['source'], self.convertNameToFileName(object['name']))
file_path_test = os.path.join(os.path.dirname(__file__), '../../../../tests', object['source'], self.convertNameToTestFileName(object['name'],object['tags']['product']))
YmlWriter.writeYmlFile(file_path_test, test_obj)
object.pop('source')
elif type == SecurityContentType.stories:
file_path = os.path.join(os.path.dirname(__file__), '../../../../stories', self.convertNameToFileName(object['name']))
file_path = os.path.join(os.path.dirname(__file__), '../../../../stories', self.convertNameToFileName(object['name'],object['tags']['product']))
YmlWriter.writeYmlFile(file_path, object)
def convertNameToFileName(self, name: str):
def convertNameToFileName(self, name: str, product: list):
file_name = name \
.replace(' ', '_') \
.replace('-','_') \
.replace('.','_') \
.replace('/','_') \
.lower()
file_name = file_name + '.yml'
if 'Splunk Behavioral Analytics' in product:
file_name = 'ssa___' + file_name + '.yml'
else:
file_name = file_name + '.yml'
return file_name
def convertNameToTestFileName(self, name: str, product: list):
file_name = name \
.replace(' ', '_') \
.replace('-','_') \
.replace('.','_') \
.replace('/','_') \
.lower()
if 'Splunk Behavioral Analytics' in product:
file_name = 'ssa___' + file_name + '.test.yml'
else:
file_name = file_name + '.test.yml'
return file_name