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:
srv-rr-gh-researchbt
@@ -54,6 +54,7 @@ The Content Control tool allows you to manipulate Splunk Security Content via th
|
||||
5. **inspect** - Uses a local version of appinspect to ensure that the app you built meets basic quality standards.
|
||||
6. **cloud_deploy** - Using ACS, deploy your custom app to a running Splunk Cloud Instance.
|
||||
7. **convert** - Convert a detection rule with sigma syntax to a Splunk SPL detection
|
||||
8. **content_changer** - Perform changes on security content
|
||||
|
||||
### pre-requisites
|
||||
Make sure you use python version 3.9.
|
||||
@@ -90,6 +91,11 @@ Detection rule using raw:
|
||||
Detection rule converted to Windows Security Event Code 4688:
|
||||
`python contentctl.py -p . convert -dm raw -lo "Windows Security 4688" -o detections/endpoint/ -dp dev/endpoint/attempted_credential_dump_from_registry_via_reg_exe.yml`
|
||||
|
||||
### perform changes on security content
|
||||
Content changer will perform a change function defined in [here](https://github.com/splunk/security_content/blob/add_content_changer/bin/contentctl_project/contentctl_core/application/use_cases/content_changer.py) on all content or the content defined through the filter condition:
|
||||
`python contentctl.py -p detections/endpoint content_changer --change_function update_description --filter_key name --filter_value "3CX Supply Chain Attack Network Indicators" "Hello World"`
|
||||
|
||||
|
||||
# MITRE ATT&CK ⚔️
|
||||
### Detection Coverage
|
||||
To view an up-to-date detection coverage map for all the content tagged with MITRE techniques visit: [https://mitremap.splunkresearch.com/](https://mitremap.splunkresearch.com/) under the **Detection Coverage** layer. Below is a snapshot in time of what technique we currently have some detection coverage for. The darker the shade of blue the more detections we have for this particular technique. This map is automatically updated on every release and generated from the [generate-coverage-map.py](https://github.com/splunk/security_content/blob/develop/bin/generate-coverage-map.py).
|
||||
|
||||
@@ -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