Merge pull request #116 from splunk/updating_docs

updating readme
This commit is contained in:
Bhavin Patel
2019-06-05 11:23:11 -07:00
committed by GitHub
7 changed files with 39 additions and 870 deletions
-1
View File
@@ -2,7 +2,6 @@ repos:
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v2.1.0 # Use the ref you want to point at
hooks:
- id: trailing-whitespace
- id: check-executables-have-shebangs
exclude: 'package/bin/da_ess_contentupdate/|package/bin/splunklib/|venv/'
- id: check-json
+39 -5
View File
@@ -11,32 +11,66 @@ Contains a collection of security stories with their corresponding detection, in
# Usage
Can be consumed using:
#### [Splunk App](https://github.com/splunk/security-content/releases)
grab the latest release of DA-ESS-ContentUpdate and install it in a Splunk Server.
#### [API](https://github.com/splunk/security-content-api)
`curl https://g7jbilqdth.execute-api.us-west-2.amazonaws.com/api/`
#### [CLI](https://github.com/splunk/security-content-api/blob/master/content-update.py)
`python content-update.py -o $SPLUNK_HOME/etc/apps/DA-ESS-ContentUpdate --splunk_user admin --splunk_password xxxx`
# Structure
# Writing Content
Make sure you followed step 1 to 3 under [developing](https://github.com/splunk/security-content#developing) before starting.
1. select which content [piece](https://github.com/splunk/security-content#content-parts) you want to write.
2. copy an example and edit to your needs, most sure you at minium write a [story](stories/), [detection](detections/) and [investigation](investigations/)
3. make a pull request .. if CI failed refer to [troubleshooting](https://github.com/splunk/security-content#troubleshooting)
# Security Content Layout
![](static/structure.png)
#### Content Parts
* [stories/](stories/) - contains all analytics stories/use cases for ESCU
* [detections/](detections/) - splunk, uba and phantom detections that power stories
* [investigations/](investigations/) - splunk, and phantom investigation content that are used in stories
* [responses/](responses/) - automated splunk and phantom responses that are used in stories
* [baselines/](baselines/) - phantom and Splunk baseline needed to support detections in stories
#### Supporting parts
* [package/](package/) - splunk content app source files, includes lookups, binaries, and defaul config files
* [bin/](bin/) - where all binaries to produce, and test content lives
* [spec/](spec/) - location of all spec files that describe ESCU content
# Docs
* [docs/](docs/) - documentation for all of the spec files
* [spec/](spec/) - location of all spec files that describe ESCU content
# Developing
For getting pre-commit checks, install the hooks see below for steps:
1. Install circleci [CLI Tool](https://circleci.com/docs/2.0/local-cli/#installation)
2. create virtualenv and install requirements: `virtualenv venv && source venv/bin/activate && pip install -r requirements.txt`
3. install pre-commit `pre-commit install`
1. create virtualenv and install requirements: `virtualenv venv && source venv/bin/activate && pip install -r requirements.txt`
2. install pre-commit `pre-commit install`
3. Install circleci [CLI Tool](https://circleci.com/docs/2.0/local-cli/#installation)
To test a local change to CI or build make sure you are running docker and then
`circleci local execute -e GITHUB_TOKEN=$GITHUB_TOKEN --branch <your branch>`
To generate docs from schema automatically
1. install https://github.com/adobe/jsonschema2md
2. `jsonschema2md -d spec/v2/detections.json.spec -o docs`
# Troubleshooting
#### Our Automated Tests
1. [CI](https://github.com/splunk/security-content/blob/44946063173f7bc9921f0da0aa62139c084d1c51/.circleci/config.yml#L27) validates that the content was written to spec using [`validate.py`](https://github.com/splunk/security-content/blob/runstory/bin/generate.py), to run validation manually execute: `python bin/generate.py --path . --output package --storiesv1 --use_case_lib -v`
2. [CI](https://github.com/splunk/security-content/blob/44946063173f7bc9921f0da0aa62139c084d1c51/.circleci/config.yml#L60) generates splunk configuration files using [`generate.py`](https://github.com/splunk/security-content/blob/develop/bin/generate.py). If you want to export splunk conf files manually from the content run: `python bin/generate.py --path . --output package --storiesv1 --use_case_lib -v`
3. [CI](https://github.com/splunk/security-content/blob/44946063173f7bc9921f0da0aa62139c084d1c51/.circleci/config.yml#L107) builds a DA-ESS-ContentUpdate splunk package using the [Splunk Packaging Toolkit](http://dev.splunk.com/view/packaging-toolkit/SP-CAAAE9V)
4. [CI](https://github.com/splunk/security-content/blob/44946063173f7bc9921f0da0aa62139c084d1c51/.circleci/config.yml#L145) tests the newly produce package using [Splunk Appinspect](http://dev.splunk.com/view/appinspect/SP-CAAAE9U)
# Todo's
* build cli for interacting and developing
-44
View File
@@ -1,44 +0,0 @@
import enchant
import nltk
nltk.download('stopwords')
nltk.download('punkt')
nltk.download('wordnet')
common_manifest_words = [
"App", "Splunk", "AWS", "CVE", "IAM", "ESCU", "Netbackup", "CLI", "DNS", "CloudTrail",
"DNSTwist", "IP", "DHCP", "AMIs", "EventCode", "PowerShell", "dest", "instanceId",
"malware", "exfiltration", "exfiltrate", "Spectre", "Mitre", "ul", "li", "JBoss", "admin",
"subsearch", "cleartext", "WMI", "URL", "URLs", "ATT", "CK", "SMB", "ACL", "ACLs", "CIDR",
"VPC", "TXT", "USB", "SQL", "src", "VPN", "src", "br", "phishing", "weaponized",
"Ransomware", "ransomware", "VirusTotal", "RDP", "kerberized", "IMAP", "SSL",
"instanceType", "subsearch", "SPL", "rhaegal", "drogon", "BadRabbit", "WannaCry",
"CARBANAK", "wevtutil", "UAC", "ModifiedPersistence", "whoami", "tstats", "IPs",
"sourcetype", "JexBoss", "iptables", "wget", "NTFS", "noninteractive", "DLL", "DLLs",
"faux", "eval", "userName", "outlier", "DDoS", "logon", "noninteractive", "PID", "whois",
"misconfigured", "ARN", "MITM", "UDP", "MX", "Noriaki", "Iwasaki", "API", "APIs",
"awsRegion", "wmic", "vssadmin", "IDAPro", "AppData", "lifecycle", "prepending",
"prepended", "exfiltrating", "SDKs", "FQDN", "FQDNs", "DDNS", "Multipart", "CDN",
"cryptomining", "Cryptomining", "cryptojacking", "Cryptojacking", "typosquatting",
"DHL", "Samsam", "usernames", "antivirus", "NtLmSsP", "NTLM", "dnstwist", "url",
"netsh", "Netsh", "hostname", "hostnames", "csv", "amiID", "whitelist", "whitelisting",
"CVEs", "Cyber", "backticks", "backend", "backdoor", "backdoored", "wormable",
"RunDLL", "Schtasks", "logons", "spearphishing", "driveby", "cyber", "firstTime",
"lastTime", "ARNs", "GeoIP", "MaxMind", "DHS", "localgroup","PsExec", "psexec",
"accepteula", "cmdline", "apiCalls", "latestCount", "ol","internet","numberOfBlockedConnections", "arn", "AssumedRole",
"dataPointThreshold", "deviationThreshold", "eventNames","CloudWatch","VPCs", "ICMP"
"AssumeRole", "isnotnull", "requestingAccountId",
"http", "Sysmon", "CIM", "resourceId", "NetworkACLEvents", "securityGroupAPIs",
"CIDRs", "https", "username", "pst", "ost", "AccessKeyId","AssumeRole","isnotnull","AccessKeyId","redhat","bucketName","serviceName","javascript","VBScript",
"Mimikatz","SeDebugPrivilege","sekurlsa","lsass","offline","ExecutionPolicy","NirSoft","scomma","stext","Nirsoft","Clickspeed","clickstream","IIS","fraudster","splunk","DDOS","clickstreams",
"wscript", "cscript", "WinEventLog","filesystem","Recurly","whitelisted", "POC","Playbook", "SamSam", "Bruteforce", "POSTs", "java", "RCE"
]
d = enchant.Dict("en_US")
for word in common_manifest_words:
if not d.is_added(word):
d.add(word)
-383
View File
@@ -1,383 +0,0 @@
# Take the manifest files and build files for Enterprise Security Content Updates with markdown syntax
import datetime
import glob
import json
import os
import argparse
ALL_UUIDS = []
def markdown(x):
markdown = str(x)
markdown = markdown.replace("<code>", "`")
markdown = markdown.replace("</code>", "`")
markdown = markdown.replace("<b>", "**")
markdown = markdown.replace("</b>", "**")
# list tag replacements
markdown = markdown.replace("<ol><li>", "\\\n\\\n1. ")
markdown = markdown.replace("</li><li>", "\\\n\\\n1. ")
markdown = markdown.replace("</li></ol>", "")
markdown = markdown.replace("</li></ul>", "")
markdown = markdown.replace("<ul><li>", "\\\n\\\n1. ")
# break tags replacements
markdown = markdown.replace("<br></br>", "\\\n\\\n")
markdown = markdown.replace("<br/><br/>", "\\\n\\\n")
markdown = markdown.replace("<br/>", "\\\n\\\n")
return markdown
def main():
# Create conf files from manifest files
ssconf = open(OUTPUT_DIRECTORY + 'savedsearches.conf', 'w')
asconf = open(OUTPUT_DIRECTORY + 'analytic_stories.conf', 'w')
story_manifests = []
for story_manifest_file in glob.glob(os.path.join(MANIFEST_DIRECTORY, '*/stories/*.json')):
try:
story_manifest_data = json.loads(open(story_manifest_file, 'r').read())
except Exception as e:
print "Error reading %s" % story_manifest_file
print e
continue
story_manifests.append(story_manifest_data)
search_manifests = {}
for search_manifest_file in glob.glob(os.path.join(MANIFEST_DIRECTORY, '*/searches/*.json')):
try:
search_manifest = json.loads(open(search_manifest_file, 'r').read())
except Exception as e:
print "Error reading %s" % search_manifest_file
print e
continue
search_manifests[search_manifest['search_name']] = search_manifest
full_stories = {}
for story in story_manifests:
# Start building the story for the use case
full_stories[story['name']] = {}
full_stories[story['name']]['story_name'] = story['name']
full_stories[story['name']]['id'] = story['id']
if 'modification_date' in story:
full_stories[story['name']]['modification_date'] = story['modification_date']
else:
full_stories[story['name']]['modification_date'] = story['creation_date']
full_stories[story['name']]['creation_date'] = story['creation_date']
full_stories[story['name']]['description'] = story['description']
if 'references' not in story:
story['references'] = []
full_stories[story['name']]['references'] = story['references']
full_stories[story['name']]['category'] = story['category']
full_stories[story['name']]['version'] = story['version']
full_stories[story['name']]['mappings'] = {}
full_stories[story['name']]['data_models'] = set()
full_stories[story['name']]['providing_technologies'] = set()
full_stories[story['name']]['tags'] = set()
full_stories[story['name']]['narrative'] = story['narrative']
full_stories[story['name']]['detection_searches'] = []
full_stories[story['name']]['investigative_searches'] = []
full_stories[story['name']]['contextual_searches'] = []
full_stories[story['name']]['support_searches'] = []
all_searches = []
if 'detection_searches' in story['searches']:
full_stories[story['name']]['detection_searches'] = ["ESCU - %s - Rule" % ds for ds in
story['searches']['detection_searches']]
all_searches += story['searches']['detection_searches']
if 'investigative_searches' in story['searches']:
full_stories[story['name']]['investigative_searches'] = ["ESCU - " + invs for invs in
story['searches']['investigative_searches']]
all_searches += story['searches']['investigative_searches']
if 'contextual_searches' in story['searches']:
full_stories[story['name']]['contextual_searches'] = ["ESCU - " + cs for cs in
story['searches']['contextual_searches']]
all_searches += story['searches']['contextual_searches']
if 'support_searches' in story['searches']:
full_stories[story['name']]['support_searches'] = ["ESCU - " + ss for ss in
story['searches']['support_searches']]
all_searches += story['searches']['support_searches']
# Mark each search with what analytic stories it belongs too
for search in all_searches:
if search not in search_manifests:
continue
if 'analytic_story' not in search_manifests[search]:
search_manifests[search]['analytic_story'] = []
search_manifests[search]['analytic_story'].append(story['name'])
for key, search in search_manifests.iteritems():
# Gather the next steps
contextual_searches = []
investigative_searches = []
if 'analytic_story' in search:
for analytic_story in search['analytic_story']:
if not analytic_story:
continue
if 'mappings' in search:
for map_key in search['mappings']:
if map_key not in full_stories[analytic_story]['mappings']:
full_stories[analytic_story]['mappings'][map_key] = set()
for mapping in search['mappings'][map_key]:
full_stories[analytic_story]['mappings'][map_key].add(mapping)
if 'data_models' in search['data_metadata']:
for dm in search['data_metadata']['data_models']:
full_stories[analytic_story]['data_models'].add(dm)
if 'providing_technologies' in search['data_metadata']:
for tex in search['data_metadata']['providing_technologies']:
full_stories[analytic_story]['providing_technologies'].add(tex)
search_mod = datetime.datetime.strptime(search['modification_date'], "%Y-%m-%d")
latest_story_mod = datetime.datetime.strptime(full_stories[analytic_story]['modification_date'], "%Y-%m-%d")
if search_mod > latest_story_mod:
full_stories[analytic_story]['modification_date'] = search_mod.strftime("%Y-%m-%d")
if search['search_type'] == 'detection':
contextual_searches.extend(full_stories[analytic_story]['contextual_searches'])
investigative_searches.extend(full_stories[analytic_story]['investigative_searches'])
contextual_searches = sorted(list(set(contextual_searches)))
investigative_searches = sorted(list(set(investigative_searches)))
if 'correlation_rule' in search:
stanza_name = "ESCU - %s - Rule" % search['search_name'][0:86]
else:
stanza_name = "ESCU - %s" % search['search_name'][0:93]
ssconf.write("[%s]\n" % stanza_name)
ssconf.write("action.escu = 0\n")
ssconf.write("action.escu.enabled = 1\n")
ssconf.write("action.escu.creation_date = %s\n" % search['creation_date'])
if 'modification_date' in search:
ssconf.write("action.escu.modification_date = %s\n" % search['modification_date'])
else:
ssconf.write("action.escu.modification_date = %s\n" % search['creation_date'])
if 'asset_type' in search:
ssconf.write("action.escu.asset_at_risk = %s\n" % search['asset_type'])
if 'channel' in search:
ssconf.write("action.escu.channel = %s\n" % search['channel'])
if 'confidence' in search:
ssconf.write("action.escu.confidence = %s\n" % search['confidence'])
if 'eli5' in search:
eli5 = markdown(search['eli5'])
ssconf.write("action.escu.eli5 = %s\n" % eli5)
else:
ssconf.write("action.escu.eli5 = none\n")
if 'how_to_implement' in search:
how_to_implement = markdown(search['how_to_implement'])
ssconf.write("action.escu.how_to_implement = %s\n" % how_to_implement)
else:
ssconf.write("action.escu.how_to_implement = none\n")
if 'data_models' in search['data_metadata']:
ssconf.write("action.escu.data_models = %s\n" % json.dumps(search['data_metadata']['data_models']))
if search['search_type'] == 'detection':
ssconf.write("action.escu.full_search_name = ESCU - %s - Rule\n" % search['search_name'])
else:
ssconf.write("action.escu.full_search_name = ESCU - %s\n" % search['search_name'])
if 'mappings' in search:
ssconf.write("action.escu.mappings = %s\n" % json.dumps(search['mappings']))
if 'known_false_positives' in search:
known_false_positives = markdown(search['known_false_positives'])
ssconf.write("action.escu.known_false_positives = %s\n" % known_false_positives)
else:
ssconf.write("action.escu.known_false_positives = None at this time\n")
if 'search_type' in search:
ssconf.write("action.escu.search_type = %s\n" % search['search_type'])
if 'providing_technologies' in search['data_metadata']:
ssconf.write("action.escu.providing_technologies = %s\n" %
json.dumps(search['data_metadata']['providing_technologies']))
if 'analytic_story' in search:
ssconf.write("action.escu.analytic_story = %s\n" % json.dumps(search['analytic_story']))
if 'fields_required' in search:
ssconf.write("action.escu.fields_required = %s\n" % json.dumps(search['fields_required']))
if 'search_window' in search:
ssconf.write("action.escu.earliest_time_offset = %s\n" % search['search_window']['earliest_time_offset'])
ssconf.write("action.escu.latest_time_offset = %s\n" % search['search_window']['latest_time_offset'])
if 'phantom_playbooks' in search:
for p in (search['phantom_playbooks']):
ssconf.write("action.runphantomplaybook = 1\n")
ssconf.write("action.runphantomplaybook.param.phantom_server = %s\n" % p['phantom_server'])
ssconf.write("action.runphantomplaybook.param.playbook_name = %s\n" % p['playbook_name'])
ssconf.write("action.runphantomplaybook.param.playbook_display_name = %s\n" % p['playbook_display_name'])
ssconf.write("action.runphantomplaybook.param.playbook_url = %s\n" % p['playbook_url'])
ssconf.write("action.runphantomplaybook.param.sensitivity = %s\n" % p['sensitivity'])
ssconf.write("action.runphantomplaybook.param.severity = %s\n" % p['severity'])
if 'correlation_rule' in search:
ssconf.write("action.correlationsearch.enabled = 1\n")
ssconf.write("action.correlationsearch.label = %s\n" % search['search_name'])
if 'notable' in search['correlation_rule']:
ssconf.write("action.notable = 1\n")
if 'nes_fields' in search['correlation_rule']['notable']:
ssconf.write("action.notable.param.nes_fields = %s\n" % search['correlation_rule']['notable']['nes_fields'])
ssconf.write("action.notable.param.rule_description = %s\n" %
search['correlation_rule']['notable']['rule_description'])
ssconf.write("action.notable.param.rule_title = %s\n" %
search['correlation_rule']['notable']['rule_title'])
ssconf.write("action.notable.param.security_domain = %s\n" % search['security_domain'])
ssconf.write("action.notable.param.severity = %s\n" % search['confidence'])
cs_string = ""
for cs in contextual_searches:
cs_string += " - %s\\n" % cs
invs_string = ""
for invs in investigative_searches:
invs_string += " - %s\\n" % invs
if 'phantom_playbooks' in search:
for p in (search['phantom_playbooks']):
playbook_next_steps_string = "Splunk>Phantom Response Playbook - Monitor enrichment of the \
Splunk>Phantom Playbook called " + str(p['playbook_display_name']) + " and answer any \
analyst prompt in Mission Control with a response decision. \
Link to the playbook " + str(p['playbook_url'])
next_steps = "{\"version\": 1, \"data\": \"Recommended following \
steps:\\n\\n1. [[action|runphantomplaybook]]: Phantom playbook \
recommendations:\\n%s\\n2. [[action|escu_contextualize]]: Based \
on ESCU context gathering recommendations:\\n%s\\n3. [[action|escu_investigate]]: \
Based on ESCU investigate recommendations:\\n%s\"}" % (playbook_next_steps_string,
cs_string, invs_string)
ssconf.write("action.notable.param.next_steps = %s\n" % next_steps)
ssconf.write("action.notable.param.recommended_actions = runphantomplaybook, \
escu_contextualize, escu_investigate\n")
elif 'phantom_playbooks' not in search:
next_steps = "{\"version\": 1, \"data\": \"Recommended following steps:\\n\\n1. \
[[action|escu_contextualize]]: Based on ESCU context gathering recommendations:\\n%s\\n2. \
[[action|escu_investigate]]: Based on ESCU investigate \
recommendations:\\n%s\"}" % (cs_string, invs_string)
ssconf.write("action.notable.param.next_steps = %s\n" % next_steps)
ssconf.write("action.notable.param.recommended_actions = escu_contextualize, escu_investigate\n")
if 'risk' in search['correlation_rule']:
ssconf.write("action.risk = 1\n")
ssconf.write("action.risk.param._risk_object = %s\n" % search['correlation_rule']['risk']['risk_object'])
try:
ssconf.write("action.risk.param._risk_object_type = %s\n" %
search['correlation_rule']['risk']['risk_object_type'][0])
except Exception as e:
print "Error is risk object type %s" % search['correlation_rule']['risk']['risk_object_type']
print
continue
if len(search['correlation_rule']['risk']['risk_object_type']) > 1:
print "Error there is more than 1 risk object type %s" % search['correlation_rule']['risk']['risk_object_type']
ssconf.write("action.risk.param._risk_score = %d\n" % search['correlation_rule']['risk']['risk_score'])
ssconf.write("action.risk.param.verbose = 0\n")
if 'suppress' in search['correlation_rule']:
ssconf.write("alert.digest_mode = 1\n")
ssconf.write("alert.suppress = 1\n")
ssconf.write("alert.suppress.fields = %s\n" % search['correlation_rule']['suppress']['suppress_fields'])
ssconf.write("alert.suppress.period = %s\n" % search['correlation_rule']['suppress']['suppress_period'])
if 'scheduling' in search and 'cron_schedule' in search['scheduling']:
ssconf.write("cron_schedule = %s\n" % search['scheduling']['cron_schedule'])
search_description = markdown(search['search_description'])
ssconf.write("description = %s\n" % search_description)
if 'scheduling' in search:
ssconf.write("dispatch.earliest_time = %s\n" % search['scheduling']['earliest_time'])
ssconf.write("dispatch.latest_time = %s\n" % search['scheduling']['latest_time'])
ssconf.write("disabled=true\n")
if search['search_type'] == 'detection':
ssconf.write("enableSched = 1\n")
ssconf.write("counttype = number of events\n")
ssconf.write("relation = greater than\n")
ssconf.write("quantity = 0\n")
ssconf.write("realtime_schedule = 0\n")
ssconf.write("schedule_window = auto\n")
ssconf.write("is_visible = false\n")
search = markdown(search['search'])
ssconf.write("search = %s\n" % search)
ssconf.write("\n")
# Finish the story
for story_name, story in sorted(full_stories.iteritems()):
asconf.write("[%s]\n" % story_name)
asconf.write("category = %s\n" % story['category'])
asconf.write("creation_date = %s\n" % story['creation_date'])
data_models = list(story['data_models'])
if data_models:
data_models.sort()
asconf.write("data_models = %s\n" % json.dumps(data_models))
else:
asconf.write("data_models =\n")
description = markdown(story['description'])
asconf.write("description = %s\n" % description)
asconf.write("id = %s\n" % story['id'])
asconf.write("version = %s\n" % story['version'])
if 'mappings' in story:
for key in story['mappings'].keys():
story['mappings'][key] = list(story['mappings'][key])
asconf.write("mappings = %s\n" % json.dumps(story['mappings']))
asconf.write("modification_date = %s\n" % story['modification_date'])
asconf.write("reference = %s\n" % json.dumps(story['references']))
tex = list(story['providing_technologies'])
tex.sort()
asconf.write("providing_technologies = %s\n" % json.dumps(tex))
if story['detection_searches']:
asconf.write("detection_searches = %s\n" % json.dumps(story['detection_searches']))
if story['investigative_searches']:
asconf.write("investigative_searches = %s\n" % json.dumps(story['investigative_searches']))
if story['contextual_searches']:
asconf.write("contextual_searches = %s\n" % json.dumps(story['contextual_searches']))
if story['support_searches']:
asconf.write("support_searches = %s\n" % json.dumps(story['support_searches']))
if story['narrative']:
narrative = markdown(story['narrative'])
asconf.write("narrative = %s\n" % narrative)
asconf.write("\n")
usconf = open('src/default/usage_searches.conf', 'r')
usage_searches = usconf.read()
usconf.close()
ssconf.write('####################################################################\n\n')
ssconf.write(usage_searches)
asconf.close()
ssconf.close()
print "-----> ESCU ------- analytic_stories.conf file written for ESCU with Markdown"
print "-----> ESCU------- savedsearches.conf file written for ESCU with Markdown"
if __name__ == "__main__":
# grab arguments
parser = argparse.ArgumentParser(description="converts security-content manifests to source files", epilog="""
This tool converts manifests to the source files to be used by products, specfically Splunk's.
It generates the savesearches.conf, analyticsstories.conf files for ES.""")
parser.add_argument("-p", "--path", required=True, help="path to security-security content repo")
parser.add_argument("-o", "--output", required=True, help="path to the output directory")
# parse them
args = parser.parse_args()
MANIFEST_DIRECTORY = args.path
OUTPUT_DIRECTORY = args.output
main()
-275
View File
@@ -1,275 +0,0 @@
# Take the manifest files and build files for Enterprise Security
import datetime
import glob
import json
import os
ALL_UUIDS = []
MANIFEST_DIRECTORY = "."
OUTPUT_DIRECTORY = 'src/default/'
def markdown(x):
markdown = str(x)
markdown = markdown.replace("<code>", "`")
markdown = markdown.replace("</code>", "`")
markdown = markdown.replace("<b>", "**")
markdown = markdown.replace("</b>", "**")
# list tag replacements
markdown = markdown.replace("<ol><li>", "\\\n\\\n1. ")
markdown = markdown.replace("</li><li>", "\\\n\\\n1. ")
markdown = markdown.replace("</li></ol>", "")
markdown = markdown.replace("</li></ul>", "")
markdown = markdown.replace("<ul><li>", "\\\n\\\n1. ")
# break tags replacements
markdown = markdown.replace("<br></br>", "\\\n\\\n")
markdown = markdown.replace("<br/><br/>", "\\\n\\\n")
markdown = markdown.replace("<br/>", "\\\n\\\n")
return markdown
def main():
# Create conf files from manifest files
test = open(OUTPUT_DIRECTORY + 'analyticstories.conf', 'w')
story_manifests = []
for story_manifest_file in glob.glob(os.path.join(MANIFEST_DIRECTORY, '*/stories/*.json')):
try:
story_manifest_data = json.loads(open(story_manifest_file, 'r').read())
except Exception as e:
print "Error reading %s" % story_manifest_file
print e
continue
story_manifests.append(story_manifest_data)
search_manifests = {}
for search_manifest_file in glob.glob(os.path.join(MANIFEST_DIRECTORY, '*/searches/*.json')):
try:
search_manifest = json.loads(open(search_manifest_file, 'r').read())
except Exception as e:
print "Error reading %s" % search_manifest_file
print e
continue
search_manifests[search_manifest['search_name']] = search_manifest
full_stories = {}
for story in story_manifests:
# Start building the story for the use case
full_stories[story['name']] = {}
full_stories[story['name']]['story_name'] = story['name']
full_stories[story['name']]['id'] = story['id']
full_stories[story['name']]['maintainers'] = story['maintainers']
full_stories[story['name']]['spec_version'] = story['spec_version']
if 'modification_date' in story:
full_stories[story['name']]['modification_date'] = story['modification_date']
else:
full_stories[story['name']]['modification_date'] = story['creation_date']
full_stories[story['name']]['creation_date'] = story['creation_date']
full_stories[story['name']]['description'] = story['description']
if 'references' not in story:
story['references'] = []
full_stories[story['name']]['references'] = story['references']
full_stories[story['name']]['category'] = story['category']
full_stories[story['name']]['version'] = story['version']
full_stories[story['name']]['mappings'] = {}
full_stories[story['name']]['data_models'] = set()
full_stories[story['name']]['providing_technologies'] = set()
full_stories[story['name']]['tags'] = set()
full_stories[story['name']]['narrative'] = story['narrative']
full_stories[story['name']]['detection_searches'] = []
full_stories[story['name']]['investigative_searches'] = []
full_stories[story['name']]['contextual_searches'] = []
full_stories[story['name']]['support_searches'] = []
all_searches = []
if 'detection_searches' in story['searches']:
full_stories[story['name']]['detection_searches'] = ["ESCU - %s - Rule" % ds for ds in
story['searches']['detection_searches']]
all_searches += story['searches']['detection_searches']
if 'investigative_searches' in story['searches']:
full_stories[story['name']]['investigative_searches'] = ["ESCU - " + invs for invs in
story['searches']['investigative_searches']]
all_searches += story['searches']['investigative_searches']
if 'contextual_searches' in story['searches']:
full_stories[story['name']]['contextual_searches'] = ["ESCU - " + cs for cs in
story['searches']['contextual_searches']]
all_searches += story['searches']['contextual_searches']
if 'support_searches' in story['searches']:
full_stories[story['name']]['support_searches'] = ["ESCU - " + ss for ss in
story['searches']['support_searches']]
all_searches += story['searches']['support_searches']
# Mark each search with what analytic stories it belongs too
for search in all_searches:
if search not in search_manifests:
continue
if 'analytic_story' not in search_manifests[search]:
search_manifests[search]['analytic_story'] = []
search_manifests[search]['analytic_story'].append(story['name'])
for key, search in search_manifests.iteritems():
# Gather the next steps
contextual_searches = []
investigative_searches = []
if 'analytic_story' in search:
for analytic_story in search['analytic_story']:
if not analytic_story:
continue
if 'mappings' in search:
for map_key in search['mappings']:
if map_key not in full_stories[analytic_story]['mappings']:
full_stories[analytic_story]['mappings'][map_key] = set()
for mapping in search['mappings'][map_key]:
full_stories[analytic_story]['mappings'][map_key].add(mapping)
if 'data_models' in search['data_metadata']:
for dm in search['data_metadata']['data_models']:
full_stories[analytic_story]['data_models'].add(dm)
if 'providing_technologies' in search['data_metadata']:
for tex in search['data_metadata']['providing_technologies']:
full_stories[analytic_story]['providing_technologies'].add(tex)
search_mod = datetime.datetime.strptime(search['modification_date'], "%Y-%m-%d")
latest_story_mod = datetime.datetime.strptime(full_stories[analytic_story]['modification_date'], "%Y-%m-%d")
if search_mod > latest_story_mod:
full_stories[analytic_story]['modification_date'] = search_mod.strftime("%Y-%m-%d")
if search['search_type'] == 'detection':
contextual_searches.extend(full_stories[analytic_story]['contextual_searches'])
investigative_searches.extend(full_stories[analytic_story]['investigative_searches'])
contextual_searches = sorted(list(set(contextual_searches)))
investigative_searches = sorted(list(set(investigative_searches)))
# if 'correlation_rule' in search:
# stanza_name = "ESCU - %s - Rule" % search['search_name'][0:86]
# else:
# stanza_name = "ESCU - %s" % search['search_name'][0:93]
for story_name, story in sorted(full_stories.iteritems()):
test.write("[analytic_story://%s]\n" % story_name)
test.write("category = %s\n" % story['category'])
description = markdown(story['description'])
test.write("description = %s\n" % description)
test.write("maintainers = %s\n" % json.dumps(story['maintainers']))
if story['narrative']:
narrative = markdown(story['narrative'])
test.write("narrative = %s\n" % narrative)
test.write("references = %s\n" % json.dumps(story['references']))
if story['detection_searches']:
searches = json.dumps(story['detection_searches'])
if story['investigative_searches']:
searches = searches + json.dumps(story['investigative_searches'])
if story['contextual_searches']:
searches = searches + json.dumps(story['contextual_searches'])
if story['support_searches']:
searches = searches + json.dumps(story['support_searches'])
searches = searches.replace("][", ",")
test.write("searches = %s" % searches)
test.write("\n")
test.write("spec_version = %s\n" % json.dumps(story['spec_version']))
test.write("last_updated = %s\n" % story['modification_date'])
test.write("version = %s\n" % story['version'])
test.write("\n")
test.write("\n")
# testing save searches stanza
for key, search in search_manifests.iteritems():
# Gather the next steps
contextual_searches = []
investigative_searches = []
if 'analytic_story' in search:
for analytic_story in search['analytic_story']:
if not analytic_story:
continue
if 'mappings' in search:
for map_key in search['mappings']:
if map_key not in full_stories[analytic_story]['mappings']:
full_stories[analytic_story]['mappings'][map_key] = set()
for mapping in search['mappings'][map_key]:
full_stories[analytic_story]['mappings'][map_key].add(mapping)
if 'data_models' in search['data_metadata']:
for dm in search['data_metadata']['data_models']:
full_stories[analytic_story]['data_models'].add(dm)
if 'providing_technologies' in search['data_metadata']:
for tex in search['data_metadata']['providing_technologies']:
full_stories[analytic_story]['providing_technologies'].add(tex)
search_mod = datetime.datetime.strptime(search['modification_date'], "%Y-%m-%d")
latest_story_mod = datetime.datetime.strptime(full_stories[analytic_story]['modification_date'], "%Y-%m-%d")
if search_mod > latest_story_mod:
full_stories[analytic_story]['modification_date'] = search_mod.strftime("%Y-%m-%d")
if search['search_type'] == 'detection':
contextual_searches.extend(full_stories[analytic_story]['contextual_searches'])
investigative_searches.extend(full_stories[analytic_story]['investigative_searches'])
contextual_searches = sorted(list(set(contextual_searches)))
if search['search_type'] == 'detection':
test.write("[savedsearch://ESCU - %s - Rule]\n" % search['search_name'])
else:
test.write("[savedsearch://ESCU - %s]\n" % search['search_name'])
if 'search_type' in search:
test.write("type = %s\n" % search['search_type'])
if 'asset_type' in search:
test.write("asset_type = %s\n" % search['asset_type'])
if 'confidence' in search:
test.write("confidence = %s\n" % search['confidence'])
if 'eli5' in search:
explanation = markdown(search['eli5'])
test.write("explanation = %s\n" % explanation)
else:
test.write("explanation = none\n")
if 'how_to_implement' in search:
how_to_implement = markdown(search['how_to_implement'])
test.write("how_to_implement = %s\n" % how_to_implement)
else:
test.write("how_to_implement = none\n")
if 'mappings' in search:
test.write("annotations = %s\n" % json.dumps(search['mappings']))
if 'known_false_positives' in search:
known_false_positives = markdown(search['known_false_positives'])
test.write("known_false_positives = %s\n" % known_false_positives)
else:
test.write("known_false_positives = None at this time\n")
if 'providing_technologies' in search['data_metadata']:
test.write("providing_technologies = %s\n" % json.dumps(search['data_metadata']['providing_technologies']))
if 'search_window' in search:
test.write("earliest_time_offset = %s\n" % search['search_window']['earliest_time_offset'])
test.write("latest_time_offset = %s\n" % search['search_window']['latest_time_offset'])
test.write("\n")
test.write("\n")
test.close()
print "-----> ES-Hyuara ----- analyticstories.conf file written for Hyuara with Markdown ------------"
if __name__ == "__main__":
main()
-162
View File
@@ -1,162 +0,0 @@
'''
Take the manifest files and build files for Enterprise Security
'''
import glob
import os.path
import json
import argparse
# HIGH Level Fields
# Add these lines:
import nltk
from nltk.corpus import stopwords
import enchant
import grammar_check
stop_words_en = set(stopwords.words('english'))
def check_spelling(sentences):
errors = []
sp = enchant.Dict("en_US")
words = nltk.word_tokenize(sentences)
words = [word for word in words if word.isalpha()]
for word in words:
if not sp.check(word):
errors.append("\"%s\" not recognized" % word)
return errors
def check_grammar(sentences):
errors = []
grammar_tool = grammar_check.LanguageTool('en-US')
grammar_tool.disabled.add("EN_QUOTES")
grammar_tool.disabled.add("THREE_NN")
grammar_tool.disabled.add("NON3PRS_VERB")
grammar_tool.disabled.add("MASS_AGREEMENT")
grammar_tool.disabled.add("EN_UNPAIRED_BRACKETS")
grammar_tool.disabled.add("THIS_NNS")
grammar_tool.disabled.add("MUCH_COUNTABLE")
grammar_tool.disabled.add("COMMA_PARENTHESIS_WHITESPACE")
grammar_errors = grammar_tool.check(sentences)
for grammar_error in grammar_errors:
errors.append(grammar_error)
return errors
def spell_check_search_manifest(search):
''' Validate that the search has the required fields for savedsearches.conf '''
errors = []
spelling_errors = check_spelling(search['search_name'])
errors.extend(spelling_errors)
spelling_errors = check_spelling(search['search_description'])
grammar_errors = check_grammar(search['search_description'])
errors.extend(spelling_errors)
errors.extend(grammar_errors)
if 'how_to_implement' in search:
spelling_errors = check_spelling(search['how_to_implement'])
grammar_errors = check_grammar(search['how_to_implement'])
errors.extend(spelling_errors)
errors.extend(grammar_errors)
if 'eli5' in search:
spelling_errors = check_spelling(search['eli5'])
grammar_errors = check_grammar(search['eli5'])
errors.extend(spelling_errors)
errors.extend(grammar_errors)
if 'known_false_positives' in search:
search['known_false_positives'].encode('ascii')
spelling_errors = check_spelling(search['known_false_positives'])
if 'correlation_rule' in search and 'notable' in search['correlation_rule']:
spelling_errors = check_spelling(search['correlation_rule']['notable']['rule_title'])
errors.extend(spelling_errors)
spelling_errors = check_spelling(search['correlation_rule']['notable']['rule_description'])
errors.extend(spelling_errors)
return errors
def spell_check_story_manifest(story):
''' Validate that the analytic story manifest is in the proper format '''
errors = []
spelling_errors = check_spelling(story['name'])
errors.extend(spelling_errors)
spelling_errors = check_spelling(story['description'])
grammar_errors = check_grammar(story['description'])
errors.extend(spelling_errors)
errors.extend(grammar_errors)
spelling_errors = check_spelling(story['narrative'])
grammar_errors = check_grammar(story['narrative'])
errors.extend(spelling_errors)
errors.extend(grammar_errors)
def main():
''' Open manifest file, validate required fields '''
errors = False
story_manifest_files = os.path.join(MANIFEST_DIRECTORY, "*/stories/*.json")
stories_checked = 0
for story_manifest_file in glob.glob(story_manifest_files):
try:
story_manifest_data = json.loads(open(story_manifest_file, 'r').read())
except Exception as e:
print "Error reading %s" % story_manifest_file
errors = True
continue
stories_checked += 1
story_errors = spell_check_story_manifest(story_manifest_data)
if story_errors:
errors = True
for err in story_errors:
print os.path.basename(story_manifest_file)
print "\t%s" % err
search_manifest_files = os.path.join(MANIFEST_DIRECTORY, "*/searches/*.json")
searches_checked = 0
for search_manifest_file in glob.glob(search_manifest_files):
try:
search_manifest_data = json.loads(open(search_manifest_file, 'r').read())
except Exception as e:
errors = True
print "Error reading %s" % search_manifest_file
print e
continue
searches_checked += 1
sm_errors = spell_check_search_manifest(search_manifest_data)
if sm_errors:
errors = True
for err in sm_errors:
print os.path.basename(search_manifest_file)
print "\t%s" % err
print "%d story manifests checked" % stories_checked
print "%d search manifests checked" % searches_checked
if not errors:
print "No errors found"
if __name__ == "__main__":
# grab arguments
parser = argparse.ArgumentParser(description="spell checks the security-contents manifests")
parser.add_argument("-p", "--path", required=True, help="path to security-security content repo")
# parse them
args = parser.parse_args()
MANIFEST_DIRECTORY = args.path
main()
Binary file not shown.

After

Width:  |  Height:  |  Size: 54 KiB