Merge pull request #2069 from splunk/revert-1954-refactored_security_content

Revert "Refactored security content"
This commit is contained in:
Bhavin Patel
2022-03-04 12:21:04 -08:00
committed by GitHub
3129 changed files with 1156051 additions and 653634 deletions
+99 -56
View File
@@ -10,6 +10,8 @@
#https://docs.github.com/en/actions/guides/storing-workflow-data-as-artifacts
#In CircleCI, this was different (store_artifacts vs persist_to_workspace)
name: build-and-validate
on:
push:
@@ -80,41 +82,44 @@ jobs:
source venv/bin/activate
python3 -m pip install --upgrade pip
python3 -m pip install wheel
python3 -m pip install -q -r bin/contentctl_project/requirements.txt
python3 -m pip install -q -r requirements.txt
- name: run validate
run: |
source venv/bin/activate
python3 bin/contentctl_project/main/contentctl.py -p . validate -pr ESCU
python3 bin/contentctl_project/main/contentctl.py -p . validate -pr SSA
# - name: Get CTI Repo for Mitre context
# uses: actions/checkout@v2
# with:
# repository: mitre/cti #check out https://github.com/mitre/cti.git, defaults to HEAD
# path: "cti/"
python3 contentctl.py --path . --verbose validate
- name: Get CTI Repo for Mitre context
uses: actions/checkout@v2
with:
repository: mitre/cti #check out https://github.com/mitre/cti.git, defaults to HEAD
path: "cti/"
#Now generate the documentation (uses Node)
- uses: actions/setup-node@v2
with:
node-version: '14' #can easily be changed to a different version
- name: Generate documentation
run: |
ls -lah
#Enter the virtualenv and run the docgen
source venv/bin/activate
python3 bin/contentctl_project/main/contentctl.py -p . docgen -o docs
python3 bin/doc_gen.py --path . --output docs -v
#Now generate the spec docs
npm install -g @adobe/jsonschema2md
jsonschema2md -d spec -o docs/spec -f yaml -e spec.json -x -
#Clean up extra properties on docs
rm -rf docs/spec/*-*.md
echo "****** BRANCH INFORMATION ******"
git branch
git branch --show-current
build-sources:
runs-on: ubuntu-latest
needs: validate-content
@@ -122,16 +127,11 @@ jobs:
- name: Checkout Repo
uses: actions/checkout@v2
- uses: actions/setup-python@v2
with:
python-version: '3.9' #Available versions here - https://github.com/actions/python-versions/releases easy to change/make a matrix/use pypy
architecture: 'x64' # optional x64 or x86. Defaults to x64 if not specified
- name: Install System Packages
run: |
sudo apt update -qq
sudo apt install jq -qq
- name: Install Python Dependencies
run: |
#Get the virtualenv set up
@@ -140,14 +140,23 @@ jobs:
source venv/bin/activate
python3 -m pip install --upgrade pip
python3 -m pip install wheel
python3 -m pip install -q -r bin/contentctl_project/requirements.txt
python3 -m pip install -q -r requirements.txt
- name: Run Generate
run: |
source venv/bin/activate
python3 bin/contentctl_project/main/contentctl.py --path . generate --product ESCU --output dist/escu
python3 bin/contentctl_project/main/contentctl.py --path . generate --product SSA --output dist/ssa
python3 contentctl.py --path . --verbose generate --product ESCU --output dist/escu
python3 contentctl.py --path . --verbose generate --product SSA --output dist/ssa
- name: Copy lookups .csv files
run: |
# clean up current lookups
rm -rf dist/escu/lookups
mkdir dist/escu/lookups
#copy over lookups
cd lookups
cp -rv *.csv ../dist/escu/lookups
- name: Copy lookups .mlmodel files
run: |
cd lookups
@@ -156,7 +165,7 @@ jobs:
then cp -rv *.mlmodel ../dist/escu/lookups
else echo "No mlmodel files to copy"
fi
#Tag is '' for non-tagged push and the tag name for a tagged release
- name: Set tag
id: vars
@@ -170,7 +179,7 @@ jobs:
echo "Release is NOT TAGGED!"
echo "::set-output name=tag::"
fi
- name: Update Version and Build number
run : |
# check if tag is set, get build number from the tag if set
@@ -187,9 +196,10 @@ jobs:
sed -i "s/\"version\": .*$/\"version\": \"$CONTENT_VERSION\"/g" dist/escu/app.manifest
sed -i "s/version = .*$/version = $CONTENT_VERSION/g" dist/escu/default/content-version.conf
tar -czf content-pack-build-escu.tar.gz dist/escu/*
# update build number and version for ssa
tar -czf content-pack-build-ssa.tar.gz dist/ssa/*
- name: Persist to Workspace
uses: actions/upload-artifact@v2
with:
@@ -197,7 +207,8 @@ jobs:
path: |
content-pack-build-escu.tar.gz
content-pack-build-ssa.tar.gz
build-package:
runs-on: ubuntu-latest
needs: [validate-content, build-sources]
@@ -217,13 +228,13 @@ jobs:
- name: Get virtualenv for Python 2.7
run: |
sudo apt install virtualenv
- name: Grab Splunk Packaging Toolkit
run : |
curl -Ls https://download.splunk.com/misc/packaging-toolkit/splunk-packaging-toolkit-0.9.0.tar.gz -o splunk-packaging-toolkit-latest.tar.gz
mkdir slim-latest
tar -zxf splunk-packaging-toolkit-latest.tar.gz -C slim-latest --strip-components=1
- name: Install Splunk Packaging Toolkit (slim)
run: |
cd slim-latest
@@ -233,32 +244,36 @@ jobs:
python2 -m pip install wheel
python2 -m pip install semantic_version
python2 -m pip install .
- name: Create a .spl for this Build Using Slim
run: |
source slim-latest/venv/bin/activate
cd build
tar -zxf content-pack-build-escu.tar.gz
tar -zxf content-pack-build-ssa.tar.gz
mv dist/escu DA-ESS-ContentUpdate
mv dist/ssa SSA_Content
slim package -o upload DA-ESS-ContentUpdate
cp upload/DA-ESS-ContentUpdate-*.tar.gz DA-ESS-ContentUpdate-latest.tar.gz
sha256sum DA-ESS-ContentUpdate-latest.tar.gz > checksum.txt
#Do this copy so that we conform as much as possible, and have to make
#as few changes as possible, once we start generating this as a real,
#properly packaged app
tar -zcf upload/SSA_Content-NO_SLIM.tar.gz SSA_Content
cp upload/SSA_Content-*.tar.gz SSA_Content-latest.tar.gz
sha256sum SSA_Content-latest.tar.gz >> checksum.txt
- name: store_artifacts
uses: actions/upload-artifact@v2
with:
name: package
path: |
build/upload
- name: store_artifacts_two
uses: actions/upload-artifact@v2
with:
@@ -267,8 +282,13 @@ jobs:
build/DA-ESS-ContentUpdate-latest.tar.gz
build/SSA_Content-latest.tar.gz
build/checksum.txt
#Everything below this line should ONLY run on a tag and nothing else
#We still want all of the above checks to run and pass before running these
run-appinspect:
@@ -289,11 +309,16 @@ jobs:
name: content-latest
path: build/
- name: Install System Packages
run: |
sudo apt update -qq
sudo apt install jq -qq
- name: Submit ESCU to AppInspect API
env:
APPINSPECT_USERNAME: ${{ secrets.AppInspectUsername }}
@@ -303,20 +328,20 @@ jobs:
#Enclose in quotes in case there are any special characters in the username/password
#Better not to pass these arguments on the command line, if possible
./appinspect.sh ../ DA-ESS-ContentUpdate-latest.tar.gz "$APPINSPECT_USERNAME" "$APPINSPECT_PASSWORD"
- name: Create report artifact
if: always()
run: |
#Always create this, regardless of whether success or failure above
tar -cvzf report.tar.gz report/
- name: store_artifacts
uses: actions/upload-artifact@v2
with:
name: appinspect_reports
path: |
report.tar.gz
#Still store the report, even if we have failed (otherwise we don't know why/how we failed)
- name: store_artifacts_on_failure
uses: actions/upload-artifact@v2
@@ -325,8 +350,7 @@ jobs:
name: appinspect_reports_failure
path: |
report.tar.gz
create-report:
runs-on: ubuntu-latest
needs: [validate-content, build-sources, build-package, run-appinspect]
@@ -338,11 +362,12 @@ jobs:
with:
ref: 'develop'
- name: Install System Packages
run: |
sudo apt update -qq
sudo apt install jq -qq
- uses: actions/setup-python@v2
with:
python-version: '3.9' #Available versions here - https://github.com/actions/python-versions/releases easy to change/make a matrix/use pypy
@@ -356,13 +381,13 @@ jobs:
source venv/bin/activate
python3 -m pip install --upgrade pip
python3 -m pip install wheel
python3 -m pip install -q -r bin/contentctl_project/requirements.txt
python3 -m pip install -q -r requirements.txt
- name: run reporting
run: |
source venv/bin/activate
python3 bin/contentctl_project/main/contentctl.py -p . reporting
python3 bin/reporting.py
#Official, Verified Amazon-AWS Github Account Provided Action
- uses: aws-actions/configure-aws-credentials@v1
with:
@@ -374,7 +399,7 @@ jobs:
- name: Upload Reporting
run: |
aws s3 cp bin/reporting s3://security-content/reporting --recursive --exclude "*" --include "*.svg"
update-sources-github:
runs-on: ubuntu-latest
needs: [validate-content, build-sources, build-package, run-appinspect, create-report]
@@ -397,13 +422,13 @@ jobs:
with:
name: content-latest
- name: Stage artifacts in proper directories
run: |
mkdir latest-escu
tar -zxf DA-ESS-ContentUpdate-latest.tar.gz -C latest-escu --strip-components=1
mkdir latest-ssa
tar -zxf SSA_Content-latest.tar.gz -C latest-ssa --strip-components=1
- name: Install Python Dependencies
run: |
#Get the virtualenv set up
@@ -412,13 +437,19 @@ jobs:
source venv/bin/activate
python3 -m pip install --upgrade pip
python3 -m pip install wheel
python3 -m pip install -q -r bin/contentctl_project/requirements.txt
python3 -m pip install -q -r requirements.txt
- name: Get CTI Repo for Mitre context
uses: actions/checkout@v2
with:
repository: mitre/cti #check out https://github.com/mitre/cti.git, defaults to HEAD
path: "cti/"
- name: Get branch and PR required for detection testing main.py
id: vars
run: |
echo "::set-output name=branch::${GITHUB_REF#refs/heads/}"
- name: Run doc-gen
run: |
source venv/bin/activate
@@ -461,7 +492,8 @@ jobs:
# Push quietly to prevent showing the token in log
#No need to provide any credentials
git push
publish-github-release:
#Github-maintained release action is in archived state: https://github.com/actions/create-release
#They recommend several and we use the following with the most stars: https://github.com/softprops/action-gh-release
@@ -480,6 +512,8 @@ jobs:
with:
name: appinspect_reports
#Rename those artifacts appropriately
- name: Set tag
id: vars
@@ -489,10 +523,12 @@ jobs:
run: |
cp DA-ESS-ContentUpdate-latest.tar.gz DA-ESS-ContentUpdate-${{ steps.vars.outputs.tag }}.tar.gz
cp SSA_Content-latest.tar.gz SSA_Content-${{ steps.vars.outputs.tag }}.tar.gz
#No checksum on the reports
cp report.tar.gz report-${{ steps.vars.outputs.tag }}.tar.gz
cp checksum.txt checksum-${{ steps.vars.outputs.tag }}.txt
#Upload all of the artifacts that we have created using the third party
#action recommended bu Github
- name: Upload Release Artifacts
@@ -503,7 +539,9 @@ jobs:
SSA_Content-${{ steps.vars.outputs.tag }}.tar.gz
report-${{ steps.vars.outputs.tag }}.tar.gz
checksum-${{ steps.vars.outputs.tag }}.txt
attack-range-update:
runs-on: ubuntu-latest
needs: [validate-content, build-sources, build-package, run-appinspect, create-report, update-sources-github, publish-github-release]
@@ -516,6 +554,7 @@ jobs:
with:
name: content-latest
#Official, Verified Amazon-AWS Github Account Provided Action
- uses: aws-actions/configure-aws-credentials@v1
with:
@@ -529,7 +568,7 @@ jobs:
aws s3 cp DA-ESS-ContentUpdate-latest.tar.gz s3://attack-range-appbinaries/
# make the file public since it is not by default
aws s3api put-object-acl --bucket attack-range-appbinaries --key DA-ESS-ContentUpdate-latest.tar.gz --acl public-read
master-api-update:
runs-on: ubuntu-latest
needs: [validate-content, build-sources, build-package, run-appinspect, create-report, update-sources-github, publish-github-release, attack-range-update]
@@ -555,13 +594,18 @@ jobs:
source venv/bin/activate
python3 -m pip install --upgrade pip
python3 -m pip install wheel
python3 -m pip install -q -r bin/contentctl_project/requirements.txt
python3 -m pip install -q -r requirements.txt
- name: Create Baseline Folder
run: |
source venv/bin/activate
python3 bin/create_baseline_folder.py
- name: Create YML to JSON Folder
run: |
source venv/bin/activate
python3 bin/contentctl_project/main/contentctl.py --path . generate --product API --output dist/api
python3 contentctl.py --path . --verbose generate --product API --output dist/api
#Official, Verified Amazon-AWS Github Account Provided Action
- uses: aws-actions/configure-aws-credentials@v1
with:
@@ -583,7 +627,6 @@ jobs:
aws s3 cp macros s3://security-content/macros --recursive --exclude "*" --include "*.yml"
aws s3 cp deployments s3://security-content/deployments --recursive --exclude "*" --include "*.yml"
aws s3 cp dist/api s3://security-content/json --recursive --exclude "*" --include "*.json"
- name: Security Content API Smoke Test
run: |
API_URL='https://content.splunkresearch.com/detections'
@@ -592,4 +635,4 @@ jobs:
if [ "$API_STATUS" != "200" ]; then
echo "Error [Security Content API status: $API_STATUS]"
exit 1
fi
fi
-6
View File
@@ -185,12 +185,6 @@ jobs:
bin/automated_detection_testing/ci/detection_testing_batch/test_results/combined.json
bin/automated_detection_testing/ci/detection_testing_batch/test_results/summary.json
- name: Print Docker Logs on Failure
if: failure()
run: |
echo "PRINTING OUT THE DOCKER LOGS"
docker logs splunk_test_0
docker-detection-testing-execution-merge-results:
runs-on: ubuntu-latest
+1 -1
View File
@@ -115,7 +115,7 @@ If you have questions or need support, you can:
## License
Copyright 2022 Splunk Inc.
Copyright 2020 Splunk Inc.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
+75
View File
@@ -0,0 +1,75 @@
import glob
import yaml
import sys
import re
import argparse
from os import path
def load_objects(file_path, REPO_PATH):
files = []
manifest_files = path.join(path.expanduser(REPO_PATH), file_path)
for file in sorted(glob.glob(manifest_files)):
files.append(load_file(file))
return files
def load_file(file_path):
with open(file_path, 'r', encoding="utf-8") as stream:
try:
file = list(yaml.safe_load_all(stream))[0]
except yaml.YAMLError as exc:
print(exc)
sys.exit("ERROR: reading {0}".format(file_path))
return file
def load_content(old_project):
stories = load_objects("stories/*.yml", old_project)
macros = load_objects("macros/*.yml", old_project)
lookups = load_objects("lookups/*.yml", old_project)
baselines = load_objects("baselines/*.yml", old_project)
responses = load_objects("responses/*.yml", old_project)
response_tasks = load_objects("response_tasks/*.yml", old_project)
deployments = load_objects("deployments/*.yml", old_project)
# process all detections
detections = []
detections = load_objects("detections/*/*.yml", old_project)
detections.extend(load_objects("detections/*/*/*.yml", old_project))
#print(len(detections))
return detections, stories, macros, lookups, baselines, responses, response_tasks, deployments
def add_required_field(detections, new_project):
#for detection in detections:
matches = re.findall(r'(?<key>[^\s]*)=', detections[0])
for match in matches:
print(match)
def main(new_project, old_project, change):
detections, stories, macros, lookups, baselines, responses, response_tasks, deployments = load_content(old_project)
if change == "add_required_field":
add_required_field(detections, new_project)
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="applies security content changes to the whole project")
parser.add_argument("-np", "--new_project", required=True, help="the security content project to write the new configs in to")
parser.add_argument("-op", "--old_project", required=True, help="the security content project to read the files from")
parser.add_argument("-c", "--change", required=True, help="the name of your change")
# parse them
args = parser.parse_args()
new_project = args.new_project
old_project = args.old_project
change = args.change
main(new_project, old_project, change)
+218
View File
@@ -0,0 +1,218 @@
import glob
import yaml
import sys
import re
import argparse
import requests
import csv
from requests.auth import HTTPBasicAuth
from urllib3.exceptions import InsecureRequestWarning
from os import path
BASE_URL = f"https://ip:8089"
SEARCH_PARSER_ENDPOINT = f"/services/search/parser"
USER = f"admin"
PASSWORD = f"password"
parsed_fields = dict()
def load_objects(file_path, REPO_PATH):
files = []
manifest_files = path.join(path.expanduser(REPO_PATH), file_path)
for file in sorted(glob.glob(manifest_files)):
files.append(load_file(file))
return files
def load_file(file_path):
with open(file_path, 'r', encoding="utf-8") as stream:
try:
file = list(yaml.safe_load_all(stream))[0]
except yaml.YAMLError as exc:
print(exc)
sys.exit("ERROR: reading {0}".format(file_path))
return file
def load_content(old_project):
# process all detections
detections = []
detections = load_objects("detections/*/*.yml", old_project)
detections.extend(load_objects("detections/*/*/*.yml", old_project))
#print(len(detections))
return detections
def analysis_detection(detections):
for detection in detections:#
if detection['type'] != 'streaming':
#if detection['name'] == 'Attempted Credential Dump From Registry via Reg exe':
print('Analysis Detection: ' + detection['name'])
call_splunk_parser_api(detection)
# sort parsed fields by occurence
sorted_dict = {k: v for k, v in sorted(parsed_fields.items(), key=lambda item: item[1], reverse=True)}
with open('output_fields_ordered_by_usage.csv', mode='w') as csv_file:
writer = csv.writer(csv_file, delimiter=',', quotechar='"', quoting=csv.QUOTE_ALL)
writer.writerow(['field_name', 'occurence'])
for field_name in sorted_dict:
writer.writerow([field_name, sorted_dict[field_name]])
# sort parsed fields by name
sorted_dict_2 = sorted(parsed_fields.items())
with open('output_fields_ordered_by_keys.csv', mode='w') as csv_file:
writer = csv.writer(csv_file, delimiter=',', quotechar='"', quoting=csv.QUOTE_ALL)
writer.writerow(['field_name', 'occurence'])
for field_name in sorted_dict_2:
writer.writerow([field_name[0], field_name[1]])
def call_splunk_parser_api(detection):
requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning)
spl = ''
if detection['search'].startswith('| tstats'):
spl = detection['search']
else:
spl = 'search ' + detection['search']
data = {
"output_mode": "json",
"q": spl,
"parse_only": "true"
}
#Have semgrep ignore the following line. It will complain about the verify=false, but the server is hosted on localhost
response = requests.post(BASE_URL + SEARCH_PARSER_ENDPOINT, data=data, auth=(USER, PASSWORD), verify=False, headers={"Content-Type": "application/x-www-form-urlencoded"}) # nosemgrep
if response.status_code != 200:
print(response.json())
print('ERROR: parser endpoint problems')
return
parse_commands(response.json())
def parse_commands(api_response):
tmp_parsed_fields = {}
last_stat_command = ''
rename_command_after_stats_arr = []
for command in api_response['commands']:
if command['command'] in ['tstats', 'stats', 'table']:
last_stat_command = command
if (command['command'] == 'rename') and last_stat_command:
rename_command_after_stats_arr.append(command)
if not last_stat_command:
print('ERROR: could not find stats table or tasts command')
return
# last command table
if last_stat_command['command'] == 'table':
matches = re.findall(r'([0-9a-zA-Z_]+)', last_stat_command['rawargs'])
for match in matches:
if match in tmp_parsed_fields:
tmp_parsed_fields[match] = tmp_parsed_fields[match] + 1
else:
tmp_parsed_fields[match] = 1
# last command stats
if last_stat_command['command'] == 'stats':
match = re.match(r'(.*)by', last_stat_command['rawargs'])
if match:
args_one = match.group(1)
matches = re.findall(r'(?:min|max|values)\(([0-9a-zA-Z_]+)\)', args_one)
if matches:
for match in matches:
if match in tmp_parsed_fields:
tmp_parsed_fields[match] = tmp_parsed_fields[match] + 1
else:
tmp_parsed_fields[match] = 1
match = re.match(r'.*by(.*)$', last_stat_command['rawargs'])
if match:
args_two = match.group(1)
matches = re.findall(r'([0-9a-zA-Z_]+)', args_two)
if matches:
for match in matches:
if match in tmp_parsed_fields:
tmp_parsed_fields[match] = tmp_parsed_fields[match] + 1
else:
tmp_parsed_fields[match] = 1
# tstats command
if last_stat_command['command'] == 'tstats':
match = re.match(r'(.*)(?:from|FROM)', last_stat_command['rawargs'])
if match:
args_one = match.group(1)
matches = re.findall(r'(?:min|max|values)\(([0-9a-zA-Z_]+)\.([0-9a-zA-Z_]+)\)', args_one)
if matches:
for match in matches:
field = match[0] + '.' + match[1]
if field in tmp_parsed_fields:
tmp_parsed_fields[field] = tmp_parsed_fields[field] + 1
else:
tmp_parsed_fields[field] = 1
match = re.match(r'.*by(.*)$', last_stat_command['rawargs'])
if match:
args_two = match.group(1)
matches = re.findall(r'([0-9_a-zA-Z]+)\.([0-9a-zA-Z_]+)', args_two)
if matches:
for match in matches:
field = match[0] + '.' + match[1]
if field in tmp_parsed_fields:
tmp_parsed_fields[field] = tmp_parsed_fields[field] + 1
else:
tmp_parsed_fields[field] = 1
match = re.match(r'.*where(.*)by.*$', last_stat_command['rawargs'])
if match:
args_three = match.group(1)
matches = re.findall(r'([0-9_a-zA-Z]+)\.([0-9a-zA-Z_]+)=', args_three)
if matches:
for match in matches:
field = match[0] + '.' + match[1]
if field in tmp_parsed_fields:
tmp_parsed_fields[field] = tmp_parsed_fields[field] + 1
else:
tmp_parsed_fields[field] = 1
# rename occured
for rename_command_after_stats in rename_command_after_stats_arr:
if rename_command_after_stats:
renamed_field = {}
matches = re.findall(r'(?:(([0-9a-zA-Z_]+)\s+as\s+([0-9a-zA-Z_]+)))', rename_command_after_stats['rawargs'])
for match in matches:
renamed_field[match[1]] = match[2]
for key in renamed_field:
if key in tmp_parsed_fields:
tmp_parsed_fields[renamed_field[key]] = tmp_parsed_fields.pop(key)
# write to global parsed fields var
for key in tmp_parsed_fields:
if key in parsed_fields:
parsed_fields[key] = parsed_fields[key] + tmp_parsed_fields[key]
else:
parsed_fields[key] = tmp_parsed_fields[key]
def main(project):
detections = load_content(project)
analysis_detection(detections)
if __name__ == "__main__":
main("../")
@@ -1,9 +0,0 @@
import abc
from contentctl_core.domain.entities.enums.enums import SecurityContentType
class Adapter(abc.ABC):
@abc.abstractmethod
def writeObjects(self, objects: list, output_path: str, type: SecurityContentType = None) -> None:
pass
@@ -1,25 +0,0 @@
import abc
from contentctl_core.domain.entities.enums.enums import SecurityContentType
from contentctl_core.domain.entities.baseline import Baseline
from contentctl_core.domain.entities.enums.enums import SecurityContentProduct
from contentctl_core.domain.entities.security_content_object import SecurityContentObject
class BaselineBuilder(abc.ABC):
@abc.abstractmethod
def addDeployment(self, deployments: list) -> None:
pass
@abc.abstractmethod
def setObject(self, path: str) -> None:
pass
@abc.abstractmethod
def reset(self) -> None:
pass
@abc.abstractmethod
def getObject(self) -> SecurityContentObject:
pass
@@ -1,22 +0,0 @@
import abc
from contentctl_core.domain.entities.security_content_object import SecurityContentObject
from contentctl_core.domain.entities.enums.enums import SecurityContentType
# https://refactoring.guru/design-patterns/builder
class BasicBuilder(abc.ABC):
@abc.abstractmethod
def setObject(self, path: str, type: SecurityContentType) -> None:
pass
@abc.abstractmethod
def reset(self) -> None:
pass
@abc.abstractmethod
def getObject(self) -> SecurityContentObject:
pass
@@ -1,69 +0,0 @@
import abc
from contentctl_core.domain.entities.enums.enums import SecurityContentProduct
from contentctl_core.domain.entities.security_content_object import SecurityContentObject
from contentctl_core.domain.entities.enums.enums import SecurityContentType
# https://refactoring.guru/design-patterns/builder
class DetectionBuilder(abc.ABC):
@abc.abstractmethod
def addDeployment(self, deployments: list) -> None:
pass
@abc.abstractmethod
def addRBA(self) -> None:
pass
@abc.abstractmethod
def addNesFields(self) -> None:
pass
@abc.abstractmethod
def addMappings(self) -> None:
pass
@abc.abstractmethod
def addAnnotations(self) -> None:
pass
@abc.abstractmethod
def addPlaybook(self, playbooks: list) -> None:
pass
@abc.abstractmethod
def addBaseline(self, baselines: list) -> None:
pass
@abc.abstractmethod
def addUnitTest(self, tests: list) -> None:
pass
@abc.abstractmethod
def addMitreAttackEnrichment(self) -> None:
pass
@abc.abstractmethod
def addMacros(self, macros: list) -> None:
pass
@abc.abstractmethod
def addLookups(self, lookups: list) -> None:
pass
@abc.abstractmethod
def addCve(self) -> None:
pass
@abc.abstractmethod
def setObject(self, path: str) -> None:
pass
@abc.abstractmethod
def reset(self) -> None:
pass
@abc.abstractmethod
def getObject(self) -> SecurityContentObject:
pass
@@ -1,51 +0,0 @@
import abc
from contentctl_core.application.builder.basic_builder import BasicBuilder
from contentctl_core.application.builder.detection_builder import DetectionBuilder
from contentctl_core.application.builder.baseline_builder import BaselineBuilder
from contentctl_core.application.builder.investigation_builder import InvestigationBuilder
from contentctl_core.application.builder.story_builder import StoryBuilder
from contentctl_core.application.builder.playbook_builder import PlaybookBuilder
from contentctl_core.domain.entities.enums.enums import SecurityContentProduct
class Director(abc.ABC):
@abc.abstractmethod
def constructDetection(self, builder: DetectionBuilder, path: str, deployments: list, playbooks: list, baselines: list, tests: list, attack_enrichment: dict, macros: list) -> None:
pass
@abc.abstractmethod
def constructBaseline(self, builder: BaselineBuilder, path: str, deployments: list) -> None:
pass
@abc.abstractmethod
def constructDeployment(self, builder: BasicBuilder, path: str) -> None:
pass
@abc.abstractmethod
def constructLookup(self, builder: BasicBuilder, path: str) -> None:
pass
@abc.abstractmethod
def constructMacro(self, builder: BasicBuilder, path: str) -> None:
pass
@abc.abstractmethod
def constructPlaybook(self, builder: PlaybookBuilder, path: str, detections: list) -> None:
pass
@abc.abstractmethod
def constructTest(self, builder: BasicBuilder, path: str) -> None:
pass
@abc.abstractmethod
def constructStory(self, builder: StoryBuilder, path: str, detections: list, baselines: list, investigations: list) -> None:
pass
@abc.abstractmethod
def constructInvestigation(self, builder: InvestigationBuilder, path: str) -> None:
pass
@abc.abstractmethod
def constructObjects(self, builder: BasicBuilder, path: str) -> None:
pass
@@ -1,25 +0,0 @@
import abc
from contentctl_core.domain.entities.security_content_object import SecurityContentObject
class InvestigationBuilder(abc.ABC):
@abc.abstractmethod
def setObject(self, path: str) -> None:
pass
@abc.abstractmethod
def reset(self) -> None:
pass
@abc.abstractmethod
def getObject(self) -> SecurityContentObject:
pass
@abc.abstractmethod
def addInputs(self) -> None:
pass
@abc.abstractmethod
def addLowercaseName(self) -> None:
pass
@@ -1,23 +0,0 @@
import abc
from contentctl_core.domain.entities.security_content_object import SecurityContentObject
# https://refactoring.guru/design-patterns/builder
class PlaybookBuilder(abc.ABC):
@abc.abstractmethod
def setObject(self, path: str) -> None:
pass
@abc.abstractmethod
def addDetections(self, detections : list) -> None:
pass
@abc.abstractmethod
def reset(self) -> None:
pass
@abc.abstractmethod
def getObject(self) -> SecurityContentObject:
pass
@@ -1,39 +0,0 @@
import abc
from contentctl_core.domain.entities.security_content_object import SecurityContentObject
from contentctl_core.domain.entities.enums.enums import SecurityContentType
class StoryBuilder(abc.ABC):
@abc.abstractmethod
def addDetections(self, detections: list) -> None:
pass
@abc.abstractmethod
def addInvestigations(self, investigations: list) -> None:
pass
@abc.abstractmethod
def addAuthorCompanyName(self) -> None:
pass
@abc.abstractmethod
def addBaselines(self, baselines: list) -> None:
pass
@abc.abstractmethod
def addInvestigations(self, investigations: list) -> None:
pass
@abc.abstractmethod
def setObject(self, path: str) -> None:
pass
@abc.abstractmethod
def reset(self) -> None:
pass
@abc.abstractmethod
def getObject(self) -> SecurityContentObject:
pass
@@ -1,57 +0,0 @@
import os
from dataclasses import dataclass
from contentctl_core.domain.entities.enums.enums import SecurityContentType
from contentctl_core.application.builder.basic_builder import BasicBuilder
from contentctl_core.application.builder.detection_builder import DetectionBuilder
from contentctl_core.application.builder.story_builder import StoryBuilder
from contentctl_core.application.builder.director import Director
from contentctl_core.application.factory.utils.utils import Utils
@dataclass(frozen=True)
class BAFactoryInputDto:
input_path: str
basic_builder: BasicBuilder
detection_builder: DetectionBuilder
director: Director
@dataclass(frozen=True)
class BAFactoryOutputDto:
detections: list
tests: list
class BAFactory():
input_dto: BAFactoryInputDto
output_dto: BAFactoryOutputDto
def __init__(self, output_dto: BAFactoryOutputDto) -> None:
self.output_dto = output_dto
def execute(self, input_dto: BAFactoryInputDto) -> None:
self.input_dto = input_dto
self.createSecurityContent(SecurityContentType.unit_tests)
self.createSecurityContent(SecurityContentType.detections)
def createSecurityContent(self, type: SecurityContentType) -> list:
objects = []
if type == SecurityContentType.unit_tests:
files = Utils.get_all_yml_files_from_directory(os.path.join(self.input_dto.input_path, 'tests'))
else:
files = Utils.get_all_yml_files_from_directory(os.path.join(self.input_dto.input_path, str(type.name)))
for file in files:
if 'ssa__' in file:
if type == SecurityContentType.detections:
self.input_dto.director.constructDetection(self.input_dto.detection_builder, file, [], [], [], self.output_dto.tests, {}, [], [])
detection = self.input_dto.detection_builder.getObject()
if not detection.deprecated and not detection.experimental:
self.output_dto.detections.append(detection)
elif type == SecurityContentType.unit_tests:
self.input_dto.director.constructTest(self.input_dto.basic_builder, file)
test = self.input_dto.basic_builder.getObject()
self.output_dto.tests.append(test)
@@ -1,124 +0,0 @@
import os
from dataclasses import dataclass
from contentctl_core.domain.entities.enums.enums import SecurityContentProduct
from contentctl_core.domain.entities.enums.enums import SecurityContentType
from contentctl_core.application.builder.basic_builder import BasicBuilder
from contentctl_core.application.builder.detection_builder import DetectionBuilder
from contentctl_core.application.builder.story_builder import StoryBuilder
from contentctl_core.application.builder.baseline_builder import BaselineBuilder
from contentctl_core.application.builder.investigation_builder import InvestigationBuilder
from contentctl_core.application.builder.playbook_builder import PlaybookBuilder
from contentctl_core.application.builder.director import Director
from contentctl_core.application.factory.utils.utils import Utils
@dataclass(frozen=True)
class FactoryInputDto:
input_path: str
basic_builder: BasicBuilder
detection_builder: DetectionBuilder
story_builder: StoryBuilder
baseline_builder: BaselineBuilder
investigation_builder: InvestigationBuilder
playbook_builder: PlaybookBuilder
director: Director
attack_enrichment: dict
@dataclass()
class FactoryOutputDto:
detections: list
stories: list
baselines: list
investigations: list
playbooks: list
deployments: list
macros: list
lookups: list
tests: list
class Factory():
input_dto: FactoryInputDto
output_dto: FactoryOutputDto
def __init__(self, output_dto: FactoryOutputDto) -> None:
self.output_dto = output_dto
def execute(self, input_dto: FactoryInputDto) -> None:
self.input_dto = input_dto
# order matters to load and enrich security content types
self.createSecurityContent(SecurityContentType.unit_tests)
self.createSecurityContent(SecurityContentType.lookups)
self.createSecurityContent(SecurityContentType.macros)
self.createSecurityContent(SecurityContentType.deployments)
self.createSecurityContent(SecurityContentType.baselines)
self.createSecurityContent(SecurityContentType.investigations)
self.createSecurityContent(SecurityContentType.detections) # execution only for playbook enrichment
self.createSecurityContent(SecurityContentType.playbooks)
self.output_dto.detections = []
self.createSecurityContent(SecurityContentType.detections)
self.createSecurityContent(SecurityContentType.stories)
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'))
elif type == SecurityContentType.unit_tests:
files = Utils.get_all_yml_files_from_directory(os.path.join(self.input_dto.input_path, 'tests'))
else:
files = Utils.get_all_yml_files_from_directory(os.path.join(self.input_dto.input_path, str(type.name)))
for file in files:
if not 'ssa__' in file:
if type == SecurityContentType.lookups:
self.input_dto.director.constructLookup(self.input_dto.basic_builder, file)
self.output_dto.lookups.append(self.input_dto.basic_builder.getObject())
elif type == SecurityContentType.macros:
self.input_dto.director.constructMacro(self.input_dto.basic_builder, file)
self.output_dto.macros.append(self.input_dto.basic_builder.getObject())
elif type == SecurityContentType.deployments:
self.input_dto.director.constructDeployment(self.input_dto.basic_builder, file)
self.output_dto.deployments.append(self.input_dto.basic_builder.getObject())
elif type == SecurityContentType.playbooks:
self.input_dto.director.constructPlaybook(self.input_dto.playbook_builder, file, self.output_dto.detections)
self.output_dto.playbooks.append(self.input_dto.playbook_builder.getObject())
elif type == SecurityContentType.baselines:
self.input_dto.director.constructBaseline(self.input_dto.baseline_builder, file, self.output_dto.deployments)
baseline = self.input_dto.baseline_builder.getObject()
self.output_dto.baselines.append(baseline)
elif type == SecurityContentType.investigations:
self.input_dto.director.constructInvestigation(self.input_dto.investigation_builder, file)
investigation = self.input_dto.investigation_builder.getObject()
self.output_dto.investigations.append(investigation)
elif type == SecurityContentType.stories:
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()
self.output_dto.stories.append(story)
elif type == SecurityContentType.detections:
self.input_dto.director.constructDetection(self.input_dto.detection_builder, file,
self.output_dto.deployments, self.output_dto.playbooks, self.output_dto.baselines,
self.output_dto.tests, self.input_dto.attack_enrichment, self.output_dto.macros,
self.output_dto.lookups)
detection = self.input_dto.detection_builder.getObject()
self.output_dto.detections.append(detection)
elif type == SecurityContentType.unit_tests:
self.input_dto.director.constructTest(self.input_dto.basic_builder, file)
test = self.input_dto.basic_builder.getObject()
self.output_dto.tests.append(test)
@@ -1,79 +0,0 @@
import os
import uuid
from dataclasses import dataclass
from PyInquirer import prompt
from datetime import datetime
from contentctl_core.domain.entities.enums.enums import SecurityContentType
from contentctl_core.application.factory.utils.new_content_questions import NewContentQuestions
@dataclass(frozen=True)
class NewContentFactoryInputDto:
type: SecurityContentType
@dataclass(frozen=True)
class NewContentFactoryOutputDto:
obj: dict
class NewContentFactory():
def __init__(self, output_dto: NewContentFactoryOutputDto) -> None:
self.output_dto = output_dto
def execute(self, input_dto: NewContentFactoryInputDto) -> None:
if input_dto.type == SecurityContentType.detections:
questions = NewContentQuestions.get_questions_detection()
answers = prompt(questions)
self.output_dto.obj['name'] = answers['detection_name']
self.output_dto.obj['id'] = str(uuid.uuid4())
self.output_dto.obj['version'] = 1
self.output_dto.obj['date'] = datetime.today().strftime('%Y-%m-%d')
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`'
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']
self.output_dto.obj['tags'] = dict()
self.output_dto.obj['tags']['analytic_story'] = ['UPDATE_STORY_NAME']
self.output_dto.obj['tags']['asset_type'] = 'UPDATE asset_type'
self.output_dto.obj['tags']['cis20'] = ['CIS 3', 'CIS 5', 'CIS 16']
self.output_dto.obj['tags']['confidence'] = 'UPDATE value between 1-100'
self.output_dto.obj['tags']['context'] = ['Update context']
self.output_dto.obj['tags']['dataset'] = ['UPDATE_DATASET_URL']
self.output_dto.obj['tags']['impact'] = 'UPDATE value between 1-100'
self.output_dto.obj['tags']['kill_chain_phases'] = answers['kill_chain_phases']
self.output_dto.obj['tags']['message'] = 'UPDATE message'
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']
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()
answers = prompt(questions)
self.output_dto.obj['name'] = answers['story_name']
self.output_dto.obj['id'] = str(uuid.uuid4())
self.output_dto.obj['version'] = 1
self.output_dto.obj['date'] = datetime.today().strftime('%Y-%m-%d')
self.output_dto.obj['author'] = answers['story_author']
self.output_dto.obj['description'] = 'UPDATE_DESCRIPTION'
self.output_dto.obj['narrative'] = 'UPDATE_NARRATIVE'
self.output_dto.obj['references'] = []
self.output_dto.obj['tags']['analytic_story'] = self.output_dto.obj['name']
self.output_dto.obj['tags']['category'] = answers['category']
self.output_dto.obj['tags']['product'] = ['Splunk Enterprise','Splunk Enterprise Security','Splunk Cloud']
self.output_dto.obj['tags']['usecase'] = answers['usecase']
@@ -1,29 +0,0 @@
import os
from dataclasses import dataclass
from contentctl_core.application.builder.basic_builder import BasicBuilder
from contentctl_core.application.builder.director import Director
from contentctl_core.domain.entities.enums.enums import SecurityContentType
from contentctl_core.application.factory.utils.utils import Utils
@dataclass(frozen=True)
class ObjectFactoryInputDto:
input_path: str
builder: BasicBuilder
director: Director
class ObjectFactory():
objects: list
def __init__(self, objects: list) -> None:
self.objects = objects
def execute(self, input_dto: ObjectFactoryInputDto) -> None:
self.input_path = input_dto.input_path
files = Utils.get_all_yml_files_from_directory(input_dto.input_path)
for file in files:
input_dto.director.constructObjects(input_dto.builder, file)
self.objects.append(input_dto.builder.getObject())
@@ -1,266 +0,0 @@
class NewContentQuestions():
@classmethod
def get_questions_detection(self) -> list:
questions = [
{
'type': 'list',
'message': 'what kind of detection is this',
'name': 'detection_kind',
'choices': [
{
'name': 'endpoint'
},
{
'name': 'cloud'
},
{
'name': 'application'
},
{
'name': 'network'
},
{
'name': 'web'
},
{
'name': 'experimental'
},
],
'default': 'endpoint'
},
{
'type': 'input',
'message': 'enter detection name',
'name': 'detection_name',
'default': 'Powershell Encoded Command',
},
{
'type': 'input',
'message': 'enter author name',
'name': 'detection_author',
},
{
'type': 'list',
'message': 'select a detection type',
'name': 'detection_type',
'choices': [
{
'name': 'TTP'
},
{
'name': 'Anomaly'
},
{
'name': 'Hunting'
},
{
'name': 'Baseline'
},
{
'name': 'Investigation'
},
{
'name': 'Correlation'
}
],
'default': 'TTP'
},
{
'type': 'checkbox',
'message': 'select the datamodels used in the detection',
'name': 'datamodels',
'choices': [
{
'name': 'Endpoint',
'checked': True
},
{
'name': 'Authentication'
},
{
'name': 'Change'
},
{
'name': 'Email'
},
{
'name': 'Network_Resolution'
},
{
'name': 'Network_Traffic'
},
{
'name': 'Network_Sessions'
},
{
'name': 'Updates'
},
{
'name': 'Vulnerabilities'
},
{
'name': 'Web'
},
{
'name': 'Risk'
},
],
},
{
'type': 'input',
'message': 'enter search (spl)',
'name': 'detection_search',
'default': '| UPDATE_SPL'
},
{
'type': 'input',
'message': 'enter MITRE ATT&CK Technique IDs related to the detection, comma delimited for multiple',
'name': 'mitre_attack_ids',
'default': 'T1003.002'
},
{
'type': 'checkbox',
'message': 'select kill chain phases related to the detection',
'name': 'kill_chain_phases',
'choices': [
{
'name': 'Reconnaissance'
},
{
'name': 'Intrusion'
},
{
'name': 'Exploitation',
'checked': True
},
{
'name': 'Privilege Escalation'
},
{
'name': 'Lateral Movement'
},
{
'name': 'Obfuscation'
},
{
'name': 'Denial of Service'
},
{
'name': 'Exfiltration'
},
],
},
{
'type': 'list',
'message': 'security_domain for detection',
'name': 'security_domain',
'choices': [
{
'name': 'access'
},
{
'name': 'endpoint'
},
{
'name': 'network'
},
{
'name': 'threat'
},
{
'name': 'identity'
},
{
'name': 'audit'
},
],
'default': 'endpoint'
},
]
return questions
@classmethod
def get_questions_story(self) -> list:
questions = [
{
'type': 'input',
'message': 'enter story name',
'name': 'story_name',
'default': 'Suspicious Powershell Behavior',
},
{
'type': 'input',
'message': 'enter author name',
'name': 'story_author',
},
{
'type': 'checkbox',
'message': 'select a category',
'name': 'category',
'choices': [
{
'name': 'Adversary Tactics',
'checked': True
},
{
'name': 'Account Compromise'
},
{
'name': 'Unauthorized Software'
},
{
'name': 'Best Practices'
},
{
'name': 'Cloud Security'
},
{
'name': 'Command and Control'
},
{
'name': 'Lateral Movement'
},
{
'name': 'Ransomware'
},
{
'name': 'Privilege Escalation'
},
],
},
{
'type': 'list',
'message': 'select a use case',
'name': 'usecase',
'choices': [
{
'name': 'Advanced Threat Detection',
'checked': True
},
{
'name': 'Security Monitoring'
},
{
'name': 'Compliance'
},
{
'name': 'Insider Threat'
},
{
'name': 'Application Security'
},
{
'name': 'Other'
},
],
},
]
return questions
@@ -1,13 +0,0 @@
import os
class Utils:
@staticmethod
def get_all_yml_files_from_directory(path: str) -> list:
listOfFiles = list()
for (dirpath, dirnames, filenames) in os.walk(path):
for file in filenames:
if file.endswith(".yml"):
listOfFiles.append(os.path.join(dirpath, file))
return sorted(listOfFiles)
@@ -1,183 +0,0 @@
import re
import uuid
from dataclasses import dataclass
from contentctl_core.application.factory.object_factory import ObjectFactory, ObjectFactoryInputDto
from contentctl_core.application.adapter.adapter import Adapter
@dataclass(frozen=True)
class ContentChangerInputDto:
adapter : Adapter
factory_input_dto : ObjectFactoryInputDto
converter_func_name : str
class ContentChanger:
def execute(self, input_dto: ContentChangerInputDto) -> None:
objects = list()
factory = ObjectFactory(objects)
factory.execute(input_dto.factory_input_dto)
converter_func = getattr(self, input_dto.converter_func_name)
converter_func(objects)
input_dto.adapter.writeObjectsInPlace(objects)
# Define Converter Functions here
def example_converter_func(self, objects : list) -> None:
for obj in objects:
obj['author'] = obj['author'].upper()
def add_default_risk_values(self, objects : list) -> None:
for obj in objects:
if not 'confidence' in obj['tags']:
obj['tags']['confidence'] = 50
if not 'impact' in obj['tags']:
obj['tags']['impact'] = 50
if not 'risk_score' in obj['tags']:
obj['tags']['risk_score'] = 25
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_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_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_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 fix_wrong_calculated_risk_score(self, objects : list) -> None:
for obj in objects:
calculated_risk_score = (int(obj['tags']['impact']))*(int(obj['tags']['confidence']))/100
if calculated_risk_score != int(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 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)
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 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_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 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:
if 'confidence' not in obj['tags']:
obj['tags']['confidence'] = 50
if 'impact' not in obj['tags']:
obj['tags']['impact'] = 50
if 'risk_score' not in obj['tags']:
calculated_risk_score = (int(obj['tags']['impact']))*(int(obj['tags']['confidence']))/100
obj['tags']['risk_score'] = calculated_risk_score
@@ -1,23 +0,0 @@
from dataclasses import dataclass
from contentctl_core.application.factory.factory import FactoryInputDto, Factory, FactoryOutputDto
from contentctl_core.application.adapter.adapter import Adapter
@dataclass(frozen=True)
class DocGenInputDto:
output_path: str
factory_input_dto: FactoryInputDto
adapter : Adapter
class DocGen:
def execute(self, input_dto: DocGenInputDto) -> None:
factory_output_dto = FactoryOutputDto([],[],[],[],[],[],[],[],[])
factory = Factory(factory_output_dto)
factory.execute(input_dto.factory_input_dto)
input_dto.adapter.writeObjects([factory_output_dto.stories, factory_output_dto.detections, factory_output_dto.playbooks], input_dto.output_path)
@@ -1,58 +0,0 @@
import os
import shutil
from dataclasses import dataclass
from contentctl_core.domain.entities.enums.enums import SecurityContentProduct, SecurityContentType
from contentctl_core.application.adapter.adapter import Adapter
from contentctl_core.application.factory.factory import FactoryInputDto, Factory, FactoryOutputDto
from contentctl_core.application.factory.ba_factory import BAFactoryInputDto, BAFactory, BAFactoryOutputDto
@dataclass(frozen=True)
class GenerateInputDto:
output_path: str
factory_input_dto: FactoryInputDto
ba_factory_input_dto: BAFactoryInputDto
adapter : Adapter
product: SecurityContentProduct
class Generate:
def execute(self, input_dto: GenerateInputDto) -> None:
if input_dto.product == SecurityContentProduct.ESCU:
factory_output_dto = FactoryOutputDto([],[],[],[],[],[],[],[],[])
factory = Factory(factory_output_dto)
factory.execute(input_dto.factory_input_dto)
input_dto.adapter.writeHeaders(input_dto.output_path)
input_dto.adapter.writeObjects(factory_output_dto.detections, input_dto.output_path, SecurityContentType.detections)
input_dto.adapter.writeObjects(factory_output_dto.stories, input_dto.output_path, SecurityContentType.stories)
input_dto.adapter.writeObjects(factory_output_dto.baselines, input_dto.output_path, SecurityContentType.baselines)
input_dto.adapter.writeObjects(factory_output_dto.investigations, input_dto.output_path, SecurityContentType.investigations)
input_dto.adapter.writeObjects(factory_output_dto.lookups, input_dto.output_path, SecurityContentType.lookups)
input_dto.adapter.writeObjects(factory_output_dto.macros, input_dto.output_path, SecurityContentType.macros)
elif input_dto.product == SecurityContentProduct.SSA:
shutil.rmtree(input_dto.output_path + '/srs/', ignore_errors=True)
shutil.rmtree(input_dto.output_path + '/complex/', ignore_errors=True)
os.makedirs(input_dto.output_path + '/complex/')
os.makedirs(input_dto.output_path + '/srs/')
factory_output_dto = BAFactoryOutputDto([],[])
factory = BAFactory(factory_output_dto)
factory.execute(input_dto.ba_factory_input_dto)
input_dto.adapter.writeObjects(factory_output_dto.detections, input_dto.output_path)
elif input_dto.product == SecurityContentProduct.API:
factory_output_dto = FactoryOutputDto([],[],[],[],[],[],[],[],[])
factory = Factory(factory_output_dto)
factory.execute(input_dto.factory_input_dto)
input_dto.adapter.writeObjects(factory_output_dto.detections, input_dto.output_path, SecurityContentType.detections)
input_dto.adapter.writeObjects(factory_output_dto.stories, input_dto.output_path, SecurityContentType.stories)
input_dto.adapter.writeObjects(factory_output_dto.baselines, input_dto.output_path, SecurityContentType.baselines)
input_dto.adapter.writeObjects(factory_output_dto.investigations, input_dto.output_path, SecurityContentType.investigations)
input_dto.adapter.writeObjects(factory_output_dto.lookups, input_dto.output_path, SecurityContentType.lookups)
input_dto.adapter.writeObjects(factory_output_dto.macros, input_dto.output_path, SecurityContentType.macros)
input_dto.adapter.writeObjects(factory_output_dto.deployments, input_dto.output_path, SecurityContentType.deployments)
@@ -1,22 +0,0 @@
from dataclasses import dataclass
from contentctl_core.application.factory.new_content_factory import NewContentFactory, NewContentFactoryInputDto, NewContentFactoryOutputDto
from contentctl_core.application.adapter.adapter import Adapter
@dataclass(frozen=True)
class NewContentInputDto:
factory_input_dto: NewContentFactoryInputDto
adapter : Adapter
class NewContent:
def execute(self, input_dto: NewContentInputDto) -> None:
factory_output_dto = NewContentFactoryOutputDto(dict())
factory = NewContentFactory(factory_output_dto)
factory.execute(input_dto.factory_input_dto)
input_dto.adapter.writeObjectNewContent(factory_output_dto.obj, input_dto.factory_input_dto.type)
@@ -1,24 +0,0 @@
import os
from dataclasses import dataclass
from contentctl_core.application.factory.factory import FactoryInputDto, Factory, FactoryOutputDto
from contentctl_core.application.adapter.adapter import Adapter
@dataclass(frozen=True)
class ReportingInputDto:
factory_input_dto: FactoryInputDto
adapter_svg : Adapter
adapter_attack : Adapter
class Reporting:
def execute(self, input_dto: ReportingInputDto) -> None:
factory_output_dto = FactoryOutputDto([],[],[],[],[],[],[],[],[])
factory = Factory(factory_output_dto)
factory.execute(input_dto.factory_input_dto)
input_dto.adapter_svg.writeObjects(factory_output_dto.detections, os.path.join(os.path.dirname(__file__), '../../../../reporting'))
input_dto.adapter_attack.writeObjects(factory_output_dto.detections, os.path.join(os.path.dirname(__file__), '../../../../../docs/mitre-map'))
@@ -1,48 +0,0 @@
from dataclasses import dataclass
from pydantic import ValidationError
from contentctl_core.domain.entities.enums.enums import SecurityContentProduct, SecurityContentType
from contentctl_core.application.factory.factory import FactoryInputDto, Factory, FactoryOutputDto
from contentctl_core.application.factory.ba_factory import BAFactoryInputDto, BAFactory, BAFactoryOutputDto
@dataclass(frozen=True)
class ValidateInputDto:
factory_input_dto: FactoryInputDto
ba_factory_input_dto: BAFactoryInputDto
product: SecurityContentProduct
class Validate:
def execute(self, input_dto: ValidateInputDto) -> None:
if input_dto.product == SecurityContentProduct.ESCU:
factory_output_dto = FactoryOutputDto([],[],[],[],[],[],[],[],[])
factory = Factory(factory_output_dto)
factory.execute(input_dto.factory_input_dto)
elif input_dto.product == SecurityContentProduct.SSA:
factory_output_dto = BAFactoryOutputDto([],[])
factory = BAFactory(factory_output_dto)
factory.execute(input_dto.ba_factory_input_dto)
# validate detections
# validate tests
self.validate_detection_exist_for_test(factory_output_dto.tests, factory_output_dto.detections)
def validate_detection_exist_for_test(self, tests : list, detections: list):
for test in tests:
found_detection = False
for detection in detections:
if test.tests[0].file in detection.file_path:
found_detection = True
if not found_detection:
ValueError("detection doesn't exist for test file: " + test.name)
@@ -1,100 +0,0 @@
import string
import uuid
import requests
from pydantic import BaseModel, validator, ValidationError
from dataclasses import dataclass
from datetime import datetime
from contentctl_core.domain.entities.security_content_object import SecurityContentObject
from contentctl_core.domain.entities.enums.enums import DataModel
from contentctl_core.domain.entities.baseline_tags import BaselineTags
from contentctl_core.domain.entities.deployment import Deployment
class Baseline(BaseModel, SecurityContentObject):
# baseline spec
name: str
id: str
version: int
date: str
author: str
type: str
datamodel: list
description: str
search: str
how_to_implement: str
known_false_positives: str
references: list
tags: BaselineTags
# enrichment
deployment: Deployment = None
@validator('name')
def name_max_length(cls, v):
if len(v) > 75:
raise ValueError('name is longer then 75 chars: ' + v)
return v
@validator('name')
def name_invalid_chars(cls, v):
invalidChars = set(string.punctuation.replace("-", ""))
if any(char in invalidChars for char in v):
raise ValueError('invalid chars used in name: ' + v)
return v
@validator('id')
def id_check(cls, v, values):
try:
uuid.UUID(str(v))
except:
raise ValueError('uuid is not valid: ' + values["name"])
return v
@validator('date')
def date_valid(cls, v, values):
try:
datetime.strptime(v, "%Y-%m-%d")
except:
raise ValueError('date is not in format YYYY-MM-DD: ' + values["name"])
return v
@validator('type')
def type_valid(cls, v, values):
if v != "Baseline":
raise ValueError('not valid analytics type: ' + values["name"])
return v
@validator('datamodel')
def datamodel_valid(cls, v, values):
for datamodel in v:
if datamodel not in [el.name for el in DataModel]:
raise ValueError('not valid data model: ' + values["name"])
return v
@validator('description', 'how_to_implement')
def encode_error(cls, v, values, field):
try:
v.encode('ascii')
except UnicodeEncodeError:
raise ValueError('encoding error in ' + field.name + ': ' + values["name"])
return v
@validator('references')
def references_check(cls, v, values):
for reference in v:
try:
get = requests.get(reference)
if not get.status_code == 200:
raise ValueError('Reference ' + reference + ' is not reachable: ' + values["name"])
except requests.exceptions.RequestException as e:
raise ValueError('Reference ' + reference + ' is not reachable: ' + values["name"])
return v
@validator('search')
def search_validate(cls, v, values):
# write search validator
return v
@@ -1,25 +0,0 @@
from pydantic import BaseModel, validator, ValidationError
class BaselineTags(BaseModel):
analytic_story: list
deployments: list = None
detections: list
product: list
required_fields: list
security_domain: str
@validator('product')
def tags_product(cls, v, values):
valid_products = [
"Splunk Enterprise", "Splunk Enterprise Security", "Splunk Cloud",
"Splunk Security Analytics for AWS", "Splunk Behavioral Analytics"
]
for value in v:
if value not in valid_products:
raise ValueError('product is not valid for ' + values['name'] + '. valid products are ' + str(valid_products))
return v
@@ -1,60 +0,0 @@
import uuid
import string
from pydantic import BaseModel, validator, ValidationError
from datetime import datetime
from contentctl_core.domain.entities.security_content_object import SecurityContentObject
from contentctl_core.domain.entities.deployment_scheduling import DeploymentScheduling
from contentctl_core.domain.entities.deployment_email import DeploymentEmail
from contentctl_core.domain.entities.deployment_notable import DeploymentNotable
from contentctl_core.domain.entities.deployment_rba import DeploymentRBA
from contentctl_core.domain.entities.deployment_slack import DeploymentSlack
from contentctl_core.domain.entities.deployment_phantom import DeploymentPhantom
class Deployment(BaseModel, SecurityContentObject):
name: str
id: str
date: str
author: str
description: str
scheduling: DeploymentScheduling = None
email: DeploymentEmail = None
notable: DeploymentNotable = None
rba: DeploymentRBA = None
slack: DeploymentSlack = None
phantom: DeploymentPhantom = None
tags: dict
@validator('name')
def name_invalid_chars(cls, v):
invalidChars = set(string.punctuation.replace("-", ""))
if any(char in invalidChars for char in v):
raise ValueError('invalid chars used in name: ' + v)
return v
@validator('id')
def id_check(cls, v, values):
try:
uuid.UUID(str(v))
except:
raise ValueError('uuid is not valid: ' + values["name"])
return v
@validator('date')
def date_valid(cls, v, values):
try:
datetime.strptime(v, "%Y-%m-%d")
except:
raise ValueError('date is not in format YYYY-MM-DD: ' + values["name"])
return v
@validator('description')
def encode_error(cls, v, values, field):
try:
v.encode('ascii')
except UnicodeEncodeError:
raise ValueError('encoding error in ' + field.name + ': ' + values["name"])
return v
@@ -1,8 +0,0 @@
from pydantic import BaseModel, validator, ValidationError
class DeploymentEmail(BaseModel):
message: str
subject: str
to: str
@@ -1,8 +0,0 @@
from pydantic import BaseModel, validator, ValidationError
class DeploymentNotable(BaseModel):
rule_description: str
rule_title: str
nes_fields: list
@@ -1,10 +0,0 @@
from pydantic import BaseModel, validator, ValidationError
class DeploymentPhantom(BaseModel):
cam_workers : str
label : str
phantom_server : str
sensitivity : str
severity : str
@@ -1,7 +0,0 @@
from pydantic import BaseModel, validator, ValidationError
class DeploymentRBA(BaseModel):
enabled: str
@@ -1,10 +0,0 @@
from pydantic import BaseModel, validator, ValidationError
class DeploymentScheduling(BaseModel):
cron_schedule: str
earliest_time: str
latest_time: str
schedule_window: str
@@ -1,7 +0,0 @@
from pydantic import BaseModel, validator, ValidationError
class DeploymentSlack(BaseModel):
channel: str
message: str
@@ -1,133 +0,0 @@
import enum
import uuid
import string
import re
import requests
from pydantic import BaseModel, validator, root_validator
from dataclasses import dataclass
from datetime import datetime
from contentctl_core.domain.entities.security_content_object import SecurityContentObject
from contentctl_core.domain.entities.enums.enums import AnalyticsType
from contentctl_core.domain.entities.enums.enums import DataModel
from contentctl_core.domain.entities.detection_tags import DetectionTags
from contentctl_core.domain.entities.deployment import Deployment
from contentctl_core.domain.entities.unit_test import UnitTest
from contentctl_core.domain.entities.macro import Macro
from contentctl_core.domain.entities.lookup import Lookup
from contentctl_core.domain.entities.baseline import Baseline
from contentctl_core.domain.entities.playbook import Playbook
class Detection(BaseModel, SecurityContentObject):
# detection spec
name: str
id: str
version: int
date: str
author: str
type: str
datamodel: list
description: str
search: str
how_to_implement: str
known_false_positives: str
references: list
tags: DetectionTags
# enrichments
deprecated: bool = None
experimental: bool = None
deployment: Deployment = None
annotations: dict = None
risk: list = None
playbooks: list[Playbook] = None
baselines: list[Baseline] = None
mappings: dict = None
test: UnitTest = None
macros: list[Macro] = None
lookups: list[Lookup] = None
cve_enrichment: list = None
file_path: str = None
source: str = None
@validator('name')
def name_max_length(cls, v):
if len(v) > 75:
raise ValueError('name is longer then 75 chars: ' + v)
return v
@validator('name')
def name_invalid_chars(cls, v):
invalidChars = set(string.punctuation.replace("-", ""))
if any(char in invalidChars for char in v):
raise ValueError('invalid chars used in name: ' + v)
return v
@validator('id')
def id_check(cls, v, values):
try:
uuid.UUID(str(v))
except:
raise ValueError('uuid is not valid: ' + values["name"])
return v
@validator('date')
def date_valid(cls, v, values):
try:
datetime.strptime(v, "%Y-%m-%d")
except:
raise ValueError('date is not in format YYYY-MM-DD: ' + values["name"])
return v
@validator('type')
def type_valid(cls, v, values):
if v.lower() not in [el.name.lower() for el in AnalyticsType]:
raise ValueError('not valid analytics type: ' + values["name"])
return v
@validator('datamodel')
def datamodel_valid(cls, v, values):
for datamodel in v:
if datamodel not in [el.name for el in DataModel]:
raise ValueError('not valid data model: ' + values["name"])
return v
@validator('description', 'how_to_implement')
def encode_error(cls, v, values, field):
try:
v.encode('ascii')
except UnicodeEncodeError:
raise ValueError('encoding error in ' + field.name + ': ' + values["name"])
return v
@root_validator
def search_validation(cls, values):
if 'ssa_' not in values['file_path']:
if not '_filter' in values['search']:
raise ValueError('filter macro missing in: ' + values["name"])
if any(x in values['search'] for x in ['eventtype=', 'sourcetype=', ' source=', 'index=']):
if not 'index=_internal' in values['search']:
raise ValueError('Use source macro instead of eventtype, sourcetype, source or index in detection: ' + values["name"])
return values
# @validator('references')
# def references_check(cls, v, values):
# for reference in v:
# try:
# get = requests.get(reference)
# if not get.status_code == 200:
# raise ValueError('Reference ' + reference + ' is not reachable: ' + values["name"])
# except requests.exceptions.RequestException as e:
# raise ValueError('Reference ' + reference + ' is not reachable: ' + values["name"])
# return v
@validator('search')
def search_validate(cls, v, values):
# write search validator
return v
@@ -1,145 +0,0 @@
import re
from pydantic import BaseModel, validator, ValidationError
from contentctl_core.domain.entities.mitre_attack_enrichment import MitreAttackEnrichment
class DetectionTags(BaseModel):
# detection spec
name: str
analytic_story: list
asset_type: str
automated_detection_testing: str = None
cis20: list = None
confidence: str
context: list
dataset: list = None
impact: int
kill_chain_phases: list
message: str
mitre_attack_id: list = None
nist: list = None
observable: list
product: list
required_fields: list
risk_score: int
security_domain: str
risk_severity: str = None
cve: list = None
# enrichment
mitre_attack_enrichments: list[MitreAttackEnrichment] = None
@validator('cis20')
def tags_cis20(cls, v, values):
pattern = 'CIS [0-9]{1,2}'
for value in v:
if not re.match(pattern, value):
raise ValueError('CIS controls are not following the pattern CIS xx: ' + values["name"])
return v
@validator('confidence')
def tags_confidence(cls, v, values):
v = int(v)
if not (v > 0 and v <= 100):
raise ValueError('confidence score is out of range 1-100: ' + values["name"])
else:
return v
@validator('context')
def tags_context(cls, v, values):
context_list = [
"Other", "Unknown", "Source:Endpoint",
"Source:AD", "Source:Firewall", "Source:Application Log",
"Source:IPS", "Source:Cloud Data", "Source:Correlation",
"Source:Printer", "Source:Badge", "Scope:Internal",
"Scope:External", "Scope:Inbound", "Scope:Outbound",
"Scope:Local", "Scope:Network", "Outcome:Blocked",
"Outcome:Allowed", "Stage:Recon", "Stage:Initial Access",
"Stage:Execution", "Stage:Persistence", "Stage:Privilege Escalation",
"Stage:Defense Evasion", "Stage:Credential Access", "Stage:Discovery",
"Stage:Lateral Movement", "Stage:Collection", "Stage:Exfiltration",
"Stage:Command And Control", "Consequence:Infection", "Consequence:Reduced Visibility",
"Consequence:Data Destruction", "Consequence:Denial Of Service", "Consequence:Loss Of Control",
"Rares:Rare User", "Rares:Rare Process", "Rares:Rare Device",
"Rares:Rare Domain", "Rares:Rare Network", "Rares:Rare Location",
"Other:Peer Group", "Other:Brute Force", "Other:Policy Violation",
"Other:Threat Intelligence", "Other:Flight Risk", "Other:Removable Storage"
]
for value in v:
if value not in context_list:
raise ValueError('context value not valid for ' + values["name"] + '. valid options are ' + str(context_list) )
return v
@validator('impact')
def tags_impact(cls, v, values):
if not (v > 0 and v <= 100):
raise ValueError('impact score is out of range 1-100: ' + values["name"])
else:
return v
@validator('kill_chain_phases')
def tags_kill_chain_phases(cls, v, values):
valid_kill_chain_phases = [
'Reconnaissance', 'Weaponization', 'Delivery',
'Exploitation', 'Installation', 'Command and Control',
'Actions on Objectives']
for value in v:
if value not in valid_kill_chain_phases:
raise ValueError('kill chain phase not valid for ' + values["name"] + '. valid options are ' + str(valid_kill_chain_phases))
return v
@validator('mitre_attack_id')
def tags_mitre_attack_id(cls, v, values):
pattern = 'T[0-9]{4}'
for value in v:
if not re.match(pattern, value):
raise ValueError('Mitre Attack ID are not following the pattern Txxxx: ' + values["name"])
return v
@validator('observable')
def tags_observable(cls,v,values):
valid_roles = [
"Other", "Unknown", "Actor",
"Target", "Attacker", "Victim",
"Parent Process", "Child Process", "Known Bad",
"Data Loss", "Observer"
]
valid_types = [
"Other", "Unknown", "Device",
"Container", "Endpoint", "Hostname",
"IP Address", "User", "Username",
"Email", "Email Address", "URL",
"URL Domain", "File", "File Name",
"File Hash", "Process", "Process Name",
"Location"
]
for value in v:
if value['type'] in valid_types:
for role in value['role']:
if role not in valid_roles:
raise ValueError('Observable role ' + role + ' not valid for ' + values["name"] + '. valid options are ' + str(valid_roles))
else:
raise ValueError('Observable type ' + value['type'] + ' not valid for ' + values["name"] + '. valid options are ' + str(valid_types))
return v
@validator('product')
def tags_product(cls, v, values):
valid_products = [
"Splunk Enterprise", "Splunk Enterprise Security", "Splunk Cloud",
"Splunk Security Analytics for AWS", "Splunk Behavioral Analytics"
]
for value in v:
if value not in valid_products:
raise ValueError('product is not valid for ' + values['name'] + '. valid products are ' + str(valid_products))
return v
@validator('risk_score')
def tags_calculate_risk_score(cls, v, values):
calculated_risk_score = (int(values['impact']))*(int(values['confidence']))/100
if calculated_risk_score != int(v):
raise ValueError('risk_score is calculated wrong: ' + values["name"])
return v
@@ -1,42 +0,0 @@
import enum
class AnalyticsType(enum.Enum):
TTP = 1
anomaly = 2
hunting = 3
correlation = 4
class DataModel(enum.Enum):
Endpoint = 1
Network_Traffic = 2
Authentication = 3
Change = 4
Change_Analysis = 5
Email = 6
Network_Resolution = 7
Network_Sessions = 8
UEBA = 9
Updates = 10
Vulnerabilities = 11
Web = 12
Endpoint_Processes = 13
Endpoint_Filesystem = 14
Endpoint_Registry = 15
Risk = 16
class SecurityContentType(enum.Enum):
detections = 1
baselines = 2
stories = 3
playbooks = 4
macros = 5
lookups = 6
deployments = 7
investigations = 8
unit_tests = 9
class SecurityContentProduct(enum.Enum):
ESCU = 1
SSA = 2
API = 3
@@ -1,97 +0,0 @@
import enum
import uuid
import string
import re
import requests
from pydantic import BaseModel, validator, ValidationError
from dataclasses import dataclass
from datetime import datetime
from contentctl_core.domain.entities.security_content_object import SecurityContentObject
from contentctl_core.domain.entities.enums.enums import AnalyticsType
from contentctl_core.domain.entities.enums.enums import DataModel
from contentctl_core.domain.entities.investigation_tags import InvestigationTags
class Investigation(BaseModel, SecurityContentObject):
# investigation spec
name: str
id: str
version: int
date: str
author: str
type: str
datamodel: list
description: str
search: str
how_to_implement: str
known_false_positives: str
references: list
inputs: list = None
tags: InvestigationTags
# enrichment
lowercase_name: str = None
@validator('name')
def name_max_length(cls, v):
if len(v) > 75:
raise ValueError('name is longer then 75 chars: ' + v)
return v
@validator('name')
def name_invalid_chars(cls, v):
invalidChars = set(string.punctuation.replace("-", ""))
if any(char in invalidChars for char in v):
raise ValueError('invalid chars used in name: ' + v)
return v
@validator('id')
def id_check(cls, v, values):
try:
uuid.UUID(str(v))
except:
raise ValueError('uuid is not valid: ' + values["name"])
return v
@validator('date')
def date_valid(cls, v, values):
try:
datetime.strptime(v, "%Y-%m-%d")
except:
raise ValueError('date is not in format YYYY-MM-DD: ' + values["name"])
return v
@validator('datamodel')
def datamodel_valid(cls, v, values):
for datamodel in v:
if datamodel not in [el.name for el in DataModel]:
raise ValueError('not valid data model: ' + values["name"])
return v
@validator('description', 'how_to_implement')
def encode_error(cls, v, values, field):
try:
v.encode('ascii')
except UnicodeEncodeError:
raise ValueError('encoding error in ' + field.name + ': ' + values["name"])
return v
@validator('references')
def references_check(cls, v, values):
for reference in v:
try:
get = requests.get(reference)
if not get.status_code == 200:
raise ValueError('Reference ' + reference + ' is not reachable: ' + values["name"])
except requests.exceptions.RequestException as e:
raise ValueError('Reference ' + reference + ' is not reachable: ' + values["name"])
return v
@validator('search')
def search_validate(cls, v, values):
# write search validator
return v
@@ -1,9 +0,0 @@
from pydantic import BaseModel, validator, ValidationError
class InvestigationTags(BaseModel):
analytic_story: list
product: list
required_fields: list
security_domain: str
@@ -1,16 +0,0 @@
from pydantic import BaseModel, validator, ValidationError
from contentctl_core.domain.entities.security_content_object import SecurityContentObject
class Lookup(BaseModel, SecurityContentObject):
name: str
description: str
collection: str = None
fields_list: str = None
filename: str = None
default_match: str = None
match_type: str = None
min_matches: int = None
case_sensitive_match: str = None
@@ -1,13 +0,0 @@
from pydantic import BaseModel, validator, ValidationError
from contentctl_core.domain.entities.security_content_object import SecurityContentObject
class Macro(BaseModel, SecurityContentObject):
name: str
definition: str
description: str
arguments: list = None
@@ -1,8 +0,0 @@
from pydantic import BaseModel, validator, ValidationError
class MitreAttackEnrichment(BaseModel):
mitre_attack_id: str
mitre_attack_technique: str
mitre_attack_tactics: list
mitre_attack_groups: list
@@ -1,25 +0,0 @@
import uuid
import string
from pydantic import BaseModel, validator, ValidationError
from contentctl_core.domain.entities.security_content_object import SecurityContentObject
from contentctl_core.domain.entities.playbook_tags import PlaybookTag
class Playbook(BaseModel, SecurityContentObject):
name: str
id: str
version: int
date: str
author: str
type: str
description: str
how_to_implement: str
playbook: str
references: list
app_list: list
tags: PlaybookTag
@@ -1,13 +0,0 @@
from pydantic import BaseModel, validator, ValidationError
class PlaybookTag(BaseModel):
analytic_story: list = None
detections: list = None
platform_tags: list = None
playbook_fields: list = None
product: list = None
playbook_fields: list = None
detection_objects: list = None
@@ -1,7 +0,0 @@
import abc
from contentctl_core.domain.entities.enums.enums import SecurityContentType
class SecurityContentObject(abc.ABC):
type: SecurityContentType
@@ -1,74 +0,0 @@
import string
import uuid
import requests
from pydantic import BaseModel, validator, ValidationError
from datetime import datetime
from contentctl_core.domain.entities.security_content_object import SecurityContentObject
from contentctl_core.domain.entities.story_tags import StoryTags
class Story(BaseModel, SecurityContentObject):
# story spec
name: str
id: str
version: int
date: str
author: str
description: str
narrative: str
references: list
tags: StoryTags
# enrichments
detection_names: list = None
investigation_names: list = None
baseline_names: list = None
author_company: str = None
author_name: str = None
detections: list = None
investigations: list = None
@validator('name')
def name_invalid_chars(cls, v):
invalidChars = set(string.punctuation.replace("-", ""))
if any(char in invalidChars for char in v):
raise ValueError('invalid chars used in name: ' + v)
return v
@validator('id')
def id_check(cls, v, values):
try:
uuid.UUID(str(v))
except:
raise ValueError('uuid is not valid: ' + values["name"])
return v
@validator('date')
def date_valid(cls, v, values):
try:
datetime.strptime(v, "%Y-%m-%d")
except:
raise ValueError('date is not in format YYYY-MM-DD: ' + values["name"])
return v
@validator('description', 'narrative')
def encode_error(cls, v, values, field):
try:
v.encode('ascii')
except UnicodeEncodeError:
raise ValueError('encoding error in ' + field.name + ': ' + values["name"])
return v
# @validator('references')
# def references_check(cls, v, values):
# for reference in v:
# try:
# get = requests.get(reference)
# if not get.status_code == 200:
# raise ValueError('Reference ' + reference + ' is not reachable: ' + values["name"])
# except requests.exceptions.RequestException as e:
# raise ValueError('Reference ' + reference + ' is not reachable: ' + values["name"])
# return v
@@ -1,32 +0,0 @@
from pydantic import BaseModel, validator, ValidationError
from contentctl_core.domain.entities.mitre_attack_enrichment import MitreAttackEnrichment
class StoryTags(BaseModel):
# story spec
name: str
analytic_story: str
category: list
product: list
usecase: str
# enrichment
mitre_attack_enrichments: list[MitreAttackEnrichment] = None
mitre_attack_tactics: list = None
datamodels: list = None
kill_chain_phases: list = None
@validator('product')
def tags_product(cls, v, values):
valid_products = [
"Splunk Enterprise", "Splunk Enterprise Security", "Splunk Cloud",
"Splunk Security Analytics for AWS", "Splunk Behavioral Analytics"
]
for value in v:
if value not in valid_products:
raise ValueError('product is not valid for ' + values['name'] + '. valid products are ' + str(valid_products))
return v
@@ -1,11 +0,0 @@
from pydantic import BaseModel, validator, ValidationError
from contentctl_core.domain.entities.security_content_object import SecurityContentObject
from contentctl_core.domain.entities.unit_test_test import UnitTestTest
class UnitTest(BaseModel, SecurityContentObject):
name: str
tests: list[UnitTestTest]
@@ -1,11 +0,0 @@
from pydantic import BaseModel, validator, ValidationError
class UnitTestAttackData(BaseModel):
file_name: str
data: str
source: str
sourcetype: str = None
update_timestamp: bool = None
@@ -1,11 +0,0 @@
from pydantic import BaseModel, validator, ValidationError
class UnitTestBaseline(BaseModel):
name: str
file: str
pass_condition: str
earliest_time: str
latest_time: str
@@ -1,15 +0,0 @@
from pydantic import BaseModel, validator, ValidationError
from contentctl_core.domain.entities.unit_test_attack_data import UnitTestAttackData
from contentctl_core.domain.entities.unit_test_baseline import UnitTestBaseline
class UnitTestTest(BaseModel):
name: str
file: str
pass_condition: str
earliest_time: str = None
latest_time: str = None
baselines: list[UnitTestBaseline] = None
attack_data: list[UnitTestAttackData]
@@ -1,28 +0,0 @@
import os
import pytest
from contentctl_core.application.factory.ba_factory import BAFactory, BAFactoryInputDto, BAFactoryOutputDto
from contentctl_infrastructure.builder.security_content_director import SecurityContentDirector
from contentctl_infrastructure.builder.security_content_basic_builder import SecurityContentBasicBuilder
from contentctl_infrastructure.builder.security_content_detection_builder import SecurityContentDetectionBuilder
def test_factory_BA():
input_path = os.path.join(os.path.dirname(__file__), '../../../../../..')
input_dto = BAFactoryInputDto(
input_path,
SecurityContentBasicBuilder(),
SecurityContentDetectionBuilder(),
SecurityContentDirector()
)
output_dto = BAFactoryOutputDto([],[])
factory = BAFactory(output_dto)
factory.execute(input_dto)
for detection in output_dto.detections:
if not detection.test:
raise AssertionError("test file missing for ssa detection: " + detection.name)
@@ -1,37 +0,0 @@
import os
from re import A
from contentctl_core.application.factory.factory import FactoryInputDto
from contentctl_core.application.factory.factory import FactoryOutputDto
from contentctl_core.application.factory.factory import Factory
from contentctl_infrastructure.builder.security_content_director import SecurityContentDirector
from contentctl_infrastructure.builder.security_content_basic_builder import SecurityContentBasicBuilder
from contentctl_infrastructure.builder.security_content_detection_builder import SecurityContentDetectionBuilder
from contentctl_infrastructure.builder.security_content_story_builder import SecurityContentStoryBuilder
from contentctl_core.domain.entities.enums.enums import SecurityContentProduct
from contentctl_infrastructure.builder.security_content_investigation_builder import SecurityContentInvestigationBuilder
from contentctl_infrastructure.builder.security_content_baseline_builder import SecurityContentBaselineBuilder
from contentctl_infrastructure.builder.attack_enrichment import AttackEnrichment
from contentctl_infrastructure.builder.security_content_playbook_builder import SecurityContentPlaybookBuilder
def test_factory_ESCU():
input_path = os.path.join(os.path.dirname(__file__), '../../../../../..')
input_dto = FactoryInputDto(
input_path,
SecurityContentBasicBuilder(),
SecurityContentDetectionBuilder(),
SecurityContentStoryBuilder(),
SecurityContentBaselineBuilder(),
SecurityContentInvestigationBuilder(),
SecurityContentPlaybookBuilder(),
SecurityContentDirector(),
AttackEnrichment.get_attack_lookup()
)
output_dto = FactoryOutputDto([],[],[],[],[],[],[],[],[])
factory = Factory(output_dto)
factory.execute(input_dto)
@@ -1,25 +0,0 @@
import os
from contentctl_core.domain.entities.enums.enums import SecurityContentType
from contentctl_core.application.factory.object_factory import ObjectFactoryInputDto
from contentctl_core.application.factory.object_factory import ObjectFactory
from contentctl_infrastructure.builder.security_content_object_builder import SecurityContentObjectBuilder
from contentctl_infrastructure.builder.security_content_director import SecurityContentDirector
def test_object_factory():
input_path = os.path.join(os.path.dirname(__file__), '../../../../../../detections')
input_dto = ObjectFactoryInputDto(
input_path,
SecurityContentObjectBuilder(),
SecurityContentDirector()
)
objects = list()
factory = ObjectFactory(objects)
factory.execute(input_dto)
#assert len(objects) == 959
@@ -1,88 +0,0 @@
name: Attempted Credential Dump From Registry via Reg exe
id: e9fb4a59-c5fb-440a-9f24-191fbc6b2911
version: 6
date: '2021-09-16'
author: PATRICK BAREISS, SPLUNK
type: TTP
datamodel:
- Endpoint
description: Monitor for execution of reg.exe with parameters specifying an export
of keys that contain hashed credentials that attackers may try to crack offline.
search: '| tstats `security_content_summariesonly` count min(_time) as firstTime max(_time)
as lastTime from datamodel=Endpoint.Processes where `process_reg` OR `process_cmd`
Processes.process=*save* (Processes.process=*HKEY_LOCAL_MACHINE\\Security* OR Processes.process=*HKEY_LOCAL_MACHINE\\SAM*
OR Processes.process=*HKEY_LOCAL_MACHINE\\System* OR Processes.process=*HKLM\\Security*
OR Processes.process=*HKLM\\System* OR Processes.process=*HKLM\\SAM*) by Processes.dest
Processes.user Processes.parent_process Processes.process_name Processes.original_file_name
Processes.process Processes.process_id Processes.parent_process_id | `drop_dm_object_name(Processes)`
| `security_content_ctime(firstTime)`| `security_content_ctime(lastTime)` | `attempted_credential_dump_from_registry_via_reg_exe_filter`'
how_to_implement: To successfully implement this search you need to be ingesting information
on process that include the name of the process responsible for the changes from
your endpoints into the `Endpoint` datamodel in the `Processes` node. In addition,
confirm the latest CIM App 4.20 or higher is installed and the latest TA for the
endpoint product.
known_false_positives: None identified.
references:
- https://github.com/redcanaryco/atomic-red-team/blob/master/atomics/T1003.002/T1003.002.md#atomic-test-1---registry-dump-of-sam-creds-and-secrets
tags:
analytic_story:
- Credential Dumping
- DarkSide Ransomware
asset_type: Endpoint
automated_detection_testing: passed
cis20:
- CIS 3
- CIS 5
- CIS 16
confidence: 100
context:
- Source:Endpoint
- Stage:Credential Access
dataset:
- https://media.githubusercontent.com/media/splunk/attack_data/master/datasets/attack_techniques/T1003.002/atomic_red_team/windows-sysmon.log
impact: 90
kill_chain_phases:
- Actions on Objectives
message: An instance of $parent_process_name$ spawning $process_name$ was identified
on endpoint $dest$ by user $user$ attempting to export the registry keys.
mitre_attack_id:
- T1003.002
- T1003
nist:
- DE.CM
observable:
- name: user
type: User
role:
- Victim
- name: dest
type: Hostname
role:
- Victim
- name: parent_process_name
type: Process
role:
- Parent Process
- name: process_name
type: Process
role:
- Child Process
product:
- Splunk Enterprise
- Splunk Enterprise Security
- Splunk Cloud
required_fields:
- _time
- Processes.dest
- Processes.user
- Processes.parent_process_name
- Processes.parent_process
- Processes.original_file_name
- Processes.process_name
- Processes.process
- Processes.process_id
- Processes.parent_process_path
- Processes.process_path
- Processes.parent_process_id
risk_score: 90
security_domain: endpoint
@@ -1,88 +0,0 @@
name: Attempted Credential Dump From Registry via Reg exe
id: e9fb4a59-c5fb-440a-9f24-191fbc6b2911
version: 6
date: '2021-09-16'
author: PATRICK BAREISS, SPLUNK
type: TTP
datamodel:
- Endpoint
description: Monitor for execution of reg.exe with parameters specifying an export
of keys that contain hashed credentials that attackers may try to crack offline.
search: '| tstats `security_content_summariesonly` count min(_time) as firstTime max(_time)
as lastTime from datamodel=Endpoint.Processes where `process_reg` OR `process_cmd`
Processes.process=*save* (Processes.process=*HKEY_LOCAL_MACHINE\\Security* OR Processes.process=*HKEY_LOCAL_MACHINE\\SAM*
OR Processes.process=*HKEY_LOCAL_MACHINE\\System* OR Processes.process=*HKLM\\Security*
OR Processes.process=*HKLM\\System* OR Processes.process=*HKLM\\SAM*) by Processes.dest
Processes.user Processes.parent_process Processes.process_name Processes.original_file_name
Processes.process Processes.process_id Processes.parent_process_id | `drop_dm_object_name(Processes)`
| `security_content_ctime(firstTime)`| `security_content_ctime(lastTime)` | `attempted_credential_dump_from_registry_via_reg_exe_filter`'
how_to_implement: To successfully implement this search you need to be ingesting information
on process that include the name of the process responsible for the changes from
your endpoints into the `Endpoint` datamodel in the `Processes` node. In addition,
confirm the latest CIM App 4.20 or higher is installed and the latest TA for the
endpoint product.
known_false_positives: None identified.
references:
- https://github.com/redcanaryco/atomic-red-team/blob/master/atomics/T1003.002/T1003.002.md#atomic-test-1---registry-dump-of-sam-creds-and-secrets
tags:
analytic_story:
- Credential Dumping
- DarkSide Ransomware
asset_type: Endpoint
automated_detection_testing: passed
cis20:
- CIS 3
- CIS 5
- CIS 16
confidence: 100
context:
- Source:Endpoint
- Stage:Credential Access
dataset:
- https://media.githubusercontent.com/media/splunk/attack_data/master/datasets/attack_techniques/T1003.002/atomic_red_team/windows-sysmon.log
impact: 90
kill_chain_phases:
- Actions on Objectives
message: An instance of $parent_process_name$ spawning $process_name$ was identified
on endpoint $dest$ by user $user$ attempting to export the registry keys.
mitre_attack_id:
- T1003.002
- T1003
nist:
- DE.CM
observable:
- name: user
type: User
role:
- Victim
- name: dest
type: Hostname
role:
- Victim
- name: parent_process_name
type: Process
role:
- Parent Process
- name: process_name
type: Process
role:
- Child Process
product:
- Splunk Enterprise
- Splunk Enterprise Security
- Splunk Cloud
required_fields:
- _time
- Processes.dest
- Processes.user
- Processes.parent_process_name
- Processes.parent_process
- Processes.original_file_name
- Processes.process_name
- Processes.process
- Processes.process_id
- Processes.parent_process_path
- Processes.process_path
- Processes.parent_process_id
risk_score: 90
security_domain: endpoint
@@ -1,34 +0,0 @@
import os
from contentctl_core.application.use_cases.content_changer import ContentChanger, ContentChangerInputDto
from contentctl_core.application.factory.object_factory import ObjectFactoryInputDto
from contentctl_infrastructure.builder.security_content_object_builder import SecurityContentObjectBuilder
from contentctl_infrastructure.builder.security_content_director import SecurityContentDirector
from contentctl_infrastructure.adapter.obj_to_yml_adapter import ObjToYmlAdapter
from contentctl_infrastructure.builder.yml_reader import YmlReader
def test_content_changer_author_uppercase():
input_path = os.path.join(os.path.dirname(__file__),
'data_content_changer')
output_path = os.path.join(os.path.dirname(__file__),
'data_content_changer_ref')
factory_input_dto = ObjectFactoryInputDto(
input_path,
SecurityContentObjectBuilder(),
SecurityContentDirector()
)
input_dto = ContentChangerInputDto(
ObjToYmlAdapter(),
factory_input_dto,
'example_converter_func'
)
content_changer = ContentChanger()
content_changer.execute(input_dto)
yml_obj = YmlReader.load_file(os.path.join(output_path, 'attempted_credential_dump_from_registry_via_reg_exe.yml'))
assert yml_obj['author'] == 'PATRICK BAREISS, SPLUNK'
@@ -1,68 +0,0 @@
name: Unsigned Image Loaded by LSASS
id: 56ef054c-76ef-45f9-af4a-a634695dcd65
version: 1
date: '2019-12-06'
author: Patrick Bareiss, Splunk
type: TTP
datamodel: []
description: This search detects loading of unsigned images by LSASS. Deprecated because
too noisy.
search: '`sysmon` EventID=7 Image=*lsass.exe Signed=false | stats count min(_time)
as firstTime max(_time) as lastTime by Computer, Image, ImageLoaded, Signed, SHA1
| rename Computer as dest | `security_content_ctime(firstTime)`| `security_content_ctime(lastTime)`
| `unsigned_image_loaded_by_lsass_filter` '
how_to_implement: This search needs Sysmon Logs with a sysmon configuration, which
includes EventCode 7 with lsass.exe. This search uses an input macro named `sysmon`.
We strongly recommend that you specify your environment-specific configurations
(index, source, sourcetype, etc.) for Windows Sysmon logs. Replace the macro definition
with configurations for your Splunk environment. The search also uses a post-filter
macro designed to filter out known false positives.
known_false_positives: Other tools could load images into LSASS for legitimate reason.
But enterprise tools should always use signed DLLs.
references:
- https://attack.mitre.org/techniques/T1003/001/
tags:
analytic_story:
- Credential Dumping
asset_type: Windows
cis20:
- CIS 8
- CIS 16
confidence: 100
context:
- Source:Endpoint
- Stage:Credential Access
impact: 90
kill_chain_phases:
- Actions on Objectives
message: An instance of $parent_process_name$ spawning $process_name$ was identified
on endpoint $dest$ by user $user$ attempting to export the registry keys.
mitre_attack_id:
- T1003.001
nist:
- DE.CM
observable:
- name: user
type: User
role:
- Victim
- name: dest
type: Hostname
role:
- Victim
- name: parent_process_name
type: Process
role:
- Parent Process
- name: process_name
type: Process
role:
- Child Process
product:
- Splunk Enterprise
- Splunk Enterprise Security
- Splunk Cloud
required_fields:
- _time
risk_score: 90
security_domain: endpoint
@@ -1,88 +0,0 @@
name: Attempted Credential Dump From Registry via Reg exe
id: e9fb4a59-c5fb-440a-9f24-191fbc6b2911
version: 6
date: '2021-09-16'
author: Patrick Bareiss, Splunk
type: TTP
datamodel:
- Endpoint
description: Monitor for execution of reg.exe with parameters specifying an export
of keys that contain hashed credentials that attackers may try to crack offline.
search: '| tstats `security_content_summariesonly` count min(_time) as firstTime max(_time)
as lastTime from datamodel=Endpoint.Processes where `process_reg` OR `process_cmd`
Processes.process=*save* (Processes.process=*HKEY_LOCAL_MACHINE\\Security* OR Processes.process=*HKEY_LOCAL_MACHINE\\SAM*
OR Processes.process=*HKEY_LOCAL_MACHINE\\System* OR Processes.process=*HKLM\\Security*
OR Processes.process=*HKLM\\System* OR Processes.process=*HKLM\\SAM*) by Processes.dest
Processes.user Processes.parent_process Processes.process_name Processes.original_file_name
Processes.process Processes.process_id Processes.parent_process_id | `drop_dm_object_name(Processes)`
| `security_content_ctime(firstTime)`| `security_content_ctime(lastTime)` | `attempted_credential_dump_from_registry_via_reg_exe_filter`'
how_to_implement: To successfully implement this search you need to be ingesting information
on process that include the name of the process responsible for the changes from
your endpoints into the `Endpoint` datamodel in the `Processes` node. In addition,
confirm the latest CIM App 4.20 or higher is installed and the latest TA for the
endpoint product.
known_false_positives: None identified.
references:
- https://github.com/redcanaryco/atomic-red-team/blob/master/atomics/T1003.002/T1003.002.md#atomic-test-1---registry-dump-of-sam-creds-and-secrets
tags:
analytic_story:
- Credential Dumping
- DarkSide Ransomware
asset_type: Endpoint
automated_detection_testing: passed
cis20:
- CIS 3
- CIS 5
- CIS 16
confidence: 100
context:
- Source:Endpoint
- Stage:Credential Access
dataset:
- https://media.githubusercontent.com/media/splunk/attack_data/master/datasets/attack_techniques/T1003.002/atomic_red_team/windows-sysmon.log
impact: 90
kill_chain_phases:
- Actions on Objectives
message: An instance of $parent_process_name$ spawning $process_name$ was identified
on endpoint $dest$ by user $user$ attempting to export the registry keys.
mitre_attack_id:
- T1003.002
- T1003
nist:
- DE.CM
observable:
- name: user
type: User
role:
- Victim
- name: dest
type: Hostname
role:
- Victim
- name: parent_process_name
type: Process
role:
- Parent Process
- name: process_name
type: Process
role:
- Child Process
product:
- Splunk Enterprise
- Splunk Enterprise Security
- Splunk Cloud
required_fields:
- _time
- Processes.dest
- Processes.user
- Processes.parent_process_name
- Processes.parent_process
- Processes.original_file_name
- Processes.process_name
- Processes.process
- Processes.process_id
- Processes.parent_process_path
- Processes.process_path
- Processes.parent_process_id
risk_score: 90
security_domain: endpoint
@@ -1,89 +0,0 @@
name: Cobalt Strike Named Pipes
id: 5876d429-0240-4709-8b93-ea8330b411b5
version: 1
date: '2021-02-22'
author: Michael Haag, Splunk
type: TTP
datamodel: []
description: 'The following analytic identifies the use of default or publicly known
named pipes used with Cobalt Strike. A named pipe is a named, one-way or duplex
pipe for communication between the pipe server and one or more pipe clients. Cobalt
Strike uses named pipes in many ways and has default values used with the Artifact
Kit and Malleable C2 Profiles. The following query assists with identifying these
default named pipes. Each EDR product presents named pipes a little different. Consider
taking the values and generating a query based on the product of choice. \
Upon triage, review the process performing the named pipe. If it is explorer.exe,
It is possible it was injected into by another process. Review recent parallel processes
to identify suspicious patterns or behaviors. A parallel process may have a network
connection, review and follow the connection back to identify any file modifications.'
search: '`sysmon` EventID=17 OR EventID=18 PipeName IN (\\msagent_*, \\wkssvc*, \\DserNamePipe*,
\\srvsvc_*, \\mojo.*, \\postex_*, \\status_*, \\MSSE-*, \\spoolss_*, \\win_svc*,
\\ntsvcs*, \\winsock*, \\UIA_PIPE*) | stats count min(_time) as firstTime max(_time)
as lastTime by Computer, process_name, process_id process_path, PipeName | rename
Computer as dest | `security_content_ctime(firstTime)` | `security_content_ctime(lastTime)`
| `cobalt_strike_named_pipes_filter`'
how_to_implement: To successfully implement this search, you need to be ingesting
logs with the process name, parent process, and command-line executions from your
endpoints. If you are using Sysmon, you must have at least version 6.0.4 of the
Sysmon TA.
known_false_positives: The idea of using named pipes with Cobalt Strike is to blend
in. Therefore, some of the named pipes identified and added may cause false positives.
Filter by process name or pipe name to reduce false positives.
references:
- https://attack.mitre.org/techniques/T1218/009/
- https://docs.microsoft.com/en-us/windows/win32/ipc/named-pipes
- https://gist.github.com/MHaggis/6c600e524045a6d49c35291a21e10752
- https://www.fireeye.com/blog/threat-research/2021/05/shining-a-light-on-darkside-ransomware-operations.html
tags:
analytic_story:
- Cobalt Strike
- Trickbot
- DarkSide Ransomware
asset_type: Endpoint
automated_detection_testing: passed
cis20:
- CIS 8
confidence: 90
context:
- Source:Endpoint
- Stage:Defense Evasion
dataset:
- https://media.githubusercontent.com/media/splunk/attack_data/master/datasets/attack_techniques/T1055/cobalt_strike/windows-sysmon.log
impact: 80
kill_chain_phases:
- Actions on Objectives
message: An instance of $process_name$ was identified on endpoint $Computer$ by
user $user$ accessing known suspicious named pipes related to Cobalt Strike.
mitre_attack_id:
- T1055
nist:
- PR.PT
- DE.CM
observable:
- name: user
type: User
role:
- Victim
- name: Computer
type: Hostname
role:
- Victim
- name: process_name
type: Process
role:
- Parent Process
product:
- Splunk Enterprise
- Splunk Enterprise Security
- Splunk Cloud
required_fields:
- _time
- EventID
- PipeName
- Computer
- process_name
- process_path
- process_id
risk_score: 72
security_domain: endpoint
@@ -1,5 +0,0 @@
arguments:
- field
definition: 'convert timeformat="%Y-%m-%dT%H:%M:%S" ctime($field$)'
description: convert epoch time to string
name: security_content_ctime
@@ -1,3 +0,0 @@
definition: summariesonly=false allow_old_summaries=true
description: search data model's summaries only
name: security_content_summariesonly
@@ -1,4 +0,0 @@
definition: sourcetype=XmlWinEventLog:Microsoft-Windows-Sysmon/Operational OR source=XmlWinEventLog:Microsoft-Windows-Sysmon/Operational
description: customer specific splunk configurations(eg- index, source, sourcetype).
Replace the macro definition with configurations for your Splunk Environmnent.
name: sysmon
@@ -1,48 +0,0 @@
name: Cobalt Strike
id: bcfd17e8-5461-400a-80a2-3b7d1459220c
version: 1
date: '2021-02-16'
author: Michael Haag, Splunk
description: Cobalt Strike is threat emulation software. Red teams and penetration testers use Cobalt Strike to demonstrate the risk of a breach and evaluate mature security programs. Most recently, Cobalt Strike has become the choice tool by threat groups due to its ease of use and extensibility.
narrative: 'This Analytic Story supports you to detect Tactics, Techniques and Procedures
(TTPs) from Cobalt Strike. Cobalt Strike has many ways to be enhanced by using aggressor scripts, malleable C2 profiles, default attack packages, and much more.
For endpoint behavior, Cobalt Strike is most commonly identified via named pipes, spawn to processes, and DLL function names. Many additional variables are provided for in memory operation of the beacon implant.
On the network, depending on the malleable C2 profile used, it is near infinite in the amount of ways to conceal the C2 traffic with Cobalt Strike.
Not every query may be specific to Cobalt Strike the tool, but the methodologies and techniques used by it.\
Splunk Threat Research reviewed all publicly available instances of Malleabe C2 Profiles and generated a list of the most commonly used spawnto and pipenames.\
`Spawnto_x86` and `spawnto_x64` is the process that Cobalt Strike will spawn and injects shellcode into.\
Pipename sets the named pipe name used in Cobalt Strikes Beacon SMB C2 traffic.\
With that, new detections were generated focused on these spawnto processes spawning without command line arguments. Similar, the named pipes most commonly used by Cobalt Strike added as a detection.
In generating content for Cobalt Strike, the following is considered:\
- Is it normal for spawnto_ value to have no command line arguments? No command line arguments and a network connection?\
- What is the default, or normal, process lineage for spawnto_ value?\
- Does the spawnto_ value make network connections?\
- Is it normal for spawnto_ value to load jscript, vbscript, Amsi.dll, and clr.dll?\
While investigating a detection related to this Analytic Story, keep in mind the parent process, process path, and any file modifications that may occur. Tuning may need to occur to remove any false positives.'
references:
- https://www.cobaltstrike.com/
- https://www.infocyte.com/blog/2020/09/02/cobalt-strike-the-new-favorite-among-thieves/
- https://bluescreenofjeff.com/2017-01-24-how-to-write-malleable-c2-profiles-for-cobalt-strike/
- https://blog.talosintelligence.com/2020/09/coverage-strikes-back-cobalt-strike-paper.html
- https://www.fireeye.com/blog/threat-research/2020/12/unauthorized-access-of-fireeye-red-team-tools.html
- https://github.com/MichaelKoczwara/Awesome-CobaltStrike-Defence
- https://github.com/zer0yu/Awesome-CobaltStrike
tags:
analytic_story: Cobalt Strike
category:
- Adversary Tactics
product:
- Splunk Enterprise
- Splunk Enterprise Security
- Splunk Cloud
usecase: Advanced Threat Detection
@@ -1,37 +0,0 @@
name: Credential Dumping
id: 854d78bf-d0e2-4f4e-b05c-640905f86d7a
version: 3
date: '2020-02-04'
author: Rico Valdez, Splunk
description: Uncover activity consistent with credential dumping, a technique wherein
attackers compromise systems and attempt to obtain and exfiltrate passwords. The
threat actors use these pilfered credentials to further escalate privileges and
spread throughout a target environment. The included searches in this Analytic Story
are designed to identify attempts to credential dumping.
narrative: 'Credential dumping&#151;gathering credentials from a target system, often
hashed or encrypted&#151;is a common attack technique. Even though the credentials
may not be in plain text, an attacker can still exfiltrate the data and set to cracking
it offline, on their own systems. The threat actors target a variety of sources
to extract them, including the Security Accounts Manager (SAM), Local Security Authority
(LSA), NTDS from Domain Controllers, or the Group Policy Preference (GPP) files.\
Once attackers obtain valid credentials, they use them to move throughout a target
network with ease, discovering new systems and identifying assets of interest. Credentials
obtained in this manner typically include those of privileged users, which may provide
access to more sensitive information and system operations.\
The detection searches in this Analytic Story monitor access to the Local Security
Authority Subsystem Service (LSASS) process, the usage of shadowcopies for credential
dumping and some other techniques for credential dumping.'
references:
- https://attack.mitre.org/wiki/Technique/T1003
- https://cyberwardog.blogspot.com/2017/03/chronicles-of-threat-hunter-hunting-for.html
tags:
analytic_story: Credential Dumping
category:
- Adversary Tactics
product:
- Splunk Enterprise
- Splunk Enterprise Security
- Splunk Cloud
usecase: Advanced Threat Detection
@@ -1,20 +0,0 @@
name: DarkSide Ransomware
id: 507edc74-13d5-4339-878e-b9114ded1f35
version: 1
date: '2021-05-12'
author: Bhavin Patel, Splunk
description: Leverage searches that allow you to detect and investigate unusual activities
that might relate to the DarkSide Ransomware
narrative: 'This story addresses Darkside ransomware. This ransomware payload has many similarities to common ransomware however there are certain items particular to it. The creation of a .TXT log that shows every item being encrypted as well as the creation of ransomware notes and files adding a machine ID created based on CRC32 checksum algorithm. This ransomware payload leaves machines in minimal operation level,enough to browse the attackers websites. A customized URI with leaked information is presented to each victim.This is the ransomware payload that shut down the Colonial pipeline. The story is composed of several detection searches covering similar items to other ransomware payloads and those particular to Darkside payload.'
references:
- https://www.splunk.com/en_us/blog/security/the-darkside-of-the-ransomware-pipeline.htmlbig-game-hunting-with-ryuk-another-lucrative-targeted-ransomware/
- https://www.fireeye.com/blog/threat-research/2021/05/shining-a-light-on-darkside-ransomware-operations.html
tags:
analytic_story: DarkSide Ransomware
category:
- Malware
product:
- Splunk Enterprise
- Splunk Enterprise Security
- Splunk Cloud
usecase: Advanced Threat Detection
@@ -1,23 +0,0 @@
name: Trickbot
id: 16f93769-8342-44c0-9b1d-f131937cce8e
version: 1
date: '2021-04-20'
author: Rod Soto, Teoderick Contreras, Splunk
description: Leverage searches that allow you to detect and investigate unusual activities
that might relate to the trickbot banking trojan, including looking for file writes associated
with its payload, process injection, shellcode execution and data collection even in LDAP environment.
narrative: trickbot banking trojan campaigns targeting banks and other vertical sectors.This malware is known
in Microsoft Windows OS where target security Microsoft Defender to prevent its detection and removal. steal
Verizon credentials and targeting banks using its multi component modules that collect and exfiltrate data.
references:
- https://en.wikipedia.org/wiki/Trickbot
- https://blog.checkpoint.com/2021/03/11/february-2021s-most-wanted-malware-trickbot-takes-over-following-emotet-shutdown/
tags:
analytic_story: Trickbot
category:
- Malware
product:
- Splunk Enterprise
- Splunk Enterprise Security
- Splunk Cloud
usecase: Advanced Threat Detection
@@ -1,88 +0,0 @@
name: Attempted Credential Dump From Registry via Reg exe
id: e9fb4a59-c5fb-440a-9f24-191fbc6b2911
version: 6
date: '2021-09-16'
author: PATRICK BAREISS, SPLUNK
type: TTP
datamodel:
- Endpoint
description: Monitor for execution of reg.exe with parameters specifying an export
of keys that contain hashed credentials that attackers may try to crack offline.
search: '| tstats `security_content_summariesonly` count min(_time) as firstTime max(_time)
as lastTime from datamodel=Endpoint.Processes where `process_reg` OR `process_cmd`
Processes.process=*save* (Processes.process=*HKEY_LOCAL_MACHINE\\Security* OR Processes.process=*HKEY_LOCAL_MACHINE\\SAM*
OR Processes.process=*HKEY_LOCAL_MACHINE\\System* OR Processes.process=*HKLM\\Security*
OR Processes.process=*HKLM\\System* OR Processes.process=*HKLM\\SAM*) by Processes.dest
Processes.user Processes.parent_process Processes.process_name Processes.original_file_name
Processes.process Processes.process_id Processes.parent_process_id | `drop_dm_object_name(Processes)`
| `security_content_ctime(firstTime)`| `security_content_ctime(lastTime)` | `attempted_credential_dump_from_registry_via_reg_exe_filter`'
how_to_implement: To successfully implement this search you need to be ingesting information
on process that include the name of the process responsible for the changes from
your endpoints into the `Endpoint` datamodel in the `Processes` node. In addition,
confirm the latest CIM App 4.20 or higher is installed and the latest TA for the
endpoint product.
known_false_positives: None identified.
references:
- https://github.com/redcanaryco/atomic-red-team/blob/master/atomics/T1003.002/T1003.002.md#atomic-test-1---registry-dump-of-sam-creds-and-secrets
tags:
analytic_story:
- Credential Dumping
- DarkSide Ransomware
asset_type: Endpoint
automated_detection_testing: passed
cis20:
- CIS 3
- CIS 5
- CIS 16
confidence: 100
context:
- Source:Endpoint
- Stage:Credential Access
dataset:
- https://media.githubusercontent.com/media/splunk/attack_data/master/datasets/attack_techniques/T1003.002/atomic_red_team/windows-sysmon.log
impact: 90
kill_chain_phases:
- Actions on Objectives
message: An instance of $parent_process_name$ spawning $process_name$ was identified
on endpoint $dest$ by user $user$ attempting to export the registry keys.
mitre_attack_id:
- T1003.002
- T1003
nist:
- DE.CM
observable:
- name: user
type: User
role:
- Victim
- name: dest
type: Hostname
role:
- Victim
- name: parent_process_name
type: Process
role:
- Parent Process
- name: process_name
type: Process
role:
- Child Process
product:
- Splunk Enterprise
- Splunk Enterprise Security
- Splunk Cloud
required_fields:
- _time
- Processes.dest
- Processes.user
- Processes.parent_process_name
- Processes.parent_process
- Processes.original_file_name
- Processes.process_name
- Processes.process
- Processes.process_id
- Processes.parent_process_path
- Processes.process_path
- Processes.parent_process_id
risk_score: 90
security_domain: endpoint
@@ -1,89 +0,0 @@
name: Cobalt Strike Named Pipes
id: 5876d429-0240-4709-8b93-ea8330b411b5
version: 1
date: '2021-02-22'
author: MICHAEL HAAG, SPLUNK
type: TTP
datamodel: []
description: 'The following analytic identifies the use of default or publicly known
named pipes used with Cobalt Strike. A named pipe is a named, one-way or duplex
pipe for communication between the pipe server and one or more pipe clients. Cobalt
Strike uses named pipes in many ways and has default values used with the Artifact
Kit and Malleable C2 Profiles. The following query assists with identifying these
default named pipes. Each EDR product presents named pipes a little different. Consider
taking the values and generating a query based on the product of choice. \
Upon triage, review the process performing the named pipe. If it is explorer.exe,
It is possible it was injected into by another process. Review recent parallel processes
to identify suspicious patterns or behaviors. A parallel process may have a network
connection, review and follow the connection back to identify any file modifications.'
search: '`sysmon` EventID=17 OR EventID=18 PipeName IN (\\msagent_*, \\wkssvc*, \\DserNamePipe*,
\\srvsvc_*, \\mojo.*, \\postex_*, \\status_*, \\MSSE-*, \\spoolss_*, \\win_svc*,
\\ntsvcs*, \\winsock*, \\UIA_PIPE*) | stats count min(_time) as firstTime max(_time)
as lastTime by Computer, process_name, process_id process_path, PipeName | rename
Computer as dest | `security_content_ctime(firstTime)` | `security_content_ctime(lastTime)`
| `cobalt_strike_named_pipes_filter`'
how_to_implement: To successfully implement this search, you need to be ingesting
logs with the process name, parent process, and command-line executions from your
endpoints. If you are using Sysmon, you must have at least version 6.0.4 of the
Sysmon TA.
known_false_positives: The idea of using named pipes with Cobalt Strike is to blend
in. Therefore, some of the named pipes identified and added may cause false positives.
Filter by process name or pipe name to reduce false positives.
references:
- https://attack.mitre.org/techniques/T1218/009/
- https://docs.microsoft.com/en-us/windows/win32/ipc/named-pipes
- https://gist.github.com/MHaggis/6c600e524045a6d49c35291a21e10752
- https://www.fireeye.com/blog/threat-research/2021/05/shining-a-light-on-darkside-ransomware-operations.html
tags:
analytic_story:
- Cobalt Strike
- Trickbot
- DarkSide Ransomware
asset_type: Endpoint
automated_detection_testing: passed
cis20:
- CIS 8
confidence: 90
context:
- Source:Endpoint
- Stage:Defense Evasion
dataset:
- https://media.githubusercontent.com/media/splunk/attack_data/master/datasets/attack_techniques/T1055/cobalt_strike/windows-sysmon.log
impact: 80
kill_chain_phases:
- Actions on Objectives
message: An instance of $process_name$ was identified on endpoint $Computer$ by
user $user$ accessing known suspicious named pipes related to Cobalt Strike.
mitre_attack_id:
- T1055
nist:
- PR.PT
- DE.CM
observable:
- name: user
type: User
role:
- Victim
- name: Computer
type: Hostname
role:
- Victim
- name: process_name
type: Process
role:
- Parent Process
product:
- Splunk Enterprise
- Splunk Enterprise Security
- Splunk Cloud
required_fields:
- _time
- EventID
- PipeName
- Computer
- process_name
- process_path
- process_id
risk_score: 72
security_domain: endpoint
@@ -1,75 +0,0 @@
import json
VERSION = "4.2"
NAME = "Detection Coverage"
DESCRIPTION = "security_content detection coverage"
DOMAIN = "mitre-enterprise"
class AttackNavWriter():
@staticmethod
def writeAttackNavFile(mitre_techniques : dict, output_path : str) -> None:
max_count = 0
for technique_id in mitre_techniques.keys():
if mitre_techniques[technique_id]['score'] > max_count:
max_count = mitre_techniques[technique_id]['score']
layer_json = {
"version": VERSION,
"name": NAME,
"description": DESCRIPTION,
"domain": DOMAIN,
"techniques": []
}
layer_json["gradient"] = {
"colors": [
"#ffffff",
"#66b1ff",
"#096ed7"
],
"minValue": 0,
"maxValue": max_count
}
layer_json["filters"] = {
"platforms":
["Windows",
"Linux",
"macOS",
"AWS",
"GCP",
"Azure",
"Office 365",
"SaaS"
]
}
layer_json["legendItems"] = [
{
"label": "NO available detections",
"color": "#ffffff"
},
{
"label": "Some detections available",
"color": "#66b1ff"
}
]
layer_json['showTacticRowBackground'] = True
layer_json['tacticRowBackground'] = "#dddddd"
layer_json["sorting"] = 3
for technique_id in mitre_techniques.keys():
layer_technique = {
"techniqueID": technique_id,
"score": mitre_techniques[technique_id]['score'],
"comment": "\n\n".join(mitre_techniques[technique_id]['file_paths'])
}
layer_json["techniques"].append(layer_technique)
with open(output_path, 'w') as outfile:
json.dump(layer_json, outfile, ensure_ascii=False, indent=4)
@@ -1,55 +0,0 @@
import datetime
import os
from jinja2 import Environment, FileSystemLoader
from contentctl_core.domain.entities.security_content_object import SecurityContentObject
class ConfWriter():
@staticmethod
def writeConfFileHeader(output_path : str) -> None:
utc_time = datetime.datetime.utcnow().replace(microsecond=0).isoformat()
j2_env = Environment(
loader=FileSystemLoader(os.path.join(os.path.dirname(__file__), 'templates')),
trim_blocks=True)
template = j2_env.get_template('header.j2')
output = template.render(time=utc_time)
with open(output_path, 'w') as f:
output = output.encode('ascii', 'ignore').decode('ascii')
f.write(output)
@staticmethod
def writeConfFile(template_name : str, output_path : str, objects : list) -> None:
def custom_jinja2_enrichment_filter(string, object):
customized_string = string
for key in dir(object):
if type(key) is not str:
key = key.decode()
if not key.startswith('__') and not key == "_abc_impl" and not callable(getattr(object, key)):
if hasattr(object, key):
customized_string = customized_string.replace("%" + key + "%", str(getattr(object, key)))
for key in dir(object.tags):
if type(key) is not str:
key = key.decode()
if not key.startswith('__') and not key == "_abc_impl" and not callable(getattr(object.tags, key)):
if hasattr(object.tags, key):
customized_string = customized_string.replace("%" + key + "%", str(getattr(object.tags, key)))
return customized_string
j2_env = Environment(
loader=FileSystemLoader(os.path.join(os.path.dirname(__file__), 'templates')),
trim_blocks=True)
j2_env.filters['custom_jinja2_enrichment_filter'] = custom_jinja2_enrichment_filter
template = j2_env.get_template(template_name)
output = template.render(objects=objects)
with open(output_path, 'a') as f:
output = output.encode('ascii', 'ignore').decode('ascii')
f.write(output)
@@ -1,33 +0,0 @@
import os
from jinja2 import Environment, FileSystemLoader
class JinjaWriter:
@staticmethod
def writeObjectsList(template_name : str, output_path : str, objects : list) -> None:
j2_env = Environment(
loader=FileSystemLoader(os.path.join(os.path.dirname(__file__), 'templates')),
trim_blocks=False)
template = j2_env.get_template(template_name)
output = template.render(objects=objects)
with open(output_path, 'w') as f:
output = output.encode('ascii', 'ignore').decode('ascii')
f.write(output)
@staticmethod
def writeObject(template_name : str, output_path : str, object : dict) -> None:
j2_env = Environment(
loader=FileSystemLoader(os.path.join(os.path.dirname(__file__), 'templates')),
trim_blocks=False)
template = j2_env.get_template(template_name)
output = template.render(object=object)
with open(output_path, 'w') as f:
output = output.encode('ascii', 'ignore').decode('ascii')
f.write(output)
@@ -1,10 +0,0 @@
import json
class JsonWriter():
@staticmethod
def writeJsonObject(file_path : str, obj) -> None:
with open(file_path, 'w') as outfile:
json.dump(obj, outfile, ensure_ascii=False, indent=4)
@@ -1,36 +0,0 @@
import os
from contentctl_core.application.adapter.adapter import Adapter
from contentctl_core.domain.entities.enums.enums import SecurityContentType
from contentctl_infrastructure.adapter.attack_nav_writer import AttackNavWriter
class ObjToAttackNavAdapter(Adapter):
def writeObjects(self, objects: list, output_path: str, type: SecurityContentType = None) -> None:
techniques = dict()
for detection in objects:
if detection.tags.mitre_attack_enrichments:
for mitre_attack_enrichment in detection.tags.mitre_attack_enrichments:
if not mitre_attack_enrichment.mitre_attack_id in techniques:
techniques[mitre_attack_enrichment.mitre_attack_id] = {
'score': 1,
'file_paths': ['https://github.com/splunk/security_content/blob/develop/detections/' + detection.source + '/' + self.convertNameToFileName(detection.name)]
}
else:
techniques[mitre_attack_enrichment.mitre_attack_id]['score'] = techniques[mitre_attack_enrichment.mitre_attack_id]['score'] + 1
techniques[mitre_attack_enrichment.mitre_attack_id]['file_paths'].append('https://github.com/splunk/security_content/blob/develop/detections/' + detection.source + '/' + self.convertNameToFileName(detection.name))
AttackNavWriter.writeAttackNavFile(techniques, os.path.join(output_path, 'coverage.json'))
def convertNameToFileName(self, name: str):
file_name = name \
.replace(' ', '_') \
.replace('-','_') \
.replace('.','_') \
.replace('/','_') \
.lower()
file_name = file_name + '.yml'
return file_name
@@ -1,95 +0,0 @@
import os
import glob
import shutil
from contentctl_core.application.adapter.adapter import Adapter
from contentctl_infrastructure.adapter.conf_writer import ConfWriter
from contentctl_core.domain.entities.enums.enums import SecurityContentType
class ObjToConfAdapter(Adapter):
def writeHeaders(self, output_folder: str) -> None:
ConfWriter.writeConfFileHeader(os.path.join(output_folder, 'default/analyticstories.conf'))
ConfWriter.writeConfFileHeader(os.path.join(output_folder, 'default/savedsearches.conf'))
ConfWriter.writeConfFileHeader(os.path.join(output_folder, 'default/collections.conf'))
ConfWriter.writeConfFileHeader(os.path.join(output_folder, 'default/es_investigations.conf'))
ConfWriter.writeConfFileHeader(os.path.join(output_folder, 'default/macros.conf'))
ConfWriter.writeConfFileHeader(os.path.join(output_folder, 'default/transforms.conf'))
ConfWriter.writeConfFileHeader(os.path.join(output_folder, 'default/workflow_actions.conf'))
def writeObjects(self, objects: list, output_path: str, type: SecurityContentType = None) -> None:
if type == SecurityContentType.detections:
ConfWriter.writeConfFile('savedsearches_detections.j2',
os.path.join(output_path, 'default/savedsearches.conf'),
objects)
ConfWriter.writeConfFile('analyticstories_detections.j2',
os.path.join(output_path, 'default/analyticstories.conf'),
objects)
ConfWriter.writeConfFile('macros_detections.j2',
os.path.join(output_path, 'default/macros.conf'),
objects)
elif type == SecurityContentType.stories:
ConfWriter.writeConfFile('analyticstories_stories.j2',
os.path.join(output_path, 'default/analyticstories.conf'),
objects)
elif type == SecurityContentType.baselines:
ConfWriter.writeConfFile('savedsearches_baselines.j2',
os.path.join(output_path, 'default/savedsearches.conf'),
objects)
elif type == SecurityContentType.investigations:
ConfWriter.writeConfFile('savedsearches_investigations.j2',
os.path.join(output_path, 'default/savedsearches.conf'),
objects)
ConfWriter.writeConfFile('analyticstories_investigations.j2',
os.path.join(output_path, 'default/analyticstories.conf'),
objects)
workbench_panels = []
for investigation in objects:
if investigation.inputs:
response_file_name_xml = investigation.lowercase_name + "___response_task.xml"
workbench_panels.append(investigation)
investigation.search = investigation.search.replace(">","&gt;")
investigation.search = investigation.search.replace("<","&lt;")
ConfWriter.writeConfFileHeader(os.path.join(output_path,
'default/data/ui/panels/', str("workbench_panel_" + response_file_name_xml)))
ConfWriter.writeConfFile('panel.j2',
os.path.join(output_path,
'default/data/ui/panels/', str("workbench_panel_" + response_file_name_xml)),
[investigation.search])
ConfWriter.writeConfFile('es_investigations_investigations.j2',
os.path.join(output_path, 'default/es_investigations.conf'),
workbench_panels)
ConfWriter.writeConfFile('workflow_actions.j2',
os.path.join(output_path, 'default/workflow_actions.conf'),
workbench_panels)
elif type == SecurityContentType.lookups:
ConfWriter.writeConfFile('collections.j2',
os.path.join(output_path, 'default/collections.conf'),
objects)
ConfWriter.writeConfFile('transforms.j2',
os.path.join(output_path, 'default/transforms.conf'),
objects)
files = glob.iglob(os.path.join(os.path.dirname(__file__), '../../../..' , 'lookups', '*.csv'))
for file in files:
if os.path.isfile(file):
shutil.copy(file, os.path.join(output_path, 'lookups'))
elif type == SecurityContentType.macros:
ConfWriter.writeConfFile('macros.j2',
os.path.join(output_path, 'default/macros.conf'),
objects)
@@ -1,79 +0,0 @@
import os
from contentctl_core.application.adapter.adapter import Adapter
from contentctl_infrastructure.adapter.json_writer import JsonWriter
from contentctl_core.domain.entities.enums.enums import SecurityContentType
class ObjToJsonAdapter(Adapter):
def writeObjects(self, objects: list, output_path: str, type: SecurityContentType = None) -> None:
if type == SecurityContentType.detections:
obj_array = []
for detection in objects:
obj_array.append(detection.dict(exclude_none=True,
exclude =
{
"deprecated": True,
"experimental": True,
"annotations": True,
"risk": True,
"playbooks": True,
"baselines": True,
"mappings": True,
"test": True,
"deployment": True
}
))
JsonWriter.writeJsonObject(os.path.join(output_path, 'detections.json'), obj_array)
elif type == SecurityContentType.stories:
obj_array = []
for story in objects:
obj_array.append(story.dict(exclude_none=True))
JsonWriter.writeJsonObject(os.path.join(output_path, 'stories.json'), obj_array)
elif type == SecurityContentType.baselines:
obj_array = []
for baseline in objects:
obj_array.append(baseline.dict(
exclude =
{
"deployment": True
}
))
JsonWriter.writeJsonObject(os.path.join(output_path, 'baselines.json'), obj_array)
elif type == SecurityContentType.investigations:
obj_array = []
for investigation in objects:
obj_array.append(investigation.dict(exclude_none=True))
JsonWriter.writeJsonObject(os.path.join(output_path, 'response_tasks.json'), obj_array)
elif type == SecurityContentType.lookups:
obj_array = []
for lookup in objects:
obj_array.append(lookup.dict(exclude_none=True))
JsonWriter.writeJsonObject(os.path.join(output_path, 'lookups.json'), obj_array)
elif type == SecurityContentType.macros:
obj_array = []
for macro in objects:
obj_array.append(macro.dict(exclude_none=True))
JsonWriter.writeJsonObject(os.path.join(output_path, 'macros.json'), obj_array)
elif type == SecurityContentType.deployments:
obj_array = []
for deployment in objects:
obj_array.append(deployment.dict(exclude_none=True))
JsonWriter.writeJsonObject(os.path.join(output_path, 'deployments.json'), obj_array)
@@ -1,64 +0,0 @@
import os
from contentctl_core.application.adapter.adapter import Adapter
from contentctl_core.domain.entities.enums.enums import SecurityContentType
from contentctl_infrastructure.adapter.jinja_writer import JinjaWriter
class ObjToMdAdapter(Adapter):
def writeObjects(self, objects: list, output_path: str, type: SecurityContentType = None) -> None:
attack_tactics = set()
datamodels = set()
categories = set()
for story in objects[0]:
if story.tags.category:
categories.update(story.tags.category)
for detection in objects[1]:
if detection.tags.mitre_attack_enrichments:
for attack in detection.tags.mitre_attack_enrichments:
attack_tactics.update(attack.mitre_attack_tactics)
if detection.datamodel:
datamodels.update(detection.datamodel)
JinjaWriter.writeObjectsList('doc_navigation.j2', os.path.join(output_path, '_data/navigation.yml'),
{
'attack_tactics': sorted(list(attack_tactics)),
'datamodels': sorted(list(datamodels)),
'categories': sorted(list(categories))
}
)
self.writeNavigationPageObjects(sorted(list(datamodels)), output_path)
self.writeNavigationPageObjects(sorted(list(attack_tactics)), output_path)
self.writeNavigationPageObjects(sorted(list(categories)), output_path)
JinjaWriter.writeObjectsList('doc_story_page.j2', os.path.join(output_path, '_pages/stories.md'), sorted(objects[0], key=lambda x: x.name))
self.writeObjectsMd(objects[0], os.path.join(output_path, '_stories'), 'doc_stories.j2')
JinjaWriter.writeObjectsList('doc_detection_page.j2', os.path.join(output_path, '_pages/detections.md'), sorted(objects[1], key=lambda x: x.name))
self.writeDetectionsMd(objects[1], os.path.join(output_path, '_posts'), 'doc_detections.j2')
JinjaWriter.writeObjectsList('doc_playbooks_page.j2', os.path.join(output_path, '_pages/paybooks.md'), sorted(objects[2], key=lambda x: x.name))
self.writeObjectsMd(objects[2], os.path.join(output_path, '_playbooks'), 'doc_playbooks.j2')
def writeNavigationPageObjects(self, objects: list, output_path: str) -> None:
for obj in objects:
JinjaWriter.writeObject('doc_navigation_pages.j2', os.path.join(output_path, '_pages', obj.lower().replace(' ', '_') + '.md'),
{
'name': obj
}
)
def writeObjectsMd(self, objects, output_path: str, template_name: str) -> None:
for obj in objects:
JinjaWriter.writeObject(template_name, os.path.join(output_path, obj.name.lower().replace(' ', '_') + '.md'), obj)
def writeDetectionsMd(self, objects, output_path: str, template_name: str) -> None:
for obj in objects:
JinjaWriter.writeObject(template_name, os.path.join(output_path, obj.date + '-' + obj.name.lower().replace(' ', '_') + '.md'), obj)
@@ -1,33 +0,0 @@
import os
from contentctl_core.application.adapter.adapter import Adapter
from contentctl_core.domain.entities.enums.enums import SecurityContentType
from contentctl_infrastructure.adapter.jinja_writer import JinjaWriter
class ObjToSvgAdapter(Adapter):
def writeObjects(self, objects: list, output_path: str, type: SecurityContentType = None) -> None:
detections_tmp = objects
detection_without_test = 0
detections = []
obj = dict()
for detection in detections_tmp:
if not detection.deprecated:
detections.append(detection)
if not detection.test and not detection.experimental:
detection_without_test = detection_without_test + 1
obj['count'] = len(detections)
obj['coverage'] = (obj['count'] - detection_without_test)/obj['count']
obj['coverage'] = "{:.0%}".format(obj['coverage'])
JinjaWriter.writeObject('detection_count.j2', os.path.join(output_path, 'detection_count.svg'), obj)
JinjaWriter.writeObject('detection_coverage.j2', os.path.join(output_path, 'detection_coverage.svg'), obj)
@@ -1,121 +0,0 @@
import os
import re
from contentctl_infrastructure.adapter.yml_writer import YmlWriter
from contentctl_core.application.adapter.adapter import Adapter
from contentctl_core.domain.entities.enums.enums import SecurityContentType
class ObjToYmlAdapter(Adapter):
def writeObjectsInPlace(self, objects: list) -> None:
for object in objects:
file_path = object['file_path']
object.pop('file_path')
object.pop('deprecated')
object.pop('experimental')
YmlWriter.writeYmlFile(file_path, object)
def writeObjects(self, objects: list, output_path: str, type: SecurityContentType = None) -> None:
for obj in objects:
file_name = "ssa___" + self.convertNameToFileName(obj.name)
if self.isComplexBARule(obj.search):
file_path = os.path.join(output_path, 'complex', file_name)
else:
file_path = os.path.join(output_path, 'srs', file_name)
# remove unncessary fields
YmlWriter.writeYmlFile(file_path, obj.dict(
include =
{
"name": True,
"id": True,
"version": True,
"description": True,
"search": True,
"how_to_implement": True,
"known_false_positives": True,
"references": True,
"tags":
{
"analytic_story": True,
"cis20" : True,
"nist": True,
"kill_chain_phases": True,
"mitre_attack_id": True,
"risk_severity": True,
"security_domain": True,
"required_fields": True
},
"test":
{
"name": True,
"tests": {
'__all__':
{
"name": True,
"file": True,
"pass_condition": True,
"attack_data": {
'__all__':
{
"file_name": True,
"data": True,
"source": True
}
}
}
}
}
}
))
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']))
test_obj = {}
test_obj['name'] = object['name'] + ' Unit Test'
test_obj['tests'] = [
{
'name': object['name'],
'file': object['source'] + '/' + self.convertNameToFileName(object['name']),
'pass_condition': '| stats count | where count > 0',
'earliest_time': '-24h',
'latest_time': 'now',
'attack_data': [
{
'file_name': 'UPDATE',
'data': 'UPDATE',
'source': 'UPDATE',
'sourcetype': 'UPDATE'
}
]
}
]
file_path_test = os.path.join(os.path.dirname(__file__), '../../../../tests', object['source'], self.convertNameToFileName(object['name']))
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']))
YmlWriter.writeYmlFile(file_path, object)
def convertNameToFileName(self, name: str):
file_name = name \
.replace(' ', '_') \
.replace('-','_') \
.replace('.','_') \
.replace('/','_') \
.lower()
file_name = file_name + '.yml'
return file_name
def isComplexBARule(self, search):
return re.findall("stats|first_time_event|adaptive_threshold", search)
@@ -1,22 +0,0 @@
### DETECTIONS ###
{% for detection in objects %}
{% if (detection.type == 'TTP' or detection.type == 'Anomaly' or detection.type == 'Hunting' or detection.type == 'Correlation') %}
[savedsearch://ESCU - {{ detection.name }} - Rule]
type = detection
asset_type = {{ detection.tags.asset_type }}
confidence = medium
explanation = {{ detection.description }}
{% if detection.how_to_implement is defined %}
how_to_implement = {{ detection.how_to_implement }}
{% else %}
how_to_implement = none
{% endif %}
annotations = {{ detection.mappings | tojson }}
known_false_positives = {{ detection.known_false_positives }}
providing_technologies = []
{% endif %}
{% endfor %}
### END DETECTIONS ###
@@ -1,21 +0,0 @@
### RESPONSE TASKS ###
{% for detection in objects %}
{% if (detection.type == 'Investigation') %}
[savedsearch://ESCU - {{ detection.name }} - Response Task]
type = investigation
explanation = none
{% if detection.how_to_implement is defined %}
how_to_implement = {{ detection.how_to_implement }}
{% else %}
how_to_implement = none
{% endif %}
known_false_positives = not defined
earliest_time_offset = 14400
latest_time_offset = 0
{% endif %}
{% endfor %}
### END RESPONSE TASKS ###
@@ -1,21 +0,0 @@
### STORIES ###
{% for story in objects %}
[analytic_story://{{ story.name }}]
category = {{ story.tags.category[0] }}
last_updated = {{ story.date }}
version = {{ story.version }}
references = {{ story.references | tojson }}
maintainers = [{"company": "{{ story.author_company }}", "email": "-", "name": "{{ story.author_name }}"}]
spec_version = 3
searches = {{ (story.detection_names + story.investigation_names) | tojson }}
description = {{ story.description }}
{% if story.narrative is defined %}
narrative = {{ story.narrative }}
{% endif %}
{% endfor %}
### END STORIES ###
@@ -1,7 +0,0 @@
{% for lookup in objects %}
[{{ lookup.name }}]
enforceTypes = false
replicate = false
{% endfor %}
@@ -1,18 +0,0 @@
<?xml version="1.0"?>
<svg xmlns="http://www.w3.org/2000/svg" width="105" height="20">
<linearGradient id="a" x2="0" y2="100%">
<stop offset="0" stop-color="#bbb" stop-opacity=".1"/>
<stop offset="2" stop-opacity=".1"/>
</linearGradient>
<rect rx="3" width="65" height="20" fill="#555"/> <!-- Comment -->
<rect rx="3" x="65" width="40" height="20" fill="#4c1"/>
<path fill="#4c1" d="M63 0h4v20h-4z"/>
<rect rx="3" width="105" height="20" fill="url(#a)"/>
<g fill="#fff" text-anchor="middle" font-family="DejaVu Sans,Verdana,Geneva,sans-serif" font-size="11">
<text x="30" y="14">detections</text>
<text x="83" y="14">{{ object.count }}</text>
</g>
</svg>

Before

Width:  |  Height:  |  Size: 670 B

@@ -1,18 +0,0 @@
<?xml version="1.0"?>
<svg xmlns="http://www.w3.org/2000/svg" width="100" height="20">
<linearGradient id="a" x2="0" y2="100%">
<stop offset="0" stop-color="#bbb" stop-opacity=".1"/>
<stop offset="2" stop-opacity=".1"/>
</linearGradient>
<rect rx="3" width="60" height="20" fill="#555"/> <!-- Comment -->
<rect rx="3" x="60" width="40" height="20" fill="#4c1"/>
<path fill="#4c1" d="M58 0h4v20h-4z"/>
<rect rx="3" width="100" height="20" fill="url(#a)"/>
<g fill="#fff" text-anchor="middle" font-family="DejaVu Sans,Verdana,Geneva,sans-serif" font-size="11">
<text x="30" y="14">coverage</text>
<text x="80" y="14">{{ object.coverage }}</text>
</g>
</svg>

Before

Width:  |  Height:  |  Size: 671 B

@@ -1,19 +0,0 @@
---
title: "Detections"
layout: categories
author_profile: false
permalink: /detections/
classes: wide
sidebar:
nav: "detections"
---
| Name | Technique | Type |
| -------------- | --------------- | --------------- |
{%- for detection in objects -%}
{% if detection.tags.mitre_attack_enrichments %}
| [{{ detection.name }}](/{{ detection.source }}/{{ detection.name | lower | replace(' ', '_') }}/) | {% for attack in detection.tags.mitre_attack_enrichments -%} [{{ attack.mitre_attack_technique }}](/tags/#{{ attack.mitre_attack_technique | lower | replace(" ", "-") }}){% if not loop.last -%}, {% endif -%}{%- endfor %} | {{ detection.type }} |
{%- else %}
| [{{ detection.name }}]() | None | {{ detection.type }} |
{%- endif -%}
{%- endfor -%}
@@ -1,153 +0,0 @@
---
title: "{{object.name}}"
excerpt: "{% if object.tags.mitre_attack_enrichments %}{% for attack in object.tags.mitre_attack_enrichments -%}
{{ attack.mitre_attack_technique }}
{% if not loop.last -%}, {% endif -%}
{% endfor %}{% endif -%}"
categories:
- {{object.source|capitalize}}
last_modified_at: {{object.date}}
toc: true
toc_label: ""
tags:
{% if object.tags.mitre_attack_enrichments %}
{%- for attack in object.tags.mitre_attack_enrichments %}
- {{ attack.mitre_attack_technique }}
{%- endfor -%}
{%- for attack in object.tags.mitre_attack_enrichments %}
{%- for tactic in attack.mitre_attack_tactics %}
- {{ tactic }}
{%- endfor -%}
{%- endfor -%}
{% endif -%}
{%- for product in object.tags.product %}
- {{ product }}
{%- endfor -%}
{% if object.cve_enrichment -%}
{%- for cve in object.cve_enrichment %}
- {{ cve.id }}
{%- endfor -%}
{% endif -%}
{%- for datamodel in object.datamodel %}
- {{ datamodel }}
{%- endfor %}
---
{% if object.experimental is sameas true -%}
### ⚠️ WARNING THIS IS A EXPERIMENTAL object
We have not been able to test, simulate, or build datasets for this object. Use at your own risk. This analytic is **NOT** supported.
{% endif %}
[Try in Splunk Security Cloud](https://www.splunk.com/en_us/cyber-security.html){: .btn .btn--success}
#### Description
{{ object.description }}
- **Type**: [{{ object.type }}](https://github.com/splunk/security_content/wiki/object-Analytic-Types)
- **Product**: {{ object.tags.product|join(', ') }}
- **Datamodel**: {% for datamodel in object.datamodel %}[{{ datamodel }}](https://docs.splunk.com/Documentation/CIM/latest/User/{{ datamodel|replace("_", "")}}){% if not loop.last %}, {% endif %}{%-endfor %}
- **Last Updated**: {{ object.date }}
- **Author**: {{object.author}}
- **ID**: {{ object.id }}
{% if object.tags.mitre_attack_id %}
#### [ATT&CK](https://attack.mitre.org/)
| ID | Technique | Tactic |
| -------------- | ---------------- |-------------------- |
{%- for attack in object.tags.mitre_attack_enrichments %}
{% if attack.mitre_attack_id -%}
{% set sub_technique = attack.mitre_attack_id.split('.') %}{% if sub_technique | length > 1 -%}
| [{{ attack.mitre_attack_id }}](https://attack.mitre.org/techniques/{{sub_technique[0]}}/{{sub_technique[1]}}/) | {{ attack.mitre_attack_technique }} | {{ attack.mitre_attack_tactics|join(', ') }} |
{% else -%}
| [{{ attack.mitre_attack_id }}](https://attack.mitre.org/techniques/{{attack.mitre_attack_id}}/) | {{ attack.mitre_attack_technique }} | {{ attack.mitre_attack_tactics|join(', ') }} |
{% endif -%}
{% endif -%}
{% endfor %}
{% endif -%}
#### Search
```
{{ object.search|replace("|", "\n|")|safe }}
```
{% if object.macros -%}
#### Macros
The SPL above uses the following Macros:
{% for macro in object.macros -%}
{% if not '_filter' in macro.name -%}
* [{{ macro.name }}](https://github.com/splunk/security_content/blob/develop/macros/{{macro.name}}.yml)
{% endif -%}
{% endfor %}
{% endif -%}
Note that `{{object.name | lower | replace(" ", "_") }}_filter` is a empty macro by default. It allows the user to filter out any results (false positives) without editing the SPL.
{% if object.lookups -%}
#### Lookups
The SPL above uses the following Lookups:
{% for lookup in object.lookups -%}
* [{{lookup.name}}](https://github.com/splunk/security_content/blob/develop/lookups/{{lookup.name}}.yml) with [data](https://github.com/splunk/security_content/tree/develop/lookups/{{lookup.name}}.csv)
{% endfor %}
{% endif -%}
#### Required field
{% for field in object.tags.required_fields -%}
* {{ field }}
{% endfor %}
#### How To Implement
{{ object.how_to_implement}}
#### Known False Positives
{{ object.known_false_positives}}
#### Associated Analytic story
{% for story in object.tags.analytic_story -%}
* [{{ story }}](/stories/{{story|lower|replace(" ", "_")}})
{% endfor %}
#### Kill Chain Phase
{% for phase in object.tags.kill_chain_phases -%}
* {{ phase }}
{% endfor %}
{% if object.tags.observable %}
#### RBA
| Risk Score | Impact | Confidence | Message |
| ----------- | ----------- |--------------|--------------|
| {{(object.tags.impact * object.tags.confidence)/100}} | {{ object.tags.impact }} | {{ object.tags.confidence }} | {{object.tags.message}} |
{% endif %}
{% if object.cve_enrichment -%}
#### CVE
| ID | Summary | [CVSS](https://nvd.nist.gov/vuln-metrics/cvss) |
| ----------- | ----------- | -------------- |
{% for cve in object.cve_enrichment -%}
| [{{ cve.id }}](https://nvd.nist.gov/vuln/detail/{{cve.id}}) | {{ cve.summary }} | {{ cve.cvss }} |
{% endfor %}
{% endif %}
#### Reference
{% if object.references %}
{% for reference in object.references -%}
* [{{ reference }}]({{ reference }})
{% endfor %}
{% endif %}
#### Test Dataset
Replay any dataset to Splunk Enterprise by using our [`replay.py`](https://github.com/splunk/attack_data#using-replaypy) tool or the [UI](https://github.com/splunk/attack_data#using-ui).
Alternatively you can replay a dataset into a [Splunk Attack Range](https://github.com/splunk/attack_range#replay-dumps-into-attack-range-splunk-server)
{% if object.tags.dataset %}
{% for dataset in object.tags.dataset -%}
* [{{dataset}}]({{ dataset }})
{% endfor %}
{% endif %}
[*source*](https://github.com/splunk/security_content/tree/develop/detections/{% if object.experimental is sameas true -%}experimental/{%- endif -%}{{object.source}}/{{ object.name | lower | replace (" ", "_") }}.yml) \| *version*: **{{object.version}}**
@@ -1,46 +0,0 @@
main:
- title: "Detections"
url: /detections/
- title: "Analytic Stories"
url: /stories/
- title: "Playbooks"
url: /playbooks/
- title: "Tags"
url: /tags/
- title: "About"
url: https://www.splunk.com/en_us/cyber-security/threat-research.html
detections:
- title: "Tactic"
children:
{%- for tactic in objects['attack_tactics'] %}
- title: {{ tactic }}
url: /detections/{{ tactic | lower | replace(" ", "_") }}/
{%- endfor %}
- title: "Datamodel"
children:
{%- for datamodel in objects['datamodels'] %}
- title: {{ datamodel }}
url: /detections/{{ datamodel | lower | replace(" ", "_") }}/
{%- endfor %}
- title: "Product"
children:
- title: "Splunk Enterprise Security"
url: /tags/#splunk-enterprise-security
- title: "Splunk Behavioral Analytics"
url: /tags/#splunk-behavioral-analytics
- title: "Splunk Security Analytics for AWS"
url: /tags/#splunk-security-analytics-for-aws
stories:
- title: "Use Case"
children:
{%- for category in objects['categories'] %}
- title: {{ category }}
url: /stories/{{ category | lower | replace(" ", "_") }}/
{%- endfor %}
playbooks:
- title: "Type"
children:
- title: "Response"
url: /tags/#response/
- title: "Investigation"
url: /tags/#investigation/
@@ -1,9 +0,0 @@
---
title: {{ object['name'] }}
layout: tag
author_profile: false
taxonomy: {{ object['name'] }}
permalink: /detections/{{ object['name'] | lower | replace(' ', '_') }}/
sidebar:
nav: "detections"
---
@@ -1,58 +0,0 @@
---
title: "{{object.name}}"
last_modified_at: {{object.date}}
toc: true
toc_label: ""
tags:
- {{ object.type }}
{%- for product in object.tags.product %}
- {{ product }}
{%- endfor -%}
{%- for app in object.app_list %}
- {{ app }}
{%- endfor %}
{%- for tag in object.tags.platform_tags %}
- {{ tag }}
{%- endfor %}
---
[Try in Splunk SOAR](https://www.splunk.com/en_us/software/splunk-security-orchestration-and-automation.html){: .btn .btn--success}
#### Description
{{ object.description }}
- **Type**: {{ object.type }}
- **Product**: {{ object.tags.product|join(', ') }}
- **Apps**: {% for app in object.app_list %}[{{ app }}](https://splunkbase.splunk.com/apps/#/search/{{ app }}/product/soar){% if not loop.last %}, {% endif %}{%-endfor %}
- **Last Updated**: {{ object.date }}
- **Author**: {{object.author}}
- **ID**: {{ object.id }}
#### Associated Detections
{% if object.tags.detection_objects %}
{% for detection in object.tags.detection_objects -%}
* [{{ detection.name }}](/{{ detection.soure }}/{{detection.name|lower|replace(" ", "_")}}/)
{% endfor %}
{% endif %}
#### How To Implement
{{ object.how_to_implement}}
#### Playbooks
![](https://raw.githubusercontent.com/splunk/security_content/develop/playbooks/{{ object.playbook | lower | replace(" ", "_")}}.png)
#### Required field
{% for field in object.tags.object_fields -%}
* {{ field }}
{% endfor %}
#### Reference
{% if object.references %}
{% for reference in object.references -%}
* [{{ reference }}]({{ reference }})
{% endfor %}
{% endif %}
[*source*](https://github.com/splunk/security_content/tree/develop/playbooks/{{ object.name | lower | replace (" ", "_") }}.yml) \| *version*: **{{object.version}}**
@@ -1,19 +0,0 @@
---
title: "Playbooks"
layout: collection
author_profile: false
permalink: /playbooks/
classes: wide
sidebar:
nav: "playbooks"
---
| Name | Detections | Type |
| --------| ---------- | ----------- |
{% for playbook in objects -%}
{% if playbook.tags.detection_objects -%}
| [{{ playbook.name }}](/playbooks/{{ playbook.name|lower|replace(' ', '_') }}/)|{% for detection in playbook.tags.detection_objects -%}[{{ detection.name }}](/{{ detection.source }}/{{detection.name|lower|replace(" ", "_")}}/){%- endfor -%} | {{ playbook.type }} |
{% else -%}
| [{{ playbook.name }}](/playbooks/{{ playbook.name|lower|replace(' ', '_') }}/)| None | {{ playbook.type }} |
{% endif -%}
{%- endfor -%}
@@ -1,51 +0,0 @@
---
title: "{{object.name}}"
last_modified_at: {{object.date}}
toc: true
toc_label: ""
tags:
{%- for product in object.tags.product %}
- {{ product }}
{%- endfor -%}
{%- for datamodel in object.tags.datamodels %}
- {{ datamodel }}
{%- endfor -%}
{%- for phase in object.tags.kill_chain_phases %}
- {{ phase }}
{%- endfor %}
---
[Try in Splunk Security Cloud](https://www.splunk.com/en_us/cyber-security.html){: .btn .btn--success}
#### Description
{{ object.description }}
- **Product**: {{ object.tags.product|join(', ') }}
- **Datamodel**: {% for datamodel in object.tags.datamodels %}[{{ datamodel }}](https://docs.splunk.com/Documentation/CIM/latest/User/{{ datamodel|replace("_", "")}}){% if not loop.last %}, {% endif %}{%-endfor %}
- **Last Updated**: {{ object.date }}
- **Author**: {{object.author}}
- **ID**: {{ object.id }}
#### Narrative
{{ object.narrative }}
#### Detections
| Name | Technique | Type |
| ----------- | ----------- |--------------|
{%- if object.detections %}
{%- for detection in object.detections %}
| [{{ detection.name }}](/{{ detection.source }}/{{ detection.name | lower | replace(' ', '_') }}/) | {% if detection.tags.mitre_attack_enrichments %}{% for attack in detection.tags.mitre_attack_enrichments -%}[{{ attack.mitre_attack_technique }}](/tags/#{{ attack.mitre_attack_technique | lower | replace(" ", "-") }}){% if not loop.last %}, {% endif %}{%- endfor %}{% else %}None{%- endif -%} | {{ detection.type }} |
{%- endfor %}
{%- endif %}
#### Reference
{% if object.references %}
{% for reference in object.references -%}
* [{{ reference }}]({{ reference }})
{% endfor %}
{% endif %}
[*source*](https://github.com/splunk/security_content/tree/develop/stories/{{ object.name | lower | replace (" ", "_") }}.yml) \| *version*: **{{object.version}}**
@@ -1,19 +0,0 @@
---
title: Analytic Stories
layout: collection
permalink: /stories/
collection: stories
classes: wide
sidebar:
nav: "stories"
---
| Name | Technique | Tactic |
| ----------- | ----------- |--------------|
{%- for story in objects -%}
{% if story.tags.mitre_attack_enrichments %}
| [{{ story.name }}]({{ story.name | lower | replace(" ", "_") }}) | {% for attack in story.tags.mitre_attack_enrichments -%} [{{ attack.mitre_attack_technique }}](/tags/#{{ attack.mitre_attack_technique | lower | replace(" ", "-") }}){% if not loop.last -%}, {% endif -%}{%- endfor %} | {% for tactic in story.tags.mitre_attack_tactics -%} [{{ tactic }}](/tags/#{{ tactic | lower | replace(" ", "-") }}){% if not loop.last -%}, {% endif -%}{%- endfor %} |
{%- else %}
| [{{ story.name }}]() | None | None |
{%- endif -%}
{%- endfor -%}

Some files were not shown because too many files have changed in this diff Show More