mirror of
https://github.com/splunk/security_content
synced 2026-06-08 17:32:49 +00:00
divious1
810 lines
30 KiB
Python
810 lines
30 KiB
Python
#!/usr/bin/python
|
|
|
|
'''
|
|
Validates Manifest file under the security-content repo for correctness.
|
|
'''
|
|
|
|
import glob
|
|
import json
|
|
import jsonschema
|
|
import sys
|
|
import argparse
|
|
from os import path
|
|
|
|
|
|
def validate_detection_contentv2(detection, DETECTION_UUIDS, errors):
|
|
|
|
if detection['id'] == '':
|
|
errors.append('ERROR: Blank ID')
|
|
|
|
if detection['id'] in DETECTION_UUIDS:
|
|
errors.append('ERROR: Duplicate UUID found: %s' % detection['id'])
|
|
else:
|
|
DETECTION_UUIDS.append(detection['id'])
|
|
|
|
if detection['name'].endswith(" "):
|
|
errors.append(
|
|
"ERROR: Detection name has trailing spaces: '%s'" %
|
|
detection['name'])
|
|
|
|
try:
|
|
detection['description'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("ERROR: description not ascii")
|
|
|
|
if 'how_to_implement' in detection:
|
|
try:
|
|
detection['how_to_implement'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("ERROR: how_to_implement not ascii")
|
|
|
|
if 'eli5' in detection:
|
|
try:
|
|
detection['eli5'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("ERROR: eli5 not ascii")
|
|
|
|
if 'known_false_positives' in detection:
|
|
try:
|
|
detection['known_false_positives'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("ERROR: known_false_positives not ascii")
|
|
|
|
if detection['detect']['splunk']:
|
|
|
|
# do a regex match here instead of key values
|
|
if (detection['detect']['splunk']['correlation_rule']['search'].find('tstats') != -1) or \
|
|
(detection['detect']['splunk']['correlation_rule']['search'].find('datamodel') != -1):
|
|
|
|
if 'data_models' not in detection['data_metadata']:
|
|
errors.append("ERROR: The Splunk search uses a data model but 'data_models' field is not set")
|
|
|
|
if not detection['data_metadata']['data_models']:
|
|
errors.append("ERROR: The Splunk search uses a data model but 'data_models' is empty")
|
|
|
|
# do a regex match here instead of key values
|
|
if (detection['detect']['splunk']['correlation_rule']['search'].find('sourcetype') != -1):
|
|
if 'data_sourcetypes' not in detection['data_metadata']:
|
|
errors.append("ERROR: The Splunk search specifies a sourcetype but 'data_sourcetypes' \
|
|
field is not set")
|
|
|
|
if not detection['data_metadata']['data_sourcetypes']:
|
|
errors.append("ERROR: The Splunk search specifies a sourcetype but \
|
|
'data_sourcetypes' is empty")
|
|
|
|
return errors
|
|
|
|
|
|
def validate_investigation_contentv2(investigation, investigation_uuids, errors):
|
|
|
|
if investigation['id'] == '':
|
|
errors.append('ERROR: Blank ID')
|
|
|
|
if investigation['id'] in investigation_uuids:
|
|
errors.append('ERROR: Duplicate UUID found: %s' % investigation['id'])
|
|
else:
|
|
investigation_uuids.append(investigation['id'])
|
|
|
|
if investigation['name'].endswith(" "):
|
|
errors.append(
|
|
"ERROR: Investigation name has trailing spaces: '%s'" %
|
|
investigation['name'])
|
|
|
|
try:
|
|
investigation['description'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("ERROR: description not ascii")
|
|
|
|
if 'how_to_implement' in investigation:
|
|
try:
|
|
investigation['how_to_implement'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("ERROR: how_to_implement not ascii")
|
|
|
|
if 'eli5' in investigation:
|
|
try:
|
|
investigation['eli5'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("ERROR: eli5 not ascii")
|
|
|
|
if 'known_false_positives' in investigation:
|
|
try:
|
|
investigation['known_false_positives'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("ERROR: known_false_positives not ascii")
|
|
|
|
if 'splunk' in investigation['investigate']:
|
|
|
|
# do a regex match here instead of key values
|
|
if (investigation['investigate']['splunk']['search'].find('tstats') != -1) or \
|
|
(investigation['investigate']['splunk']['search'].find('datamodel') != -1):
|
|
|
|
if 'data_models' not in investigation['data_metadata']:
|
|
errors.append("ERROR: The Splunk search uses a data model but 'data_models' field is not set")
|
|
|
|
if not investigation['data_metadata']['data_models']:
|
|
errors.append("ERROR: The Splunk search uses a data model but 'data_models' is empty")
|
|
|
|
# do a regex match here instead of key values
|
|
if (investigation['investigate']['splunk']['search'].find('sourcetype') != -1):
|
|
if 'data_sourcetypes' not in investigation['data_metadata']:
|
|
errors.append("ERROR: The Splunk search specifies a sourcetype but 'data_sourcetypes' \
|
|
field is not set")
|
|
|
|
if not investigation['data_metadata']['data_sourcetypes']:
|
|
errors.append("ERROR: The Splunk search specifies a sourcetype but \
|
|
'data_sourcetypes' is empty")
|
|
|
|
return errors
|
|
|
|
|
|
def validate_baselines_contentv2(baseline, baselines_uuids, errors):
|
|
|
|
if baseline['id'] == '':
|
|
errors.append('ERROR: Blank ID')
|
|
|
|
if baseline['id'] in baselines_uuids:
|
|
errors.append('ERROR: Duplicate UUID found: %s' % baseline['id'])
|
|
else:
|
|
baselines_uuids.append(baseline['id'])
|
|
|
|
if baseline['name'].endswith(" "):
|
|
errors.append(
|
|
"ERROR: Investigation name has trailing spaces: '%s'" %
|
|
baseline['name'])
|
|
|
|
try:
|
|
baseline['description'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("ERROR: description not ascii")
|
|
|
|
if 'how_to_implement' in baseline:
|
|
try:
|
|
baseline['how_to_implement'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("ERROR: how_to_implement not ascii")
|
|
|
|
if 'eli5' in baseline:
|
|
try:
|
|
baseline['eli5'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("ERROR: eli5 not ascii")
|
|
|
|
if 'known_false_positives' in baseline:
|
|
try:
|
|
baseline['known_false_positives'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("ERROR: known_false_positives not ascii")
|
|
|
|
if baseline['baseline']['splunk']:
|
|
|
|
# do a regex match here instead of key values
|
|
if (baseline['baseline']['splunk']['search'].find('tstats') != -1) or \
|
|
(baseline['baseline']['splunk']['search'].find('datamodel') != -1):
|
|
|
|
if 'data_models' not in baseline['data_metadata']:
|
|
errors.append("ERROR: The Splunk search uses a data model but 'data_models' field is not set")
|
|
|
|
if not baseline['data_metadata']['data_models']:
|
|
errors.append("ERROR: The Splunk search uses a data model but 'data_models' is empty")
|
|
|
|
# do a regex match here instead of key values
|
|
if (baseline['baseline']['splunk']['search'].find('sourcetype') != -1):
|
|
if 'data_sourcetypes' not in baseline['data_metadata']:
|
|
errors.append("ERROR: The Splunk search specifies a sourcetype but 'data_sourcetypes' \
|
|
field is not set")
|
|
|
|
if not baseline['data_metadata']['data_sourcetypes']:
|
|
errors.append("ERROR: The Splunk search specifies a sourcetype but \
|
|
'data_sourcetypes' is empty")
|
|
|
|
return errors
|
|
|
|
|
|
def validate_detection_contentv1(detection, DETECTION_UUIDS, errors):
|
|
|
|
try:
|
|
detection['search_description'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("ERROR: description not ascii")
|
|
|
|
if detection['search_name'].endswith(" "):
|
|
errors.append(
|
|
"ERROR: Detection name has trailing spaces: '%s'" %
|
|
detection['search_name'])
|
|
|
|
if detection['search_id'] == '':
|
|
errors.append('ERROR: Blank ID')
|
|
|
|
if detection['search_id'] in DETECTION_UUIDS:
|
|
errors.append('ERROR: Duplicate UUID found: %s' % detection['search_id'])
|
|
else:
|
|
DETECTION_UUIDS.append(detection['search_id'])
|
|
|
|
if '| tstats' in detection['search'] or 'datamodel' in detection['search']:
|
|
if 'data_models' not in detection['data_metadata']:
|
|
errors.append(
|
|
"ERROR: The search uses a data model but 'data_models' \
|
|
field is not set")
|
|
|
|
if 'data_models' in detection and not \
|
|
detection['data_metadata']['data_models']:
|
|
errors.append(
|
|
"ERROR: The search uses a data model but 'data_models' is empty")
|
|
|
|
if 'sourcetype' in detection['search']:
|
|
if 'data_sourcetypes' not in detection['data_metadata']:
|
|
errors.append(
|
|
"ERROR: The search specifies a sourcetype but 'data_sourcetypes' \
|
|
field is not set")
|
|
|
|
if 'data_sourcetypes' in detection and not \
|
|
detection['data_metadata']['data_sourcetypes']:
|
|
errors.append(
|
|
"ERROR: The search specifies a sourcetype but \
|
|
'data_sourcetypes' is empty")
|
|
|
|
try:
|
|
detection['search_description'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("ERROR: search_description not ascii")
|
|
|
|
if 'how_to_implement' in detection:
|
|
try:
|
|
detection['how_to_implement'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("ERROR: how_to_implement not ascii")
|
|
|
|
if 'eli5' in detection:
|
|
try:
|
|
detection['eli5'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("eli5 not ascii")
|
|
|
|
if 'known_false_positives' in detection:
|
|
try:
|
|
detection['known_false_positives'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("ERROR: known_false_positives not ascii")
|
|
|
|
if 'correlation_rule' in detection and 'notable' in \
|
|
detection['correlation_rule']:
|
|
try:
|
|
detection['correlation_rule']['notable']['rule_title'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("ERROR: rule_title not ascii")
|
|
|
|
try:
|
|
detection['correlation_rule']['notable']['rule_description'].encode(
|
|
'ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("ERROR: rule_description not ascii")
|
|
|
|
return errors
|
|
|
|
|
|
def validate_investigation_contentv1(investigation, investigation_uuids, errors):
|
|
|
|
try:
|
|
investigation['search_description'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("ERROR: description not ascii")
|
|
|
|
if investigation['search_name'].endswith(" "):
|
|
errors.append(
|
|
"ERROR: Investigation name has trailing spaces: '%s'" %
|
|
investigation['search_name'])
|
|
|
|
if investigation['search_id'] == '':
|
|
errors.append('ERROR: Blank ID')
|
|
|
|
if investigation['search_id'] in investigation_uuids:
|
|
errors.append('ERROR: Duplicate UUID found: %s' % investigation['search_id'])
|
|
else:
|
|
investigation_uuids.append(investigation['search_id'])
|
|
|
|
if '| tstats' in investigation['search'] or 'datamodel' in investigation['search']:
|
|
if 'data_models' not in investigation['data_metadata']:
|
|
errors.append(
|
|
"ERROR: The search uses a data model but 'data_models' \
|
|
field is not set")
|
|
|
|
if 'data_models' in investigation and not \
|
|
investigation['data_metadata']['data_models']:
|
|
errors.append(
|
|
"ERROR: The search uses a data model but 'data_models' is empty")
|
|
|
|
if 'sourcetype' in investigation['search']:
|
|
if 'data_sourcetypes' not in investigation['data_metadata']:
|
|
errors.append(
|
|
"ERROR: The search specifies a sourcetype but 'data_sourcetypes' \
|
|
field is not set")
|
|
|
|
if 'data_sourcetypes' in investigation and not \
|
|
investigation['data_metadata']['data_sourcetypes']:
|
|
errors.append(
|
|
"ERROR: The search specifies a sourcetype but \
|
|
'data_sourcetypes' is empty")
|
|
|
|
try:
|
|
investigation['search_description'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("ERROR: search_description not ascii")
|
|
|
|
if 'how_to_implement' in investigation:
|
|
try:
|
|
investigation['how_to_implement'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("ERROR: how_to_implement not ascii")
|
|
|
|
if 'eli5' in investigation:
|
|
try:
|
|
investigation['eli5'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("eli5 not ascii")
|
|
|
|
if 'known_false_positives' in investigation:
|
|
try:
|
|
investigation['known_false_positives'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("ERROR: known_false_positives not ascii")
|
|
|
|
return errors
|
|
|
|
|
|
def validate_baselines_contentv1(baseline, baselines_uuids, errors):
|
|
|
|
try:
|
|
baseline['search_description'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("ERROR: description not ascii")
|
|
|
|
if baseline['search_name'].endswith(" "):
|
|
errors.append(
|
|
"ERROR: Baseline name has trailing spaces: '%s'" %
|
|
baseline['search_name'])
|
|
|
|
if baseline['search_id'] == '':
|
|
errors.append('ERROR: Blank ID')
|
|
|
|
if baseline['search_id'] in baselines_uuids:
|
|
errors.append('ERROR: Duplicate UUID found: %s' % baseline['search_id'])
|
|
else:
|
|
baselines_uuids.append(baseline['search_id'])
|
|
|
|
if '| tstats' in baseline['search'] or 'datamodel' in baseline['search']:
|
|
if 'data_models' not in baseline['data_metadata']:
|
|
errors.append(
|
|
"ERROR: The search uses a data model but 'data_models' \
|
|
field is not set")
|
|
|
|
if 'data_models' in baseline and not \
|
|
baseline['data_metadata']['data_models']:
|
|
errors.append(
|
|
"ERROR: The search uses a data model but 'data_models' is empty")
|
|
|
|
if 'sourcetype' in baseline['search']:
|
|
if 'data_sourcetypes' not in baseline['data_metadata']:
|
|
errors.append(
|
|
"ERROR: The search specifies a sourcetype but 'data_sourcetypes' \
|
|
field is not set")
|
|
|
|
if 'data_sourcetypes' in baseline and not \
|
|
baseline['data_metadata']['data_sourcetypes']:
|
|
errors.append(
|
|
"ERROR: The search specifies a sourcetype but \
|
|
'data_sourcetypes' is empty")
|
|
|
|
try:
|
|
baseline['search_description'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("ERROR: search_description not ascii")
|
|
|
|
if 'how_to_implement' in baseline:
|
|
try:
|
|
baseline['how_to_implement'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("ERROR: how_to_implement not ascii")
|
|
|
|
if 'eli5' in baseline:
|
|
try:
|
|
baseline['eli5'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("eli5 not ascii")
|
|
|
|
if 'known_false_positives' in baseline:
|
|
try:
|
|
baseline['known_false_positives'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("ERROR: known_false_positives not ascii")
|
|
|
|
return errors
|
|
|
|
|
|
def validate_investigation_content(investigation, investigation_uuids):
|
|
'''Validate that the content of a investigation manifest is correct'''
|
|
errors = []
|
|
|
|
# run v1 content validation
|
|
if investigation["spec_version"] == 1:
|
|
errors = validate_investigation_contentv1(investigation, investigation_uuids, errors)
|
|
|
|
if investigation["spec_version"] == 2:
|
|
errors = validate_investigation_contentv2(investigation, investigation_uuids, errors)
|
|
|
|
return errors
|
|
|
|
|
|
def validate_detection_content(detection, DETECTION_UUIDS):
|
|
'''Validate that the content of a detection manifest is correct'''
|
|
errors = []
|
|
|
|
# run v1 content validation
|
|
if detection["spec_version"] == 1:
|
|
errors = validate_detection_contentv1(detection, DETECTION_UUIDS, errors)
|
|
|
|
if detection["spec_version"] == 2:
|
|
errors = validate_detection_contentv2(detection, DETECTION_UUIDS, errors)
|
|
|
|
return errors
|
|
|
|
|
|
def validate_story_content(story, STORY_UUIDS):
|
|
''' Validate that the content of a story manifest is correct'''
|
|
errors = []
|
|
|
|
if story['id'] == '':
|
|
errors.append('ERROR: Blank ID')
|
|
|
|
if story['id'] in STORY_UUIDS:
|
|
errors.append('ERROR: Duplicate UUID found: %s' % story['id'])
|
|
else:
|
|
STORY_UUIDS.append(story['id'])
|
|
|
|
try:
|
|
story['description'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("ERROR: description not ascii")
|
|
|
|
try:
|
|
story['narrative'].encode('ascii')
|
|
except UnicodeEncodeError:
|
|
errors.append("ERROR: narrative not ascii")
|
|
|
|
return errors
|
|
|
|
|
|
def validate_baselines_content(baseline, baselines_uuids):
|
|
'''Validate that the content of a baseline manifest is correct'''
|
|
errors = []
|
|
|
|
# run v1 content validation
|
|
if baseline["spec_version"] == 1:
|
|
errors = validate_baselines_contentv1(baseline, baselines_uuids, errors)
|
|
|
|
if baseline["spec_version"] == 2:
|
|
errors = validate_baselines_contentv2(baseline, baselines_uuids, errors)
|
|
|
|
return errors
|
|
|
|
|
|
def validate_investigation(REPO_PATH, verbose):
|
|
''' Validates Investigation'''
|
|
|
|
INVESTIGATION_UUIDS = []
|
|
# retrive
|
|
v1_schema_file_investigative = path.join(path.expanduser(REPO_PATH), 'spec/v1/investigative_search.json.spec')
|
|
try:
|
|
v1_schema_investigative = json.loads(open(v1_schema_file_investigative, 'rb').read())
|
|
except IOError:
|
|
print "ERROR: reading version 1 investigations schema file {0}".format(v1_schema_file_investigative)
|
|
|
|
v1_schema_file_contexual = path.join(path.expanduser(REPO_PATH), 'spec/v1/contextual_search.json.spec')
|
|
try:
|
|
v1_schema_contexual = json.loads(open(v1_schema_file_contexual, 'rb').read())
|
|
except IOError:
|
|
print "ERROR: reading version 1 investigations schema file {0}".format(v1_schema_file_contexual)
|
|
|
|
v2_schema_file = path.join(path.expanduser(REPO_PATH), 'spec/v2/investigations.spec.json')
|
|
try:
|
|
v2_schema = json.loads(open(v2_schema_file, 'rb').read())
|
|
except IOError:
|
|
print "ERROR: reading version 2 investigations schema file {0}".format(v2_schema_file)
|
|
|
|
error = False
|
|
manifest_files = path.join(path.expanduser(REPO_PATH), "investigations/*.json")
|
|
|
|
for manifest_file in glob.glob(manifest_files):
|
|
if verbose:
|
|
print "processing investigation {0}".format(manifest_file)
|
|
# read in each investigation
|
|
try:
|
|
investigation = json.loads(
|
|
open(manifest_file, 'r').read())
|
|
except IOError:
|
|
print "ERROR: reading {0}".format(manifest_file)
|
|
error = True
|
|
continue
|
|
except ValueError:
|
|
print "ERROR: File is not proper JSON {0}".format(manifest_file)
|
|
error = True
|
|
continue
|
|
|
|
# validate v1 and v2 stories against spec for both investigations and old contexual searches
|
|
|
|
if investigation['spec_version'] == 1 and investigation['search_type'] == "contextual":
|
|
try:
|
|
jsonschema.validate(instance=investigation, schema=v1_schema_contexual)
|
|
except jsonschema.exceptions.ValidationError as json_ve:
|
|
print "ERROR: {0} at:\n\t{1}".format(json.dumps(json_ve.message), manifest_file)
|
|
print "\tAffected Object: {}".format(json.dumps(json_ve.instance))
|
|
error = True
|
|
elif investigation['spec_version'] == 1 and investigation['search_type'] == "investigative":
|
|
try:
|
|
jsonschema.validate(instance=investigation, schema=v1_schema_investigative)
|
|
except jsonschema.exceptions.ValidationError as json_ve:
|
|
print "ERROR: {0} at:\n\t{1}".format(json.dumps(json_ve.message), manifest_file)
|
|
print "\tAffected Object: {}".format(json.dumps(json_ve.instance))
|
|
error = True
|
|
elif investigation['spec_version'] == 2:
|
|
try:
|
|
jsonschema.validate(instance=investigation, schema=v2_schema)
|
|
except jsonschema.exceptions.ValidationError as json_ve:
|
|
print "ERROR: {0} at:\n\t{1}".format(json.dumps(json_ve.message), manifest_file)
|
|
print "\tAffected Object: {}".format(json.dumps(json_ve.instance))
|
|
error = True
|
|
else:
|
|
print "ERROR: Story {0} does not contain a spec_version which is required".format(manifest_file)
|
|
error = True
|
|
continue
|
|
|
|
# now lets validate the content
|
|
investigation_errors = validate_investigation_content(investigation, INVESTIGATION_UUIDS)
|
|
if investigation_errors:
|
|
error = True
|
|
for err in investigation_errors:
|
|
print "{0} at:\n\t {1}".format(err, manifest_file)
|
|
|
|
return error
|
|
|
|
|
|
def validate_detection(REPO_PATH, verbose):
|
|
''' Validates Detections'''
|
|
|
|
DETECTION_UUIDS = []
|
|
# retrive
|
|
v1_schema_file = path.join(path.expanduser(REPO_PATH), 'spec/v1/detection_search.json.spec')
|
|
try:
|
|
v1_schema = json.loads(open(v1_schema_file, 'rb').read())
|
|
except IOError:
|
|
print "ERROR: reading version 1 detection schema file {0}".format(v1_schema_file)
|
|
except ValueError:
|
|
print "ERROR: File is not proper JSON {0}".format(v1_schema_file)
|
|
|
|
v2_schema_file = path.join(path.expanduser(REPO_PATH), 'spec/v2/detections.spec.json')
|
|
try:
|
|
v2_schema = json.loads(open(v2_schema_file, 'rb').read())
|
|
except IOError:
|
|
print "ERROR: reading version 2 detection schema file {0}".format(v2_schema_file)
|
|
except ValueError:
|
|
print "ERROR: File is not proper JSON {0}".format(v2_schema_file)
|
|
|
|
error = False
|
|
manifest_files = path.join(path.expanduser(REPO_PATH), "detections/*.json")
|
|
|
|
for manifest_file in glob.glob(manifest_files):
|
|
if verbose:
|
|
print "processing detection {0}".format(manifest_file)
|
|
|
|
# read in each story
|
|
try:
|
|
detection = json.loads(
|
|
open(manifest_file, 'r').read())
|
|
except IOError:
|
|
print "Error reading {0}".format(manifest_file)
|
|
error = True
|
|
continue
|
|
|
|
# validate v1 and v2 stories against spec
|
|
if detection['spec_version'] == 1:
|
|
try:
|
|
jsonschema.validate(instance=detection, schema=v1_schema)
|
|
except jsonschema.exceptions.ValidationError as json_ve:
|
|
print "ERROR: {0} at:\n\t{1}".format(json.dumps(json_ve.message), manifest_file)
|
|
print "\tAffected Object: {}".format(json.dumps(json_ve.instance))
|
|
error = True
|
|
|
|
elif detection['spec_version'] == 2:
|
|
try:
|
|
jsonschema.validate(instance=detection, schema=v2_schema)
|
|
except jsonschema.exceptions.ValidationError as json_ve:
|
|
print "ERROR: {0} at:\n\t{1}".format(json.dumps(json_ve.message), manifest_file)
|
|
print "\tAffected Object: {}".format(json.dumps(json_ve.instance))
|
|
error = True
|
|
|
|
else:
|
|
print "ERROR: Story {0} does not contain a spec_version which is required".format(manifest_file)
|
|
error = True
|
|
continue
|
|
|
|
# now lets validate the content
|
|
detection_errors = validate_detection_content(detection, DETECTION_UUIDS)
|
|
if detection_errors:
|
|
error = True
|
|
for err in detection_errors:
|
|
print "{0} at:\n\t {1}".format(err, manifest_file)
|
|
|
|
return error
|
|
|
|
|
|
def validate_story(REPO_PATH, verbose):
|
|
''' Validates Stories'''
|
|
|
|
STORY_UUIDS = []
|
|
|
|
# retrive
|
|
v1_schema_file = path.join(path.expanduser(REPO_PATH), 'spec/v1/analytic_story.json.spec')
|
|
try:
|
|
v1_schema = json.loads(open(v1_schema_file, 'rb').read())
|
|
except IOError:
|
|
print "ERROR: reading version 1 story schema file {0}".format(v1_schema_file)
|
|
except ValueError:
|
|
print "ERROR: File is not proper JSON {0}".format(v1_schema_file)
|
|
|
|
v2_schema_file = path.join(path.expanduser(REPO_PATH), 'spec/v2/story.spec.json')
|
|
try:
|
|
v2_schema = json.loads(open(v2_schema_file, 'rb').read())
|
|
except IOError:
|
|
print "ERROR: reading version 2 story schema file {0}".format(v2_schema_file)
|
|
except ValueError:
|
|
print "ERROR: File is not proper JSON {0}".format(v2_schema_file)
|
|
|
|
error = False
|
|
story_manifest_files = path.join(path.expanduser(REPO_PATH), "stories/*.json")
|
|
|
|
for story_manifest_file in glob.glob(story_manifest_files):
|
|
if verbose:
|
|
print "processing story {0}".format(story_manifest_file)
|
|
|
|
# read in each story
|
|
try:
|
|
story = json.loads(
|
|
open(story_manifest_file, 'r').read())
|
|
except IOError:
|
|
print "Error reading {0}".format(story_manifest_file)
|
|
error = True
|
|
continue
|
|
|
|
# validate v1 and v2 stories against spec
|
|
if story['spec_version'] == 1:
|
|
try:
|
|
jsonschema.validate(instance=story, schema=v1_schema)
|
|
except jsonschema.exceptions.ValidationError as json_ve:
|
|
print "ERROR: {0} at:\n\t{1}".format(json.dumps(json_ve.message), story_manifest_file)
|
|
print "\tAffected Object: {}".format(json.dumps(json_ve.instance))
|
|
error = True
|
|
|
|
elif story['spec_version'] == 2:
|
|
try:
|
|
jsonschema.validate(instance=story, schema=v2_schema)
|
|
except jsonschema.exceptions.ValidationError as json_ve:
|
|
print "ERROR: {0} at:\n\t{1}".format(json.dumps(json_ve.message), story_manifest_file)
|
|
print "\tAffected Object: {}".format(json.dumps(json_ve.instance))
|
|
error = True
|
|
|
|
else:
|
|
print "ERROR: Story {0} does not contain a spec_version which is required".format(story_manifest_file)
|
|
error = True
|
|
continue
|
|
|
|
# now lets validate the content
|
|
story_errors = validate_story_content(story, STORY_UUIDS)
|
|
if story_errors:
|
|
error = True
|
|
for err in story_errors:
|
|
print "{0} at:\n\t {1}".format(err, story_manifest_file)
|
|
|
|
return error
|
|
|
|
|
|
def validate_baselines(REPO_PATH, verbose):
|
|
''' Validates Baselines'''
|
|
|
|
BASELINE_UUIDS = []
|
|
|
|
# retrive
|
|
v1_schema_file = path.join(path.expanduser(REPO_PATH), 'spec/v1/support_search.json.spec')
|
|
try:
|
|
v1_schema = json.loads(open(v1_schema_file, 'rb').read())
|
|
except IOError:
|
|
print "ERROR: reading version 1 baseline schema file {0}".format(v1_schema_file)
|
|
except ValueError:
|
|
print "ERROR: File is not proper JSON {0}".format(v1_schema_file)
|
|
|
|
v2_schema_file = path.join(path.expanduser(REPO_PATH), 'spec/v2/baselines.spec.json')
|
|
try:
|
|
v2_schema = json.loads(open(v2_schema_file, 'rb').read())
|
|
except IOError:
|
|
print "ERROR: reading version 2 baseline schema file {0}".format(v2_schema_file)
|
|
except ValueError:
|
|
print "ERROR: File is not proper JSON {0}".format(v2_schema_file)
|
|
|
|
error = False
|
|
baselines_manifest_files = path.join(path.expanduser(REPO_PATH), "baselines/*.json")
|
|
|
|
for baselines_manifest_file in glob.glob(baselines_manifest_files):
|
|
if verbose:
|
|
print "processing story {0}".format(baselines_manifest_file)
|
|
|
|
# read in each baseline
|
|
try:
|
|
baseline = json.loads(
|
|
open(baselines_manifest_file, 'r').read())
|
|
except IOError:
|
|
print "Error reading {0}".format(baselines_manifest_file)
|
|
error = True
|
|
continue
|
|
|
|
# validate v1 and v2 stories against spec
|
|
if baseline['spec_version'] == 1:
|
|
try:
|
|
jsonschema.validate(instance=baseline, schema=v1_schema)
|
|
except jsonschema.exceptions.ValidationError as json_ve:
|
|
print "ERROR: {0} at:\n\t{1}".format(json.dumps(json_ve.message), baselines_manifest_file)
|
|
print "\tAffected Object: {}".format(json.dumps(json_ve.instance))
|
|
error = True
|
|
|
|
elif baseline['spec_version'] == 2:
|
|
try:
|
|
jsonschema.validate(instance=baseline, schema=v2_schema)
|
|
except jsonschema.exceptions.ValidationError as json_ve:
|
|
print "ERROR: {0} at:\n\t{1}".format(json.dumps(json_ve.message), baselines_manifest_file)
|
|
print "\tAffected Object: {}".format(json.dumps(json_ve.instance))
|
|
error = True
|
|
|
|
else:
|
|
print "ERROR: Baseline {0} does not contain a spec_version which is required".format(baselines_manifest_file)
|
|
error = True
|
|
continue
|
|
|
|
# now lets validate the content
|
|
baselines_errors = validate_baselines_content(baseline, BASELINE_UUIDS)
|
|
if baselines_errors:
|
|
error = True
|
|
for err in baselines_errors:
|
|
print "{0} at:\n\t {1}".format(err, baselines_manifest_file)
|
|
|
|
return error
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# grab arguments
|
|
parser = argparse.ArgumentParser(description="validates security content manifest files", epilog="""
|
|
Validates security manifest for correctness, adhering to spec and other common items.
|
|
VALIDATE DOES NOT PROCESS RESPONSES SPEC for the moment.""")
|
|
parser.add_argument("-p", "--path", required=True, help="path to security-security content repo")
|
|
parser.add_argument("-v", "--verbose", required=False, action='store_true', help="prints verbose output")
|
|
# parse them
|
|
args = parser.parse_args()
|
|
REPO_PATH = args.path
|
|
verbose = args.verbose
|
|
|
|
story_error = validate_story(REPO_PATH, verbose)
|
|
|
|
detection_error = validate_detection(REPO_PATH, verbose)
|
|
|
|
investigation_error = validate_investigation(REPO_PATH, verbose)
|
|
|
|
baseline_error = validate_baselines(REPO_PATH, verbose)
|
|
|
|
if story_error:
|
|
sys.exit("Errors found")
|
|
elif detection_error:
|
|
sys.exit("Errors found")
|
|
elif investigation_error:
|
|
sys.exit("Errors found")
|
|
elif baseline_error:
|
|
sys.exit("Errors found")
|
|
else:
|
|
print "No Errors found"
|