mirror of
https://github.com/splunk/security_content
synced 2026-06-08 17:32:49 +00:00
Add content changer
This commit is contained in:
P4T12ICK
@@ -14,6 +14,9 @@ class ContentChangerInputDto:
|
||||
adapter : Adapter
|
||||
factory_input_dto : ObjectFactoryInputDto
|
||||
converter_func_name : str
|
||||
filter_key: str
|
||||
filter_value: str
|
||||
variables: list
|
||||
|
||||
class ContentChanger:
|
||||
|
||||
@@ -22,10 +25,14 @@ class ContentChanger:
|
||||
factory = ObjectFactory(objects)
|
||||
factory.execute(input_dto.factory_input_dto)
|
||||
|
||||
filtered_objects = objects
|
||||
if input_dto.filter_key and input_dto.filter_value:
|
||||
filtered_objects = self.apply_key_value_filter(objects, input_dto.filter_key, input_dto.filter_value)
|
||||
|
||||
converter_func = getattr(self, input_dto.converter_func_name)
|
||||
converter_func(objects)
|
||||
converter_func(filtered_objects, input_dto.variables)
|
||||
|
||||
input_dto.adapter.writeObjectsInPlace(objects)
|
||||
input_dto.adapter.writeObjectsInPlace(filtered_objects)
|
||||
|
||||
@staticmethod
|
||||
def enumerate_content_changer_functions(exclude_functions: list[str] = ["enumerate_content_changer_functions", "execute", "example_converter_func"]) -> list[str]:
|
||||
@@ -33,6 +40,15 @@ class ContentChanger:
|
||||
function_names = [function_object[0] for function_object in members if function_object[0] not in exclude_functions]
|
||||
return function_names
|
||||
|
||||
def apply_key_value_filter(self, objects: list, key: str, value: str) -> None:
|
||||
new_list = list()
|
||||
for obj in objects:
|
||||
if str(obj[key]) == value:
|
||||
new_list.append(obj)
|
||||
|
||||
return new_list
|
||||
|
||||
|
||||
def all(self, objects : list) -> None:
|
||||
for func_name in ContentChanger.enumerate_content_changer_functions():
|
||||
if func_name not in ["all", "change_test_file_format"]:
|
||||
@@ -42,158 +58,162 @@ class ContentChanger:
|
||||
|
||||
|
||||
# Define Converter Functions here
|
||||
def example_converter_func(self, objects : list) -> None:
|
||||
for obj in objects:
|
||||
obj['author'] = obj['author'].upper()
|
||||
# def example_converter_func(self, objects : list) -> None:
|
||||
# for obj in objects:
|
||||
# obj['author'] = obj['author'].upper()
|
||||
|
||||
def add_unknown_context(self, objects : list) -> None:
|
||||
for obj in objects:
|
||||
if not 'context' in obj['tags']:
|
||||
obj['tags']['context'] = ['Unknown']
|
||||
# def add_unknown_context(self, objects : list) -> None:
|
||||
# for obj in objects:
|
||||
# if not 'context' in obj['tags']:
|
||||
# obj['tags']['context'] = ['Unknown']
|
||||
|
||||
def add_default_message(self, objects : list) -> None:
|
||||
for obj in objects:
|
||||
if not 'message' in obj['tags']:
|
||||
obj['tags']['message'] = 'tbd'
|
||||
# def add_default_message(self, objects : list) -> None:
|
||||
# for obj in objects:
|
||||
# if not 'message' in obj['tags']:
|
||||
# obj['tags']['message'] = 'tbd'
|
||||
|
||||
def add_default_observable(self, objects : list) -> None:
|
||||
for obj in objects:
|
||||
if not 'observable' in obj['tags'] or ('observable' in obj['tags'] and len(obj['tags']['observable']) == 0):
|
||||
observables = []
|
||||
regexp_user = re.compile(r'user')
|
||||
if regexp_user.search(obj['search']):
|
||||
observables.append({'name': 'user', 'type': 'User', 'role': ['Victim']})
|
||||
regexp_user = re.compile(r'dest')
|
||||
if regexp_user.search(obj['search']):
|
||||
observables.append({'name': 'dest', 'type': 'Hostname', 'role': ['Victim']})
|
||||
if len(observables) == 0:
|
||||
observables.append({'name': 'dest', 'type': 'Other', 'role': ['Other']})
|
||||
obj['tags']['observable'] = observables
|
||||
# def add_default_observable(self, objects : list) -> None:
|
||||
# for obj in objects:
|
||||
# if not 'observable' in obj['tags'] or ('observable' in obj['tags'] and len(obj['tags']['observable']) == 0):
|
||||
# observables = []
|
||||
# regexp_user = re.compile(r'user')
|
||||
# if regexp_user.search(obj['search']):
|
||||
# observables.append({'name': 'user', 'type': 'User', 'role': ['Victim']})
|
||||
# regexp_user = re.compile(r'dest')
|
||||
# if regexp_user.search(obj['search']):
|
||||
# observables.append({'name': 'dest', 'type': 'Hostname', 'role': ['Victim']})
|
||||
# if len(observables) == 0:
|
||||
# observables.append({'name': 'dest', 'type': 'Other', 'role': ['Other']})
|
||||
# obj['tags']['observable'] = observables
|
||||
|
||||
def add_default_cis(self, objects : list) -> None:
|
||||
for obj in objects:
|
||||
if not 'cis20' in obj['tags']:
|
||||
obj['tags']['cis20'] = ['CIS 3', 'CIS 5', 'CIS 16']
|
||||
# def add_default_cis(self, objects : list) -> None:
|
||||
# for obj in objects:
|
||||
# if not 'cis20' in obj['tags']:
|
||||
# obj['tags']['cis20'] = ['CIS 3', 'CIS 5', 'CIS 16']
|
||||
|
||||
def add_default_nist(self, objects : list) -> None:
|
||||
for obj in objects:
|
||||
if not 'nist' in obj['tags']:
|
||||
obj['tags']['nist'] = ['DE.CM']
|
||||
# def add_default_nist(self, objects : list) -> None:
|
||||
# for obj in objects:
|
||||
# if not 'nist' in obj['tags']:
|
||||
# obj['tags']['nist'] = ['DE.CM']
|
||||
|
||||
def fix_broken_uuids(self, objects : list) -> None:
|
||||
for obj in objects:
|
||||
try:
|
||||
uuid.UUID(str(obj['id']))
|
||||
except:
|
||||
obj['id'] = str(uuid.uuid4())
|
||||
# def fix_broken_uuids(self, objects : list) -> None:
|
||||
# for obj in objects:
|
||||
# try:
|
||||
# uuid.UUID(str(obj['id']))
|
||||
# except:
|
||||
# obj['id'] = str(uuid.uuid4())
|
||||
|
||||
def fix_wrong_kill_chain_phases(self, objects : list) -> None:
|
||||
valid_kill_chain_phases = [
|
||||
'Reconnaissance', 'Weaponization', 'Delivery',
|
||||
'Exploitation', 'Installation', 'Command And Control',
|
||||
'Actions on Objectives']
|
||||
for obj in objects:
|
||||
if 'kill_chain_phases' in obj['tags']:
|
||||
for value in obj['tags']['kill_chain_phases']:
|
||||
if value not in valid_kill_chain_phases:
|
||||
obj['tags']['kill_chain_phases'] = ['Exploitation']
|
||||
break
|
||||
# def fix_wrong_kill_chain_phases(self, objects : list) -> None:
|
||||
# valid_kill_chain_phases = [
|
||||
# 'Reconnaissance', 'Weaponization', 'Delivery',
|
||||
# 'Exploitation', 'Installation', 'Command And Control',
|
||||
# 'Actions on Objectives']
|
||||
# for obj in objects:
|
||||
# if 'kill_chain_phases' in obj['tags']:
|
||||
# for value in obj['tags']['kill_chain_phases']:
|
||||
# if value not in valid_kill_chain_phases:
|
||||
# obj['tags']['kill_chain_phases'] = ['Exploitation']
|
||||
# break
|
||||
|
||||
def add_default_kill_chain_phases(self, objects : list) -> None:
|
||||
for obj in objects:
|
||||
if 'kill_chain_phases' not in obj['tags']:
|
||||
obj['tags']['kill_chain_phases'] = ['Exploitation']
|
||||
if obj['tags']['kill_chain_phases'] == ['Privilege Escalation']:
|
||||
obj['tags']['kill_chain_phases'] = ['Exploitation']
|
||||
# def add_default_kill_chain_phases(self, objects : list) -> None:
|
||||
# for obj in objects:
|
||||
# if 'kill_chain_phases' not in obj['tags']:
|
||||
# obj['tags']['kill_chain_phases'] = ['Exploitation']
|
||||
# if obj['tags']['kill_chain_phases'] == ['Privilege Escalation']:
|
||||
# obj['tags']['kill_chain_phases'] = ['Exploitation']
|
||||
|
||||
def fix_wrong_calculated_risk_score(self, objects : list) -> None:
|
||||
for obj in objects:
|
||||
#Risk score must be an integer, so we round it to the nearest integer
|
||||
calculated_risk_score = round((obj['tags']['impact'] * obj['tags']['confidence'])/100)
|
||||
if calculated_risk_score != round(obj['tags']['risk_score']):
|
||||
obj['tags']['risk_score'] = calculated_risk_score
|
||||
# def fix_wrong_calculated_risk_score(self, objects : list) -> None:
|
||||
# for obj in objects:
|
||||
# #Risk score must be an integer, so we round it to the nearest integer
|
||||
# calculated_risk_score = round((obj['tags']['impact'] * obj['tags']['confidence'])/100)
|
||||
# if calculated_risk_score != round(obj['tags']['risk_score']):
|
||||
# obj['tags']['risk_score'] = calculated_risk_score
|
||||
|
||||
def add_asset_type_to_endpoint_detections(self, objects : list) -> None:
|
||||
for obj in objects:
|
||||
if 'asset_type' not in obj['tags']:
|
||||
if '/endpoint/' in obj['file_path']:
|
||||
obj['tags']['asset_type'] = 'Endpoint'
|
||||
# def add_asset_type_to_endpoint_detections(self, objects : list) -> None:
|
||||
# for obj in objects:
|
||||
# if 'asset_type' not in obj['tags']:
|
||||
# if '/endpoint/' in obj['file_path']:
|
||||
# obj['tags']['asset_type'] = 'Endpoint'
|
||||
|
||||
def fix_observables(self, objects : list) -> None:
|
||||
for obj in objects:
|
||||
if 'observable' in obj['tags']:
|
||||
for observable in obj['tags']['observable']:
|
||||
if observable['type'] == 'Parent Process':
|
||||
observable['type'] = 'Process'
|
||||
if observable['type'] == 'user':
|
||||
observable['type'] = 'User'
|
||||
if observable['type'] == 'process name':
|
||||
observable['type'] = 'Process'
|
||||
# def fix_observables(self, objects : list) -> None:
|
||||
# for obj in objects:
|
||||
# if 'observable' in obj['tags']:
|
||||
# for observable in obj['tags']['observable']:
|
||||
# if observable['type'] == 'Parent Process':
|
||||
# observable['type'] = 'Process'
|
||||
# if observable['type'] == 'user':
|
||||
# observable['type'] = 'User'
|
||||
# if observable['type'] == 'process name':
|
||||
# observable['type'] = 'Process'
|
||||
|
||||
def fix_context(self, objects : list) -> None:
|
||||
for obj in objects:
|
||||
if 'context' in obj['tags']:
|
||||
new_context = []
|
||||
for context in obj['tags']['context']:
|
||||
if context == 'Stage:Exploitation':
|
||||
context = 'Stage:Execution'
|
||||
new_context.append(context)
|
||||
# def fix_context(self, objects : list) -> None:
|
||||
# for obj in objects:
|
||||
# if 'context' in obj['tags']:
|
||||
# new_context = []
|
||||
# for context in obj['tags']['context']:
|
||||
# if context == 'Stage:Exploitation':
|
||||
# context = 'Stage:Execution'
|
||||
# new_context.append(context)
|
||||
|
||||
obj['tags']['context'] = list(dict.fromkeys(new_context))
|
||||
# obj['tags']['context'] = list(dict.fromkeys(new_context))
|
||||
|
||||
def add_default_values_deprecated(self, objects : list) -> None:
|
||||
for obj in objects:
|
||||
if 'context' not in obj['tags']:
|
||||
obj['tags']['context'] = ['Unknown']
|
||||
if 'message' not in obj['tags']:
|
||||
obj['tags']['message'] = 'tbd'
|
||||
if 'observable' not in obj['tags']:
|
||||
obj['tags']['observable'] = [{'name': 'field', 'type': 'Unknown', 'role': ['Unknown']}]
|
||||
# def add_default_values_deprecated(self, objects : list) -> None:
|
||||
# for obj in objects:
|
||||
# if 'context' not in obj['tags']:
|
||||
# obj['tags']['context'] = ['Unknown']
|
||||
# if 'message' not in obj['tags']:
|
||||
# obj['tags']['message'] = 'tbd'
|
||||
# if 'observable' not in obj['tags']:
|
||||
# obj['tags']['observable'] = [{'name': 'field', 'type': 'Unknown', 'role': ['Unknown']}]
|
||||
|
||||
def fix_story(self, objects : list) -> None:
|
||||
for obj in objects:
|
||||
if 'type' not in obj:
|
||||
print(obj['name'])
|
||||
if isinstance(obj['tags']['analytic_story'], list):
|
||||
obj['tags']['analytic_story'] = obj['tags']['analytic_story'][0]
|
||||
# def fix_story(self, objects : list) -> None:
|
||||
# for obj in objects:
|
||||
# if 'type' not in obj:
|
||||
# print(obj['name'])
|
||||
# if isinstance(obj['tags']['analytic_story'], list):
|
||||
# obj['tags']['analytic_story'] = obj['tags']['analytic_story'][0]
|
||||
|
||||
def remove_SAAWS(self, objects : list) -> None:
|
||||
for obj in objects:
|
||||
if 'Splunk Security Analytics for AWS' in obj['tags']['product']:
|
||||
obj['tags']['product'].remove('Splunk Security Analytics for AWS')
|
||||
# def remove_SAAWS(self, objects : list) -> None:
|
||||
# for obj in objects:
|
||||
# if 'Splunk Security Analytics for AWS' in obj['tags']['product']:
|
||||
# obj['tags']['product'].remove('Splunk Security Analytics for AWS')
|
||||
|
||||
def remove_testing_passed(self, objects : list) -> None:
|
||||
for obj in objects:
|
||||
if 'automated_detection_testing' in obj['tags']:
|
||||
obj['tags'].pop('automated_detection_testing')
|
||||
# def remove_testing_passed(self, objects : list) -> None:
|
||||
# for obj in objects:
|
||||
# if 'automated_detection_testing' in obj['tags']:
|
||||
# obj['tags'].pop('automated_detection_testing')
|
||||
|
||||
def change_test_file_format(self, objects : list) -> None:
|
||||
for obj in objects:
|
||||
obj['name'] = obj['name'] + ' Unit Test'
|
||||
# def change_test_file_format(self, objects : list) -> None:
|
||||
# for obj in objects:
|
||||
# obj['name'] = obj['name'] + ' Unit Test'
|
||||
|
||||
def fix_kill_chain(self, objects : list) -> None:
|
||||
for obj in objects:
|
||||
if 'kill_chain_phases' in obj['tags']:
|
||||
if obj['tags']['kill_chain_phases'] == 'Exploitation':
|
||||
obj['tags']['kill_chain_phases'] = ['Exploitation']
|
||||
# def fix_kill_chain(self, objects : list) -> None:
|
||||
# for obj in objects:
|
||||
# if 'kill_chain_phases' in obj['tags']:
|
||||
# if obj['tags']['kill_chain_phases'] == 'Exploitation':
|
||||
# obj['tags']['kill_chain_phases'] = ['Exploitation']
|
||||
|
||||
def add_default_confidence_impact_risk_score(self, objects : list) -> None:
|
||||
for obj in objects:
|
||||
updated = True
|
||||
if not 'confidence' in obj['tags']:
|
||||
updated = True
|
||||
obj['tags']['confidence'] = 50
|
||||
if not 'impact' in obj['tags']:
|
||||
updated = True
|
||||
obj['tags']['impact'] = 50
|
||||
if not 'risk_score' in obj['tags'] or updated == True:
|
||||
#Recalculate the risk score if we have added/updated
|
||||
#the confidence or impact fields OR the risk_score
|
||||
#was missing in the first place
|
||||
self.fix_wrong_calculated_risk_score([obj])
|
||||
# def add_default_confidence_impact_risk_score(self, objects : list) -> None:
|
||||
# for obj in objects:
|
||||
# updated = True
|
||||
# if not 'confidence' in obj['tags']:
|
||||
# updated = True
|
||||
# obj['tags']['confidence'] = 50
|
||||
# if not 'impact' in obj['tags']:
|
||||
# updated = True
|
||||
# obj['tags']['impact'] = 50
|
||||
# if not 'risk_score' in obj['tags'] or updated == True:
|
||||
# #Recalculate the risk score if we have added/updated
|
||||
# #the confidence or impact fields OR the risk_score
|
||||
# #was missing in the first place
|
||||
# self.fix_wrong_calculated_risk_score([obj])
|
||||
|
||||
def fix_cc(self, objects : list) -> None:
|
||||
# def fix_cc(self, objects : list) -> None:
|
||||
# for obj in objects:
|
||||
# if 'Command & Control' in obj['tags']['analytic_story']:
|
||||
# obj['tags']['analytic_story'].remove('Command & Control')
|
||||
# obj['tags']['analytic_story'].append('Command And Control')
|
||||
|
||||
def update_description(self, objects : list, input_vars: list) -> None:
|
||||
for obj in objects:
|
||||
if 'Command & Control' in obj['tags']['analytic_story']:
|
||||
obj['tags']['analytic_story'].remove('Command & Control')
|
||||
obj['tags']['analytic_story'].append('Command And Control')
|
||||
obj["description"] = input_vars[0]
|
||||
+15
-6
@@ -84,7 +84,10 @@ def content_changer(args) -> None:
|
||||
input_dto = ContentChangerInputDto(
|
||||
ObjToYmlAdapter(args.path),
|
||||
factory_input_dto,
|
||||
args.change_function
|
||||
args.change_function,
|
||||
args.filter_key,
|
||||
args.filter_value,
|
||||
args.variables
|
||||
)
|
||||
|
||||
content_changer = ContentChanger()
|
||||
@@ -354,7 +357,7 @@ def main(args):
|
||||
# "This allows a user to easily add their own content and, eventually, "
|
||||
# "build a custom application consisting of their custom content.")
|
||||
new_content_parser = actions_parser.add_parser("new_content", help="Create new security content object")
|
||||
#content_changer_parser = actions_parser.add_parser("content_changer", help="Change Security Content based on defined rules")
|
||||
content_changer_parser = actions_parser.add_parser("content_changer", help="Change Security Content based on defined rules")
|
||||
validate_parser = actions_parser.add_parser("validate", help="Validates written content")
|
||||
generate_parser = actions_parser.add_parser("generate", help="Generates a deployment package for different platforms (splunk_app)")
|
||||
#docgen_parser = actions_parser.add_parser("docgen", help="Generates documentation")
|
||||
@@ -391,11 +394,17 @@ def main(args):
|
||||
help="Type of package to create, choose between `ESCU`, `SSA` or `API`.")
|
||||
generate_parser.set_defaults(func=generate)
|
||||
|
||||
# content_changer_choices = ContentChanger.enumerate_content_changer_functions()
|
||||
# content_changer_parser.add_argument("-cf", "--change_function", required=True, metavar='{ ' + ', '.join(content_changer_choices) +' }' , type=str, choices=content_changer_choices,
|
||||
# help= "Choose from the functions above defined in \nbin/contentctl_core/contentctl/application/use_cases/content_changer.py")
|
||||
content_changer_choices = ContentChanger.enumerate_content_changer_functions()
|
||||
content_changer_parser.add_argument("-cf", "--change_function", required=True, metavar='{ ' + ', '.join(content_changer_choices) +' }' , type=str, choices=content_changer_choices,
|
||||
help= "Choose from the functions above defined in \nbin/contentctl_core/contentctl/application/use_cases/content_changer.py")
|
||||
content_changer_parser.add_argument("-fk", "--filter_key", required=False, type=str,
|
||||
help= "Limit the chnage only on objects which contains the given key and value")
|
||||
content_changer_parser.add_argument("-fv", "--filter_value", required=False, type=str,
|
||||
help= "Limit the chnage only on objects which contains the given key and value")
|
||||
content_changer_parser.add_argument("variables", metavar="N", type=str, nargs='*',
|
||||
help= "List of input variables for content changer function")
|
||||
|
||||
# content_changer_parser.set_defaults(func=content_changer)
|
||||
content_changer_parser.set_defaults(func=content_changer)
|
||||
|
||||
# docgen_parser.add_argument("-o", "--output", required=True, type=str,
|
||||
# help="Path where to store the documentation")
|
||||
|
||||
Reference in New Issue
Block a user