admin/splunk-security_content
1
0
Fork 0
mirror of https://github.com/splunk/security_content synced 2026-06-08 17:32:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Branch was auto-updated.

Browse Source
This commit is contained in:
pyth0n1c
2022-07-07 17:08:06 -07:00
committed by GitHub
parent c220041d5c 4bae19f854
commit 87a807c19a
7 changed files with 141 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/build-and-validate.yml
+6 -6
View File
@@ -97,8 +97,8 @@ jobs:
- name: content_ctl validate
run: |
source .venv/bin/activate
python contentctl.py -p . ${{ steps.vars.outputs.skip_enrichment_var }} validate -pr ESCU > /dev/null
python contentctl.py -p . ${{ steps.vars.outputs.skip_enrichment_var }} validate -pr SSA > /dev/null
python contentctl.py -p . ${{ steps.vars.outputs.skip_enrichment_var }} validate -pr ESCU
python contentctl.py -p . ${{ steps.vars.outputs.skip_enrichment_var }} validate -pr SSA
#Now generate the documentation (uses Node)
@@ -120,8 +120,8 @@ jobs:
run: |
source .venv/bin/activate
rm -rf dist/escu/default/data/ui/panels/*.xml
python contentctl.py --path . ${{ steps.vars.outputs.skip_enrichment_var }} generate --product ESCU --output dist/escu > /dev/null
python contentctl.py --path . ${{ steps.vars.outputs.skip_enrichment_var }} generate --product SSA --output dist/ssa > /dev/null
python contentctl.py --path . ${{ steps.vars.outputs.skip_enrichment_var }} generate --product ESCU --output dist/escu
python contentctl.py --path . ${{ steps.vars.outputs.skip_enrichment_var }} generate --product SSA --output dist/ssa
- name: Copy lookups .mlmodel files
run: |
@@ -353,12 +353,12 @@ jobs:
- name: Run doc-gen
run: |
source venv/bin/activate
python3 contentctl.py -p . docgen -o docs > /dev/null
python3 contentctl.py -p . docgen -o docs
- name: Run reporting
run: |
source venv/bin/activate
python3 contentctl.py -p . reporting > /dev/null
python3 contentctl.py -p . reporting
- name: Update github with new docs and package bits
run: |
bin/contentctl_project/contentctl_core/application/factory/ba_factory.py
+16 -3
View File
@@ -50,6 +50,10 @@ class BAFactory():
validation_error_found = False
files_with_ssa = [f for f in files if 'ssa___' in f]
already_ran = False
progress_percent = 0
type_string = "UNKNOWN TYPE"
for index,file in enumerate(files_with_ssa):
#Index + 1 because we are zero indexed, not 1 indexed. This ensures
@@ -59,22 +63,31 @@ class BAFactory():
if 'ssa__' in file:
progress_percent = ((index+1)/len(files_with_ssa)) * 100
try:
type_string = "UNKNOWN TYPE"
if type == SecurityContentType.detections:
print(f"\r{'Detections Progress'.rjust(23)}: [{progress_percent:3.0f}%]...", end="", flush=True)
type_string = "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:
print(f"\r{'Unit Tests Progress'.rjust(23)}: [{progress_percent:3.0f}%]...", end="", flush=True)
type_string = "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)
else:
raise(Exception(f"Unsupported content type: [{type}]"))
if (sys.stdout.isatty() and sys.stdin.isatty() and sys.stderr.isatty()) or not already_ran:
already_ran = True
print(f"\r{f'{type_string} Progress'.rjust(23)}: [{progress_percent:3.0f}%]...", end="", flush=True)
except ValidationError as e:
print('\nValidation Error for file ' + file)
print(e)
validation_error_found = True
print(f"\r{f'{type_string} Progress'.rjust(23)}: [{progress_percent:3.0f}%]...", end="", flush=True)
print("Done!")
if validation_error_found:
bin/contentctl_project/contentctl_core/application/factory/factory.py
+23 -9
View File
@@ -157,6 +157,10 @@ class Factory():
thread.join()
'''
already_ran = False
progress_percent = 0
type_string = "UNKNOWN TYPE"
#Non threaded, production version of the construction code
files_without_ssa = [f for f in files if 'ssa___' not in f]
for index,file in enumerate(files_without_ssa):
@@ -165,47 +169,48 @@ class Factory():
# that printouts end at 100%, not some other number
progress_percent = ((index+1)/len(files_without_ssa)) * 100
try:
type_string = "UNKNOWN TYPE"
if type == SecurityContentType.lookups:
print(f"\r{'Lookups Progress'.rjust(23)}: [{progress_percent:3.0f}%]...", end="", flush=True)
type_string = "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:
print(f"\r{'Playbooks Progress'.rjust(23)}: [{progress_percent:3.0f}%]...", end="", flush=True)
type_string = "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:
print(f"\r{'Deployments Progress'.rjust(23)}: [{progress_percent:3.0f}%]...", end="", flush=True)
type_string = "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:
print(f"\r{'Playbooks Progress'.rjust(23)}: [{progress_percent:3.0f}%]...", end="", flush=True)
type_string = "Playbooks"
self.input_dto.director.constructPlaybook(self.input_dto.playbook_builder, file)
self.output_dto.playbooks.append(self.input_dto.playbook_builder.getObject())
elif type == SecurityContentType.baselines:
print(f"\r{'Baselines Progress'.rjust(23)}: [{progress_percent:3.0f}%]...", end="", flush=True)
type_string = "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:
print(f"\r{'Investigations Progress'.rjust(23)}: [{progress_percent:3.0f}%]...", end="", flush=True)
type_string = "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:
print(f"\r{'Stories Progress'.rjust(23)}: [{progress_percent:3.0f}%]...", end="", flush=True)
type_string = "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:
print(f"\r{'Detections Progress'.rjust(23)}: [{progress_percent:3.0f}%]...", end="", flush=True)
type_string = "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,
@@ -214,15 +219,24 @@ class Factory():
self.output_dto.detections.append(detection)
elif type == SecurityContentType.unit_tests:
print(f"\r{'Unit Tests Progress'.rjust(23)}: [{progress_percent:3.0f}%]...", end="", flush=True)
type_string = "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)
else:
raise Exception(f"Unsupported type: [{type}]")
if (sys.stdout.isatty() and sys.stdin.isatty() and sys.stderr.isatty()) or not already_ran:
already_ran = True
print(f"\r{f'{type_string} Progress'.rjust(23)}: [{progress_percent:3.0f}%]...", end="", flush=True)
except ValidationError as e:
print('\nValidation Error for file ' + file)
print(e)
validation_error_found = True
print(f"\r{f'{type_string} Progress'.rjust(23)}: [{progress_percent:3.0f}%]...", end="", flush=True)
print("Done!")
if validation_error_found:
bin/contentctl_project/contentctl_infrastructure/adapter/obj_to_md_adapter.py
+7 -4
View File
@@ -1,6 +1,6 @@
import os
import asyncio
import sys
from bin.contentctl_project.contentctl_core.application.adapter.adapter import Adapter
from bin.contentctl_project.contentctl_core.domain.entities.enums.enums import SecurityContentType
from bin.contentctl_project.contentctl_infrastructure.adapter.jinja_writer import JinjaWriter
@@ -14,7 +14,8 @@ class ObjToMdAdapter(Adapter):
self.files_to_write = sum([len(obj) for obj in objects])
self.index = 0
progress_percent = ((self.index+1)/self.files_to_write) * 100
print(f"\r{'Docgen Progress'.rjust(23)}: [{progress_percent:3.0f}%]...", end="", flush=True)
if (sys.stdout.isatty() and sys.stdin.isatty() and sys.stderr.isatty()):
print(f"\r{'Docgen Progress'.rjust(23)}: [{progress_percent:3.0f}%]...", end="", flush=True)
attack_tactics = set()
datamodels = set()
@@ -65,7 +66,8 @@ class ObjToMdAdapter(Adapter):
for obj in objects:
progress_percent = ((self.index+1)/self.files_to_write) * 100
self.index+=1
print(f"\r{'Docgen Progress'.rjust(23)}: [{progress_percent:3.0f}%]...", end="", flush=True)
if (sys.stdout.isatty() and sys.stdin.isatty() and sys.stderr.isatty()):
print(f"\r{'Docgen Progress'.rjust(23)}: [{progress_percent:3.0f}%]...", end="", flush=True)
JinjaWriter.writeObject(template_name, os.path.join(output_path, obj.name.lower().replace(' ', '_') + '.md'), obj)
@@ -73,6 +75,7 @@ class ObjToMdAdapter(Adapter):
for obj in objects:
progress_percent = ((self.index+1)/self.files_to_write) * 100
self.index+=1
print(f"\r{'Docgen Progress'.rjust(23)}: [{progress_percent:3.0f}%]...", end="", flush=True)
if (sys.stdout.isatty() and sys.stdin.isatty() and sys.stderr.isatty()):
print(f"\r{'Docgen Progress'.rjust(23)}: [{progress_percent:3.0f}%]...", end="", flush=True)
JinjaWriter.writeObject(template_name, os.path.join(output_path, obj.date + '-' + obj.name.lower().replace(' ', '_') + '.md'), obj)
bin/contentctl_project/contentctl_infrastructure/builder/attack_enrichment.py
+3 -2
View File
@@ -3,7 +3,7 @@ import csv
import os
from posixpath import split
from typing import Optional
import sys
from attackcti import attack_client
import logging
@@ -43,7 +43,8 @@ class AttackEnrichment():
for index, technique in enumerate(all_enterprise['techniques']):
progress_percent = ((index+1)/len(all_enterprise['techniques'])) * 100
print(f"\r\t{'MITRE Technique Progress'.rjust(23)}: [{progress_percent:3.0f}%]...", end="", flush=True)
if (sys.stdout.isatty() and sys.stdin.isatty() and sys.stderr.isatty()):
print(f"\r\t{'MITRE Technique Progress'.rjust(23)}: [{progress_percent:3.0f}%]...", end="", flush=True)
apt_groups = []
for relationship in enterprise_relationships:
if (relationship['target_ref'] == technique['id']) and relationship['source_ref'].startswith('intrusion-set'):
bin/contentctl_project/contentctl_infrastructure/builder/cve_enrichment.py
+49 -28
View File
@@ -3,32 +3,55 @@ from pycvesearch import CVESearch
import functools
import os
import shelve
import time
import sys
CVESSEARCH_API_URL = 'https://cve.circl.lu'
CVE_CACHE_FILENAME = "lookups/CVE_CACHE.db"
@functools.cache
def cvesearch_helper(url:str, cve_id:str):
if not os.path.exists(CVE_CACHE_FILENAME):
print(f"Cache at {CVE_CACHE_FILENAME} not found - Creating it.")
cache = shelve.open(CVE_CACHE_FILENAME, flag='c', writeback=True)
if cve_id in cache:
req = cache[cve_id]
cache.close()
return req
NON_PERSISTENT_CACHE = {}
try:
cve = cvesearch_id_helper(url)
result = cve.id(cve_id)
except Exception as e:
raise(Exception(f"The option 'force_cached_or_offline' was used, but {cve_id} not found in {CVE_CACHE_FILENAME} and unable to connect to {CVESSEARCH_API_URL}"))
if result is None:
raise(Exception(f'CveEnrichment for [ {cve_id} ] failed - CVE does not exist'))
cache[cve_id] = result
cache.close()
@functools.cache
def cvesearch_helper(url:str, cve_id:str, force_cached_or_offline:bool=False, max_api_attempts:int=3, retry_sleep_seconds:int=5):
if max_api_attempts < 1:
raise(Exception(f"The minimum number of CVESearch API attempts is 1. You have passed {max_api_attempts}"))
if force_cached_or_offline:
if not os.path.exists(CVE_CACHE_FILENAME):
print(f"Cache at {CVE_CACHE_FILENAME} not found - Creating it.")
cache = shelve.open(CVE_CACHE_FILENAME, flag='c', writeback=True)
else:
cache = NON_PERSISTENT_CACHE
if cve_id in cache:
result = cache[cve_id]
#print(f"hit cve_enrichment: {time.time() - start:.2f}")
else:
api_attempts_remaining = max_api_attempts
while api_attempts_remaining > 0:
api_attempts_remaining -= 1
start = time.time()
try:
cve = cvesearch_id_helper(url)
result = cve.id(cve_id)
break
except Exception as e:
if api_attempts_remaining > 0:
print(f"The option 'force_cached_or_offline' was used, but {cve_id} not found in {CVE_CACHE_FILENAME} and unable to connect to {CVESSEARCH_API_URL}: {str(e)}")
print(f"Retrying the CVESearch API up to {api_attempts_remaining} more times after a sleep of {retry_sleep_seconds} seconds...")
time.sleep(retry_sleep_seconds)
else:
raise(Exception(f"The option 'force_cached_or_offline' was used, but {cve_id} not found in {CVE_CACHE_FILENAME} and unable to connect to {CVESSEARCH_API_URL} after {max_api_attempts} attempts: {str(e)}"))
if result is None:
raise(Exception(f'CveEnrichment for [ {cve_id} ] failed - CVE does not exist'))
cache[cve_id] = result
if force_cached_or_offline:
cache.close()
return result
@@ -47,20 +70,18 @@ class CveEnrichment():
def enrich_cve(self, cve_id: str, force_cached_or_offline: bool = False) -> dict:
cve_enriched = dict()
try:
if force_cached_or_offline is True:
result = cvesearch_helper(CVESSEARCH_API_URL, cve_id)
else:
cve = CVESearch(CVESSEARCH_API_URL)
result = cve.id(cve_id)
result = cvesearch_helper(CVESSEARCH_API_URL, cve_id, force_cached_or_offline)
cve_enriched['id'] = cve_id
cve_enriched['cvss'] = result['cvss']
cve_enriched['summary'] = result['summary']
except TypeError as TypeErr:
# there was a error calling the circl api lets just empty the object
print("WARNING, issue enriching {0}, with error: {1}".format(cve_id, str(TypeErr)))
cve_enriched = dict()
sys.exit(1)
except Exception as e:
print("WARNING - {0}".format(str(e)))
cve_enriched = dict()
sys.exit(1)
return cve_enriched
bin/contentctl_project/contentctl_infrastructure/builder/splunk_app_enrichment.py
+37 -25
View File
@@ -5,43 +5,52 @@ import functools
import pickle
import shelve
import os
import time
SPLUNKBASE_API_URL = "https://apps.splunk.com/api/apps/entriesbyid/"
APP_ENRICHMENT_CACHE_FILENAME = "lookups/APP_ENRICHMENT_CACHE.db"
NON_PERSISTENT_CACHE = {}
@functools.cache
def requests_get_helper(url:str)->bytes:
if not os.path.exists(APP_ENRICHMENT_CACHE_FILENAME):
print(f"Cache at {APP_ENRICHMENT_CACHE_FILENAME} not found - Creating it.")
cache = shelve.open(APP_ENRICHMENT_CACHE_FILENAME, flag='c', writeback=True)
def requests_get_helper(url:str, force_cached_or_offline:bool = False)->bytes:
if force_cached_or_offline:
if not os.path.exists(APP_ENRICHMENT_CACHE_FILENAME):
print(f"Cache at {APP_ENRICHMENT_CACHE_FILENAME} not found - Creating it.")
cache = shelve.open(APP_ENRICHMENT_CACHE_FILENAME, flag='c', writeback=True)
else:
cache = NON_PERSISTENT_CACHE
if url in cache:
req_content = cache[url]
else:
try:
req = requests.get(url)
req_content = req.content
cache[url] = req_content
except Exception as e:
raise(Exception(f"ERROR - Failed to get Splunk App Enrichment at {SPLUNKBASE_API_URL}"))
if force_cached_or_offline:
cache.close()
return req_content
try:
req = requests.get(url)
req_content = req.content
cache[url] = req_content
cache.close()
return req_content
except Exception as e:
raise(Exception(f"ERROR - Failed to get Splunk App Enrichment at {SPLUNKBASE_API_URL}"))
return req_content
class SplunkAppEnrichment():
@classmethod
def enrich_splunk_app(self, splunk_ta: str, force_cached_or_offline: bool = False) -> dict:
appurl = SPLUNKBASE_API_URL + splunk_ta
splunk_app_enriched = dict()
try:
if force_cached_or_offline is True:
content = requests_get_helper(appurl)
response_dict = xmltodict.parse(content)
else:
response = requests.get(appurl)
response_dict = xmltodict.parse(response.content)
content = requests_get_helper(appurl, force_cached_or_offline)
response_dict = xmltodict.parse(content)
# check if list since data changes depending on answer
url, results = self._parse_splunkbase_response(response_dict)
# grab the app name
@@ -50,20 +59,23 @@ class SplunkAppEnrichment():
splunk_app_enriched['name'] = i['#text']
# grab out the splunkbase url
if 'entriesbyid' in url:
if force_cached_or_offline is True:
content = requests_get_helper(url)
response_dict = xmltodict.parse(content)
else:
response = requests.get(url)
response_dict = xmltodict.parse(response.content)
content = requests_get_helper(url, force_cached_or_offline)
response_dict = xmltodict.parse(content)
#print(json.dumps(response_dict, indent=2))
url, results = self._parse_splunkbase_response(response_dict)
# chop the url so we grab the splunkbase portion but not direct download
splunk_app_enriched['url'] = url.rsplit('/', 4)[0]
except requests.exceptions.ConnectionError as connErr:
print(f"There was a connErr for ta {splunk_ta}: {connErr}")
# there was a connection error lets just capture the name
splunk_app_enriched['name'] = splunk_ta
splunk_app_enriched['url'] = ''
except Exception as e:
print(f"There was an unknown error enriching the Splunk TA [{splunk_ta}]: {str(e)}")
splunk_app_enriched['name'] = splunk_ta
splunk_app_enriched['url'] = ''
return splunk_app_enriched